工件库：内容寻址存储、节点输出与工具快照落盘、运行清单哈希

事件流里的输出是给画布看的截断预览，出具溯源需要完整证据。
新增 data/artifacts/ 内容寻址工件库（sha256 即地址，取回时复验哈希，
篡改即报错）；节点完整输出与工具调用快照（query+结果集）自动落盘，
事件只携带工件 id。运行终态对全部事件计算清单哈希存入 Run，
事后对事件流的增删改会与它对不上。

graph_hash 剔除 viewport 后计算——挪画布不改变图的身份。
Run/Workflow/WorkflowVersion/Approval 预置分级与治理列，
init_db 附带 SQLite 轻量列迁移。

backend/app/core/artifact_store.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from app.core.config import settings
+
+# --------------------------------------------------------------------------
+# 内容寻址的工件库
+#
+# 事件流里的输出是给画布看的截断预览；这里存的是完整证据。
+# 工件按内容 sha256 寻址：同一份内容只存一份，且任何篡改都会改变地址本身，
+# 这比"相信数据库没被改过"强得多。出具溯源、周对比 diff、基线回归、
+# 轨迹提模板，全部踩在这一层上。
+# --------------------------------------------------------------------------
+
+
+def canonical_json(obj: Any) -> str:
+    """规范化 JSON：键排序、紧凑分隔符。同一数据永远得到同一字节串，哈希才稳定。"""
+    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
+
+
+def content_hash(data: bytes | str) -> str:
+    if isinstance(data, str):
+        data = data.encode()
+    return hashlib.sha256(data).hexdigest()
+
+
+def graph_hash(graph: dict[str, Any]) -> str:
+    """图定义的指纹。formal run 钉住的就是这个值。
+
+    viewport 是纯视觉状态，挪一下画布不该改变图的身份，剔除后再哈希。
+    """
+    slim = {k: v for k, v in graph.items() if k != "viewport"}
+    return content_hash(canonical_json(slim))
+
+
+def _path_of(artifact_id: str) -> Path:
+    # 按前两位分桶，避免单目录塞几万个文件
+    return settings.data_dir / "artifacts" / artifact_id[:2] / f"{artifact_id}.json"
+
+
+async def put_json(
+    obj: Any,
+    *,
+    kind: str,
+    run_id: str | None = None,
+    node_id: str | None = None,
+    meta: dict[str, Any] | None = None,
+) -> str:
+    """存一个 JSON 工件，返回其内容哈希（即工件 id）。
+
+    幂等：同内容重复写只落一次盘；同一工件被多个 run 引用时，
+    数据库里各记一行引用（溯源需要知道"谁产出的"），文件只有一份。
+    """
+    from app.db.base import SessionLocal
+    from app.db.models import Artifact
+
+    text = canonical_json(obj)
+    artifact_id = content_hash(text)
+    path = _path_of(artifact_id)
+    if not path.exists():
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(".tmp")
+        tmp.write_text(text, encoding="utf-8")
+        tmp.rename(path)  # 原子落盘，写一半崩了不会留下半个工件
+
+    async with SessionLocal() as session:
+        existing = await session.get(Artifact, (artifact_id, run_id or ""))
+        if existing is None:
+            session.add(
+                Artifact(
+                    id=artifact_id,
+                    run_id=run_id or "",
+                    kind=kind,
+                    node_id=node_id,
+                    size=len(text.encode()),
+                    meta=meta or {},
+                )
+            )
+            await session.commit()
+    return artifact_id
+
+
+def load(artifact_id: str) -> Any | None:
+    """按 id 取回工件内容。取回时重新校验哈希——工件库的承诺就是内容不可变。"""
+    if not artifact_id or not artifact_id.isalnum():
+        return None
+    path = _path_of(artifact_id)
+    if not path.exists():
+        return None
+    text = path.read_text(encoding="utf-8")
+    if content_hash(text) != artifact_id:
+        raise ValueError(f"工件 {artifact_id[:12]}… 内容与哈希不符，疑似被篡改")
+    return json.loads(text)
+
+
+def manifest_hash(rows: list[tuple[int, str, str | None, dict[str, Any]]]) -> str:
+    """把一次运行的全部事件压成一个清单哈希。
+
+    rows: [(seq, type, node_id, data)]。存在 Run 上之后，事后任何对
+    事件流的增删改都能被这一个值戳穿。
+    """
+    lines = [
+        f"{seq}|{etype}|{node_id or ''}|{content_hash(canonical_json(data))}"
+        for seq, etype, node_id, data in sorted(rows, key=lambda r: r[0])
+    ]
+    return content_hash("\n".join(lines))

