kin: auto-commit after pipeline

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": "<clear explanation>", "blocked_at": "<ISO-8601 datetime>"}
+```
+
+Use current datetime for `blocked_at`. Do NOT guess or partially validate — return blocked immediately.

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:

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]

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}"
+        )