测试(可视化): 使用确定性持久化运行夹具

移除对已清理 Build 900029 和 900031 的本地数据依赖，改用 tmp_path 构造 Root Trace、TaskLedger 和 Worker Trace。\n\n覆盖无 goal.json 的 explicit 运行发现、旧 GoalTree 回退、动态任务树、生命周期证据与 404 边界。

visualization/backend/tests/test_api.py

@@ -1,62 +1,302 @@
+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")
+
+
+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": [
+                {
+                    "id": call_id,
+                    "function": {"name": name, "arguments": json.dumps(arguments)},
+                }
+            ],
+        },
+        "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)
 
-def test_health_declares_real_persisted_mode() -> None:
+    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_real_e2e_runs_900029_and_900031_are_discoverable() -> None:
+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
-    by_id = {run["script_build_id"]: run for run in response.json()}
-    assert {900029, 900031} <= set(by_id)
-    assert by_id[900029]["task_count"] >= 11
-    assert by_id[900031]["task_count"] >= 8
-    assert by_id[900029]["planner_turn_count"] > 10
-    assert by_id[900031]["planner_turn_count"] > 10
-
-
-def test_execution_projects_global_planner_and_dynamic_task_tree() -> None:
-    for build_id in (900029, 900031):
-        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 any(task["task_kind"] == "direction" for task in body["tasks"])
-        assert any(task["task_kind"] == "compose" for task in body["tasks"])
-        assert {turn["tool_name"] for turn in body["planner_turns"]} >= {
-            "inspect_script_plan",
-            "plan_script_tasks",
-            "dispatch_script_tasks",
-            "decide_script_task",
-        }
-        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() -> None:
-    execution = client.get("/api/runs/900029/execution").json()
-    paragraph = next(task for task in execution["tasks"] if task["task_kind"] == "paragraph")
-    response = client.get(f"/api/runs/900029/tasks/{paragraph['task_id']}")
+    assert [run["script_build_id"] for run in response.json()] == [BUILD_ID]
+
+
+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
+
+
+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",
+    }
+    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"]
-    steps = body["attempts"][0]["steps"]
-    assert any(step["step_kind"] == "tool_call" for step in steps)
-    assert body["attempts"][0]["prompt"].startswith("Produce one independently testable Paragraph")
+    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"]
+    )
     assert body["data_usage"]["confirmed_reads"]
     assert body["data_usage"]["produced_outputs"]
 
 
-def test_unknown_run_and_task_are_404() -> None:
+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("/api/runs/900029/tasks/missing").status_code == 404
+    assert client.get(f"/api/runs/{BUILD_ID}/tasks/missing").status_code == 404