可视化：以 Host Journey API 替换本地运行扫描

后端 API 改为代理真实 Script Build Host 并输出 Journey，删除本地 real_runs 文件扫描和旧响应拼装；同步更新 API 健康、列表、详情与错误测试。

visualization/backend/app/main.py

@@ -1,15 +1,29 @@
-from __future__ import annotations
+from contextlib import asynccontextmanager
+from typing import Any
 
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 
-from .models import ExecutionView, RunSummary, TaskDetail
-from .real_runs import RunNotFound, agent_data_root, execution_view, list_runs, task_detail
+from .host_client import HostApiError, HostClient
+from .journey import build_journey
+from .models import JourneyView, RunSummary
 
-app = FastAPI(title="Script Build Execution Observer", version="2.0.0")
+host = HostClient()
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    yield
+    close = getattr(host, "aclose", None)
+    if close is not None:
+        await close()
+
+
+app = FastAPI(title="Script Build Journey", version="0.1.0", lifespan=lifespan)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["http://127.0.0.1:3008", "http://localhost:3008"],
+    allow_credentials=True,
     allow_methods=["GET"],
     allow_headers=["*"],
 )
@@ -17,31 +31,44 @@ app.add_middleware(
 
 @app.get("/api/health")
 def health() -> dict[str, str]:
-    root = agent_data_root()
     return {
-        "status": "ok" if root.exists() else "degraded",
-        "data_mode": "persisted-real-run",
-        "schema_version": "script-build-execution-view/v1",
-        "agent_data_root": str(root),
+        "status": "ok",
+        "service": "script-build-journey",
     }
 
 
 @app.get("/api/runs", response_model=list[RunSummary])
-def runs() -> list[RunSummary]:
-    return list_runs()
+async def runs(request: Request) -> list[RunSummary]:
+    payload = await _host_call(host.list_runs(request.headers))
+    return [
+        RunSummary(
+            script_build_id=int(item["id"]),
+            status=str(item.get("status") or "unknown"),
+            summary=str(item.get("summary") or f"Script Build {item['id']}"),
+            started_at=item.get("start_time"),
+            completed_at=item.get("end_time"),
+        )
+        for item in payload.get("items") or []
+        if isinstance(item, dict) and item.get("id") is not None
+    ]
 
 
-@app.get("/api/runs/{script_build_id}/execution", response_model=ExecutionView)
-def execution(script_build_id: int) -> ExecutionView:
-    try:
-        return execution_view(script_build_id)
-    except RunNotFound as exc:
-        raise HTTPException(status_code=404, detail=str(exc)) from exc
+@app.get("/api/runs/{script_build_id}/journey", response_model=JourneyView)
+async def journey(script_build_id: int, request: Request) -> JourneyView:
+    source = await _host_call(host.journey_source(script_build_id, request.headers))
+    return build_journey(script_build_id, source)
+
+
+@app.get("/api/runs/{script_build_id}/artifacts/{artifact_version_id}")
+async def artifact(
+    script_build_id: int, artifact_version_id: int, request: Request
+) -> dict[str, Any]:
+    return await _host_call(host.artifact(script_build_id, artifact_version_id, request.headers))
 
 
-@app.get("/api/runs/{script_build_id}/tasks/{task_id}", response_model=TaskDetail)
-def detail(script_build_id: int, task_id: str) -> TaskDetail:
+async def _host_call(awaitable: Any) -> dict[str, Any]:
     try:
-        return task_detail(script_build_id, task_id)
-    except RunNotFound as exc:
-        raise HTTPException(status_code=404, detail=str(exc)) from exc
+        value = await awaitable
+    except HostApiError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc
+    return value

visualization/backend/app/real_runs.py

@@ -1,505 +0,0 @@
-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:
-    meta_path = root / "traces" / root_trace_id / "meta.json"
-    if meta_path.exists():
-        context = _read_json(meta_path).get("context")
-        if isinstance(context, dict):
-            build_id = context.get("script_build_id")
-            if isinstance(build_id, int) and not isinstance(build_id, bool):
-                return build_id
-
-    # Legacy traces may only expose the Build ID through GoalTree or early messages.
-    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),
-    )

