可视化：从真实任务账本投影创作旅程

新增 Journey 聚合器和中文叙述器，将 Host 的 Root、阶段边界、任务、回环、验收和证据转换为稳定展示模型，并覆盖成功、返工、失败与空数据场景。

visualization/backend/app/journey.py

@@ -0,0 +1,540 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, Literal
+
+from .models import (
+    DataRef,
+    InputSummary,
+    JourneyEdge,
+    JourneyStep,
+    JourneyView,
+    ProcessView,
+    StepEvidence,
+)
+from .narrator import narrate_contract, task_label
+
+
+def build_journey(script_build_id: int, source: dict[str, Any]) -> JourneyView:
+    snapshot = source["snapshot"]
+    tasks = {str(item["task_id"]): item for item in snapshot.get("tasks") or []}
+    attempts = {str(item["attempt_id"]): item for item in snapshot.get("attempts") or []}
+    validations = {str(item["validation_id"]): item for item in snapshot.get("validations") or []}
+    decisions = {str(item["decision_id"]): item for item in snapshot.get("decisions") or []}
+    contracts = source.get("contracts") or {}
+    messages = source.get("messages") or {}
+    root_trace_id = str(snapshot.get("root_trace_id") or "")
+    planner_calls = _planner_calls(messages.get(root_trace_id) or [])
+    operations = snapshot.get("operations") or []
+    parallel = _parallel_attempts(operations)
+    event_sequences = _event_sequences(source.get("events") or [])
+    steps: list[JourneyStep] = []
+
+    plan_by_task: dict[str, str] = {}
+    execute_by_attempt: dict[str, str] = {}
+    validation_by_id: dict[str, str] = {}
+
+    for task_id, task in tasks.items():
+        if task.get("parent_task_id") is None or task_id not in contracts:
+            continue
+        contract = contracts[task_id]
+        narration = narrate_contract(contract)
+        call = planner_calls.get(task_id)
+        observable = _plain((call or {}).get("assistant_text"))
+        step_id = f"plan:{task_id}"
+        plan_by_task[task_id] = step_id
+        steps.append(
+            JourneyStep(
+                step_id=step_id,
+                step_type="plan",
+                title=f"Planner 规划·{narration.output}",
+                status=str(task.get("status") or "planned"),
+                task_id=task_id,
+                reasoning=observable or narration.reasoning,
+                reasoning_source="observable" if observable else "contract",
+                process=ProcessView(
+                    label=narration.action,
+                    actor="Planner",
+                    tool_name="plan_script_tasks",
+                    trace_id=root_trace_id,
+                    details={"task_kind": contract.get("task_kind")},
+                ),
+                input_data=_contract_inputs(contract),
+                output_data=[
+                    DataRef(label="TaskContract", kind="contract", ref=contract.get("contract_ref"))
+                ],
+                contract=narration,
+                created_at=str(task.get("created_at") or ""),
+                evidence=StepEvidence(
+                    contract=contract,
+                    tool_calls=[call["tool_call"]] if call and call.get("tool_call") else [],
+                ),
+            )
+        )
+
+    for attempt_id, attempt in attempts.items():
+        task_id = str(attempt.get("task_id") or "")
+        task = tasks.get(task_id) or {}
+        contract = contracts.get(task_id)
+        worker_trace_id = str(attempt.get("worker_trace_id") or "")
+        worker_messages = messages.get(worker_trace_id) or []
+        observable = _assistant_text(worker_messages)
+        narration = narrate_contract(contract) if contract else None
+        tools = _tool_calls(worker_messages)
+        position, total, group_id = parallel.get(attempt_id, (None, None, None))
+        stats = attempt.get("execution_stats") or {}
+        step_id = f"execute:{attempt_id}"
+        execute_by_attempt[attempt_id] = step_id
+        steps.append(
+            JourneyStep(
+                step_id=step_id,
+                step_type="execute",
+                title=f"Worker 执行·{_task_name(task, contract)}",
+                status=str(attempt.get("status") or "unknown"),
+                task_id=task_id,
+                attempt_id=attempt_id,
+                reasoning=observable
+                or (
+                    f"执行任务合同：{contract.get('objective')}"
+                    if contract
+                    else "执行已冻结的任务合同"
+                ),
+                reasoning_source="observable" if observable else "contract",
+                process=ProcessView(
+                    label=_worker_process_label(tools),
+                    actor="Worker",
+                    trace_id=worker_trace_id or None,
+                    details={"execution_mode": attempt.get("execution_mode")},
+                ),
+                input_data=_contract_inputs(contract or {}),
+                output_data=_attempt_outputs(attempt),
+                contract=narration,
+                parallel_group_id=group_id,
+                parallel_position=position,
+                parallel_total=total,
+                model=stats.get("primary_model"),
+                tokens=stats.get("total_tokens"),
+                cost=stats.get("total_cost"),
+                duration_ms=attempt.get("duration_ms"),
+                created_at=str(attempt.get("started_at") or attempt.get("created_at") or ""),
+                evidence=StepEvidence(
+                    contract=contract,
+                    tool_calls=tools,
+                    artifact_refs=_artifact_refs(attempt),
+                ),
+            )
+        )
+
+    for validation_id, validation in validations.items():
+        attempt_id = str(validation.get("attempt_id") or "")
+        attempt = attempts.get(attempt_id) or {}
+        task_id = str(validation.get("task_id") or attempt.get("task_id") or "")
+        plan = validation.get("validation_plan") or {}
+        stats = validation.get("execution_stats") or {}
+        reasoning = _validation_reason(validation)
+        task_name = _task_name(tasks.get(task_id) or {}, contracts.get(task_id))
+        step_id = f"validate:{validation_id}"
+        validation_by_id[validation_id] = step_id
+        steps.append(
+            JourneyStep(
+                step_id=step_id,
+                step_type="validate",
+                title=f"Validator 验收·{task_name}",
+                status=str(validation.get("verdict") or validation.get("status") or "unknown"),
+                task_id=task_id,
+                attempt_id=attempt_id or None,
+                validation_id=validation_id,
+                reasoning=reasoning,
+                reasoning_source="validation",
+                process=ProcessView(
+                    label=(
+                        "确定性规则预检"
+                        if plan.get("mode") == "deterministic"
+                        else "独立 Validator 语义质检"
+                    ),
+                    actor="Validator",
+                    trace_id=validation.get("validator_trace_id"),
+                    details={
+                        "mode": plan.get("mode"),
+                        "preflight_rule_ids": plan.get("preflight_rule_ids") or [],
+                    },
+                ),
+                input_data=_attempt_outputs(attempt),
+                output_data=[
+                    DataRef(
+                        label=f"Validation·{validation.get('verdict') or validation.get('status')}",
+                        kind="validation",
+                        ref=validation_id,
+                        detail=_plain(validation.get("summary")),
+                    )
+                ],
+                model=stats.get("primary_model"),
+                tokens=stats.get("total_tokens"),
+                cost=stats.get("total_cost"),
+                duration_ms=validation.get("duration_ms"),
+                created_at=str(validation.get("started_at") or validation.get("created_at") or ""),
+                evidence=StepEvidence(validation=validation, artifact_refs=_artifact_refs(attempt)),
+            )
+        )
+
+    for decision_id, decision in decisions.items():
+        task_id = str(decision.get("task_id") or "")
+        action = str(decision.get("action") or "unknown").upper()
+        validation_id = str(decision.get("validation_id") or "")
+        attempt_id = str(decision.get("attempt_id") or "")
+        steps.append(
+            JourneyStep(
+                step_id=f"decide:{decision_id}",
+                step_type="decide",
+                title=f"Planner 决定·{_action_label(action)}",
+                status=action.lower(),
+                task_id=task_id,
+                attempt_id=attempt_id or None,
+                validation_id=validation_id or None,
+                decision_id=decision_id,
+                reasoning=_plain(decision.get("reason")) or "Planner 根据验收结果选择下一步",
+                reasoning_source="decision",
+                process=ProcessView(
+                    label=_action_process(action),
+                    actor="Planner",
+                    tool_name="decide_script_task",
+                    trace_id=root_trace_id,
+                    details={
+                        "from_status": decision.get("from_status"),
+                        "to_status": decision.get("to_status"),
+                    },
+                ),
+                input_data=[
+                    DataRef(label="Validation", kind="validation", ref=validation_id or None)
+                ],
+                output_data=[
+                    DataRef(label=_action_label(action), kind="decision", ref=decision_id)
+                ],
+                created_at=str(decision.get("created_at") or ""),
+                evidence=StepEvidence(decision=decision),
+            )
+        )
+
+    steps.sort(key=lambda step: _step_sort_key(step, event_sequences))
+    steps = [step.model_copy(update={"sequence": index}) for index, step in enumerate(steps, 1)]
+    edges = _edges(steps, tasks, plan_by_task, execute_by_attempt, validation_by_id)
+    root_task = next((item for item in tasks.values() if item.get("parent_task_id") is None), {})
+    raw_input = source["input_summary"]
+    return JourneyView(
+        schema_version="script-build-journey/v1",
+        generated_at=datetime.now(timezone.utc).isoformat(),
+        script_build_id=script_build_id,
+        root_trace_id=root_trace_id,
+        status=str(root_task.get("status") or "unknown"),
+        objective=_plain(snapshot.get("root_objective")) or "完成脚本创作与交付",
+        input_summary=InputSummary(
+            topic=_plain(raw_input.get("topic")) or "未记录选题",
+            account_name=(raw_input.get("persona") or {}).get("resolved_account_name")
+            or (raw_input.get("persona") or {}).get("account_name"),
+            persona_point_count=int((raw_input.get("persona") or {}).get("point_count") or 0),
+            strategy_names=[
+                str(item["name"])
+                for item in raw_input.get("strategies") or []
+                if isinstance(item, dict) and item.get("name")
+            ],
+            snapshot_id=str(raw_input.get("input_snapshot_id") or ""),
+            digest=str(raw_input.get("input_digest") or ""),
+        ),
+        steps=steps,
+        edges=edges,
+        warnings=_warnings(steps),
+    )
+
+
+def _planner_calls(messages: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    results = {
+        str(message.get("tool_call_id")): _parse_result(message.get("content"))
+        for message in messages
+        if message.get("role") == "tool" and message.get("tool_call_id")
+    }
+    output: dict[str, dict[str, Any]] = {}
+    last_text = ""
+    for message in sorted(messages, key=lambda value: int(value.get("sequence") or 0)):
+        if message.get("role") != "assistant":
+            continue
+        content = message.get("content")
+        if not isinstance(content, dict):
+            continue
+        if _plain(content.get("text")):
+            last_text = _plain(content.get("text"))
+        for call in content.get("tool_calls") or []:
+            function = call.get("function") or {}
+            if function.get("name") != "plan_script_tasks":
+                continue
+            result = results.get(str(call.get("id"))) or {}
+            for task_id in result.get("task_ids") or []:
+                output[str(task_id)] = {
+                    "assistant_text": last_text,
+                    "tool_call": {
+                        "name": "plan_script_tasks",
+                        "arguments": _parse_json(function.get("arguments")),
+                    },
+                }
+    return output
+
+
+def _tool_calls(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    output: list[dict[str, Any]] = []
+    for message in messages:
+        content = message.get("content")
+        if message.get("role") != "assistant" or not isinstance(content, dict):
+            continue
+        for call in content.get("tool_calls") or []:
+            function = call.get("function") or {}
+            output.append(
+                {
+                    "name": str(function.get("name") or "unknown"),
+                    "arguments": _parse_json(function.get("arguments")),
+                }
+            )
+    return output
+
+
+def _assistant_text(messages: list[dict[str, Any]]) -> str:
+    values = []
+    for message in messages:
+        content = message.get("content")
+        if message.get("role") == "assistant" and isinstance(content, dict):
+            text = _plain(content.get("text"))
+            if text:
+                values.append(text)
+    return " ".join(values[:3])[:1_200]
+
+
+def _contract_inputs(contract: dict[str, Any]) -> list[DataRef]:
+    values = []
+    for item in contract.get("input_decision_refs") or []:
+        if not isinstance(item, dict):
+            continue
+        artifact = item.get("artifact_ref") or {}
+        label = f"已接受的{task_label(str(item.get('expected_task_kind') or '上游成果'))}"
+        values.append(
+            DataRef(
+                label=label,
+                kind=str(artifact.get("kind") or "artifact"),
+                ref=artifact.get("uri"),
+                detail=str(item.get("decision_id") or ""),
+            )
+        )
+    if not values:
+        values.append(DataRef(label="冻结创作输入", kind="input-snapshot"))
+    return values
+
+
+def _attempt_outputs(attempt: dict[str, Any]) -> list[DataRef]:
+    submission = attempt.get("submission") or {}
+    values = []
+    for item in submission.get("artifact_refs") or []:
+        if isinstance(item, dict):
+            values.append(
+                DataRef(
+                    label=str(item.get("summary") or item.get("kind") or "Artifact"),
+                    kind=str(item.get("kind") or "artifact"),
+                    ref=item.get("uri"),
+                )
+            )
+    return values or [DataRef(label="待提交的任务成果", kind="artifact")]
+
+
+def _artifact_refs(attempt: dict[str, Any]) -> list[dict[str, Any]]:
+    submission = attempt.get("submission") or {}
+    return [item for item in submission.get("artifact_refs") or [] if isinstance(item, dict)]
+
+
+def _validation_reason(validation: dict[str, Any]) -> str:
+    summary = _plain(validation.get("summary"))
+    failures = [
+        f"{item.get('criterion_id')}：{_plain(item.get('reason'))}"
+        for item in validation.get("criterion_results") or []
+        if isinstance(item, dict) and str(item.get("verdict")) not in {"passed", "pass"}
+    ]
+    return "；".join(failures) or summary or "Validator 已逐项检查验收标准"
+
+
+def _parallel_attempts(operations: list[dict[str, Any]]) -> dict[str, tuple[int, int, str]]:
+    output: dict[str, tuple[int, int, str]] = {}
+    for operation in operations:
+        attempt_ids = [str(value) for value in operation.get("attempt_ids") or []]
+        if len(attempt_ids) < 2:
+            continue
+        group_id = str(operation.get("operation_id") or "")
+        for index, attempt_id in enumerate(attempt_ids, start=1):
+            output[attempt_id] = (index, len(attempt_ids), group_id)
+    return output
+
+
+def _loop_target(
+    action: str,
+    task_id: str,
+    attempt_id: str,
+    tasks: dict[str, dict[str, Any]],
+    plan_by_task: dict[str, str],
+    execute_by_attempt: dict[str, str],
+) -> str | None:
+    if action == "RETRY":
+        return execute_by_attempt.get(attempt_id)
+    if action == "REVISE":
+        return plan_by_task.get(task_id)
+    if action == "SPLIT":
+        children = tasks.get(task_id, {}).get("child_task_ids") or []
+        return next(
+            (plan_by_task.get(str(value)) for value in children if plan_by_task.get(str(value))),
+            None,
+        )
+    return None
+
+
+def _edges(
+    steps: list[JourneyStep],
+    tasks: dict[str, dict[str, Any]],
+    plan_by_task: dict[str, str],
+    execute_by_attempt: dict[str, str],
+    validation_by_id: dict[str, str],
+) -> list[JourneyEdge]:
+    edges: list[JourneyEdge] = []
+    for left, right in zip(steps, steps[1:]):
+        edges.append(
+            JourneyEdge(
+                edge_id=f"sequence:{left.step_id}:{right.step_id}",
+                source=left.step_id,
+                target=right.step_id,
+                relation="sequence",
+                label="然后",
+            )
+        )
+    for step in steps:
+        if step.step_type == "execute" and step.attempt_id:
+            source = plan_by_task.get(step.task_id)
+            if source:
+                edges.append(_edge(source, step.step_id, "creates", "派发执行"))
+        elif step.step_type == "validate" and step.attempt_id:
+            source = execute_by_attempt.get(step.attempt_id)
+            if source:
+                edges.append(_edge(source, step.step_id, "validates", "提交验收"))
+        elif step.step_type == "decide" and step.validation_id:
+            source = validation_by_id.get(step.validation_id)
+            if source:
+                edges.append(_edge(source, step.step_id, "decides", "验收结论"))
+            target = _loop_target(
+                step.status.upper(),
+                step.task_id,
+                step.attempt_id or "",
+                tasks,
+                plan_by_task,
+                execute_by_attempt,
+            )
+            if target:
+                edges.append(_edge(step.step_id, target, "loop", _loop_label(step.status.upper())))
+    return edges
+
+
+def _edge(
+    source: str,
+    target: str,
+    relation: Literal["creates", "validates", "decides", "loop"],
+    label: str,
+) -> JourneyEdge:
+    return JourneyEdge(
+        edge_id=f"{relation}:{source}:{target}",
+        source=source,
+        target=target,
+        relation=relation,
+        label=label,
+    )
+
+
+def _warnings(steps: list[JourneyStep]) -> list[str]:
+    warnings = []
+    if not steps:
+        warnings.append("当前运行还没有可展示的创作步骤。")
+    if any(step.reasoning_source == "contract" for step in steps):
+        warnings.append("部分“思考”由冻结任务合同转述，不是隐藏思维链。")
+    return warnings
+
+
+def _worker_process_label(tools: list[dict[str, Any]]) -> str:
+    names = [str(item.get("name")) for item in tools if item.get("name")]
+    if not names:
+        return "根据任务合同生成候选成果"
+    shown = "、".join(names[:3])
+    suffix = f"等 {len(names)} 次工具调用" if len(names) > 3 else ""
+    return f"执行 {shown}{suffix}"
+
+
+def _task_name(task: dict[str, Any], contract: dict[str, Any] | None) -> str:
+    if contract:
+        return task_label(str(contract.get("task_kind") or "task"))
+    return _plain((task.get("current_spec") or {}).get("objective"))[:60] or "任务"
+
+
+def _action_label(action: str) -> str:
+    return {
+        "ACCEPT": "接受成果",
+        "RETRY": "重新执行",
+        "REVISE": "修订任务",
+        "SPLIT": "拆分子任务",
+        "BLOCK": "暂停在边界",
+        "CANCEL": "取消任务",
+    }.get(action, action)
+
+
+def _action_process(action: str) -> str:
+    return {
+        "ACCEPT": "接受当前成果，继续规划下一步",
+        "RETRY": "保留任务合同，让 Worker 再执行一次",
+        "REVISE": "修订任务合同后重新执行",
+        "SPLIT": "把当前问题拆成多个子任务",
+        "BLOCK": "保存当前成果，等待下一阶段",
+    }.get(action, "更新任务状态")
+
+
+def _loop_label(action: str) -> str:
+    return {"RETRY": "重试", "REVISE": "修订后返工", "SPLIT": "拆分后继续"}.get(action, "返工")
+
+
+def _event_sequences(events: list[dict[str, Any]]) -> dict[str, int]:
+    result: dict[str, int] = {}
+    for event in events:
+        sequence = int(event.get("sequence") or 0)
+        payload = event.get("payload") or {}
+        ids = list(payload.get("task_ids") or [])
+        ids.extend(
+            payload.get(key)
+            for key in ("task_id", "attempt_id", "validation_id", "decision_id")
+            if payload.get(key)
+        )
+        for entity_id in ids:
+            result.setdefault(str(entity_id), sequence)
+    return result
+
+
+def _step_sort_key(step: JourneyStep, event_sequences: dict[str, int]) -> tuple[int, str, int, str]:
+    priority = {"plan": 0, "execute": 1, "validate": 2, "decide": 3}
+    entity_id = step.decision_id or step.validation_id or step.attempt_id or step.task_id
+    event_sequence = event_sequences.get(entity_id, 2**31)
+    return event_sequence, step.created_at, priority[step.step_type], step.step_id
+
+
+def _parse_result(content: Any) -> dict[str, Any]:
+    if isinstance(content, dict):
+        return _parse_json(content.get("result", content))
+    return _parse_json(content)
+
+
+def _parse_json(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return value
+
+
+def _plain(value: Any) -> str:
+    return " ".join(str(value or "").split())

visualization/backend/app/narrator.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .models import ContractNarration
+
+TASK_LABELS = {
+    "direction": "创作方向",
+    "pattern-retrieval": "模式检索",
+    "decode-retrieval": "案例检索",
+    "external-retrieval": "外部资料检索",
+    "knowledge-retrieval": "知识检索",
+    "structure": "整体结构",
+    "paragraph": "内容段落",
+    "element-set": "内容元素",
+    "compare": "候选比较",
+    "compose": "完整脚本组合",
+    "candidate-portfolio": "候选方案集",
+    "root": "最终交付",
+}
+
+
+def task_label(task_kind: str) -> str:
+    return TASK_LABELS.get(task_kind, task_kind.replace("-", " "))
+
+
+def narrate_contract(contract: dict[str, Any]) -> ContractNarration:
+    task_kind = str(contract.get("task_kind") or "task")
+    label = task_label(task_kind)
+    objective = _plain(contract.get("objective")) or f"完成{label}"
+    goals = [str(value) for value in contract.get("goal_ids") or [] if str(value).strip()]
+    inputs = []
+    for item in contract.get("input_decision_refs") or []:
+        if not isinstance(item, dict):
+            continue
+        source_kind = task_label(str(item.get("expected_task_kind") or "上游成果"))
+        inputs.append(f"已接受的{source_kind}")
+    if contract.get("base_artifact_ref"):
+        inputs.append("已冻结的基础成果")
+    if not inputs:
+        inputs.append("当前已冻结的创作输入")
+    criteria = [
+        _plain(item.get("description"))
+        for item in contract.get("criteria") or []
+        if isinstance(item, dict) and _plain(item.get("description"))
+    ]
+    return ContractNarration(
+        action=f"创建{label} Task",
+        reasoning=f"根据任务合同：{objective}",
+        goals=goals,
+        inputs=_unique(inputs),
+        output=label,
+        criteria=_unique(criteria),
+    )
+
+
+def _plain(value: Any) -> str:
+    return " ".join(str(value or "").split())
+
+
+def _unique(values: list[str]) -> list[str]:
+    return list(dict.fromkeys(value for value in values if value))

visualization/backend/tests/test_journey.py

@@ -0,0 +1,214 @@
+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))