feat(可视化后端): 从持久化 Ledger 与 Trace 投影真实运行

扫描 agent-data 中的 task-ledger 和 trace message，按 script_build_id 定位真实 Root。解析任务类型、动态层级、Planner 工具调用、Attempt/Validation、数据读取证据与生命周期事件，并提供运行列表、全局执行视图和任务详情。

visualization/backend/app/real_runs.py

@@ -0,0 +1,496 @@
+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),
+    )