469 lines
22 KiB
Python
469 lines
22 KiB
Python
"""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 не вернул его"
|
||
)
|