script-build-and-framework-tmp/visualization/backend/app/inspector_projection.py

from __future__ import annotations

import re
from typing import Any

from .business_detail_text import (
    assessment_text,
    batches_markdown,
    branch_task_sections,
    branch_task_text,
    branches_markdown,
    business_text,
    decisions_markdown,
    event_business_sections,
    source_markdown,
    clean_report,
)
from .sanitizer import sanitize
from .retrieval_detail_projection import (
    is_retrieval_detail_event,
    project_retrieval_event,
)


def project_round_detail(
    script_build_id: int,
    round_: dict[str, Any],
    bundle: dict[str, Any] | None = None,
) -> dict[str, Any]:
    raw_round = next(
        (
            item
            for item in (bundle or {}).get("rounds", [])
            if int(item.get("round_index") or 0) == int(round_.get("roundIndex") or 0)
        ),
        None,
    )
    sections = [
        _section(
            "goal",
            "本轮目标",
            (raw_round or {}).get("goal")
            or (round_.get("summary", {}).get("goal") or {}).get("headline"),
        ),
        _section("plan", "当前保存的规划", _plan_text(round_.get("plan") or {})),
        _section("branches", "候选方案", branches_markdown(round_.get("branches") or [])),
        _section(
            "decisions",
            "候选评审与主 Agent 决策",
            _convergence_markdown(round_.get("convergenceBatches") or []),
        ),
        _section(
            "evaluation",
            "主脚本整体评审",
            _node_value(round_.get("overallEvaluation")) or "未找到明确整轮评审结论",
        ),
        _section("result", "本轮产出", _node_value(round_.get("result"))),
    ]
    result = round_.get("result") or {}
    changes = None
    if result.get("hasChanges"):
        changes = {
            "summary": _node_value(result),
            "sections": [
                item
                for item in [
                    _section(
                        "content",
                        "内容方案",
                        _round_content_changes(round_.get("branches") or []),
                    ),
                    _section(
                        "domain",
                        "领域信息",
                        _round_domain_changes(round_.get("branches") or []),
                    ),
                ]
                if item
            ],
            "artifactRef": "artifact:base:current",
            "exactness": "current-only",
        }
    return _payload(
        sections,
        {
            "scriptBuildId": script_build_id,
            "round": raw_round,
            "multipathDecisions": [
                item
                for item in (bundle or {}).get("multipathDecisions", [])
                if int(item.get("round_index") or 0) == int(round_.get("roundIndex") or 0)
            ],
        },
        changes,
    )


def project_activity_detail(
    script_build_id: int,
    activity_id: str,
    view: dict[str, Any],
    bundle: dict[str, Any],
    event_detail: dict[str, Any] | None = None,
) -> dict[str, Any]:
    if activity_id == "run:final-result":
        return _final_result_detail(script_build_id, view, bundle)

    missing_convergence = _find_missing_convergence(view, activity_id)
    if missing_convergence is not None:
        node, batch = missing_convergence
        return _payload(
            [
                _section(
                    "status",
                    str(node.get("title") or "记录状态"),
                    _node_value(node),
                    "summary",
                )
            ],
            {
                "scriptBuildId": script_build_id,
                "roundIndex": batch.get("roundIndex"),
                "branchIds": batch.get("branchIds") or [],
                "association": batch.get("association"),
                "unscopedReviewRefs": node.get("unscopedReviewRefs") or [],
            },
        )

    decision = _find_decision_projection(view, activity_id)
    if decision is not None:
        return _decision_detail_payload(
            script_build_id, decision, bundle, activity_id, event_detail
        )

    if activity_id.startswith("domain-info:"):
        record_id = _suffix_int(activity_id)
        row = _find_by_id(bundle.get("domainInfo") or [], record_id)
        if not row:
            raise KeyError(activity_id)
        return _payload(
            [
                _section("fact", "已核实领域事实", row.get("content")),
                _section("verification", "核实说明", row.get("note")),
                _section("source", "信息出处", _source_value(row.get("source"))),
                _section(
                    "use",
                    "用途与存疑状态",
                    "已作为领域事实独立保存；是否进入具体候选脚本文案未记录。",
                ),
            ],
            {"scriptBuildId": script_build_id, "table": "script_build_domain_info", "record": row},
            {
                "summary": "这条领域事实已经独立写入领域信息库。",
                "sections": [],
                "exactness": "exact",
            },
        )

    if activity_id.startswith("retrieval-direct:"):
        group = _find_retrieval_item(view, "directToolGroups", activity_id)
        if not group:
            raise KeyError(activity_id)
        calls = group.get("calls") or []
        payload = _payload(
            [
                _section("purpose", "读取方式", "实现 Agent 直接调用只读工具，补齐创作所需的上下文。"),
                _section("sources", "读取了什么", _direct_sources_markdown(group)),
                _section("result", "得到什么", _direct_result_markdown(calls), "summary"),
            ],
            {
                "scriptBuildId": script_build_id,
                "retrievalGroup": group,
                "events": _events_for_ids(bundle, [item.get("eventId") for item in calls]),
            },
        )
        payload.update(
            {
                "detailKind": "retrieval-group",
                "retrievalKind": "direct-tools",
                "activities": [
                    {
                        "id": item.get("id") or item.get("detailRef"),
                        "kind": "tool",
                        "label": item.get("businessLabel") or "读取相关信息",
                        "status": item.get("status") or "unknown",
                        "summary": item.get("resultSummary") or "",
                        "detailRef": item.get("detailRef"),
                    }
                    for item in calls
                ],
            }
        )
        return payload

    if activity_id.startswith("retrieval-agent:"):
        run = _find_retrieval_item(view, "agentRuns", activity_id)
        if not run:
            raise KeyError(activity_id)
        event = event_detail or _event_by_id(bundle, run.get("eventId"))
        screening = run.get("screening") or {}
        full_task = _event_nested_value(event, "input", "task")
        raw_screening_text = _event_nested_value(event, "output", "summary")
        if not raw_screening_text:
            raw_screening_text = screening.get("preview")
        screening_text = clean_report(raw_screening_text)
        payload = _payload(
            [
                _section("task", "完整取数任务", full_task or run.get("taskPreview") or "取数目标未记录"),
                _section("queries", "查询过程", _agent_query_markdown(run)),
                _section("hits", "原始命中", _agent_hit_markdown(run)),
                _section(
                    "screening",
                    "初筛整理",
                    screening_text or _screening_fallback(screening.get("state")),
                    "report",
                ),
            ],
            {
                "scriptBuildId": script_build_id,
                "agentRun": run,
                "agentEvent": event,
                "queryEvents": _events_for_ids(bundle, [item.get("eventId") for item in run.get("attempts") or []]),
            },
        )
        activities = [
            {
                "id": item.get("id") or item.get("detailRef"),
                "kind": "query",
                "label": item.get("queryLabel") or "查询内容详见详情",
                "status": item.get("status") or "unknown",
                "resultCount": item.get("resultCount"),
                "detailRef": item.get("detailRef"),
            }
            for item in run.get("attempts") or []
        ]
        payload.update(
            {
                "detailKind": "retrieval-agent",
                "retrievalKind": "agent",
                "activities": activities,
            }
        )
        if not activities:
            if script_build_id <= 433:
                payload["missingActivityNotice"] = "该历史运行没有结构化查询明细。"
            elif run.get("agentType") == "retrieve_data_pattern_relation":
                payload["missingActivityNotice"] = (
                    "本次运行未保存可下钻的结构化查询记录。"
                )
        return payload

    round_index, branch_id, role = _activity_parts(activity_id)
    round_ = _find_round(view, round_index) if round_index is not None else None
    if round_ is None:
        raise KeyError(activity_id)
    if branch_id is None:
        if role is None:
            return project_round_detail(script_build_id, round_, bundle)
        node = _round_node(round_, role)
        if not node:
            raise KeyError(activity_id)
        raw_round = next(
            (
                item
                for item in bundle.get("rounds") or []
                if int(item.get("round_index") or 0) == round_index
            ),
            None,
        )
        if role == "result":
            return _round_result_detail(script_build_id, round_, raw_round, bundle)
        return _node_detail(script_build_id, node, raw_round)

    branch = next(
        (
            item
            for item in round_.get("branches") or []
            if int(item.get("branchId") or 0) == branch_id
        ),
        None,
    )
    raw_branch = next(
        (
            item
            for item in bundle.get("branches") or []
            if int(item.get("branch_id") or 0) == branch_id
        ),
        None,
    )
    if not branch:
        raise KeyError(activity_id)
    if role is None:
        return _branch_detail(script_build_id, branch, raw_branch, bundle)
    if role == "retrieval":
        return _retrieval_stage_detail(script_build_id, branch, raw_branch, bundle)
    node = _branch_node(branch, role)
    if not node:
        raise KeyError(activity_id)
    if role == "output":
        return _branch_output_detail(script_build_id, branch, raw_branch, bundle)
    if role == "task":
        return _branch_task_detail(
            script_build_id,
            branch,
            raw_branch,
            bundle,
            round_index=round_index,
            branch_id=branch_id,
        )
    return _node_detail(script_build_id, node, raw_branch)


