运行分级与版本钉死：formal/exploratory、方法卡钉版、口径升版被迫处置

formal 运行的全部语义就一条规则：必须引用一个不可变的已发布
WorkflowVersion（按 graph_hash 钉住）；传裸 graph 的运行自动是
exploratory。发布后改图不影响已发布版本——formal 永远执行快照。

子图节点支持 workflow_version 钉版，这就是方法卡的机制核心：
钉住后上游改方法卡不会让口径悄悄漂移。钉住的方法卡出现新版本时，
formal 启动被拒，除非节点已声明 upgrade_policy（recompute 回算 /
dual 双印 / incomparable 标注不可比，无默认值）；已声明的升版
以 caliber.upgrade 事件记入流水。

发布端点带两级门禁：published 只拦基础错误，governed 额外过治理
lint（禁 supervisor、方法卡必须钉版、agent 不得关闭全部审批、
output 必须声明出具契约）。X-Actor 记录发布人与运行发起人。

backend/app/api/artifacts.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+
+from app.core import artifact_store
+
+router = APIRouter(prefix="/api/artifacts", tags=["artifacts"])
+
+_MAX_INLINE = 2 * 1024 * 1024
+
+
+@router.get("/{artifact_id}")
+async def get_artifact(artifact_id: str) -> dict[str, Any]:
+    """按内容哈希取回一个工件。取回时会复验哈希——内容不可变是这层的全部承诺。"""
+    try:
+        content = artifact_store.load(artifact_id)
+    except ValueError as e:
+        # 哈希对不上：明确报给调用方，这正是工件库存在的意义
+        raise HTTPException(409, str(e)) from e
+    if content is None:
+        raise HTTPException(404, "工件不存在")
+    return {"id": artifact_id, "content": content}

backend/app/api/runs.py

@@ -4,15 +4,20 @@ import asyncio
 import contextlib
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, WebSocket, WebSocketDisconnect
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from sqlalchemy import select as _select
+
 from app.core.bus import bus
+from app.core import artifact_store
 from app.db.base import SessionLocal, get_session
-from app.db.models import Approval, Run, RunEvent, Workflow
+from app.db.models import Approval, Artifact, Run, RunEvent, Workflow, WorkflowVersion
+from app.engine.governance import unresolved_caliber_upgrades
 from app.engine.runner import run_manager
+from app.engine.schema import GraphSpec
 
 router = APIRouter(prefix="/api/runs", tags=["runs"])
 
@@ -23,6 +28,9 @@ class StartRunIn(BaseModel):
     input: dict[str, Any] = Field(default_factory=dict)
     memory_scope: str = "default"
     collection: str = "default"
+    # formal：从已发布的不可变版本发起，可复现可出具；exploratory：随便玩
+    run_class: str = "exploratory"
+    version: int | None = None  # formal 时可指定版本，默认用已发布版本
 
 
 class RunOut(BaseModel):
@@ -34,6 +42,11 @@ class RunOut(BaseModel):
     output: dict[str, Any]
     error: str | None
     usage: dict[str, Any]
+    run_class: str | None = "exploratory"
+    version: int | None = None
+    version_hash: str | None = None
+    manifest_hash: str | None = None
+    started_by: str | None = None
     created_at: Any = None
     started_at: Any = None
     finished_at: Any = None
@@ -42,30 +55,83 @@ class RunOut(BaseModel):
 
 
 @router.post("", response_model=RunOut, status_code=201)
-async def start_run(payload: StartRunIn, session: AsyncSession = Depends(get_session)) -> Run:
-    graph = payload.graph
+async def start_run(
+    payload: StartRunIn,
+    session: AsyncSession = Depends(get_session),
+    x_actor: str | None = Header(default=None),
+) -> Run:
+    actor = (x_actor or "").strip() or None
     name = "临时图"
