script-build-and-framework-tmp/visualization/backend/app/fake_data.py

"""Deterministic, internally consistent fake run for the current pipeline."""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from decimal import Decimal
from typing import Any

from .contracts import ENUM_CATALOG, SCHEMA_CATALOG
from .models import (
    FlowEdge,
    FlowNode,
    Metric,
    NodeDetail,
    PhaseFrame,
    RecordEnvelope,
    RunGraph,
    RunSummary,
)

RUN_ID = 10001
ROOT_TRACE_ID = "trace_fake_phase3_10001"
STAMP = "2026-07-20T10:24:36.482000+08:00"
START = "2026-07-20T10:18:04.103000+08:00"


def digest(label: str) -> str:
    return hashlib.sha256(label.encode()).hexdigest()


def wire_digest(label: str) -> str:
    return f"sha256:{digest(label)}"


def normalize_json(value: Any) -> Any:
    if value is None or isinstance(value, (str, bool, int)):
        return value
    if isinstance(value, float):
        rendered = format(Decimal(str(value)), "f")
        if "." in rendered:
            rendered = rendered.rstrip("0").rstrip(".")
        return "0" if rendered in {"-0", ""} else rendered
    if isinstance(value, dict):
        return {key: normalize_json(value[key]) for key in sorted(value)}
    if isinstance(value, (list, tuple)):
        return [normalize_json(item) for item in value]
    raise TypeError(f"unsupported fake canonical JSON value: {type(value).__name__}")