def project_event_detail(
    script_build_id: int,
    event: dict[str, Any],
    *,
    run_status: str | None = None,
) -> dict[str, Any]:
    if is_retrieval_detail_event(event):
        payload = project_retrieval_event(event, run_status=run_status)
        payload["technical"] = sanitize(
            {
                "source": {
                    "scriptBuildId": script_build_id,
                    "table": "script_build_event / script_build_event_body",
                    "detailRef": f"event:{event.get('id')}",
                },
                **_event_technical_groups(event),
            },
            max_text=None,
        )
        return sanitize(payload, max_text=None)
    if (
        event.get("event_name") == "think_and_plan"
        and event.get("agent_role") == "main"
        and int(event.get("agent_depth") or 0) == 0
    ):
        raw_input = event.get("input")
        content = raw_input.get("content") if isinstance(raw_input, dict) else None
        data = content if isinstance(content, dict) else event.get("inputData")
        data = data if isinstance(data, dict) else {}
        return _payload(
            [
                _section(
                    "summary",
                    "规划概要",
                    data.get("thought_summary"),
                    "summary",
                    visual_role="key-result",
                ),
                _section(
                    "thought",
                    "完整思考",
                    data.get("thought"),
                    "report",
                    visual_role="section",
                ),
                _section(
                    "plan",
                    "执行计划",
                    data.get("plan"),
                    visual_role="section",
                ),
                _section(
                    "action",
                    "下一步",
                    data.get("action"),
                    visual_role="reason",
                ),
            ],
            {
                "scriptBuildId": script_build_id,
                "table": "script_build_event / script_build_event_body",
                "event": event,
            },
        )
    return _payload(
        event_business_sections(event),
        {"scriptBuildId": script_build_id, "table": "script_build_event / script_build_event_body", "event": event},
    )


def project_artifact_detail(
    script_build_id: int, snapshot_ref: str, artifact: dict[str, Any]
) -> dict[str, Any]:
    paragraphs, elements, links = _normalized_artifact_records(artifact)
    exactness = "current-only" if "base:current" in snapshot_ref else _artifact_exactness(artifact)
    context = artifact.get("artifactContext") if isinstance(artifact.get("artifactContext"), dict) else {}
    context = {
        "type": context.get("type") or ("final" if "base:current" in snapshot_ref else "candidate"),
        **context,
        "exactness": exactness,
    }
    candidate = context.get("type") == "candidate"
    scale_title = "候选脚本规模" if candidate else "主脚本规模"
    content_summary = "候选脚本内容" if candidate else "当前主脚本内容"
    payload = _payload(
        [
            _section("summary", scale_title, f"{len(paragraphs)} 个段落、{len(elements)} 个元素、{len(links)} 条关联"),
        ],
        {
            "scriptBuildId": script_build_id,
            "snapshotRef": snapshot_ref,
            "sourceOrigin": artifact.get("sourceOrigin") or "database",
            "exactness": exactness,
            "artifactContext": context,
            "recordCounts": {
                "paragraphs": len(paragraphs),
                "elements": len(elements),
                "links": len(links),
            },
        },
        {
            "summary": content_summary,
            "sections": [
                _section(
                    "counts",
                    "内容规模",
                    f"{len(paragraphs)} 个段落、{len(elements)} 个元素、{len(links)} 条关联",
                ),
            ],
            "artifactRef": snapshot_ref,
            "exactness": exactness,
        },
    )
    payload["scriptTable"] = sanitize(
        _script_table_projection(snapshot_ref, artifact, exactness),
        max_text=None,
    )
    payload["artifactContext"] = sanitize(context, max_text=None)
    return payload


_ELEMENT_REFERENCE = re.compile(r"\[([^\]:]+):(\d+)\]")
_SCRIPT_SECTIONS = (
    ("theme", "主题"),
    ("form", "形式"),
    ("function", "作用"),
    ("feeling", "感受"),
)


