diff --git a/backend/app/api/artifacts.py b/backend/app/api/artifacts.py new file mode 100644 index 00000000..d9862f8a --- /dev/null +++ b/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} diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index 31bb5804..2ab39856 100644 --- a/backend/app/api/runs.py +++ b/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) diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 4f94ce93..72e0d146 100644 --- a/backend/app/api/workflows.py +++ b/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] diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index b1b2ce4f..5f6ea4c9 100644 --- a/backend/app/engine/compiler.py +++ b/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 diff --git a/backend/app/engine/governance.py b/backend/app/engine/governance.py new file mode 100644 index 00000000..d69db24b --- /dev/null +++ b/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 diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 3fbf89b1..56c999db 100644 --- a/backend/app/engine/nodes/multi.py +++ b/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}」完成") diff --git a/backend/app/main.py b/backend/app/main.py index 730268b8..251f5c19 100644 --- a/backend/app/main.py +++ b/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)