def canonical_digest(payload: Any) -> str:
    encoded = json.dumps(
        normalize_json(payload),
        ensure_ascii=False,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def canonical_wire_digest(payload: Any) -> str:
    return f"sha256:{canonical_digest(payload)}"


def input_canonical_payload() -> dict[str, Any]:
    return {
        "schema_version": "script-build-input/v1",
        "script_build_id": RUN_ID,
        "execution_id": 88001,
        "topic_build_id": 67001,
        "topic_id": 7701,
        "topic": {"id": 7701, "title": "为什么 AI Agent 项目总在最后一步失控"},
        "account": {"id": 230, "name": "产品深水区"},
        "persona_points": [
            {
                "point_id": 7,
                "point_type": "voice",
                "point_result": "克制、具体、敢于下判断",
            }
        ],
        "section_patterns": [{"pattern_id": 14, "name": "事故切入—框架拆解—行动清单"}],
        "strategies": [{"strategy_id": 3, "version": 5, "name": "反常识开场"}],
        "prompt_manifest": [
            {
                "role": "planner",
                "name": "script_planner",
                "sha256": digest("planner-prompt"),
            }
        ],
        "datasource_manifest": {"adapter": "deterministic-fake", "version": "1"},
        "model_manifest": {
            "planner": "deterministic-fake-model",
            "worker": "deterministic-fake-model",
        },
    }


ARTIFACT_DIGESTS: dict[int, str] = {}
ARTIFACT_KINDS: dict[int, str] = {}


def artifact_uri(version: int) -> str:
    return f"script-build://artifact-versions/{version}"


def artifact_ref(version: int, kind: str | None = None) -> dict[str, Any]:
    return {
        "uri": artifact_uri(version),
        "kind": kind or ARTIFACT_KINDS[version],
        "version": str(version),
        "digest": ARTIFACT_DIGESTS.get(version, digest(f"artifact-{version}")),
        "summary": None,
        "metadata": {},
    }


def criterion(identifier: str, description: str) -> dict[str, Any]:
    return {"criterion_id": identifier, "description": description, "hard": True}


def decision_ref(decision_id: str, version: int, task_kind: str) -> dict[str, Any]:
    return {
        "decision_id": decision_id,
        "artifact_ref": artifact_ref(version),
        "scope_ref": f"script-build://decisions/{decision_id}",
        "expected_task_kind": task_kind,
    }


def task_spec(objective: str, kind: str) -> dict[str, Any]:
    return {
        "version": 1,
        "objective": objective,
        "acceptance_criteria": [criterion("content_complete", "产物完整且引用闭合")],
        "context_refs": [f"script-build://task-kinds/{kind}"],
        "created_at": START,
    }


def task_payload(
    task_id: str, objective: str, kind: str, parent: str | None = "task-root"
) -> dict[str, Any]:
    is_root = task_id == "task-root"
    return {
        "task_id": task_id,
        "goal_id": "goal-script-delivery",
        "parent_task_id": parent,
        "display_path": "Root" if is_root else f"Root / {objective[:24]}",
        "specs": [task_spec(objective, kind)],
        "current_spec_version": 1,
        "status": "completed",
        "child_task_ids": ["direction", "portfolio"] if is_root else [],
        "attempt_ids": [f"attempt-{task_id}"],
        "validation_ids": [f"validation-{task_id}"],
        "decision_ids": [f"decision-{task_id}"],
        "repair_count_by_version": {"1": 0},
        "blocked_reason": None,
        "superseded_by": None,
        "created_at": START,
        "updated_at": STAMP,
    }


def contract_payload(
    task_id: str, objective: str, kind: str, output_schema: str
) -> dict[str, Any]:
    intent = {
        "compare": "compare",
        "compose": "compose",
        "candidate-portfolio": "portfolio",
        "root-delivery": "deliver",
        "paragraph": "expand",
        "element-set": "expand",
        "structure": "expand",
    }.get(kind, "explore")
    payload: dict[str, Any] = {
        "schema_version": "script-task-contract/v1",
        "task_kind": kind,
        "scope_ref": f"script-build://tasks/{task_id}",
        "intent_class": intent,
        "objective": objective,
        "input_decision_refs": [],
        "base_artifact_ref": None,
        "write_scope": [f"script-build://tasks/{task_id}/workspace"],
        "gap_ref": None,
        "output_schema": output_schema,
        "criteria": [criterion("content_complete", "产物完整且引用闭合")],
        "budget": {
            "max_attempts": 4,
            "max_tokens": 32000,
            "max_seconds": 900,
            "max_external_queries": 40,
            "max_no_improvement": 3,
        },
        "supersedes_decision_ids": [],
        "candidate_closure_decision_refs": [],
        "adopted_decision_ids": [],
        "held_or_rejected_decision_ids": [],
        "compose_order": [],
        "comparison_decision_refs": [],
    }
    if kind == "compare":
        payload["comparison_decision_refs"] = [
            decision_ref("decision-paragraph:a", 121, "paragraph"),
            decision_ref("decision-elements:a", 122, "element-set"),
        ]
    elif kind == "compose":
        closure = [
            decision_ref("decision-paragraph:a", 121, "paragraph"),
            decision_ref("decision-elements:a", 122, "element-set"),
        ]
        payload["candidate_closure_decision_refs"] = closure
        payload["adopted_decision_ids"] = [item["decision_id"] for item in closure]
        payload["compose_order"] = [item["decision_id"] for item in closure]
        payload["comparison_decision_refs"] = [
            decision_ref("decision-compare:ab", 130, "compare")
        ]
    elif kind == "candidate-portfolio":
        closure = [
            decision_ref("decision-compose:final", 140, "compose"),
            decision_ref("decision-compose:alternate", 141, "compose"),
        ]
        payload["candidate_closure_decision_refs"] = closure
        payload["adopted_decision_ids"] = ["decision-compose:final"]
        payload["held_or_rejected_decision_ids"] = ["decision-compose:alternate"]
        payload["compose_order"] = ["decision-compose:final"]
    elif kind == "root-delivery":
        payload["input_decision_refs"] = [
            decision_ref("decision-direction", 110, "direction"),
            decision_ref("decision-portfolio", 150, "candidate-portfolio"),
        ]
        payload["scope_ref"] = "script-build://scope/root"
        payload["write_scope"] = []
    return payload


def operation_payload(task_id: str, index: int) -> dict[str, Any]:
    return {
        "operation_id": f"operation-{index:02d}",
        "root_trace_id": ROOT_TRACE_ID,
        "kind": "dispatch",
        "status": "completed",
        "task_ids": [task_id],
        "attempt_ids": [f"attempt-{task_id}"],
        "validation_ids": [f"validation-{task_id}"],
        "result_ref": {"task_id": task_id, "status": "completed"},
        "execution_epoch": 1,
        "deadline_at": "2026-07-20T10:34:00+08:00",
        "error": None,
        "started_at": START,
        "completed_at": STAMP,
        "created_at": START,
        "updated_at": STAMP,
    }


def attempt_payload(
    task_id: str, preset: str, artifact_version: int, index: int
) -> dict[str, Any]:
    return {
        "attempt_id": f"attempt-{task_id}",
        "task_id": task_id,
        "spec_version": 1,
        "worker_trace_id": f"worker-trace-{index:02d}",
        "worker_preset": preset,
        "execution_mode": "agent",
        "accepted_child_decision_ids": (
            ["decision-direction", "decision-portfolio"]
            if task_id == "root-delivery"
            else []
        ),
        "status": "submitted",
        "operation_id": f"operation-{index:02d}",
        "execution_epoch": 1,
        "continue_from_trace_id": None,
        "snapshot_id": f"artifact-snapshot-{index:02d}",
        "submission": {
            "summary": "已保存唯一冻结产物并提交验证。",
            "artifact_refs": [artifact_ref(artifact_version)],
            "evidence_refs": [],
        },
        "execution_stats": {
            "primary_model": "deterministic-fake-model",
            "total_tokens": 1840 + index * 31,
            "total_cost": 0.012 + index * 0.001,
            "failure_code": None,
        },
        "error": None,
        "started_at": START,
        "completed_at": STAMP,
        "duration_ms": 2180 + index * 73,
        "created_at": START,
        "updated_at": STAMP,
    }


def validation_payload(task_id: str, preset: str, index: int) -> dict[str, Any]:
    return {
        "validation_id": f"validation-{task_id}",
        "task_id": task_id,
        "attempt_id": f"attempt-{task_id}",
        "spec_version": 1,
        "snapshot_id": f"artifact-snapshot-{index:02d}",
        "validator_trace_id": f"validator-trace-{index:02d}",
        "validator_preset": preset,
        "validation_plan": {
            "mode": "agent",
            "validator_preset": preset,
            "rule_ids": [],
            "max_evidence_queries": 10,
            "max_evidence_items_per_query": 50,
            "evidence_timeout_seconds": 10.0,
        },
        "evidence_queries_used": 0,
        "status": "completed",
        "operation_id": f"operation-{index:02d}",
        "execution_epoch": 1,
        "verdict": "passed",
        "criterion_results": [
            {
                "criterion_id": "content_complete",
                "verdict": "passed",
                "reason": "字段、引用、digest 与冻结快照一致。",
                "evidence_refs": [],
            }
        ],
        "summary": "所有硬性标准均通过。",
        "evidence_refs": [],
        "unverified_claims": [],
        "risks": [],
        "recommendation": "accept",
        "execution_stats": {
            "primary_model": "deterministic-fake-model",
            "total_tokens": 620 + index * 11,
            "total_cost": 0.004,
            "failure_code": None,
        },
        "error": None,
        "started_at": START,
        "completed_at": STAMP,
        "duration_ms": 840 + index * 29,
        "created_at": START,
        "updated_at": STAMP,
    }


def decision_payload(
    task_id: str, index: int, action: str = "accept"
) -> dict[str, Any]:
    return {
        "decision_id": f"decision-{task_id}",
        "task_id": task_id,
        "action": action,
        "reason": "Validator 已通过，冻结产物满足当前合同。",
        "from_status": "awaiting_decision",
        "to_status": "completed" if action == "accept" else "running",
        "attempt_id": f"attempt-{task_id}",
        "validation_id": f"validation-{task_id}",
        "payload": {"artifact_ref": artifact_uri(100 + index), "fenced_epoch": 1},
        "created_at": STAMP,
    }


def lineage(scope: str) -> dict[str, Any]:
    return {
        "scope_ref": scope,
        "input_snapshot_ref": "script-build://input-snapshots/snapshot-10001-v1",
        "input_closure_digest": wire_digest(f"closure-{scope}"),
        "base_artifact_ref": None,
        "base_artifact_digest": None,
        "base_revision": None,
        "write_scope": [f"{scope}/workspace"],
        "supersedes_decision_ids": [],
        "source_lineage": [
            {"source": "accepted-decision", "decision_id": "decision-direction"}
        ],
    }


PARAGRAPHS = [
    {
        "paragraph_id": 1,
        "paragraph_index": 1,
        "level": 1,
        "parent_id": None,
        "name": "反常识开场",
        "content_range": {"min_chars": 80, "max_chars": 130},
        "theme": "AI 不是答案机",
        "form": "场景反转",
        "function": "建立冲突",
        "feeling": "惊讶",
        "theme_elements": [{"name": "认知落差"}],
        "form_elements": [{"name": "短问句"}],
        "function_elements": [{"name": "钩子"}],
        "feeling_elements": [{"name": "紧迫"}],
        "description": "用失败场景破题",
        "full_description": "从团队盲目堆 Agent 的失败现场切入，指出流程边界比模型数量更重要。",
        "is_active": True,
    },
    {
        "paragraph_id": 2,
        "paragraph_index": 2,
        "level": 2,
        "parent_id": 1,
        "name": "三个边界",
        "content_range": {"min_chars": 180, "max_chars": 260},
        "theme": "合同、验证、发布",
        "form": "递进清单",
        "function": "交付方法",
        "feeling": "笃定",
        "theme_elements": [{"name": "可验证"}],
        "form_elements": [{"name": "三段式"}],
        "function_elements": [{"name": "解释"}],
        "feeling_elements": [{"name": "掌控"}],
        "description": "拆解三层工程边界",
        "full_description": "说明冻结合同、独立验证与原子发布如何形成可恢复的内容生产线。",
        "is_active": True,
    },
]

ELEMENTS = [
    {
        "element_id": 1,
        "name": "失败现场",
        "dimension_primary": "实质",
        "dimension_secondary": "案例",
        "commonality_analysis": {"pattern": "先展示代价"},
        "topic_support": {"score": 0.94, "reason": "直接支撑流程治理主题"},
        "weight_score": {"value": 9.2, "scale": 10},
        "support_elements": [{"type": "quote", "ref": "evidence-pattern-1"}],
        "is_active": True,
    },
    {
        "element_id": 2,
        "name": "三层护栏",
        "dimension_primary": "形式",
        "dimension_secondary": "结构",
        "commonality_analysis": {"pattern": "三段递进"},
        "topic_support": {"score": 0.91, "reason": "形成清晰记忆点"},
        "weight_score": {"value": 8.8, "scale": 10},
        "support_elements": [{"type": "strategy", "ref": "strategy-3"}],
        "is_active": True,
    },
]

LINKS = [{"paragraph_id": 1, "element_id": 1}, {"paragraph_id": 2, "element_id": 2}]


def evidence_payload(kind: str, index: int) -> dict[str, Any]:
    return {
        "schema_version": "evidence-record/v1",
        "evidence_id": f"evidence-{kind}-{index}",
        "source_type": kind,
        "tool_name": f"retrieve_{kind.replace('-', '_')}",
        "query": {"topic_id": 7701, "limit": 12, "intent": "AI Agent 内容生产流程"},
        "source_refs": [f"{kind}://fake/source/{index}"],
        "raw_artifact_ref": f"script-build://raw-artifacts/sha256/{digest(kind)}",
        "summary": f"{kind} 返回了可用于方向与结构决策的高置信证据。",
        "supports": ["goal-retention", "criterion-evidence"],
        "confidence": "high",
        "limitations": ["fake 数据不代表真实检索覆盖率"],
        "created_at": STAMP,
    }


def direction_payload() -> dict[str, Any]:
    return {
        "schema_version": "script-direction/v1",
        "goals": [
            {
                "goal_id": "goal-retention",
                "statement": "用工程事故切入，讲清智能创作系统的三层边界",
                "rationale": "目标用户需要可落地的流程而非泛泛 AI 概念。",
            }
        ],
        "criteria": [
            {"criterion_id": "criterion-evidence", "description": "关键主张均可追溯"}
        ],
        "domain_criteria": [
            {
                "criterion_id": "criterion-fencing",
                "description": "正确描述 fencing 与原子发布",
            }
        ],
        "topic_refs": ["script-build://inputs/topics/7701"],
        "persona_refs": ["script-build://inputs/personas/230"],
        "strategy_refs": ["script-build://inputs/strategies/3"],
        "evidence_refs": [artifact_uri(101), artifact_uri(102)],
        "legacy_markdown": "# 创作方向\n用一个失控的多 Agent 项目开场，落到合同、验证、发布三层护栏。",
    }


def structured_payload() -> dict[str, Any]:
    return {
        "schema_version": "structured-script/v1",
        "direction_ref": artifact_uri(110),
        "input_closure_digest": wire_digest("structured-input-closure"),
        "paragraphs": PARAGRAPHS,
        "elements": ELEMENTS,
        "paragraph_element_links": LINKS,
        "source_artifact_refs": [
            artifact_uri(120),
            artifact_uri(121),
            artifact_uri(122),
        ],
        "evidence_refs": [
            artifact_uri(101),
            artifact_uri(102),
            artifact_uri(103),
            artifact_uri(104),
        ],
        "acceptance_notes": ["采用候选 A 的冲突开场", "保留候选 B 的三层递进结构"],
    }


def legacy_canonical_payload() -> dict[str, Any]:
    paragraph_keys = {
        item["paragraph_id"]: item["paragraph_index"] for item in PARAGRAPHS
    }
    paragraphs = []
    for item in PARAGRAPHS:
        paragraphs.append(
            {
                "logical_key": item["paragraph_index"],
                "paragraph_index": item["paragraph_index"],
                "level": item["level"],
                "parent_logical_key": (
                    paragraph_keys[item["parent_id"]]
                    if item["parent_id"] is not None
                    else None
                ),
                "name": item["name"],
                "content_range": item["content_range"],
                "theme": item["theme"],
                "form": item["form"],
                "function": item["function"],
                "feeling": item["feeling"],
                "theme_elements": item["theme_elements"],
                "form_elements": item["form_elements"],
                "function_elements": item["function_elements"],
                "feeling_elements": item["feeling_elements"],
                "description": item["description"],
                "full_description": item["full_description"],
            }
        )
    element_keys: dict[int, str] = {}
    elements = []
    for item in ELEMENTS:
        key = canonical_digest(
            [item["dimension_primary"], item["dimension_secondary"], item["name"]]
        )
        element_keys[item["element_id"]] = key
        elements.append(
            {
                "logical_key": key,
                "name": item["name"],
                "dimension_primary": item["dimension_primary"],
                "dimension_secondary": item["dimension_secondary"],
                "commonality_analysis": item["commonality_analysis"],
                "topic_support": item["topic_support"],
                "weight_score": item["weight_score"],
                "support_elements": item["support_elements"],
            }
        )
    links = [
        {
            "paragraph_logical_key": paragraph_keys[item["paragraph_id"]],
            "element_logical_key": element_keys[item["element_id"]],
        }
        for item in LINKS
    ]
    paragraphs.sort(key=lambda item: int(item["logical_key"]))
    elements.sort(key=lambda item: str(item["logical_key"]))
    links.sort(
        key=lambda item: (
            int(item["paragraph_logical_key"]),
            str(item["element_logical_key"]),
        )
    )
    return {
        "schema_version": "legacy-projection-canonical/v1",
        "direction": "用失控案例讲清合同、验证与原子发布。",
        "summary": "以事故切入，完整解释智能创作系统的合同、验证和原子发布。",
        "paragraph_count": len(paragraphs),
        "element_count": len(elements),
        "link_count": len(links),
        "paragraphs": [normalize_json(item) for item in paragraphs],
        "elements": [normalize_json(item) for item in elements],
        "paragraph_element_links": [normalize_json(item) for item in links],
    }


def root_input_closure_digest() -> str:
    return canonical_wire_digest(
        {
            "schema_version": "root-delivery-input-closure/v1",
            "direction": {
                "ref": artifact_uri(110),
                "digest": ARTIFACT_DIGESTS[110],
            },
            "candidate_portfolio": {
                "ref": artifact_uri(150),
                "digest": ARTIFACT_DIGESTS[150],
            },
            "structured_script": {
                "ref": artifact_uri(140),
                "digest": ARTIFACT_DIGESTS[140],
            },
        }
    )


def record(model_name: str, payload: dict[str, Any]) -> RecordEnvelope:
    expected = set(SCHEMA_CATALOG[model_name])
    actual = set(payload)
    if actual != expected:
        missing = sorted(expected - actual)
        extra = sorted(actual - expected)
        raise AssertionError(
            f"{model_name} fake field mismatch: missing={missing}, extra={extra}"
        )
    schema_version = payload.get("schema_version")
    return RecordEnvelope(
        model_name=model_name,
        schema_version=str(schema_version) if schema_version is not None else None,
        payload=payload,
    )


@dataclass
class Builder:
    nodes: list[FlowNode]
    edges: list[FlowEdge]
    sequence: int = 0

    def add(
        self,
        node_id: str,
        node_type: str,
        phase: str,
        lane: str,
        title: str,
        subtitle: str,
        status: str,
        x: int,
        y: int,
        model_name: str,
        payload: dict[str, Any],
        *,
        agent_role: str | None = None,
        task_kind: str | None = None,
        tags: list[str] | None = None,
        metrics: list[tuple[str, str]] | None = None,
    ) -> str:
        self.sequence += 1
        self.nodes.append(
            FlowNode(
                id=node_id,
                node_type=node_type,  # type: ignore[arg-type]
                phase=phase,  # type: ignore[arg-type]
                lane=lane,
                title=title,
                subtitle=subtitle,
                status=status,
                agent_role=agent_role,
                task_kind=task_kind,
                sequence=self.sequence,
                x=x,
                y=y,
                tags=tags or [],
                metrics=[
                    Metric(label=label, value=value) for label, value in (metrics or [])
                ],
                record=record(model_name, payload),
            )
        )
        return node_id

    def link(
        self,
        source: str,
        target: str,
        label: str | None = None,
        kind: str = "control",
        animated: bool = False,
    ) -> None:
        self.edges.append(
            FlowEdge(
                id=f"edge:{source}:{target}",
                source=source,
                target=target,
                label=label,
                kind=kind,
                animated=animated,  # type: ignore[arg-type]
            )
        )


def _pipeline_unit(
    b: Builder,
    *,
    prefix: str,
    phase: str,
    lane: str,
    kind: str,
    output_schema: str,
    objective: str,
    artifact_model: str,
    artifact_payload: dict[str, Any],
    x: int,
    y: int,
    index: int,
    artifact_version: int,
    preset: str,
    validator_preset: str = "script_candidate_validator",
) -> tuple[str, str]:
    artifact_kind = {
        "EvidenceRecordV1": "evidence",
        "ScriptDirectionArtifactV1": "direction",
        "StructureArtifactV1": "structure",
        "ParagraphArtifactV1": "paragraph",
        "ElementSetArtifactV1": "element_set",
        "ComparisonArtifactV1": "comparison",
        "StructuredScriptArtifactV1": "structured_script",
        "CandidatePortfolioArtifactV1": "candidate_portfolio",
        "RootDeliveryManifestV1": "root_delivery_manifest",
    }[artifact_model]
    ARTIFACT_KINDS[artifact_version] = artifact_kind
    ARTIFACT_DIGESTS[artifact_version] = canonical_wire_digest(artifact_payload)
    contract_id = b.add(
        f"{prefix}:contract",
        "contract",
        phase,
        lane,
        "冻结 TaskContract",
        objective,
        "frozen",
        x,
        y,
        "ScriptTaskContractV1",
        contract_payload(prefix, objective, kind, output_schema),
        task_kind=kind,
        tags=["immutable", output_schema],
        metrics=[("spec", "v1"), ("预算", "32k tokens")],
    )
    op_id = b.add(
        f"{prefix}:operation",
        "operation",
        phase,
        lane,
        "Durable Operation",
        "dispatch · execution_epoch=1",
        "completed",
        x + 300,
        y,
        "OperationView",
        operation_payload(prefix, index),
        task_kind=kind,
        tags=["durable", "fenced"],
        metrics=[("耗时", f"{2 + index % 4}.2s")],
    )
    attempt_id = b.add(
        f"{prefix}:attempt",
        "attempt",
        phase,
        lane,
        f"{preset} Attempt",
        "受控工具白名单内执行",
        "submitted",
        x + 600,
        y,
        "AttemptView",
        attempt_payload(prefix, preset, artifact_version, index),
        agent_role="worker",
        task_kind=kind,
        tags=["worker", "snapshot"],
        metrics=[("tokens", str(1840 + index * 31)), ("attempt", "1 / 4")],
    )
    artifact_id = b.add(
        f"{prefix}:artifact",
        "artifact",
        phase,
        lane,
        artifact_model.removesuffix("V1"),
        f"{artifact_uri(artifact_version)} · canonical",
        "frozen",
        x + 900,
        y,
        artifact_model,
        artifact_payload,
        task_kind=kind,
        tags=["frozen", "sha256"],
        metrics=[("version", str(artifact_version)), ("state", "frozen")],
    )
    validation_id = b.add(
        f"{prefix}:validation",
        "validation",
        phase,
        lane,
        f"{validator_preset} Validation",
        "独立快照验证 · passed",
        "passed",
        x + 1200,
        y,
        "ValidationView",
        validation_payload(prefix, validator_preset, index),
        agent_role="validator",
        task_kind=kind,
        tags=["independent", "passed"],
        metrics=[("verdict", "passed"), ("缺陷", "0")],
    )
    decision_id = b.add(
        f"{prefix}:decision",
        "decision",
        phase,
        lane,
        "Planner ACCEPT",
        "显式接受，不由 Validator 自动发布",
        "accepted",
        x + 1500,
        y,
        "PlannerDecisionView",
        decision_payload(prefix, index),
        agent_role="planner",
        task_kind=kind,
        tags=["authority", "accept"],
        metrics=[("action", "ACCEPT")],
    )
    b.link(contract_id, op_id, "dispatch")
    b.link(op_id, attempt_id, "worker", animated=True)
    b.link(attempt_id, artifact_id, "freeze", "artifact")
    b.link(artifact_id, validation_id, "snapshot", "validation")
    b.link(validation_id, decision_id, "passed", "validation")
    return contract_id, decision_id


def build_graph() -> RunGraph:
    b = Builder([], [])

    build = b.add(
        "run:record",
        "input",
        "phase-1",
        "mission",
        "Script Build #10001",
        "当前 Build 业务记录",
        "success",
        80,
        150,
        "ScriptBuildRecord",
        {
            "id": RUN_ID,
            "execution_id": 88001,
            "topic_build_id": 67001,
            "topic_id": 7701,
            "agent_type": "AigcAgent",
            "agent_config": {"model": "deterministic-fake-model", "temperature": 0},
            "data_source_url": None,
            "status": "success",
            "reson_trace_id": ROOT_TRACE_ID,
            "is_deleted": False,
            "is_favorited": True,
            "summary": "完成一篇讲解智能创作系统工程边界的结构化脚本。",
            "error_message": None,
            "script_direction": "用失控案例讲清合同、验证与原子发布。",
            "build_workflow": {
                "checkpoint": "PHASE_THREE_PUBLISHED",
                "schema_version": 3,
            },
            "paragraph_count": 2,
            "element_count": 2,
            "input_tokens": 28420,
            "output_tokens": 9360,
            "cost_usd": 0.1842,
            "strategies_config": {"always_on": [3], "on_demand": [8]},
            "start_time": START,
            "end_time": STAMP,
        },
        tags=["fake", "success"],
        metrics=[("段落", "2"), ("元素", "2"), ("成本", "$0.1842")],
    )
    input_payload = input_canonical_payload()
    input_sha256 = canonical_digest(input_payload)
    snapshot = b.add(
        "mission:input-snapshot",
        "input",
        "phase-1",
        "mission",
        "冻结 InputSnapshot",
        "选题 · 账号 · 人设 · 策略 · Prompt 清单",
        "frozen",
        380,
        150,
        "ScriptBuildInputSnapshotV1",
        {
            "snapshot_id": "301",
            "script_build_id": input_payload["script_build_id"],
            "execution_id": input_payload["execution_id"],
            "topic_build_id": input_payload["topic_build_id"],
            "topic_id": input_payload["topic_id"],
            "topic": input_payload["topic"],
            "account": input_payload["account"],
            "persona_points": input_payload["persona_points"],
            "section_patterns": input_payload["section_patterns"],
            "strategies": input_payload["strategies"],
            "prompt_manifest": input_payload["prompt_manifest"],
            "datasource_manifest": input_payload["datasource_manifest"],
            "model_manifest": input_payload["model_manifest"],
            "canonical_sha256": f"sha256:{input_sha256}",
            "created_at": START,
            "parent_snapshot_id": None,
            "parent_snapshot_sha256": None,
            "business_input_sha256": None,
        },
        tags=["immutable", "v1"],
        metrics=[("persona", "1"), ("strategy", "1"), ("prompt", "1")],
    )
    snapshot_row = b.add(
        "storage:input-snapshot-row",
        "artifact",
        "phase-1",
        "storage",
        "InputSnapshot 持久化行",
        "script_build_input_snapshot · version=1",
        "frozen",
        80,
        500,
        "InputSnapshotRow",
        {
            "id": 301,
            "script_build_id": RUN_ID,
            "version": 1,
            "canonical_json": input_payload,
            "canonical_sha256": input_sha256,
            "prompt_manifest_json": input_payload["prompt_manifest"],
            "schema_version": "script-build-input/v1",
            "created_at": START,
        },
        tags=["physical-row", "immutable"],
        metrics=[("table", "input_snapshot"), ("version", "1")],
    )
    binding = b.add(
        "mission:binding",
        "checkpoint",
        "phase-1",
        "mission",
        "Mission Binding",
        "Build ↔ Root Trace ↔ Snapshot 唯一绑定",
        "bound",
        680,
        150,
        "MissionBinding",
        {
            "binding_id": 501,
            "script_build_id": RUN_ID,
            "root_trace_id": ROOT_TRACE_ID,
            "input_snapshot_id": 301,
            "active_direction_artifact_version_id": 110,
            "accepted_root_artifact_version_id": 160,
            "engine_version": "host-phase3/1",
            "schema_version": "3",
            "created_at": START,
            "updated_at": STAMP,
            "owner_epoch": 1,
            "stop_epoch": 0,
            "owner_instance_id": "host-fake-01",
            "owner_acquired_at": START,
        },
        tags=["unique", "binding"],
        metrics=[("owner epoch", "1"), ("stop epoch", "0")],
    )
    owner = b.add(
        "mission:owner",
        "checkpoint",
        "phase-1",
        "mission",
        "Owner Lease + Fencing",
        "fd lock + owner_epoch CAS",
        "active",
        980,
        150,
        "MissionOwnerToken",
        {
            "script_build_id": RUN_ID,
            "root_trace_id": ROOT_TRACE_ID,
            "owner_epoch": 1,
            "stop_epoch": 0,
            "owner_instance_id": "host-fake-01",
        },
        tags=["lease", "fenced"],
        metrics=[("epoch", "1"), ("stop", "0")],
    )
    b.link(build, snapshot, "freeze")
    b.link(snapshot, snapshot_row, "persist", "artifact")
    b.link(snapshot, binding, "bind")
    b.link(binding, owner, "acquire")

    direction_contract, direction_decision = _pipeline_unit(
        b,
        prefix="direction",
        phase="phase-1",
        lane="direction",
        kind="direction",
        output_schema="script-direction/v1",
        objective="确定脚本目标、标准与证据闭包",
        artifact_model="ScriptDirectionArtifactV1",
        artifact_payload=direction_payload(),
        x=380,
        y=500,
        index=1,
        artifact_version=110,
        preset="script_direction_worker",
    )
    b.link(owner, direction_contract, "plan")
    direction_publication = b.add(
        "direction:publication",
        "publication",
        "phase-1",
        "direction",
        "Direction Publication",
        "独立 direction publication identity",
        "published",
        2180,
        500,
        "Publication",
        {
            "publication_id": 701,
            "script_build_id": RUN_ID,
            "publication_type": "direction",
            "accept_decision_id": "decision-direction",
            "artifact_version_id": 110,
            "expected_sha256": ARTIFACT_DIGESTS[110],
            "state": "published",
            "attempt_count": 1,
            "publication_revision": 1,
            "last_error_code": None,
            "last_error_summary": None,
            "created_at": START,
            "updated_at": STAMP,
            "published_at": STAMP,
        },
        tags=["published", "idempotent"],
        metrics=[("revision", "1")],
    )
    b.link(direction_decision, direction_publication, "publish", "publication")

    retrieval_specs = [
        ("pattern", "pattern-retrieval", "Pattern 结构模式", 180, 2, 101),
        ("decode", "decode-retrieval", "Case 解构证据", 500, 3, 102),
        ("knowledge", "knowledge-retrieval", "知识库事实", 820, 4, 103),
        ("external", "external-retrieval", "外部检索证据", 1140, 5, 104),
    ]
    retrieval_decisions: list[str] = []
    for name, kind, label, y, index, version in retrieval_specs:
        contract_id, decision_id = _pipeline_unit(
            b,
            prefix=f"retrieval:{name}",
            phase="phase-2",
            lane=f"retrieval-{name}",
            kind=kind,
            output_schema="evidence-record/v1",
            objective=f"检索并冻结{label}",
            artifact_model="EvidenceRecordV1",
            artifact_payload=evidence_payload(kind, index),
            x=2540,
            y=y,
            index=index,
            artifact_version=version,
            preset="script_retrieval_worker",
            validator_preset="script_retrieval_validator",
        )
        b.link(direction_decision, contract_id, "accepted direction")
        retrieval_decisions.append(decision_id)

    structure_artifact = {
        "schema_version": "structure-artifact/v1",
        **lineage("script-build://scopes/structure-a"),
        "paragraphs": PARAGRAPHS,
    }
    structure_contract, structure_decision = _pipeline_unit(
        b,
        prefix="structure:a",
        phase="phase-2",
        lane="creative-a",
        kind="structure",
        output_schema="structure-artifact/v1",
        objective="生成冲突递进式脚本结构候选 A",
        artifact_model="StructureArtifactV1",
        artifact_payload=structure_artifact,
        x=4540,
        y=1500,
        index=6,
        artifact_version=120,
        preset="script_structure_worker",
    )
    for item in retrieval_decisions:
        b.link(item, structure_contract, "evidence", "artifact")

    paragraph_artifact = {
        "schema_version": "paragraph-artifact/v1",
        **lineage("script-build://scopes/paragraph-a"),
        "paragraphs": PARAGRAPHS,
        "elements": [],
        "paragraph_element_links": [],
        "change_manifest": {"created": [1, 2], "updated": [], "deleted": []},
    }
    paragraph_contract, paragraph_decision = _pipeline_unit(
        b,
        prefix="paragraph:a",
        phase="phase-2",
        lane="creative-a",
        kind="paragraph",
        output_schema="paragraph-artifact/v1",
        objective="细化候选 A 的段落内容与父子关系",
        artifact_model="ParagraphArtifactV1",
        artifact_payload=paragraph_artifact,
        x=4840,
        y=1810,
        index=7,
        artifact_version=121,
        preset="script_paragraph_worker",
    )
    b.link(structure_decision, paragraph_contract, "base structure", "artifact")

    element_artifact = {
        "schema_version": "element-set-artifact/v1",
        **lineage("script-build://scopes/elements-a"),
        "paragraphs": PARAGRAPHS,
        "elements": ELEMENTS,
        "paragraph_element_links": LINKS,
        "change_manifest": {"created": [1, 2], "linked": 2},
    }
    element_contract, element_decision = _pipeline_unit(
        b,
        prefix="elements:a",
        phase="phase-2",
        lane="creative-a",
        kind="element-set",
        output_schema="element-set-artifact/v1",
        objective="生成元素集合并闭合 Paragraph–Element Link",
        artifact_model="ElementSetArtifactV1",
        artifact_payload=element_artifact,
        x=5140,
        y=1500,
        index=8,
        artifact_version=122,
        preset="script_element_set_worker",
    )
    b.link(paragraph_decision, element_contract, "paragraph closure", "artifact")

    comparison_artifact = {
        "schema_version": "comparison-artifact/v1",
        **lineage("script-build://scopes/comparison"),
        "candidate_artifact_refs": [artifact_uri(121), artifact_uri(122)],
        "criterion_results": [
            {
                "criterion_id": "content_complete",
                "candidate_results": [
                    {
                        "artifact_ref": artifact_uri(121),
                        "verdict": "passed",
                        "reason": "段落闭合",
                    },
                    {
                        "artifact_ref": artifact_uri(122),
                        "verdict": "passed",
                        "reason": "元素关联闭合",
                    },
                ],
            }
        ],
        "conflicts": ["候选 A 更强冲突，候选 B 信息密度更高"],
        "recommendation": artifact_uri(122),
    }
    compare_contract, compare_decision = _pipeline_unit(
        b,
        prefix="compare:ab",
        phase="phase-2",
        lane="convergence",
        kind="compare",
        output_schema="comparison-artifact/v1",
        objective="按同一标准比较候选并给出建议",
        artifact_model="ComparisonArtifactV1",
        artifact_payload=comparison_artifact,
        x=5440,
        y=1810,
        index=9,
        artifact_version=130,
        preset="script_comparison_worker",
    )
    b.link(paragraph_decision, compare_contract, "candidate A", "artifact")
    b.link(element_decision, compare_contract, "candidate B", "artifact")

    compose_contract, compose_decision = _pipeline_unit(
        b,
        prefix="compose:final",
        phase="phase-2",
        lane="convergence",
        kind="compose",
        output_schema="structured-script/v1",
        objective="按 accepted closure 合成 StructuredScript",
        artifact_model="StructuredScriptArtifactV1",
        artifact_payload=structured_payload(),
        x=5740,
        y=1500,
        index=10,
        artifact_version=140,
        preset="script_compose_worker",
    )
    b.link(compare_decision, compose_contract, "comparison accepted")

    alternate_script = structured_payload()
    alternate_script["acceptance_notes"] = [
        "保留候选 B 的信息密度",
        "作为 Portfolio 中未采用但完整闭合的备选脚本",
    ]
    alternate_contract, alternate_decision = _pipeline_unit(
        b,
        prefix="compose:alternate",
        phase="phase-2",
        lane="convergence-alternate",
        kind="compose",
        output_schema="structured-script/v1",
        objective="生成完整备选 StructuredScript 供 Portfolio 显式处置",
        artifact_model="StructuredScriptArtifactV1",
        artifact_payload=alternate_script,
        x=5740,
        y=2110,
        index=13,
        artifact_version=141,
        preset="script_compose_worker",
    )
    b.link(compare_decision, alternate_contract, "alternate candidate")

    portfolio_payload = {
        "schema_version": "candidate-portfolio/v1",
        "adopted_structured_script_ref": artifact_uri(140),
        "candidate_structured_script_refs": [artifact_uri(140), artifact_uri(141)],
        "accepted_decision_ids": ["decision-compose:final"],
        "superseded_decision_ids": [],
        "rejected_or_held_decision_ids": ["decision-compose:alternate"],
        "input_closure_digest": wire_digest("portfolio-input-closure"),
        "unresolved_defects": [],
        "compose_order": ["decision-compose:final"],
    }
    portfolio_contract, portfolio_decision = _pipeline_unit(
        b,
        prefix="portfolio",
        phase="phase-2",
        lane="convergence",
        kind="candidate-portfolio",
        output_schema="candidate-portfolio/v1",
        objective="恰好采用一个 StructuredScript 并处置其余候选",
        artifact_model="CandidatePortfolioArtifactV1",
        artifact_payload=portfolio_payload,
        x=6040,
        y=1810,
        index=11,
        artifact_version=150,
        preset="script_portfolio_worker",
    )
    b.link(compose_decision, portfolio_contract, "candidate closure")
    b.link(alternate_decision, portfolio_contract, "candidate closure")
    checkpoint = b.add(
        "phase2:checkpoint",
        "checkpoint",
        "phase-2",
        "boundary",
        "Phase 2 Checkpoint",
        "PHASE_TWO_CANDIDATE_PORTFOLIO_READY",
        "partial",
        7840,
        1810,
        "TaskView",
        task_payload("task-root", "交付最终脚本", "root-delivery", parent=None),
        agent_role="planner",
        tags=["partial", "closure-ready"],
        metrics=[
            ("accepted direction", "1"),
            ("accepted portfolio", "1"),
            ("active op", "0"),
        ],
    )
    b.link(portfolio_decision, checkpoint, "closure ready")

    migration = b.add(
        "phase3:migration",
        "checkpoint",
        "phase-3",
        "continuation",
        "Policy Continuation",
        "script-build-phase-three/v1 + 唯一 migration event",
        "migrated",
        8200,
        180,
        "TaskSpecView",
        task_spec("执行 Root Delivery 并发布", "root-delivery"),
        agent_role="planner",
        task_kind="root-delivery",
        tags=["reentrant", "policy-v1"],
        metrics=[("message parent", "closed"), ("migration", "unique")],
    )
    b.link(checkpoint, migration, "advance")

    canonical_payload = legacy_canonical_payload()
    canonical_sha256 = canonical_wire_digest(canonical_payload)
    root_manifest_payload = {
        "schema_version": "root-delivery-manifest/v1",
        "direction_ref": artifact_uri(110),
        "candidate_portfolio_ref": artifact_uri(150),
        "structured_script_ref": artifact_uri(140),
        "input_closure_digest": root_input_closure_digest(),
        "legacy_projection_digest": canonical_sha256,
        "build_summary": "以事故切入，完整解释智能创作系统的合同、验证和原子发布。",
        "blocking_defects": [],
    }
    root_contract, root_decision = _pipeline_unit(
        b,
        prefix="root-delivery",
        phase="phase-3",
        lane="root-delivery",
        kind="root-delivery",
        output_schema="root-delivery-manifest/v1",
        objective="锁定 Direction、Portfolio、StructuredScript 并生成唯一交付 Manifest",
        artifact_model="RootDeliveryManifestV1",
        artifact_payload=root_manifest_payload,
        x=8200,
        y=500,
        index=12,
        artifact_version=160,
        preset="script_root_worker",
        validator_preset="script_root_validator",
    )
    b.link(migration, root_contract, "host-frozen contract")

    artifact_version_row = b.add(
        "storage:artifact-version-160",
        "artifact",
        "phase-3",
        "storage",
        "ArtifactVersion #160",
        "root_delivery_manifest · frozen canonical_json",
        "frozen",
        9100,
        1020,
        "ArtifactVersionRow",
        {
            "id": 160,
            "script_build_id": RUN_ID,
            "task_id": "root-delivery",
            "attempt_id": "attempt-root-delivery",
            "spec_version": 1,
            "artifact_type": "root_delivery_manifest",
            "legacy_branch_id": None,
            "base_revision": None,
            "parent_artifact_version_id": None,
            "canonical_json": root_manifest_payload,
            "canonical_sha256": ARTIFACT_DIGESTS[160].removeprefix("sha256:"),
            "state": "frozen",
            "created_at": STAMP,
            "frozen_at": STAMP,
            "published_at": None,
        },
        task_kind="root-delivery",
        tags=["physical-row", "artifact-version"],
        metrics=[("artifact_type", "manifest"), ("branch", "null")],
    )
    b.link("root-delivery:artifact", artifact_version_row, "persist", "artifact")

    canonical = b.add(
        "publication:canonical-dry-run",
        "readback",
        "phase-3",
        "root-delivery",
        "Legacy Projection Dry-run",
        "物理 ID 无关 canonical graph",
        "matched",
        9100,
        820,
        "LegacyProjectionCanonicalV1",
        canonical_payload,
        tags=["canonical", "sha256"],
        metrics=[("digest", canonical_sha256[:12]), ("links", "2")],
    )
    b.link("root-delivery:attempt", canonical, "dry run", "artifact")
    b.link(canonical, "root-delivery:artifact", "digest", "artifact")

    journal = b.add(
        "http:finalize-command",
        "checkpoint",
        "delivery",
        "http",
        "HTTP Idempotency Journal",
        "POST /finalize · bounded response replay",
        "completed",
        10100,
        180,
        "HttpCommandRecord",
        {
            "id": 801,
            "principal_scope_sha256": digest("tenant:fake"),
            "route_family": "script-build-finalize",
            "idempotency_key": "fake-finalize-10001",
            "request_fingerprint": digest("finalize-request"),
            "preallocated_root_trace_id": ROOT_TRACE_ID,
            "resource_id": RUN_ID,
            "state": "completed",
            "response_status": 200,
            "response_json": {"script_build_id": RUN_ID, "state": "published"},
            "created_at": START,
            "updated_at": STAMP,
        },
        tags=["idempotent", "replay"],
        metrics=[("HTTP", "200"), ("state", "completed")],
    )
    b.link(root_decision, journal, "finalize")

    publication = b.add(
        "publication:identity",
        "publication",
        "delivery",
        "publisher",
        "Final Publication Identity",
        "(script_build_id, publication_type) 唯一",
        "published",
        10400,
        500,
        "Publication",
        {
            "publication_id": 702,
            "script_build_id": RUN_ID,
            "publication_type": "final",
            "accept_decision_id": "decision-root-delivery",
            "artifact_version_id": 160,
            "expected_sha256": ARTIFACT_DIGESTS[160],
            "state": "published",
            "attempt_count": 1,
            "publication_revision": 1,
            "last_error_code": None,
            "last_error_summary": None,
            "created_at": START,
            "updated_at": STAMP,
            "published_at": STAMP,
        },
        tags=["unique", "final"],
        metrics=[("attempt", "1"), ("revision", "1")],
    )
    b.link(journal, publication, "reserve", "publication")

    tx_steps = [
        ("locks", "固定锁序", "binding → build → publication → content", 10700, 260),
        ("clear", "清理 branch 0", "Link → Paragraph → Element", 11000, 260),
        ("paragraphs", "插入 Paragraph", "父段 → 子段 · logical map", 11300, 260),
        ("elements", "插入 Element", "natural key → physical ID", 11600, 260),
        ("links", "插入 Link", "paragraph/element logical pair", 11900, 260),
        (
            "readback",
            "同 Session canonical 回读",
            "digest 必须与 Manifest 一致",
            12200,
            260,
        ),
        (
            "pointers",
            "发布指针与状态",
            "accepted pointer + artifacts + build",
            12500,
            260,
        ),
        ("commit", "单次 Commit", "全部成功或全部 rollback", 12800, 260),
    ]
    physical_rows: dict[str, tuple[str, dict[str, Any]]] = {
        "paragraphs": (
            "ParagraphRow",
            {
                "id": 9001,
                "script_build_id": RUN_ID,
                "branch_id": 0,
                "base_ref_id": None,
                "paragraph_index": 1,
                "level": 1,
                "parent_id": None,
                "name": PARAGRAPHS[0]["name"],
                "content_range": PARAGRAPHS[0]["content_range"],
                "theme": PARAGRAPHS[0]["theme"],
                "form": PARAGRAPHS[0]["form"],
                "function": PARAGRAPHS[0]["function"],
                "feeling": PARAGRAPHS[0]["feeling"],
                "theme_elements": PARAGRAPHS[0]["theme_elements"],
                "form_elements": PARAGRAPHS[0]["form_elements"],
                "function_elements": PARAGRAPHS[0]["function_elements"],
                "feeling_elements": PARAGRAPHS[0]["feeling_elements"],
                "description": PARAGRAPHS[0]["description"],
                "full_description": PARAGRAPHS[0]["full_description"],
                "is_active": True,
                "created_at": STAMP,
                "updated_at": STAMP,
            },
        ),
        "elements": (
            "ElementRow",
            {
                "id": 9101,
                "script_build_id": RUN_ID,
                "branch_id": 0,
                "base_ref_id": None,
                "name": ELEMENTS[0]["name"],
                "dimension_primary": ELEMENTS[0]["dimension_primary"],
                "dimension_secondary": ELEMENTS[0]["dimension_secondary"],
                "commonality_analysis": ELEMENTS[0]["commonality_analysis"],
                "topic_support": ELEMENTS[0]["topic_support"],
                "weight_score": ELEMENTS[0]["weight_score"],
                "support_elements": ELEMENTS[0]["support_elements"],
                "is_active": True,
                "created_at": STAMP,
                "updated_at": STAMP,
            },
        ),
        "links": (
            "ParagraphElementLinkRow",
            {
                "id": 9201,
                "script_build_id": RUN_ID,
                "branch_id": 0,
                "paragraph_id": 9001,
                "element_id": 9101,
            },
        ),
    }
    previous = publication
    for offset, (key, title, subtitle, x, y) in enumerate(tx_steps, 20):
        default_record = (
            "MissionOwnerToken",
            {
                "script_build_id": RUN_ID,
                "root_trace_id": ROOT_TRACE_ID,
                "owner_epoch": 1,
                "stop_epoch": 0,
                "owner_instance_id": "host-fake-01",
            },
        )
        model_name, row_payload = physical_rows.get(key, default_record)
        node = b.add(
            f"uow:{key}",
            "transaction",
            "delivery",
            "final-uow",
            title,
            subtitle,
            "committed",
            x,
            y,
            model_name,
            row_payload,
            tags=[
                "one-session",
                "physical-row" if key in physical_rows else "fenced",
            ],
            metrics=[("step", str(offset - 19)), ("tx", "same")],
        )
        b.link(previous, node, None, "publication", animated=True)
        previous = node

    paragraph_two = dict(physical_rows["paragraphs"][1])
    paragraph_two.update(
        {
            "id": 9002,
            "paragraph_index": 2,
            "level": 2,
            "parent_id": 9001,
            "name": PARAGRAPHS[1]["name"],
            "content_range": PARAGRAPHS[1]["content_range"],
            "theme": PARAGRAPHS[1]["theme"],
            "form": PARAGRAPHS[1]["form"],
            "function": PARAGRAPHS[1]["function"],
            "feeling": PARAGRAPHS[1]["feeling"],
            "theme_elements": PARAGRAPHS[1]["theme_elements"],
            "form_elements": PARAGRAPHS[1]["form_elements"],
            "function_elements": PARAGRAPHS[1]["function_elements"],
            "feeling_elements": PARAGRAPHS[1]["feeling_elements"],
            "description": PARAGRAPHS[1]["description"],
            "full_description": PARAGRAPHS[1]["full_description"],
        }
    )
    element_two = dict(physical_rows["elements"][1])
    element_two.update(
        {
            "id": 9102,
            "name": ELEMENTS[1]["name"],
            "dimension_primary": ELEMENTS[1]["dimension_primary"],
            "dimension_secondary": ELEMENTS[1]["dimension_secondary"],
            "commonality_analysis": ELEMENTS[1]["commonality_analysis"],
            "topic_support": ELEMENTS[1]["topic_support"],
            "weight_score": ELEMENTS[1]["weight_score"],
            "support_elements": ELEMENTS[1]["support_elements"],
        }
    )
    extra_rows = [
        ("paragraphs", "ParagraphRow #2", "ParagraphRow", paragraph_two, 11300),
        ("elements", "ElementRow #2", "ElementRow", element_two, 11600),
        (
            "links",
            "ParagraphElementLinkRow #2",
            "ParagraphElementLinkRow",
            {
                "id": 9202,
                "script_build_id": RUN_ID,
                "branch_id": 0,
                "paragraph_id": 9002,
                "element_id": 9102,
            },
            11900,
        ),
    ]
    for key, title, model_name, row_payload, x in extra_rows:
        extra = b.add(
            f"uow:{key}-row-2",
            "transaction",
            "delivery",
            "final-uow-rows",
            title,
            "完整发布集合 · row 2 / 2",
            "committed",
            x,
            620,
            model_name,
            row_payload,
            tags=["one-session", "physical-row"],
            metrics=[("row", "2 / 2"), ("branch", "0")],
        )
        b.link(f"uow:{key}", extra, "row 2", "artifact")

    result = b.add(
        "publication:result",
        "publication",
        "delivery",
        "publisher",
        "Publication Result",
        "Manifest 与 StructuredScript 已 published",
        "published",
        13100,
        500,
        "PublicationResult",
        {
            "script_build_id": RUN_ID,
            "publication_id": 702,
            "publication_type": "final",
            "state": "published",
            "legacy_projection_digest": canonical_sha256,
            "committed": True,
            "post_commit_readback_ok": True,
        },
        tags=["committed", "readback-ok"],
        metrics=[("build", "success"), ("readback", "OK")],
    )
    b.link(previous, result, "commit ack", "publication")
    legacy = b.add(
        "legacy:http-readback",
        "readback",
        "delivery",
        "compatibility",
        "Legacy HTTP Detail",
        "GET /api/pattern/script_builds/10001",
        "200",
        13400,
        500,
        "ScriptBuildRecord",
        {
            "id": RUN_ID,
            "execution_id": 88001,
            "topic_build_id": 67001,
            "topic_id": 7701,
            "agent_type": "AigcAgent",
            "agent_config": {"model": "deterministic-fake-model"},
            "data_source_url": None,
            "status": "success",
            "reson_trace_id": ROOT_TRACE_ID,
            "is_deleted": False,
            "is_favorited": True,
            "summary": "完成一篇讲解智能创作系统工程边界的结构化脚本。",
            "error_message": None,
            "script_direction": "用失控案例讲清合同、验证与原子发布。",
            "build_workflow": {"checkpoint": "PHASE_THREE_PUBLISHED"},
            "paragraph_count": 2,
            "element_count": 2,
            "input_tokens": 28420,
            "output_tokens": 9360,
            "cost_usd": 0.1842,
            "strategies_config": {"always_on": [3], "on_demand": [8]},
            "start_time": START,
            "end_time": STAMP,
        },
        tags=["compat", "200"],
        metrics=[("paragraphs", "2"), ("elements", "2")],
    )
    b.link(result, legacy, "health readback", "publication")

    frames = [
        PhaseFrame(
            id="frame-phase-1",
            title="Phase 1 · Mission 与方向",
            subtitle="冻结输入、绑定 Root、方向 Worker / Validator / ACCEPT",
            x=0,
            y=0,
            width=2460,
            height=1040,
            tone="coral",
        ),
        PhaseFrame(
            id="frame-phase-2",
            title="Phase 2 · Evidence 与候选闭包",
            subtitle="四路取数、结构、段落、元素、比较、合成与 Portfolio",
            x=2480,
            y=0,
            width=5600,
            height=2400,
            tone="blue",
        ),
        PhaseFrame(
            id="frame-phase-3",
            title="Phase 3 · Root Delivery",
            subtitle="Policy continuation、唯一 Manifest、Root Validation 与 ACCEPT",
            x=8120,
            y=0,
            width=1920,
            height=1320,
            tone="purple",
        ),
        PhaseFrame(
            id="frame-delivery",
            title="Final Delivery · 原子发布",
            subtitle="HTTP journal、唯一 publication、单 Session / 单 Commit 与旧 API 回读",
            x=10060,
            y=0,
            width=3740,
            height=1120,
            tone="green",
        ),
    ]
    summary = RunSummary(
        script_build_id=RUN_ID,
        title="AI Agent 项目为什么总在最后一步失控",
        status="success",
        checkpoint="PHASE_THREE_PUBLISHED",
        root_trace_id=ROOT_TRACE_ID,
        publication_state="published",
        node_count=len(b.nodes),
        created_at=START,
        updated_at=STAMP,
    )
    return RunGraph(
        schema_version="script-build-visualization/v1",
        data_mode="fake",
        generated_at=STAMP,
        run=summary,
        frames=frames,
        nodes=b.nodes,
        edges=b.edges,
        schema_catalog={key: list(value) for key, value in SCHEMA_CATALOG.items()},
        enum_catalog={key: list(value) for key, value in ENUM_CATALOG.items()},
        declared_gaps=[
            "当前 Host 没有面向可视化的聚合 execution-view API；此后端只返回 fake graph。",
            "TraceStore 的消息父链、policy migration 与 fencing 审计尚无统一只读投影。",
            "真实运行中模型 prompt 全量快照、工具调用参数与 Operation 时间线需要专用脱敏接口。",
            "候选 Artifact 的 logical lineage 可以由现有字段还原，但尚无稳定分页/批量读取 API。",
        ],
    )


GRAPH = build_graph()


def node_detail(node_id: str) -> NodeDetail | None:
    node = next((item for item in GRAPH.nodes if item.id == node_id), None)
    if node is None:
        return None
    return NodeDetail(
        data_mode="fake",
        run_id=RUN_ID,
        node=node,
        incoming=[edge for edge in GRAPH.edges if edge.target == node_id],
        outgoing=[edge for edge in GRAPH.edges if edge.source == node_id],
    )