def _script_table_projection(
    snapshot_ref: str, artifact: dict[str, Any], exactness: str
) -> dict[str, Any]:
    snapshot = artifact.get("snapshot") if isinstance(artifact.get("snapshot"), dict) else artifact
    snapshot = snapshot if isinstance(snapshot, dict) else {}
    paragraphs, elements, links = _normalized_artifact_records(artifact)

    element_names: dict[int, str] = {}
    for element in elements:
        element_id = _as_int(element.get("effectiveId") or element.get("id"))
        row_id = _as_int(element.get("rowId") or element.get("id"))
        name = str(element.get("name") or "").strip()
        if element_id is not None:
            element_names[element_id] = name
        if row_id is not None:
            element_names[row_id] = name

    paragraph_rows = _ordered_paragraphs(
        [_script_paragraph(row, element_names) for row in paragraphs]
    )

    paragraph_lookup: dict[int, dict[str, Any]] = {}
    for paragraph in paragraph_rows:
        for key in ("id", "rowId"):
            value = _as_int(paragraph.get(key))
            if value is not None:
                paragraph_lookup[value] = paragraph

    linked_paragraphs: dict[int, list[dict[str, Any]]] = {}
    for link in links:
        element_id = _as_int(link.get("element_id") or link.get("elementId"))
        paragraph_id = _as_int(link.get("paragraph_id") or link.get("paragraphId"))
        paragraph = paragraph_lookup.get(paragraph_id) if paragraph_id is not None else None
        if element_id is None or paragraph is None:
            continue
        linked_paragraphs.setdefault(element_id, []).append(
            {
                "id": paragraph.get("id"),
                "index": paragraph.get("index"),
                "name": paragraph.get("name"),
            }
        )

    element_rows = []
    for element in elements:
        element_id = element.get("effectiveId") or element.get("id")
        row_id = element.get("rowId") or element.get("id")
        link_keys = {
            value
            for value in (_as_int(row_id), _as_int(element_id))
            if value is not None
        }
        linked = [
            paragraph
            for key in link_keys
            for paragraph in linked_paragraphs.get(key, [])
        ]
        element_rows.append(
            {
                "id": element_id,
                "rowId": row_id,
                "name": element.get("name") or "未命名元素",
                "primaryDimension": element.get("dimension_primary"),
                "secondaryDimension": element.get("dimension_secondary"),
                "paragraphs": _unique_records(linked, "id"),
            }
        )
    element_rows.sort(key=lambda row: (str(row.get("name") or ""), _sort_number(row.get("id"))))

    return {
        "snapshotRef": snapshot_ref,
        "exactness": exactness,
        "sourceOrigin": artifact.get("sourceOrigin") or snapshot.get("sourceOrigin") or "database",
        "counts": {
            "paragraphs": len(paragraph_rows),
            "elements": len(element_rows),
            "links": len(links),
        },
        "maxDepth": (
            max((int(row.get("depth") or 0) for row in paragraph_rows), default=0) + 1
            if paragraph_rows
            else 0
        ),
        "paragraphs": paragraph_rows,
        "elements": element_rows,
    }


def _normalized_artifact_records(
    artifact: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]:
    """Normalize current overlays and frozen branch snapshots to one table shape.

    Current overlays store a flat ``paragraphs`` list plus top-level
    ``paragraphElements``.  Frozen ``content_snapshot`` rows store second-level
    paragraphs under ``sub_paragraphs`` and links under each paragraph's
    ``linked_elements``.  The visualizer must preserve both without pretending
    that a truncated top-level list is the complete candidate script.
    """
    snapshot = artifact.get("snapshot") if isinstance(artifact.get("snapshot"), dict) else artifact
    snapshot = snapshot if isinstance(snapshot, dict) else {}
    paragraphs: list[dict[str, Any]] = []
    seen_paragraphs: set[tuple[str, str]] = set()

    def visit(row: Any, parent_id: Any = None) -> None:
        if not isinstance(row, dict):
            return
        current = dict(row)
        children = current.pop("sub_paragraphs", None) or current.pop("subParagraphs", None) or []
        if parent_id is not None and current.get("parent_id") is None and current.get("parentId") is None:
            current["parent_id"] = parent_id
        row_id = current.get("rowId") or current.get("id")
        key = (str(row_id), str(current.get("baseRefId") or current.get("base_ref_id") or ""))
        if key not in seen_paragraphs:
            seen_paragraphs.add(key)
            paragraphs.append(current)
        effective_parent = current.get("effectiveId") or current.get("id") or row_id
        for child in children if isinstance(children, list) else []:
            visit(child, effective_parent)

    for paragraph in snapshot.get("paragraphs") or []:
        visit(paragraph)

    elements = [item for item in snapshot.get("elements") or [] if isinstance(item, dict)]
    links: list[dict[str, Any]] = []
    seen_links: set[tuple[str, str]] = set()

    def append_link(link: Any, paragraph_id: Any = None) -> None:
        if not isinstance(link, dict):
            return
        current = dict(link)
        if paragraph_id is not None and current.get("paragraph_id") is None and current.get("paragraphId") is None:
            current["paragraph_id"] = paragraph_id
        element_id = current.get("element_id") or current.get("elementId")
        linked_paragraph = current.get("paragraph_id") or current.get("paragraphId")
        if element_id is None or linked_paragraph is None:
            return
        key = (str(linked_paragraph), str(element_id))
        if key in seen_links:
            return
        seen_links.add(key)
        links.append(current)

    for link in snapshot.get("paragraphElements") or snapshot.get("paragraph_elements") or []:
        append_link(link)
    for paragraph in paragraphs:
        paragraph_id = paragraph.get("effectiveId") or paragraph.get("id") or paragraph.get("rowId")
        nested_links = paragraph.get("linked_elements") or paragraph.get("linkedElements") or []
        if isinstance(nested_links, list) and nested_links:
            for link in nested_links:
                append_link(link, paragraph_id)
        else:
            for element_id in paragraph.get("linked_element_ids") or paragraph.get("linkedElementIds") or []:
                append_link({"element_id": element_id}, paragraph_id)

    return paragraphs, elements, links


def _script_paragraph(row: dict[str, Any], element_names: dict[int, str]) -> dict[str, Any]:
    full_description = str(row.get("full_description") or "")
    return {
        "id": row.get("effectiveId") or row.get("id"),
        "rowId": row.get("rowId") or row.get("id"),
        "parentId": row.get("parent_id"),
        "depth": 0,
        "index": row.get("paragraph_index"),
        "level": row.get("level"),
        "name": row.get("name") or "未命名段落",
        "contentRange": _content_range(row.get("content_range")),
        "sections": {
            key: {
                "label": label,
                "description": row.get(key),
                "descriptionParts": _description_parts(
                    str(row.get(key) or ""), element_names
                ),
                "dimensions": _dimensions(row.get(f"{key}_elements")),
            }
            for key, label in _SCRIPT_SECTIONS
        },
        "description": row.get("description"),
        "fullDescription": full_description,
        "fullDescriptionParts": _description_parts(full_description, element_names),
    }