backend/app/core/events.py

@@ -47,6 +47,10 @@ class EventType(StrEnum):
     HUMAN_REQUESTED = "human.requested"
     HUMAN_RESOLVED = "human.resolved"
 
+    # 出具与口径治理
+    ISSUANCE = "issuance"  # 出具契约的判定结果（档位 / 缺数据 / 未回指数字）
+    CALIBER_UPGRADE = "caliber.upgrade"  # 钉住的方法卡有新版本，按声明的策略处置
+
     # 其他
     LOG = "log"
     USAGE = "usage"  # token / 成本累计

backend/app/db/base.py

@@ -62,8 +62,34 @@ async def get_session() -> AsyncIterator[AsyncSession]:
         yield session
 
 
+# 轻量迁移：create_all 只建新表，不给已有表加列。SQLite 的 ADD COLUMN
+# 便宜且带默认值回填，够用；等真需要改列类型再考虑 alembic。
+_COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [
+    ("runs", "run_class", "TEXT DEFAULT 'exploratory'"),
+    ("runs", "version", "INTEGER"),
+    ("runs", "version_hash", "VARCHAR(64)"),
+    ("runs", "manifest_hash", "VARCHAR(64)"),
+    ("runs", "started_by", "VARCHAR(100)"),
+    ("workflows", "status", "TEXT DEFAULT 'draft'"),
+    ("workflows", "published_version", "INTEGER"),
+    ("workflows", "published_by", "VARCHAR(100)"),
+    ("workflow_versions", "graph_hash", "VARCHAR(64)"),
+    ("approvals", "resolved_by", "VARCHAR(100)"),
+]
+
+
+def _migrate(conn) -> None:  # pragma: no cover - 同步回调
+    from sqlalchemy import text
+
+    for table, column, ddl in _COLUMN_MIGRATIONS:
+        cols = {row[1] for row in conn.execute(text(f"PRAGMA table_info({table})"))}
+        if cols and column not in cols:
+            conn.execute(text(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}"))
+
+
 async def init_db() -> None:
     from app.db import models  # noqa: F401  确保模型已注册到 metadata
 
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
+        await conn.run_sync(_migrate)

backend/app/db/models.py

@@ -84,6 +84,11 @@ class Workflow(Base, TimestampMixin):
     tags: Mapped[list[Any]] = mapped_column(default=list)
     version: Mapped[int] = mapped_column(Integer, default=1)
     is_template: Mapped[bool] = mapped_column(default=False)
+    # 治理状态机：draft（随便改）→ published（有正式版本可跑）→ governed（发布需过治理 lint）
+    status: Mapped[str] = mapped_column(String(20), default="draft")
+    # 正式运行默认解析到这个版本；为空表示还没发布过
+    published_version: Mapped[int | None] = mapped_column(Integer, default=None)
+    published_by: Mapped[str | None] = mapped_column(String(100), default=None)
 
     versions: Mapped[list["WorkflowVersion"]] = relationship(
         back_populates="workflow", cascade="all, delete-orphan", lazy="selectin"
@@ -101,6 +106,8 @@ class WorkflowVersion(Base, TimestampMixin):
     version: Mapped[int] = mapped_column(Integer)
     graph: Mapped[dict[str, Any]] = mapped_column(default=dict)
     note: Mapped[str] = mapped_column(Text, default="")
+    # 图内容的 sha256 指纹（剔除 viewport）。formal run 钉住的就是它。
+    graph_hash: Mapped[str | None] = mapped_column(String(64), default=None)
 
     workflow: Mapped[Workflow] = relationship(back_populates="versions")
 
@@ -130,6 +137,14 @@ class Run(Base, TimestampMixin):
     started_at: Mapped[datetime | None] = mapped_column(default=None)
     finished_at: Mapped[datetime | None] = mapped_column(default=None)
     last_seq: Mapped[int] = mapped_column(Integer, default=0)
+    # formal：从不可变的 WorkflowVersion 发起，可复现、可出具
+    # exploratory：画布试跑 / 裸 graph / 追问，结果只标探索性
+    run_class: Mapped[str] = mapped_column(String(20), default="exploratory")
+    version: Mapped[int | None] = mapped_column(Integer, default=None)
+    version_hash: Mapped[str | None] = mapped_column(String(64), default=None)
+    # 结束时对全部事件计算的清单哈希，事后改动事件流会被它戳穿
+    manifest_hash: Mapped[str | None] = mapped_column(String(64), default=None)
+    started_by: Mapped[str | None] = mapped_column(String(100), default=None)
 
     events: Mapped[list["RunEvent"]] = relationship(
         back_populates="run", cascade="all, delete-orphan"
@@ -169,6 +184,27 @@ class Approval(Base, TimestampMixin):
     status: Mapped[str] = mapped_column(String(20), default="pending", index=True)
     response: Mapped[dict[str, Any]] = mapped_column(default=dict)
     resolved_at: Mapped[datetime | None] = mapped_column(default=None)
+    # 责任归属：谁批的。没有 actor 的治理是空转。
+    resolved_by: Mapped[str | None] = mapped_column(String(100), default=None)
+
+
+class Artifact(Base, TimestampMixin):
+    """工件引用表。
+
+    文件本体按内容哈希存在 data/artifacts/ 下（同内容只一份）；
+    这张表记录"哪个 run 的哪个节点产出了它"——溯源需要的正是这层归属。
+    复合主键：同一工件被多个 run 产出时各记一行引用。
+    """
+
+    __tablename__ = "artifacts"
+    __table_args__ = (Index("ix_artifacts_run", "run_id"),)
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    run_id: Mapped[str] = mapped_column(String(32), primary_key=True, default="")
+    kind: Mapped[str] = mapped_column(String(32), default="node_output")  # node_output | tool_snapshot | metric_set
+    node_id: Mapped[str | None] = mapped_column(String(64), default=None)
+    size: Mapped[int] = mapped_column(Integer, default=0)
+    meta: Mapped[dict[str, Any]] = mapped_column(default=dict)
 
 
 # --------------------------------------------------------------------------

backend/app/engine/compiler.py

@@ -71,11 +71,33 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita
             try:
                 updates = await runner(state, ctx)
                 elapsed = int((time.perf_counter() - started) * 1000)
+
+                # 完整输出落工件库。事件里只放截断预览是给画布看的；
+                # 出具溯源、周对比、轨迹提模板需要的是这份完整证据。
+                artifact_id: str | None = None
+                payload = (updates.get("nodes") or {}).get(node.id)
+                if payload is not None:
+                    from app.core.artifact_store import put_json
+
+                    try:
+                        artifact_id = await put_json(
+                            payload,
+                            kind="node_output",
+                            run_id=run_ctx.run_id,
+                            node_id=node.id,
+                            meta={"type": str(node.type), "attempt": attempt + 1},
+                        )
+                        if isinstance(payload, dict):
+                            payload["__artifact__"] = artifact_id
+                    except Exception:  # noqa: BLE001 - 工件写失败不该毁掉运行本身
+                        artifact_id = None
+
                 ctx.emit(
                     EventType.NODE_FINISHED,
                     duration_ms=elapsed,
                     attempt=attempt + 1,
                     preview=_preview(updates, node.id),
+                    artifact=artifact_id,
                 )
                 trail = {
                     "node_id": node.id,

backend/app/engine/nodes/llm.py

@@ -326,12 +326,24 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 content = f"工具执行失败：{type(e).__name__}: {e}"
                 ok = False
             elapsed = int((time.perf_counter() - started) * 1000)
+            snapshot_id: str | None = None
+            if ok:
+                from app.core.artifact_store import put_json
+
+                try:
+                    snapshot_id = await put_json(
+                        {"tool": name, "args": args, "result": content},
+                        kind="tool_snapshot", run_id=ctx.run.run_id, node_id=ctx.node.id,
+                    )
+                except Exception:  # noqa: BLE001
+                    snapshot_id = None
             ctx.emit(
                 EventType.TOOL_END if ok else EventType.TOOL_ERROR,
                 tool=name,
                 call_id=call_id,
                 duration_ms=elapsed,
                 preview=content[:2000],
+                artifact=snapshot_id,
             )
             transcript.append(
                 {"tool": name, "args": args, "ok": ok, "duration_ms": elapsed, "result": content[:4000]}

backend/app/engine/nodes/tools.py

@@ -67,8 +67,19 @@ async def run_tool(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
         result = {"error": f"{type(e).__name__}: {e}"}
 
     elapsed = int((time.perf_counter() - started) * 1000)
+    # 取数快照：query（args）和结果集一起进工件库，正式出具时数字回指的就是它
+    from app.core.artifact_store import put_json
+
+    try:
+        snapshot_id = await put_json(
+            {"tool": name, "args": args, "result": result},
+            kind="tool_snapshot", run_id=ctx.run.run_id, node_id=ctx.node.id,
+        )
+    except Exception:  # noqa: BLE001
+        snapshot_id = None
     ctx.emit(EventType.TOOL_END, tool=name, duration_ms=elapsed,
-             preview=json.dumps(result, ensure_ascii=False, default=str)[:2000])
+             preview=json.dumps(result, ensure_ascii=False, default=str)[:2000],
+             artifact=snapshot_id)
 
     updates: dict[str, Any] = {"nodes": {ctx.node.id: result}}
     var_name = ctx.cfg("assign_to", "")

backend/app/engine/runner.py

@@ -100,6 +100,13 @@ class RunManager:
                 )
                 await session.commit()
 
+    async def note(
+        self, run_id: str, event_type: EventType | str, *, node_id: str | None = None,
+        **data: Any,
+    ) -> None:
+        """给 API 层用的事件入口（比如 formal 启动时记录口径升版处置）。"""
+        await self._emit(run_id, event_type, node_id=node_id, data=data)
+
     # ---------------- 启动 ----------------
 
     async def start(
@@ -111,6 +118,10 @@ class RunManager:
         workflow_name: str = "",
         memory_scope: str = "default",
         collection: str = "default",
+        run_class: str = "exploratory",
+        version: int | None = None,
+        version_hash: str | None = None,
+        started_by: str | None = None,
     ) -> Run:
         spec = GraphSpec.model_validate(graph)
         report = validate_graph(spec)
@@ -126,6 +137,10 @@ class RunManager:
                 status="queued",
                 graph=graph,
                 input=input_payload,
+                run_class=run_class,
+                version=version,
+                version_hash=version_hash,
+                started_by=started_by,
             )
             run.thread_id = run.id  # 一个 run 一条 checkpoint 线程
             session.add(run)
@@ -343,6 +358,16 @@ class RunManager:
                 run.usage = usage
                 if status != "interrupted":
                     run.finished_at = datetime.now(timezone.utc)
+                    # 终态时对全部已落库事件计算清单哈希：事后任何对事件流的
+                    # 增删改都会与这个值对不上。中断态不算——事件还会继续追加。
+                    from app.core.artifact_store import manifest_hash as _manifest
+
+                    rows = await session.execute(
+                        select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data)
+                        .where(RunEvent.run_id == run_id)
+                        .order_by(RunEvent.seq)
+                    )
+                    run.manifest_hash = _manifest([tuple(r) for r in rows])
                 await session.commit()
 
         if status == "succeeded":