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

from __future__ import annotations

import re
from collections import Counter, defaultdict
from datetime import datetime, timezone
from typing import Any
from zoneinfo import ZoneInfo

from .business_detail_text import branch_task_sections, branch_task_text, business_text
from .decision_projection import AgentDecisionProjector
from .main_decision_text import (
    parse_goal_text,
    parse_objective_text,
    plan_mode_label,
    plan_path_summary,
    preview as decision_preview,
)
from .runtime_event_projection import RuntimeEventProjector


_DECISIONS = AgentDecisionProjector()


TERMINAL_STATUSES = {
    "success",
    "partial",
    "failed",
    "stopped",
    "completed",
    "cancelled",
    "cancel",
    "cancle",
}


class ExecutionViewBuilder:
    """Deterministic V8 business narrative projector.

    The projector consumes plain dictionaries and has no database, HTTP or
    framework dependencies.  Business rows win over runtime events; events may
    only fill clearly labelled evaluation/activity detail.
    """

    def __init__(self, event_projector: RuntimeEventProjector | None = None):
        self._event_projector = event_projector or RuntimeEventProjector()

    def from_bundle(self, bundle: dict[str, Any]) -> dict[str, Any]:
        record = bundle.get("record") or {}
        raw_rounds = sorted(
            bundle.get("rounds") or [],
            key=lambda item: (_integer(item.get("round_index")) or 0, _integer(item.get("id")) or 0),
        )
        raw_branches = sorted(
            bundle.get("branches") or [],
            key=lambda item: (
                _integer(item.get("round_index")) or 0,
                _integer(item.get("branch_id")) or 0,
            ),
        )
        data_decisions = [
            item
            for item in bundle.get("dataDecisions") or []
            if (_integer(item.get("branch_id")) or 0) > 0
        ]
        multipath_decisions = bundle.get("multipathDecisions") or []
        domain_info = bundle.get("domainInfo") or []
        events = bundle.get("events") or []

        valid_rounds = {
            value
            for value in (_integer(item.get("round_index")) for item in raw_rounds)
            if value is not None
        }
        valid_branches = {
            value
            for value in (_integer(item.get("branch_id")) for item in raw_branches)
            if value is not None
        }
        captured_at = datetime.now(timezone.utc)
        runtime = self._event_projector.project(
            events,
            valid_rounds=valid_rounds,
            valid_branches=valid_branches,
            captured_at=captured_at,
            run_status=str(record.get("status") or "unknown"),
            script_direction=record.get("script_direction"),
            rounds=raw_rounds,
            multipath_decisions=multipath_decisions,
        )
        implementer_tasks = _implementer_tasks_by_branch(events)

        decisions_by_branch: dict[int, list[dict[str, Any]]] = defaultdict(list)
        for item in data_decisions:
            branch_id = _integer(item.get("branch_id"))
            if branch_id is not None:
                decisions_by_branch[branch_id].append(item)
        domain_by_branch: dict[int, list[dict[str, Any]]] = defaultdict(list)
        domain_by_round: dict[int, list[dict[str, Any]]] = defaultdict(list)
        for item in domain_info:
            branch_id = _integer(item.get("branch_id"))
            round_index = _integer(item.get("round_index"))
            if branch_id is not None:
                domain_by_branch[branch_id].append(item)
            if round_index is not None:
                domain_by_round[round_index].append(item)
        multipath_by_round: dict[int, list[dict[str, Any]]] = defaultdict(list)
        unassigned_business: list[dict[str, Any]] = []
        for item in multipath_decisions:
            round_index = _integer(item.get("round_index"))
            if round_index in valid_rounds:
                multipath_by_round[round_index].append(item)
            else:
                unassigned_business.append(
                    {
                        "type": "multipath-decision",
                        "id": item.get("id"),
                        "reason": "没有可确认的所属轮次",
                    }
                )

        projected_rounds: list[dict[str, Any]] = []
        for index, raw_round in enumerate(raw_rounds):
            round_index = _integer(raw_round.get("round_index")) or index + 1
            branch_rows = [
                item
                for item in raw_branches
                if _integer(item.get("round_index")) == round_index
            ]
            next_round = raw_rounds[index + 1] if index + 1 < len(raw_rounds) else None
            projected_rounds.append(
                self._round_story(
                    record,
                    raw_round,
                    branch_rows,
                    decisions_by_branch,
                    domain_by_branch,
                    domain_by_round.get(round_index, []),
                    multipath_by_round.get(round_index, []),
                    runtime,
                    implementer_tasks,
                    next_round,
                )
            )

        data_shape = _data_shape(
            raw_branches, multipath_decisions, domain_info, events
        )
        warnings = _warnings(
            data_shape=data_shape,
            ignored_legacy_count=_integer(
                bundle.get("ignoredLegacyDataDecisionCount")
            )
            or 0,
            rounds=projected_rounds,
            runtime_unassigned=runtime.get("unassigned") or [],
            business_unassigned=unassigned_business,
        )
        artifact = bundle.get("currentArtifact") or {}
        current_round = projected_rounds[-1]["roundIndex"] if projected_rounds else None
        view = {
            "schemaVersion": "8",
            "capturedAt": captured_at.isoformat(),
            "dataShape": data_shape,
            "header": _header(record, current_round),
            "planning": _planning_analysis(
                (runtime.get("mainAgentDecisions") or {}).get("planningAnalysis"),
            ),
            "objective": _objective(
                record,
                (runtime.get("mainAgentDecisions") or {}).get("objective"),
            ),
            "rounds": projected_rounds,
            "finalResult": _final_result(record, projected_rounds, artifact),
            "unassigned": {
                "runtimeEvents": runtime.get("unassigned") or [],
                "businessRecords": unassigned_business,
            },
            "warnings": warnings,
        }
        if view["planning"] is None:
            view.pop("planning")
        return view

    def _round_story(
        self,
        record: dict[str, Any],
        raw_round: dict[str, Any],
        branch_rows: list[dict[str, Any]],
        decisions_by_branch: dict[int, list[dict[str, Any]]],
        domain_by_branch: dict[int, list[dict[str, Any]]],
        round_domain_info: list[dict[str, Any]],
        multipath_rows: list[dict[str, Any]],
        runtime: dict[str, Any],
        implementer_tasks: dict[int, str],
        next_round: dict[str, Any] | None,
    ) -> dict[str, Any]:
        round_index = _integer(raw_round.get("round_index")) or 0
        main_decisions = runtime.get("mainAgentDecisions") or {}
        goal_context = (main_decisions.get("roundGoalsByRound") or {}).get(round_index)
        plan_context = (main_decisions.get("implementationPlansByRound") or {}).get(
            round_index
        )
        plan = _plan_projection(
            raw_round,
            plan_context,
        )
        branches = [
            _branch_story(
                round_index,
                row,
                decisions_by_branch.get(_integer(row.get("branch_id")) or 0, []),
                domain_by_branch.get(_integer(row.get("branch_id")) or 0, []),
                runtime.get("retrievalStagesByBranch", {}).get(
                    _integer(row.get("branch_id")) or 0
                ),
                runtime.get("creativeByBranch", {}).get(
                    _integer(row.get("branch_id")) or 0
                ),
                implementer_tasks.get(_integer(row.get("branch_id")) or 0),
                str(record.get("status") or "unknown"),
            )
            for row in branch_rows
        ]
        branch_ids = {branch["branchId"] for branch in branches}
        decision_batches = [
            _decision_batch(round_index, row, branch_ids, branches)
            for row in sorted(multipath_rows, key=_record_order)
        ]
        convergence_batches, unassigned_reviews = _convergence_batches(
            round_index,
            runtime.get("multipathReviewsByRound", {}).get(round_index, []),
            decision_batches,
            str(record.get("status") or "unknown"),
        )
        if not convergence_batches and branches:
            convergence_batches = [
                _convergence_batch(
                    round_index,
                    None,
                    None,
                    "records-missing",
                    str(record.get("status") or "unknown"),
                    branch_ids=sorted(branch_ids),
                )
            ]
        _annotate_unscoped_round_reviews(
            convergence_batches,
            runtime.get("unassigned") or [],
            round_index,
        )
        if unassigned_reviews:
            runtime.setdefault("unassigned", []).extend(unassigned_reviews)
        overall_evaluation = _round_evaluation(
            round_index,
            runtime.get("overallEvaluationsByRound", {}).get(round_index, []),
            str(record.get("status") or "unknown"),
        )
        state = _round_state(record, raw_round, overall_evaluation)
        result = _round_result(
            record,
            round_index,
            branches,
            round_domain_info,
            state,
            next_round,
        )
        goal_summary = _goal_summary(raw_round.get("goal"))
        round_summary = _round_summary(
            round_index=round_index,
            goal=goal_summary,
            plan=plan,
            branches=branches,
            batches=decision_batches,
            domain_info=round_domain_info,
            run_status=str(record.get("status") or "unknown"),
        )
        goal_parsed = parse_goal_text(raw_round.get("goal"))
        goal_lines: list[dict[str, Any]] = []
        focus_items = goal_parsed.get("focusItems") or []
        if focus_items:
            focus_summary = "；".join(str(item) for item in focus_items[:2])
            if len(focus_items) > 2:
                focus_summary += f"；另有 {len(focus_items) - 2} 项，详情查看"
            goal_lines.append(_line("focus", "本轮聚焦", focus_summary))
        visible_inputs = _context_input_labels(goal_context)
        if visible_inputs:
            goal_lines.append(_line("visible-before", "形成前可见", " · ".join(visible_inputs)))
        goal_decision = _direction_decision(
            id=f"round:{round_index}:goal",
            subtype="round-goal",
            context=goal_context,
            decision_items=(goal_context or {}).get("decisionItems")
            or [goal_parsed["headline"]],
            primary_label="要解决",
            primary_value=goal_parsed["headline"],
            secondary=goal_lines,
            prompt_ref="prompt:main",
        )
        return {
            "id": f"round:{round_index}",
            "roundIndex": round_index,
            "state": state,
            "summary": round_summary,
            "goal": _story(
                id=f"round:{round_index}:goal",
                role="goal",
                kind="decision",
                title="本轮目标",
                label="要解决",
                value=goal_parsed["headline"],
                secondary=goal_lines,
                detail_ref=f"round:{round_index}:goal",
                source_notice=None if raw_round.get("goal") else "record-missing",
                decision=goal_decision,
            ),
            "plan": plan,
            "branches": branches,
            "convergenceBatches": convergence_batches,
            "overallEvaluation": overall_evaluation,
            "result": result,
            "detailRef": f"round:{round_index}",
        }


