video-image-production-build/visualization/web/store.py

"""把本项目按 Run 落盘的业务文件投影成上游查看器使用的只读视图。"""

from __future__ import annotations

import hashlib
import json
import mimetypes
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable


PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_RUN_ROOT = PROJECT_ROOT / "demo_output"


def _load_json(path: Path, default: Any = None) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        return default


def _stable_id(value: str) -> int:
    """生成低于 JavaScript MAX_SAFE_INTEGER 的稳定数字 ID。"""

    return int(hashlib.sha256(value.encode("utf-8")).hexdigest()[:13], 16)


def _iso(timestamp: float | None) -> str | None:
    if timestamp is None:
        return None
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()


def _clip(value: Any, limit: int = 240) -> str:
    if value is None:
        return ""
    if not isinstance(value, str):
        value = json.dumps(value, ensure_ascii=False, default=str)
    return value if len(value) <= limit else value[: limit - 1] + "…"


def _viewer_status(value: Any) -> str:
    normalized = str(value or "").upper()
    if normalized in {"COMPLETED", "PASS", "SUCCESS", "SUCCEEDED"} or (
        normalized.endswith("_COMPLETED")
    ):
        return "success"
    if normalized in {"FAILED", "FAIL", "ERROR"}:
        return "failed"
    return "running"


def _files(root: Path, pattern: str) -> list[Path]:
    return sorted(root.glob(pattern))


def _mtime_bounds(root: Path) -> tuple[float | None, float | None]:
    values = [
        path.stat().st_mtime
        for path in root.rglob("*")
        if path.is_file() and path.name != ".run.lock"
    ]
    return (min(values), max(values)) if values else (None, None)


def _input_block(
    key: str,
    title: str,
    value: Any,
    *,
    source: str = "",
    images: Iterable[str] = (),
) -> dict[str, Any]:
    return {
        "key": key,
        "name": key,
        "title": title,
        "hint": source,
        "source": source,
        "filled": value not in (None, "", [], {}),
        "text": _clip(value, 1600),
        "images": list(images),
    }


def _node_card(
    index: int,
    *,
    node_type: str,
    label: str,
    summary: Any,
    ok: bool | None = None,
    detail: Any = None,
    tool_call_id: str = "",
) -> dict[str, Any]:
    return {
        "seq": index,
        "kind": "node",
        "fanout_group": "",
        "card": {
            "i": index,
            "type": node_type,
            "label": label,
            "summary": _clip(summary, 300),
            "ok": ok,
            "images": [],
            "branch": "",
            "node_id": None,
            "tool_call_id": tool_call_id,
            "io": None,
            "calls": [],
            "reasoning": "",
            "tokens": "",
            "detail": detail if detail is not None else summary,
        },
    }


@dataclass(frozen=True)
class RunRef:
    numeric_id: int
    key: str
    path: Path
    kind: str


