pelmen/kin
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kin: auto-commit after pipeline

Browse source
This commit is contained in:
Gros Frumos 2026-03-19 20:35:50 +02:00
parent 39b6f38414
commit 5750b72e8b
1 changed files with 469 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

469
tests/test_kin_docs_004_regression.py Normal file
View file

@ -0,0 +1,469 @@
"""Regression tests for KIN-DOCS-004 — context_packet как артефакт handoff.
Acceptance criteria:
1. Структурный тест agents/prompts/department_head.md: context_packet присутствует
в ## Return Format с 5 обязательными полями, упомянут в ## Focus On и ## Quality Checks.
2. Структурный тест agents/specialists.yaml: все 7 dept head ролей имеют
output_schema.context_packet.
3. Структурные тесты промптов-получателей (architect.md, backend_dev.md, frontend_dev.md):
явная инструкция читать context_packet первым (шаг 0 в ## Working Mode).
4. Unit-тест models.create_handoff: context_packet сохраняется как JSON и декодируется обратно.
5. Runner-тесты _execute_department_head_step: context_packet включён в summary output,
размещён первым ключом в initial context для воркеров; fail-open при отсутствии.
"""
import json
from pathlib import Path
from unittest.mock import patch, MagicMock, call
import pytest
import yaml
from core.db import init_db
from core import models
PROMPTS_DIR = Path(__file__).parent.parent / "agents" / "prompts"
SPECIALISTS_YAML = Path(__file__).parent.parent / "agents" / "specialists.yaml"
DEPT_HEAD_ROLES = [
"backend_head",
"frontend_head",
"qa_head",
"security_head",
"infra_head",
"research_head",
"marketing_head",
]
CONTEXT_PACKET_FIELDS = [
"architecture_notes",
"key_files",
"constraints",
"unknowns",
"handoff_for",
]
def _load_yaml():
return yaml.safe_load(SPECIALISTS_YAML.read_text(encoding="utf-8"))
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def conn():
c = init_db(":memory:")
models.create_project(c, "proj", "TestProject", "~/projects/test",
tech_stack=["python"])
models.create_task(c, "PROJ-001", "proj", "Test task",
brief={"route_type": "dept_feature"})
yield c
c.close()
def _success_sub_result(pipeline_id=99):
return {
"success": True,
"total_cost_usd": 0.01,
"total_tokens": 100,
"total_duration_seconds": 2.0,
"steps_completed": 1,
"results": [{"role": "backend_dev", "output": "done"}],
"pipeline_id": pipeline_id,
}
def _dept_head_raw_output(with_context_packet=True, extra_fields=None):
"""Сборка raw_output для dept head, опционально с context_packet."""
out = {
"status": "done",
"sub_pipeline": [
{"role": "backend_dev", "model": "sonnet", "brief": "Implement feature"},
],
"artifacts": {"files_changed": ["core/models.py"], "notes": "done"},
"handoff_notes": "Backend complete",
}
if with_context_packet:
out["context_packet"] = {
"architecture_notes": "Pure functions, raw sqlite3",
"key_files": ["core/models.py", "core/db.py"],
"constraints": ["All DB columns need DEFAULT values"],
"unknowns": ["Frontend integration pending"],
"handoff_for": "backend_dev",
}
if extra_fields:
out.update(extra_fields)
return {"raw_output": json.dumps(out)}
# ===========================================================================
# 1. Структурные тесты agents/prompts/department_head.md
# ===========================================================================
class TestDepartmentHeadPromptStructure:
"""department_head.md содержит context_packet во всех нужных секциях."""
def _content(self):
return (PROMPTS_DIR / "department_head.md").read_text(encoding="utf-8")
def test_context_packet_in_return_format(self):
"""department_head.md содержит 'context_packet' в секции ## Return Format."""
content = self._content()
# Найти секцию Return Format и убедиться, что context_packet там есть
return_format_idx = content.find("## Return Format")
assert return_format_idx != -1, "department_head.md не содержит секцию '## Return Format'"
return_format_section = content[return_format_idx:]
assert "context_packet" in return_format_section, (
"department_head.md не содержит 'context_packet' в ## Return Format"
)
@pytest.mark.parametrize("field", CONTEXT_PACKET_FIELDS)
def test_context_packet_required_field_present(self, field):
"""department_head.md упоминает каждое из 5 обязательных полей context_packet."""
content = self._content()
assert field in content, (
f"department_head.md не содержит обязательного поля context_packet: {field!r}"
)
def test_context_packet_in_focus_on(self):
"""department_head.md упоминает context_packet в секции ## Focus On."""
content = self._content()
focus_on_idx = content.find("## Focus On")
assert focus_on_idx != -1, "department_head.md не содержит секцию '## Focus On'"
focus_on_section = content[focus_on_idx:]
quality_checks_idx = focus_on_section.find("## Quality Checks")
focus_on_body = focus_on_section[:quality_checks_idx] if quality_checks_idx != -1 else focus_on_section
assert "context_packet" in focus_on_body, (
"department_head.md не упоминает 'context_packet' в ## Focus On"
)
def test_context_packet_in_quality_checks(self):
"""department_head.md валидирует context_packet в ## Quality Checks."""
content = self._content()
qc_idx = content.find("## Quality Checks")
assert qc_idx != -1, "department_head.md не содержит секцию '## Quality Checks'"
qc_section = content[qc_idx:]
assert "context_packet" in qc_section, (
"department_head.md не упоминает 'context_packet' в ## Quality Checks"
)
# ===========================================================================
# 2. Структурный тест agents/specialists.yaml — 7 dept head ролей
# ===========================================================================
class TestSpecialistsYamlContextPacket:
"""Все 7 dept head ролей имеют output_schema.context_packet в specialists.yaml."""
@pytest.mark.parametrize("role", DEPT_HEAD_ROLES)
def test_dept_head_has_output_schema_context_packet(self, role):
"""specialists.yaml: роль {role} имеет output_schema.context_packet."""
data = _load_yaml()
specialists = data.get("specialists", {})
assert role in specialists, (
f"Роль {role!r} не найдена в specialists.yaml"
)
output_schema = specialists[role].get("output_schema", {})
assert "context_packet" in output_schema, (
f"specialists.yaml: {role} не имеет output_schema.context_packet"
)
def test_exactly_7_dept_heads_have_context_packet(self):
"""Ровно 7 dept head ролей имеют output_schema.context_packet."""
data = _load_yaml()
specialists = data.get("specialists", {})
roles_with_cp = [
role for role, spec in specialists.items()
if "context_packet" in spec.get("output_schema", {})
]
assert len(roles_with_cp) == 7, (
f"Ожидалось 7 ролей с output_schema.context_packet, найдено {len(roles_with_cp)}: {roles_with_cp}"
)
# ===========================================================================
# 3. Структурные тесты промптов-получателей context_packet
# ===========================================================================
class TestWorkerPromptsContextPacketInstruction:
"""architect.md, backend_dev.md, frontend_dev.md содержат инструкцию читать context_packet первым."""
@pytest.mark.parametrize("prompt_file", ["architect.md", "backend_dev.md", "frontend_dev.md"])
def test_worker_prompt_has_context_packet_read_first_instruction(self, prompt_file):
"""Промпт {prompt_file} содержит явную инструкцию читать context_packet первым."""
content = (PROMPTS_DIR / prompt_file).read_text(encoding="utf-8")
assert "context_packet" in content, (
f"{prompt_file} не содержит инструкцию про context_packet"
)
@pytest.mark.parametrize("prompt_file", ["architect.md", "backend_dev.md", "frontend_dev.md"])
def test_worker_prompt_context_packet_instruction_in_working_mode(self, prompt_file):
"""Инструкция про context_packet находится в секции ## Working Mode промпта."""
content = (PROMPTS_DIR / prompt_file).read_text(encoding="utf-8")
wm_idx = content.find("## Working Mode")
assert wm_idx != -1, f"{prompt_file} не содержит секцию '## Working Mode'"
wm_section = content[wm_idx:]
# Ограничиваем до следующей секции
next_section_idx = wm_section.find("\n## ", 1)
wm_body = wm_section[:next_section_idx] if next_section_idx != -1 else wm_section
assert "context_packet" in wm_body, (
f"{prompt_file}: инструкция читать context_packet отсутствует в ## Working Mode"
)
@pytest.mark.parametrize("prompt_file", ["architect.md", "backend_dev.md", "frontend_dev.md"])
def test_worker_prompt_context_packet_instruction_is_step_0(self, prompt_file):
"""Инструкция про context_packet пронумерована как шаг 0 в ## Working Mode."""
content = (PROMPTS_DIR / prompt_file).read_text(encoding="utf-8")
wm_idx = content.find("## Working Mode")
wm_section = content[wm_idx:]
next_section_idx = wm_section.find("\n## ", 1)
wm_body = wm_section[:next_section_idx] if next_section_idx != -1 else wm_section
# Шаг 0 должен содержать context_packet
assert "0." in wm_body and "context_packet" in wm_body, (
f"{prompt_file}: шаг 0 с инструкцией context_packet не найден в ## Working Mode"
)
# ===========================================================================
# 4. Unit-тесты models.create_handoff с context_packet
# ===========================================================================
class TestCreateHandoffContextPacket:
"""Unit-тесты models.create_handoff — сохранение и чтение context_packet."""
def test_create_handoff_stores_context_packet_dict(self, conn):
"""create_handoff с context_packet dict: возвращает его как dict."""
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
cp = {
"architecture_notes": "Pure functions",
"key_files": ["core/models.py"],
"constraints": ["No ORM"],
"unknowns": [],
"handoff_for": "backend_dev",
}
handoff = models.create_handoff(
conn, pipeline["id"], "PROJ-001",
from_department="backend",
context_packet=cp,
)
assert handoff["context_packet"] == cp, (
"create_handoff не вернул context_packet как dict"
)
def test_create_handoff_context_packet_none_returns_none(self, conn):
"""create_handoff без context_packet: поле возвращается как None."""
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
handoff = models.create_handoff(
conn, pipeline["id"], "PROJ-001",
from_department="backend",
context_packet=None,
)
assert handoff["context_packet"] is None, (
"create_handoff без context_packet должен возвращать None, не что-то другое"
)
def test_create_handoff_context_packet_persisted_in_db(self, conn):
"""create_handoff с context_packet: поле сохраняется в БД и декодируется при SELECT."""
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
cp = {"architecture_notes": "test", "key_files": ["a.py"], "constraints": [],
"unknowns": ["q1"], "handoff_for": "backend_dev"}
handoff = models.create_handoff(
conn, pipeline["id"], "PROJ-001",
from_department="backend",
context_packet=cp,
)
# Перечитать из БД через get_handoffs_for_task
handoffs = models.get_handoffs_for_task(conn, "PROJ-001")
assert len(handoffs) == 1
assert handoffs[0]["context_packet"] == cp, (
"context_packet не совпадает после повторного чтения из БД"
)
def test_context_packet_in_json_columns(self):
"""'context_packet' присутствует в models._JSON_COLUMNS."""
assert "context_packet" in models._JSON_COLUMNS, (
"'context_packet' должен быть в models._JSON_COLUMNS для автодекодирования"
)
# ===========================================================================
# 5. Runner-тесты _execute_department_head_step
# ===========================================================================
class TestExecuteDeptHeadStepContextPacket:
"""Тесты _execute_department_head_step: context_packet в output и initial context."""
def test_context_packet_present_in_summary_output(self, conn):
"""Если dept head вернул context_packet — он включён в summary output для следующего шага."""
from agents.runner import _execute_department_head_step
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
with patch("agents.runner.run_pipeline", return_value=_success_sub_result(pipeline["id"])), \
patch("agents.runner.models.create_handoff"):
result = _execute_department_head_step(
conn=conn,
task_id="PROJ-001",
project_id="proj",
parent_pipeline_id=pipeline["id"],
step={"role": "backend_head", "brief": "plan"},
dept_head_result=_dept_head_raw_output(with_context_packet=True),
next_department="frontend",
)
assert result["success"] is True
summary = json.loads(result["output"])
assert "context_packet" in summary, (
"context_packet должен быть в summary output для следующего dept head"
)
cp = summary["context_packet"]
assert cp["handoff_for"] == "backend_dev"
assert "core/models.py" in cp["key_files"]
def test_context_packet_placed_first_in_worker_initial_context(self, conn):
"""Если context_packet есть — он первый ключ в initial_previous_output для воркеров."""
from agents.runner import _execute_department_head_step
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
captured_initial_output = {}
def capture_run_pipeline(conn, task_id, steps, *args, **kwargs):
initial = kwargs.get("initial_previous_output", "{}")
captured_initial_output["value"] = json.loads(initial)
return _success_sub_result(pipeline["id"])
with patch("agents.runner.run_pipeline", side_effect=capture_run_pipeline), \
patch("agents.runner.models.create_handoff"):
_execute_department_head_step(
conn=conn,
task_id="PROJ-001",
project_id="proj",
parent_pipeline_id=pipeline["id"],
step={"role": "backend_head", "brief": "plan"},
dept_head_result=_dept_head_raw_output(with_context_packet=True),
)
ctx = captured_initial_output.get("value", {})
assert "context_packet" in ctx, (
"context_packet отсутствует в initial_previous_output для воркеров"
)
keys = list(ctx.keys())
assert keys[0] == "context_packet", (
f"context_packet должен быть первым ключом в initial context, но первый: {keys[0]!r}"
)
def test_missing_context_packet_does_not_block_pipeline(self, conn):
"""Если context_packet отсутствует — pipeline продолжается (fail-open)."""
from agents.runner import _execute_department_head_step
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
with patch("agents.runner.run_pipeline", return_value=_success_sub_result(pipeline["id"])), \
patch("agents.runner.models.create_handoff"):
result = _execute_department_head_step(
conn=conn,
task_id="PROJ-001",
project_id="proj",
parent_pipeline_id=pipeline["id"],
step={"role": "backend_head", "brief": "plan"},
dept_head_result=_dept_head_raw_output(with_context_packet=False),
next_department="frontend",
)
assert result["success"] is True, (
"Pipeline должен продолжаться при отсутствии context_packet (fail-open)"
)
def test_missing_context_packet_not_in_worker_initial_context(self, conn):
"""Если context_packet отсутствует — его ключа нет в initial context для воркеров."""
from agents.runner import _execute_department_head_step
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
captured = {}
def capture_run_pipeline(conn, task_id, steps, *args, **kwargs):
initial = kwargs.get("initial_previous_output", "{}")
captured["value"] = json.loads(initial)
return _success_sub_result(pipeline["id"])
with patch("agents.runner.run_pipeline", side_effect=capture_run_pipeline), \
patch("agents.runner.models.create_handoff"):
_execute_department_head_step(
conn=conn,
task_id="PROJ-001",
project_id="proj",
parent_pipeline_id=pipeline["id"],
step={"role": "backend_head", "brief": "plan"},
dept_head_result=_dept_head_raw_output(with_context_packet=False),
)
ctx = captured.get("value", {})
assert "context_packet" not in ctx, (
"context_packet не должен быть в initial context, если dept head его не вернул"
)
# Но department_head_plan должен всё равно присутствовать
assert "department_head_plan" in ctx, (
"department_head_plan должен присутствовать в initial context даже без context_packet"
)
def test_context_packet_passed_to_create_handoff(self, conn):
"""context_packet передаётся в models.create_handoff при сохранении хэндоффа."""
from agents.runner import _execute_department_head_step
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
with patch("agents.runner.run_pipeline", return_value=_success_sub_result(pipeline["id"])), \
patch("agents.runner.models.create_handoff") as mock_handoff:
_execute_department_head_step(
conn=conn,
task_id="PROJ-001",
project_id="proj",
parent_pipeline_id=pipeline["id"],
step={"role": "backend_head", "brief": "plan"},
dept_head_result=_dept_head_raw_output(with_context_packet=True),
)
assert mock_handoff.called, "models.create_handoff должен быть вызван"
_, kwargs = mock_handoff.call_args
cp = kwargs.get("context_packet")
assert cp is not None, "context_packet должен быть передан в create_handoff"
assert cp["handoff_for"] == "backend_dev"
def test_missing_context_packet_handoff_receives_none(self, conn):
"""Если context_packet отсутствует — create_handoff вызывается с context_packet=None."""
from agents.runner import _execute_department_head_step
pipeline = models.create_pipeline(
conn, "PROJ-001", "proj", "dept_feature", [{"role": "backend_head"}]
)
with patch("agents.runner.run_pipeline", return_value=_success_sub_result(pipeline["id"])), \
patch("agents.runner.models.create_handoff") as mock_handoff:
_execute_department_head_step(
conn=conn,
task_id="PROJ-001",
project_id="proj",
parent_pipeline_id=pipeline["id"],
step={"role": "backend_head", "brief": "plan"},
dept_head_result=_dept_head_raw_output(with_context_packet=False),
)
assert mock_handoff.called
_, kwargs = mock_handoff.call_args
assert kwargs.get("context_packet") is None, (
"create_handoff должен получить context_packet=None, если dept head не вернул его"
)