def _header(record: dict[str, Any], current_round: int | None) -> dict[str, Any]:
    agent_config = record.get("agent_config")
    model = None
    if isinstance(agent_config, dict):
        model = agent_config.get("model_name") or agent_config.get("model")
    return {
        "id": str(record.get("id") or ""),
        "title": f"脚本构建 #{record.get('id')}",
        "status": str(record.get("status") or "unknown"),
        "summary": _plain(record.get("summary")),
        "errorMessage": _plain(record.get("error_message")),
        "model": model,
        "totalCost": record.get("cost_usd"),
        "currentRound": current_round,
        "createdAt": record.get("start_time"),
        "updatedAt": record.get("end_time"),
        "direction": {
            "currentValue": _plain(record.get("script_direction")),
            "note": "这是数据库当前保存的创作方向。",
        },
    }


def _objective(
    record: dict[str, Any], decision_context: dict[str, Any] | None
) -> dict[str, Any]:
    raw_direction = record.get("script_direction")
    direction = _plain(raw_direction)
    # Section parsing needs the persisted Markdown line breaks. ``direction``
    # is only the compact truthy/display fallback used by the surrounding view.
    parsed = parse_objective_text(raw_direction)
    secondary: list[dict[str, Any]] = []
    input_labels = _context_input_labels(decision_context, relation="direct-read")
    if input_labels:
        secondary.append(
            _line("inputs", "决策前读取", " · ".join(input_labels))
        )
    structure: list[str] = []
    target_count = int(parsed.get("targetCount") or 0)
    dimension_count = int(parsed.get("domainDimensionCount") or 0)
    if target_count:
        structure.append(f"{target_count} 个创作目标")
    if dimension_count:
        structure.append(f"{dimension_count} 个领域评估维度")
    if structure:
        secondary.append(_line("structure", "目标结构", " · ".join(structure)))
    decision = _direction_decision(
        id="run:objective",
        subtype="objective",
        context=decision_context,
        decision_items=(decision_context or {}).get("decisionItems")
        or ([parsed.get("headline")] if parsed.get("headline") else []),
        primary_label="核心目标",
        primary_value=parsed.get("headline"),
        secondary=secondary,
        prompt_ref="prompt:main",
    )
    return _story(
        id="run:objective",
        role="objective",
        kind="decision" if direction else "missing",
        title="创作目标",
        label="核心目标",
        value=parsed.get("headline") or "创作方向未记录",
        secondary=secondary,
        detail_ref="run:objective",
        source_notice=None if direction else "record-missing",
        decision=decision,
    )


def _planning_analysis(context: dict[str, Any] | None) -> dict[str, Any] | None:
    if not context or context.get("completeness") != "complete" or not context.get("eventRef"):
        return None
    secondary = [
        _line("plan", "执行计划", context.get("plan")),
        _line("next", "下一步", context.get("action")),
    ]
    return _story(
        id="run:planning-analysis",
        role="planning-analysis",
        kind="execution",
        title="创作规划解析",
        label="规划概要",
        value=context.get("summary") or "主 Agent 已完成首次创作规划",
        secondary=secondary,
        detail_ref=context.get("eventRef"),
        status="completed",
    )


def _plan_projection(
    raw_round: dict[str, Any], decision_context: dict[str, Any] | None
) -> dict[str, Any]:
    round_index = _integer(raw_round.get("round_index")) or 0
    paths = raw_round.get("multipath_plan")
    if not isinstance(paths, list):
        paths = []
    raw_mode = _plain(raw_round.get("race_or_divide"))
    mode = plan_mode_label(raw_mode, len(paths))
    summary = (
        f"{mode or '方式未记录'} · {len(paths)} 个方案"
        if paths
        else "多路规划未记录"
    )
    decision = _direction_decision(
        id=f"round:{round_index}:plan",
        subtype="implementation-plan",
        context=decision_context,
        decision_items=(decision_context or {}).get("decisionItems")
        or [summary],
        primary_label="实现方式",
        primary_value=summary,
        secondary=_plan_card_lines(paths, raw_round.get("plan_note")),
        prompt_ref="prompt:main",
    )
    node = _story(
        id=f"round:{round_index}:plan",
        role="plan",
        kind="decision" if paths else "missing",
        title="实现规划",
        label="实现方式",
        value=summary,
        secondary=_plan_card_lines(paths, raw_round.get("plan_note")),
        detail_ref=f"round:{round_index}:plan",
        source_notice=None if paths else "record-missing",
        decision=decision,
    )
    return {
        "mode": mode,
        "rawMode": raw_mode,
        "paths": paths,
        "note": _plain(raw_round.get("plan_note")),
        "currentOnly": True,
        "runtimeRevisionCount": len((decision_context or {}).get("revisionRefs") or []),
        "runtimeRevisionRefs": list((decision_context or {}).get("revisionRefs") or []),
        "node": node,
    }


