script-build-and-framework-tmp/visualization-old/backend/app/card_details/retrieval.py

from __future__ import annotations

from typing import Any

from .common import (
    EventLoader,
    event_completeness,
    event_content,
    event_notices,
    event_unit,
    find_branch,
    find_detail,
    integer,
    load_events,
    suffix_int,
)
from .schema import field, payload


def project_event(detail_ref: str, load_event: EventLoader) -> dict[str, Any]:
    event_id = suffix_int(detail_ref)
    if event_id is None:
        raise KeyError(detail_ref)
    event = load_event(event_id)
    event_name = str(event.get("event_name") or "")
    input_value = event_content(event.get("input"))
    output_value = event_content(event.get("output"))
    inputs = [field(
        f"event:{event_id}:input",
        "运行输入",
        input_value,
        source_kind="runtime-event",
        source_label="Event Body 输入",
        source_ref=detail_ref,
        field_path="input.content",
        relation="direct-input",
        completeness="complete" if input_value not in (None, "", [], {}) else "missing",
    )]
    outputs = [field(
        f"event:{event_id}:output",
        "运行输出",
        output_value,
        source_kind="runtime-event",
        source_label="Event Body 输出",
        source_ref=detail_ref,
        field_path="output.content",
        relation="persisted-output",
        completeness="complete" if output_value not in (None, "", [], {}) else "missing",
    )]
    card_kind = _event_card_kind(event)
    modules = _event_modules(event)
    return payload(
        card_kind,
        "single-event",
        inputs=inputs,
        units=[event_unit(event)],
        outputs=outputs,
        runtime_summary=f"{event_name or '运行事件'} 是一次可独立下钻的运行 I/O。",
        notices=event_notices([event]),
        completeness=event_completeness(event),  # type: ignore[arg-type]
        business_modules=modules(inputs[0]["id"], outputs[0]["id"]),
    )


def project_retrieval_stage(
    detail_ref: str,
    round_index: int,
    branch_id: int,
    view: dict[str, Any],
    load_event: EventLoader,
) -> dict[str, Any]:
    branch = find_branch(view, round_index, branch_id)
    if branch is None:
        raise KeyError(detail_ref)
    stage = branch.get("retrievalStage") or {}
    refs: list[Any] = [stage.get("implementerEventId")]
    for group in stage.get("directToolGroups") or []:
        refs.extend(item.get("eventId") for item in group.get("calls") or [])
    for run in stage.get("agentRuns") or []:
        refs.append(run.get("eventId"))
        refs.extend(item.get("eventId") for item in run.get("attempts") or [])
    events = load_events(load_event, refs)
    inputs = [field(
        f"round:{round_index}:branch:{branch_id}:retrieval:purpose",
        "取数目的",
        ((branch.get("task") or {}).get("card") or {}).get("primary", {}).get("value"),
        source_kind="database",
        source_label="实现任务",
        source_ref=f"round:{round_index}:branch:{branch_id}:task",
        field_path="impl_task",
        relation="standing-constraint",
    )]
    outputs = [
        field(f"round:{round_index}:branch:{branch_id}:retrieval:mode", "执行方式", stage.get("observedMode") or "unknown", source_kind="calculation", source_label="按执行时间和依赖关系确定", source_ref=detail_ref, field_path="observedMode", relation="persisted-output", completeness="complete"),
        field(f"round:{round_index}:branch:{branch_id}:retrieval:waves", "执行波次", stage.get("waves") or [], source_kind="calculation", source_label="按明确 Event 时间聚合", source_ref=detail_ref, field_path="waves", relation="persisted-output", completeness="complete"),
        field(f"round:{round_index}:branch:{branch_id}:retrieval:groups", "工具取数", stage.get("directToolGroups") or [], source_kind="calculation", source_label="按明确 Event scope 分组", source_ref=detail_ref, field_path="directToolGroups", relation="persisted-output", completeness="complete"),
        field(f"round:{round_index}:branch:{branch_id}:retrieval:agents", "Agent 取数", stage.get("agentRuns") or [], source_kind="calculation", source_label="按 parent_event_id 聚合", source_ref=detail_ref, field_path="agentRuns", relation="persisted-output", completeness="complete"),
        field(f"round:{round_index}:branch:{branch_id}:retrieval:result", "阶段结果与失败", {"status": stage.get("status"), "directToolGroups": stage.get("directToolGroups") or [], "agentRuns": stage.get("agentRuns") or []}, source_kind="calculation", source_label="取数阶段状态确定性聚合", source_ref=detail_ref, field_path="status", relation="persisted-output", completeness="complete"),
    ]
    return payload(
        "retrieval-stage",
        "aggregate",
        inputs=inputs,
        units=[event_unit(item) for item in events],
        outputs=outputs,
        runtime_summary=f"按实现 Agent 的明确 scope 聚合 {len(events)} 个运行事件，观测模式为 {stage.get('observedMode') or 'unknown'}。",
        notices=event_notices(events),
        completeness="complete" if events and stage.get("status") not in {"missing", "unknown"} else "partial" if events else "missing",
        business_modules=[
            {"id": "purpose", "title": "取数目的", "sourceIds": [inputs[0]["id"]]},
            {"id": "mode", "title": "执行方式", "sourceIds": [outputs[0]["id"]]},
            {"id": "waves", "title": "执行波次", "sourceIds": [outputs[1]["id"]]},
            {"id": "direct", "title": "工具取数", "sourceIds": [outputs[2]["id"]]},
            {"id": "agents", "title": "Agent 取数", "sourceIds": [outputs[3]["id"]]},
            {"id": "result", "title": "阶段结果与失败", "sourceIds": [outputs[4]["id"]]},
        ],
    )