class FileRunStore:
    """扫描 ``demo_output``，不读取或修改数据库、Checkpoint。"""

    def __init__(self, run_root: Path | None = None):
        self.run_root = (run_root or DEFAULT_RUN_ROOT).resolve()

    def discover(self) -> list[RunRef]:
        if not self.run_root.is_dir():
            return []
        refs: list[RunRef] = []
        for path in sorted(self.run_root.iterdir()):
            if not path.is_dir():
                continue
            kind = self._kind(path)
            if kind is None:
                continue
            refs.append(
                RunRef(
                    numeric_id=_stable_id(path.name),
                    key=path.name,
                    path=path.resolve(),
                    kind=kind,
                )
            )
        return sorted(
            refs,
            key=lambda item: _mtime_bounds(item.path)[1] or 0,
            reverse=True,
        )

    def get(self, numeric_id: int) -> RunRef:
        for ref in self.discover():
            if ref.numeric_id == numeric_id:
                return ref
        raise KeyError(numeric_id)

    @staticmethod
    def _kind(path: Path) -> str | None:
        if (path / "run_summary.json").is_file() or (
            path / "global_data_stage_delivery.json"
        ).is_file():
            return "global_data"
        if _files(path, "production_plans/production_dag.v*.json"):
            return "segment"
        return None

    def brief(self, ref: RunRef) -> dict[str, Any]:
        start, end = _mtime_bounds(ref.path)
        if ref.kind == "global_data":
            summary = _load_json(ref.path / "run_summary.json", {}) or {}
            stage_delivery = _load_json(
                ref.path / "global_data_stage_delivery.json", {}
            ) or {}
            metrics = _load_json(ref.path / "run_metrics.json", {}) or {}
            totals = metrics.get("model_totals") or {}
            invocation = metrics.get("run_invocations") or {}
            duration_ms = invocation.get("total_duration_ms")
            plan = self._latest_plan(ref.path)
            raw_status = (
                summary.get("status")
                or stage_delivery.get("status")
                or "RUNNING"
            )
            return {
                "id": ref.numeric_id,
                "run_key": ref.key,
                "run_kind": ref.kind,
                "agent_name": "global_data",
                "model_name": self._model_names(metrics),
                "objective": (plan or {}).get("goal") or ref.key,
                "status": _viewer_status(raw_status),
                "raw_status": raw_status,
                "parent_run_id": None,
                "start_time": _iso(start),
                "end_time": _iso(end),
                "duration_sec": (
                    round(float(duration_ms) / 1000, 1)
                    if isinstance(duration_ms, (int, float))
                    else None
                ),
                "step_count": len(summary.get("event_log") or []),
                "replan_count": summary.get("replan_count") or 0,
                "input_tokens": totals.get("input_tokens") or 0,
                "output_tokens": totals.get("output_tokens") or 0,
                "cost_usd": totals.get("reported_cost_usd") or 0,
                "cost_status": (
                    "reported"
                    if totals.get("reported_cost_usd") is not None
                    else "unavailable"
                ),
            }

        plan = self._latest_production_plan(ref.path) or {}
        reports = _files(
            ref.path, "segment_validation_results/segment.*.v*.json"
        )
        summaries = _files(ref.path, "segment_summaries/segment.*.v*.json")
        metrics = _load_json(ref.path / "run_metrics.json", {}) or {}
        totals = metrics.get("model_totals") or {}
        raw_status = "RUNNING"
        if reports:
            verdicts = [
                (_load_json(path, {}) or {}).get("verdict") for path in reports
            ]
            raw_status = "COMPLETED" if verdicts and all(
                value == "PASS" for value in verdicts
            ) else "FAILED"
        elif summaries:
            raw_status = (_load_json(summaries[-1], {}) or {}).get(
                "status", "RUNNING"
            )
        return {
            "id": ref.numeric_id,
            "run_key": ref.key,
            "run_kind": ref.kind,
            "agent_name": "segment_production",
            "model_name": self._model_names(metrics),
            "objective": plan.get("summary") or f"Production Segment · {ref.key}",
            "status": _viewer_status(raw_status),
            "raw_status": raw_status,
            "parent_run_id": None,
            "start_time": _iso(start),
            "end_time": _iso(end),
            "duration_sec": (
                round(end - start, 1)
                if start is not None and end is not None
                else None
            ),
            "step_count": len(_files(ref.path, "tool_operations/*.json")),
            "replan_count": 0,
            "input_tokens": totals.get("input_tokens") or 0,
            "output_tokens": totals.get("output_tokens") or 0,
            "cost_usd": totals.get("reported_cost_usd") or 0,
            "cost_status": (
                "reported"
                if totals.get("reported_cost_usd") is not None
                else "unavailable"
            ),
        }

    @staticmethod
    def _model_names(metrics: dict[str, Any]) -> str:
        names: list[str] = []
        for item in (metrics.get("models") or {}).values():
            for name in item.get("models") or []:
                if name and name not in names:
                    names.append(str(name))
        return " · ".join(names)

    @staticmethod
    def _latest_plan(run_dir: Path) -> dict[str, Any] | None:
        plans = _files(run_dir, "plans/global_data_dag.v*.json")
        return _load_json(plans[-1], {}) if plans else None

    @staticmethod
    def _latest_production_plan(
        run_dir: Path,
    ) -> dict[str, Any] | None:
        plans = _files(
            run_dir,
            "production_plans/production_dag.v*.json",
        )
        if not plans:
            return None
        latest = max(
            plans,
            key=lambda path: int(
                re.search(r"\.v(\d+)\.json$", path.name).group(1)
            ),
        )
        return _load_json(latest, {})

    def detail(self, ref: RunRef) -> dict[str, Any]:
        result = self.brief(ref)
        if ref.kind == "global_data":
            summary = _load_json(ref.path / "run_summary.json", {}) or {}
            delivery = _load_json(
                ref.path / "global_data_stage_delivery.json", {}
            ) or {}
            result.update(
                {
                    "final_output": delivery.get("summary"),
                    "error_message": summary.get("error"),
                    "input_payload": {
                        "input_path": summary.get("input_path"),
                        "input_sha256": summary.get("input_sha256"),
                        "protocol_version": summary.get("protocol_version"),
                    },
                    "overview": self._global_overview(ref, summary),
                    "steps": self._global_steps(summary),
                }
            )
        else:
            plan = self._latest_production_plan(ref.path) or {}
            result.update(
                {
                    "final_output": self._segment_final_output(ref),
                    "error_message": None,
                    "input_payload": plan,
                    "overview": self._segment_overview(ref),
                    "steps": self._segment_steps(ref),
                }
            )
        return result

    def _global_overview(
        self, ref: RunRef, summary: dict[str, Any]
    ) -> dict[str, int]:
        plans = _files(ref.path, "plans/global_data_dag.v*.json")
        latest = _load_json(plans[-1], {}) if plans else {}
        tasks = (latest or {}).get("tasks") or []
        version = int((latest or {}).get("plan_version") or len(plans) or 1)
        reports = [
            _load_json(
                ref.path
                / "validation_results"
                / f"{task.get('task_id')}.v{version}.json",
                {},
            )
            or {}
            for task in tasks
        ]
        result_files = _files(ref.path, "executor_results/Task*.v*.json")
        distinct_tasks = {
            re.sub(r"\.v\d+$", "", path.stem) for path in result_files
        }
        return {
            "plan_rounds": len(plans),
            "execute_rounds": int(
                summary.get("executor_calls") or len(result_files)
            ),
            "task_count": len(tasks),
            "passed": sum(report.get("verdict") == "PASS" for report in reports),
            "failed": sum(report.get("verdict") == "FAIL" for report in reports),
            "retries": max(len(result_files) - len(distinct_tasks), 0),
        }

    def _segment_overview(self, ref: RunRef) -> dict[str, int]:
        plan = self._latest_production_plan(ref.path) or {}
        packages = _files(ref.path, "segment_packages/*.json")
        latest_reports: dict[str, dict[str, Any]] = {}
        for path in _files(
            ref.path, "segment_validation_results/segment.*.v*.json"
        ):
            parts = path.stem.split(".")
            segment_id = parts[1] if len(parts) > 2 else path.stem
            latest_reports[segment_id] = _load_json(path, {}) or {}
        reports = list(latest_reports.values())
        return {
            "plan_rounds": len(
                _files(
                    ref.path,
                    "production_plans/production_dag.v*.json",
                )
            ),
            "execute_rounds": len(_files(ref.path, "tool_operations/*.json")),
            "task_count": len(plan.get("segments") or packages),
            "passed": sum(report.get("verdict") == "PASS" for report in reports),
            "failed": sum(report.get("verdict") == "FAIL" for report in reports),
            "retries": 0,
        }

    @staticmethod
    def _global_steps(summary: dict[str, Any]) -> list[dict[str, Any]]:
        steps = []
        for index, message in enumerate(summary.get("event_log") or [], start=1):
            lower = message.lower()
            if "planner" in lower:
                node_type = "plan"
            elif "executor" in lower:
                node_type = "execute"
            elif "validator" in lower or "验收" in message:
                node_type = "validate"
            elif "预处理" in message:
                node_type = "preprocess"
            elif "完成" in message:
                node_type = "finalize"
            else:
                node_type = "route"
            steps.append(
                {
                    "step_id": index,
                    "step_index": index,
                    "node_type": node_type,
                    "content": message,
                    "delta": {"event": message},
                    "input_tokens": 0,
                    "output_tokens": 0,
                    "created_at": None,
                }
            )
        return steps

    def _segment_steps(self, ref: RunRef) -> list[dict[str, Any]]:
        steps: list[dict[str, Any]] = []
        package_paths = _files(ref.path, "segment_packages/*.json")
        if package_paths:
            package = _load_json(package_paths[0], {}) or {}
            steps.append(
                {
                    "step_id": 1,
                    "step_index": 1,
                    "node_type": "prepare_segment",
                    "content": f"物化 {package.get('segment_id', 'Segment')} Package",
                    "delta": package,
                    "input_tokens": 0,
                    "output_tokens": 0,
                    "created_at": _iso(package_paths[0].stat().st_mtime),
                }
            )
        for index, path in enumerate(
            _files(ref.path, "tool_operations/*.json"), start=2
        ):
            operation = _load_json(path, {}) or {}
            steps.append(
                {
                    "step_id": index,
                    "step_index": index,
                    "node_type": "execute",
                    "content": (
                        f"{operation.get('tool_name', 'tool')} · "
                        f"{operation.get('status', 'UNKNOWN')}"
                    ),
                    "delta": operation,
                    "input_tokens": 0,
                    "output_tokens": 0,
                    "created_at": _iso(path.stat().st_mtime),
                }
            )
        return steps

    @staticmethod
    def _segment_final_output(ref: RunRef) -> Any:
        reports = _files(
            ref.path, "segment_validation_results/segment.*.v*.json"
        )
        if reports:
            return _load_json(reports[-1], {})
        packages = _files(ref.path, "segment_packages/*.json")
        if packages:
            package = _load_json(packages[-1], {}) or {}
            return {
                "status": "RUNNING",
                "segment_id": package.get("segment_id"),
                "message": "Segment 已物化，尚无终态 ValidationReport",
            }
        return None

    def snapshots(self, ref: RunRef) -> dict[str, Any]:
        if ref.kind == "global_data":
            snapshots = self._global_snapshots(ref)
        else:
            snapshots = self._segment_snapshots(ref)
        return {
            "run_id": ref.numeric_id,
            "run_key": ref.key,
            "objective": self.brief(ref)["objective"],
            "status": self.brief(ref)["status"],
            "snapshots": snapshots,
        }

    def _global_snapshots(self, ref: RunRef) -> list[dict[str, Any]]:
        paths = _files(ref.path, "plans/global_data_dag.v*.json")
        output: list[dict[str, Any]] = []
        previous_ids: set[str] = set()
        for index, path in enumerate(paths):
            plan = _load_json(path, {}) or {}
            version = plan.get("plan_version") or index + 1
            tasks = plan.get("tasks") or []
            current_ids = {str(item.get("task_id")) for item in tasks}
            added = sorted(current_ids - previous_ids)
            removed = sorted(previous_ids - current_ids)
            nodes = [
                self._snapshot_task(ref, task, version) for task in tasks
            ]
            output.append(
                {
                    "loop_index": index,
                    "kind": "initial" if index == 0 else "replan",
                    "title": f"Plan v{version}",
                    "step_id": index + 1,
                    "step_index": index + 1,
                    "is_final": index == len(paths) - 1,
                    "created_at": _iso(path.stat().st_mtime),
                    "reasoning": plan.get("revision_summary") or plan.get("goal"),
                    "tree_changed": bool(added or removed),
                    "diff": {
                        "added": added,
                        "removed": removed,
                        "modified": [],
                    },
                    "lessons_added": [],
                    "lessons_cumulative": [],
                    "tree": {"nodes": nodes},
                    "node_change": {
                        **{item: "added" for item in added},
                        **{item: "removed" for item in removed},
                    },
                    "round_execution": None,
                }
            )
            previous_ids = current_ids
        return output

    def _snapshot_task(
        self, ref: RunRef, task: dict[str, Any], version: int
    ) -> dict[str, Any]:
        task_id = str(task.get("task_id") or "")
        delivery = _load_json(
            ref.path / "executor_results" / f"{task_id}.v{version}.json", {}
        ) or {}
        report = _load_json(
            ref.path / "validation_results" / f"{task_id}.v{version}.json", {}
        ) or {}
        verdict = report.get("verdict")
        status = "completed" if verdict == "PASS" else (
            "failed" if verdict == "FAIL" else "pending"
        )
        return {
            "step_id": task_id,
            "task_id": task_id,
            "parent_id": None,
            "name": task.get("objective") or task_id,
            "goal": task.get("reason") or "",
            "status": status,
            "kind": task.get("skill_id"),
            "acceptance": task.get("expectation_ids") or [],
            "inputs": task.get("depends_on") or [],
            "stage": "GLOBAL_DATA",
            "category": task.get("deliverable_type") or "",
            "executing": False,
            "result": (
                {
                    "summary": delivery.get("summary") or report.get("summary"),
                    "product_url": None,
                    "product_text": None,
                    "verdict": report,
                    "attempts": 1,
                }
                if delivery or report
                else None
            ),
            "exec_step_id": None,
            "tools": [
                item.get("tool_name")
                for item in delivery.get("tool_calls") or []
                if item.get("tool_name")
            ],
            "trace_summary": {
                "llm": 0,
                "tool": len(delivery.get("tool_calls") or []),
            },
        }

    def _segment_snapshots(self, ref: RunRef) -> list[dict[str, Any]]:
        plan = self._latest_production_plan(ref.path) or {}
        packages = [
            _load_json(path, {}) or {}
            for path in _files(ref.path, "segment_packages/*.json")
        ]
        nodes = []
        for package in packages:
            segment_id = package.get("segment_id") or "Segment"
            reports = _files(
                ref.path,
                f"segment_validation_results/segment.{segment_id}.v*.json",
            )
            report = _load_json(reports[-1], {}) if reports else {}
            verdict = (report or {}).get("verdict")
            nodes.append(
                {
                    "step_id": segment_id,
                    "task_id": segment_id,
                    "parent_id": None,
                    "name": segment_id,
                    "goal": f"生产并验收 {segment_id}",
                    "status": (
                        "completed" if verdict == "PASS"
                        else "failed" if verdict == "FAIL"
                        else "running"
                    ),
                    "kind": "segment-production",
                    "acceptance": [],
                    "inputs": [
                        item.get("input_id")
                        for item in package.get("production_inputs") or []
                    ],
                    "stage": "PRODUCTION",
                    "category": "video",
                    "executing": verdict is None,
                    "result": report or None,
                    "exec_step_id": None,
                    "tools": [],
                    "trace_summary": {"llm": 0, "tool": 0},
                }
            )
        return [
            {
                "loop_index": 0,
                "kind": "initial",
                "title": f"Production Plan v{plan.get('plan_version', 1)}",
                "step_id": 1,
                "step_index": 1,
                "is_final": False,
                "created_at": None,
                "reasoning": plan.get("summary") or "",
                "tree_changed": True,
                "diff": {
                    "added": [node["step_id"] for node in nodes],
                    "removed": [],
                    "modified": [],
                },
                "lessons_added": [],
                "lessons_cumulative": [],
                "tree": {"nodes": nodes},
                "node_change": {
                    node["step_id"]: "added" for node in nodes
                },
                "round_execution": None,
            }
        ]

    def flat(self, ref: RunRef, round_index: int | None) -> dict[str, Any]:
        if ref.kind == "global_data":
            return self._global_flat(ref, round_index)
        return self._segment_flat(ref, round_index)

    @staticmethod
    def _module_def(
        module_id: int,
        key: str,
        *,
        name: str,
        summary: str,
        tools: Iterable[str] = (),
        inputs: Iterable[tuple[str, str]] = (),
        model: str = "",
        refs: Iterable[tuple[str, str]] = (),
    ) -> dict[str, Any]:
        return {
            "id": module_id,
            "module_key": key,
            "kind": "agent",
            "name": name,
            "summary": summary,
            "spec": {
                "node_kind": "agent",
                "summary": summary,
                "system_prompt": "",
                "input_schema": [
                    {
                        "key": input_key,
                        "name": input_key,
                        "title": title,
                        "required": True,
                        "source": title,
                    }
                    for input_key, title in inputs
                ],
                "tools": [{"name": value} for value in tools],
                "model": model,
            },
            "view": None,
            "fingerprint": hashlib.sha256(
                f"{key}:{summary}".encode("utf-8")
            ).hexdigest()[:8],
            "refs": [
                {
                    "node_key": node_key,
                    "to_module_key": target,
                    "seq": index,
                    "title": target,
                    "cardinality": "one",
                    "fanout_over": "",
                }
                for index, (node_key, target) in enumerate(refs)
            ],
        }

    @staticmethod
    def _instance(
        instance_id: int,
        module_id: int,
        module_key: str,
        *,
        label: str,
        task_id: str | None,
        ok: bool | None,
        inputs: list[dict[str, Any]],
        output: Any,
        flow: list[dict[str, Any]],
        used_tools: Iterable[str] = (),
        images: Iterable[str] = (),
    ) -> dict[str, Any]:
        return {
            "id": instance_id,
            "module_id": module_id,
            "module_key": module_key,
            "label": label,
            "task_id": task_id,
            "step_id": task_id,
            "step_index": instance_id,
            "branch": "",
            "ok": ok,
            "input": {"blocks": inputs},
            "output": {
                "data": output,
                "ok": ok,
                "images": list(images),
            } if output is not None else None,
            "timing": {"sec": None},
            "tokens": "",
            "caller": None,
            "used_tools": sorted(set(used_tools)),
            "flow": flow,
        }

    def _global_flat(
        self, ref: RunRef, round_index: int | None
    ) -> dict[str, Any]:
        plans = _files(ref.path, "plans/global_data_dag.v*.json")
        metrics = _load_json(ref.path / "run_metrics.json", {}) or {}
        model = self._model_names(metrics)
        module_defs = {
            1: self._module_def(
                1, "production.global_data.plan",
                name="Global Data Planner",
                summary="读取 Production Brief，生成或修订完整 Global Data DAG。",
                inputs=(("brief", "Production Brief"),),
                model=model,
            ),
            2: self._module_def(
                2, "production.global_data.execute",
                name="Global Data Executor",
                summary="按 TaskPackage 与 Skill 白名单执行一个 Task。",
                inputs=(("task", "TaskPackage"),),
                model=model,
            ),
            3: self._module_def(
                3, "production.global_data.validate_task",
                name="Task Validator",
                summary="在独立上下文中逐项验收 Executor Delivery。",
                inputs=(("delivery", "Executor Delivery"),),
                model=model,
            ),
            4: self._module_def(
                4, "production.global_data.validate_stage",
                name="Stage Validator",
                summary="全部 Task 通过后检查阶段完整性、冲突和漏项。",
                inputs=(("plan", "Global Data Plan"), ("tasks", "PASS Deliveries")),
                model=model,
            ),
        }
        rounds: list[dict[str, Any]] = []
        for index, path in enumerate(plans):
            if round_index is not None and index != round_index:
                continue
            plan = _load_json(path, {}) or {}
            version = int(plan.get("plan_version") or index + 1)
            instances: list[dict[str, Any]] = []
            next_id = (index + 1) * 1000
            next_id += 1
            planner_id = next_id
            instances.append(
                self._instance(
                    planner_id, 1, "production.global_data.plan",
                    label=f"Plan v{version}",
                    task_id=None,
                    ok=True,
                    inputs=[
                        _input_block(
                            "brief", "Production Brief",
                            plan.get("goal"), source="production_brief.json",
                        )
                    ],
                    output=plan,
                    flow=[
                        _node_card(
                            0, node_type="llm", label="规划",
                            summary=plan.get("revision_summary") or plan.get("goal"),
                            ok=True, detail=plan,
                        )
                    ],
                )
            )
            entry_ids = [planner_id]
            for task in self._ordered_tasks(ref, plan.get("tasks") or []):
                task_id = str(task.get("task_id"))
                delivery = _load_json(
                    ref.path / "executor_results" / f"{task_id}.v{version}.json",
                    {},
                ) or {}
                report = _load_json(
                    ref.path / "validation_results" / f"{task_id}.v{version}.json",
                    {},
                ) or {}
                next_id += 1
                executor_id = next_id
                tools = [
                    item.get("tool_name")
                    for item in delivery.get("tool_calls") or []
                    if item.get("tool_name")
                ]
                flow = [
                    _node_card(
                        flow_index,
                        node_type="tool",
                        label=call.get("tool_name") or "tool",
                        summary=call.get("output_excerpt"),
                        ok=call.get("success"),
                        detail={
                            "input": call.get("arguments"),
                            "output": call.get("output_excerpt"),
                        },
                        tool_call_id=call.get("tool_call_id") or "",
                    )
                    for flow_index, call in enumerate(
                        delivery.get("tool_calls") or []
                    )
                ]
                if not flow and delivery:
                    flow = [
                        _node_card(
                            0,
                            node_type="tool",
                            label="确定性交付",
                            summary=delivery.get("summary") or "已写入交付结果",
                            ok=True,
                            detail=delivery,
                        )
                    ]
                delivery_view = self._delivery_view(ref, delivery)
                image_urls = [
                    artifact["browser_uri"]
                    for artifact in (delivery_view or {}).get("artifacts") or []
                    if str(
                        artifact.get("mime_type")
                        or artifact.get("media_type")
                        or ""
                    ).startswith("image/")
                    and artifact.get("browser_uri")
                ]
                instances.append(
                    self._instance(
                        executor_id, 2, "production.global_data.execute",
                        label=task.get("objective") or task_id,
                        task_id=task_id,
                        ok=bool(delivery),
                        inputs=[
                            _input_block(
                                "task", "TaskPackage", task,
                                source=f"tasks/{task_id}.v{version}.json",
                            )
                        ],
                        output=delivery_view,
                        flow=flow or [
                            _node_card(
                                0, node_type="note", label="等待执行",
                                summary="尚无 Executor Delivery", ok=None,
                            )
                        ],
                        used_tools=tools,
                        images=image_urls,
                    )
                )
                entry_ids.append(executor_id)
                next_id += 1
                validator_id = next_id
                instances.append(
                    self._instance(
                        validator_id, 3,
                        "production.global_data.validate_task",
                        label=f"{task_id} 验收",
                        task_id=task_id,
                        ok=report.get("verdict") == "PASS" if report else None,
                        inputs=[
                            _input_block(
                                "delivery", "Executor Delivery",
                                delivery.get("summary"),
                                source=f"executor_results/{task_id}.v{version}.json",
                            )
                        ],
                        output=report or None,
                        flow=[
                            _node_card(
                                0, node_type="llm", label="独立验收",
                                summary=report.get("summary") or "尚无验收报告",
                                ok=(
                                    report.get("verdict") == "PASS"
                                    if report else None
                                ),
                                detail=report or None,
                            )
                        ],
                    )
                )
                entry_ids.append(validator_id)
            stage_paths = _files(
                ref.path,
                f"stage_validation_results/global_data.v{version}.json",
            )
            if stage_paths:
                stage = _load_json(stage_paths[-1], {}) or {}
                next_id += 1
                stage_id = next_id
                instances.append(
                    self._instance(
                        stage_id, 4, "production.global_data.validate_stage",
                        label="Global Data Stage 验收",
                        task_id=None,
                        ok=stage.get("verdict") == "PASS",
                        inputs=[
                            _input_block(
                                "plan", "Global Data Plan", f"v{version}",
                                source=path.name,
                            ),
                            _input_block(
                                "tasks", "PASS Deliveries",
                                [item.get("task_id") for item in plan.get("tasks") or []],
                                source="executor_results/",
                            ),
                        ],
                        output=stage,
                        flow=[
                            _node_card(
                                0, node_type="llm", label="阶段总验收",
                                summary=stage.get("summary"),
                                ok=stage.get("verdict") == "PASS",
                                detail=stage,
                            )
                        ],
                    )
                )
                entry_ids.append(stage_id)
            rounds.append(
                {"index": index, "entry_ids": entry_ids, "instances": instances}
            )
        root = {
            "id": 0,
            "module_id": 0,
            "module_key": "production.global_data.workflow",
            "label": "Global Data 0.3",
            "task_id": None,
            "step_id": None,
            "step_index": 0,
            "branch": "",
            "ok": self.brief(ref)["status"] == "success",
            "input": {"blocks": []},
            "output": None,
            "timing": {"sec": self.brief(ref)["duration_sec"]},
            "tokens": "",
            "caller": None,
            "used_tools": [],
            "flow": [],
            "module": {
                "id": 0,
                "module_key": "production.global_data.workflow",
                "kind": "workflow",
                "name": "Global Data Workflow",
                "summary": "Preprocess → Plan → Execute → Validate → Replan/Finalize",
                "spec": {},
                "view": None,
                "fingerprint": "0.3",
                "refs": [],
            },
        }
        return {
            "run_id": ref.numeric_id,
            "run_key": ref.key,
            "root": root,
            "modules": module_defs,
            "round_count": len(plans),
            "rounds": rounds,
        }

    @staticmethod
    def _ordered_tasks(
        ref: RunRef, tasks: list[dict[str, Any]]
    ) -> list[dict[str, Any]]:
        summary = _load_json(ref.path / "run_summary.json", {}) or {}
        order: list[str] = []
        for event in summary.get("event_log") or []:
            match = re.search(r"(Task\d+)", event)
            if match and match.group(1) not in order:
                order.append(match.group(1))
        rank = {task_id: index for index, task_id in enumerate(order)}
        return sorted(
            tasks,
            key=lambda item: (
                rank.get(str(item.get("task_id")), len(rank)),
                str(item.get("task_id")),
            ),
        )

    def _delivery_view(
        self, ref: RunRef, delivery: dict[str, Any]
    ) -> dict[str, Any] | None:
        if not delivery:
            return None
        result = dict(delivery)
        result["artifacts"] = [
            {
                **artifact,
                "browser_uri": self.media_url(ref, artifact.get("uri")),
            }
            for artifact in delivery.get("artifacts") or []
        ]
        return result

    def _segment_flat(
        self, ref: RunRef, round_index: int | None
    ) -> dict[str, Any]:
        plan = self._latest_production_plan(ref.path) or {}
        packages = [
            (path, _load_json(path, {}) or {})
            for path in _files(ref.path, "segment_packages/*.json")
        ]
        modules = {
            11: self._module_def(
                11, "production.segment.execute",
                name="Segment Executor",
                summary="按 SegmentPackage 生产镜头、声音、字幕和最终片段。",
                inputs=(("package", "SegmentPackage"),),
            ),
            12: self._module_def(
                12, "production.segment.validate",
                name="Segment Validator",
                summary="确定性证据前检后，独立判断六项 Segment 质量。",
                inputs=(("delivery", "Segment Delivery"),),
            ),
        }
        instances: list[dict[str, Any]] = []
        entry_ids: list[int] = []
        operations = [
            _load_json(path, {}) or {}
            for path in _files(ref.path, "tool_operations/*.json")
        ]
        for index, (package_path, package) in enumerate(packages):
            segment_id = str(package.get("segment_id") or f"Segment{index + 1}")
            executor_id = 2000 + index * 10 + 1
            flow = [
                _node_card(
                    flow_index,
                    node_type="tool",
                    label=operation.get("tool_name") or "tool",
                    summary=operation.get("status"),
                    ok=(
                        True
                        if operation.get("status") == "SUCCEEDED"
                        else False
                        if operation.get("status") == "FAILED"
                        else None
                    ),
                    detail=operation,
                )
                for flow_index, operation in enumerate(operations)
            ]
            deliveries = _files(
                ref.path, f"segment_deliveries/segment.{segment_id}.v*.json"
            )
            delivery = _load_json(deliveries[-1], {}) if deliveries else {}
            instances.append(
                self._instance(
                    executor_id, 11, "production.segment.execute",
                    label=segment_id,
                    task_id=segment_id,
                    ok=bool(delivery) if deliveries else None,
                    inputs=[
                        _input_block(
                            "package", "SegmentPackage", package,
                            source=str(package_path.relative_to(ref.path)),
                        )
                    ],
                    output=delivery or None,
                    flow=flow or [
                        _node_card(
                            0, node_type="note", label="等待执行",
                            summary="尚未形成 Segment Delivery", ok=None,
                        )
                    ],
                    used_tools=[
                        item.get("tool_name")
                        for item in operations
                        if item.get("tool_name")
                    ],
                )
            )
            entry_ids.append(executor_id)
            reports = _files(
                ref.path,
                f"segment_validation_results/segment.{segment_id}.v*.json",
            )
            if reports:
                report = _load_json(reports[-1], {}) or {}
                validator_id = executor_id + 1
                instances.append(
                    self._instance(
                        validator_id, 12, "production.segment.validate",
                        label=f"{segment_id} 验收",
                        task_id=segment_id,
                        ok=report.get("verdict") == "PASS",
                        inputs=[
                            _input_block(
                                "delivery", "Segment Delivery",
                                delivery, source=deliveries[-1].name,
                            )
                        ],
                        output=report,
                        flow=[
                            _node_card(
                                0, node_type="llm", label="六项独立验收",
                                summary=report.get("summary"),
                                ok=report.get("verdict") == "PASS",
                                detail=report,
                            )
                        ],
                    )
                )
                entry_ids.append(validator_id)
        rounds = (
            [{"index": 0, "entry_ids": entry_ids, "instances": instances}]
            if round_index in (None, 0)
            else []
        )
        root = {
            "id": 10,
            "module_id": 10,
            "module_key": "production.segment.workflow",
            "label": f"Production 0.6 · {plan.get('plan_id', ref.key)}",
            "task_id": None,
            "step_id": None,
            "step_index": 0,
            "branch": "",
            "ok": self.brief(ref)["status"] == "success",
            "input": {"blocks": []},
            "output": None,
            "timing": {"sec": self.brief(ref)["duration_sec"]},
            "tokens": "",
            "caller": None,
            "used_tools": [],
            "flow": [],
            "module": {
                "id": 10,
                "module_key": "production.segment.workflow",
                "kind": "workflow",
                "name": "Segment Workflow",
                "summary": "SegmentPackage → Executor → Validator",
                "spec": {},
                "view": None,
                "fingerprint": "0.6",
                "refs": [],
            },
        }
        return {
            "run_id": ref.numeric_id,
            "run_key": ref.key,
            "root": root,
            "modules": modules,
            "round_count": 1,
            "rounds": rounds,
        }

    def media_path(self, ref: RunRef, relative: str) -> Path:
        candidate = (ref.path / relative).resolve()
        if not candidate.is_relative_to(ref.path) or not candidate.is_file():
            raise FileNotFoundError(relative)
        return candidate

    def media_url(self, ref: RunRef, uri: Any) -> str | None:
        if not isinstance(uri, str) or not uri:
            return None
        try:
            path = Path(uri).resolve()
            relative = path.relative_to(ref.path)
        except (ValueError, OSError):
            return uri if uri.startswith(("http://", "https://")) else None
        return f"/api/runs/{ref.numeric_id}/media/{relative.as_posix()}"

    def tools(self) -> dict[str, Any]:
        from production_build_agents.tools.registry import (
            create_default_tool_registry,
        )

        # ToolRegistry 没有只读定义接口；从 Skill 白名单和工具 README 返回稳定说明，
        # 不实例化远端客户端，也不触发环境校验。
        names = {
            "read_production_brief",
            "search_tool",
            "inspect_tool",
            "run_tool",
            "probe_media",
            "view_images",
            "extract_frames",
            "publish_media_reference",
            "transcribe_audio",
            "create_ass_subtitles",
            "inspect_ass_subtitles",
            "render_ass_subtitles",
            "audio_trim",
            "mix_audio_tracks",
            "video_trim",
            "video_concat",
            "video_mux_audio",
        }
        _ = create_default_tool_registry  # 保留到实现正式定义投影时使用。
        return {
            name: {
                "description": "Production Build 运行时工具",
                "params": [],
            }
            for name in sorted(names)
        }


def media_type(path: Path) -> str:
    return mimetypes.guess_type(path.name)[0] or "application/octet-stream"