def _branch_story(
    round_index: int,
    row: dict[str, Any],
    decisions: list[dict[str, Any]],
    domain_info: list[dict[str, Any]],
    retrieval_stage: dict[str, Any] | None,
    creative_decision: dict[str, Any] | None,
    implementer_task: str | None,
    run_status: str,
) -> dict[str, Any]:
    branch_id = _integer(row.get("branch_id")) or 0
    path_type = str(row.get("path_type") or "unknown")
    if path_type not in {"内容", "领域信息"}:
        path_type = "unknown"
    task = (
        branch_task_text(implementer_task)
        or branch_task_text(row.get("impl_task"))
        or business_text(row.get("target"))
    )
    retrieval_stage = retrieval_stage or _missing_retrieval_stage(
        round_index, branch_id, run_status
    )
    projected_data_decisions = [
        _DECISIONS.data_tradeoff(
            {
                **item,
                "decision": business_text(item.get("decision")),
                "reasoning": business_text(item.get("reasoning")),
                "id": f"data-decision:{item.get('id')}",
                "detailRef": f"data-decision:{item.get('id')}",
                "promptRef": _implementer_prompt_ref(retrieval_stage),
            }
        )
        for item in decisions
    ]
    for item, projection in zip(decisions, projected_data_decisions):
        full_conclusion = business_text(item.get("decision"))
        if full_conclusion:
            projection.setdefault("card", {}).setdefault("primary", {})[
                "value"
            ] = full_conclusion
    decision_nodes = [
        _story(
            id=f"data-decision:{item.get('id')}",
            role="data-decision",
            kind="decision",
            title="数据取舍",
            label="取舍结论",
            value=projection["card"]["primary"]["value"],
            secondary=projection["card"]["secondary"],
            detail_ref=f"data-decision:{item.get('id')}",
            decision=projection,
        )
        for item, projection in zip(decisions, projected_data_decisions)
    ]
    output = _branch_output(row, path_type, domain_info)
    if creative_decision and output.get("type") == "script-artifact":
        creative_decision = dict(creative_decision)
        explicit_basis = [
            {
                "label": business_text(item.get("decision")) or "已记录一项数据取舍",
                "value": business_text(item.get("reasoning")) or None,
                "detailRef": f"data-decision:{item.get('id')}",
            }
            for item in decisions
            if item.get("decision")
        ]
        creative_decision["artifactRef"] = output.get("snapshotRef")
        creative_decision["body"] = {
            **(creative_decision.get("body") or {}),
            "output": output.get("summary"),
            "explicitBasis": explicit_basis,
            "candidateSnapshot": {
                "snapshotRef": output.get("snapshotRef"),
                "paragraphCount": output.get("paragraphCount"),
                "elementCount": output.get("elementCount"),
                "linkCount": output.get("linkCount"),
                "exactness": _candidate_output_exactness(output),
            },
        }
        detail = creative_decision.get("detail")
        if isinstance(detail, dict):
            blocks = [
                block
                for block in detail.get("blocks") or []
                if not (
                    isinstance(block, dict)
                    and block.get("title") in {"实际产出", "产出的候选表"}
                )
            ]
            if explicit_basis:
                blocks.insert(
                    0,
                    {
                        "type": "items",
                        "title": "明确采用的数据取舍",
                        "items": explicit_basis,
                        "visualRole": "evidence",
                        "presentation": "list",
                        "collapsible": True,
                    },
                )
            blocks.append(
                {
                    "type": "summary",
                    "title": "产出的候选表",
                    "value": output.get("summary"),
                    "visualRole": "key-result",
                    "presentation": "prose",
                }
            )
            creative_decision["detail"] = {**detail, "blocks": blocks}
        secondary = list((creative_decision.get("card") or {}).get("secondary") or [])
        if explicit_basis:
            secondary.append(_line("basis", "采用数据", f"{len(explicit_basis)} 项明确取舍"))
        secondary.append(_line("output", "产出的候选表", output.get("summary")))
        creative_decision["card"] = {
            **(creative_decision.get("card") or {}),
            "secondary": secondary[:3],
        }
        output["node"] = _story(
            id=creative_decision["id"],
            role="candidate-output",
            kind="decision",
            title="产生候选创作表",
            label="创作处理",
            value=creative_decision["card"]["primary"]["value"],
            secondary=creative_decision["card"]["secondary"],
            detail_ref=creative_decision["detailRef"],
            artifact_ref=output.get("snapshotRef"),
            has_changes=bool(output.get("snapshotRef")),
            decision=creative_decision,
        )
    status = str(row.get("status") or "open")
    disposition_context = _branch_disposition_context(status, run_status)
    title = _plain(row.get("target")) or f"方案 {branch_id}"
    return {
        "id": f"round:{round_index}:branch:{branch_id}",
        "branchId": branch_id,
        "pathType": path_type,
        "status": status,
        "title": title,
        "summary": {
            "task": task or "实现任务未记录",
            "output": output["summary"],
        },
        "task": _branch_task_story(
            id=f"round:{round_index}:branch:{branch_id}:task",
            task=task,
            detail_ref=f"round:{round_index}:branch:{branch_id}:task",
        ),
        "retrievalStage": retrieval_stage,
        "dataDecisions": [
            {
                "id": f"data-decision:{item.get('id')}",
                "recordId": item.get("id"),
                "decision": _plain(item.get("decision")),
                "reasoning": _plain(item.get("reasoning")),
                "sources": item.get("sources") or [],
                "createdAt": item.get("created_at"),
                "node": node,
                "decisionProjection": projection,
            }
            for item, node, projection in zip(
                decisions, decision_nodes, projected_data_decisions
            )
        ],
        "output": output,
        "outcome": {
            "status": status,
            "label": disposition_context["label"],
            "reasoning": _plain(row.get("decision_reasoning")),
            "sourceNotice": disposition_context["sourceNotice"],
        },
        "detailRef": f"round:{round_index}:branch:{branch_id}",
    }


def _branch_output(
    row: dict[str, Any], path_type: str, domain_info: list[dict[str, Any]]
) -> dict[str, Any]:
    branch_id = _integer(row.get("branch_id")) or 0
    round_index = _integer(row.get("round_index")) or 0
    if path_type == "领域信息":
        facts = [
            item
            for item in domain_info
            if _integer(item.get("branch_id")) == branch_id
            and _integer(item.get("round_index")) in {None, round_index}
        ]
        summary = (
            f"新增 {len(facts)} 条已核实领域事实"
            if facts
            else "已查询完成，本方案没有新增领域事实"
        )
        node = _story(
            id=f"round:{round_index}:branch:{branch_id}:output",
            role="domain-output",
            kind="result",
            title="领域事实",
            label="产出",
            value=summary,
            detail_ref=f"round:{round_index}:branch:{branch_id}:output",
            has_changes=bool(facts),
            source_notice=None,
        )
        return {
            "type": "domain-info",
            "summary": summary,
            "factCount": len(facts),
            "factRefs": [f"domain-info:{item.get('id')}" for item in facts],
            "node": node,
        }

    candidate = row.get("candidate_snapshot") or {}
    snapshot = candidate.get("snapshot") if isinstance(candidate, dict) else None
    counts = _snapshot_counts(snapshot)
    if any(counts.values()):
        summary = (
            f"候选脚本包含 {counts['paragraphs']} 个段落、"
            f"{counts['elements']} 个元素"
        )
    elif row.get("self_assessment"):
        assessment = business_text(row.get("self_assessment"))
        first_item = next(
            (line.strip(" -*") for line in assessment.splitlines() if line.strip()),
            assessment,
        )
        summary = f"实现 Agent 自评：{first_item}"
    else:
        summary = "候选脚本内容未记录"
    has_output = bool(snapshot) or bool(row.get("self_assessment"))
    node = _story(
        id=f"round:{round_index}:branch:{branch_id}:output",
        role="candidate-output",
        kind="result" if has_output else "missing",
        title="候选产出",
        label="产出",
        value=summary,
        detail_ref=f"round:{round_index}:branch:{branch_id}:output",
        has_changes=bool(snapshot),
        source_notice=None if has_output else "record-missing",
        artifact_ref=f"artifact:branch:{branch_id}" if snapshot else None,
    )
    return {
        "type": "script-artifact" if has_output else "missing",
        "summary": summary,
        "snapshotRef": f"artifact:branch:{branch_id}" if snapshot else None,
        "paragraphCount": counts["paragraphs"],
        "elementCount": counts["elements"],
        "linkCount": counts["links"],
        "accuracy": {
            "historical": candidate.get("historicalAccuracy")
            if isinstance(candidate, dict)
            else "unknown",
            "currentProjection": candidate.get("currentProjectionAccuracy")
            if isinstance(candidate, dict)
            else "unknown",
        },
        "node": node,
    }