def _ordered_paragraphs(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Keep every child next to its real parent without guessing from level labels."""
    by_reference: dict[int, dict[str, Any]] = {}
    for row in rows:
        for value in (row.get("rowId"), row.get("id")):
            reference = _as_int(value)
            if reference is not None:
                by_reference[reference] = row

    children: dict[int, list[dict[str, Any]]] = {}
    roots: list[dict[str, Any]] = []
    for row in rows:
        parent = by_reference.get(_as_int(row.get("parentId")))
        if parent is None or parent is row:
            row["parentId"] = None
            roots.append(row)
            continue
        row["parentId"] = parent.get("id")
        children.setdefault(id(parent), []).append(row)

    ordered: list[dict[str, Any]] = []
    visited: set[int] = set()

    def visit(row: dict[str, Any], depth: int, ancestry: set[int]) -> None:
        marker = id(row)
        if marker in visited:
            return
        if marker in ancestry:
            row["parentId"] = None
            depth = 0
        visited.add(marker)
        row["depth"] = depth
        ordered.append(row)
        next_ancestry = {*ancestry, marker}
        for child in sorted(children.get(marker, []), key=_paragraph_sort_key):
            visit(child, depth + 1, next_ancestry)

    for root in sorted(roots, key=_paragraph_sort_key):
        visit(root, 0, set())
    for row in sorted(rows, key=_paragraph_sort_key):
        if id(row) not in visited:
            row["parentId"] = None
            visit(row, 0, set())
    return ordered


def _paragraph_sort_key(row: dict[str, Any]) -> tuple[Any, ...]:
    return (_sort_number(row.get("index")), _sort_number(row.get("id")))


def _content_range(value: Any) -> list[str]:
    if not isinstance(value, dict):
        return [_display_value(value)] if value not in (None, "") else []

    tags: list[str] = []
    for key, item in value.items():
        if key == "标题":
            tags.append("标题")
        elif key == "图片" and isinstance(item, list):
            if not item:
                continue
            values = [str(entry) for entry in item]
            tags.append(f"图{values[0]}" if len(values) == 1 else f"图{values[0]}-{values[-1]}")
        elif key == "视频":
            tags.append(_display_value(item))
        elif key == "正文":
            body = "\n".join(str(entry) for entry in item) if isinstance(item, list) else _display_value(item)
            tags.append(f"正文: {body}")
        else:
            rendered = ",".join(str(entry) for entry in item) if isinstance(item, list) else _display_value(item)
            tags.append(f"{key}: {rendered}")
    return tags


def _dimensions(value: Any) -> list[dict[str, Any]]:
    rows = value if isinstance(value, list) else [value] if isinstance(value, dict) else []
    return [
        {
            "type": row.get("维度类型") or row.get("dimension_type"),
            "name": row.get("维度") or row.get("dimension"),
            "value": row.get("原子点") or row.get("value"),
            **({"id": row.get("id")} if row.get("id") is not None else {}),
        }
        for row in rows
        if isinstance(row, dict)
    ]


def _description_parts(value: str, element_names: dict[int, str]) -> list[dict[str, Any]]:
    parts: list[dict[str, Any]] = []
    cursor = 0
    for match in _ELEMENT_REFERENCE.finditer(value):
        if match.start() > cursor:
            parts.append({"type": "text", "text": value[cursor : match.start()]})
        element_id = int(match.group(2))
        name = element_names.get(element_id)
        parts.append(
            {
                "type": "element",
                "id": element_id,
                "name": name,
                "resolved": name is not None,
                "text": match.group(0),
            }
        )
        cursor = match.end()
        if name:
            duplicate = re.match(r"\s*" + re.escape(name), value[cursor:])
            if duplicate:
                cursor += duplicate.end()
    if cursor < len(value):
        parts.append({"type": "text", "text": value[cursor:]})
    return parts


def _display_value(value: Any) -> str:
    if value is True:
        return "包含"
    if value is False:
        return "不包含"
    if value is None:
        return ""
    return str(value)


def _as_int(value: Any) -> int | None:
    try:
        return int(value) if value is not None else None
    except (TypeError, ValueError):
        return None


def _sort_number(value: Any) -> tuple[int, str]:
    number = _as_int(value)
    return (0, str(number).zfill(12)) if number is not None else (1, str(value or ""))


def _unique_records(rows: list[dict[str, Any]], key: str) -> list[dict[str, Any]]:
    seen: set[Any] = set()
    result = []
    for row in rows:
        value = row.get(key)
        if value in seen:
            continue
        seen.add(value)
        result.append(row)
    return result


def _branch_detail(script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any]) -> dict[str, Any]:
    facts = _branch_facts(branch, bundle)
    return _payload(
        [
            _section("task", "实现任务", branch_task_text(branch.get("summary", {}).get("task"))),
            _section("data", "数据取舍", decisions_markdown([item.get("decision") for item in branch.get("dataDecisions") or []])),
            _section("output", "候选产出", branch.get("summary", {}).get("output")),
        ],
        {"scriptBuildId": script_build_id, "branch": raw},
        {
            "summary": branch.get("summary", {}).get("output"),
            "sections": [_section("facts", "领域事实", _domain_facts_markdown(facts))] if facts else [],
            "artifactRef": branch.get("output", {}).get("snapshotRef"),
            "exactness": _output_exactness(branch.get("output") or {}),
        }
        if branch.get("output", {}).get("node", {}).get("hasChanges")
        else None,
    )


def _branch_task_detail(
    script_build_id: int,
    branch: dict[str, Any],
    raw: Any,
    bundle: dict[str, Any],
    *,
    round_index: int,
    branch_id: int,
) -> dict[str, Any]:
    task_event = next(
        (
            event
            for event in reversed(bundle.get("events") or [])
            if str(event.get("event_type") or "") == "agent_invoke"
            and str(event.get("event_name") or "") == "script_implementer"
            and int(event.get("round_index") or 0) == round_index
            and int(event.get("branch_id") or 0) == branch_id
        ),
        None,
    )
    event_input = (task_event or {}).get("inputData")
    event_task = event_input.get("task") if isinstance(event_input, dict) else None
    raw_task = (raw or {}).get("impl_task") if isinstance(raw, dict) else None
    complete_task = branch_task_text(
        event_task
        or raw_task
        or branch.get("summary", {}).get("task")
    )
    task_sections = branch_task_sections(
        event_task
        or raw_task
        or branch.get("summary", {}).get("task")
    )
    visual_roles = {
        "goal": ("summary", "key-result"),
        "materials": ("report", "evidence"),
        "avoid": ("report", "notice"),
        "context": ("report", "evidence"),
    }
    return _payload(
        [
            _section(
                f"task-{section['kind']}-{index}",
                section["title"],
                section["content"],
                visual_roles.get(section["kind"], ("default", "section"))[0],
                visual_role=visual_roles.get(section["kind"], ("default", "section"))[1],
                presentation="document" if section["kind"] in {"materials", "constraints", "avoid", "context"} else "prose",
            )
            for index, section in enumerate(task_sections, 1)
        ] or [_section("task-notes-1", "任务说明", complete_task)],
        {
            "scriptBuildId": script_build_id,
            "branch": raw,
            "taskEvent": task_event,
            "taskSource": (
                "script_build_event.inputData.task"
                if event_task
                else "script_build_branch.impl_task"
                if raw_task
                else "execution-view summary"
            ),
        },
    )


def _branch_output_detail(script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any]) -> dict[str, Any]:
    output = branch.get("output") or {}
    facts = _branch_facts(branch, bundle)
    changes = {
        "summary": output.get("summary"),
        "sections": [_section("facts", "已核实领域事实", _domain_facts_markdown(facts))] if facts else [],
        "artifactRef": output.get("snapshotRef"),
        "exactness": _output_exactness(output),
    }
    return _payload(
        [
            _section("output", "形成了什么", output.get("summary")),
            _section("assessment", "实现者说明", assessment_text((raw or {}).get("self_assessment")), "report"),
        ],
        {"scriptBuildId": script_build_id, "branch": raw, "output": output},
        changes if output.get("node", {}).get("hasChanges") else None,
    )


def _retrieval_stage_detail(script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any]) -> dict[str, Any]:
    stage = branch.get("retrievalStage") or {}
    direct = stage.get("directToolGroups") or []
    agents = stage.get("agentRuns") or []
    return _payload(
        [
            _section(
                "purpose",
                "取数目的",
                branch_task_text(branch.get("summary", {}).get("task"))
                or "为当前实现任务补齐可用的上游信息。",
            ),
            _section("mode", "执行方式", _retrieval_mode_text(stage.get("observedMode"))),
            _section("waves", "执行波次", _retrieval_waves_markdown(stage.get("waves") or [])),
            _section("direct", "工具取数", _direct_groups_markdown(direct)),
            _section("agents", "Agent 取数", _agent_runs_markdown(agents)),
            _section("result", "阶段结果与失败", _retrieval_stage_result(stage)),
        ],
        {
            "scriptBuildId": script_build_id,
            "branch": raw,
            "retrievalStage": stage,
            "note": "取数和初筛来自结构化运行事件；正式数据取舍来自独立业务表，不在本详情内反推。",
        },
    )


def _node_detail(script_build_id: int, node: dict[str, Any], technical: Any) -> dict[str, Any]:
    fields = []
    primary = node.get("card", {}).get("primary")
    if primary:
        fields.append(_section("primary", primary.get("label") or node.get("title"), primary.get("value")))
    for line in node.get("card", {}).get("secondary") or []:
        fields.append(_section(str(line.get("key")), line.get("label") or "补充说明", line.get("value")))
    return _payload(fields, {"scriptBuildId": script_build_id, "record": technical})


def _round_result_detail(
    script_build_id: int,
    round_: dict[str, Any],
    raw_round: dict[str, Any] | None,
    bundle: dict[str, Any],
) -> dict[str, Any]:
    result = round_.get("result") or {}
    branches = round_.get("branches") or []
    round_index = int(round_.get("roundIndex") or 0)
    facts = [
        item
        for item in bundle.get("domainInfo") or []
        if int(item.get("round_index") or 0) == round_index
    ]
    evaluation = round_.get("overallEvaluation") or {}
    next_step = next(
        (
            line.get("value")
            for line in (result.get("card") or {}).get("secondary") or []
            if line.get("key") == "next"
        ),
        None,
    )
    sections = [
        _section(
            "goal",
            "本轮目标完成情况",
            "\n".join(
                part
                for part in (
                    business_text((raw_round or {}).get("goal")),
                    f"收口状态：{result.get('status') or round_.get('state') or '未记录'}",
                )
                if part
            ),
        ),
        _section("outcomes", "方案处置", _branch_outcomes_markdown(branches)),
        _section("facts", "领域事实", _domain_facts_markdown(facts) or "本轮没有新增领域事实记录。"),
        _section(
            "evaluation",
            "整体评审",
            _node_value(evaluation) or "未找到明确的本轮整体评审结论。",
        ),
        _section(
            "changes",
            "主脚本变化",
            _round_content_changes(branches),
        ),
        _section("next", "下一步", next_step or "下一步未记录。"),
    ]
    changes = None
    if result.get("hasChanges"):
        changes = {
            "summary": _node_value(result),
            "sections": [
                _section("content", "内容方案", _round_content_changes(branches)),
                _section("domain", "领域信息", f"本轮独立保存 {len(facts)} 条领域事实。"),
            ],
            "artifactRef": "artifact:base:current",
            "exactness": "current-only",
        }
    return _payload(
        sections,
        {
            "scriptBuildId": script_build_id,
            "round": raw_round,
            "resultProjection": result,
            "note": "本轮主脚本历史快照无法完整还原，产物入口指向当前主脚本。",
        },
        changes,
    )


def _final_result_detail(
    script_build_id: int, view: dict[str, Any], bundle: dict[str, Any]
) -> dict[str, Any]:
    final = view.get("finalResult") or {}
    record = bundle.get("record") or {}
    branches = final.get("branchCounts") or {}
    artifact = final.get("artifact") or {}
    status = str(final.get("status") or record.get("status") or "unknown")
    conclusion = {
        "success": "构建已完成。",
        "completed": "构建已完成。",
        "partial": "构建部分完成。",
        "failed": "构建失败。",
        "stopped": "构建已停止。",
        "running": "构建仍在进行。",
    }.get(status.lower(), f"构建状态：{status}。")
    summary_or_error = final.get("summary")
    if final.get("errorMessage"):
        summary_or_error = f"失败原因：{final['errorMessage']}"
    sections = [
        _section("conclusion", "构建结论", conclusion, "summary"),
        _section("summary", "构建摘要或失败原因", summary_or_error or "未另外记录构建摘要。"),
        _section(
            "statistics",
            "轮次与方案统计",
            (
                f"共 {int(final.get('roundCount') or 0)} 轮，"
                f"{int(branches.get('total') or 0)} 个方案；"
                f"采用 {int(branches.get('merged') or 0)}，"
                f"暂存 {int(branches.get('parked') or 0)}，"
                f"未采用 {int(branches.get('discarded') or 0)}，"
                f"待处理 {int(branches.get('open') or 0)}。"
            ),
        ),
        _section(
            "artifact",
            "最终主脚本规模",
            (
                f"{int(artifact.get('paragraphCount') or 0)} 个段落、"
                f"{int(artifact.get('elementCount') or 0)} 个元素、"
                f"{int(artifact.get('linkCount') or 0)} 条关联。"
            ),
        ),
        _section(
            "version",
            "版本说明",
            "当前只能确认数据库保存的最新主脚本，不将它伪装成某一轮结束时的历史快照。",
        ),
    ]
    return _payload(
        sections,
        {
            "scriptBuildId": script_build_id,
            "table": "script_build_record",
            "record": record,
            "finalResultProjection": final,
        },
        {
            "summary": "打开当前主脚本。",
            "sections": [],
            "artifactRef": artifact.get("snapshotRef") or "artifact:base:current",
            "exactness": "current-only",
        },
    )


def _find_decision_projection(
    view: dict[str, Any], detail_ref: str
) -> dict[str, Any] | None:
    candidates: list[dict[str, Any]] = []
    objective = view.get("objective") or {}
    if isinstance(objective.get("decision"), dict):
        candidates.append(objective["decision"])
    for round_ in view.get("rounds") or []:
        for node in (
            round_.get("goal"),
            (round_.get("plan") or {}).get("node"),
            round_.get("overallEvaluation"),
        ):
            if isinstance(node, dict) and isinstance(node.get("decision"), dict):
                candidates.append(node["decision"])
        for branch in round_.get("branches") or []:
            for item in branch.get("dataDecisions") or []:
                projection = item.get("decisionProjection")
                if isinstance(projection, dict):
                    candidates.append(projection)
            output_node = (branch.get("output") or {}).get("node") or {}
            if isinstance(output_node.get("decision"), dict):
                candidates.append(output_node["decision"])
        for batch in round_.get("convergenceBatches") or []:
            review = batch.get("review") or {}
            decision = batch.get("decision") or {}
            if isinstance(review.get("decisionProjection"), dict):
                candidates.append(review["decisionProjection"])
            if isinstance(decision.get("decisionProjection"), dict):
                candidates.append(decision["decisionProjection"])
    return next(
        (
            item
            for item in candidates
            if detail_ref in {item.get("id"), item.get("detailRef")}
        ),
        None,
    )


def _find_missing_convergence(
    view: dict[str, Any], detail_ref: str
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    for round_ in view.get("rounds") or []:
        for batch in round_.get("convergenceBatches") or []:
            for key in ("missingReview", "missingDecision"):
                node = batch.get(key)
                if isinstance(node, dict) and detail_ref in {
                    node.get("id"),
                    node.get("detailRef"),
                }:
                    return node, batch
    return None


def _decision_detail_payload(
    script_build_id: int,
    decision: dict[str, Any],
    bundle: dict[str, Any],
    activity_id: str,
    event_detail: dict[str, Any] | None = None,
) -> dict[str, Any]:
    detail = decision.get("detail") if isinstance(decision.get("detail"), dict) else {}
    blocks = [
        block
        for block in detail.get("blocks") or []
        if isinstance(block, dict)
    ]
    for notice in decision.get("notices") or []:
        if not isinstance(notice, dict) or not notice.get("message"):
            continue
        blocks.append(
            {
                "type": "notice",
                "level": "warning" if "missing" in str(notice.get("code")) else "info",
                "title": "记录说明",
                "value": notice["message"],
                "visualRole": "notice",
                "presentation": "prose",
            }
        )
    event_refs = [
        ref
        for ref in [
            *(decision.get("eventRefs") or []),
            *(decision.get("technicalRefs") or []),
        ]
        if isinstance(ref, str) and ref.startswith("event:")
    ]
    event_refs = list(dict.fromkeys(event_refs))
    technical: dict[str, Any] = {
        "source": {
            "scriptBuildId": script_build_id,
            "detailRef": activity_id,
            "technicalRefs": event_refs,
        },
        "association": {
            "completeness": decision.get("completeness"),
            "notices": decision.get("notices") or [],
        },
        "rawDecisionProjection": decision,
        "rawBusinessRecord": _decision_raw_record(bundle, activity_id),
    }
    if event_refs:
        technical["runtimeEvents"] = _events_for_ids(
            bundle, [_suffix_int(ref) for ref in event_refs]
        )
    if event_detail:
        technical.update(_event_technical_groups(event_detail))
    payload: dict[str, Any] = {
        "detailKind": "agent-decision",
        "decisionType": decision.get("decisionType"),
        "actor": decision.get("actor"),
        "authority": decision.get("authority"),
        "question": decision.get("question"),
        "promptRef": decision.get("promptRef"),
        "blocks": sanitize(blocks, max_text=None),
        "technical": sanitize(technical, max_text=None),
    }
    artifact_ref = decision.get("artifactRef")
    if artifact_ref:
        payload["changes"] = {
            "summary": (decision.get("body") or {}).get("output")
            or "已形成可查看的候选产出。",
            "sections": [],
            "artifactRef": artifact_ref,
            "exactness": "historical-snapshot",
        }
    return payload


def _event_technical_groups(event: dict[str, Any]) -> dict[str, Any]:
    """Split a lazily loaded event body into the Inspector's technical groups.

    The execution bundle only carries bounded previews.  Once a user opens an
    event-backed decision, the full (sanitized) body belongs here rather than
    being silently replaced by those previews.
    """

    metadata = {
        key: value
        for key, value in event.items()
        if key not in {"input", "output", "duration_ms"}
    }
    groups: dict[str, Any] = {
        "input": event.get("input"),
        "output": event.get("output"),
        "rawEvent": metadata,
    }
    if event.get("duration_ms") is not None:
        groups["cost"] = {"durationMs": event.get("duration_ms")}
    return groups


def _decision_raw_record(bundle: dict[str, Any], activity_id: str) -> Any:
    if activity_id.startswith("data-decision:"):
        return _find_by_id(bundle.get("dataDecisions") or [], _suffix_int(activity_id))
    if activity_id.startswith("multipath-decision:"):
        return _find_by_id(
            bundle.get("multipathDecisions") or [], _suffix_int(activity_id)
        )
    if activity_id.startswith("event:"):
        return _event_by_id(bundle, _suffix_int(activity_id))
    if activity_id.startswith("creative:"):
        event_id = _suffix_int(activity_id)
        event = _event_by_id(bundle, event_id)
        branch_id = int((event or {}).get("branch_id") or 0)
        branch = next(
            (
                item
                for item in bundle.get("branches") or []
                if int(item.get("branch_id") or 0) == branch_id
            ),
            None,
        )
        return {
            "anchorEvent": event,
            "candidateSnapshot": (branch or {}).get("candidate_snapshot"),
        }
    if activity_id == "run:objective":
        return bundle.get("record")
    round_index, _, role = _activity_parts(activity_id)
    if role in {"goal", "plan"}:
        return next(
            (
                item
                for item in bundle.get("rounds") or []
                if int(item.get("round_index") or 0) == int(round_index or 0)
            ),
            None,
        )
    return None


def _payload(sections: list[dict[str, Any] | None], technical: Any, changes: dict[str, Any] | None = None) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "businessSections": [item for item in sections if item],
        "technical": sanitize(technical, max_text=None),
    }
    if changes:
        changes["sections"] = [item for item in changes.get("sections", []) if item]
        payload["changes"] = sanitize(changes, max_text=None)
    return payload


def _section(
    section_id: str,
    title: str,
    content: Any,
    variant: str = "default",
    *,
    visual_role: str | None = None,
    presentation: str = "prose",
    collapsible: bool = False,
    default_open: bool = False,
) -> dict[str, Any] | None:
    if content is None or content == "" or content == []:
        return None
    if not isinstance(content, str):
        return None
    content = business_text(content)
    if not content:
        return None
    return {
        "id": section_id,
        "title": title,
        "content": sanitize(content, max_text=None),
        "variant": variant,
        "visualRole": visual_role or ("key-result" if variant == "summary" else "section"),
        "presentation": presentation,
        "collapsible": collapsible,
        "defaultOpen": default_open,
    }


def _items_section(
    section_id: str,
    title: str,
    items: list[dict[str, Any] | None],
    *,
    fallback: Any = None,
    variant: str = "default",
    visual_role: str | None = None,
    collapsible: bool = False,
    default_open: bool = False,
) -> dict[str, Any] | None:
    values = [item for item in items if item]
    if not values and fallback in (None, "", []):
        return None
    content = "\n".join(
        f"{item.get('label')}：{item.get('value')}" for item in values
    ) or str(fallback)
    section = _section(
        section_id,
        title,
        content,
        variant,
        visual_role=visual_role,
        presentation="list",
        collapsible=collapsible,
        default_open=default_open,
    )
    if section is not None and values:
        section["items"] = sanitize(values, max_text=None)
    return section


def _detail_item(label: Any, value: Any) -> dict[str, Any] | None:
    if value in (None, "", []):
        return None
    return {
        "label": sanitize(business_text(str(label)), max_text=None),
        "value": sanitize(business_text(str(value)), max_text=None),
    }


def _activity_parts(activity_id: str) -> tuple[int | None, int | None, str | None]:
    parts = activity_id.split(":")
    if len(parts) < 2 or parts[0] != "round":
        return None, None, None
    try:
        round_index = int(parts[1])
    except ValueError:
        return None, None, None
    if len(parts) >= 4 and parts[2] == "branch":
        try:
            branch_id = int(parts[3])
        except ValueError:
            return round_index, None, None
        return round_index, branch_id, parts[4] if len(parts) > 4 else None
    return round_index, None, parts[2] if len(parts) > 2 else None


def _round_node(round_: dict[str, Any], role: str) -> dict[str, Any] | None:
    if role == "goal":
        return round_.get("goal")
    if role == "plan":
        return (round_.get("plan") or {}).get("node")
    if role == "result":
        return round_.get("result")
    if role == "evaluation":
        return round_.get("overallEvaluation")
    return None


def _branch_node(branch: dict[str, Any], role: str) -> dict[str, Any] | None:
    if role == "task":
        return branch.get("task")
    if role == "output":
        return branch.get("output", {}).get("node")
    return None


def _convergence_markdown(batches: list[dict[str, Any]]) -> str:
    if not batches:
        return "未找到候选评审或主 Agent 多路决策记录。"
    rows: list[str] = []
    for index, batch in enumerate(batches, 1):
        review = batch.get("review") or {}
        decision = batch.get("decision") or {}
        branch_ids = batch.get("branchIds") or []
        rows.append(
            f"### 第 {index} 批 · "
            + ("、".join(f"方案 {value}" for value in branch_ids) or "方案范围未记录")
        )
        rows.append(
            f"**多方案评审：** {review.get('recommendation') or review.get('comparison') or review.get('conclusion') or '评审结论未记录'}"
        )
        rows.append(
            f"**主 Agent 决策：** {business_text(decision.get('decision')) or '决策尚未记录'}"
        )
    return "\n\n".join(rows)


def _find_retrieval_item(view: dict[str, Any], collection: str, item_id: str) -> dict[str, Any] | None:
    for round_ in view.get("rounds") or []:
        for branch in round_.get("branches") or []:
            for item in (branch.get("retrievalStage") or {}).get(collection) or []:
                if item.get("id") == item_id:
                    return item
    return None


def _event_by_id(bundle: dict[str, Any], event_id: Any) -> dict[str, Any] | None:
    return next(
        (item for item in bundle.get("events") or [] if int(item.get("id") or 0) == int(event_id or 0)),
        None,
    )


def _events_for_ids(bundle: dict[str, Any], event_ids: list[Any]) -> list[dict[str, Any]]:
    wanted = {int(value) for value in event_ids if value is not None}
    return [item for item in bundle.get("events") or [] if int(item.get("id") or 0) in wanted]


def _event_nested_value(
    event: dict[str, Any] | None, side: str, key: str
) -> Any:
    if not isinstance(event, dict):
        return None
    aliases = (
        ("inputData", "input_data")
        if side == "input"
        else ("agentOutputData", "outputData", "output_data")
    )
    for alias in aliases:
        value = event.get(alias)
        if isinstance(value, dict) and value.get(key) not in (None, ""):
            return value.get(key)
    wrapped = event.get(side)
    if isinstance(wrapped, dict):
        content = wrapped.get("content")
        if isinstance(content, dict):
            return content.get(key)
        if key == "summary" and isinstance(content, str):
            return content
    return None


def _direct_sources_markdown(group: dict[str, Any]) -> str:
    rows = group.get("sources") or []
    return "\n".join(
        f"- {item.get('businessLabel') or '相关信息'}：{int(item.get('callCount') or 0)} 次"
        for item in rows
    ) or "未记录具体读取项"


def _direct_result_markdown(calls: list[dict[str, Any]]) -> str:
    success = sum(1 for item in calls if item.get("status") == "success")
    failure = sum(1 for item in calls if item.get("status") == "failure")
    running = sum(1 for item in calls if item.get("status") == "running")
    interrupted = sum(1 for item in calls if item.get("status") == "interrupted")
    return f"共 {len(calls)} 次读取：{success} 次完成，{failure} 次失败，{running} 次进行中，{interrupted} 次因构建中断。"


def _agent_query_markdown(run: dict[str, Any]) -> str:
    summary = run.get("querySummary") or {}
    total = int(summary.get("total") or 0)
    if total == 0 and not (run.get("attempts") or []):
        return "Agent 已完成，没有发起查询。"
    header = (
        f"共 {total} 次查询："
        f"{int(summary.get('hit') or 0)} 次命中，"
        f"{int(summary.get('empty') or 0)} 次无结果，"
        f"{int(summary.get('failure') or 0)} 次失败，"
        f"{int(summary.get('interrupted') or 0)} 次因构建中断。"
    )
    rows = []
    for index, attempt in enumerate(run.get("attempts") or [], 1):
        query = attempt.get("queryLabel") or attempt.get("queryPreview") or attempt.get("businessLabel") or "查询"
        label = {"hit": "命中", "empty": "无结果", "failure": "失败", "running": "进行中", "interrupted": "构建中断", "unknown": "结果数不明"}.get(attempt.get("status"), "结果数不明")
        count = attempt.get("resultCount")
        suffix = f"· {count} 条" if isinstance(count, int) and count > 0 else f"· {label}"
        rows.append(f"{index:02d}. {query} {suffix}")
    return "\n".join([header, *rows])


def _agent_hit_markdown(run: dict[str, Any]) -> str:
    attempts = run.get("attempts") or []
    hits = [item for item in attempts if item.get("status") == "hit"]
    empty = [item for item in attempts if item.get("status") == "empty"]
    failures = [
        item
        for item in attempts
        if item.get("status") in {"failure", "interrupted"}
    ]
    rows = [
        f"- {item.get('queryLabel') or '查询'}：{int(item.get('resultCount') or 0)} 条"
        for item in hits
    ]
    rows.append(f"成功命中 {len(hits)} 次，成功为空 {len(empty)} 次，技术失败或中断 {len(failures)} 次。")
    return "\n".join(rows)


def _screening_fallback(state: Any) -> str:
    return {"running": "Agent 仍在整理查询结果。", "missing": "未找到 Agent 的初筛整理记录。"}.get(str(state), "初筛整理未记录。")


def _retrieval_mode_text(mode: Any) -> str:
    return {
        "single": "本次只有一个取数操作。",
        "sequential": "各项取数依次执行。",
        "parallel": "各项取数在同一时间段并行执行。",
        "mixed": "本次同时存在并行取数和后续取数。",
        "unknown": "运行时间记录不足，无法确认执行先后。",
    }.get(str(mode), "执行方式未记录。")


def _retrieval_waves_markdown(waves: list[dict[str, Any]]) -> str:
    if not waves:
        return "运行时间或并发信息不足，无法安全还原执行波次。"
    rows = []
    for index, wave in enumerate(waves, 1):
        operations = wave.get("operations") or wave.get("operationIds") or []
        rows.append(
            f"- 第 {index} 波：{len(operations)} 个取数操作"
        )
    return "\n".join(rows)


def _retrieval_stage_result(stage: dict[str, Any]) -> str:
    direct_calls = [
        call
        for group in stage.get("directToolGroups") or []
        for call in group.get("calls") or []
    ]
    attempts = [
        attempt
        for run in stage.get("agentRuns") or []
        for attempt in run.get("attempts") or []
    ]
    failures = sum(
        1
        for item in [*direct_calls, *attempts]
        if item.get("status") in {"failure", "interrupted"}
    )
    hits = sum(1 for item in attempts if item.get("status") == "hit")
    empty = sum(1 for item in attempts if item.get("status") == "empty")
    return (
        f"阶段状态：{stage.get('status') or '未记录'}。"
        f"工具读取 {len(direct_calls)} 次，Agent 查询 {len(attempts)} 次；"
        f"命中 {hits} 次，成功为空 {empty} 次，失败或中断 {failures} 次。"
    )


def _direct_groups_markdown(groups: list[dict[str, Any]]) -> str:
    if not groups:
        return "本次未记录实现 Agent 直接读取工具。"
    return "\n".join(
        f"- 第 {index} 组：{'、'.join(item.get('businessLabel') or '相关信息' for item in group.get('sources') or [])}（{int(group.get('callCount') or 0)} 次）"
        for index, group in enumerate(groups, 1)
    )


def _agent_runs_markdown(runs: list[dict[str, Any]]) -> str:
    if not runs:
        return "本次未记录委派取数 Agent。"
    return "\n".join(
        f"- {item.get('businessLabel') or '取数 Agent'}：{int((item.get('querySummary') or {}).get('total') or 0)} 次查询，初筛{'已记录' if (item.get('screening') or {}).get('state') == 'available' else '未完整记录'}"
        for item in runs
    )


def _find_round(view: dict[str, Any], round_index: int | None) -> dict[str, Any] | None:
    return next((item for item in view.get("rounds") or [] if int(item.get("roundIndex") or 0) == int(round_index or 0)), None)


def _find_by_id(rows: list[dict[str, Any]], record_id: int) -> dict[str, Any] | None:
    return next((item for item in rows if int(item.get("id") or 0) == record_id), None)


def _suffix_int(value: str) -> int:
    try:
        return int(value.rsplit(":", 1)[1])
    except (IndexError, ValueError) as exc:
        raise KeyError(value) from exc


def _node_value(node: dict[str, Any] | None) -> str | None:
    return ((node or {}).get("card") or {}).get("primary", {}).get("value")


def _plan_text(plan: dict[str, Any]) -> str:
    paths = plan.get("paths") or []
    parts = [f"方式：{plan.get('mode') or '未记录'}", f"当前保存 {len(paths)} 个方案"]
    if plan.get("note"):
        parts.append(f"说明：{plan['note']}")
    revisions = int(plan.get("runtimeRevisionCount") or 0)
    if revisions > 1:
        parts.append(f"运行事件中找到 {revisions} 次规划记录；历史版本只在技术详情中查看。")
    return "\n".join(parts)


def _round_content_changes(branches: list[dict[str, Any]]) -> str:
    merged = [item for item in branches if item.get("pathType") != "领域信息" and item.get("status") == "merged"]
    return f"确认采用 {len(merged)} 个内容方案；无法据此还原本轮结束时的完整主脚本快照。"


def _branch_outcomes_markdown(branches: list[dict[str, Any]]) -> str:
    if not branches:
        return "本轮没有可展示的方案处置记录。"
    rows = []
    for item in branches:
        outcome = item.get("outcome") or {}
        line = (
            f"- 方案 {item.get('branchId')}："
            f"{outcome.get('label') or item.get('status') or '状态未记录'}"
        )
        if outcome.get("reasoning"):
            line += f"。{business_text(outcome.get('reasoning'))}"
        rows.append(line)
    return "\n".join(rows)


def _round_domain_changes(branches: list[dict[str, Any]]) -> str:
    count = sum(int(item.get("output", {}).get("factCount") or 0) for item in branches if item.get("pathType") == "领域信息")
    return f"新增 {count} 条独立保存的领域事实。"


def _branch_facts(branch: dict[str, Any], bundle: dict[str, Any]) -> list[dict[str, Any]]:
    refs = set(branch.get("output", {}).get("factRefs") or [])
    return [item for item in bundle.get("domainInfo") or [] if f"domain-info:{item.get('id')}" in refs]


def _output_exactness(output: dict[str, Any]) -> str:
    if output.get("type") == "domain-info":
        return "exact"
    historical = output.get("accuracy", {}).get("historical")
    return "historical-snapshot" if historical == "exact" else "current-only" if output.get("snapshotRef") else "unknown"


def _artifact_exactness(artifact: dict[str, Any]) -> str:
    return "historical-snapshot" if artifact.get("historicalAccuracy") == "exact" else "current-only"


def _domain_facts_markdown(rows: list[dict[str, Any]]) -> str:
    values = [business_text(row.get("content")) for row in rows]
    return "\n".join(f"- {value}" for value in values if value)


def _source_value(value: Any) -> str:
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        return source_markdown(value)
    if isinstance(value, dict):
        return source_markdown([value]) or business_text(value.get("url") or value.get("name"))
    return ""