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

import json

from app.journey import build_journey
from app.narrator import narrate_contract


def _contract(task_id: str, task_kind: str, goal_ids: list[str]) -> dict:
    return {
        "task_id": task_id,
        "contract_ref": f"script-build://task-contracts/sha256/{task_id}",
        "task_kind": task_kind,
        "objective": "组织目标并形成可执行的内容结构",
        "goal_ids": goal_ids,
        "input_decision_refs": [
            {
                "decision_id": "direction-accepted",
                "expected_task_kind": "direction",
                "artifact_ref": {
                    "uri": "script-build://artifact-versions/9",
                    "kind": "direction",
                },
            }
        ],
        "base_artifact_ref": None,
        "criteria": [
            {"criterion_id": "complete", "description": "结构覆盖所有核心目标", "hard": True}
        ],
    }


def test_contract_narrator_uses_only_frozen_contract_facts() -> None:
    narration = narrate_contract(_contract("task-1", "structure", ["goal-1", "goal-2"]))

    assert narration.action == "创建整体结构 Task"
    assert narration.goals == ["goal-1", "goal-2"]
    assert narration.inputs == ["已接受的创作方向"]
    assert narration.output == "整体结构"
    assert narration.criteria == ["结构覆盖所有核心目标"]


def test_journey_links_real_lifecycle_and_marks_retry_and_parallel_work() -> None:
    source = {
        "snapshot": {
            "root_trace_id": "root-7",
            "root_objective": "为选题生成一份可发布的脚本",
            "tasks": [
                {
                    "task_id": "root-task",
                    "parent_task_id": None,
                    "status": "running",
                    "created_at": "2026-07-21T01:00:00+00:00",
                    "current_spec": {"objective": "root", "context_refs": []},
                    "child_task_ids": ["task-1", "task-2"],
                },
                {
                    "task_id": "task-1",
                    "parent_task_id": "root-task",
                    "status": "needs_replan",
                    "created_at": "2026-07-21T01:01:00+00:00",
                    "current_spec": {"objective": "structure", "context_refs": []},
                    "child_task_ids": [],
                },
                {
                    "task_id": "task-2",
                    "parent_task_id": "root-task",
                    "status": "running",
                    "created_at": "2026-07-21T01:01:00+00:00",
                    "current_spec": {"objective": "paragraph", "context_refs": []},
                    "child_task_ids": [],
                },
            ],
            "attempts": [
                {
                    "attempt_id": "attempt-1",
                    "task_id": "task-1",
                    "worker_trace_id": "worker-1",
                    "status": "submitted",
                    "operation_id": "operation-1",
                    "started_at": "2026-07-21T01:02:00+00:00",
                    "duration_ms": 2000,
                    "execution_stats": {"primary_model": "qwen", "total_tokens": 120},
                    "submission": {
                        "artifact_refs": [
                            {
                                "uri": "script-build://artifact-versions/11",
                                "kind": "structure",
                            }
                        ]
                    },
                },
                {
                    "attempt_id": "attempt-2",
                    "task_id": "task-2",
                    "worker_trace_id": "worker-2",
                    "status": "running",
                    "operation_id": "operation-1",
                    "started_at": "2026-07-21T01:02:00+00:00",
                    "execution_stats": {},
                    "submission": None,
                },
            ],
            "validations": [
                {
                    "validation_id": "validation-1",
                    "task_id": "task-1",
                    "attempt_id": "attempt-1",
                    "status": "completed",
                    "verdict": "failed",
                    "started_at": "2026-07-21T01:03:00+00:00",
                    "duration_ms": 500,
                    "validation_plan": {
                        "mode": "deterministic",
                        "preflight_rule_ids": ["goal-coverage"],
                    },
                    "criterion_results": [
                        {
                            "criterion_id": "complete",
                            "verdict": "failed",
                            "reason": "缺少 goal-2",
                        }
                    ],
                    "execution_stats": {"total_tokens": 0},
                }
            ],
            "decisions": [
                {
                    "decision_id": "decision-1",
                    "task_id": "task-1",
                    "attempt_id": "attempt-1",
                    "validation_id": "validation-1",
                    "action": "retry",
                    "reason": "补齐 goal-2 后重新生成",
                    "created_at": "2026-07-21T01:04:00+00:00",
                }
            ],
            "operations": [
                {
                    "operation_id": "operation-1",
                    "attempt_ids": ["attempt-1", "attempt-2"],
                }
            ],
        },
        "input_summary": {
            "topic": "一次反转如何改变信念",
            "persona": {"account_name": "每天心理学", "point_count": 3},
            "strategies": [{"name": "先冲突后解法"}],
            "input_snapshot_id": "11",
            "input_digest": "sha256:input",
        },
        "contracts": {
            "task-1": _contract("task-1", "structure", ["goal-1", "goal-2"]),
            "task-2": _contract("task-2", "paragraph", ["goal-2"]),
        },
        "messages": {
            "root-7": [
                {
                    "role": "assistant",
                    "sequence": 1,
                    "content": {
                        "text": "当前缺少整体结构，需要组织目标。",
                        "tool_calls": [
                            {
                                "id": "call-1",
                                "function": {
                                    "name": "plan_script_tasks",
                                    "arguments": json.dumps(
                                        {"contracts": [{"task_kind": "structure"}]}
                                    ),
                                },
                            }
                        ],
                    },
                },
                {
                    "role": "tool",
                    "tool_call_id": "call-1",
                    "content": {"result": json.dumps({"task_ids": ["task-1", "task-2"]})},
                },
            ],
            "worker-1": [
                {
                    "role": "assistant",
                    "content": {
                        "text": "我先建立结构骨架。",
                        "tool_calls": [
                            {"function": {"name": "create_script_structure", "arguments": "{}"}}
                        ],
                    },
                }
            ],
            "worker-2": [],
        },
        "events": [
            {"sequence": 10, "payload": {"task_ids": ["task-1", "task-2"]}},
            {"sequence": 20, "payload": {"attempt_id": "attempt-1"}},
            {"sequence": 21, "payload": {"attempt_id": "attempt-2"}},
            {"sequence": 30, "payload": {"validation_id": "validation-1"}},
            {"sequence": 40, "payload": {"decision_id": "decision-1"}},
        ],
    }

    journey = build_journey(7, source)

    assert [step.step_type for step in journey.steps].count("plan") == 2
    execute = next(step for step in journey.steps if step.step_id == "execute:attempt-1")
    assert execute.parallel_total == 2
    assert execute.reasoning_source == "observable"
    validation = next(step for step in journey.steps if step.step_type == "validate")
    assert "缺少 goal-2" in validation.reasoning
    loop = next(edge for edge in journey.edges if edge.relation == "loop")
    assert loop.source == "decide:decision-1"
    assert loop.target == "execute:attempt-1"
    assert any(edge.relation == "validates" for edge in journey.edges)
    assert [step.sequence for step in journey.steps] == list(range(1, len(journey.steps) + 1))