def _branch_task_story(
    *, id: str, task: str | None, detail_ref: str
) -> dict[str, Any]:
    if not task:
        return _story(
            id=id,
            role="branch-task",
            kind="missing",
            title="实现任务",
            label="实现任务",
            value="实现任务未记录",
            detail_ref=detail_ref,
            source_notice="record-missing",
        )
    sections = branch_task_sections(task)
    preferred = next(
        (
            item
            for kind in ("scope", "method", "goal", "notes")
            for item in sections
            if item.get("kind") == kind and item.get("content")
        ),
        None,
    )
    primary = (preferred or {}).get("content") or task
    secondary = [
        _line(str(item.get("kind") or index), item.get("title") or "任务说明", item.get("content"))
        for index, item in enumerate(sections, 1)
        if item is not preferred and item.get("content")
    ][:3]
    node = _story(
        id=id,
        role="branch-task",
        kind="execution",
        title="实现任务",
        label=(preferred or {}).get("title") or "实现任务",
        value=primary,
        secondary=secondary,
        detail_ref=detail_ref,
    )
    node["taskStructure"] = {
        "sectionCount": len(sections),
        "sectionKinds": [item.get("kind") for item in sections],
    }
    return node


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


def _missing_retrieval_stage(
    round_index: int, branch_id: int, run_status: str
) -> dict[str, Any]:
    running = str(run_status or "").lower() not in TERMINAL_STATUSES
    return {
        "id": f"round:{round_index}:branch:{branch_id}:retrieval",
        "roundIndex": round_index,
        "branchId": branch_id,
        "implementerEventId": None,
        "observedMode": "unknown",
        "waves": [],
        "directToolGroups": [],
        "agentRuns": [],
        "status": "running" if running else "missing",
        "startedAt": None,
        "endedAt": None,
        "detailRef": f"round:{round_index}:branch:{branch_id}:retrieval",
        "sourceNotice": None if running else "record-missing",
    }


def _decision_batch(
    round_index: int,
    row: dict[str, Any],
    valid_branch_ids: set[int],
    branches: list[dict[str, Any]],
) -> dict[str, Any]:
    branch_ids = _ids(row.get("branch_ids"))
    missing = [branch_id for branch_id in branch_ids if branch_id not in valid_branch_ids]
    record_id = row.get("id")
    decision = _plain(row.get("decision")) or "决策内容未记录"
    display_decision = _business_terms(decision)
    secondary = [
        _line(
            "branches",
            "比较方案",
            "、".join(f"方案 {branch_id}" for branch_id in branch_ids)
            or "涉及方案未记录",
        )
    ]
    if row.get("reasoning"):
        secondary.append(
            _line("reason", "理由", _preview(row.get("reasoning"), 120))
        )
    branch_by_id = {branch["branchId"]: branch for branch in branches}
    outcomes = [
        {
            "branchId": branch_id,
            "status": branch_by_id.get(branch_id, {}).get("outcome", {}).get("status", "unknown"),
            "label": branch_by_id.get(branch_id, {}).get("outcome", {}).get("label", "状态未记录"),
            "reasoning": branch_by_id.get(branch_id, {}).get("outcome", {}).get("reasoning"),
        }
        for branch_id in branch_ids
    ]
    if outcomes:
        secondary.insert(
            1,
            _line(
                "outcomes",
                "各方案结果",
                "；".join(
                    f"方案 {item['branchId']} · {item['label']}" for item in outcomes
                ),
            ),
        )
    projection = _DECISIONS.multipath_tradeoff(
        {
            "id": f"multipath-decision:{record_id}",
            "detailRef": f"multipath-decision:{record_id}",
            "promptRef": "prompt:main",
            "branchIds": branch_ids,
            "scopeItems": [f"方案 {branch_id}" for branch_id in branch_ids],
            "decision": display_decision,
            "reasoning": business_text(row.get("reasoning")),
            "selected": [
                f"方案 {item['branchId']}"
                for item in outcomes
                if item["status"] == "merged"
            ],
            "parked": [
                f"方案 {item['branchId']}"
                for item in outcomes
                if item["status"] == "parked"
            ],
            "rejected": [
                f"方案 {item['branchId']}"
                for item in outcomes
                if item["status"] == "discarded"
            ],
            "completeness": "partial" if missing or not branch_ids else "complete",
        }
    )
    projection.setdefault("card", {}).setdefault("primary", {})[
        "value"
    ] = display_decision
    return {
        "id": f"multipath-decision:{record_id}",
        "recordId": record_id,
        "roundIndex": round_index,
        "branchIds": branch_ids,
        "decision": decision,
        "reasoning": _plain(row.get("reasoning")),
        "createdAt": row.get("created_at"),
        "completeness": "partial" if missing or not branch_ids else "complete",
        "missingBranchIds": missing,
        "branchOutcomes": outcomes,
        "decisionProjection": projection,
        "node": _story(
            id=f"multipath-decision:{record_id}",
            role="multipath-decision",
            kind="decision",
            title="主 Agent 多路决策",
            label="决定",
            value=projection["card"]["primary"]["value"],
            secondary=projection["card"]["secondary"],
            detail_ref=f"multipath-decision:{record_id}",
            source_notice="record-missing" if missing else None,
            decision=projection,
        ),
    }