-    if payload.workflow_id:
+    version: int | None = None
+    version_hash: str | None = None
+
+    if payload.run_class == "formal":
+        # 正式运行的全部语义就这一条规则：必须引用一个不可变的已发布版本。
+        # 传裸 graph、或工作流还没发布过，都进不了 formal。
+        if not payload.workflow_id:
+            raise HTTPException(400, "正式运行必须指定 workflow_id")
+        if payload.graph is not None:
+            raise HTTPException(400, "正式运行不接受临时 graph——先保存并发布版本")
         workflow = await session.get(Workflow, payload.workflow_id)
         if not workflow:
             raise HTTPException(404, "工作流不存在")
-        graph = graph or workflow.graph
+        if workflow.status not in ("published", "governed") or not workflow.published_version:
+            raise HTTPException(409, f"「{workflow.name}」还没有发布版本，先在编排页发布")
+        version = payload.version or workflow.published_version
+        snapshot = (
+            await session.execute(
+                _select(WorkflowVersion).where(
+                    WorkflowVersion.workflow_id == workflow.id,
+                    WorkflowVersion.version == version,
+                )
+            )
+        ).scalar_one_or_none()
+        if not snapshot:
+            raise HTTPException(404, f"版本 v{version} 不存在")
+        graph = snapshot.graph
+        version_hash = snapshot.graph_hash or artifact_store.graph_hash(graph)
         name = workflow.name
+
+        # 口径升版是被迫处置的事件：钉住的方法卡有新版本而未声明策略，拒绝启动
+        spec = GraphSpec.model_validate(graph)
+        errors, upgrade_events = await unresolved_caliber_upgrades(session, spec)
+        if errors:
+            raise HTTPException(409, "；".join(errors))
+    else:
+        graph = payload.graph
+        upgrade_events = []
+        if payload.workflow_id:
+            workflow = await session.get(Workflow, payload.workflow_id)
+            if not workflow:
+                raise HTTPException(404, "工作流不存在")
+            graph = graph or workflow.graph
+            name = workflow.name
     if not graph:
         raise HTTPException(400, "需要提供 workflow_id 或 graph")
 
     try:
