from __future__ import annotations import json import sys from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts.lib.model_config import resolve_model_profile from scripts.runtime.roles import resolve_runtime_profile from scripts.runtime.methods import ResearchMethodRegistry from scripts.runtime.phase1 import build_chapter_planning from scripts.runtime.skills import SkillRegistry from scripts.runtime.tasks import ( TaskCard, detect_dependency_cycles, generate_task_cards, generate_task_cards_from_research_brief, validate_packet, validate_task_cards, ) def test_skill_registry_uses_agents_skills_as_canonical() -> None: registry = SkillRegistry() names = registry.list_names() assert "search-strategy" in names assert "search-gateway" in names assert "source-quality" in names assert "document-ingest" in names assert "deep-research" in names assert "antigravity-surface-adapter" in names assert "method-selection" in names assert "research-quality-gates" in names assert registry.validate()["ok"] is True def test_model_profile_exposes_task_types_and_role_defaults() -> None: resolved = resolve_model_profile(profile="medium") assert resolved["roles"]["dr_pm"] assert resolved["task_types"]["source_discovery"] == "dr_searcher" assert resolved["task_types"]["chapter_assembly"] == "dr_analyst" def test_runtime_profile_resolves_task_model_and_skills() -> None: runtime = resolve_runtime_profile(profile="medium") worker = runtime.role_for_task("evidence_packet") assert worker.name == "dr_analyst" assert worker.model assert "evidence-table" in worker.skills assert "search-gateway" in worker.skills assert worker.max_concurrency >= 1 def test_generate_task_cards_from_chinese_framework() -> None: framework = """ # 研究框架 ## 第1章 GLP-1 产业链的增量来自适应症扩张 研究思路：围绕临床、监管、竞争格局和生产供应链展开。 ## 第2章 供应链瓶颈决定国产替代窗口 研究思路：围绕专利、上游原料、产能和中国市场展开。 """ cards = generate_task_cards("glp1-test", framework, axes=["clinical", "regulatory"]) assert [card.task_id for card in cards] == [ "ch01-clinical", "ch01-regulatory", "ch02-clinical", "ch02-regulatory", ] assert cards[0].output_packet == "phase2/packets/ch01-clinical.json" assert cards[0].chapter_title == "GLP-1 产业链的增量来自适应症扩张" assert cards[0].research_goal assert "search-gateway" in cards[0].required_skills assert cards[0].expected_evidence["min_tier_1_2_sources"] == 2 assert cards[0].stop_conditions def test_generate_task_cards_from_research_brief_carries_prompt_and_skills() -> None: brief = { "research_method": "gmp_quality_operations_diagnosis", "work_language": "zh", "tone": "面向管理层的事实型整改诊断", "task_planning": { "required_skills": ["search-gateway", "evidence-table", "source-quality"], "search_routes_by_axis": { "quality_system_gap": ["general", "news"], "counter": ["scholar", "general"], }, "axis_prompt_briefs": { "quality_system_gap": "把现场发现映射到质量体系流程缺口和法规要求。", "counter": "主动寻找能削弱或限定结论的反方证据。", }, "stop_conditions": ["每张卡至少形成 3 条可追溯证据。"], }, "materials": [{"path": "phase0/extracted/audit.md", "role": "site_evidence"}], } framework = "## 第1章 质量体系闭环能力决定整改可信度\n\n研究思路。" cards = generate_task_cards_from_research_brief( "baifan-test", framework, brief, axes=["quality_system_gap", "counter"], ) assert [card.task_id for card in cards] == ["ch01-quality_system_gap", "ch01-counter"] assert cards[0].prompt_brief == "把现场发现映射到质量体系流程缺口和法规要求。" assert cards[0].research_method == "gmp_quality_operations_diagnosis" assert cards[0].allowed_materials == ["phase0/extracted/audit.md"] assert cards[0].preferred_model_role == "dr_analyst" assert cards[1].preferred_model_role == "dr_verifier" assert "search-gateway" in cards[0].required_skills def test_gmp_task_cards_include_fda_enforcement_route() -> None: brief = { "research_method": "gmp_quality_operations_diagnosis", "task_planning": { "search_routes_by_axis": { "quality_system_gap": ["fda", "general"], }, }, } framework = "## 第1章 偏差和 CAPA 闭环能力决定质量体系可信度\n\n研究思路。" cards = generate_task_cards_from_research_brief( "baifan-test", framework, brief, axes=["quality_system_gap"], ) assert cards[0].search_routes == ["fda", "general"] assert "FDA Warning Letters" in " ".join(cards[0].questions) assert "fda_warning_letter_or_meeting_record" in cards[0].expected_evidence["preferred_evidence_types"] def test_integrated_chapter_mode_is_method_driven_not_gmp_hardcoded() -> None: brief = { "research_method": "mckinsey_market", "phase2_mode": "chapter_integrated", "task_planning": {}, } framework = "## 第1章 市场需求正在被支付政策重塑\n\n研究思路。" cards = generate_task_cards_from_research_brief( "market-test", framework, brief, ) assert [card.task_id for card in cards] == ["ch01-chapter_integrated"] assert "literature evidence" in " ".join(cards[0].questions) assert "FDA Warning Letters" not in " ".join(cards[0].questions) def test_integrated_task_card_uses_phase1_chapter_planning() -> None: brief = { "research_method": "gmp_quality_operations_diagnosis", "phase2_mode": "chapter_integrated", "chapter_planning": [ { "chapter_id": "ch01", "title": "审计发现应先转化为商业化阶段门缺口", "core_question": "本章要判断审计发现是否反映阶段门缺口。", "bold_hypothesis": "大胆假设：风险项计数低估了商业化 readiness 缺口。", "verification_plan": ["提取现场材料原文", "检索官方法规和执法案例"], "evidence_lanes": ["site audit findings", "official baseline"], "minimum_evidence": {"local_material_quotes": 2}, "phase2_prompt_context": "章节：ch01\n必须围绕阶段门缺口求证。", } ], "task_planning": {"phase2_mode": "chapter_integrated"}, } framework = "## 第1章 审计发现应先转化为商业化阶段门缺口\n\n研究思路。" cards = generate_task_cards_from_research_brief("baifan-test", framework, brief) assert cards[0].prompt_brief == "章节：ch01\n必须围绕阶段门缺口求证。" assert cards[0].research_goal == "本章要判断审计发现是否反映阶段门缺口。" assert "大胆假设：风险项计数低估了商业化 readiness 缺口。" in cards[0].questions assert cards[0].expected_evidence["phase1_minimum_evidence"] == {"local_material_quotes": 2} assert cards[0].expected_evidence["must_address_phase1_hypothesis"] is True assert any("Phase1 的大胆假设" in item for item in cards[0].stop_conditions) def test_phase1_gmp_hypotheses_are_not_title_restatements(tmp_path: Path) -> None: project = tmp_path / "baifan" project.mkdir() (project / "phase0/extracted").mkdir(parents=True) material = project / "phase0/extracted/audit.md" material.write_text( "人员培训记录齐全，但无菌操作动作违反 First Air 原则，需要进一步培训。\n" "复盘显示 owner、关闭证据和问题升级机制仍需补齐。\n", encoding="utf-8", ) manifest = { "topic": "白帆生物 GMP 与运营诊断", "material_inventory": [{"extracted_to": "phase0/extracted/audit.md"}], } method = ResearchMethodRegistry().get("gmp_quality_operations_diagnosis") plans = build_chapter_planning( project, manifest, method, ["人员能力：培训有效性比培训记录更关键", "运营节奏：从临时协调转向管理系统"], quota=2000, ) assert "关键解释变量" not in plans[0]["bold_hypothesis"] assert "不缺培训台账" in plans[0]["bold_hypothesis"] assert "固定节奏和可视化管理系统" in plans[1]["bold_hypothesis"] assert plans[0]["core_question"] != plans[0]["title"] assert "章节成稿应围绕这一观点展开" not in plans[1]["writing_claim"] def test_research_brief_without_materials_falls_back_to_material_digest() -> None: brief = { "research_method": "mckinsey_market", "phase2_mode": "chapter_integrated", "phase1_inputs": {"material_digest": "phase1/material_digest.md"}, } framework = "## 第1章 市场需求正在被支付政策重塑\n\n研究思路。" cards = generate_task_cards_from_research_brief("market-test", framework, brief) assert cards[0].allowed_materials == ["phase1/material_digest.md"] def test_task_card_validation_rejects_duplicates_and_cycles() -> None: cards = [ TaskCard(task_id="a", chapter_ids=["ch01"], topic_axis="clinical", questions=["q"], search_routes=["scholar"], output_packet="phase2/packets/a.json", dependencies=["b"]), TaskCard(task_id="b", chapter_ids=["ch01"], topic_axis="regulatory", questions=["q"], search_routes=["general"], output_packet="phase2/packets/b.json", dependencies=["a"]), ] with pytest.raises(ValueError, match="dependency cycle"): detect_dependency_cycles(cards) with pytest.raises(ValueError, match="duplicate task_id"): validate_task_cards([cards[0], cards[0]]) def test_packet_validation_requires_sources_and_counter_evidence() -> None: packet = { "task_id": "ch01-clinical", "claims": [{"claim": "结论", "source_ids": ["src_001"]}], "evidence_items": [{"source_id": "src_001", "summary": "证据"}], "counter_evidence": [], "source_ids": ["src_001"], "source_quality_notes": ["Tier 1"], "open_questions": [], "raw_quotes_or_notes": ["Original English excerpt allowed."], } with pytest.raises(ValueError, match="counter_evidence"): validate_packet(packet) packet["counter_evidence"] = [{"claim": "限制", "source_ids": ["src_002"]}] with pytest.raises(ValueError, match="not declared"): validate_packet(packet) packet["source_ids"].append("src_002") packet["sources"] = [ {"id": "src_001", "title": "来源1", "url": "https://example.com/1"}, {"id": "src_002", "title": "来源2", "url": "https://example.com/2"}, ] validate_packet(packet) def test_skill_sync_copies_to_adapter_dirs(tmp_path: Path) -> None: canonical = tmp_path / "skills" target = tmp_path / "adapter" / "skills" source_skill = canonical / "demo" source_skill.mkdir(parents=True) (source_skill / "SKILL.md").write_text("---\nname: demo\n---\n\nBody\n", encoding="utf-8") registry = SkillRegistry(canonical_dir=canonical) copied = registry.sync_to([target]) assert copied == 1 assert (target / "demo" / "SKILL.md").read_text(encoding="utf-8").endswith("Body\n") def test_skill_sync_skips_canonical_dir_to_avoid_deleting_source(tmp_path: Path) -> None: canonical = tmp_path / "skills" source_skill = canonical / "demo" source_skill.mkdir(parents=True) (source_skill / "SKILL.md").write_text("---\nname: demo\n---\n\nBody\n", encoding="utf-8") registry = SkillRegistry(canonical_dir=canonical) copied = registry.sync_to([canonical]) assert copied == 0 assert (source_skill / "SKILL.md").exists()