def _convergence_batches(
    round_index: int,
    reviews: list[dict[str, Any]],
    decisions: list[dict[str, Any]],
    run_status: str,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    ordered_reviews = sorted(reviews, key=_runtime_order)
    ordered_decisions = sorted(decisions, key=_decision_order)
    unmatched_reviews = set(range(len(ordered_reviews)))
    batches: list[dict[str, Any]] = []

    for decision in ordered_decisions:
        decision_set = set(decision.get("branchIds") or [])
        decision_time = _date_value(decision.get("createdAt"))
        candidates = [
            index
            for index in unmatched_reviews
            if set(ordered_reviews[index].get("branchIds") or []) == decision_set
            and _not_after(ordered_reviews[index].get("endedAt"), decision_time)
        ]
        review_index = max(
            candidates,
            key=lambda index: _runtime_order(ordered_reviews[index]),
            default=None,
        )
        review = ordered_reviews[review_index] if review_index is not None else None
        if review_index is not None:
            unmatched_reviews.remove(review_index)
        batches.append(
            _convergence_batch(
                round_index,
                review,
                decision,
                "exact-branch-set-and-time" if review else "decision-only",
                run_status,
            )
        )

    for index in sorted(unmatched_reviews, key=lambda value: _runtime_order(ordered_reviews[value])):
        review = ordered_reviews[index]
        batches.append(
            _convergence_batch(
                round_index, review, None, "review-only", run_status
            )
        )

    batches.sort(key=_convergence_order)
    return batches, []


def _annotate_unscoped_round_reviews(
    batches: list[dict[str, Any]],
    unassigned: list[dict[str, Any]],
    round_index: int,
) -> None:
    """Keep unscoped reviews visible without fabricating a candidate link."""

    reviews = [
        item
        for item in unassigned
        if str(item.get("name") or "") == "script_multipath_evaluator"
        and _integer(item.get("roundIndex")) == round_index
        and str(item.get("association") or "") == "missing-branch-scope"
    ]
    if not reviews:
        return
    refs = [item.get("detailRef") for item in reviews if item.get("detailRef")]
    for batch in batches:
        missing = batch.get("missingReview")
        if not isinstance(missing, dict):
            continue
        missing["card"] = {
            "primary": {
                "label": "评审关联",
                "value": "发现同轮评审，但无法安全关联到具体候选",
            },
            "secondary": [
                _line("records", "未关联记录", f"{len(refs)} 条评审记录")
            ],
        }
        missing["sourceNotice"] = "association-incomplete"
        missing["unscopedReviewRefs"] = refs


def _convergence_batch(
    round_index: int,
    review: dict[str, Any] | None,
    decision: dict[str, Any] | None,
    association: str,
    run_status: str,
    branch_ids: list[int] | None = None,
) -> dict[str, Any]:
    branch_ids = branch_ids or list(
        dict.fromkeys(
            (review or {}).get("branchIds")
            or (decision or {}).get("branchIds")
            or []
        )
    )
    suffix = (
        f"review-{review.get('eventId')}"
        if review
        else f"decision-{decision.get('recordId')}"
        if decision
        else "records-missing"
    )
    missing_context = _missing_multipath_context(run_status)
    if review is not None and not review.get("decisionProjection"):
        review = dict(review)
        review["decisionProjection"] = _evaluation_decision(
            review,
            subtype="multipath-evaluation",
            detail_ref=review.get("detailRef"),
            prompt_ref=review.get("promptRef"),
        )
    missing_review = None
    if review is None:
        missing_review_id = f"round:{round_index}:{suffix}:review-missing"
        missing_review = _story(
            id=missing_review_id,
            role="multipath-review",
            kind=missing_context["kind"],
            title="多方案评审",
            label="评审记录",
            value="未找到可与本次决策安全关联的多方案评审记录",
            detail_ref=missing_review_id,
            source_notice=missing_context["sourceNotice"],
            status=missing_context.get("status"),
        )
    missing_decision = None
    if decision is None:
        missing_decision_id = f"round:{round_index}:{suffix}:decision-missing"
        missing_decision = _story(
            id=missing_decision_id,
            role="multipath-decision",
            kind=missing_context["kind"],
            title="主 Agent 多路决策",
            label=missing_context["label"],
            value=missing_context["value"],
            detail_ref=missing_decision_id,
            source_notice=missing_context["sourceNotice"],
            status=missing_context.get("status"),
        )
    if review is not None and decision is not None:
        _attach_review_comparison(review, decision, association)
    return {
        "id": f"round:{round_index}:convergence:{suffix}",
        "roundIndex": round_index,
        "branchIds": branch_ids,
        "review": review,
        "decision": decision,
        "missingReview": missing_review,
        "missingDecision": missing_decision,
        "association": association,
    }


def _round_evaluation(
    round_index: int, evaluations: list[dict[str, Any]], run_status: str
) -> dict[str, Any]:
    if not evaluations:
        context = _missing_evaluation_context(run_status)
        status = context.get("status")
        if not status:
            status = (
                "not-evaluated"
                if context["sourceNotice"] == "record-missing"
                else "unknown"
            )
        return _story(
            id=f"round:{round_index}:evaluation",
            role="round-evaluation",
            kind=context["kind"],
            title="主脚本整体评审",
            label=context["label"],
            value=context["value"],
            detail_ref=f"round:{round_index}:evaluation",
            source_notice=context["sourceNotice"],
            status=status,
        )
    latest = evaluations[-1]
    conclusion = _plain(latest.get("conclusion"))
    projection = _evaluation_decision(
        latest,
        subtype="overall-evaluation",
        detail_ref=latest.get("detailRef"),
        prompt_ref=latest.get("promptRef"),
    )
    return _story(
        id=f"round:{round_index}:evaluation",
        role="round-evaluation",
        kind="decision" if conclusion else "missing",
        title="主脚本整体评审",
        label="评审结论",
        value=projection["card"]["primary"]["value"],
        secondary=projection["card"]["secondary"],
        detail_ref=latest.get("detailRef"),
        source_notice="runtime-associated",
        status=_evaluation_state(conclusion),
        decision=projection,
    )


def _evaluation_decision(
    evaluation: dict[str, Any],
    *,
    subtype: str,
    detail_ref: Any,
    prompt_ref: Any,
) -> dict[str, Any]:
    projection = _DECISIONS.evaluation(
        {
            "id": evaluation.get("id") or detail_ref,
            "detailRef": detail_ref,
            "promptRef": prompt_ref,
            "subtype": subtype,
            "subjects": evaluation.get("subjects") or [
                f"方案 {value}" for value in evaluation.get("branchIds") or []
            ],
            "criteria": evaluation.get("criteria") or [],
            "itemConclusions": evaluation.get("itemConclusions")
            or evaluation.get("branchConclusions")
            or [],
            "branchEvaluations": evaluation.get("branchEvaluations") or [],
            "comparisonRows": evaluation.get("comparisonRows") or [],
            "achievements": evaluation.get("achievements") or [],
            "problems": evaluation.get("problems") or [],
            "conclusion": evaluation.get("conclusion"),
            "recommendation": evaluation.get("recommendation")
            or evaluation.get("comparison"),
            "nextGoal": evaluation.get("nextGoal"),
            "completeness": "partial"
            if evaluation.get("notices")
            else "complete",
        }
    )
    full_conclusion = business_text(
        evaluation.get("conclusion")
        or evaluation.get("recommendation")
        or evaluation.get("comparison")
    )
    if full_conclusion:
        projection.setdefault("card", {}).setdefault("primary", {})[
            "value"
        ] = full_conclusion
    return projection


def _attach_review_comparison(
    review: dict[str, Any], decision: dict[str, Any], association: str
) -> None:
    projection = decision.get("decisionProjection")
    if not isinstance(projection, dict):
        return
    conclusions = {
        int(item.get("branchId") or 0): item
        for item in review.get("branchConclusions") or []
        if isinstance(item, dict)
    }
    outcomes = {
        int(item.get("branchId") or 0): item
        for item in decision.get("branchOutcomes") or []
        if isinstance(item, dict)
    }
    reasoning = business_text(decision.get("reasoning"))
    rows: list[dict[str, Any]] = []
    for branch_id in decision.get("branchIds") or []:
        row = {
            "candidate": f"方案 {branch_id}",
            "recommendation": (conclusions.get(branch_id) or {}).get("conclusion")
            or review.get("recommendation")
            or "评审未给出单独结论",
            "finalDecision": (outcomes.get(branch_id) or {}).get("label")
            or "最终状态未记录",
        }
        handling = _explicit_recommendation_handling(reasoning, branch_id)
        if handling:
            row["handling"] = handling
        rows.append(row)
    projection.setdefault("body", {})["reviewComparison"] = rows
    projection["body"]["reviewAssociation"] = association
    review_input = {
        "label": "同批候选方案的评审建议",
        "summary": business_text(
            review.get("recommendation") or review.get("comparison")
        ),
        "observedRelation": "returned-to-actor",
        "decisionUse": "not-recorded",
        "detailRef": review.get("detailRef"),
    }
    projection.setdefault("inputs", []).append(review_input)
    projection.setdefault("notices", []).append(
        {
            "code": "same-batch-association",
            "message": "评审与最终决定按同一批候选方案和发生顺序关联；系统没有保存直接的一对一关系。",
        }
    )
    if rows:
        blocks = projection.setdefault("detail", {}).setdefault("blocks", [])
        blocks.insert(
            0,
            {
                "type": "items",
                "title": "参与判断",
                "items": [
                    {
                        "label": review_input["label"],
                        "value": review_input["summary"],
                        "note": "可确认评审在最终决定前返回；是否采用应以下方的逐方案对照为准。",
                        "detailRef": review_input["detailRef"],
                    }
                ],
                "visualRole": "evidence",
                "presentation": "list",
                "collapsible": True,
            },
        )
        blocks.insert(
            3,
            {
                "type": "comparison",
                "title": "评审建议与最终决定",
                "rows": rows,
                "visualRole": "section",
                "presentation": "comparison",
            },
        )


def _explicit_recommendation_handling(reasoning: str, branch_id: int) -> str | None:
    if not reasoning:
        return None
    for sentence in re.split(r"[。；;\n]+", reasoning):
        if not re.search(rf"方案\s*{branch_id}\b", sentence):
            continue
        if any(
            token in sentence
            for token in (
                "采纳",
                "吸收",
                "保留",
                "延后",
                "转入",
                "下一轮",
                "转译",
                "不直接合并",
                "未采用",
                "拒绝",
                "不采纳",
            )
        ):
            return sentence.strip()
    return None


_DATABASE_TIMEZONE = ZoneInfo("Asia/Shanghai")


def _date_value(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, str) and value:
        try:
            parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        except ValueError:
            return None
    else:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=_DATABASE_TIMEZONE)
    return parsed.astimezone(timezone.utc)


