test(可视化后端): 用真实 E2E 运行验证投影接口

基于持久化 build 900029 与 900031 验证运行发现、全局 Planner 轴、动态 Task 树、任务生命周期、Worker loop、数据证明和 404 行为，不再依赖构造的假图数据。

visualization/backend/tests/test_api.py

@@ -1,204 +1,62 @@
 from fastapi.testclient import TestClient
 
-from app.contracts import SCHEMA_CATALOG
-from app.fake_data import (
-    ARTIFACT_DIGESTS,
-    GRAPH,
-    RUN_ID,
-    canonical_digest,
-    canonical_wire_digest,
-)
 from app.main import app
 
 client = TestClient(app)
 
 
-def test_health_declares_fake_mode() -> None:
+def test_health_declares_real_persisted_mode() -> None:
     response = client.get("/api/health")
     assert response.status_code == 200
-    assert response.json()["data_mode"] == "fake"
+    assert response.json()["data_mode"] == "persisted-real-run"
 
 
-def test_graph_is_fine_grained_and_connected() -> None:
-    response = client.get(f"/api/runs/{RUN_ID}/graph")
+def test_real_e2e_runs_900029_and_900031_are_discoverable() -> None:
+    response = client.get("/api/runs")
     assert response.status_code == 200
-    body = response.json()
-    assert body["data_mode"] == "fake"
-    assert len(body["nodes"]) >= 55
-    assert len(body["edges"]) >= 50
-    assert {item["phase"] for item in body["nodes"]} == {
-        "phase-1",
-        "phase-2",
-        "phase-3",
-        "delivery",
-    }
-    assert {item["node_type"] for item in body["nodes"]} >= {
-        "contract",
-        "operation",
-        "attempt",
-        "artifact",
-        "validation",
-        "decision",
-        "publication",
-        "transaction",
-        "readback",
-    }
-
-
-def test_every_fake_record_uses_exact_contract_fields() -> None:
-    used_models: set[str] = set()
-    for node in GRAPH.nodes:
-        model = node.record.model_name
-        used_models.add(model)
-        assert set(node.record.payload) == set(SCHEMA_CATALOG[model]), node.id
-    required = {
-        "ScriptBuildRecord",
-        "ScriptBuildInputSnapshotV1",
-        "MissionBinding",
-        "MissionOwnerToken",
-        "ScriptTaskContractV1",
-        "OperationView",
-        "AttemptView",
-        "ValidationView",
-        "PlannerDecisionView",
-        "EvidenceRecordV1",
-        "ScriptDirectionArtifactV1",
-        "StructureArtifactV1",
-        "ParagraphArtifactV1",
-        "ElementSetArtifactV1",
-        "ComparisonArtifactV1",
-        "StructuredScriptArtifactV1",
-        "CandidatePortfolioArtifactV1",
-        "RootDeliveryManifestV1",
-        "LegacyProjectionCanonicalV1",
-        "Publication",
-        "PublicationResult",
-        "HttpCommandRecord",
-        "InputSnapshotRow",
-        "ArtifactVersionRow",
-        "ParagraphRow",
-        "ElementRow",
-        "ParagraphElementLinkRow",
-    }
-    assert required <= used_models
-
-
-def test_manifest_and_legacy_projection_use_real_logical_closure() -> None:
-    records = {node.record.model_name: node.record.payload for node in GRAPH.nodes}
-    canonical = records["LegacyProjectionCanonicalV1"]
-    manifest = records["RootDeliveryManifestV1"]
-    assert manifest["legacy_projection_digest"] == canonical_wire_digest(canonical)
-    assert manifest["input_closure_digest"] != manifest["legacy_projection_digest"]
-    for paragraph in canonical["paragraphs"]:
-        assert "logical_key" in paragraph
-        assert "parent_logical_key" in paragraph
-        assert not {"paragraph_id", "parent_id", "is_active"} & set(paragraph)
-    for element in canonical["elements"]:
-        assert "logical_key" in element
-        assert not {"element_id", "is_active"} & set(element)
-    for link in canonical["paragraph_element_links"]:
-        assert set(link) == {"paragraph_logical_key", "element_logical_key"}
-
-
-def test_artifact_refs_pin_specific_kind_and_canonical_digest() -> None:
-    attempts = [
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.record.model_name == "AttemptView"
-    ]
-    for attempt in attempts:
-        for ref in attempt["submission"]["artifact_refs"]:
-            version = int(ref["version"])
-            assert ref["kind"] != "business-artifact"
-            assert ref["digest"] == ARTIFACT_DIGESTS[version]
-    node_ids = {node.id for node in GRAPH.nodes}
-    assert "compose:alternate:artifact" in node_ids
-    assert "compose:alternate:decision" in node_ids
-
-
-def test_root_task_has_no_self_parent_and_declares_children() -> None:
-    root = next(
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.record.model_name == "TaskView"
-        and node.record.payload["task_id"] == "task-root"
-    )
-    assert root["parent_task_id"] is None
-    assert root["child_task_ids"] == ["direction", "portfolio"]
-    contract = next(
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.id == "root-delivery:contract"
-    )
-    assert contract["scope_ref"] == "script-build://scope/root"
-    assert contract["write_scope"] == []
-    assert [item["expected_task_kind"] for item in contract["input_decision_refs"]] == [
-        "direction",
-        "candidate-portfolio",
-    ]
-    attempt = next(
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.id == "root-delivery:attempt"
-    )
-    assert attempt["accepted_child_decision_ids"] == [
-        "decision-direction",
-        "decision-portfolio",
-    ]
-
-
-def test_input_snapshot_row_and_logical_view_share_one_canonical_payload() -> None:
-    row = next(
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.record.model_name == "InputSnapshotRow"
-    )
-    logical = next(
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.record.model_name == "ScriptBuildInputSnapshotV1"
-    )
-    assert logical["snapshot_id"] == str(row["id"]) == "301"
-    assert row["canonical_sha256"] == canonical_digest(row["canonical_json"])
-    assert logical["canonical_sha256"] == f"sha256:{row['canonical_sha256']}"
-    for key in (
-        "script_build_id",
-        "execution_id",
-        "topic_build_id",
-        "topic_id",
-        "topic",
-        "account",
-        "persona_points",
-        "section_patterns",
-        "strategies",
-        "prompt_manifest",
-        "datasource_manifest",
-        "model_manifest",
-    ):
-        assert logical[key] == row["canonical_json"][key]
-
-
-def test_final_physical_row_counts_match_canonical_counts() -> None:
-    canonical = next(
-        node.record.payload
-        for node in GRAPH.nodes
-        if node.record.model_name == "LegacyProjectionCanonicalV1"
-    )
-    counts = {
-        model: sum(1 for node in GRAPH.nodes if node.record.model_name == model)
-        for model in ("ParagraphRow", "ElementRow", "ParagraphElementLinkRow")
-    }
-    assert counts == {
-        "ParagraphRow": canonical["paragraph_count"],
-        "ElementRow": canonical["element_count"],
-        "ParagraphElementLinkRow": canonical["link_count"],
-    }
-
-
-def test_node_detail_and_not_found() -> None:
-    node_id = GRAPH.nodes[10].id
-    response = client.get(f"/api/runs/{RUN_ID}/nodes/{node_id}")
+    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 response.status_code == 200
-    assert response.json()["node"]["id"] == node_id
-    assert client.get(f"/api/runs/999/nodes/{node_id}").status_code == 404
-    assert client.get(f"/api/runs/{RUN_ID}/nodes/missing").status_code == 404
+    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 body["data_usage"]["confirmed_reads"]
+    assert body["data_usage"]["produced_outputs"]
+
+
+def test_unknown_run_and_task_are_404() -> None:
+    assert client.get("/api/runs/999999/execution").status_code == 404
+    assert client.get("/api/runs/900029/tasks/missing").status_code == 404