"""Regression tests for KIN-DOCS-003 — knowledge_synthesizer role. Acceptance criteria: 1. Структурный тест промпта knowledge_synthesizer: 5 обязательных секций + специфические поля (#940) 2. Тест регистрации роли в specialists.yaml (model, tools, permissions, output_schema) 3. Тест корректности pipeline research_head с aggregation шагом: - build_phase_order с ≥2 исследователями вставляет knowledge_synthesizer перед architect - create_project_with_phases НЕ создаёт knowledge_synthesizer как фазу (авто-управляемая роль) 4. Тест context_builder: knowledge_synthesizer получает decisions, но не modules 5. Тест activate_phase: при ≥2 одобренных фазах инжектирует phases_context в brief 6. Проверка exclusion-листов (#921, #917): knowledge_synthesizer.md не исключён из проверок структуры """ from pathlib import Path from unittest.mock import patch import pytest import yaml from core.db import init_db from core import models from core.phases import build_phase_order, create_project_with_phases, activate_phase from core.context_builder import build_context SPECIALISTS_YAML = Path(__file__).parent.parent / "agents" / "specialists.yaml" PROMPTS_DIR = Path(__file__).parent.parent / "agents" / "prompts" REQUIRED_SECTIONS = [ "## Working Mode", "## Focus On", "## Quality Checks", "## Return Format", "## Constraints", ] OUTPUT_SCHEMA_FIELDS = [ "unified_findings", "confidence_rated_conclusions", "unresolved_conflicts", "prioritized_actions", "phases_context_used", ] def _load_yaml(): return yaml.safe_load(SPECIALISTS_YAML.read_text(encoding="utf-8")) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def conn(): c = init_db(":memory:") yield c c.close() @pytest.fixture def conn_with_project(conn): models.create_project(conn, "p1", "TestProject", "/p1", tech_stack=["python"]) models.create_task(conn, "P1-001", "p1", "Research task") yield conn # =========================================================================== # 1. Структурный тест промпта (#940) # =========================================================================== class TestKnowledgeSynthesizerPrompt: """Структурный тест agents/prompts/knowledge_synthesizer.md.""" def test_prompt_file_exists(self): """Файл agents/prompts/knowledge_synthesizer.md существует.""" path = PROMPTS_DIR / "knowledge_synthesizer.md" assert path.exists(), "knowledge_synthesizer.md не найден в agents/prompts/" @pytest.mark.parametrize("section", REQUIRED_SECTIONS) def test_prompt_has_required_section(self, section): """Промпт содержит все 5 обязательных секций (REQUIRED_SECTIONS).""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert section in content, ( f"knowledge_synthesizer.md не содержит обязательную секцию {section!r}" ) def test_prompt_has_input_section(self): """Промпт содержит секцию ## Input — агент-специфичная секция.""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert "## Input" in content, ( "knowledge_synthesizer.md не содержит секцию '## Input'" ) def test_prompt_contains_confidence_rated_conclusions(self): """Промпт содержит поле confidence_rated_conclusions в Return Format.""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert "confidence_rated_conclusions" in content, ( "knowledge_synthesizer.md не содержит 'confidence_rated_conclusions'" ) def test_prompt_contains_phases_context_used(self): """Промпт содержит поле phases_context_used в Return Format.""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert "phases_context_used" in content, ( "knowledge_synthesizer.md не содержит 'phases_context_used'" ) def test_prompt_contains_unresolved_conflicts(self): """Промпт содержит поле unresolved_conflicts в Return Format.""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert "unresolved_conflicts" in content, ( "knowledge_synthesizer.md не содержит 'unresolved_conflicts'" ) def test_prompt_contains_blocked_protocol(self): """Промпт содержит Blocked Protocol с инструкцией blocked_reason.""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert "blocked_reason" in content, ( "knowledge_synthesizer.md не содержит 'blocked_reason' — Blocked Protocol обязателен" ) def test_prompt_no_legacy_output_format_header(self): """Промпт НЕ содержит устаревшей секции '## Output format'.""" content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8") assert "## Output format" not in content, ( "knowledge_synthesizer.md содержит устаревшую секцию '## Output format'" ) # =========================================================================== # 2. Регистрация в specialists.yaml # =========================================================================== class TestKnowledgeSynthesizerSpecialists: """Тесты регистрации knowledge_synthesizer в agents/specialists.yaml.""" def test_role_exists_in_specialists(self): """specialists.yaml содержит роль knowledge_synthesizer.""" data = _load_yaml() assert "knowledge_synthesizer" in data.get("specialists", {}), ( "knowledge_synthesizer отсутствует в specialists.yaml" ) def test_role_model_is_sonnet(self): """knowledge_synthesizer использует модель sonnet.""" data = _load_yaml() role = data["specialists"]["knowledge_synthesizer"] assert role.get("model") == "sonnet", ( f"Ожидался model=sonnet, получили: {role.get('model')}" ) def test_role_tools_include_read_grep_glob(self): """knowledge_synthesizer имеет инструменты Read, Grep, Glob.""" data = _load_yaml() tools = data["specialists"]["knowledge_synthesizer"].get("tools", []) for required_tool in ("Read", "Grep", "Glob"): assert required_tool in tools, ( f"knowledge_synthesizer должен иметь инструмент {required_tool!r}" ) def test_role_has_no_write_tools(self): """knowledge_synthesizer НЕ имеет write-инструментов (Read-only роль).""" data = _load_yaml() tools = set(data["specialists"]["knowledge_synthesizer"].get("tools", [])) write_tools = {"Write", "Edit", "Bash"} unexpected = write_tools & tools assert not unexpected, ( f"knowledge_synthesizer не должен иметь write-инструменты: {unexpected}" ) def test_role_permissions_is_read_only(self): """knowledge_synthesizer имеет permissions=read_only.""" data = _load_yaml() role = data["specialists"]["knowledge_synthesizer"] assert role.get("permissions") == "read_only", ( f"Ожидался permissions=read_only, получили: {role.get('permissions')}" ) def test_role_context_rules_decisions_all(self): """knowledge_synthesizer получает все decisions (context_rules.decisions=all).""" data = _load_yaml() role = data["specialists"]["knowledge_synthesizer"] decisions = role.get("context_rules", {}).get("decisions") assert decisions == "all", ( f"Ожидался context_rules.decisions=all, получили: {decisions}" ) def test_role_has_output_schema(self): """knowledge_synthesizer имеет поле output_schema.""" data = _load_yaml() role = data["specialists"]["knowledge_synthesizer"] assert "output_schema" in role, ( "knowledge_synthesizer должен иметь output_schema" ) @pytest.mark.parametrize("field", OUTPUT_SCHEMA_FIELDS) def test_output_schema_has_required_field(self, field): """output_schema содержит каждое из 5 обязательных полей.""" data = _load_yaml() schema = data["specialists"]["knowledge_synthesizer"]["output_schema"] assert field in schema, ( f"output_schema knowledge_synthesizer не содержит обязательного поля {field!r}" ) # =========================================================================== # 3. Корректность pipeline с aggregation шагом # =========================================================================== class TestResearchPipelineWithSynthesizer: """Тесты корректности research pipeline с knowledge_synthesizer aggregation шагом.""" def test_build_phase_order_two_researchers_includes_synthesizer(self): """build_phase_order с 2 исследователями вставляет knowledge_synthesizer перед architect.""" result = build_phase_order(["business_analyst", "market_researcher"]) assert "knowledge_synthesizer" in result, ( "knowledge_synthesizer должен быть в pipeline при ≥2 исследователях" ) def test_build_phase_order_synthesizer_before_architect(self): """knowledge_synthesizer всегда стоит непосредственно перед architect.""" result = build_phase_order(["tech_researcher", "ux_designer"]) idx_syn = result.index("knowledge_synthesizer") idx_arch = result.index("architect") assert idx_syn == idx_arch - 1, ( f"knowledge_synthesizer (pos={idx_syn}) должен идти прямо перед architect (pos={idx_arch})" ) def test_build_phase_order_one_researcher_no_synthesizer(self): """build_phase_order с 1 исследователем НЕ вставляет knowledge_synthesizer.""" result = build_phase_order(["business_analyst"]) assert "knowledge_synthesizer" not in result, ( "knowledge_synthesizer не должен вставляться при 1 исследователе" ) def test_create_project_two_researchers_excludes_synthesizer_from_phases(self, conn): """create_project_with_phases НЕ создаёт фазу knowledge_synthesizer — авто-управляемая роль.""" result = create_project_with_phases( conn, "p1", "P1", "/p1", description="Test", selected_roles=["market_researcher", "tech_researcher"], ) phase_roles = [ph["role"] for ph in result["phases"]] assert "knowledge_synthesizer" not in phase_roles, ( f"knowledge_synthesizer не должен быть в pipeline-фазах. Фазы: {phase_roles}" ) def test_create_project_two_researchers_phase_count(self, conn): """create_project_with_phases с 2 исследователями создаёт 3 фазы (без synthesizer).""" result = create_project_with_phases( conn, "p1", "P1", "/p1", description="Test", selected_roles=["market_researcher", "tech_researcher"], ) assert len(result["phases"]) == 3, ( f"Ожидалось 3 фазы (market_researcher, tech_researcher, architect), " f"получили {len(result['phases'])}: {[ph['role'] for ph in result['phases']]}" ) def test_create_project_two_researchers_architect_last(self, conn): """create_project_with_phases с 2 исследователями ставит architect последним.""" result = create_project_with_phases( conn, "p1", "P1", "/p1", description="Test", selected_roles=["market_researcher", "tech_researcher"], ) assert result["phases"][-1]["role"] == "architect" # =========================================================================== # 4. context_builder — knowledge_synthesizer # =========================================================================== class TestKnowledgeSynthesizerContextBuilder: """Тесты context_builder для роли knowledge_synthesizer.""" def test_knowledge_synthesizer_gets_decisions(self, conn_with_project): """build_context для knowledge_synthesizer включает decisions.""" models.add_decision( conn_with_project, "p1", "gotcha", "Some decision", "Details" ) ctx = build_context(conn_with_project, "P1-001", "knowledge_synthesizer", "p1") assert "decisions" in ctx, ( "knowledge_synthesizer должен получать decisions в контексте" ) def test_knowledge_synthesizer_does_not_get_modules(self, conn_with_project): """build_context для knowledge_synthesizer НЕ включает modules (в отличие от architect).""" models.add_module(conn_with_project, "p1", "api", "backend", "src/api/") ctx = build_context(conn_with_project, "P1-001", "knowledge_synthesizer", "p1") assert "modules" not in ctx, ( "knowledge_synthesizer не должен получать modules — только decisions" ) def test_architect_gets_both_decisions_and_modules(self, conn_with_project): """Регрессия: architect по-прежнему получает decisions И modules.""" models.add_module(conn_with_project, "p1", "api", "backend", "src/api/") models.add_decision( conn_with_project, "p1", "convention", "Use WAL", "SQLite WAL" ) ctx = build_context(conn_with_project, "P1-001", "architect", "p1") assert "decisions" in ctx assert "modules" in ctx # =========================================================================== # 5. activate_phase — phases_context injection # =========================================================================== class TestActivatePhaseKnowledgeSynthesizer: """Тесты activate_phase для роли knowledge_synthesizer.""" def test_activate_phase_synthesizer_no_approved_phases_no_context(self, conn): """activate_phase для knowledge_synthesizer без одобренных фаз не добавляет phases_context.""" models.create_project(conn, "p1", "P1", "/p1", description="Test") phase = models.create_phase(conn, "p1", "knowledge_synthesizer", 0) result = activate_phase(conn, phase["id"]) task = models.get_task(conn, result["task_id"]) brief = task["brief"] assert "phases_context" not in brief, ( "phases_context не должен быть в brief, если нет одобренных фаз" ) def test_activate_phase_synthesizer_with_approved_phases_injects_context(self, conn): """activate_phase для knowledge_synthesizer собирает phases_context из одобренных фаз.""" models.create_project(conn, "p1", "P1", "/p1", description="Test") # Создать и одобрить исследовательскую фазу researcher_phase = models.create_phase(conn, "p1", "business_analyst", 0) researcher_phase = activate_phase(conn, researcher_phase["id"]) researcher_task_id = researcher_phase["task_id"] # Записать лог агента с output_summary для задачи исследователя conn.execute( """INSERT INTO agent_logs (project_id, task_id, agent_role, action, output_summary, success) VALUES (?, ?, ?, ?, ?, ?)""", ("p1", researcher_task_id, "business_analyst", "run", "Market shows strong demand for product.", 1), ) conn.commit() # Одобрить исследовательскую фазу models.update_phase(conn, researcher_phase["id"], status="approved") # Создать и активировать фазу knowledge_synthesizer synth_phase = models.create_phase(conn, "p1", "knowledge_synthesizer", 1) result = activate_phase(conn, synth_phase["id"]) task = models.get_task(conn, result["task_id"]) brief = task["brief"] assert "phases_context" in brief, ( "phases_context должен быть в brief при наличии одобренных фаз с agent_logs" ) assert "business_analyst" in brief["phases_context"], ( "phases_context должен содержать ключ 'business_analyst'" ) assert "Market shows strong demand" in brief["phases_context"]["business_analyst"], ( "phases_context['business_analyst'] должен содержать output_summary из agent_logs" ) def test_activate_phase_synthesizer_ignores_pending_phases(self, conn): """activate_phase для knowledge_synthesizer не включает в phases_context неодобренные фазы.""" models.create_project(conn, "p1", "P1", "/p1", description="Test") # Создать исследовательскую фазу в статусе active (не approved) researcher_phase = models.create_phase(conn, "p1", "tech_researcher", 0) researcher_phase = activate_phase(conn, researcher_phase["id"]) researcher_task_id = researcher_phase["task_id"] # Записать лог агента conn.execute( """INSERT INTO agent_logs (project_id, task_id, agent_role, action, output_summary, success) VALUES (?, ?, ?, ?, ?, ?)""", ("p1", researcher_task_id, "tech_researcher", "run", "Tech analysis done.", 1), ) conn.commit() # Фаза остаётся в статусе active (не approved) # Активировать knowledge_synthesizer synth_phase = models.create_phase(conn, "p1", "knowledge_synthesizer", 1) result = activate_phase(conn, synth_phase["id"]) task = models.get_task(conn, result["task_id"]) brief = task["brief"] # phases_context должен отсутствовать или быть пустым phases_context = brief.get("phases_context", {}) assert "tech_researcher" not in phases_context, ( "phases_context не должен включать фазы без статуса approved" ) # =========================================================================== # 6. Exclusion list guard (#921, #917) # =========================================================================== class TestExclusionListRegression: """Верифицирует, что knowledge_synthesizer.md не добавлен в exclusion list (#921, #917).""" def test_knowledge_synthesizer_not_in_exclusion_list(self): """knowledge_synthesizer.md не включён в EXCLUDED_FROM_STRUCTURE_CHECK.""" from tests.test_kin_docs_002_regression import EXCLUDED_FROM_STRUCTURE_CHECK assert "knowledge_synthesizer.md" not in EXCLUDED_FROM_STRUCTURE_CHECK, ( "knowledge_synthesizer.md не должен быть в EXCLUDED_FROM_STRUCTURE_CHECK — " "роль должна проходить все стандартные структурные проверки" ) def test_exclusion_list_is_empty(self): """EXCLUDED_FROM_STRUCTURE_CHECK остаётся пустым — guard от #929.""" from tests.test_kin_docs_002_regression import EXCLUDED_FROM_STRUCTURE_CHECK assert EXCLUDED_FROM_STRUCTURE_CHECK == [], ( f"EXCLUDED_FROM_STRUCTURE_CHECK должен быть пустым, " f"но содержит: {EXCLUDED_FROM_STRUCTURE_CHECK}" )