def _not_after(value: Any, boundary: datetime | None) -> bool:
    current = _date_value(value)
    return boundary is None or current is None or current <= boundary


def _record_order(row: dict[str, Any]) -> tuple[str, int]:
    return (str(row.get("created_at") or ""), _integer(row.get("id")) or 0)


def _runtime_order(row: dict[str, Any]) -> tuple[str, int]:
    return (
        str(row.get("startedAt") or row.get("endedAt") or ""),
        _integer(row.get("eventId")) or 0,
    )


def _decision_order(row: dict[str, Any]) -> tuple[str, int]:
    return (str(row.get("createdAt") or ""), _integer(row.get("recordId")) or 0)


def _convergence_order(row: dict[str, Any]) -> tuple[str, int]:
    review = row.get("review") or {}
    decision = row.get("decision") or {}
    return (
        str(review.get("startedAt") or decision.get("createdAt") or ""),
        _integer(review.get("eventId") or decision.get("recordId")) or 0,
    )


def _round_result(
    record: dict[str, Any],
    round_index: int,
    branches: list[dict[str, Any]],
    domain_info: list[dict[str, Any]],
    state: str,
    next_round: dict[str, Any] | None,
) -> dict[str, Any]:
    merged_content = [
        branch
        for branch in branches
        if branch["pathType"] != "领域信息" and branch["status"] == "merged"
    ]
    script_summary = (
        f"确认采用 {len(merged_content)} 个内容方案"
        if merged_content
        else "没有可确认的内容方案合入记录"
    )
    domain_summary = (
        f"新增 {len(domain_info)} 条领域事实"
        if domain_info
        else "本轮没有新增领域事实记录"
    )
    if next_round:
        next_step = f"数据库已记录第 {next_round.get('round_index')} 轮"
    elif str(record.get("status") or "").lower() in TERMINAL_STATUSES:
        next_step = "构建已经结束"
    else:
        next_step = "等待本轮收口"
    primary = f"{script_summary}；{domain_summary}"
    return _story(
        id=f"round:{round_index}:result",
        role="round-result",
        kind="result",
        title="本轮产出",
        label="产出",
        value=primary,
        secondary=[_line("next", "下一步", next_step)],
        detail_ref=f"round:{round_index}:result",
        artifact_ref=f"artifact:round:{round_index}",
        has_changes=bool(merged_content or domain_info),
        status=state,
    )


def _round_state(
    record: dict[str, Any],
    raw_round: dict[str, Any],
    evaluation: dict[str, Any] | None,
) -> str:
    if evaluation and evaluation.get("status") not in {None, "unknown"}:
        return str(evaluation["status"])
    run_status = str(record.get("status") or "").lower()
    if str(raw_round.get("status") or "").lower() == "open" and run_status not in TERMINAL_STATUSES:
        return "running"
    return "unknown"


def _evaluation_state(conclusion: str | None) -> str:
    text = str(conclusion or "")
    if any(token in text for token in ("未通过", "不通过", "回退", "重试")):
        return "needs-retry"
    if "部分通过" in text:
        return "partially-passed"
    if "通过" in text:
        return "passed"
    return "unknown"


def _final_result(
    record: dict[str, Any], rounds: list[dict[str, Any]], artifact: dict[str, Any]
) -> dict[str, Any]:
    branches = [branch for round_ in rounds for branch in round_["branches"]]
    statuses = Counter(branch["status"] for branch in branches)
    counts = _snapshot_counts(artifact)
    status = str(record.get("status") or "unknown")
    summary = _plain(record.get("summary"))
    error = _plain(record.get("error_message"))
    primary = summary or error or _final_status_text(status)
    return {
        "id": "run:final-result",
        "role": "final-result",
        "kind": "result",
        "title": "最终结果",
        "status": status,
        "terminal": status.lower() in TERMINAL_STATUSES,
        "roundCount": len(rounds),
        "branchCounts": {
            "total": len(branches),
            "merged": statuses.get("merged", 0),
            "parked": statuses.get("parked", 0),
            "discarded": statuses.get("discarded", 0),
            "open": statuses.get("open", 0),
        },
        "summary": _plain(record.get("summary")),
        "errorMessage": _plain(record.get("error_message")),
        "artifact": {
            "snapshotRef": "artifact:base:current",
            "paragraphCount": counts["paragraphs"],
            "elementCount": counts["elements"],
            "linkCount": counts["links"],
            "historicalVersion": "current-only",
        },
        "card": {
            "primary": {"label": "构建结论", "value": primary},
            "secondary": [
                _line("rounds", "轮次", f"{len(rounds)} 轮"),
                _line(
                    "branches",
                    "方案处置",
                    f"采用 {statuses.get('merged', 0)} · 暂存 {statuses.get('parked', 0)} · 未采用 {statuses.get('discarded', 0)}",
                ),
                _line(
                    "artifact",
                    "主脚本规模",
                    f"{counts['paragraphs']} 段落 · {counts['elements']} 元素 · {counts['links']} 关联",
                ),
            ],
        },
        "artifactRef": "artifact:base:current",
        "detailRef": "run:final-result",
    }


def _final_status_text(status: str) -> str:
    return {
        "success": "构建已完成",
        "completed": "构建已完成",
        "partial": "构建部分完成",
        "failed": "构建失败",
        "stopped": "构建已停止",
        "cancelled": "构建已取消",
        "running": "构建进行中",
    }.get(str(status or "").lower(), "构建状态已记录")


