"""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], )