def project_direct_group(
    detail_ref: str, view: dict[str, Any], load_event: EventLoader
) -> dict[str, Any]:
    group = find_detail(view, detail_ref)
    if group is None:
        raise KeyError(detail_ref)
    calls = group.get("calls") or []
    events = load_events(load_event, [item.get("eventId") for item in calls])
    inputs = [field(
        f"{detail_ref}:purpose",
        "读取目的",
        [item.get("businessLabel") for item in calls if item.get("businessLabel")],
        source_kind="calculation",
        source_label="工具类型业务标签",
        source_ref=detail_ref,
        field_path="calls.businessLabel",
        relation="direct-input",
    )]
    outputs = [field(
        f"{detail_ref}:results",
        "实际返回的信息",
        [event_content(item.get("output")) for item in events],
        source_kind="runtime-event",
        source_label="逐次工具 Event Body",
        source_ref=detail_ref,
        field_path="events[].output.content",
        relation="persisted-output",
        completeness="complete" if events and all(event_content(item.get("output")) is not None for item in events) else "partial",
    )]
    return payload(
        "retrieval-direct-group",
        "event-sequence",
        inputs=inputs,
        units=[event_unit(item) for item in events],
        outputs=outputs,
        runtime_summary=f"该工具取数组包含 {len(events)} 次直接读取。",
        notices=event_notices(events),
        completeness="complete" if events and len(events) == len(calls) else "partial" if events else "missing",
        business_modules=[
            {"id": "purpose", "title": "读取目的", "sourceIds": [inputs[0]["id"]]},
            {"id": "calls", "title": "逐次工具调用", "sourceIds": [outputs[0]["id"]]},
            {"id": "results", "title": "实际返回的信息", "sourceIds": [outputs[0]["id"]]},
            {"id": "failures", "title": "失败与缺失", "sourceIds": [outputs[0]["id"]]},
        ],
    )


def project_retrieval_agent(
    detail_ref: str, view: dict[str, Any], load_event: EventLoader
) -> dict[str, Any]:
    run = find_detail(view, detail_ref)
    if run is None:
        raise KeyError(detail_ref)
    agent_id = integer(run.get("eventId")) or suffix_int(detail_ref)
    refs = [agent_id, *[item.get("eventId") for item in run.get("attempts") or []]]
    events = load_events(load_event, refs)
    agent = next((item for item in events if integer(item.get("id")) == agent_id), {})
    queries = [item for item in events if integer(item.get("id")) != agent_id]
    task = _agent_task(agent) or run.get("taskPreview")
    screening = _agent_screening(agent) or (run.get("screening") or {}).get("preview")
    inputs = [field(
        f"{detail_ref}:task",
        "完整取数任务",
        task,
        source_kind="runtime-event",
        source_label="取数 Agent Event Body",
        source_ref=f"event:{agent_id}",
        field_path="input.content.task",
        relation="direct-input",
        completeness="complete" if _agent_task(agent) else "partial" if task else "missing",
    )]
    outputs = [
        field(f"{detail_ref}:raw-results", "原始命中", [event_content(item.get("output")) for item in queries], source_kind="runtime-event", source_label="查询 Event Body", source_ref=detail_ref, field_path="queryEvents[].output.content", relation="persisted-output", completeness="complete" if queries and all(event_content(item.get("output")) is not None for item in queries) else "partial" if queries else "missing"),
        field(f"{detail_ref}:screening", "初筛整理", screening, source_kind="runtime-event", source_label="取数 Agent Event Body", source_ref=f"event:{agent_id}", field_path="output.content", relation="persisted-output", completeness="complete" if _agent_screening(agent) else "partial" if screening else "missing"),
        field(f"{detail_ref}:query-summary", "查询统计", run.get("querySummary"), source_kind="calculation", source_label="查询 Event 状态聚合", source_ref=detail_ref, field_path="querySummary", relation="persisted-output"),
    ]
    return payload(
        "retrieval-agent",
        "aggregate",
        inputs=inputs,
        units=[event_unit(item, label=(run.get("businessLabel") if integer(item.get("id")) == agent_id else None)) for item in events],
        outputs=outputs,
        runtime_summary=f"一个取数 Agent 和 {len(queries)} 次查询组成该运行过程。",
        notices=event_notices(events),
        completeness="complete" if _agent_task(agent) and _agent_screening(agent) else "partial" if events else "missing",
        business_modules=[
            {"id": "task", "title": "完整取数任务", "sourceIds": [inputs[0]["id"]]},
            {"id": "queries", "title": "查询过程", "sourceIds": [outputs[2]["id"]]},
            {"id": "raw", "title": "原始命中", "sourceIds": [outputs[0]["id"]]},
            {"id": "screening", "title": "初筛整理", "sourceIds": [outputs[1]["id"]]},
            {"id": "failures", "title": "失败与空结果", "sourceIds": [outputs[2]["id"]]},
        ],
        card_fields=[
            {"key": "target", "label": "取数目标", "sourceIds": [inputs[0]["id"]], "transform": "使用完整任务的结构化目标"},
            {"key": "queryStats", "label": "查询统计", "sourceIds": [outputs[2]["id"]], "transform": "命中/空/失败数量"},
            {"key": "screening", "label": "初筛结论", "sourceIds": [outputs[1]["id"]], "transform": "使用 Agent 完整结果中的 summary"},
        ],
    )