def _story(
    *,
    id: str,
    role: str,
    kind: str,
    title: str,
    label: str,
    value: Any,
    secondary: list[dict[str, Any]] | None = None,
    detail_ref: str | None = None,
    has_changes: bool = False,
    source_notice: str | None = None,
    status: str | None = None,
    artifact_ref: str | None = None,
    decision: dict[str, Any] | None = None,
) -> dict[str, Any]:
    node = {
        "id": id,
        "role": role,
        "kind": kind,
        "title": title,
        "card": (decision or {}).get("card")
        or {
            # Callers pass a deliberately selected card field. Keep that field
            # intact; line clamping is a view concern, not a data projection.
            "primary": {"label": label, "value": _plain(value)},
            "secondary": [item for item in (secondary or []) if item.get("value")],
        },
        "hasChanges": bool(has_changes),
    }
    if detail_ref:
        node["detailRef"] = detail_ref
    if source_notice:
        node["sourceNotice"] = source_notice
    if status:
        node["status"] = status
    if artifact_ref:
        node["artifactRef"] = artifact_ref
    if decision:
        node["decision"] = decision
    return node


def _line(key: str, label: str, value: Any, tone: str = "normal") -> dict[str, Any]:
    return {"key": key, "label": label, "value": _plain(value), "tone": tone}


def _plan_card_lines(paths: list[Any], note: Any) -> list[dict[str, Any]]:
    lines: list[dict[str, Any]] = []
    if paths:
        route_values = [plan_path_summary(path, index) for index, path in enumerate(paths[:2], 1)]
        route_summary = "；".join(route_values)
        if len(paths) > 2:
            route_summary += f"；另有 {len(paths) - 2} 路，详情查看"
        lines.append(_line("routes", "各路任务", route_summary))
    reason = _plain(note)
    if reason:
        lines.append(_line("reason", "规划理由", reason))
    return lines[:2]


def _context_input_labels(
    context: dict[str, Any] | None, *, relation: str | None = None
) -> list[str]:
    result: list[str] = []
    for item in (context or {}).get("inputs") or []:
        if not isinstance(item, dict):
            continue
        if relation and item.get("relation") != relation:
            continue
        label = _plain(item.get("label"))
        if label and label not in result:
            result.append(label)
    return result[:3]


def _direction_decision(
    *,
    id: str,
    subtype: str,
    context: dict[str, Any] | None,
    decision_items: list[Any],
    primary_label: str,
    primary_value: Any,
    secondary: list[dict[str, Any]],
    prompt_ref: str,
) -> dict[str, Any]:
    context = context if isinstance(context, dict) else {}
    decision = _DECISIONS.direction(
        {
            "id": id,
            "detailRef": id,
            "promptRef": prompt_ref,
            "subtype": subtype,
            "decisionItems": [str(item) for item in decision_items if item],
            "explicitReasoning": context.get("explicitReasoning"),
            "constraints": context.get("constraints") or [],
            "inputs": context.get("inputs") or [],
            "sourceDocument": context.get("sourceDocument"),
            "implementationPlan": {
                "mode": context.get("displayMode"),
                "rawMode": context.get("rawMode"),
                "routes": context.get("routeItems") or [],
                "reasoning": context.get("explicitReasoning"),
            }
            if subtype == "implementation-plan"
            else None,
            "completeness": context.get("completeness"),
        }
    )
    decision["card"] = {
        "primary": {
            "label": primary_label,
            "value": _plain(primary_value) or "决策结论未记录",
        },
        "secondary": [item for item in secondary if item.get("value")][:3],
    }
    revisions = []
    for revision in context.get("revisions") or []:
        if not isinstance(revision, dict):
            continue
        paths = revision.get("paths") if isinstance(revision.get("paths"), list) else []
        parts = [str(revision.get("mode"))] if revision.get("mode") else []
        if paths:
            parts.append(f"{len(paths)} 路方案")
        revisions.append(
            {
                "current": bool(revision.get("current")),
                "detailRef": revision.get("detailRef"),
                "summary": " · ".join(parts) or "已保存一次版本",
            }
        )
    if revisions:
        decision["body"]["revisions"] = revisions
    decision["technicalRefs"] = list(context.get("technicalRefs") or [])
    return decision


def _implementer_prompt_ref(retrieval_stage: dict[str, Any] | None) -> str | None:
    event_id = _integer((retrieval_stage or {}).get("implementerEventId"))
    return f"prompt:event:{event_id}" if event_id is not None else None


def _snapshot_counts(value: Any) -> dict[str, int]:
    if isinstance(value, dict) and isinstance(value.get("snapshot"), dict):
        value = value["snapshot"]
    if not isinstance(value, dict):
        return {"paragraphs": 0, "elements": 0, "links": 0}
    stats = value.get("stats") if isinstance(value.get("stats"), dict) else {}
    if stats:
        return {
            "paragraphs": int(stats.get("paragraph_count") or 0),
            "elements": int(stats.get("element_count") or 0),
            "links": int(stats.get("link_count") or 0),
        }
    return {
        "paragraphs": len(value.get("paragraphs") or []),
        "elements": len(value.get("elements") or []),
        "links": len(value.get("paragraphElements") or value.get("paragraph_elements") or []),
    }


def _ids(value: Any) -> list[int]:
    if not isinstance(value, list):
        return []
    result: list[int] = []
    for item in value:
        number = _integer(item)
        if number is not None and number not in result:
            result.append(number)
    return result


def _implementer_tasks_by_branch(
    events: list[dict[str, Any]],
) -> dict[int, str]:
    """Return complete dispatch tasks when the lightweight event carries them.

    Historical bundles may only retain previews; callers deliberately keep the
    branch business record as a labelled fallback in that case.
    """

    result: dict[int, str] = {}
    for event in events:
        if (
            str(event.get("event_type") or "") != "agent_invoke"
            or str(event.get("event_name") or "") != "script_implementer"
        ):
            continue
        branch_id = _integer(event.get("branch_id"))
        if branch_id is None:
            continue
        data = event.get("inputData")
        task = data.get("task") if isinstance(data, dict) else None
        if not task:
            wrapped = event.get("input")
            content = wrapped.get("content") if isinstance(wrapped, dict) else None
            task = content.get("task") if isinstance(content, dict) else None
        text = branch_task_text(task)
        if text:
            result[branch_id] = text
    return result


def _round_summary(
    *,
    round_index: int,
    goal: dict[str, Any],
    plan: dict[str, Any],
    branches: list[dict[str, Any]],
    batches: list[dict[str, Any]],
    domain_info: list[dict[str, Any]],
    run_status: str,
) -> dict[str, Any]:
    paths = plan.get("paths") or []
    branch_ids = sorted(
        {
            branch_id
            for batch in batches
            for branch_id in batch.get("branchIds") or []
        }
    )
    decision_context = _missing_multipath_context(run_status)
    decision_summary = (
        "；".join(_preview(_business_terms(batch.get("decision")), 64) for batch in batches[:2])
        if batches
        else decision_context["value"]
    )
    if len(batches) > 2:
        decision_summary += f"；另有 {len(batches) - 2} 个决策批次"
    merged_content_count = sum(
        1
        for branch in branches
        if branch.get("pathType") != "领域信息" and branch.get("status") == "merged"
    )
    script_summary = (
        f"{merged_content_count} 个内容方案进入主脚本"
        if merged_content_count
        else "没有可确认的内容方案合入记录"
    )
    domain_summary = (
        f"新增 {len(domain_info)} 条领域事实"
        if domain_info
        else "没有新增领域事实记录"
    )
    return {
        "title": f"第 {round_index} 轮",
        "goal": goal,
        "approach": {
            "mode": plan.get("mode"),
            "candidateCount": len(branches) or len(paths),
            "plannedPathCount": len(paths),
            "contentBranchCount": sum(
                1 for branch in branches if branch.get("pathType") == "内容"
            ),
            "domainBranchCount": sum(
                1 for branch in branches if branch.get("pathType") == "领域信息"
            ),
            "summary": _plan_summary(plan),
        },
        "decision": {
            "batchCount": len(batches),
            "branchIds": branch_ids,
            "summary": decision_summary,
        },
        "output": {
            "mergedContentCount": merged_content_count,
            "domainFactCount": len(domain_info),
            "scriptSummary": script_summary,
            "domainSummary": domain_summary,
            "summary": f"{script_summary}；{domain_summary}",
        },
    }


