"""把本项目按 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"