def _agent_task(event: dict[str, Any]) -> Any:
    content = event_content(event.get("input"))
    return content.get("task") if isinstance(content, dict) else content


def _agent_screening(event: dict[str, Any]) -> Any:
    content = event_content(event.get("output"))
    return content.get("summary") if isinstance(content, dict) and content.get("summary") else content


def _event_card_kind(event: dict[str, Any]) -> str:
    name = str(event.get("event_name") or "")
    if name == "think_and_plan" and str(event.get("agent_role") or "") == "main":
        return "planning-analysis"
    if name == "script_multipath_evaluator":
        return "multipath-review"
    if name == "script_evaluator":
        return "overall-review"
    if name == "script_implementer":
        return "implementation-task"
    if str(event.get("event_type") or "") == "tool_call":
        return "query" if name.startswith(("search_", "query_", "get_")) else "tool-call"
    return "runtime-event"


def _event_modules(event: dict[str, Any]):
    kind = _event_card_kind(event)
    def modules(input_id: str, output_id: str) -> list[dict[str, Any]]:
        if kind == "planning-analysis":
            return [
                {"id": "summary", "title": "规划概要", "sourceIds": [input_id]},
                {"id": "thought", "title": "完整思考", "sourceIds": [input_id]},
                {"id": "plan", "title": "执行计划", "sourceIds": [input_id]},
                {"id": "action", "title": "下一步", "sourceIds": [input_id]},
                {"id": "completeness", "title": "记录完整性", "sourceIds": [input_id, output_id]},
            ]
        if kind == "multipath-review":
            return [
                {"id": "scope", "title": "评审范围", "sourceIds": [input_id]},
                {"id": "criteria", "title": "评审标准", "sourceIds": [input_id]},
                {"id": "per-branch", "title": "逐方案评审", "sourceIds": [output_id]},
                {"id": "comparison", "title": "方案对比", "sourceIds": [output_id]},
                {"id": "recommendation", "title": "评审建议", "sourceIds": [output_id]},
            ]
        if kind == "overall-review":
            return [
                {"id": "subject", "title": "评审对象", "sourceIds": [input_id]},
                {"id": "criteria", "title": "标准", "sourceIds": [input_id]},
                {"id": "achieved", "title": "已达成", "sourceIds": [output_id]},
                {"id": "issues", "title": "问题", "sourceIds": [output_id]},
                {"id": "conclusion", "title": "总结论", "sourceIds": [output_id]},
                {"id": "next", "title": "下一轮建议", "sourceIds": [output_id]},
            ]
        if kind == "implementation-task":
            return [
                {"id": "task", "title": "完整实现任务", "sourceIds": [input_id]},
                {"id": "result", "title": "实现 Agent 返回", "sourceIds": [output_id]},
            ]
        if kind in {"query", "tool-call"}:
            return [
                {"id": "purpose", "title": "目的 / 查询条件", "sourceIds": [input_id]},
                {"id": "result", "title": "返回内容 / 结果列表", "sourceIds": [output_id]},
                {"id": "status", "title": "状态、耗时与完整性", "sourceIds": [input_id, output_id]},
            ]
        return [
            {"id": "input", "title": "运行输入", "sourceIds": [input_id]},
            {"id": "output", "title": "运行输出", "sourceIds": [output_id]},
        ]
    return modules