def _goal_summary(value: Any) -> dict[str, Any]:
    parsed = parse_goal_text(value)
    if not parsed.get("raw"):
        return {
            "headline": "本轮目标未记录",
            "itemCount": 0,
            "previewItems": [],
            "truncated": False,
        }
    items = list(parsed.get("focusItems") or [])
    headline_source = str(parsed.get("headline") or parsed.get("raw"))
    headline = _preview(headline_source, 96)
    preview_items = [_preview(item, 72) for item in items[:3]]
    return {
        "headline": headline,
        "itemCount": len(items),
        "previewItems": preview_items,
        "truncated": bool(
            items
            or len(headline_source) > len(headline)
            or len(items) > len(preview_items)
        ),
    }


def _business_terms(value: Any) -> str:
    text = _plain(value) or "未记录"
    text = re.sub(r"(?i)\s*\bbranch\s*#?(\d+)\b\s*", r"方案 \1 ", text)
    text = re.sub(r"(?i)\s*\bbase\b", "主脚本", text)
    text = re.sub(r"\s+([，。；！？、：）】,.])", r"\1", text)
    return text


def _missing_multipath_context(run_status: str) -> dict[str, Any]:
    status = str(run_status or "unknown").lower()
    if status == "running":
        return {
            "kind": "execution",
            "label": "当前进度",
            "value": "等待主 Agent 比较候选方案",
            "sourceNotice": None,
            "status": "running",
        }
    interrupted = {
        "stopped": "构建停止前尚未形成多路决策",
        "cancelled": "构建取消前尚未形成多路决策",
        "cancel": "构建取消前尚未形成多路决策",
        "cancle": "构建取消前尚未形成多路决策",
        "failed": "构建失败前尚未形成多路决策",
    }
    if status in interrupted:
        return {
            "kind": "missing",
            "label": "流程进度",
            "value": interrupted[status],
            "sourceNotice": "process-incomplete",
        }
    return {
        "kind": "missing",
        "label": "记录状态",
        "value": "构建已经结束，但未找到主 Agent 多路决策记录",
        "sourceNotice": "record-missing",
    }


def _missing_evaluation_context(run_status: str) -> dict[str, Any]:
    status = str(run_status or "unknown").lower()
    if status == "running":
        return {
            "kind": "execution",
            "label": "当前进度",
            "value": "等待整体评审",
            "sourceNotice": None,
            "status": "running",
        }
    interrupted = {
        "stopped": "构建停止前尚未进行整体评审",
        "cancelled": "构建取消前尚未进行整体评审",
        "cancel": "构建取消前尚未进行整体评审",
        "cancle": "构建取消前尚未进行整体评审",
        "failed": "构建失败前尚未形成整体评审结论",
    }
    if status in interrupted:
        return {
            "kind": "missing",
            "label": "流程进度",
            "value": interrupted[status],
            "sourceNotice": "process-incomplete",
        }
    return {
        "kind": "missing",
        "label": "记录状态",
        "value": "构建已经结束，但未找到明确的整体评审结论",
        "sourceNotice": "record-missing",
    }


def _branch_disposition_context(branch_status: str, run_status: str) -> dict[str, Any]:
    if branch_status != "open":
        return {
            "kind": "decision",
            "label": _status_label(branch_status),
            "sourceNotice": None,
        }
    status = str(run_status or "unknown").lower()
    if status == "running":
        return {
            "kind": "execution",
            "label": "等待主 Agent 处置",
            "sourceNotice": None,
        }
    interrupted = {
        "stopped": "构建停止时尚未处置",
        "cancelled": "构建取消时尚未处置",
        "cancel": "构建取消时尚未处置",
        "cancle": "构建取消时尚未处置",
        "failed": "构建失败时尚未处置",
    }
    if status in interrupted:
        return {
            "kind": "missing",
            "label": interrupted[status],
            "sourceNotice": "process-incomplete",
        }
    return {
        "kind": "missing",
        "label": "构建结束时仍未处置",
        "sourceNotice": "record-missing",
    }


def _plan_summary(plan: dict[str, Any]) -> str:
    paths = plan.get("paths") or []
    if not paths:
        return "多路规划未记录"
    return f"{plan.get('mode') or '方式未记录'}：{len(paths)} 个候选方案"


def _status_label(status: str) -> str:
    return {
        "merged": "已采用",
        "parked": "暂存",
        "discarded": "未采用",
        "open": "尚未处置",
    }.get(status, status or "状态未记录")


def _data_shape(
    branches: list[dict[str, Any]],
    multipath_decisions: list[dict[str, Any]],
    domain_info: list[dict[str, Any]],
    events: list[dict[str, Any]],
) -> str:
    if (
        multipath_decisions
        or domain_info
        or events
        or any(branch.get("path_type") for branch in branches)
    ):
        return "current"
    return "legacy-incomplete"


def _warnings(
    *,
    data_shape: str,
    ignored_legacy_count: int,
    rounds: list[dict[str, Any]],
    runtime_unassigned: list[dict[str, Any]],
    business_unassigned: list[dict[str, Any]],
) -> list[dict[str, str]]:
    warnings: list[dict[str, str]] = []
    if data_shape == "legacy-incomplete":
        warnings.append(
            {
                "id": "legacy-shape",
                "level": "warning",
                "message": "这是旧结构记录，只展示新数据契约能够直接确认的内容。",
            }
        )
    if ignored_legacy_count:
        warnings.append(
            {
                "id": "ignored-legacy-data-decisions",
                "level": "info",
                "message": f"已忽略 {ignored_legacy_count} 条旧语义主决策记录。",
            }
        )
    missing_rounds = [
        round_["roundIndex"]
        for round_ in rounds
        if round_["branches"]
        and not any(
            batch.get("decision")
            for batch in round_.get("convergenceBatches") or []
        )
        and any(
            (batch.get("missingDecision") or {}).get("sourceNotice")
            == "record-missing"
            for batch in round_.get("convergenceBatches") or []
        )
    ]
    if missing_rounds:
        warnings.append(
            {
                "id": "missing-multipath-decisions",
                "level": "warning",
                "message": "第 " + "、".join(map(str, missing_rounds)) + " 轮未记录主 Agent 多路决策。",
            }
        )
    if runtime_unassigned:
        warnings.append(
            {
                "id": "unassigned-runtime-events",
                "level": "info",
                "message": f"有 {len(runtime_unassigned)} 条运行记录无法安全归属，已留在技术详情中。",
            }
        )
    if business_unassigned:
        warnings.append(
            {
                "id": "unassigned-business-records",
                "level": "warning",
                "message": f"有 {len(business_unassigned)} 条多路决策缺少可确认的轮次归属。",
            }
        )
    return warnings


def _plain(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).replace("\r\n", "\n").strip()
    if not text:
        return None
    text = re.sub(r"```[a-zA-Z0-9_-]*", "", text).replace("```", "")
    text = re.sub(r"^\s{0,3}#{1,6}\s*", "", text, flags=re.M)
    text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text)
    text = re.sub(r"(?m)^\s*[-*]\s+", "• ", text)
    return text.strip()


def _preview(value: Any, limit: int = 140) -> str:
    text = _plain(value) or "未记录"
    text = re.sub(r"\s+", " ", text)
    if len(text) <= limit:
        return text
    window = text[:limit]
    cut = max(window.rfind(mark) for mark in "。！？；")
    if cut >= max(30, limit // 2):
        return window[: cut + 1]
    return window.rstrip("，、；：,. ") + "…"


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