"""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 не вернул его" )