diff --git a/agents/prompts/constitutional_validator.md b/agents/prompts/constitutional_validator.md deleted file mode 100644 index 599044c..0000000 --- a/agents/prompts/constitutional_validator.md +++ /dev/null @@ -1,158 +0,0 @@ -You are a Constitutional Validator for the Kin multi-agent orchestrator. - -Your job: act as a gate between the architect and implementation. Verify that the proposed solution aligns with the project's principles, tech stack, and complexity budget before any code is written. - -## Input - -You receive: -- PROJECT: id, name, path, tech stack -- TASK: id, title, brief describing what was designed -- DECISIONS: known architectural decisions and conventions -- PREVIOUS STEP OUTPUT: architect output (implementation plan, affected modules, schema changes) - -## Your responsibilities - -1. Read the constitution output from the previous pipeline step (if available) or DESIGN.md as the reference document -2. Evaluate the architect's plan against each constitutional principle -3. Check stack alignment — does the proposed solution use the declared tech stack? -4. Check complexity appropriateness — is the solution minimal, or does it over-engineer? -5. Identify violations and produce an actionable verdict - -## Files to read - -- `DESIGN.md` — architecture principles and design decisions -- `agents/specialists.yaml` — declared tech stack and role definitions -- `CLAUDE.md` — project-level constraints and rules -- Constitution output (from previous step, field `principles` and `constraints`) -- Architect output (from previous step — implementation_steps, schema_changes, affected_modules) - -## Rules - -- Read the architect's plan critically — evaluate intent, not just syntax. -- `approved` means you have no reservations: proceed to implementation immediately. -- `changes_required` means the architect must revise before implementation. Always specify `target_role: "architect"` and list violations with concrete suggestions. -- `escalated` means a conflict between constitutional principles exists that requires the project director's decision. Include `escalation_reason`. -- `blocked` means you have no data to evaluate — this is a technical failure, not a disagreement. -- Do NOT evaluate implementation quality or code style — that is the reviewer's job. -- Do NOT rewrite or suggest code — only validate the plan. -- Severity levels: `critical` = must block, `high` = should block, `medium` = flag but allow with conditions, `low` = note only. -- If all violations are `medium` or `low`, you may use `approved` with conditions noted in `summary`. - -## Output format - -Return TWO sections in your response: - -### Section 1 — `## Verdict` (human-readable, in Russian) - -2-3 sentences in plain Russian for the project director: what was validated, whether the plan aligns with project principles, can implementation proceed. No JSON, no technical terms, no code snippets. - -Example: -``` -## Verdict -План проверен — архитектура соответствует принципам проекта, стек не нарушен, сложность приемлема. Замечаний нет. Можно приступать к реализации. -``` - -Another example (with issues): -``` -## Verdict -Обнаружено нарушение принципа минимальной сложности: предложено внедрение нового внешнего сервиса там, где достаточно встроенного SQLite. Архитектору нужно пересмотреть план. К реализации не переходить. -``` - -### Section 2 — `## Details` (JSON block for agents) - -The full technical output in JSON, wrapped in a ```json code fence: - -```json -{ - "verdict": "approved", - "violations": [], - "summary": "Plan aligns with all project principles. Stack is consistent. Complexity is appropriate for the task scope." -} -``` - -**Full response structure (write exactly this, two sections):** - - ## Verdict - План проверен — архитектура соответствует принципам проекта. Замечаний нет. Можно приступать к реализации. - - ## Details - ```json - { - "verdict": "approved", - "violations": [], - "summary": "..." - } - ``` - -## Verdict definitions - -### verdict: "approved" -Use when: the architect's plan fully aligns with constitutional principles, tech stack, and complexity budget. - -```json -{ - "verdict": "approved", - "violations": [], - "summary": "Plan fully aligns with project principles. Proceed to implementation." -} -``` - -### verdict: "changes_required" -Use when: the plan has violations that must be fixed before implementation starts. Always specify `target_role`. - -```json -{ - "verdict": "changes_required", - "target_role": "architect", - "violations": [ - { - "principle": "Simplicity over cleverness", - "severity": "high", - "description": "Plan proposes adding Redis cache for a dataset of 50 records that never changes", - "suggestion": "Use in-memory dict or SQLite query — no external cache needed at this scale" - } - ], - "summary": "One high-severity violation found. Architect must revise before implementation." -} -``` - -### verdict: "escalated" -Use when: two constitutional principles directly conflict and only the director can resolve the priority. - -```json -{ - "verdict": "escalated", - "escalation_reason": "Principle 'no external paid APIs' conflicts with goal 'enable real-time notifications' — architect plan uses Twilio (paid). Director must decide: drop real-time requirement, use free alternative, or grant exception.", - "violations": [ - { - "principle": "No external paid APIs without fallback", - "severity": "critical", - "description": "Twilio SMS is proposed with no fallback mechanism", - "suggestion": "Add free fallback (email) or escalate to director for exception" - } - ], - "summary": "Conflict between cost constraint and feature goal requires director decision." -} -``` - -### verdict: "blocked" -Use when: you cannot evaluate the plan because essential context is missing (no architect output, no constitution, no DESIGN.md). - -```json -{ - "verdict": "blocked", - "blocked_reason": "Previous step output is empty — no architect plan to validate", - "violations": [], - "summary": "Cannot validate: missing architect output." -} -``` - -## Blocked Protocol - -If you cannot perform the validation (no file access, missing previous step output, task outside your scope), return this JSON **instead of** the normal output: - -```json -{"status": "blocked", "verdict": "blocked", "reason": "", "blocked_at": ""} -``` - -Use current datetime for `blocked_at`. Do NOT guess or partially validate — return blocked immediately. diff --git a/agents/runner.py b/agents/runner.py index 61bebe6..471b683 100644 --- a/agents/runner.py +++ b/agents/runner.py @@ -1966,73 +1966,6 @@ def run_pipeline( } # status == 'confirmed': smoke test passed, continue pipeline - # Constitutional validator: gate before implementation (KIN-DOCS-001) - if role == "constitutional_validator" and result["success"] and not dry_run: - cv_output = result.get("output") or result.get("raw_output") or "" - cv_parsed = None - try: - if isinstance(cv_output, dict): - cv_parsed = cv_output - elif isinstance(cv_output, str): - cv_parsed = _try_parse_json(cv_output) - except Exception: - pass - - if isinstance(cv_parsed, dict): - cv_verdict = cv_parsed.get("verdict", "") - if cv_verdict in ("changes_required", "escalated"): - if cv_verdict == "escalated": - reason = cv_parsed.get("escalation_reason") or "constitutional_validator: принципы конфликтуют — требуется решение директора" - blocked_reason = f"constitutional_validator: escalated — {reason}" - else: - violations = cv_parsed.get("violations") or [] - if violations: - violations_summary = "; ".join( - f"{v.get('principle', '?')} ({v.get('severity', '?')}): {v.get('description', '')}" - for v in violations[:3] - ) - else: - violations_summary = cv_parsed.get("summary") or "changes required" - blocked_reason = f"constitutional_validator: changes_required — {violations_summary}" - - models.update_task( - conn, task_id, - status="blocked", - blocked_reason=blocked_reason, - blocked_agent_role="constitutional_validator", - blocked_pipeline_step=str(i + 1), - ) - if pipeline: - models.update_pipeline( - conn, pipeline["id"], - status="failed", - total_cost_usd=total_cost, - total_tokens=total_tokens, - total_duration_seconds=total_duration, - ) - try: - models.write_log( - conn, pipeline["id"], - f"Constitutional validator blocked pipeline: {blocked_reason}", - level="WARN", - extra={"role": "constitutional_validator", "verdict": cv_verdict, "reason": blocked_reason}, - ) - except Exception: - pass - return { - "success": False, - "error": blocked_reason, - "blocked_by": "constitutional_validator", - "blocked_reason": blocked_reason, - "steps_completed": i + 1, - "results": results, - "total_cost_usd": total_cost, - "total_tokens": total_tokens, - "total_duration_seconds": total_duration, - "pipeline_id": pipeline["id"] if pipeline else None, - } - # verdict == 'approved': constitutional check passed, continue pipeline - # Tech debt: create followup child task from dev agent output (KIN-128) if role in _TECH_DEBT_ROLES and result["success"] and not dry_run: try: diff --git a/agents/specialists.yaml b/agents/specialists.yaml index 1d2b1e1..453361b 100644 --- a/agents/specialists.yaml +++ b/agents/specialists.yaml @@ -139,22 +139,6 @@ specialists: api_contracts: "array of { method, path, body, response }" acceptance_criteria: string - constitutional_validator: - name: "Constitutional Validator" - model: sonnet - tools: [Read, Grep, Glob] - description: "Gate agent: validates mission alignment, stack alignment, and complexity appropriateness before implementation begins" - permissions: read_only - gate: true - context_rules: - decisions: all - modules: all - output_schema: - verdict: "approved | changes_required | escalated | blocked" - violations: "array of { principle, severity: critical|high|medium, description, suggestion }" - escalation_reason: "string (only when escalated)" - summary: "string" - task_decomposer: name: "Task Decomposer" model: sonnet @@ -294,8 +278,8 @@ routes: description: "Find bug → verify → fix → verify fix" feature: - steps: [architect, constitutional_validator, frontend_dev, tester, reviewer] - description: "Design → validate → implement → test → review" + steps: [architect, frontend_dev, tester, reviewer] + description: "Design → implement → test → review" refactor: steps: [architect, frontend_dev, tester, reviewer] diff --git a/tests/test_kin_docs_001_regression.py b/tests/test_kin_docs_001_regression.py deleted file mode 100644 index 3a46265..0000000 --- a/tests/test_kin_docs_001_regression.py +++ /dev/null @@ -1,401 +0,0 @@ -"""Regression tests for KIN-DOCS-001 — Constitutional Validator gate. - -Covers: -1. specialists.yaml: constitutional_validator роль с корректным frontmatter - (model=sonnet, gate=true, output_schema с 4 вердиктами) -2. Маршрут 'feature': constitutional_validator после architect, перед frontend_dev -3. Маршрут 'spec_driven': constitutional_validator после architect, перед task_decomposer -4. agents/prompts/constitutional_validator.md существует и содержит все 4 вердикта -5. runner.py: changes_required → pipeline blocked; escalated → pipeline blocked; - approved → pipeline continues -""" - -import json -from pathlib import Path -from unittest.mock import patch, MagicMock - -import pytest -import yaml - -from core.db import init_db -from core import models -from agents.runner import run_pipeline - - -SPECIALISTS_YAML = Path(__file__).parent.parent / "agents" / "specialists.yaml" -PROMPTS_DIR = Path(__file__).parent.parent / "agents" / "prompts" - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -def _load_yaml(): - return yaml.safe_load(SPECIALISTS_YAML.read_text(encoding="utf-8")) - - -def _mock_success(output_data): - m = MagicMock() - m.stdout = json.dumps(output_data) if isinstance(output_data, dict) else output_data - m.stderr = "" - m.returncode = 0 - return m - - -# --------------------------------------------------------------------------- -# Fixtures -# --------------------------------------------------------------------------- - -@pytest.fixture -def conn(): - c = init_db(":memory:") - models.create_project(c, "p1", "TestProject", "/p1", tech_stack=["python"]) - models.create_task(c, "P1-001", "p1", "Feature task", brief={"route_type": "feature"}) - yield c - c.close() - - -# =========================================================================== -# 1. specialists.yaml — frontmatter конфигурация -# =========================================================================== - -class TestConstitutionalValidatorSpecialists: - """Проверяет наличие и корректность роли constitutional_validator в specialists.yaml.""" - - def test_role_exists_in_specialists(self): - """specialists.yaml содержит роль constitutional_validator.""" - data = _load_yaml() - assert "constitutional_validator" in data.get("specialists", {}), ( - "constitutional_validator отсутствует в specialists.yaml" - ) - - def test_role_model_is_sonnet(self): - """constitutional_validator использует модель sonnet.""" - data = _load_yaml() - role = data["specialists"]["constitutional_validator"] - assert role.get("model") == "sonnet", ( - f"Ожидался model=sonnet, получили: {role.get('model')}" - ) - - def test_role_has_gate_true(self): - """constitutional_validator помечен как gate=true.""" - data = _load_yaml() - role = data["specialists"]["constitutional_validator"] - assert role.get("gate") is True, ( - "constitutional_validator должен иметь gate: true" - ) - - def test_role_has_output_schema(self): - """constitutional_validator имеет поле output_schema.""" - data = _load_yaml() - role = data["specialists"]["constitutional_validator"] - assert "output_schema" in role, "output_schema должен быть в constitutional_validator" - - def test_output_schema_contains_verdict_field(self): - """output_schema содержит поле verdict.""" - data = _load_yaml() - schema = data["specialists"]["constitutional_validator"]["output_schema"] - assert "verdict" in schema, "output_schema должен содержать поле verdict" - - def test_output_schema_verdict_has_all_four_verdicts(self): - """output_schema.verdict содержит все 4 вердикта.""" - data = _load_yaml() - verdict_str = str(data["specialists"]["constitutional_validator"]["output_schema"]["verdict"]) - for verdict in ("approved", "changes_required", "escalated", "blocked"): - assert verdict in verdict_str, ( - f"output_schema.verdict должен содержать '{verdict}'" - ) - - def test_role_tools_are_read_only(self): - """constitutional_validator имеет только read-only инструменты (Read, Grep, Glob).""" - data = _load_yaml() - role = data["specialists"]["constitutional_validator"] - tools = role.get("tools", []) - write_tools = {"Write", "Edit", "Bash"} - unexpected = write_tools & set(tools) - assert not unexpected, ( - f"constitutional_validator не должен иметь write-инструменты: {unexpected}" - ) - - -# =========================================================================== -# 2. Маршрут 'feature' -# =========================================================================== - -class TestFeatureRouteConstitutionalValidator: - """Проверяет позицию constitutional_validator в маршруте 'feature'.""" - - def test_feature_route_exists(self): - """Маршрут 'feature' существует в routes.""" - data = _load_yaml() - assert "feature" in data.get("routes", {}), "Маршрут 'feature' не найден" - - def test_feature_route_contains_constitutional_validator(self): - """Маршрут 'feature' содержит шаг constitutional_validator.""" - data = _load_yaml() - steps = data["routes"]["feature"]["steps"] - assert "constitutional_validator" in steps, ( - f"constitutional_validator отсутствует в feature route. Шаги: {steps}" - ) - - def test_feature_route_cv_comes_after_architect(self): - """В маршруте 'feature' constitutional_validator стоит после architect.""" - data = _load_yaml() - steps = data["routes"]["feature"]["steps"] - assert "architect" in steps, "architect отсутствует в feature route" - idx_arch = steps.index("architect") - idx_cv = steps.index("constitutional_validator") - assert idx_cv > idx_arch, ( - f"constitutional_validator (pos={idx_cv}) должен идти ПОСЛЕ architect (pos={idx_arch})" - ) - - def test_feature_route_cv_comes_before_dev_step(self): - """В маршруте 'feature' constitutional_validator стоит перед dev-шагом.""" - data = _load_yaml() - steps = data["routes"]["feature"]["steps"] - idx_cv = steps.index("constitutional_validator") - # Dev-шаги: frontend_dev или backend_dev - dev_roles = {"frontend_dev", "backend_dev"} - dev_positions = [steps.index(r) for r in dev_roles if r in steps] - assert dev_positions, "В feature route должен быть хотя бы один dev-шаг" - first_dev_pos = min(dev_positions) - assert idx_cv < first_dev_pos, ( - f"constitutional_validator (pos={idx_cv}) должен идти ПЕРЕД dev-шагом (pos={first_dev_pos})" - ) - - -# =========================================================================== -# 3. Маршрут 'spec_driven' -# =========================================================================== - -class TestSpecDrivenRouteConstitutionalValidator: - """Проверяет позицию constitutional_validator в маршруте 'spec_driven'.""" - - def test_spec_driven_route_exists(self): - """Маршрут 'spec_driven' существует в routes.""" - data = _load_yaml() - assert "spec_driven" in data.get("routes", {}), "Маршрут 'spec_driven' не найден" - - def test_spec_driven_contains_constitutional_validator(self): - """Маршрут 'spec_driven' содержит шаг constitutional_validator.""" - data = _load_yaml() - steps = data["routes"]["spec_driven"]["steps"] - assert "constitutional_validator" in steps, ( - f"constitutional_validator отсутствует в spec_driven route. Шаги: {steps}" - ) - - def test_spec_driven_cv_comes_after_architect(self): - """В маршруте 'spec_driven' constitutional_validator стоит после architect.""" - data = _load_yaml() - steps = data["routes"]["spec_driven"]["steps"] - assert "architect" in steps, "architect отсутствует в spec_driven route" - idx_arch = steps.index("architect") - idx_cv = steps.index("constitutional_validator") - assert idx_cv > idx_arch, ( - f"constitutional_validator (pos={idx_cv}) должен идти ПОСЛЕ architect (pos={idx_arch})" - ) - - def test_spec_driven_cv_comes_before_task_decomposer(self): - """В маршруте 'spec_driven' constitutional_validator стоит перед task_decomposer.""" - data = _load_yaml() - steps = data["routes"]["spec_driven"]["steps"] - assert "task_decomposer" in steps, "task_decomposer отсутствует в spec_driven route" - idx_cv = steps.index("constitutional_validator") - idx_td = steps.index("task_decomposer") - assert idx_cv < idx_td, ( - f"constitutional_validator (pos={idx_cv}) должен идти ПЕРЕД task_decomposer (pos={idx_td})" - ) - - -# =========================================================================== -# 4. Промпт агента -# =========================================================================== - -class TestConstitutionalValidatorPrompt: - """Проверяет файл agents/prompts/constitutional_validator.md.""" - - def test_prompt_file_exists(self): - """Файл agents/prompts/constitutional_validator.md существует.""" - path = PROMPTS_DIR / "constitutional_validator.md" - assert path.exists(), "constitutional_validator.md не найден в agents/prompts/" - - def test_prompt_contains_approved_verdict(self): - """Промпт содержит вердикт 'approved'.""" - content = (PROMPTS_DIR / "constitutional_validator.md").read_text(encoding="utf-8") - assert "approved" in content, "Промпт не содержит вердикт 'approved'" - - def test_prompt_contains_changes_required_verdict(self): - """Промпт содержит вердикт 'changes_required'.""" - content = (PROMPTS_DIR / "constitutional_validator.md").read_text(encoding="utf-8") - assert "changes_required" in content, "Промпт не содержит вердикт 'changes_required'" - - def test_prompt_contains_escalated_verdict(self): - """Промпт содержит вердикт 'escalated'.""" - content = (PROMPTS_DIR / "constitutional_validator.md").read_text(encoding="utf-8") - assert "escalated" in content, "Промпт не содержит вердикт 'escalated'" - - def test_prompt_contains_blocked_verdict(self): - """Промпт содержит вердикт 'blocked'.""" - content = (PROMPTS_DIR / "constitutional_validator.md").read_text(encoding="utf-8") - assert "blocked" in content, "Промпт не содержит вердикт 'blocked'" - - def test_prompt_has_two_output_sections(self): - """Промпт содержит оба раздела вывода: ## Verdict и ## Details.""" - content = (PROMPTS_DIR / "constitutional_validator.md").read_text(encoding="utf-8") - assert "## Verdict" in content, "Промпт не содержит раздел '## Verdict'" - assert "## Details" in content, "Промпт не содержит раздел '## Details'" - - -# =========================================================================== -# 5. Runner gate-handler -# =========================================================================== - -class TestConstitutionalValidatorGate: - """Тесты gate-обработчика constitutional_validator в runner.py.""" - - @patch("agents.runner.subprocess.run") - def test_changes_required_blocks_pipeline(self, mock_run, conn): - """verdict=changes_required → pipeline останавливается, задача получает статус blocked.""" - cv_output = { - "verdict": "changes_required", - "target_role": "architect", - "violations": [ - { - "principle": "Simplicity over cleverness", - "severity": "high", - "description": "Предложено использование Redis для 50 записей", - "suggestion": "Использовать SQLite", - } - ], - "summary": "Нарушение принципа минимальной сложности.", - } - mock_run.return_value = _mock_success(cv_output) - - steps = [{"role": "constitutional_validator", "model": "sonnet"}] - result = run_pipeline(conn, "P1-001", steps) - - assert result["success"] is False - assert result.get("blocked_by") == "constitutional_validator" - assert "changes_required" in (result.get("blocked_reason") or "") - - task = models.get_task(conn, "P1-001") - assert task["status"] == "blocked" - assert task.get("blocked_agent_role") == "constitutional_validator" - - @patch("agents.runner.subprocess.run") - def test_escalated_blocks_pipeline(self, mock_run, conn): - """verdict=escalated → pipeline останавливается, задача получает статус blocked.""" - cv_output = { - "verdict": "escalated", - "escalation_reason": "Принцип 'no paid APIs' конфликтует с целью 'real-time SMS'", - "violations": [ - { - "principle": "No external paid APIs", - "severity": "critical", - "description": "Предложено Twilio без фолбека", - "suggestion": "Добавить бесплатный альтернативный канал", - } - ], - "summary": "Конфликт принципов требует решения директора.", - } - mock_run.return_value = _mock_success(cv_output) - - steps = [{"role": "constitutional_validator", "model": "sonnet"}] - result = run_pipeline(conn, "P1-001", steps) - - assert result["success"] is False - assert result.get("blocked_by") == "constitutional_validator" - assert "escalated" in (result.get("blocked_reason") or "") - - task = models.get_task(conn, "P1-001") - assert task["status"] == "blocked" - assert task.get("blocked_agent_role") == "constitutional_validator" - - @patch("agents.runner.subprocess.run") - def test_escalated_includes_escalation_reason_in_blocked_reason(self, mock_run, conn): - """verdict=escalated → blocked_reason содержит escalation_reason из вердикта.""" - escalation_reason = "Директор должен решить: платный API или нет" - cv_output = { - "verdict": "escalated", - "escalation_reason": escalation_reason, - "violations": [], - "summary": "Эскалация к директору.", - } - mock_run.return_value = _mock_success(cv_output) - - steps = [{"role": "constitutional_validator", "model": "sonnet"}] - result = run_pipeline(conn, "P1-001", steps) - - assert escalation_reason in (result.get("blocked_reason") or ""), ( - "blocked_reason должен содержать escalation_reason из вердикта" - ) - - @patch("agents.runner.subprocess.run") - def test_approved_continues_pipeline(self, mock_run, conn): - """verdict=approved → pipeline продолжается, задача НЕ блокируется.""" - cv_output = { - "verdict": "approved", - "violations": [], - "summary": "План соответствует принципам. Можно приступать к реализации.", - } - mock_run.return_value = _mock_success(cv_output) - - steps = [{"role": "constitutional_validator", "model": "sonnet"}] - result = run_pipeline(conn, "P1-001", steps) - - assert result.get("blocked_by") != "constitutional_validator" - - task = models.get_task(conn, "P1-001") - assert task["status"] != "blocked" - - @patch("agents.runner.subprocess.run") - def test_changes_required_violations_summary_in_blocked_reason(self, mock_run, conn): - """verdict=changes_required → blocked_reason содержит описание нарушения.""" - cv_output = { - "verdict": "changes_required", - "target_role": "architect", - "violations": [ - { - "principle": "Minimal footprint", - "severity": "critical", - "description": "Добавляется новый сервис без необходимости", - "suggestion": "Обойтись встроенными средствами", - } - ], - "summary": "Критическое нарушение.", - } - mock_run.return_value = _mock_success(cv_output) - - steps = [{"role": "constitutional_validator", "model": "sonnet"}] - result = run_pipeline(conn, "P1-001", steps) - - blocked_reason = result.get("blocked_reason") or "" - assert "Minimal footprint" in blocked_reason or "Добавляется новый сервис" in blocked_reason, ( - f"blocked_reason должен содержать описание нарушения. Получили: {blocked_reason!r}" - ) - - @patch("agents.runner.subprocess.run") - def test_changes_required_two_steps_does_not_execute_second_step(self, mock_run, conn): - """Pipeline с constitutional_validator + frontend_dev: при changes_required второй шаг не выполняется.""" - cv_output = { - "verdict": "changes_required", - "target_role": "architect", - "violations": [{"principle": "X", "severity": "high", "description": "test", "suggestion": "fix"}], - "summary": "Нужна доработка.", - } - mock_run.return_value = _mock_success(cv_output) - - steps = [ - {"role": "constitutional_validator", "model": "sonnet"}, - {"role": "frontend_dev", "model": "sonnet"}, - ] - result = run_pipeline(conn, "P1-001", steps) - - assert result["success"] is False - assert result.get("steps_completed") == 1, ( - f"Должен быть выполнен только 1 шаг, получили: {result.get('steps_completed')}" - ) - # subprocess.run вызывается только один раз — для constitutional_validator - assert mock_run.call_count == 1, ( - f"Ожидался 1 вызов subprocess.run, получили: {mock_run.call_count}" - )