from __future__ import annotations import json import os import re from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable from .models import ( AgentStep, AttemptDetail, DataUsage, ExecutionView, LifecycleEvent, PlannerTurn, RunSummary, TaskDetail, TaskNode, ) VISUALIZATION_ROOT = Path(__file__).resolve().parents[2] DEFAULT_AGENT_DATA_ROOT = VISUALIZATION_ROOT.parent / "script_build_host" / ".local" / "agent-data" BUILD_ID_PATTERN = re.compile(r"Script build (\d+)", re.IGNORECASE) TASK_KIND_PREFIX = "script-build://task-kinds/" CONTRACT_PREFIX = "script-build://task-contracts/sha256/" PLANNER_TOOLS = { "inspect_script_plan", "plan_script_tasks", "dispatch_script_tasks", "validate_attempt", "decide_script_task", "read_input_snapshot", } ACTION_LABELS = { "inspect_script_plan": "查看整棵任务树", "plan_script_tasks": "规划新 Task", "dispatch_script_tasks": "派发 Task 执行", "validate_attempt": "读取独立验收", "decide_script_task": "决定下一步", "read_input_snapshot": "读取冻结输入", } class RunNotFound(FileNotFoundError): pass def agent_data_root() -> Path: configured = os.getenv("SCRIPT_BUILD_AGENT_DATA_ROOT") return Path(configured).expanduser().resolve() if configured else DEFAULT_AGENT_DATA_ROOT.resolve() def _read_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: value = json.load(handle) return value if isinstance(value, dict) else {} def _message_files(root: Path, trace_id: str) -> list[Path]: return sorted((root / "traces" / trace_id / "messages").glob("*.json")) def _messages(root: Path, trace_id: str) -> list[dict[str, Any]]: return [_read_json(path) for path in _message_files(root, trace_id)] def _build_id_for_root(root: Path, root_trace_id: str) -> int | None: goal_path = root / "traces" / root_trace_id / "goal.json" if goal_path.exists(): match = BUILD_ID_PATTERN.search(str(_read_json(goal_path).get("mission", ""))) if match: return int(match.group(1)) for message in _messages(root, root_trace_id)[:6]: content = message.get("content") if isinstance(content, str): try: payload = json.loads(content) except json.JSONDecodeError: continue if isinstance(payload, dict) and isinstance(payload.get("script_build_id"), int): return payload["script_build_id"] return None def _roots_by_build() -> dict[int, str]: root = agent_data_root() values: dict[int, str] = {} for ledger in sorted((root / "task-ledger").glob("*/orchestration/ledger.json")): root_trace_id = ledger.parents[1].name build_id = _build_id_for_root(root, root_trace_id) if build_id is not None: values[build_id] = root_trace_id return values def _load_run(script_build_id: int) -> tuple[Path, str, dict[str, Any], list[dict[str, Any]]]: root = agent_data_root() root_trace_id = _roots_by_build().get(script_build_id) if not root_trace_id: raise RunNotFound(f"Persisted script build {script_build_id} was not found") ledger_path = root / "task-ledger" / root_trace_id / "orchestration" / "ledger.json" return root, root_trace_id, _read_json(ledger_path), _messages(root, root_trace_id) def _task_kind(task: dict[str, Any]) -> str: if task.get("parent_task_id") is None: return "root" specs = task.get("specs") or [] refs = (specs[-1].get("context_refs") or []) if specs else [] for ref in refs: if isinstance(ref, str) and ref.startswith(TASK_KIND_PREFIX): return ref.removeprefix(TASK_KIND_PREFIX) return "task" def _phase(display_path: str, task_kind: str) -> str: if task_kind == "root": return "mission" if display_path.startswith("0.1"): return "phase-one" if display_path.startswith("0.2"): return "phase-two" return "delivery" def _task_view(task: dict[str, Any]) -> TaskNode: specs = task.get("specs") or [{}] spec = specs[-1] kind = _task_kind(task) path = str(task.get("display_path", "?")) return TaskNode( task_id=str(task.get("task_id", "")), display_path=path, parent_task_id=task.get("parent_task_id"), child_task_ids=list(task.get("child_task_ids") or []), task_kind=kind, phase=_phase(path, kind), objective=str(spec.get("objective") or "未记录目标"), status=str(task.get("status") or "unknown"), blocked_reason=task.get("blocked_reason"), attempt_count=len(task.get("attempt_ids") or []), validation_count=len(task.get("validation_ids") or []), decision_count=len(task.get("decision_ids") or []), current_spec_version=int(task.get("current_spec_version") or 1), acceptance_criteria=list(spec.get("acceptance_criteria") or []), context_refs=list(spec.get("context_refs") or []), created_at=str(task.get("created_at") or ""), updated_at=str(task.get("updated_at") or ""), ) def _parse_object(value: Any) -> Any: if not isinstance(value, str): return value try: return json.loads(value) except json.JSONDecodeError: return value def _bounded_text(value: Any, limit: int = 3200) -> str: if value is None: return "" text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=2) normalized = text.strip() return normalized if len(normalized) <= limit else f"{normalized[:limit]}\n…（已截断）" def _ids_from(value: Any, known_ids: set[str]) -> set[str]: found: set[str] = set() if isinstance(value, dict): for key, item in value.items(): if key in {"task_id", "parent_task_id"} and isinstance(item, str) and item in known_ids: found.add(item) elif key in {"task_ids", "child_task_ids"} and isinstance(item, list): found.update(entry for entry in item if isinstance(entry, str) and entry in known_ids) found.update(_ids_from(item, known_ids)) elif isinstance(value, list): for item in value: found.update(_ids_from(item, known_ids)) return found def _result_status(result: Any) -> str: if isinstance(result, dict): encoded = json.dumps(result, ensure_ascii=False).lower() if any(token in encoded for token in ('"error"', "protocol_violation", "failed")) and '"error": null' not in encoded: return "failed" return "succeeded" return "unknown" def _planner_turns( messages: list[dict[str, Any]], task_views: list[TaskNode] ) -> tuple[list[PlannerTurn], str | None]: known_ids = {task.task_id for task in task_views} root_ids = {task.task_id for task in task_views if task.parent_task_id is None} tool_results = { message.get("tool_call_id"): message for message in messages if message.get("role") == "tool" and message.get("tool_call_id") } visible_ids = set(root_ids) current_phase = "mission" turns: list[PlannerTurn] = [] current_focus: str | None = None for message in messages: if message.get("role") == "user": payload = _parse_object(message.get("content")) if isinstance(payload, dict) and payload.get("phase"): current_phase = f"phase-{payload['phase']}" continue if message.get("role") != "assistant" or not isinstance(message.get("content"), dict): continue content = message["content"] for call in content.get("tool_calls") or []: function = call.get("function") or {} tool_name = str(function.get("name") or "") if tool_name not in PLANNER_TOOLS: continue arguments = _parse_object(function.get("arguments") or "{}") arguments = arguments if isinstance(arguments, dict) else {"raw": arguments} result_message = tool_results.get(call.get("id"), {}) result_content = result_message.get("content") or {} raw_result = result_content.get("result") if isinstance(result_content, dict) else result_content result = _parse_object(raw_result) affected = _ids_from(arguments, known_ids) | _ids_from(result, known_ids) newly_visible = _ids_from(result, known_ids) visible_ids.update(newly_visible) explicit_focus = arguments.get("task_id") or arguments.get("parent_task_id") if not explicit_focus and isinstance(arguments.get("task_ids"), list) and arguments["task_ids"]: explicit_focus = arguments["task_ids"][0] if explicit_focus in known_ids: current_focus = explicit_focus elif affected: current_focus = sorted(affected)[0] order = len(turns) + 1 visible_snapshot = sorted(visible_ids) turns.append( PlannerTurn( turn_id=f"planner-{message.get('sequence', order)}-{len(turns)}", order=order, message_sequence=int(message.get("sequence") or order), phase=current_phase, action=tool_name, action_label=ACTION_LABELS[tool_name], summary=_bounded_text(content.get("text"), 520) or ACTION_LABELS[tool_name], observable_reasoning=_bounded_text(content.get("text"), 2200), result_summary=_bounded_text(result, 900), status=_result_status(result), focus_task_id=current_focus, affected_task_ids=sorted(affected), visible_task_ids=visible_snapshot, tool_name=tool_name, tool_arguments=arguments, tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0), created_at=str(message.get("created_at") or ""), ) ) if turns: turns[-1].visible_task_ids = sorted(known_ids) return turns, current_focus def _checkpoint(tasks: list[TaskNode]) -> str: kinds = {task.task_kind for task in tasks} if "candidate-portfolio" in kinds: if "compose" in kinds: return "Phase 2 · Compose 子任务回流" return "Phase 2 · Candidate Portfolio" if any(task.task_kind == "direction" and task.status == "completed" for task in tasks): return "Phase 1 · Direction 已验收" return "Phase 1 · 方向与证据" def _current_gap(tasks: list[TaskNode]) -> str: candidates = [task for task in tasks if task.status in {"needs_replan", "failed", "blocked"}] candidates.sort(key=lambda item: (item.display_path.count("."), item.updated_at), reverse=True) if candidates: task = candidates[0] reason = task.blocked_reason or task.status return f"{task.display_path} {task.task_kind} 等待 Planner 处理：{reason}" waiting = [task for task in tasks if task.status == "waiting_children"] if waiting: task = sorted(waiting, key=lambda item: item.display_path.count("."), reverse=True)[0] return f"{task.display_path} 正在等待子 Task 结果返回" return "当前持久化快照没有未处理缺口" def _run_summary( script_build_id: int, root_trace_id: str, ledger: dict[str, Any], tasks: list[TaskNode], turns: list[PlannerTurn], focus: str | None, ) -> RunSummary: root_id = ledger.get("root_task_id") root_task = next((task for task in tasks if task.task_id == root_id), tasks[0]) return RunSummary( script_build_id=script_build_id, title=root_task.objective, status=root_task.status, checkpoint=_checkpoint(tasks), root_trace_id=root_trace_id, revision=int(ledger.get("revision") or 0), task_count=len(tasks), attempt_count=len(ledger.get("attempts") or {}), planner_turn_count=len(turns), current_focus_task_id=focus, current_gap=_current_gap(tasks), created_at=root_task.created_at, updated_at=max((task.updated_at for task in tasks), default=root_task.updated_at), ) def execution_view(script_build_id: int) -> ExecutionView: _, root_trace_id, ledger, messages = _load_run(script_build_id) tasks = [_task_view(item) for item in (ledger.get("tasks") or {}).values()] tasks.sort(key=lambda item: [int(part) for part in item.display_path.split(".")]) turns, focus = _planner_turns(messages, tasks) run = _run_summary(script_build_id, root_trace_id, ledger, tasks, turns, focus) warnings = [ "树的生长顺序来自 Planner 工具结果；Task 状态来自当前 ledger，不伪造历史状态回放。", "Reasoning 仅展示模型主动输出的说明、工具参数与验收理由，不等同于隐藏思维链。", ] return ExecutionView( schema_version="script-build-execution-view/v1", data_mode="persisted-real-run", generated_at=datetime.now(timezone.utc).isoformat(), run=run, planner_turns=turns, tasks=tasks, snapshot_basis="Planner 工具调用顺序 + 当前 TaskLedger 持久化快照", warnings=warnings, ) def list_runs() -> list[RunSummary]: values: list[RunSummary] = [] for build_id in sorted(_roots_by_build(), reverse=True): try: values.append(execution_view(build_id).run) except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): continue return values def _contract(root: Path, root_trace_id: str, task: TaskNode) -> dict[str, Any] | None: for ref in task.context_refs: if ref.startswith(CONTRACT_PREFIX): digest = ref.removeprefix(CONTRACT_PREFIX) path = root / "script-task-contracts" / root_trace_id / "sha256" / digest if path.exists(): return _read_json(path) return None def _step_title(tool_name: str) -> str: if tool_name.startswith("read_"): return f"读取数据 · {tool_name}" if tool_name.startswith(("create_", "batch_create_", "append_")): return f"写入草稿 · {tool_name}" if tool_name.startswith(("update_", "batch_update_")): return f"修改草稿 · {tool_name}" if tool_name == "submit_attempt": return "提交 Attempt" return f"调用工具 · {tool_name}" def _agent_steps(root: Path, trace_id: str | None) -> tuple[str, list[AgentStep]]: if not trace_id: return "", [] prompt = "" steps: list[AgentStep] = [] for message in _messages(root, trace_id): role = str(message.get("role") or "unknown") content = message.get("content") if role == "system": prompt = _bounded_text(content, 9000) continue if role == "user": steps.append( AgentStep( sequence=int(message.get("sequence") or 0), role=role, step_kind="instruction", title="冻结的 TaskSpec", text=_bounded_text(content), tool_name=None, tool_arguments=None, tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0), duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""), ) ) continue if role == "assistant" and isinstance(content, dict): text = _bounded_text(content.get("text"), 2200) if text: steps.append( AgentStep( sequence=int(message.get("sequence") or 0), role=role, step_kind="explanation", title="Worker 的可观察说明", text=text, tool_name=None, tool_arguments=None, tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0), duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""), ) ) for call in content.get("tool_calls") or []: function = call.get("function") or {} name = str(function.get("name") or "unknown_tool") args = _parse_object(function.get("arguments") or "{}") steps.append( AgentStep( sequence=int(message.get("sequence") or 0), role=role, step_kind="tool_call", title=_step_title(name), text="", tool_name=name, tool_arguments=args if isinstance(args, dict) else {"raw": args}, tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0), duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""), ) ) elif role == "tool": tool_name = content.get("tool_name") if isinstance(content, dict) else None result = content.get("result") if isinstance(content, dict) else content steps.append( AgentStep( sequence=int(message.get("sequence") or 0), role=role, step_kind="tool_result", title=f"工具返回 · {tool_name or 'unknown'}", text=_bounded_text(_parse_object(result)), tool_name=tool_name, tool_arguments=None, tokens=0, cost=0, duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""), ) ) return prompt, steps def _values_for_task(values: dict[str, Any], task_id: str) -> list[dict[str, Any]]: return [value for value in values.values() if value.get("task_id") == task_id] def _first_for_attempt(values: Iterable[dict[str, Any]], attempt_id: str) -> dict[str, Any] | None: return next((value for value in values if value.get("attempt_id") == attempt_id), None) def task_detail(script_build_id: int, task_id: str) -> TaskDetail: root, root_trace_id, ledger, _ = _load_run(script_build_id) raw_task = (ledger.get("tasks") or {}).get(task_id) if not raw_task: raise RunNotFound(f"Task {task_id} was not found in build {script_build_id}") task = _task_view(raw_task) contract = _contract(root, root_trace_id, task) validations = _values_for_task(ledger.get("validations") or {}, task_id) decisions = _values_for_task(ledger.get("decisions") or {}, task_id) raw_attempts = _values_for_task(ledger.get("attempts") or {}, task_id) attempt_details: list[AttemptDetail] = [] reads: list[dict[str, Any]] = [] outputs: list[dict[str, Any]] = [] for attempt in sorted(raw_attempts, key=lambda item: item.get("created_at") or ""): prompt, steps = _agent_steps(root, attempt.get("worker_trace_id")) for step in steps: if step.step_kind == "tool_call" and step.tool_name and step.tool_name.startswith(("read_", "get_", "search_", "query_", "inspect_")): reads.append({"attempt_id": attempt.get("attempt_id"), "tool_name": step.tool_name, "arguments": step.tool_arguments, "proof": "worker_trace_tool_call"}) submission = attempt.get("submission") if isinstance(submission, dict): for artifact in submission.get("artifact_refs") or []: outputs.append({"attempt_id": attempt.get("attempt_id"), "artifact": artifact, "proof": "attempt_submission"}) validation = _first_for_attempt(validations, str(attempt.get("attempt_id"))) decision = _first_for_attempt(decisions, str(attempt.get("attempt_id"))) stats = attempt.get("execution_stats") or {} attempt_details.append( AttemptDetail( attempt_id=str(attempt.get("attempt_id") or ""), task_id=task_id, status=str(attempt.get("status") or "unknown"), worker_preset=attempt.get("worker_preset"), worker_trace_id=attempt.get("worker_trace_id"), execution_mode=attempt.get("execution_mode"), started_at=attempt.get("started_at"), completed_at=attempt.get("completed_at"), total_tokens=int(stats.get("total_tokens") or 0), total_cost=float(stats.get("total_cost") or 0), error=attempt.get("error"), submission=submission, validation=validation, decision=decision, prompt=prompt, steps=steps, ) ) lifecycle = [ LifecycleEvent(event_id=f"task-{task_id}", kind="task", title="Planner 创建 Task", status="created", summary=task.objective, created_at=task.created_at) ] for attempt in raw_attempts: lifecycle.append(LifecycleEvent(event_id=str(attempt.get("attempt_id")), kind="attempt", title="Worker 执行 Attempt", status=str(attempt.get("status")), summary=str(attempt.get("worker_preset") or "worker"), created_at=str(attempt.get("created_at") or ""), attempt_id=attempt.get("attempt_id"))) for validation in validations: lifecycle.append(LifecycleEvent(event_id=str(validation.get("validation_id")), kind="validation", title="Validator 独立验收", status=str(validation.get("verdict") or validation.get("status")), summary=str(validation.get("recommendation") or validation.get("summary") or ""), created_at=str(validation.get("created_at") or ""), attempt_id=validation.get("attempt_id"))) for decision in decisions: lifecycle.append(LifecycleEvent(event_id=str(decision.get("decision_id")), kind="decision", title="Planner 决定下一步", status=str(decision.get("action")), summary=str(decision.get("reason") or ""), created_at=str(decision.get("created_at") or ""), attempt_id=decision.get("attempt_id"))) lifecycle.sort(key=lambda item: item.created_at) inherited: list[dict[str, Any]] = [{"ref": ref, "proof": "task_spec_context_ref", "usage": "在冻结输入闭包中"} for ref in task.context_refs] if contract: for ref in contract.get("input_decision_refs") or []: inherited.append({"ref": ref, "proof": "task_contract_input_decision_ref", "usage": "显式接受的上游结果"}) if contract.get("base_artifact_ref"): inherited.append({"ref": contract["base_artifact_ref"], "proof": "task_contract_base_artifact_ref", "usage": "本 Attempt 的基础产物"}) adoption = [{"decision_id": value.get("decision_id"), "action": value.get("action"), "reason": value.get("reason"), "proof": "planner_decision"} for value in decisions] return TaskDetail( task=task, contract=contract, lifecycle=lifecycle, attempts=attempt_details, data_usage=DataUsage(inherited_inputs=inherited, confirmed_reads=reads, produced_outputs=outputs, adoption=adoption), )