kin: auto-commit after pipeline
This commit is contained in:
Gros Frumos
parent
d0d72cb647
commit
1600d0d471
2 changed files with 403 additions and 2 deletions
401
tests/test_kin_docs_001_regression.py
Normal file
401
tests/test_kin_docs_001_regression.py
Normal file
|
|
@ -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}"
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue