tests/test_orchestrator.py

"""AnalysisOrchestrator 测试（intent-understanding-layer 任务 11）。

覆盖 Property 5（确认门禁：未确认 / 澄清态绝不触达 compute）与需求 9.4（审计）。
"""

from __future__ import annotations

from kernel.orchestrator import AnalysisOrchestrator
from kernel.registry import SkillRegistry
from kernel.skill_base import (
    ComputeResult,
    ExtractedData,
    InputKind,
    PharmaSkill,
    RawInput,
    ReportSections,
    SkillMeta,
)
from kernel.task_sheet import AnalysisTaskSheet, ClarificationDecision, Intent


class _SpySkill(PharmaSkill):
    meta = SkillMeta(id="descriptive_summary", display_name="x", description="",
                     version="1.0.0", input_kinds={InputKind.TEXT})

    def __init__(self):
        self.compute_called = False

    def render_inputs(self, st_ctx):
        return RawInput()

    def extract(self, raw, svc):
        # 把任务单是否成功传入做个标记
        self.seen_sheet = (raw.extra or {}).get("task_sheet")
        return ExtractedData(payload={})

    def compute(self, data):
        self.compute_called = True
        return ComputeResult(summary={"ok": True})

    def explain(self, result, svc):
        return ReportSections(sections={"done": "yes"})


class _Audit:
    def __init__(self):
        self.events = []

    def record_analysis(self, user, skill_id, step=""):
        self.events.append((user, skill_id, step))


class _Svc:
    def __init__(self):
        self.audit = _Audit()
        self.llm = None
        self.report = None
        self.chart = None
        self.prompt_guard = None
        self.lang = "zh"


def _registry(skill):
    reg = SkillRegistry()
    reg.register(skill)
    return reg


def test_clarification_sheet_never_reaches_compute():
    skill = _SpySkill()
    orch = AnalysisOrchestrator(_registry(skill), _Svc())
    sheet = AnalysisTaskSheet(
        intent=Intent.UNKNOWN,
        clarification=ClarificationDecision(needed=True, reason_code="no_llm"),
    )
    sections = orch.run_confirmed(sheet, RawInput())
    assert skill.compute_called is False
    assert "clarification" in sections.sections


def test_confirmed_descriptive_sheet_runs_pipeline():
    skill = _SpySkill()
    svc = _Svc()
    orch = AnalysisOrchestrator(_registry(skill), svc)
    sheet = AnalysisTaskSheet(intent=Intent.DESCRIPTIVE_SUMMARY,
                              proposed_skill_id="descriptive_summary")
    sections = orch.run_confirmed(sheet, RawInput(goal="梳理"))
    assert skill.compute_called is True
    # 任务单通过 raw.extra 传入了 skill.extract
    assert skill.seen_sheet is not None
    # 审计记录了确认执行（需求 9.4）
    assert any(e[1] == "descriptive_summary" for e in svc.audit.events)


def test_unroutable_sheet_short_circuits():
    skill = _SpySkill()
    orch = AnalysisOrchestrator(_registry(skill), _Svc())
    sheet = AnalysisTaskSheet(intent=Intent.UNKNOWN)  # 无澄清但也无法路由
    sections = orch.run_confirmed(sheet, RawInput())
    assert skill.compute_called is False
    assert "clarification" in sections.sections


def test_build_task_sheet_no_llm_is_clarification():
    skill = _SpySkill()
    orch = AnalysisOrchestrator(_registry(skill), _Svc())  # svc.llm = None
    sheet = orch.build_task_sheet(RawInput(goal="梳理数据", extra={"text_content": "有关物质 总杂% 0"}))
    assert sheet.needs_clarification()
    assert skill.compute_called is False