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 _version(path: Path) -> int:
    match = re.search(r"\.v(\d+)\.json$", path.name)
    return int(match.group(1)) if match else 0


def _latest_by_subject(paths: Iterable[Path]) -> dict[str, Path]:
    """按 ``Subject.vN.json`` 的 Subject 选择最新正式版本。"""

    result: dict[str, Path] = {}
    for path in paths:
        subject = re.sub(r"\.v\d+$", "", path.stem)
        current = result.get(subject)
        if current is None or _version(path) > _version(current):
            result[subject] = path
    return result


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"
            )
            tasks = (plan or {}).get("tasks") or []
            plan_version = int(
                (plan or {}).get("plan_version") or 1
            )
            task_reports = [
                _load_json(
                    ref.path
                    / "validation_results"
                    / f"{task.get('task_id')}.v{plan_version}.json",
                    {},
                ) or {}
                for task in tasks
            ]
            task_passed = sum(
                report.get("verdict") == "PASS"
                for report in task_reports
            )
            task_failed = sum(
                report.get("verdict") == "FAIL"
                for report in task_reports
            )
            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,
                "business_label": f"Global Data · {len(tasks)} 个资料任务",
                "business_progress": (
                    f"{task_passed} 已通过"
                    + (f" · {task_failed} 未通过" if task_failed else "")
                ),
                "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 {}
        run_summary = _load_json(
            ref.path / "production_run_summary.json", {}
        ) or {}
        reports = _files(ref.path, "segment_validation_results/*.v*.json")
        summaries = _files(ref.path, "segment_summaries/*.v*.json")
        metrics = _load_json(ref.path / "run_metrics.json", {}) or {}
        totals = metrics.get("model_totals") or {}
        latest_reports = [
            _load_json(path, {}) or {}
            for path in _latest_by_subject(reports).values()
        ]
        segments = plan.get("segments") or []
        segment_passed = sum(
            report.get("verdict") == "PASS"
            for report in latest_reports
        )
        segment_failed = sum(
            report.get("verdict") == "FAIL"
            for report in latest_reports
        )
        raw_status = run_summary.get("status") or "RUNNING"
        if not run_summary and 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 not run_summary and 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,
            "business_label": f"Production · {len(segments)} 个 Segment",
            "business_progress": (
                (
                    f"停在 {run_summary.get('current_segment_id')} · "
                    f"{run_summary.get('phase')}"
                )
                if run_summary.get("current_segment_id")
                else (
                    f"{segment_passed} 已通过"
                    + (
                        f" · {segment_failed} 未通过"
                        if segment_failed else ""
                    )
                )
            ),
            "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 {}
            summary = _load_json(
                ref.path / "production_run_summary.json", {}
            ) or {}
            result.update(
                {
                    "final_output": self._segment_final_output(ref),
                    "error_message": summary.get("error"),
                    "input_payload": plan,
                    "overview": self._segment_overview(ref),
                    "steps": self._segment_steps(ref),
                }
            )
        result["business_story"] = self._business_story(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 = {
            segment_id: _load_json(path, {}) or {}
            for segment_id, path in _latest_by_subject(
                _files(ref.path, "segment_validation_results/*.v*.json")
            ).items()
        }
        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:
        final_delivery = _load_json(
            ref.path / "final_production_delivery.json", None
        )
        if final_delivery:
            return final_delivery
        summary = _load_json(
            ref.path / "production_run_summary.json", {}
        ) or {}
        reports = _files(ref.path, "segment_validation_results/*.v*.json")
        if reports:
            return _load_json(reports[-1], {})
        if summary.get("status") == "FAILED":
            return {
                "status": "FAILED",
                "phase": summary.get("phase"),
                "failure_code": summary.get("failure_code"),
                "message": summary.get("error") or summary.get("event"),
            }
        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 _business_story(self, ref: RunRef) -> dict[str, Any]:
        if ref.kind == "global_data":
            return self._global_business_story(ref)
        return self._production_business_story(ref)

    def _artifact_item(
        self, ref: RunRef, artifact: dict[str, Any]
    ) -> dict[str, Any]:
        uri = artifact.get("uri") or artifact.get("source_uri")
        return {
            "label": (
                artifact.get("description")
                or artifact.get("artifact_id")
                or artifact.get("artifact_type")
                or "正式产物"
            ),
            "detail": " · ".join(
                str(value)
                for value in (
                    artifact.get("artifact_type"),
                    artifact.get("mime_type"),
                )
                if value
            ),
            "source": artifact.get("artifact_id") or "",
            "media_url": self.media_url(ref, uri),
        }

    @staticmethod
    def _criteria(report: dict[str, Any]) -> list[dict[str, Any]]:
        source = (
            report.get("criterion_results")
            or report.get("requirement_results")
            or report.get("checks")
            or []
        )
        return [
            {
                "label": (
                    item.get("expectation_id")
                    or item.get("requirement_id")
                    or item.get("name")
                    or item.get("verification_capability")
                    or f"判断项 {index}"
                ),
                "verdict": item.get("verdict") or item.get("status") or "UNKNOWN",
                "reason": item.get("reason") or item.get("summary") or "",
                "evidence": item.get("evidence") or [],
            }
            for index, item in enumerate(source, start=1)
            if isinstance(item, dict)
        ]

    def _global_business_story(self, ref: RunRef) -> dict[str, Any]:
        summary = _load_json(ref.path / "run_summary.json", {}) or {}
        delivery = _load_json(
            ref.path / "global_data_stage_delivery.json", {}
        ) or {}
        plan_paths = _files(ref.path, "plans/global_data_dag.v*.json")
        rounds: list[dict[str, Any]] = []
        for round_index, plan_path in enumerate(plan_paths, start=1):
            plan = _load_json(plan_path, {}) or {}
            version = int(plan.get("plan_version") or _version(plan_path) or 1)
            steps: list[dict[str, Any]] = []
            for task in self._ordered_tasks(ref, plan.get("tasks") or []):
                task_id = str(task.get("task_id") or "")
                execution = _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 = (
                    "success" if verdict == "PASS"
                    else "failed" if verdict == "FAIL"
                    else "running" if execution
                    else "pending"
                )
                basis = [
                    {
                        "label": "业务来源",
                        "detail": source,
                        "source": source,
                    }
                    for source in task.get("source_paths") or []
                ]
                basis.extend(
                    {
                        "label": "验收目标",
                        "detail": expectation,
                        "source": expectation,
                    }
                    for expectation in task.get("expectation_ids") or []
                )
                basis.extend(
                    {
                        "label": "前置依赖",
                        "detail": dependency,
                        "source": dependency,
                    }
                    for dependency in task.get("depends_on") or []
                )
                outputs = [
                    self._artifact_item(ref, artifact)
                    for artifact in execution.get("artifacts") or []
                    if isinstance(artifact, dict)
                ]
                steps.append(
                    {
                        "id": task_id,
                        "stage": "GLOBAL DATA",
                        "title": task.get("objective") or task_id,
                        "status": status,
                        "action": (
                            execution.get("summary")
                            or "已纳入本轮计划，尚未形成正式执行交付。"
                        ),
                        "reason": task.get("reason") or "",
                        "basis": basis,
                        "judgment": {
                            "verdict": verdict or "PENDING",
                            "summary": (
                                report.get("summary")
                                or "尚无独立 Validator 判断。"
                            ),
                            "criteria": self._criteria(report),
                        },
                        "outputs": outputs,
                        "tools": [
                            {
                                "name": call.get("tool_name") or "tool",
                                "status": (
                                    "SUCCEEDED"
                                    if call.get("success") is True
                                    else "FAILED"
                                    if call.get("success") is False
                                    else "UNKNOWN"
                                ),
                                "detail": call.get("output_excerpt") or "",
                            }
                            for call in execution.get("tool_calls") or []
                            if isinstance(call, dict)
                        ],
                    }
                )
            if round_index == len(plan_paths):
                stage_paths = _files(
                    ref.path, "stage_validation_results/global_data.v*.json"
                )
                if stage_paths:
                    stage_report = _load_json(stage_paths[-1], {}) or {}
                    stage_verdict = stage_report.get("verdict")
                    steps.append(
                        {
                            "id": "GLOBAL_DATA_STAGE",
                            "stage": "STAGE GATE",
                            "title": "判断 Global Data 是否可交付给 Production",
                            "status": (
                                "success" if stage_verdict == "PASS"
                                else "failed" if stage_verdict == "FAIL"
                                else "running"
                            ),
                            "action": "汇总所有 Task 的正式产物，检查覆盖、冲突与未解决项。",
                            "reason": "单个素材通过不等于整套生产输入完整，阶段门负责做跨 Task 的最终判断。",
                            "basis": [
                                {
                                    "label": "阶段输入",
                                    "detail": f"{len(plan.get('tasks') or [])} 个 Task 的正式交付与独立验收",
                                    "source": stage_paths[-1].name,
                                }
                            ],
                            "judgment": {
                                "verdict": stage_verdict or "PENDING",
                                "summary": stage_report.get("summary") or "",
                                "criteria": self._criteria(stage_report),
                            },
                            "outputs": [
                                {
                                    "label": "Global Data Stage Delivery",
                                    "detail": delivery.get("summary") or "",
                                    "source": "global_data_stage_delivery.json",
                                    "media_url": None,
                                }
                            ] if delivery else [],
                            "tools": [],
                        }
                    )
            passed = sum(step["status"] == "success" for step in steps)
            failed = sum(step["status"] == "failed" for step in steps)
            rounds.append(
                {
                    "index": round_index,
                    "title": f"Global Data Plan v{version}",
                    "status": (
                        "failed" if failed
                        else "success" if steps and passed == len(steps)
                        else "running"
                    ),
                    "summary": (
                        plan.get("revision_summary")
                        or plan.get("goal")
                        or "Global Data 规划与验收"
                    ),
                    "steps": steps,
                }
            )
        raw_status = (
            summary.get("status") or delivery.get("status") or "RUNNING"
        )
        return {
            "kind": "global_data",
            "protocol_version": (
                summary.get("protocol_version")
                or (self._latest_plan(ref.path) or {}).get("schema_version")
                or ""
            ),
            "status": _viewer_status(raw_status),
            "phase": summary.get("phase") or "GLOBAL_DATA",
            "headline": "把 Production 需要的共享素材与约束准备完整",
            "conclusion": (
                delivery.get("summary")
                or summary.get("error")
                or "运行尚未形成阶段交付。"
            ),
            "failure": (
                {
                    "code": summary.get("failure_code") or "GLOBAL_DATA_FAILED",
                    "message": summary.get("error") or "",
                    "stage": summary.get("phase") or "GLOBAL_DATA",
                }
                if _viewer_status(raw_status) == "failed"
                else None
            ),
            "rounds": rounds,
        }

    @staticmethod
    def _operation_segment(operation: dict[str, Any]) -> str:
        operation_id = str(
            operation.get("operation_scope")
            or operation.get("operation_id")
            or ""
        )
        match = re.search(r":(Segment[^:]+):v\d+:", operation_id)
        return match.group(1) if match else ""

    def _production_business_story(self, ref: RunRef) -> dict[str, Any]:
        summary = _load_json(
            ref.path / "production_run_summary.json", {}
        ) or {}
        plan_paths = _files(
            ref.path, "production_plans/production_dag.v*.json"
        )
        all_operations = [
            _load_json(path, {}) or {}
            for path in _files(ref.path, "tool_operations/*.json")
        ]
        rounds: list[dict[str, Any]] = []
        for round_index, plan_path in enumerate(plan_paths, start=1):
            plan = _load_json(plan_path, {}) or {}
            version = int(plan.get("plan_version") or _version(plan_path) or 1)
            steps: list[dict[str, Any]] = []
            for segment in plan.get("segments") or []:
                segment_id = str(segment.get("segment_id") or "")
                package_path = (
                    ref.path
                    / "segment_packages"
                    / f"{segment_id}.v{version}.json"
                )
                package = _load_json(package_path, {}) or {}
                delivery_paths = _files(
                    ref.path,
                    f"segment_deliveries/{segment_id}.v*.json",
                )
                report_paths = _files(
                    ref.path,
                    f"segment_validation_results/{segment_id}.v*.json",
                )
                delivery = (
                    _load_json(delivery_paths[-1], {}) or {}
                    if delivery_paths else {}
                )
                report = (
                    _load_json(report_paths[-1], {}) or {}
                    if report_paths else {}
                )
                operations = [
                    item
                    for item in all_operations
                    if self._operation_segment(item) == segment_id
                ]
                is_failed_segment = (
                    summary.get("status") == "FAILED"
                    and summary.get("current_segment_id") == segment_id
                )
                verdict = report.get("verdict")
                status = (
                    "success" if verdict == "PASS"
                    else "failed" if verdict == "FAIL" or is_failed_segment
                    else "running" if package or operations
                    else "pending"
                )
                artifact_inputs = (
                    package.get("artifact_inputs")
                    or segment.get("artifact_inputs")
                    or []
                )
                basis = []
                for item in artifact_inputs:
                    if not isinstance(item, dict):
                        continue
                    basis.append(
                        {
                            "label": (
                                item.get("usage")
                                or item.get("artifact_id")
                                or "正式素材"
                            ),
                            "detail": (
                                item.get("artifact_id")
                                or item.get("input_id")
                                or _clip(item.get("value"), 180)
                            ),
                            "source": (
                                item.get("source_path")
                                or item.get("artifact_id")
                                or ""
                            ),
                            "media_url": self.media_url(
                                ref, item.get("uri") or item.get("source_uri")
                            ),
                        }
                    )
                basis.extend(
                    {
                        "label": "前置 Segment",
                        "detail": dependency,
                        "source": dependency,
                    }
                    for dependency in (
                        segment.get("depends_on")
                        or package.get("dependencies")
                        or []
                    )
                )
                outputs = [
                    self._artifact_item(ref, artifact)
                    for artifact in delivery.get("artifacts") or []
                    if isinstance(artifact, dict)
                ]
                judgment_summary = report.get("summary") or ""
                judgment_verdict = verdict or "PENDING"
                if is_failed_segment:
                    judgment_verdict = "FAIL"
                    judgment_summary = (
                        summary.get("error")
                        or summary.get("event")
                        or "Segment 执行失败。"
                    )
                steps.append(
                    {
                        "id": segment_id,
                        "stage": "PRODUCTION",
                        "title": (
                            segment.get("objective")
                            or package.get("objective")
                            or f"生产 {segment_id}"
                        ),
                        "status": status,
                        "action": (
                            delivery.get("summary")
                            or (
                                f"执行了 {len(operations)} 次媒体工具操作，"
                                "但尚未形成合法 Segment Delivery。"
                                if operations else
                                "已规划，尚未进入正式生产。"
                            )
                        ),
                        "reason": (
                            f"完成时间线 {(
                                package.get('timeline')
                                or segment.get('timeline')
                                or {}
                            ).get('start_ms', '?')}–{(
                                package.get('timeline')
                                or segment.get('timeline')
                                or {}
                            ).get('end_ms', '?')} ms 的独立可验收片段。"
                        ),
                        "basis": basis,
                        "judgment": {
                            "verdict": judgment_verdict,
                            "summary": (
                                judgment_summary
                                or "尚无正式 Segment ValidationReport。"
                            ),
                            "criteria": self._criteria(report),
                        },
                        "outputs": outputs,
                        "tools": [
                            {
                                "name": operation.get("tool_name") or "tool",
                                "status": operation.get("status") or "UNKNOWN",
                                "detail": _clip(operation.get("result"), 360),
                            }
                            for operation in sorted(
                                operations,
                                key=lambda item: item.get("ordinal") or 0,
                            )
                        ],
                    }
                )
            failed = sum(step["status"] == "failed" for step in steps)
            passed = sum(step["status"] == "success" for step in steps)
            rounds.append(
                {
                    "index": round_index,
                    "title": f"Production Plan v{version}",
                    "status": (
                        "failed" if failed
                        else "success" if steps and passed == len(steps)
                        else "running"
                    ),
                    "summary": (
                        plan.get("revision_summary")
                        or plan.get("summary")
                        or f"按 {len(steps)} 个 Segment 组织正式生产"
                    ),
                    "steps": steps,
                }
            )
        raw_status = summary.get("status") or "RUNNING"
        return {
            "kind": "production",
            "protocol_version": (
                summary.get("schema_version")
                or (self._latest_production_plan(ref.path) or {}).get(
                    "schema_version"
                )
                or ""
            ),
            "status": _viewer_status(raw_status),
            "phase": summary.get("phase") or "PRODUCTION",
            "headline": "按 Segment 生产、独立验收，再决定是否进入最终交付",
            "conclusion": (
                summary.get("event")
                or summary.get("error")
                or "运行尚未形成正式 Production 结论。"
            ),
            "failure": (
                {
                    "code": summary.get("failure_code") or "PRODUCTION_FAILED",
                    "message": summary.get("error") or summary.get("event") or "",
                    "stage": summary.get("phase") or "PRODUCTION",
                }
                if _viewer_status(raw_status) == "failed"
                else None
            ),
            "rounds": rounds,
        }

    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_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": self.brief(ref)["status"] != "running",
                "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] = []
        all_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
            operations = [
                operation
                for operation in all_operations
                if self._operation_segment(operation) == segment_id
            ]
            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_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_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 []
        )
        protocol_version = str(plan.get("schema_version") or "unknown")
        root = {
            "id": 10,
            "module_id": 10,
            "module_key": "production.segment.workflow",
            "label": (
                f"Production {protocol_version} · "
                f"{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": protocol_version,
                "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"