-        return await run_manager.start(
+        run = await run_manager.start(
             graph=graph,
             input_payload=payload.input,
             workflow_id=payload.workflow_id,
             workflow_name=name,
             memory_scope=payload.memory_scope,
             collection=payload.collection,
+            run_class="formal" if payload.run_class == "formal" else "exploratory",
+            version=version,
+            version_hash=version_hash,
+            started_by=actor,
         )
     except ValueError as e:
         raise HTTPException(400, str(e)) from e
 
+    # 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的"
+    for ev in upgrade_events:
+        payload_ev = dict(ev)
+        node = payload_ev.pop("node_id", None)
+        await run_manager.note(run.id, "caliber.upgrade", node_id=node, **payload_ev)
+    return run
+
 
 @router.get("", response_model=list[RunOut])
 async def list_runs(
@@ -116,6 +182,27 @@ async def get_events(
     ]
 
 
+@router.get("/{run_id}/artifacts")
+async def list_artifacts(
+    run_id: str, session: AsyncSession = Depends(get_session)
+) -> list[dict[str, Any]]:
+    """这次运行产出的全部工件引用。审阅出具物时从这里下钻到完整证据。"""
+    rows = await session.execute(
+        select(Artifact).where(Artifact.run_id == run_id).order_by(Artifact.created_at)
+    )
+    return [
+        {
+            "id": a.id,
+            "kind": a.kind,
+            "node_id": a.node_id,
+            "size": a.size,
+            "meta": a.meta,
+            "created_at": a.created_at.isoformat() if a.created_at else None,
+        }
+        for a in rows.scalars()
+    ]
+
+
 @router.post("/{run_id}/cancel")
 async def cancel_run(run_id: str) -> dict[str, Any]:
     stopped = await run_manager.cancel(run_id)

backend/app/api/workflows.py

@@ -2,13 +2,15 @@ from __future__ import annotations
 
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, Header, HTTPException
 from pydantic import BaseModel, Field
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.artifact_store import graph_hash
 from app.db.base import get_session
 from app.db.models import Run, Workflow, WorkflowVersion
+from app.engine.governance import lint_for_publish
 from app.engine.schema import GraphSpec, validate_graph
 
 router = APIRouter(prefix="/api/workflows", tags=["workflows"])
@@ -37,6 +39,8 @@ class WorkflowOut(BaseModel):
     tags: list[str]
     version: int
     is_template: bool
+    status: str = "draft"
+    published_version: int | None = None
     created_at: Any = None
     updated_at: Any = None
     run_count: int = 0
@@ -73,7 +77,8 @@ async def create_workflow(
     session.add(workflow)
     await session.flush()
     session.add(
-        WorkflowVersion(workflow_id=workflow.id, version=1, graph=payload.graph, note="初始版本")
+        WorkflowVersion(workflow_id=workflow.id, version=1, graph=payload.graph,
+                        graph_hash=graph_hash(payload.graph), note="初始版本")
     )
     await session.commit()
     await session.refresh(workflow)
@@ -113,6 +118,7 @@ async def update_workflow(
                 workflow_id=workflow.id,
                 version=workflow.version,
                 graph=payload.graph,
+                graph_hash=graph_hash(payload.graph),
                 note=payload.note,
             )
         )
@@ -197,6 +203,7 @@ async def restore_version(
             workflow_id=workflow.id,
             version=workflow.version,
             graph=snapshot.graph,
+            graph_hash=graph_hash(snapshot.graph),
             note=f"回滚到 v{version}",
         )
     )
@@ -205,6 +212,68 @@ async def restore_version(
     return workflow
 
 
+class PublishIn(BaseModel):
+    level: str = "published"  # published | governed
+    version: int | None = None  # 默认发布当前最新版本
+
+
+@router.post("/{workflow_id}/publish")
+async def publish_workflow(
+    workflow_id: str,
+    payload: PublishIn,
+    session: AsyncSession = Depends(get_session),
+    x_actor: str | None = Header(default=None),
+) -> dict[str, Any]:
+    """把某个版本立为正式版本。
+
+    formal 运行只能从这里发布过的版本发起。governed 级别额外过治理 lint：
+    错误会挡住发布——这就是"必经检查点"落地的地方。
+    返回 200 + ok/issues 而不是抛错，让前端能把问题列表渲染出来。
+    """
+    if payload.level not in ("published", "governed"):
+        raise HTTPException(400, "level 只能是 published 或 governed")
+    workflow = await session.get(Workflow, workflow_id)
+    if not workflow:
+        raise HTTPException(404, "工作流不存在")
+
+    version = payload.version or workflow.version
+    snapshot = (
+        await session.execute(
+            select(WorkflowVersion).where(
+                WorkflowVersion.workflow_id == workflow_id,
+                WorkflowVersion.version == version,
+            )
+        )
+    ).scalar_one_or_none()
+    if not snapshot:
+        raise HTTPException(404, f"版本 v{version} 不存在，先保存一次")
+
+    try:
+        spec = GraphSpec.model_validate(snapshot.graph)
+    except Exception as e:  # noqa: BLE001
+        return {"ok": False, "issues": [{"level": "error", "message": f"图结构非法：{e}"}]}
+
+    issues = [i.model_dump() for i in validate_graph(spec).issues]
+    issues += [i.model_dump() for i in lint_for_publish(spec, level=payload.level).issues]
+    if any(i["level"] == "error" for i in issues):
+        return {"ok": False, "level": payload.level, "version": version, "issues": issues}
+
+    if not snapshot.graph_hash:
+        snapshot.graph_hash = graph_hash(snapshot.graph)
+    workflow.status = payload.level
+    workflow.published_version = version
+    workflow.published_by = (x_actor or "").strip() or None
+    await session.commit()
+    return {
+        "ok": True,
+        "level": payload.level,
+        "version": version,
+        "graph_hash": snapshot.graph_hash,
+        "published_by": workflow.published_by,
+        "issues": issues,  # 剩下的都是警告
+    }
+
+
 class ValidateIn(BaseModel):
     graph: dict[str, Any]
 

backend/app/engine/compiler.py

@@ -80,6 +80,8 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita
                     from app.core.artifact_store import put_json
 
                     try:
+                        # 只记录、不回写：payload 和 run.output 可能是同一个对象，
+                        # 往里塞 __artifact__ 会污染最终成果。事件里携带 id 就够了。
                         artifact_id = await put_json(
                             payload,
                             kind="node_output",
@@ -87,8 +89,6 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita
                             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
 

backend/app/engine/governance.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.engine.schema import GraphSpec, NodeType, ValidationResult
+
+# 口径升版的三种合法处置。没有默认值是刻意的：升版最容易静默翻车的地方
+# 就是"上周和这周的数其实不可比，但报表看起来一切正常"。
+UPGRADE_POLICIES = {"recompute", "dual", "incomparable"}
+
+UPGRADE_POLICY_LABELS = {
+    "recompute": "用新口径回算历史",
+    "dual": "并排双印新旧口径",
+    "incomparable": "标注与历史不可比",
+}
+
+
+def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult:
+    """发布门禁。
+
+    published：基础可运行性之外只给警告。
+    governed：受管模板，出具用的那一档 —— 把关节层的自由度收进边界里，
+    错误会挡住发布。
+    """
+    result = ValidationResult()
+    strict = level == "governed"
+
+    def flag(message: str, *, node_id: str | None = None, hard: bool = False) -> None:
+        result.add(message, level="error" if (hard and strict) else "warning", node_id=node_id)
+
+    has_contract_output = False
+    for node in spec.nodes:
+        cfg = node.config
+        if node.type == NodeType.SUPERVISOR:
+            # 全动态规划属于探索层。受管模板里出现它，等于把 XAgent 搬上了生产线。
+            flag(
+                "受管模板不允许 supervisor 节点（全动态规划属于探索层）",
+                node_id=node.id, hard=True,
+            )
+        elif node.type == NodeType.SUBGRAPH:
+            if not cfg.get("workflow_version"):
+                flag(
+                    "子图（方法卡）引用没有钉版本，口径会随上游最新版漂移",
+                    node_id=node.id, hard=True,
+                )
+        elif node.type == NodeType.AGENT:
+            tools = cfg.get("tools") or []
+            if cfg.get("approval") == "never" and tools:
+                flag(
+                    "agent 关闭了全部工具审批（approval=never），受管模板要求危险工具至少人工确认",
+                    node_id=node.id, hard=True,
+                )
+            if not tools:
+                flag("agent 没有配任何工具，通常意味着它该是个 llm 节点", node_id=node.id)
+            elif len(tools) > 10:
+                flag(
+                    f"工具白名单有 {len(tools)} 个，边界过宽（白名单衰减的前兆）",
+                    node_id=node.id,
+                )
+        elif node.type == NodeType.OUTPUT:
+            if cfg.get("contract"):
+                has_contract_output = True
+        elif node.type == NodeType.CODE:
+            if cfg.get("network"):
+                flag("代码节点开了网络，受管模板建议把取数收敛到工具节点（有快照）", node_id=node.id)
+
+    if strict and not has_contract_output:
+        result.add(
+            "受管模板的 output 节点必须声明出具契约（contract），否则三档出具无从谈起",
+            level="error",
+        )
+    return result
+
+
+async def unresolved_caliber_upgrades(
+    session: AsyncSession, spec: GraphSpec
+) -> tuple[list[str], list[dict[str, Any]]]:
+    """检查钉了版本的子图（方法卡）是否有未处置的新版本。
+
+    返回 (阻断性错误, 需记录的升版事件)。有新版本但没声明 upgrade_policy
+    的直接阻断 formal 启动 —— 这是"被迫处置"，不是提醒。
+    """
+    from app.db.models import WorkflowVersion
+
+    errors: list[str] = []
+    events: list[dict[str, Any]] = []
+    for node in spec.nodes:
+        if node.type != NodeType.SUBGRAPH:
+            continue
+        pinned = node.config.get("workflow_version")
+        wf_id = node.config.get("workflow_id")
+        if not pinned or not wf_id:
+            continue
+        latest = (
+            await session.execute(
+                select(func.max(WorkflowVersion.version)).where(
+                    WorkflowVersion.workflow_id == wf_id
+                )
+            )
+        ).scalar()
+        if latest is None or latest <= int(pinned):
+            continue
+        policy = node.config.get("upgrade_policy")
+        if policy not in UPGRADE_POLICIES:
+            errors.append(
+                f"节点 {node.title}（{node.id}）钉在方法卡 v{pinned}，但已有 v{latest}。"
+                f"formal 运行前必须声明 upgrade_policy："
+                + " / ".join(f"{k}（{v}）" for k, v in UPGRADE_POLICY_LABELS.items())
+            )
+        else:
+            events.append(
+                {
+                    "node_id": node.id,
+                    "workflow_id": wf_id,
+                    "pinned": int(pinned),
+                    "latest": int(latest),
+                    "policy": policy,
+                    "policy_label": UPGRADE_POLICY_LABELS[policy],
+                }
+            )
+    return errors, events

backend/app/engine/nodes/multi.py

@@ -225,16 +225,42 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     if not workflow_id:
         raise NodeError(ctx.node.id, "子图节点没有选择工作流")
 
+    # 版本钉死：这就是"方法卡"的机制核心。钉了 workflow_version 就永远
+    # 执行那个不可变快照，上游改了方法卡也不会让这里的口径悄悄漂移；
+    # 没钉则跟随最新（画布试跑方便，但治理 lint 会在受管模板里拦下它）。
+    pinned = ctx.cfg("workflow_version")
     async with SessionLocal() as session:
         workflow = await session.get(Workflow, workflow_id)
-    if not workflow:
-        raise NodeError(ctx.node.id, f"找不到工作流 {workflow_id}")
+        if not workflow:
+            raise NodeError(ctx.node.id, f"找不到工作流 {workflow_id}")
+        if pinned:
+            from sqlalchemy import select
+
+            from app.db.models import WorkflowVersion
+
+            snapshot = (
+                await session.execute(
+                    select(WorkflowVersion).where(
+                        WorkflowVersion.workflow_id == workflow_id,
+                        WorkflowVersion.version == int(pinned),
+                    )
+                )
+            ).scalar_one_or_none()
+            if not snapshot:
+                raise NodeError(
+                    ctx.node.id, f"工作流「{workflow.name}」没有 v{pinned} 这个版本"
+                )
+            sub_graph = snapshot.graph
+            used_version = int(pinned)
+        else:
+            sub_graph = workflow.graph
+            used_version = workflow.version
 
     sub_input = ctx.render(ctx.cfg("input", {}) or {}, state)
     if not isinstance(sub_input, dict):
         sub_input = {"input": sub_input}
 
-    sub_spec = GraphSpec.model_validate(workflow.graph)
+    sub_spec = GraphSpec.model_validate(sub_graph)
     sub_run = type(ctx.run)(
         run_id=ctx.run.run_id,
         thread_id=f"{ctx.run.thread_id}:{ctx.node.id}",
@@ -270,6 +296,9 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     result = {
         "output": output,
         "workflow": workflow.name,
+        # 溯源信息：这次到底跑的是哪个版本、是不是钉死的
+        "workflow_version": used_version,
+        "pinned": bool(pinned),
         "text": json.dumps(output, ensure_ascii=False) if output else "",
     }
     ctx.emit(EventType.LOG, level="info", message=f"子图「{workflow.name}」完成")

backend/app/main.py

@@ -8,7 +8,7 @@ from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
-from app.api import copilot, knowledge, runs, sandbox, settings as settings_api, tools, workflows
+from app.api import artifacts, copilot, knowledge, runs, sandbox, settings as settings_api, tools, workflows
 from app.core.config import settings
 from app.db.base import init_db
 from app.engine.runner import run_manager
@@ -67,6 +67,7 @@ for router in (
     knowledge.skills_router,
     sandbox.router,
     copilot.router,
+    artifacts.router,
 ):
     app.include_router(router)