mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 16:41:59 +00:00
ThearchyHelios
815 lines
35 KiB
Python
815 lines
35 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Any
|
||
from urllib.parse import unquote
|
||
|
||
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.api.coded import CodedHTTPException
|
||
from app.core.bus import bus
|
||
from app.core import artifact_store
|
||
from app.data.catalog_impact import catalog_drift
|
||
from app.db.base import SessionLocal, get_session
|
||
from app.db.models import Approval, Artifact, Run, RunEvent, Workflow, WorkflowVersion
|
||
from app.engine.governance import governance_note, unresolved_caliber_upgrades
|
||
from app.engine.runner import run_manager
|
||
from app.engine.schema import GraphSpec, NodeType, type_label
|
||
|
||
router = APIRouter(prefix="/api/runs", tags=["runs"])
|
||
logger = logging.getLogger(__name__)
|
||
|
||
#: 发起时发现绑定的工具在本机不存在（见 missing_tools）
|
||
TOOL_MISSING = "run_tool_missing"
|
||
|
||
|
||
def actor_of(header: str | None) -> str | None:
|
||
"""请求头 X-Actor 里的署名（设置页填的那个），没填就是 None。
|
||
|
||
浏览器的请求头只能是 Latin-1：中文署名原样放进去，fetch 直接抛错、整个请求
|
||
发不出去。所以前端按 encodeURIComponent 编码后发，这里解回来；老客户端发的
|
||
纯 ASCII 署名解码前后一样。
|
||
"""
|
||
value = unquote((header or "").strip()).strip()
|
||
return value[:100] or None
|
||
|
||
|
||
class StartRunIn(BaseModel):
|
||
workflow_id: str | None = None
|
||
graph: dict[str, Any] | None = None # 直接跑一张未保存的图（画布上的"试运行"）
|
||
input: dict[str, Any] = Field(default_factory=dict)
|
||
# 不传就取设置里的运行默认值（settings.resolve_run_scope）
|
||
memory_scope: str | None = None
|
||
collection: str | None = None
|
||
# formal：从已发布的不可变版本发起，可复现可出具；exploratory：随便玩
|
||
run_class: str = "exploratory"
|
||
version: int | None = None # formal 时可指定版本，默认用已发布版本
|
||
|
||
|
||
class RunOut(BaseModel):
|
||
id: str
|
||
workflow_id: str | None
|
||
workflow_name: str
|
||
status: str
|
||
input: dict[str, Any]
|
||
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
|
||
#: 封存到了哪一条事件。为空 = 还没封存（没跑完或停在审批）
|
||
manifest_seq: int | None = None
|
||
started_by: str | None = None
|
||
#: 失败时能定位到的节点
|
||
error_node_id: str | None = None
|
||
#: 发起时实际生效的记忆域和知识库
|
||
memory_scope: str | None = None
|
||
collection: str | None = None
|
||
# 时间一律带时区（db.base.UTCDateTime 读出来就是 UTC aware）
|
||
created_at: Any = None
|
||
started_at: Any = None
|
||
finished_at: Any = None
|
||
|
||
model_config = {"from_attributes": True}
|
||
|
||
|
||
@router.post("", response_model=RunOut, status_code=201)
|
||
async def start_run(
|
||
payload: StartRunIn,
|
||
session: AsyncSession = Depends(get_session),
|
||
x_actor: str | None = Header(default=None),
|
||
) -> Run:
|
||
actor = actor_of(x_actor)
|
||
# 没挂在工作流上（画布没保存就跑、问数据当场搭的）。老数据里是「临时图」，前端两种都认
|
||
name = "未保存的工作流"
|
||
version: int | None = None
|
||
version_hash: str | None = None
|
||
governed = False
|
||
|
||
if payload.run_class == "formal":
|
||
# 正式运行的全部语义就这一条规则：必须引用一个不可变的已发布版本。
|
||
# 传裸 graph、或工作流还没发布过，都进不了 formal。
|
||
if not payload.workflow_id:
|
||
raise HTTPException(400, "正式运行需要从已保存的工作流发起：请求中没有指定工作流")
|
||
if payload.graph is not None:
|
||
raise HTTPException(400, "正式运行只能使用已发布的版本，画布上未发布的修改不会生效。请先保存并发布")
|
||
workflow = await session.get(Workflow, payload.workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, "工作流不存在，可能已被删除")
|
||
# 只看 published_version：status 反映的是当前画布（改过图就退回 draft），
|
||
# 而 formal 跑的是已发布的那一版，两者本来就可以不一致
|
||
if not workflow.published_version:
|
||
raise HTTPException(409, f"「{workflow.name}」还没有发布版本，请先在编排页发布")
|
||
version = payload.version or workflow.published_version
|
||
# "不可变"和"过过闸"是两件事。快照确实不可变，但 PATCH 保存出来的
|
||
# 草稿版本同样有快照——只认 version 存不存在的话，工作流只要发布过
|
||
# 任意一个版本，之后所有草稿都能以 formal 身份启动，发布期的
|
||
# validate_graph + lint_for_publish 全部绕过。
|
||
if version != workflow.published_version:
|
||
raise HTTPException(
|
||
409,
|
||
f"v{version} 不是当前发布版本（v{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
|
||
# 按这一版发布时记下的级别；记级别之前发布的老版本没有，照旧看工作流现在的 status
|
||
governed = (snapshot.level == "governed") if snapshot.level else workflow.status == "governed"
|
||
|
||
# 口径升版是被迫处置的事件：钉住的方法卡有新版本而未声明策略，拒绝启动
|
||
spec = GraphSpec.model_validate(graph)
|
||
errors, upgrade_events = await unresolved_caliber_upgrades(session, spec)
|
||
if errors:
|
||
raise HTTPException(409, "；".join(errors))
|
||
# 数据目录和口径卡不同：目录是给人和助手看的说明，不是运行的输入。发布之后变了只提醒、不拦——
|
||
# 拦下来等于让一次文字修订卡住正式出具
|
||
catalog_changes = await catalog_drift(session, snapshot.catalog_versions)
|
||
else:
|
||
graph = payload.graph
|
||
upgrade_events = []
|
||
catalog_changes = []
|
||
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, "没有要运行的内容：请求中既没有工作流 ID，也没有工作流内容")
|
||
await _refuse_missing_tools(session, graph)
|
||
|
||
from app.api.settings import resolve_run_scope
|
||
|
||
memory_scope, collection = await resolve_run_scope(
|
||
session, payload.memory_scope, payload.collection
|
||
)
|
||
try:
|
||
run = await run_manager.start(
|
||
graph=graph,
|
||
input_payload=payload.input,
|
||
workflow_id=payload.workflow_id,
|
||
workflow_name=name,
|
||
memory_scope=memory_scope,
|
||
collection=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
|
||
|
||
if payload.run_class == "formal":
|
||
# 按不按受管出具在发起这一刻定下来、记进封存范围：status 反映的是当前画布，中途改图、
|
||
# 审批停下再恢复时再去读它，报告节点和出口复核可能各读到不同的值
|
||
await run_manager.note(run.id, "log", **governance_note(
|
||
governed=governed, workflow_name=name, version=version))
|
||
if catalog_changes:
|
||
# 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表
|
||
await run_manager.note(run.id, "catalog.drift", version=version, count=len(catalog_changes),
|
||
tables=catalog_changes)
|
||
# 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的"
|
||
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
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 发起前检查绑定的工具
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def node_setting(spec: GraphSpec, node: Any, key: str) -> Any:
|
||
"""节点配置的实际取值，和运行时 NodeContext.cfg 同一个规则：节点上没写（None、空串）就取图级 defaults。"""
|
||
value = node.config.get(key)
|
||
return spec.defaults.get(key) if value in (None, "") else value
|
||
|
||
|
||
def tool_bindings(spec: GraphSpec) -> list[tuple[str, str]]:
|
||
"""图里每一处绑定的工具：(谁绑的, 工具名)。调用工具节点的 tool、Agent 的 tools、协作成员的 tools。
|
||
|
||
取值和运行时一致（node_setting）：Agent 节点上没写 tools、协作节点上没写成员，运行时用的是
|
||
图级 defaults 里的，这里也算它们——不然发起前检查放过了缺的工具，发起时固定数据版本也漏了源。
|
||
|
||
模板拼出来的名字（{{ vars.tool }}）也在内：发起前检查（_bindings）跳过它们，发起时固定数据
|
||
版本（engine/runner.py）要知道有没有这种名字——有的话说不准会用到哪个数据源。
|
||
"""
|
||
out: list[tuple[str, str]] = []
|
||
for node in spec.nodes:
|
||
who = f"「{node.title}」（{type_label(node.type)}）"
|
||
if node.type == NodeType.TOOL:
|
||
names = [node_setting(spec, node, "tool")]
|
||
out.extend((who, n) for n in names if isinstance(n, str) and n.strip())
|
||
elif node.type == NodeType.AGENT:
|
||
out.extend((who, n) for n in node_setting(spec, node, "tools") or []
|
||
if isinstance(n, str) and n.strip())
|
||
elif node.type == NodeType.SUPERVISOR:
|
||
for member in node_setting(spec, node, "agents") or []:
|
||
if not isinstance(member, dict):
|
||
continue
|
||
label = f"「{node.title}」的成员「{member.get('name') or '?'}」"
|
||
out.extend((label, n) for n in member.get("tools") or [] if isinstance(n, str) and n.strip())
|
||
return [(who, name.strip()) for who, name in out]
|
||
|
||
|
||
def _bindings(spec: GraphSpec) -> list[tuple[str, str]]:
|
||
"""发起前要检查的绑定。模板拼出来的工具名（{{ vars.tool }}）发起时还不知道是谁，交给运行时。"""
|
||
return [(who, name) for who, name in tool_bindings(spec) if "{{" not in name]
|
||
|
||
|
||
async def missing_tools(session: AsyncSession, spec: GraphSpec) -> list[tuple[str, str, str]]:
|
||
"""本机不存在的绑定：(谁绑的, 工具名, 去哪接入：data / tools)。
|
||
|
||
解析和运行时 build_tools 一致：内置（静态注册）、mcp:<服务>/<工具>、db_query__ / db_schema__
|
||
<数据源>、其余按自定义工具。数据源和自定义工具只认启用着的——停用的运行时同样拿不到。
|
||
MCP 只看服务登记着、启用着没有：连不连得上、工具还在不在，要连一次才知道，发起时不去等它，
|
||
上次没连上或者工具清单里没有的只在日志里提一句。
|
||
"""
|
||
from app.db.models import CustomTool, DataSource, McpServer
|
||
from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX
|
||
from app.tools.registry import all_specs
|
||
|
||
bindings = _bindings(spec)
|
||
if not bindings:
|
||
return []
|
||
builtin = all_specs()
|
||
sources = {n for (n,) in await session.execute(select(DataSource.name).where(DataSource.enabled.is_(True)))}
|
||
custom = {n for (n,) in await session.execute(select(CustomTool.name).where(CustomTool.enabled.is_(True)))}
|
||
servers = {r.name: r for r in (await session.execute(select(McpServer))).scalars()}
|
||
missing: list[tuple[str, str, str]] = []
|
||
for who, name in bindings:
|
||
if name.startswith("mcp:"):
|
||
server, _, tool = name[4:].partition("/")
|
||
row = servers.get(server)
|
||
if row is None or not row.enabled:
|
||
missing.append((who, name, "tools"))
|
||
elif row.status == "error":
|
||
logger.warning("%s 绑的 %s：MCP 服务 %s 上次没连上，运行时可能拿不到这个工具", who, name, server)
|
||
elif tool not in ("", "*") and row.tools_cache and tool not in row.tools_cache:
|
||
logger.warning("%s 绑的 %s：MCP 服务 %s 上次交回的工具清单里没有 %s", who, name, server, tool)
|
||
elif name in builtin:
|
||
continue
|
||
elif name.startswith((QUERY_PREFIX, SCHEMA_PREFIX)):
|
||
source = name[len(QUERY_PREFIX):] if name.startswith(QUERY_PREFIX) else name[len(SCHEMA_PREFIX):]
|
||
if source not in sources:
|
||
missing.append((who, name, "data"))
|
||
elif name not in custom:
|
||
missing.append((who, name, "tools"))
|
||
return missing
|
||
|
||
|
||
async def _refuse_missing_tools(session: AsyncSession, graph: dict[str, Any]) -> None:
|
||
"""绑定的工具在本机不存在就不发起：以前要跑到那一步才失败，前面的步骤白跑、钱白花。"""
|
||
try:
|
||
spec = GraphSpec.model_validate(graph)
|
||
except Exception: # noqa: BLE001 - 图本身不成形交给 run_manager.start 报
|
||
return
|
||
missing = await missing_tools(session, spec)
|
||
if not missing:
|
||
return
|
||
lines = "；".join(f"{who}绑定的 {name}" for who, name, _ in missing[:8])
|
||
more = f"等 {len(missing)} 处" if len(missing) > 8 else ""
|
||
places = {where for _, _, where in missing}
|
||
if places == {"data"}:
|
||
fix = "请到「数据」页接入，或在节点中重新选择"
|
||
elif places == {"tools"}:
|
||
fix = "请到「工具」页接入，或在节点中重新选择"
|
||
else:
|
||
fix = "数据源工具请到「数据」页接入，其他工具请到「工具」页接入，或在节点中重新选择"
|
||
raise CodedHTTPException(422, f"绑定的工具不存在：{lines}{more}。{fix}", TOOL_MISSING)
|
||
|
||
|
||
#: workflow_id 取这个值时只要没挂在任何工作流上的运行：画布上没保存就跑的临时图，
|
||
#: 以及工作流被删掉之后留下来的运行（外键是 SET NULL）
|
||
UNSAVED = "__none__"
|
||
|
||
|
||
@router.get("", response_model=list[RunOut])
|
||
async def list_runs(
|
||
workflow_id: str | None = Query(default=None, description=f"{UNSAVED} 表示没保存的工作流"),
|
||
status: str | None = Query(default=None, description="逗号分隔，如 failed,cancelled"),
|
||
run_class: str | None = None,
|
||
q: str | None = Query(default=None, description="按工作流名模糊匹配"),
|
||
limit: int = Query(default=50, ge=1, le=200),
|
||
before: datetime | None = Query(default=None, description="翻页游标：只要 created_at 早于它的"),
|
||
session: AsyncSession = Depends(get_session),
|
||
) -> list[Run]:
|
||
"""运行列表，新的在前。
|
||
|
||
以前只有 limit：待审批的运行排在第 145 位，列表只取 100 条，点了徽标也找不到它。
|
||
筛选都走服务端，翻页用 created_at 做游标（before 传上一页最后一条的 created_at）。
|
||
"""
|
||
stmt = select(Run).order_by(Run.created_at.desc()).limit(limit)
|
||
if workflow_id == UNSAVED:
|
||
stmt = stmt.where(Run.workflow_id.is_(None))
|
||
elif workflow_id:
|
||
stmt = stmt.where(Run.workflow_id == workflow_id)
|
||
if statuses := _csv(status):
|
||
stmt = stmt.where(Run.status.in_(statuses))
|
||
if classes := _csv(run_class):
|
||
stmt = stmt.where(Run.run_class.in_(classes))
|
||
if q and q.strip():
|
||
stmt = stmt.where(Run.workflow_name.ilike(f"%{_like(q.strip())}%", escape="\\"))
|
||
if before is not None:
|
||
stmt = stmt.where(Run.created_at < before)
|
||
return list((await session.execute(stmt)).scalars())
|
||
|
||
|
||
def _csv(value: str | None) -> list[str]:
|
||
return [v.strip() for v in (value or "").split(",") if v.strip()]
|
||
|
||
|
||
def _like(text: str) -> str:
|
||
"""用户输入里的 % 和 _ 按字面匹配，不当通配符。"""
|
||
return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||
|
||
|
||
@router.get("/{run_id}", response_model=RunOut)
|
||
async def get_run(run_id: str, session: AsyncSession = Depends(get_session)) -> RunOut:
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise HTTPException(404, "运行记录不存在")
|
||
out = RunOut.model_validate(run)
|
||
if out.status == "failed" and not out.error_node_id:
|
||
# 加这一列之前失败的老运行：从事件里找最后一个失败的节点
|
||
out.error_node_id = (await session.execute(
|
||
select(RunEvent.node_id)
|
||
.where(RunEvent.run_id == run_id, RunEvent.type == "node.failed")
|
||
.order_by(RunEvent.seq.desc()).limit(1)
|
||
)).scalar_one_or_none()
|
||
return out
|
||
|
||
|
||
@router.get("/{run_id}/graph")
|
||
async def get_run_graph(run_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||
"""运行时的那张图（快照），不是工作流现在的样子。回放历史运行时画布照它画。"""
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise HTTPException(404, "运行记录不存在")
|
||
return {"graph": run.graph or {}, "workflow_id": run.workflow_id, "version": run.version}
|
||
|
||
|
||
@router.get("/{run_id}/events")
|
||
async def get_events(
|
||
run_id: str,
|
||
after: int = 0,
|
||
session: AsyncSession = Depends(get_session),
|
||
) -> list[dict[str, Any]]:
|
||
"""拉历史事件。刷新页面后重建时间线用这个，再接 WebSocket 续上实时流。"""
|
||
rows = await session.execute(
|
||
select(RunEvent)
|
||
.where(RunEvent.run_id == run_id, RunEvent.seq > after)
|
||
.order_by(RunEvent.seq)
|
||
)
|
||
return [
|
||
{
|
||
"id": e.id,
|
||
"run_id": e.run_id,
|
||
"seq": e.seq,
|
||
"type": e.type,
|
||
"node_id": e.node_id,
|
||
"ts": e.ts,
|
||
"data": e.data,
|
||
}
|
||
for e in rows.scalars()
|
||
]
|
||
|
||
|
||
@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, x_actor: str | None = Header(default=None)) -> dict[str, Any]:
|
||
"""停止一次运行。status 是 stopping（正在执行的，取消已发出，稍后收到 run.cancelled）
|
||
或 cancelled（停在审批上、或服务重启挂起的，已经收成终态，待审批一并关闭）。"""
|
||
outcome = await run_manager.cancel(run_id, actor=actor_of(x_actor))
|
||
if outcome is None:
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if run is None:
|
||
raise HTTPException(404, "运行记录不存在")
|
||
raise HTTPException(409, _cannot_cancel(run.status))
|
||
return {"ok": True, "status": outcome}
|
||
|
||
|
||
def _cannot_cancel(status: str) -> str:
|
||
return {
|
||
"succeeded": "这次运行已完成，没有要停止的。",
|
||
"failed": "这次运行已失败结束，没有要停止的；如需继续，请点「继续运行」。",
|
||
"cancelled": "这次运行已取消。",
|
||
}.get(status, "这次运行已不在执行中（可能刚结束，或服务重启过），无需停止。请刷新页面查看最新状态")
|
||
|
||
|
||
class ResumeIn(BaseModel):
|
||
response: Any = None
|
||
approval_id: str | None = None
|
||
|
||
|
||
def _missing(e: KeyError) -> HTTPException:
|
||
# str(KeyError) 会带上 repr 的引号，界面上就成了 "'运行记录不存在'"
|
||
return HTTPException(404, str(e.args[0]) if e.args else "运行记录不存在")
|
||
|
||
|
||
@router.post("/{run_id}/resume", response_model=RunOut)
|
||
async def resume_run(
|
||
run_id: str, payload: ResumeIn, x_actor: str | None = Header(default=None)
|
||
) -> Run:
|
||
try:
|
||
# approval_id 以前定义了却从没传下去——一次运行里有多条待审批时，
|
||
# 引擎无从知道这次回复的是哪一条
|
||
return await run_manager.resume(
|
||
run_id, payload.response,
|
||
approval_id=payload.approval_id,
|
||
actor=actor_of(x_actor),
|
||
)
|
||
except KeyError as e:
|
||
raise _missing(e) from e
|
||
except ValueError as e:
|
||
raise HTTPException(409, str(e)) from e
|
||
|
||
|
||
class ContinueIn(BaseModel):
|
||
"""接着跑。graph 只能带改过配置的同一张图，结构必须一致。"""
|
||
|
||
graph: dict[str, Any] | None = None
|
||
|
||
|
||
@router.post("/{run_id}/continue", response_model=RunOut)
|
||
async def continue_run(
|
||
run_id: str, payload: ContinueIn, x_actor: str | None = Header(default=None)
|
||
) -> Run:
|
||
"""从失败的节点接着跑，前面跑过的不重来。
|
||
|
||
和 /resume 不是一回事：那条回复人工审批，这条是真的挂了（status=failed），
|
||
或者被服务重启打断、没有在等谁的 interrupted。挡住前者的原来只是 resume 里的
|
||
一行状态判定，而 checkpoint 一直都在——改个模型 id 就得把前面两分钟的取数
|
||
重跑一遍，纯属白等。
|
||
"""
|
||
try:
|
||
return await run_manager.continue_failed(
|
||
run_id, graph=payload.graph, actor=actor_of(x_actor),
|
||
)
|
||
except KeyError as e:
|
||
raise _missing(e) from e
|
||
except ValueError as e:
|
||
raise HTTPException(409, str(e)) from e
|
||
|
||
|
||
@router.get("/{run_id}/state")
|
||
async def get_state(run_id: str) -> dict[str, Any]:
|
||
"""读取 checkpoint 里的当前状态。调试中断中的 run 时很有用。"""
|
||
from app.engine.compiler import compile_graph
|
||
from app.engine.context import RunContext
|
||
from app.engine.schema import GraphSpec
|
||
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise HTTPException(404, "运行记录不存在")
|
||
|
||
spec = GraphSpec.model_validate(run.graph)
|
||
app = compile_graph(spec, RunContext(run_id=run_id, thread_id=run_id, spec=spec)).compile(
|
||
checkpointer=run_manager.checkpointer
|
||
)
|
||
snapshot = await app.aget_state({"configurable": {"thread_id": run_id}})
|
||
values = dict(snapshot.values or {})
|
||
values.pop("messages", None) # 消息对象不好直接序列化，前端也用不到
|
||
return {
|
||
"values": _jsonable(values),
|
||
"next": list(snapshot.next or ()),
|
||
"interrupts": [
|
||
{"id": getattr(i, "id", None), "value": _jsonable(getattr(i, "value", None))}
|
||
for i in (snapshot.interrupts or ())
|
||
],
|
||
}
|
||
|
||
|
||
@router.get("/{run_id}/verify")
|
||
async def verify_run(run_id: str) -> dict[str, Any]:
|
||
"""核对事件流和封存时的清单哈希是否一致：事后被改过、删过、插过都会对不上。"""
|
||
from app.engine.runner import verify_manifest
|
||
|
||
try:
|
||
return await verify_manifest(run_id)
|
||
except KeyError as e:
|
||
raise _missing(e) from e
|
||
|
||
|
||
@router.get("/{run_id}/history")
|
||
async def get_history(run_id: str, limit: int = 50) -> list[dict[str, Any]]:
|
||
"""checkpoint 历史 —— 可以看到每一步之后的状态快照，用来做时间旅行式调试。"""
|
||
from app.engine.compiler import compile_graph
|
||
from app.engine.context import RunContext
|
||
from app.engine.schema import GraphSpec
|
||
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise HTTPException(404, "运行记录不存在")
|
||
|
||
spec = GraphSpec.model_validate(run.graph)
|
||
app = compile_graph(spec, RunContext(run_id=run_id, thread_id=run_id, spec=spec)).compile(
|
||
checkpointer=run_manager.checkpointer
|
||
)
|
||
out: list[dict[str, Any]] = []
|
||
async for snapshot in app.aget_state_history({"configurable": {"thread_id": run_id}}):
|
||
values = dict(snapshot.values or {})
|
||
values.pop("messages", None)
|
||
out.append(
|
||
{
|
||
"checkpoint_id": snapshot.config.get("configurable", {}).get("checkpoint_id"),
|
||
"next": list(snapshot.next or ()),
|
||
"created_at": snapshot.created_at,
|
||
"trail": _jsonable(values.get("trail") or [])[-1:],
|
||
"nodes_done": list((values.get("nodes") or {}).keys()),
|
||
}
|
||
)
|
||
if len(out) >= limit:
|
||
break
|
||
return out
|
||
|
||
|
||
@router.delete("/{run_id}", status_code=204)
|
||
async def delete_run(
|
||
run_id: str, force: bool = False, session: AsyncSession = Depends(get_session)
|
||
) -> None:
|
||
"""删除运行记录。运行记录是出具的追溯依据，两种情况要拦：
|
||
|
||
- 还在跑的：删了会留下没人管的执行任务，它还会往一条不存在的记录里写事件；
|
||
- 封存过的正式运行：删掉之后清单再也无法核对。确实要删，带 force=true。
|
||
"""
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise HTTPException(404, "运行记录不存在")
|
||
if run.status in ("queued", "running") or run_manager.is_active(run_id):
|
||
raise HTTPException(409, "这次运行仍在进行，现在删除会留下无人管理的后台任务。请先停止运行，再删除")
|
||
if run.run_class == "formal" and run.manifest_hash and not force:
|
||
raise HTTPException(
|
||
409,
|
||
"这是一次已封存的正式运行，是出具结果的追溯凭证。删除后其事件和工件将一并删除，"
|
||
"封存清单无法再核对。如仍要删除，请选择「强制删除」",
|
||
)
|
||
await session.delete(run)
|
||
await session.commit()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 实时事件流
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
#: 不在进行中的状态：没有后续事件会来（interrupted 要等人回复，恢复会新开一段
|
||
#: 执行，客户端到时候重新接）
|
||
_SETTLED = ("succeeded", "failed", "cancelled", "interrupted")
|
||
|
||
|
||
@router.websocket("/{run_id}/stream")
|
||
async def stream_run(websocket: WebSocket, run_id: str) -> None:
|
||
"""把一次运行的事件推给前端。
|
||
|
||
先补历史再接实时，所以中途打开页面、或者刷新之后，时间线都是完整的。
|
||
运行一旦不在进行中，补完历史就发一条 stream.end（data.status 是当时的状态）
|
||
再关掉——interrupted 也发。以前停在审批或服务重启挂起的运行什么都不发，
|
||
客户端一直重连、一直以为它还在跑。
|
||
"""
|
||
await websocket.accept()
|
||
after = int(websocket.query_params.get("after", 0) or 0)
|
||
|
||
# 先占订阅位，再读历史。反过来的话，两步之间图正好跑出来的事件既不在
|
||
# 历史里、也不在实时流里——永远丢了。实测就是这样：一次跑图的开头
|
||
# 两三条 node.started/node.finished 消失，界面上表现为某个节点永远
|
||
# 在转圈、它的子步骤跑到了顶层。
|
||
subscription = bus.attach(run_id)
|
||
try:
|
||
last_seq, run = await _replay(websocket, run_id, after)
|
||
if run is not None and run.status in _SETTLED and run_manager.is_active(run_id):
|
||
# 状态已经落了、执行任务还在收尾（终态事件、封存）：等它收完，把这期间
|
||
# 落库的事件补上，按最终状态收尾。否则会在 run.finished 之前就宣布结束
|
||
await run_manager.wait_idle(run_id)
|
||
last_seq, run = await _replay(websocket, run_id, last_seq)
|
||
|
||
if run is None or run.status in _SETTLED:
|
||
await _end(websocket, run.status if run else None)
|
||
return
|
||
|
||
# 占位期间攒下的那几条历史里也有，按 seq 滤掉，否则前端同一步收两遍
|
||
pump = asyncio.create_task(_pump(websocket, subscription.stream(after=last_seq), run_id))
|
||
try:
|
||
# 客户端断开时读操作会抛异常，用它来结束这条连接
|
||
while True:
|
||
await websocket.receive_text()
|
||
except WebSocketDisconnect:
|
||
pass
|
||
finally:
|
||
pump.cancel()
|
||
with contextlib.suppress(asyncio.CancelledError):
|
||
await pump
|
||
except WebSocketDisconnect:
|
||
return
|
||
except Exception: # noqa: BLE001
|
||
with contextlib.suppress(Exception):
|
||
await websocket.close()
|
||
finally:
|
||
subscription.close()
|
||
|
||
|
||
async def _replay(websocket: WebSocket, run_id: str, after: int) -> tuple[int, Run | None]:
|
||
"""把 seq > after 的历史事件发出去，顺带取回运行当前的状态。"""
|
||
last_seq = after
|
||
async with SessionLocal() as session:
|
||
rows = await session.execute(
|
||
select(RunEvent)
|
||
.where(RunEvent.run_id == run_id, RunEvent.seq > after)
|
||
.order_by(RunEvent.seq)
|
||
)
|
||
for event in rows.scalars():
|
||
last_seq = max(last_seq, event.seq or 0)
|
||
await websocket.send_json(
|
||
{
|
||
"seq": event.seq,
|
||
"type": event.type,
|
||
"node_id": event.node_id,
|
||
"ts": event.ts,
|
||
"data": event.data,
|
||
"replay": True,
|
||
}
|
||
)
|
||
run = await session.get(Run, run_id)
|
||
return last_seq, run
|
||
|
||
|
||
async def _end(websocket: WebSocket, status: str | None) -> None:
|
||
# 顶层的 status 留给老客户端；新口径读 data.status
|
||
await websocket.send_json({"type": "stream.end", "status": status, "data": {"status": status}})
|
||
await websocket.close()
|
||
|
||
|
||
async def _pump(websocket: WebSocket, stream: Any, run_id: str) -> None:
|
||
try:
|
||
async for event in stream:
|
||
await websocket.send_json(event.to_wire())
|
||
# 实时流走完 = 运行到了终态，或者关停时挂起了：补一条结束标记再关，
|
||
# 客户端不用靠猜，也不会对着一条已经结束的运行无限重连
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
await _end(websocket, run.status if run else None)
|
||
except (WebSocketDisconnect, RuntimeError):
|
||
return
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 审批
|
||
# --------------------------------------------------------------------------
|
||
|
||
approvals_router = APIRouter(prefix="/api/approvals", tags=["approvals"])
|
||
|
||
|
||
class ApprovalOut(BaseModel):
|
||
id: str
|
||
run_id: str
|
||
node_id: str
|
||
mode: str
|
||
title: str
|
||
payload: dict[str, Any]
|
||
status: str
|
||
response: dict[str, Any]
|
||
created_at: Any = None
|
||
#: 谁批的：设置里的署名（请求头 X-Actor）。没署名就是 null，由前端写「未署名」
|
||
resolved_by: str | None = None
|
||
resolved_at: Any = None
|
||
# 下面几项来自所属的运行。全局待办列表要拼得出一行像样的字：
|
||
# 哪个工作流、卡在哪个节点、运行现在什么状态
|
||
workflow_id: str | None = None
|
||
workflow_name: str | None = None
|
||
node_label: str | None = None
|
||
run_status: str | None = None
|
||
run_class: str | None = None
|
||
|
||
|
||
@approvals_router.get("", response_model=list[ApprovalOut])
|
||
async def list_approvals(
|
||
status: str = Query(default="pending", description="pending / resolved / all，可逗号分隔"),
|
||
run_id: str | None = None,
|
||
limit: int = Query(default=100, ge=1, le=500),
|
||
session: AsyncSession = Depends(get_session),
|
||
) -> list[ApprovalOut]:
|
||
stmt = (
|
||
select(Approval, Run)
|
||
.join(Run, Run.id == Approval.run_id, isouter=True)
|
||
.order_by(Approval.created_at.desc())
|
||
.limit(limit)
|
||
)
|
||
if status != "all" and (statuses := _csv(status)):
|
||
stmt = stmt.where(Approval.status.in_(statuses))
|
||
if run_id:
|
||
stmt = stmt.where(Approval.run_id == run_id)
|
||
out: list[ApprovalOut] = []
|
||
for approval, run in (await session.execute(stmt)).all():
|
||
item = ApprovalOut.model_validate(approval, from_attributes=True)
|
||
if run is not None:
|
||
item.workflow_id = run.workflow_id
|
||
item.workflow_name = run.workflow_name
|
||
item.run_status = run.status
|
||
item.run_class = run.run_class
|
||
item.node_label = _node_label(run.graph, approval.node_id)
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
def _node_label(graph: dict[str, Any] | None, node_id: str) -> str | None:
|
||
for n in (graph or {}).get("nodes") or []:
|
||
if isinstance(n, dict) and n.get("id") == node_id:
|
||
return ((n.get("data") or {}).get("label")) or node_id
|
||
return None
|
||
|
||
|
||
class DecideIn(BaseModel):
|
||
approved: bool = True
|
||
note: str = ""
|
||
value: Any = None
|
||
args: dict[str, Any] | None = None
|
||
|
||
|
||
@approvals_router.post("/{approval_id}/decide", response_model=RunOut)
|
||
async def decide(
|
||
approval_id: str,
|
||
payload: DecideIn,
|
||
session: AsyncSession = Depends(get_session),
|
||
x_actor: str | None = Header(default=None),
|
||
) -> Run:
|
||
approval = await session.get(Approval, approval_id)
|
||
if not approval:
|
||
raise HTTPException(404, "审批记录不存在，所属的运行可能已被删除")
|
||
if approval.status != "pending":
|
||
raise HTTPException(409, "这条审批已处理过，请刷新页面查看最新状态")
|
||
|
||
# 责任归属先落库再恢复执行——恢复失败也要留下"谁试图批的"
|
||
approval.resolved_by = actor_of(x_actor)
|
||
await session.commit()
|
||
|
||
response: dict[str, Any] = {"approved": payload.approved, "note": payload.note}
|
||
if payload.value is not None:
|
||
response["value"] = payload.value
|
||
if payload.args is not None:
|
||
response["args"] = payload.args
|
||
|
||
try:
|
||
return await run_manager.resume(
|
||
approval.run_id, response,
|
||
approval_id=approval.id,
|
||
actor=approval.resolved_by,
|
||
)
|
||
except ValueError as e:
|
||
raise HTTPException(409, str(e)) from e
|
||
|
||
|
||
def _jsonable(value: Any) -> Any:
|
||
if isinstance(value, dict):
|
||
return {str(k): _jsonable(v) for k, v in value.items()}
|
||
if isinstance(value, (list, tuple)):
|
||
return [_jsonable(v) for v in value]
|
||
if isinstance(value, (str, int, float, bool)) or value is None:
|
||
return value
|
||
return str(value)
|