From 7edc66201c968386f169db91e8c7608d34772a01 Mon Sep 17 00:00:00 2001 From: Gros Frumos Date: Thu, 19 Mar 2026 13:47:49 +0200 Subject: [PATCH 1/2] kin: KIN-DOCS-001-backend_dev --- agents/prompts/constitutional_validator.md | 158 +++++++++++++++++++++ agents/runner.py | 67 +++++++++ agents/specialists.yaml | 24 +++- 3 files changed, 245 insertions(+), 4 deletions(-) create mode 100644 agents/prompts/constitutional_validator.md diff --git a/agents/prompts/constitutional_validator.md b/agents/prompts/constitutional_validator.md new file mode 100644 index 0000000..599044c --- /dev/null +++ b/agents/prompts/constitutional_validator.md @@ -0,0 +1,158 @@ +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 471b683..61bebe6 100644 --- a/agents/runner.py +++ b/agents/runner.py @@ -1966,6 +1966,73 @@ 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 453361b..6056448 100644 --- a/agents/specialists.yaml +++ b/agents/specialists.yaml @@ -139,6 +139,22 @@ 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 @@ -278,8 +294,8 @@ routes: description: "Find bug → verify → fix → verify fix" feature: - steps: [architect, frontend_dev, tester, reviewer] - description: "Design → implement → test → review" + steps: [architect, constitutional_validator, frontend_dev, tester, reviewer] + description: "Design → validate → implement → test → review" refactor: steps: [architect, frontend_dev, tester, reviewer] @@ -306,8 +322,8 @@ routes: description: "SSH diagnose → find root cause → verify fix plan" spec_driven: - steps: [constitution, spec, architect, task_decomposer] - description: "Constitution → spec → implementation plan → decompose into tasks" + steps: [constitution, spec, architect, constitutional_validator, task_decomposer] + description: "Constitution → spec → implementation plan → validate → decompose into tasks" dept_feature: steps: [backend_head, frontend_head, qa_head] From 1600d0d471d38012f62d4d6d42bbea75734ce89b Mon Sep 17 00:00:00 2001 From: Gros Frumos Date: Thu, 19 Mar 2026 13:52:36 +0200 Subject: [PATCH 2/2] kin: auto-commit after pipeline --- agents/specialists.yaml | 4 +- tests/test_kin_docs_001_regression.py | 401 ++++++++++++++++++++++++++ 2 files changed, 403 insertions(+), 2 deletions(-) create mode 100644 tests/test_kin_docs_001_regression.py diff --git a/agents/specialists.yaml b/agents/specialists.yaml index 6056448..1d2b1e1 100644 --- a/agents/specialists.yaml +++ b/agents/specialists.yaml @@ -322,8 +322,8 @@ routes: description: "SSH diagnose → find root cause → verify fix plan" spec_driven: - steps: [constitution, spec, architect, constitutional_validator, task_decomposer] - description: "Constitution → spec → implementation plan → validate → decompose into tasks" + steps: [constitution, spec, architect, task_decomposer] + description: "Constitution → spec → implementation plan → decompose into tasks" dept_feature: steps: [backend_head, frontend_head, qa_head] diff --git a/tests/test_kin_docs_001_regression.py b/tests/test_kin_docs_001_regression.py new file mode 100644 index 0000000..3a46265 --- /dev/null +++ b/tests/test_kin_docs_001_regression.py @@ -0,0 +1,401 @@ +"""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}" + )