visualization/backend/tests/test_api.py

@@ -1,302 +1,46 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-
-import pytest
 from fastapi.testclient import TestClient
 
-from app.main import app
-from app.real_runs import _build_id_for_root
-
-client = TestClient(app)
-BUILD_ID = 990001
-ROOT_TRACE_ID = "root-990001"
-NOW = "2026-07-21T00:00:00+00:00"
-
-
-def _write_json(path: Path, value: object) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+import app.main as main
 
 
-def _task(
-    task_id: str,
-    *,
-    parent_task_id: str | None,
-    display_path: str,
-    kind: str,
-    objective: str,
-    child_task_ids: list[str] | None = None,
-    attempt_ids: list[str] | None = None,
-    validation_ids: list[str] | None = None,
-    decision_ids: list[str] | None = None,
-) -> dict[str, object]:
-    context_refs = [] if kind == "root" else [f"script-build://task-kinds/{kind}"]
-    return {
-        "task_id": task_id,
-        "parent_task_id": parent_task_id,
-        "display_path": display_path,
-        "specs": [
-            {
-                "version": 1,
-                "objective": objective,
-                "acceptance_criteria": [
-                    {
-                        "criterion_id": f"{kind}-closed",
-                        "description": "产物闭合",
-                        "hard": True,
-                    }
-                ],
-                "context_refs": context_refs,
-            }
-        ],
-        "current_spec_version": 1,
-        "status": "completed",
-        "child_task_ids": child_task_ids or [],
-        "attempt_ids": attempt_ids or [],
-        "validation_ids": validation_ids or [],
-        "decision_ids": decision_ids or [],
-        "blocked_reason": None,
-        "created_at": NOW,
-        "updated_at": NOW,
-    }
-
-
-def _planner_message(sequence: int, call_id: str, name: str, arguments: dict) -> dict:
-    return {
-        "sequence": sequence,
-        "role": "assistant",
-        "content": {
-            "text": f"执行 {name}",
-            "tool_calls": [
+class FakeHost:
+    async def list_runs(self, _headers):
+        return {
+            "items": [
                 {
-                    "id": call_id,
-                    "function": {"name": name, "arguments": json.dumps(arguments)},
+                    "id": 42,
+                    "status": "running",
+                    "summary": "正在生成脚本",
+                    "start_time": "2026-07-21T01:00:00Z",
+                    "end_time": None,
                 }
-            ],
-        },
-        "tokens": 10,
-        "cost": 0,
-        "created_at": NOW,
-    }
-
-
-def _tool_result(sequence: int, call_id: str, name: str, result: dict) -> dict:
-    return {
-        "sequence": sequence,
-        "role": "tool",
-        "tool_call_id": call_id,
-        "content": {"tool_name": name, "result": result},
-        "created_at": NOW,
-    }
-
-
-@pytest.fixture
-def persisted_run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
-    root = tmp_path / "agent-data"
-    monkeypatch.setenv("SCRIPT_BUILD_AGENT_DATA_ROOT", str(root))
-    _write_json(
-        root / "traces" / ROOT_TRACE_ID / "meta.json",
-        {
-            "trace_id": ROOT_TRACE_ID,
-            "mode": "agent",
-            "agent_role": "explicit",
-            "context": {"script_build_id": BUILD_ID, "phase": "one"},
-        },
-    )
-
-    root_messages = root / "traces" / ROOT_TRACE_ID / "messages"
-    messages = [
-        {"sequence": 1, "role": "user", "content": '{"phase":"one"}', "created_at": NOW},
-        _planner_message(2, "call-plan-direction", "plan_script_tasks", {"parent_task_id": "root-task"}),
-        _tool_result(3, "call-plan-direction", "plan_script_tasks", {"task_ids": ["direction-task"]}),
-        _planner_message(4, "call-dispatch-direction", "dispatch_script_tasks", {"task_ids": ["direction-task"]}),
-        _tool_result(5, "call-dispatch-direction", "dispatch_script_tasks", {"task_ids": ["direction-task"]}),
-        {"sequence": 6, "role": "user", "content": '{"phase":"two"}', "created_at": NOW},
-        _planner_message(7, "call-plan-paragraph", "inspect_script_plan", {"task_id": "root-task"}),
-        _tool_result(8, "call-plan-paragraph", "inspect_script_plan", {"task_ids": ["paragraph-task"]}),
-        _planner_message(9, "call-decide-paragraph", "decide_script_task", {"task_id": "paragraph-task"}),
-        _tool_result(10, "call-decide-paragraph", "decide_script_task", {"task_id": "paragraph-task"}),
-    ]
-    for index, message in enumerate(messages, start=1):
-        _write_json(root_messages / f"{index:04d}.json", message)
-
-    worker_trace_id = "worker-paragraph"
-    worker_messages = root / "traces" / worker_trace_id / "messages"
-    worker_values = [
-        {
-            "sequence": 1,
-            "role": "system",
-            "content": "Produce one independently testable Paragraph from the frozen contract.",
-            "created_at": NOW,
-        },
-        {
-            "sequence": 2,
-            "role": "user",
-            "content": '{"task_spec":{"objective":"write opening"}}',
-            "created_at": NOW,
-        },
-        _planner_message(3, "call-read-input", "read_input_snapshot", {}),
-        _tool_result(4, "call-read-input", "read_input_snapshot", {"topic": "fixture"}),
-    ]
-    for index, message in enumerate(worker_values, start=1):
-        _write_json(worker_messages / f"{index:04d}.json", message)
-
-    tasks = {
-        "root-task": _task(
-            "root-task",
-            parent_task_id=None,
-            display_path="0",
-            kind="root",
-            objective="完成脚本构建",
-            child_task_ids=["direction-task", "paragraph-task"],
-        ),
-        "direction-task": _task(
-            "direction-task",
-            parent_task_id="root-task",
-            display_path="0.1",
-            kind="direction",
-            objective="确定创作方向",
-        ),
-        "paragraph-task": _task(
-            "paragraph-task",
-            parent_task_id="root-task",
-            display_path="0.2",
-            kind="paragraph",
-            objective="创作开场段落",
-            attempt_ids=["attempt-paragraph"],
-            validation_ids=["validation-paragraph"],
-            decision_ids=["decision-paragraph"],
-        ),
-    }
-    ledger = {
-        "root_task_id": "root-task",
-        "root_objective": "完成脚本构建",
-        "revision": 4,
-        "tasks": tasks,
-        "attempts": {
-            "attempt-paragraph": {
-                "attempt_id": "attempt-paragraph",
-                "task_id": "paragraph-task",
-                "status": "completed",
-                "worker_preset": "script_paragraph_worker",
-                "worker_trace_id": worker_trace_id,
-                "execution_mode": "local",
-                "created_at": NOW,
-                "started_at": NOW,
-                "completed_at": NOW,
-                "execution_stats": {"total_tokens": 20, "total_cost": 0},
-                "submission": {
-                    "artifact_refs": [
-                        {
-                            "uri": "script-build://artifacts/paragraph/1",
-                            "kind": "paragraph",
-                            "digest": "sha256:fixture",
-                        }
-                    ]
-                },
-                "error": None,
-            }
-        },
-        "validations": {
-            "validation-paragraph": {
-                "validation_id": "validation-paragraph",
-                "task_id": "paragraph-task",
-                "attempt_id": "attempt-paragraph",
-                "verdict": "passed",
-                "recommendation": "accept",
-                "created_at": NOW,
-            }
-        },
-        "decisions": {
-            "decision-paragraph": {
-                "decision_id": "decision-paragraph",
-                "task_id": "paragraph-task",
-                "attempt_id": "attempt-paragraph",
-                "action": "accept",
-                "reason": "段落通过独立验收",
-                "created_at": NOW,
-            }
-        },
-    }
-    _write_json(
-        root / "task-ledger" / ROOT_TRACE_ID / "orchestration" / "ledger.json",
-        ledger,
-    )
-    return root
-
-
-def test_health_declares_real_persisted_mode(persisted_run: Path) -> None:
-    response = client.get("/api/health")
-    assert response.status_code == 200
-    assert response.json()["data_mode"] == "persisted-real-run"
-    assert response.json()["agent_data_root"] == str(persisted_run)
-
+            ]
+        }
 
