script-build-and-framework-tmp/visualization/backend/tests/test_api.py

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)

    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_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"][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(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