-def test_explicit_run_is_discovered_from_trace_meta_without_goal_tree(
-    persisted_run: Path,
-) -> None:
-    assert not (persisted_run / "traces" / ROOT_TRACE_ID / "goal.json").exists()
-    response = client.get("/api/runs")
-    assert response.status_code == 200
-    assert [run["script_build_id"] for run in response.json()] == [BUILD_ID]
 
+def test_runs_are_normalized_from_host(monkeypatch) -> None:
+    monkeypatch.setattr(main, "host", FakeHost())
 
-def test_legacy_goal_tree_remains_a_build_id_fallback(tmp_path: Path) -> None:
-    root = tmp_path / "legacy-agent-data"
-    _write_json(
-        root / "traces" / "legacy-root" / "goal.json",
-        {"mission": "Script build 880001"},
-    )
-    assert _build_id_for_root(root, "legacy-root") == 880001
-
+    response = TestClient(main.app).get("/api/runs")
 
-def test_execution_projects_global_planner_and_dynamic_task_tree(
-    persisted_run: Path,
-) -> None:
-    del persisted_run
-    response = client.get(f"/api/runs/{BUILD_ID}/execution")
     assert response.status_code == 200
-    body = response.json()
-    assert body["data_mode"] == "persisted-real-run"
-    assert len([task for task in body["tasks"] if task["parent_task_id"] is None]) == 1
-    assert {task["task_kind"] for task in body["tasks"]} >= {"direction", "paragraph"}
-    assert {turn["tool_name"] for turn in body["planner_turns"]} >= {
-        "inspect_script_plan",
-        "plan_script_tasks",
-        "dispatch_script_tasks",
-        "decide_script_task",
+    assert response.json()[0] == {
+        "script_build_id": 42,
+        "status": "running",
+        "summary": "正在生成脚本",
+        "started_at": "2026-07-21T01:00:00Z",
+        "completed_at": None,
     }
-    visible_counts = [len(turn["visible_task_ids"]) for turn in body["planner_turns"]]
-    assert visible_counts == sorted(visible_counts)
-    assert visible_counts[-1] == len(body["tasks"])
 
 
-def test_task_detail_exposes_lifecycle_worker_loop_and_data_proof(
-    persisted_run: Path,
-) -> None:
-    del persisted_run
-    response = client.get(f"/api/runs/{BUILD_ID}/tasks/paragraph-task")
-    assert response.status_code == 200
-    body = response.json()
-    assert {event["kind"] for event in body["lifecycle"]} == {
-        "task",
-        "attempt",
-        "validation",
-        "decision",
-    }
-    assert body["attempts"][0]["prompt"].startswith(
-        "Produce one independently testable Paragraph"
-    )
-    assert any(
-        step["step_kind"] == "tool_call" for step in body["attempts"][0]["steps"]
+def test_browser_auth_credentials_are_allowed_for_the_frontend() -> None:
+    response = TestClient(main.app).options(
+        "/api/runs",
+        headers={
+            "Origin": "http://127.0.0.1:3008",
+            "Access-Control-Request-Method": "GET",
+        },
     )
-    assert body["data_usage"]["confirmed_reads"]
-    assert body["data_usage"]["produced_outputs"]
-
 
-def test_unknown_run_and_task_are_404(persisted_run: Path) -> None:
-    del persisted_run
-    assert client.get("/api/runs/999999/execution").status_code == 404
-    assert client.get(f"/api/runs/{BUILD_ID}/tasks/missing").status_code == 404
+    assert response.headers["access-control-allow-origin"] == "http://127.0.0.1:3008"
+    assert response.headers["access-control-allow-credentials"] == "true"