mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 09:01:58 +00:00
ThearchyHelios
1846 lines
102 KiB
Python
1846 lines
102 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import copy
|
||
import json
|
||
import logging
|
||
import time
|
||
from collections.abc import Callable, Sequence
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
from langgraph.checkpoint.base import BaseCheckpointSaver
|
||
from langgraph.errors import GraphRecursionError
|
||
from langgraph.types import Command
|
||
from sqlalchemy import case, func, select, update
|
||
|
||
from app.cluster import fence, identity
|
||
from app.cluster.fence import Fenced
|
||
from app.core.bus import bus
|
||
from app.core.config import settings
|
||
from app.core.errors import describe_exception, raw_detail
|
||
from app.core.events import EventType, RunEventModel
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Approval, Run, RunEvent, Workflow
|
||
from app.engine.approval import read_decision
|
||
from app.engine.compiler import compile_graph, initial_state
|
||
from app.engine.context import NodeError, NodeRejected, RunContext
|
||
from app.engine.deadline import segment
|
||
from app.engine.replay import PROTOCOL, PROTOCOL_KEY, protocol_of
|
||
from app.engine.schema import GraphSpec, NodeType, loop_steps, topology_of, validate_graph
|
||
from app.process.actions import ActionPending
|
||
from app.records.views import scope_entries
|
||
from app.tools.datasource import RUN_VERSIONS_KEY
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 流式增量（正文 token、思考 delta）量太大，只做实时推送不落库；
|
||
# 轨迹里正文靠 node.finished 的 preview，思考靠调用结束时的单条 llm.thinking 汇总
|
||
_EPHEMERAL = {EventType.LLM_TOKEN, EventType.LLM_THINKING_DELTA}
|
||
|
||
#: RunContext.extra 里放这次运行的选项（Run.options）的键。节点从这里读发起时定下的快照、不读设置的
|
||
#: 当前值：审批停下期间改了设置，恢复后的那一段也按发起时的，重放和正式运行才可复现
|
||
RUN_OPTIONS_KEY = "run_options"
|
||
|
||
|
||
#: 接着跑被拒的机器码：正式运行带图（F2），有结果未知、还没核实的写入，被人驳回了的
|
||
FORMAL_GRAPH_LOCKED = "formal_graph_locked"
|
||
ACTION_UNRESOLVED = "action_unresolved"
|
||
RUN_REJECTED = "run_rejected"
|
||
|
||
#: 运行为什么结束在 cancelled（Run.end_reason，docs/design/next-round/00-总方案.md X9）
|
||
END_REJECTED = "rejected"
|
||
END_STOPPED = "stopped"
|
||
#: 被驳回的运行不能继续运行：从断点重跑会读到同一个驳回答复、再停在同一处
|
||
REJECTED_CONTINUE = "被驳回的运行不能继续运行：改好内容后重新发起"
|
||
#: 被驳回的运行也没有可恢复的审批（恢复、继续运行同一个错误码 RUN_REJECTED）
|
||
REJECTED_RESUME = "这次运行已被驳回，没有待处理的审批，无法恢复：改好内容后重新发起"
|
||
#: 服务重启时还在排队的运行改成失败时写的话（run.error 和补发的 run.failed 是同一句）
|
||
QUEUED_AT_RESTART = "服务重启时这次运行仍在排队，没有可恢复的进度，请重新发起运行"
|
||
|
||
|
||
class ContinueRefused(ValueError):
|
||
"""接着跑被拒、带机器码的那几种。是 ValueError：老调用方按 ValueError 接（409）照旧成立。"""
|
||
|
||
def __init__(self, message: str, *, code: str) -> None:
|
||
super().__init__(message)
|
||
self.code = code
|
||
|
||
|
||
class RunManager:
|
||
"""负责一次运行的完整生命周期：启动、串流、中断、恢复、取消。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._tasks: dict[str, asyncio.Task[None]] = {}
|
||
self._checkpointer: BaseCheckpointSaver | None = None
|
||
self._cm: Any = None
|
||
# 并发上限以前是这里一把进程内的信号量（max_concurrent_runs）；现在是库里按类别发的名额（app/cluster/slots.py），
|
||
# 几个进程合起来算，评测、召回这些系统运行不再挤占问数据和试跑
|
||
#: 正在关停。此时收到的取消来自关停而不是用户，运行要记成可恢复的中断
|
||
self._closing = False
|
||
#: 已经跑完、正在落状态和封存的运行。关停不去打断它们
|
||
self._finalizing: set[str] = set()
|
||
#: 谁停的：取消执行任务时记下，run.cancelled 和关掉的审批上写得出人
|
||
self._stopped_by: dict[str, str | None] = {}
|
||
|
||
# ---------------- 生命周期 ----------------
|
||
|
||
async def setup(self) -> None:
|
||
self._closing = False
|
||
settings.ensure_dirs()
|
||
# SQLite：数据目录里的 checkpoints.db；PostgreSQL：同一个库（或 AGENTLAB_CHECKPOINT_URL），见 engine/checkpoints.py
|
||
from app.engine.checkpoints import open_saver
|
||
|
||
self._cm = open_saver()
|
||
self._checkpointer = _fenced_saver(await self._cm.__aenter__())
|
||
# 上次进程被强杀时留下的记录。running 和 queued 要分开处理：
|
||
# running 至少跑过一步、有 checkpoint，可以续；queued 还在排队，
|
||
# 一个 checkpoint 都没写过，"恢复"它只会从 START 用默认值重跑一遍，
|
||
# 还丢掉 run.input——那不是恢复，是伪造一份结果。
|
||
async with SessionLocal() as session:
|
||
# 先把落后于库里实际最后一条的 last_seq 改正（为什么会落后见 _seq_floor）。内存里的序号计数刚清零，
|
||
# 接下来各处都按 last_seq 接号：本模块的几处另外兜了底，按需裁判记判定（api/evidence.py 的
|
||
# _append_judged）没有，不改正的话它重启后第一次追加就撞上已有的序号。有 (run_id, seq) 索引，
|
||
# 每次运行一次索引查找。updated_at 不动：这是补记录，不是运行有了变化
|
||
latest = select(func.max(RunEvent.seq)).where(RunEvent.run_id == Run.id).scalar_subquery()
|
||
await session.execute(
|
||
update(Run).where(func.coalesce(Run.last_seq, 0) < latest)
|
||
.values(last_seq=latest, updated_at=Run.updated_at)
|
||
.execution_options(synchronize_session=False))
|
||
await session.commit()
|
||
# 上次进程被强杀时留下的记录：只碰属主不在了的（多进程时别的活着的进程在跑的一概不碰，见 reap）
|
||
from app.cluster import state
|
||
|
||
await self.reap(await state.current(fresh=True))
|
||
# 口径账本：最近几天封存了、一次都没比过的探索运行，在后台补比一次（ledger/match.py）。只读工件和版本表，不查业务库、
|
||
# 不调模型；和调度循环同一个开关，拷来的数据目录、测试不补
|
||
from app.ledger import match as ledger_match
|
||
from app.ledger import recall as ledger_recall
|
||
|
||
# 多进程时只由调度当班的进程做（app/cluster；单进程时本进程启动就当班）：几个进程各补一遍会重复比对、重复召回
|
||
from app.cluster.service import leading
|
||
|
||
if leading():
|
||
ledger_match.start_backfill()
|
||
# 口径召回（ledger/recall.py）：最近发布、自动那一批没发完的接着发，结束了还没收尾的召回运行补上收尾。同一个开关
|
||
ledger_recall.start_resume()
|
||
|
||
async def reap(self, live: Any) -> int:
|
||
"""接手属主不在了的运行（启动时；多进程时当班进程隔一会儿再来一次，app/cluster/reaper.py）。规矩同以前的启动整理：
|
||
|
||
- running 至少跑过一步、有断点：标 interrupted（可从断点继续运行），补一条 server_shutdown 日志——进程被强杀时连这一条
|
||
都来不及发，事件流停在半路，回放它的客户端才知道它是挂起了而不是还在跑；
|
||
- queued 还在排队，一个断点都没写过，「恢复」它只会从 START 用默认值重跑一遍、还丢掉 run.input，那不是恢复是伪造：
|
||
标 failed，补一条 run.failed（不写结束时间、不封存：它从没开始执行，同以前）。
|
||
|
||
不自动重跑有副作用的步骤：人看了再点「继续运行」。属主还活着的（别的进程正在跑、本进程里正在跑）一概不碰；认领那一下是
|
||
带条件的 UPDATE（状态、属主都还是刚才看到的），几个进程同时接手同一次运行只有一个改得动。交回整理了几次运行。"""
|
||
from app.cluster.claims import same
|
||
|
||
async with SessionLocal() as session:
|
||
rows = (await session.execute(select(Run.id, Run.owner, Run.status).where(
|
||
Run.status.in_(("running", "queued"))))).all()
|
||
done = 0
|
||
for run_id, owner, status in rows:
|
||
if live.alive(owner, local=self.is_active(run_id)):
|
||
continue
|
||
event: RunEventModel | None = None
|
||
async with SessionLocal() as session:
|
||
values: dict[str, Any] = (
|
||
{"status": "interrupted", "error": "服务重启，运行已挂起，可从断点继续运行"} if status == "running"
|
||
else {"status": "failed", "error": QUEUED_AT_RESTART})
|
||
claimed = await session.execute(update(Run).where(Run.id == run_id, Run.status == status,
|
||
same(Run.owner, owner))
|
||
.values(owner=None, **values).execution_options(synchronize_session=False))
|
||
if claimed.rowcount != 1:
|
||
await session.rollback()
|
||
continue
|
||
run = await session.get(Run, run_id)
|
||
bus.set_seq(run_id, await _seq_floor(session, run_id, run.last_seq if run else 0))
|
||
event = self._event(run_id, EventType.LOG, data={
|
||
"level": "warn", "code": "server_shutdown",
|
||
"message": "服务上次未正常关闭，这次运行已中断；可从断点继续运行",
|
||
}) if status == "running" else self._event(run_id, EventType.RUN_FAILED, data={"error": QUEUED_AT_RESTART})
|
||
await _stage(session, event, run)
|
||
await session.commit()
|
||
await bus.publish(event)
|
||
await bus.close(run_id)
|
||
done += 1
|
||
if done:
|
||
logger.warning("接手了 %d 次属主已经不在的运行（在跑的标中断、排队的标失败）", done)
|
||
return done
|
||
|
||
async def shutdown(self) -> None:
|
||
self._closing = True
|
||
# 后台的口径比对、召回先停：没做完的不记，下次启动补比、接着召回会再做
|
||
from app.ledger import match as ledger_match
|
||
from app.ledger import recall as ledger_recall
|
||
|
||
await ledger_match.shutdown()
|
||
await ledger_recall.shutdown()
|
||
# 只等还没结束的：结束了的没什么可等，而且它可能属于另一个事件循环，
|
||
# gather 会直接抛 "future belongs to a different loop"
|
||
pending = [t for t in self._tasks.values() if not t.done()]
|
||
for run_id, task in list(self._tasks.items()):
|
||
# 正在收尾的不打断：状态和封存写到一半，比哪一步都没写更难收拾
|
||
if not task.done() and run_id not in self._finalizing:
|
||
task.cancel()
|
||
await asyncio.gather(*pending, return_exceptions=True)
|
||
self._tasks.clear()
|
||
if self._cm is not None:
|
||
with contextlib.suppress(Exception):
|
||
await self._cm.__aexit__(None, None, None)
|
||
|
||
@property
|
||
def checkpointer(self) -> BaseCheckpointSaver:
|
||
if self._checkpointer is None:
|
||
raise RuntimeError("RunManager 还没初始化")
|
||
return self._checkpointer
|
||
|
||
# ---------------- 事件 ----------------
|
||
|
||
async def _emit(
|
||
self,
|
||
run_id: str,
|
||
event_type: EventType | str,
|
||
*,
|
||
node_id: str | None = None,
|
||
data: dict[str, Any] | None = None,
|
||
persist: bool = True,
|
||
ts: float | None = None,
|
||
) -> None:
|
||
# 执行这次运行的那段（cluster/fence 绑定过世代的）写之前问一句：租约续晚了就不写（可能已经被别的进程接手）
|
||
fence.check()
|
||
event = self._event(run_id, event_type, node_id=node_id, data=data, ts=ts)
|
||
# 先落库，再广播。顺序反过来会漏事件，而且是结构性的漏：
|
||
#
|
||
# 订阅者的完整性靠"读历史 + 接实时"两段拼起来。它先订阅、再读历史，
|
||
# 于是需要这条保证——**提交早于我这次 SELECT 的，在历史里；提交晚于
|
||
# 我这次 SELECT 的，广播也晚于我订阅，在实时流里**。这条保证只有在
|
||
# "提交先于广播"时才成立。
|
||
#
|
||
# 先广播的话，一条在订阅建立之前广播、却在读历史之后才提交的事件，
|
||
# 两边都拿不到。实测跑一次 25ms 的图，六条事件里的 run.started 有
|
||
# 一半几率就这么没了——界面表现为某个节点永远在转圈。
|
||
if persist and event.type not in _EPHEMERAL:
|
||
async with SessionLocal() as session:
|
||
await _stage(session, event)
|
||
await session.commit()
|
||
await bus.publish(event)
|
||
|
||
@staticmethod
|
||
def _event(
|
||
run_id: str,
|
||
event_type: EventType | str,
|
||
*,
|
||
node_id: str | None = None,
|
||
data: dict[str, Any] | None = None,
|
||
ts: float | None = None,
|
||
) -> RunEventModel:
|
||
return RunEventModel(
|
||
run_id=run_id,
|
||
seq=bus.next_seq(run_id),
|
||
type=EventType(str(event_type)),
|
||
node_id=node_id,
|
||
data=data or {},
|
||
# 节点里发出的事件带着它发生的时刻。以前一律取转发时刻：并行的成员
|
||
# 先做完的那条也要排到整轮结束才被转发，时间线上的耗时就全错了
|
||
**({"ts": ts} if ts else {}),
|
||
)
|
||
|
||
async def note(
|
||
self, run_id: str, event_type: EventType | str, *, node_id: str | None = None,
|
||
**data: Any,
|
||
) -> None:
|
||
"""给运行之外的代码往运行里记一条事件。发起时就定下的事件不走这里，交给 start 的 notes。
|
||
|
||
执行中的运行，序号由执行它的进程发（core/bus 的进程内计数）：运行在别的活着的进程里执行时交给它记（cluster/requests），
|
||
不在这里取号——这里的计数和那边的不是同一个，直接写会撞号。"""
|
||
if not self.is_active(run_id):
|
||
from app.cluster import requests, workers
|
||
|
||
async with SessionLocal() as session:
|
||
row = (await session.execute(select(Run.status, Run.owner).where(Run.id == run_id))).first()
|
||
if row is not None and row.status in ("running", "queued") and row.owner \
|
||
and row.owner != identity.owner() and await workers.alive(row.owner):
|
||
await requests.send(row.owner, "note", "run", run_id, payload={
|
||
"type": str(event_type), "node_id": node_id, "data": data})
|
||
return
|
||
await self._emit(run_id, event_type, node_id=node_id, data=data)
|
||
|
||
# ---------------- 启动 ----------------
|
||
|
||
async def start(
|
||
self,
|
||
*,
|
||
graph: dict[str, Any],
|
||
input_payload: dict[str, Any],
|
||
workflow_id: str | None = None,
|
||
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,
|
||
approval_default: str | None = None,
|
||
options: dict[str, Any] | None = None,
|
||
schedule_id: str | None = None,
|
||
notes: Sequence[tuple[EventType | str, str | None, dict[str, Any]]] = (),
|
||
origin_kind: str | None = None,
|
||
origin: dict[str, Any] | None = None,
|
||
app_id: str | None = None,
|
||
app_version: int | None = None,
|
||
) -> Run:
|
||
"""建运行记录、交给执行任务。options 是这次运行的选项快照，schedule_id 是发起它的定时任务（可空）。
|
||
origin_kind / origin / app_id / app_version 是运行来源（engine/origin.py），由 launch_run 定下，原样记在运行上；
|
||
对应的 run.origin 事件由 launch_run 放进 notes。
|
||
|
||
notes 是发起时就定下的事件 (类型, node_id, 数据)，比如正式运行按什么级别出具、口径升版怎么处置。它们和
|
||
运行记录同一个事务提交，排在 run.started 前面：写不进去就不建运行，不会出现运行已经在执行、调用方却
|
||
只拿到一个异常（定时任务会把它记成未能发起）；也不会和执行抢先后，运行结束得快时落到封存之后。
|
||
"""
|
||
spec = GraphSpec.model_validate(graph)
|
||
report = validate_graph(spec)
|
||
if not report.ok:
|
||
raise ValueError(
|
||
"工作流未通过校验，请先在画布上修正标红的节点再运行："
|
||
+ "；".join(i.message for i in report.issues if i.level == "error")
|
||
)
|
||
if approval_default is None:
|
||
approval_default = await _approval_default(run_class=run_class)
|
||
# 拷一份：调用方手上的 dict 之后再改，不该改到这次运行的快照
|
||
options = copy.deepcopy(options) if options else {}
|
||
|
||
async with SessionLocal() as session:
|
||
from app.data.table_versions import pin_lock
|
||
from app.tools.trust import run_snapshot
|
||
|
||
tool_trust = await run_snapshot(session, run_class=run_class)
|
||
agent_limits = await _agent_limits(session)
|
||
# 读上传表格的当前指针到运行记录提交，持着版本存储的锁：中间有人重传的话，发布后的回收
|
||
# 看不到这条还没提交的运行，会把刚固定的快照删掉（table_versions.pin_lock）
|
||
async with pin_lock():
|
||
data_versions = await _data_versions(session, spec)
|
||
# 记录表这次运行看哪些行：正式运行看全部，别的按发起人的授权（records/views.scope_entries）
|
||
await scope_entries(session, data_versions, run_class=run_class, started_by=started_by)
|
||
run = Run(
|
||
workflow_id=workflow_id,
|
||
workflow_name=workflow_name,
|
||
status="queued",
|
||
graph=graph,
|
||
input=input_payload,
|
||
run_class=run_class,
|
||
version=version,
|
||
version_hash=version_hash,
|
||
started_by=started_by,
|
||
memory_scope=memory_scope,
|
||
collection=collection,
|
||
approval_default=approval_default,
|
||
tool_trust=tool_trust,
|
||
agent_limits=agent_limits,
|
||
data_versions=data_versions,
|
||
options=options,
|
||
schedule_id=schedule_id,
|
||
origin_kind=origin_kind,
|
||
origin=origin,
|
||
app_id=app_id,
|
||
app_version=app_version,
|
||
)
|
||
run.thread_id = run.id # 一个 run 一条 checkpoint 线程
|
||
session.add(run)
|
||
staged: list[RunEventModel] = []
|
||
if notes:
|
||
await session.flush() # 事件要带上运行的 id
|
||
staged = [self._event(run.id, kind, node_id=node_id, data=data)
|
||
for kind, node_id, data in notes]
|
||
for event in staged:
|
||
await _stage(session, event, run)
|
||
await session.commit()
|
||
await session.refresh(run)
|
||
# 先落库再广播，理由同 _emit；执行任务还没建，run.started 必定排在它们后面
|
||
for event in staged:
|
||
await bus.publish(event)
|
||
|
||
self._tasks[run.id] = asyncio.create_task(
|
||
self._drive(
|
||
run.id,
|
||
spec,
|
||
initial_state(input_payload),
|
||
workflow_id=workflow_id,
|
||
memory_scope=memory_scope,
|
||
collection=collection,
|
||
approval_default=approval_default,
|
||
tool_trust=tool_trust,
|
||
agent_limits=agent_limits,
|
||
data_versions=data_versions,
|
||
options=options,
|
||
origin_kind=origin_kind,
|
||
)
|
||
)
|
||
return run
|
||
|
||
async def resume(
|
||
self, run_id: str, response: Any, *, approval_id: str | None = None,
|
||
actor: str | None = None,
|
||
) -> Run:
|
||
"""从人工介入的断点继续。
|
||
|
||
两件事以前是错的：
|
||
|
||
一是不看有没有 checkpoint 就发 Command。重启扫描会把 queued（从没跑过、
|
||
没有任何 checkpoint）也标成 interrupted，对这种 run 发 Command，LangGraph
|
||
会拿一个空 state 从 START 重跑整张图——run.input 完全没被注入，最后还落成
|
||
succeeded。跑出来的是一份用默认值算的假结果。
|
||
|
||
二是同一 superstep 里有多个 interrupt 时，只发标量 Command 会让 LangGraph
|
||
直接抛 RuntimeError 打爆整个 run，而所有 pending 审批还都被写成同一个回复。
|
||
现在按 interrupt_id 逐条收集，凑齐了才继续，没凑齐就继续等下一个人回复。
|
||
"""
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise KeyError("运行记录不存在")
|
||
if run.status not in ("interrupted",):
|
||
raise await _resume_refused(session, run)
|
||
spec = GraphSpec.model_validate(run.graph)
|
||
workflow_id = run.workflow_id
|
||
carried = _carried(run)
|
||
|
||
# 先问引擎：这条线程到底停在哪、还等着哪些 interrupt
|
||
run_ctx = RunContext(run_id=run_id, thread_id=run_id, spec=spec, workflow_id=workflow_id)
|
||
app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer)
|
||
config = {"configurable": {"thread_id": run_id}}
|
||
snapshot = await app.aget_state(config)
|
||
waiting = list(getattr(snapshot, "interrupts", None) or [])
|
||
|
||
if not waiting and not getattr(snapshot, "next", None):
|
||
# 没有断点可续。多半是重启时被标成 interrupted 的 queued 运行。
|
||
raise ValueError(
|
||
"这次运行没有可恢复的断点，可能在服务重启前尚未开始执行。请重新发起运行。"
|
||
)
|
||
if stale := await _stale_replay(run_id, spec, snapshot):
|
||
raise ValueError(
|
||
f"这次运行停在「{stale}」时系统已升级，断点格式不再兼容，继续会让「{stale}」中"
|
||
f"已执行的工具重复执行。请放弃这次运行并重新发起"
|
||
)
|
||
|
||
answers: dict[str, Any] = {}
|
||
async with SessionLocal() as session:
|
||
pending = list((
|
||
await session.execute(
|
||
select(Approval).where(
|
||
Approval.run_id == run_id, Approval.status == "pending"
|
||
)
|
||
)
|
||
).scalars())
|
||
|
||
# 定位这次回复的是哪一条：显式指定优先，否则只在唯一 pending 时才敢猜
|
||
target = None
|
||
if approval_id:
|
||
target = next((a for a in pending if a.id == approval_id), None)
|
||
if target is None:
|
||
raise ValueError("这条审批已处理过，或不属于这次运行。请刷新页面查看最新状态")
|
||
elif len(pending) == 1:
|
||
target = pending[0]
|
||
elif len(pending) > 1:
|
||
raise ValueError(
|
||
f"这次运行有 {len(pending)} 条待处理的审批，"
|
||
"无法确定本次回复对应哪一条，请在对应的审批卡上操作"
|
||
)
|
||
|
||
now = datetime.now(timezone.utc)
|
||
if target is not None:
|
||
# 存全文，不能用 _safe 截：多条审批并存时，交给引擎的就是这一份
|
||
# （见下面 answers）。截过的版本会把人工改过的长稿砍掉一截再送进节点
|
||
full = json.loads(json.dumps(response, ensure_ascii=False, default=str))
|
||
# 条件更新：上面读到它还是 pending，到这里之间可能已经被放弃运行关掉
|
||
# （或者另一个人先批了）。无条件写回去，会把关掉的审批又改成已回复
|
||
marked = await session.execute(
|
||
update(Approval)
|
||
.where(Approval.id == target.id, Approval.status == "pending")
|
||
.values(status="answered", # 已回复，但还没提交给引擎
|
||
response=full if isinstance(full, dict) else {"value": full},
|
||
resolved_at=now, **({"resolved_by": actor} if actor else {}))
|
||
)
|
||
if marked.rowcount != 1:
|
||
await session.rollback()
|
||
raise ValueError("这条审批已处理过，或这次运行已被放弃。请刷新页面查看最新状态")
|
||
await session.commit()
|
||
# 审批卡上点的是「始终允许」：这个工具以后改由门控把关。只有带 trust_key 的
|
||
# 审批才有这个按钮（MCP / 自定义工具、探索运行），别的审批带了 always 也不理。
|
||
# 这次运行里本节点后面的调用由节点自己从答复里认出来（见 llm.py 的 granted）
|
||
trust_key = (target.payload or {}).get("trust_key")
|
||
if isinstance(trust_key, str) and trust_key and read_decision(response).always:
|
||
from app.tools.trust import set_trust
|
||
|
||
await set_trust(session, trust_key, "gated")
|
||
|
||
# 凑齐没有？引擎还在等的每一个 interrupt 都得有答案
|
||
answered = list((
|
||
await session.execute(
|
||
select(Approval).where(
|
||
Approval.run_id == run_id,
|
||
Approval.status.in_(["answered", "pending"]),
|
||
).execution_options(populate_existing=True)
|
||
)
|
||
).scalars())
|
||
by_iid = {a.interrupt_id: a for a in answered if a.interrupt_id}
|
||
missing = []
|
||
for item in waiting:
|
||
iid = getattr(item, "id", None)
|
||
rec = by_iid.get(iid)
|
||
if rec is None or rec.status != "answered":
|
||
missing.append(iid)
|
||
continue
|
||
payload = rec.response or {}
|
||
# 只拆我们自己包的那层 {"value": x}（裸值回复）。带 approved / note / args
|
||
# 的结构体要原样交给节点——以前一律取 value，只要回复里带了内容，
|
||
# 驳回、备注、改过的参数就全在这里丢了
|
||
wrapped = isinstance(payload, dict) and set(payload) == {"value"}
|
||
answers[iid] = payload["value"] if wrapped else payload
|
||
|
||
if missing:
|
||
# 还差人没回。保持 interrupted，把已回复的那条留在 answered 上等齐
|
||
await session.commit()
|
||
run = await session.get(Run, run_id)
|
||
if run.status != "interrupted":
|
||
# 审批没了是因为运行在这期间被放弃了（或被别人驳回了）：不能回一个「已记下」
|
||
raise await _resume_refused(session, run)
|
||
return run
|
||
|
||
# 齐了：正式落 resolved，然后驱动引擎。谁批的跟着交给节点，
|
||
# human.resolved 里才写得出签批人
|
||
ready = [rec for rec in answered if rec.status == "answered"]
|
||
actors: dict[str, str | None] = {rec.node_id: rec.resolved_by for rec in ready}
|
||
# 回到 running 也用条件更新占住：从上面读状态到这里隔了好几次 await，
|
||
# 放弃运行可能正好在这中间把它收成了 cancelled（已经封存、发过 run.cancelled）。
|
||
# 无条件写 running，一次取消了的运行就又被拉起来跑了
|
||
await _claim(session, run_id, ("interrupted",), _cannot_resume)
|
||
await session.execute(
|
||
update(Approval)
|
||
.where(Approval.id.in_([rec.id for rec in ready]), Approval.status == "answered")
|
||
.values(status="resolved")
|
||
)
|
||
run = await session.get(Run, run_id)
|
||
bus.set_seq(run_id, await _seq_floor(session, run_id, run.last_seq))
|
||
await session.commit()
|
||
await session.refresh(run)
|
||
|
||
# 单个 interrupt 用标量（LangGraph 两种都认），多个必须用 {interrupt_id: value} 映射。
|
||
#
|
||
# 一个都没在等：这是服务重启时停下的运行（启动扫描或关停时标成 interrupted），
|
||
# 只能从断点接着跑（输入传 None）。以前这里照样发 Command(resume=…)——那个值
|
||
# 会被后面遇到的第一个 interrupt() 当成答复，一个还没轮到的人工关卡就这么被
|
||
# 自动作答了；当前版本的 LangGraph 则直接在内部崩掉。两样都不是"接着跑"
|
||
command: Command | None
|
||
if not waiting:
|
||
command = None
|
||
elif len(waiting) == 1:
|
||
command = Command(resume=response)
|
||
else:
|
||
command = Command(resume=answers)
|
||
await self._emit(run_id, EventType.RUN_RESUMED, data=(
|
||
{"message": "服务重启时中断的运行，从断点继续运行", "actor": actor} if command is None
|
||
else {"response": _safe(response), "actor": actor}
|
||
))
|
||
self._tasks[run_id] = asyncio.create_task(
|
||
self._drive(run_id, spec, command, workflow_id=workflow_id, **carried,
|
||
resumed=sorted(_replayed(snapshot)),
|
||
actors=actors)
|
||
)
|
||
return run
|
||
|
||
async def continue_failed(
|
||
self, run_id: str, *, graph: dict[str, Any] | None = None, actor: str | None = None
|
||
) -> Run:
|
||
"""从失败的地方接着跑，而不是整张图从头来一遍。
|
||
|
||
为什么值得单独开这条路：库里 28 条 failed 里，模型 id 写错、401、循环
|
||
条件语法错、缺必填输入加起来占了大半——全是**配置错了**，改一下就能跑。
|
||
而现在"改一下再跑"意味着前面花两分钟查完的表结构、跑完的 SQL 全部重来。
|
||
checkpointer 一直开着（thread_id = run_id），那些结果本来就在库里躺着。
|
||
|
||
graph 可以带一张改过的图，但只准改配置：节点 id、类型、连线必须和原来
|
||
一模一样。checkpoint 是按节点名存的，改了结构就对不上，续下去会拿着
|
||
错位的通道状态跑出一份似是而非的结果——那比直接报错糟得多。
|
||
|
||
服务重启挂起的运行（interrupted 且没有待审批）也走这里：它没有在等谁，
|
||
断点完好，从断点驱动和失败后接着跑是同一件事。界面上两者都叫「接着跑」。
|
||
|
||
正式运行不能带图接着跑（F2）：它钉的是发布的不可变版本、按 graph_hash 核对，换一张图接着跑，运行就不再等于它
|
||
钉住的那一版，而 version_hash 照旧写着那一版。接着跑也不改发起人：接着跑的人记在 run.resumed 里。
|
||
写外部系统结果未知、还没核实的（app/process/actions.py）也不能接着跑：核实之后系统会自动接着跑。
|
||
"""
|
||
from app.process.actions import unresolved_action
|
||
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if not run:
|
||
raise KeyError("运行记录不存在")
|
||
has_pending = run.status == "interrupted" and (await session.execute(
|
||
select(Approval.id).where(
|
||
Approval.run_id == run_id, Approval.status.in_(["pending", "answered"])
|
||
).limit(1)
|
||
)).first() is not None
|
||
if run.status == "cancelled" and run.end_reason == END_REJECTED:
|
||
raise ContinueRefused(REJECTED_CONTINUE, code=RUN_REJECTED)
|
||
if not (run.status == "failed" or (run.status == "interrupted" and not has_pending)):
|
||
raise ValueError(_cannot_continue(run.status, has_pending))
|
||
if run.status == "failed" and await _rejected_before(session, run):
|
||
# 加 end_reason 之前被驳回的运行记的是 failed（封存不改写）：接着跑会读到同一个驳回答复、再停一次
|
||
raise ContinueRefused(REJECTED_CONTINUE, code=RUN_REJECTED)
|
||
if graph is not None and run.run_class == "formal":
|
||
raise ContinueRefused(
|
||
"正式运行固定在发布的版本上，不能修改配置后继续运行。如需修改，请在画布上修改并发布新版本后重新发起",
|
||
code=FORMAL_GRAPH_LOCKED)
|
||
if run.status == "interrupted" and await unresolved_action(session, run_id):
|
||
raise ContinueRefused(
|
||
"这次运行有一次写入外部系统的结果未知，正在等待核实；核实之后运行会自动继续，无需手动继续运行",
|
||
code=ACTION_UNRESOLVED)
|
||
observed = run.status
|
||
spec = GraphSpec.model_validate(run.graph)
|
||
workflow_id = run.workflow_id
|
||
carried = _carried(run)
|
||
|
||
if graph is not None:
|
||
revised = GraphSpec.model_validate(graph)
|
||
if topology_of(revised) != topology_of(spec):
|
||
raise ValueError(
|
||
"继续运行时只能修改节点配置，不能增删节点或修改连线。"
|
||
"如需修改结构，请重新发起运行。"
|
||
)
|
||
report = validate_graph(revised)
|
||
if not report.ok:
|
||
first = next((i.message for i in report.issues if i.level == "error"), "")
|
||
raise ValueError(f"修改后的工作流未通过校验：{first}")
|
||
spec = revised
|
||
|
||
# 先问引擎到底停在哪。没有待跑的节点就不是"能接着跑"的情形——
|
||
# 硬发一个 None 下去，LangGraph 会从 START 重跑整张图
|
||
run_ctx = RunContext(run_id=run_id, thread_id=run_id, spec=spec, workflow_id=workflow_id)
|
||
app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer)
|
||
config = {"configurable": {"thread_id": run_id}}
|
||
snapshot = await app.aget_state(config)
|
||
pending = [str(n) for n in (getattr(snapshot, "next", None) or [])]
|
||
if not pending:
|
||
# 停在审批上、批了之后才失败的节点（动作确认之后结果未知、工具批准之后执行出错）：LangGraph 的 next 里
|
||
# 没有它（答复的写入挂在它身上），可断点完好，从断点驱动会重新执行它、审批的答复取自断点
|
||
pending = [str(t.name) for t in getattr(snapshot, "tasks", None) or () if getattr(t, "error", None)]
|
||
if not pending:
|
||
raise ValueError(
|
||
"这次运行没有可继续的断点，可能在执行第一个节点前就已中断。请重新发起运行。"
|
||
)
|
||
if stale := await _stale_replay(run_id, spec, snapshot):
|
||
raise ValueError(
|
||
f"这次运行停在「{stale}」时系统已升级，断点格式不再兼容，继续运行会让「{stale}」中"
|
||
f"已执行的工具重复执行。请重新发起运行"
|
||
)
|
||
|
||
done = [n for n in (snapshot.values or {}).get("nodes", {}) if n not in pending]
|
||
async with SessionLocal() as session:
|
||
# 和 resume 一样用条件更新占住：读状态之后，放弃运行、或者另一次接着跑
|
||
# 可能已经先动了它
|
||
await _claim(session, run_id, (observed,),
|
||
lambda status: _cannot_continue(status, False),
|
||
**({"graph": spec.model_dump(mode="json")} if graph is not None else {}))
|
||
run = await session.get(Run, run_id)
|
||
bus.set_seq(run_id, await _seq_floor(session, run_id, run.last_seq))
|
||
await session.commit()
|
||
await session.refresh(run)
|
||
|
||
labels = {n.id: n.title for n in spec.nodes}
|
||
await self._emit(
|
||
run_id, EventType.RUN_RESUMED,
|
||
data={
|
||
"from": pending,
|
||
# 说清楚"哪些没重来"——否则用户看到时间线又动起来，
|
||
# 分不清这次是接着跑还是整张图又跑了一遍
|
||
"message": f"从「{labels.get(pending[0], pending[0])}」继续运行"
|
||
+ (f"，保留前面 {len(done)} 个节点的结果" if done else ""),
|
||
"actor": actor,
|
||
},
|
||
)
|
||
self._tasks[run_id] = asyncio.create_task(
|
||
self._drive(run_id, spec, None, workflow_id=workflow_id, **carried, resumed=pending)
|
||
)
|
||
return run
|
||
|
||
async def cancel(self, run_id: str, *, actor: str | None = None) -> str | None:
|
||
"""停止一次运行。返回 "stopping"（执行任务已取消，终态由它自己收尾）、
|
||
"cancelled"（停在断点上的，这里直接收成终态），或者 None（已经是终态、或不存在）。
|
||
|
||
停在断点上的运行——等审批的、服务重启挂起的——手上没有执行任务。以前这里只会
|
||
去取消任务，于是它们一律「不在执行中」：待审批的那张卡只能批了它、让它跑完。
|
||
"""
|
||
if run_id in self._finalizing:
|
||
# 执行任务已经跑完、正在落状态——刚发出 run.interrupted、还在写 interrupted
|
||
# 的那一刻，界面上已经给出了「放弃这次运行」。这时取消任务会把收尾从中间
|
||
# 打断：状态停在 running、审批还挂着、任务却没了，再点一次只剩 409。
|
||
# 等它落完，再按落下的状态处理
|
||
await self.wait_idle(run_id, timeout=30.0)
|
||
if run_id in self._finalizing: # 收尾卡住了（库被锁着）：也不去打断它
|
||
return None
|
||
task = self._tasks.get(run_id)
|
||
if task and not task.done():
|
||
self._stopped_by[run_id] = actor
|
||
task.cancel()
|
||
return "stopping"
|
||
# 不在本进程里执行（多进程，app/cluster）：在别的活着的进程里跑（或排队）的给它发停止请求，它照本进程的规矩停、发
|
||
# run.cancelled（带上是谁停的），界面上同样是「正在停止」；属主不在了的照租约过期接手（标中断），再按停在断点上的放弃
|
||
from app.cluster import requests, workers
|
||
|
||
async with SessionLocal() as session:
|
||
row = (await session.execute(select(Run.status, Run.owner).where(Run.id == run_id))).first()
|
||
if row is not None and row.status in ("running", "queued") and row.owner and row.owner != identity.owner():
|
||
if await workers.alive(row.owner):
|
||
await requests.send(row.owner, "stop", "run", run_id, actor=actor)
|
||
return "stopping"
|
||
from app.cluster import state
|
||
|
||
await self.reap((await state.current(fresh=True)).trusting())
|
||
return "cancelled" if await self._abandon(run_id, actor) else None
|
||
|
||
async def _abandon(self, run_id: str, actor: str | None) -> bool:
|
||
"""把停在断点上的运行收成 cancelled：关掉待审批、发 run.cancelled、封存。"""
|
||
now = datetime.now(timezone.utc)
|
||
async with SessionLocal() as session:
|
||
# 条件更新占住这次收尾：同一时刻有人批了它（resume 把它改回 running），
|
||
# 或者另一个取消先到了，这里就什么都不做
|
||
claimed = await session.execute(
|
||
update(Run).where(Run.id == run_id, Run.status == "interrupted")
|
||
.values(status="cancelled", end_reason=END_STOPPED, finished_at=now)
|
||
)
|
||
if claimed.rowcount != 1:
|
||
await session.rollback()
|
||
return False
|
||
closed = await _close_approvals(session, run_id, actor, now)
|
||
run = await session.get(Run, run_id)
|
||
timing = await _timing(session, run, 0, now)
|
||
usage = dict(run.usage or {})
|
||
# 停下时记的用量里还没有这次停在审批期间问的质询，这里补上（上次停下时计过的不再计）
|
||
inquiry = (await _usage_of_events(session, run_id)).get("inquiry")
|
||
if inquiry or usage.get("inquiry"):
|
||
usage = _with_inquiry(usage, inquiry)
|
||
run.usage = {**usage, "duration_ms": timing["active_ms"], **timing}
|
||
run.error = "用户取消：等待审批时放弃了这次运行" if closed else "用户取消：挂起期间放弃了这次运行"
|
||
bus.set_seq(run_id, await _seq_floor(session, run_id, run.last_seq))
|
||
# 和状态同一个事务提交，理由同 _finalize
|
||
event = self._event(run_id, EventType.RUN_CANCELLED, data={
|
||
"reason": END_STOPPED, "timing": timing, "actor": actor,
|
||
"message": f"放弃了这次运行，{closed} 条待审批一并关闭" if closed
|
||
else "放弃了这次挂起的运行",
|
||
})
|
||
await _stage(session, event, run)
|
||
await session.commit()
|
||
|
||
await bus.publish(event)
|
||
await self._seal(run_id)
|
||
# 等审批时在线的连接没有关（它还在等恢复），这里让它收到终态再结束
|
||
await bus.close(run_id)
|
||
with contextlib.suppress(Exception):
|
||
from app.sandbox.manager import sandbox_manager
|
||
|
||
await sandbox_manager.cleanup(run_id)
|
||
return True
|
||
|
||
def is_active(self, run_id: str) -> bool:
|
||
task = self._tasks.get(run_id)
|
||
return bool(task and not task.done())
|
||
|
||
async def wait_idle(self, run_id: str, timeout: float = 10.0) -> None:
|
||
"""等这次运行手上的执行任务收完尾。不取消它，也不抛它的异常。"""
|
||
task = self._tasks.get(run_id)
|
||
if task is not None and not task.done():
|
||
await asyncio.wait({task}, timeout=timeout)
|
||
|
||
# ---------------- 执行 ----------------
|
||
|
||
async def _drive(
|
||
self,
|
||
run_id: str,
|
||
spec: GraphSpec,
|
||
payload: Any,
|
||
*,
|
||
workflow_id: str | None = None,
|
||
memory_scope: str = "default",
|
||
collection: str = "default",
|
||
approval_default: str | None = None,
|
||
tool_trust: dict[str, str] | None = None,
|
||
agent_limits: dict[str, Any] | None = None,
|
||
data_versions: dict[str, Any] | None = None,
|
||
options: dict[str, Any] | None = None,
|
||
resumed: list[str] | tuple[str, ...] = (),
|
||
actors: dict[str, str | None] | None = None,
|
||
origin_kind: str | None = None,
|
||
) -> None:
|
||
from app.cluster import slots
|
||
|
||
started = time.perf_counter()
|
||
status = "succeeded"
|
||
error: str | None = None
|
||
final_state: dict[str, Any] = {}
|
||
#: 终态事件（run.failed / run.cancelled）等 _finalize 算完耗时再发，timing 才带得上
|
||
terminal: tuple[EventType, dict[str, Any]] | None = None
|
||
error_node: str | None = None
|
||
#: 收成 cancelled 的原因（Run.end_reason）：rejected 人的驳回，stopped 手动停止
|
||
end_reason: str | None = None
|
||
#: 结果未知、挂起等核实的动作（ActionPending），收尾之后发 action.unknown、安排核实
|
||
pending_action: str | None = None
|
||
|
||
deadline: asyncio.Timeout | None = None
|
||
# 停在审批上的那一刻：这一段的执行算到这里为止。后面写审批记录、发 run.interrupted 是收尾，
|
||
# 等待从 run.interrupted 算起——以前执行一直计到 finally，收尾那一截执行和等待两边都算，
|
||
# active + wait 比墙钟还长（机器一忙就差出几十毫秒）
|
||
paused: float | None = None
|
||
# 这段执行属于现在这一世代（cluster/fence.py）：本进程隔离之后，它不再写。隔离之后新身份还没登记上时先等一等
|
||
try:
|
||
await fence.ready()
|
||
except Fenced:
|
||
logger.error("运行 %s：本进程隔离之后一直没重新登记上，不开始执行（由别的进程按「属主不在了」处理）", run_id)
|
||
return
|
||
fence.bind()
|
||
mine = identity.owner()
|
||
#: 被隔离了（续不上租约）：停手，不写终态（这次运行由接手的进程按「属主不在了」处理）
|
||
fenced = False
|
||
# 先要并发名额（app/cluster/slots.py）。等名额的时候被停止（或者关停）也要照实收尾，见 _stopped_in_queue
|
||
slot = slots.hold(run_id, origin_kind)
|
||
try:
|
||
await slot.__aenter__()
|
||
except asyncio.CancelledError:
|
||
await self._stopped_in_queue(run_id)
|
||
raise
|
||
try:
|
||
# 执行时长从拿到名额算起：排队等名额的那一段不算执行（墙钟从 started_at 算，started_at 也是拿到名额之后才写）
|
||
started = time.perf_counter()
|
||
try:
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row is not None:
|
||
row.status = "running"
|
||
row.owner = mine
|
||
# started_at 只在第一次开始时写。以前每段 _drive 都覆盖它，
|
||
# 审批恢复、接着跑之后，运行的"开始时间"就成了最后一段的开始
|
||
if row.started_at is None:
|
||
row.started_at = datetime.now(timezone.utc)
|
||
row.finished_at = None
|
||
row.error_node_id = None
|
||
await session.commit()
|
||
await self._emit(
|
||
run_id,
|
||
EventType.RUN_STARTED,
|
||
data={"nodes": len(spec.nodes),
|
||
"resumed": payload is None or isinstance(payload, Command),
|
||
# 实际生效的运行参数进事件流，也就进了封存清单：
|
||
# 正式运行查的是哪个库、审批默认是什么，事后可复核
|
||
"memory_scope": memory_scope, "collection": collection,
|
||
"approval_default": approval_default,
|
||
# MCP / 自定义工具的信任三档（只列不是「等审批」的；正式运行记 formal）
|
||
"tool_trust": _trust_event(tool_trust),
|
||
# agent 护栏的上限（步数兜底、令牌 / 金额预算），None 是升级前的运行
|
||
"agent_limits": agent_limits,
|
||
# 这次运行固定的上传表格版本（源 id → 快照 id、源名）。续跑、恢复记的是
|
||
# 运行记录里的那份，含运行中途第一次用到时补上的；None 是升级前的运行
|
||
"data_versions": data_versions,
|
||
# 这一段是按哪一版重放协议跑的，恢复时据此认出升级前做下的工作
|
||
"replay_protocol": PROTOCOL,
|
||
# 发起时定下的运行选项（核查设置之类），每一段都记、记的是同一份。没有选项的
|
||
# 运行不带这个键，和加选项之前的 run.started 逐字相同
|
||
**({"options": options} if options else {})},
|
||
)
|
||
|
||
run_ctx = RunContext(
|
||
run_id=run_id,
|
||
thread_id=run_id,
|
||
spec=spec,
|
||
workflow_id=workflow_id,
|
||
memory_scope=memory_scope,
|
||
collection=collection,
|
||
approval_default=approval_default,
|
||
tool_trust=tool_trust,
|
||
agent_limits=agent_limits,
|
||
# 数据版本放 extra：数据源工具经 ToolContext.data_versions 拿到（tools/datasource.py
|
||
# 的 run_versions）。复制一份：运行中途补固定的源就地写进这个 dict，同一段执行里
|
||
# 后面的节点都看得到，但不该碰到别处（运行记录、事件）手上的那个对象
|
||
extra={"resumed": set(resumed), "actors": dict(actors or {}),
|
||
RUN_VERSIONS_KEY: dict(data_versions) if data_versions is not None else None,
|
||
# 同理也是一份拷贝：节点读它，不该改到运行记录那份
|
||
RUN_OPTIONS_KEY: copy.deepcopy(options or {})},
|
||
)
|
||
app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer)
|
||
# 循环按自己声明的轮数另记一份预算，全局上限只管没人把关的环
|
||
extra_steps = loop_steps(spec)
|
||
config = {
|
||
"configurable": {"thread_id": run_id},
|
||
"recursion_limit": settings.max_graph_steps + extra_steps,
|
||
# 进每个 checkpoint 的 metadata：恢复时据此认出旧版引擎停下的断点
|
||
"metadata": {PROTOCOL_KEY: PROTOCOL},
|
||
}
|
||
|
||
interrupted = False
|
||
# 单次执行的墙钟上限。以前 max_run_seconds 只存在于配置里，没有任何
|
||
# 代码执行它——一个挂住的模型或工具能把运行和它占着的并发名额一直拖着。
|
||
# 按"这一段执行"计：等人工审批时 _drive 已经结束，那段时间不算。
|
||
deadline = asyncio.timeout(settings.max_run_seconds)
|
||
# 截止时刻同时交给节点（engine/deadline.py）：核查节点据此在上限之内收尾，不把整次运行拖到被中止
|
||
async with deadline, segment(deadline.when(), settings.max_run_seconds):
|
||
async for mode, chunk in app.astream(
|
||
payload, config=config, stream_mode=["custom", "updates"]
|
||
):
|
||
if mode == "custom":
|
||
await self._handle_custom(run_id, chunk)
|
||
elif mode == "updates":
|
||
# LangGraph 这一轮的节点都跑完了才发出中断：此刻引擎已经停下
|
||
at = time.perf_counter()
|
||
if await self._handle_updates(run_id, chunk):
|
||
interrupted = True
|
||
paused = at if paused is None else paused
|
||
|
||
snapshot = await app.aget_state(config)
|
||
final_state = dict(snapshot.values or {})
|
||
if interrupted or snapshot.interrupts:
|
||
status = "interrupted"
|
||
elif snapshot.next:
|
||
status = "interrupted"
|
||
|
||
except Fenced:
|
||
fenced = True
|
||
logger.error("运行 %s：本进程续不上租约，停手、不写终态（由接手的进程按「属主不在了」处理）", run_id)
|
||
except asyncio.CancelledError:
|
||
if fence.fenced_here():
|
||
fenced = True
|
||
raise
|
||
if self._closing:
|
||
# 服务关停（dev.sh 的 --reload 改一个文件就是一次）不是用户取消。
|
||
# checkpoint 完好，记成中断、重启后从断点接着跑——和强杀后启动
|
||
# 扫描的处理一致。以前一律记"用户取消"，resume 和 continue 都拒绝它
|
||
status = "interrupted"
|
||
error = "服务重启，运行已挂起，可从断点继续运行"
|
||
await self._emit(run_id, EventType.LOG, data={
|
||
"level": "warn", "code": "server_shutdown",
|
||
"message": "服务关闭时这次运行仍在进行，已挂起；重启后可从断点继续运行",
|
||
})
|
||
else:
|
||
status = "cancelled"
|
||
error = "用户取消"
|
||
end_reason = END_STOPPED
|
||
terminal = (EventType.RUN_CANCELLED, {"reason": END_STOPPED,
|
||
"actor": self._stopped_by.pop(run_id, None)})
|
||
raise
|
||
except ActionPending as e:
|
||
# 写外部系统的结果未知（app/process/actions.py）：不在节点里等，运行收成挂起、不封存，等核实之后
|
||
# 由系统接着跑。节点重新执行时先查生效记录，已生效的交回回执、未知的照旧挂起，不会重发
|
||
status = "interrupted"
|
||
error = str(e)
|
||
pending_action = e.action_id
|
||
except TimeoutError as e:
|
||
status = "failed"
|
||
if deadline is not None and deadline.expired():
|
||
error = (
|
||
f"本次运行超过 {settings.max_run_seconds} 秒的时长上限，已中止。"
|
||
"通常是模型或工具长时间未响应；如确需更长时间，请联系管理员调大时长上限"
|
||
"（AGENTLAB_MAX_RUN_SECONDS）"
|
||
)
|
||
terminal = (EventType.RUN_FAILED, {"error": error})
|
||
else: # 别处抛的超时，不是这道上限——按普通失败说
|
||
error, terminal = _failure(e, spec)
|
||
logger.exception("run %s 失败", run_id)
|
||
except GraphRecursionError:
|
||
status = "failed"
|
||
# LangGraph 的原文是英文，还叫人去调 recursion_limit——用户碰不到那个键。
|
||
# 循环的轮数已经另算了预算，走到这里的多半是不归 loop 节点管的环
|
||
error = (
|
||
f"运行已执行 {settings.max_graph_steps + extra_steps} 步仍未结束，已中止"
|
||
f"（工作流步数上限为 {settings.max_graph_steps}"
|
||
+ (f"，另为循环预留 {extra_steps} 步" if extra_steps else "") + "）。"
|
||
"可能存在没有轮数上限的环路：某个分支连回了上游，但没有用「循环」节点限制轮数。"
|
||
"请用「循环」节点包住这段环路；如确需更多步数，请联系管理员调大步数上限"
|
||
"（AGENTLAB_MAX_GRAPH_STEPS）"
|
||
)
|
||
terminal = (EventType.RUN_FAILED, {"error": error})
|
||
except NodeRejected as e:
|
||
# 人的决定（驳回、超时驳回），不是故障：收成 cancelled，同手动停止走一样的收尾（关掉并行支路上的审批、
|
||
# 回收沙箱、关流、封存、不许继续运行）。原话照旧记在 error 上，在哪一步、谁、结论、备注进 run.cancelled
|
||
status, end_reason = "cancelled", END_REJECTED
|
||
error, error_node = str(e), e.node_id
|
||
terminal = (EventType.RUN_CANCELLED, rejected_data(e, spec))
|
||
except Exception as e: # noqa: BLE001
|
||
status = "failed"
|
||
error, terminal = _failure(e, spec)
|
||
error_node = terminal[1].get("node_id")
|
||
logger.exception("run %s 失败", run_id)
|
||
finally:
|
||
stop = paused if status == "interrupted" and paused is not None else time.perf_counter()
|
||
elapsed = int((stop - started) * 1000)
|
||
self._finalizing.add(run_id)
|
||
try:
|
||
if not fenced:
|
||
await self._finalize(run_id, status, error, final_state, elapsed, terminal=terminal,
|
||
error_node=error_node, end_reason=end_reason, owner=mine)
|
||
# 记 action.unknown、安排核实也算收尾：做完才把自己摘掉。以前摘掉之后才做，等运行「不在跑了」
|
||
# 再去读事件的一方（接口、测试）有一小段读不到这一条；库慢一点（PostgreSQL 满载时）就读不到
|
||
if pending_action is not None and not fenced:
|
||
from app.process.actions import on_unknown
|
||
|
||
try:
|
||
await on_unknown(pending_action)
|
||
except Exception: # noqa: BLE001 - 运行已经收成挂起；没安排上的核实，下次启动时 actions.startup 补上
|
||
logger.exception("run %s 的动作 %s 结果未知，安排核实时出错", run_id, pending_action)
|
||
finally:
|
||
# 收尾被打断（关停时的取消、库连接出错）也得把自己摘掉。以前写在
|
||
# _finalize 后面，一被打断就漏掉，留下的任务让下一次 shutdown 的
|
||
# gather 在别的事件循环里炸开
|
||
self._finalizing.discard(run_id)
|
||
self._tasks.pop(run_id, None)
|
||
self._stopped_by.pop(run_id, None)
|
||
# 运行到终态就把沙箱会话收掉。thread_id 每个 run 都是新的，
|
||
# 不收的话每跑一次带代码节点的图就多一台常驻 microVM（几百 MB），
|
||
# 而且没有任何自动回收路径会碰它——闲置回收只在"下次有人执行
|
||
# 代码"时才触发。interrupted 要留着，人回来还得接着跑。
|
||
if status in ("succeeded", "failed", "cancelled") and not fenced:
|
||
with contextlib.suppress(Exception):
|
||
from app.sandbox.manager import sandbox_manager
|
||
|
||
await sandbox_manager.cleanup(run_id)
|
||
finally:
|
||
await slot.__aexit__(None, None, None)
|
||
|
||
async def _stopped_in_queue(self, run_id: str) -> None:
|
||
"""还在等并发名额时被取消。关停：什么都不写（还是排队中 / 刚认领的样子），照「属主不在了」的规矩由下次启动或别的
|
||
进程接手（排队的标失败、认领了的标中断）；人点了停止：收成 cancelled，同执行中停止（run.cancelled 带上是谁停的、封存）。"""
|
||
try:
|
||
if self._closing:
|
||
return
|
||
if fence.fenced_here():
|
||
return
|
||
actor = self._stopped_by.pop(run_id, None)
|
||
self._finalizing.add(run_id)
|
||
await self._finalize(run_id, "cancelled", "用户取消", {}, 0,
|
||
terminal=(EventType.RUN_CANCELLED, {"reason": END_STOPPED, "actor": actor}),
|
||
end_reason=END_STOPPED, owner=identity.owner())
|
||
finally:
|
||
self._finalizing.discard(run_id)
|
||
self._tasks.pop(run_id, None)
|
||
self._stopped_by.pop(run_id, None)
|
||
|
||
async def _handle_custom(self, run_id: str, chunk: Any) -> None:
|
||
"""节点通过 get_stream_writer 发来的事件。"""
|
||
if not isinstance(chunk, dict) or not chunk.get("__agentlab__"):
|
||
return
|
||
await self._emit(
|
||
run_id,
|
||
chunk.get("type", EventType.LOG),
|
||
node_id=chunk.get("node_id"),
|
||
data=chunk.get("data") or {},
|
||
ts=chunk.get("ts"),
|
||
)
|
||
|
||
async def _handle_updates(self, run_id: str, chunk: Any) -> bool:
|
||
"""LangGraph 的状态更新。这里只关心中断信号 —— 节点级事件已经由 custom 流覆盖。"""
|
||
if not isinstance(chunk, dict):
|
||
return False
|
||
interrupts = chunk.get("__interrupt__")
|
||
if not interrupts:
|
||
return False
|
||
fence.check()
|
||
|
||
for item in interrupts:
|
||
value = getattr(item, "value", item)
|
||
interrupt_id = getattr(item, "id", None)
|
||
payload = value if isinstance(value, dict) else {"value": value}
|
||
opened: str | None = None
|
||
async with SessionLocal() as session:
|
||
existing = (
|
||
await session.execute(
|
||
select(Approval).where(
|
||
Approval.run_id == run_id,
|
||
Approval.interrupt_id == interrupt_id,
|
||
Approval.status == "pending",
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is None:
|
||
approval = Approval(
|
||
run_id=run_id,
|
||
node_id=str(payload.get("node_id") or ""),
|
||
interrupt_id=interrupt_id,
|
||
mode=str(payload.get("mode") or "approve"),
|
||
title=str(payload.get("title") or "待审批"),
|
||
payload=payload,
|
||
schema_=payload.get("schema") or {},
|
||
)
|
||
session.add(approval)
|
||
await session.flush()
|
||
# 同一次中断给人的那一面（app/process/tasks.py）：和 Approval 同一个事务，有审批就有待办。
|
||
# 开待办出了错不能连审批一起丢——运行确实停在这里等人；下次启动时按「派给任何人」补建
|
||
try:
|
||
async with session.begin_nested():
|
||
from app.process.tasks import open_task
|
||
|
||
opened = (await open_task(session, approval)).id
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("run %s 的审批 %s 没能开出待办，启动时会补建", run_id, approval.id)
|
||
await session.commit()
|
||
await self._emit(
|
||
run_id,
|
||
EventType.RUN_INTERRUPTED,
|
||
node_id=payload.get("node_id"),
|
||
data={"interrupt_id": interrupt_id, "payload": _safe(payload)},
|
||
)
|
||
if opened:
|
||
from app.process.tasks import announce
|
||
|
||
await announce(opened)
|
||
return True
|
||
|
||
async def _finalize(
|
||
self,
|
||
run_id: str,
|
||
status: str,
|
||
error: str | None,
|
||
state: dict[str, Any],
|
||
elapsed_ms: int,
|
||
*,
|
||
terminal: tuple[EventType, dict[str, Any]] | None = None,
|
||
error_node: str | None = None,
|
||
end_reason: str | None = None,
|
||
owner: str | None = None,
|
||
) -> None:
|
||
"""收尾。owner：执行这一段的属主——运行那一行的属主已经不是它（被别的进程当成没了接手了，cluster/fence.py）就什么都不写。"""
|
||
usage = dict(state.get("usage") or {})
|
||
output = dict(state.get("output") or {})
|
||
now = datetime.now(timezone.utc)
|
||
timing = {"wall_ms": elapsed_ms, "active_ms": elapsed_ms, "wait_ms": 0}
|
||
|
||
# 终态事件和运行记录在同一个事务里提交，提交之后再广播。两边都有人在读：
|
||
# 读到 failed 的一方马上去翻事件，要翻得到 run.failed（报错原因和定位都在
|
||
# 那条事件里）；收到 run.finished 的一方马上 GET 这次运行取完整成果（事件里
|
||
# 的 output 是截断的），要读得到 succeeded 和 output。分两次提交，总有一边
|
||
# 会读到半截——以前先提交状态，前一种偶尔翻不到；反过来先落事件，后一种
|
||
# 每次都读到 running 和空的 output
|
||
event: RunEventModel | None = None
|
||
app_id: str | None = None
|
||
#: 收尾之后要不要放到后台比对口径（ledger/match.py）：只看探索运行
|
||
run_class: str | None = None
|
||
origin_kind: str | None = None
|
||
async with SessionLocal() as session:
|
||
# 先占住这一行再看属主（PostgreSQL 上 FOR UPDATE；SQLite 只有一个进程）：接手的进程改属主用的是带条件的 UPDATE，
|
||
# 两边一个接一个，被接手了的这边不写
|
||
run = (await session.execute(select(Run).where(Run.id == run_id).with_for_update())).scalar_one_or_none()
|
||
if run is not None and owner is not None and run.owner != owner:
|
||
logger.error("运行 %s 已经被别的进程接手（属主 %s），本进程不写终态", run_id, identity.describe(run.owner))
|
||
await session.rollback()
|
||
return
|
||
if run:
|
||
app_id = run.app_id
|
||
run_class, origin_kind = run.run_class, run.origin_kind
|
||
timing = await _timing(session, run, elapsed_ms, now)
|
||
events_usage = await _usage_of_events(session, run_id)
|
||
inquiry = events_usage.pop("inquiry", None)
|
||
if status != "succeeded":
|
||
# 状态里的用量只算跑完了的节点。失败、取消、停在审批上的那个节点已经
|
||
# 花出去的调用（校验修复、agent 的前几步）只在 llm.end 里——以前这部分
|
||
# 在运行记录上凭空消失，失败的运行看着像没花钱
|
||
usage = _spent(usage, events_usage)
|
||
elif events_usage.get("unpriced_calls"):
|
||
# 金额估不出来的调用只有 llm.end 数得出（状态里的用量只有金额，0 分不出是免费还是没价）
|
||
usage = {**usage, "unpriced_calls": events_usage["unpriced_calls"]}
|
||
if inquiry:
|
||
# 审批质询不在图里执行，状态和 llm.end 两本账里都没有它：比完再加上
|
||
usage = _with_inquiry(usage, inquiry)
|
||
# duration_ms 是各段执行时长之和（= active_ms）。以前用这一段的耗时
|
||
# 覆盖它：87 秒、中间等过一次审批的运行，列表上写着 22ms
|
||
usage.update(duration_ms=timing["active_ms"], **timing)
|
||
|
||
if status == "succeeded":
|
||
# 事件里的 output 仍然截断（事件体积要控制），但截了就要说：消费方据此
|
||
# 回源 GET /api/runs/{id} 取全文。以前截得悄无声息，问数据页把 2000 字处
|
||
# 切断的半截答案当成完整结论落了库
|
||
clipped, truncated = _clip(output)
|
||
data: dict[str, Any] = {"output": clipped, "usage": usage,
|
||
"duration_ms": timing["active_ms"], "timing": timing, **_rejected_via(state)}
|
||
if truncated:
|
||
data["output_truncated"] = True
|
||
event = self._event(run_id, EventType.RUN_FINISHED, data=data)
|
||
elif terminal is not None:
|
||
kind, data = terminal
|
||
event = self._event(run_id, kind, data={**data, "timing": timing, **_rejected_via(state)})
|
||
|
||
if run:
|
||
run.status = status
|
||
run.error = error
|
||
run.output = output
|
||
run.usage = usage
|
||
run.error_node_id = error_node
|
||
run.end_reason = end_reason
|
||
if status != "interrupted":
|
||
run.finished_at = now
|
||
else:
|
||
# 停在审批上（或挂起）的运行不属于任何进程：谁收到决定谁接着跑（resume / continue_failed 的条件 UPDATE
|
||
# 认领，同一个决定只会被一个进程执行）
|
||
run.owner = None
|
||
if status in ("failed", "cancelled"):
|
||
# 并行支路上已经停下来等人的那张审批卡，运行结束了它也就没有意义了；
|
||
# 留着的话它一直挂在全局待办里，批了也只会被告知运行不在断点上
|
||
actor = terminal[1].get("actor") if terminal else None
|
||
await _close_approvals(session, run_id, actor, now)
|
||
if event is not None:
|
||
await _stage(session, event, run)
|
||
from app.scheduler import outbox # 数据事件触发：正式运行成功结束，同一个事务记一条待派发事件
|
||
# 途经驳回出口、按驳回那条路走完的也是 succeeded：照发，事件带上途经了哪几个驳回（缺省不认途经驳回的触发）
|
||
outbox.stage_run_finished(session, run, rejected_via=_rejected_via(state).get("rejected_via"))
|
||
await session.commit()
|
||
if event is not None:
|
||
await bus.publish(event)
|
||
|
||
if status != "interrupted":
|
||
# 终态事件落库之后再封存。以前在 run.finished 之前算，清单里恰恰少了
|
||
# 那条宣布"跑完了、成果是什么"的事件——改它的成果，清单照样对得上。
|
||
# 中断态不封：事件还会继续追加
|
||
await self._seal(run_id)
|
||
if status in ("succeeded", "failed", "cancelled") or (
|
||
status == "interrupted" and self._closing
|
||
):
|
||
# 关停时挂起的也要让在线的客户端收尾：进程马上就没了，不说一声，
|
||
# 界面会一直以为它还在跑
|
||
await bus.close(run_id)
|
||
if app_id and (status in ("succeeded", "failed") or end_reason == END_REJECTED):
|
||
# 应用实例结束、失败：按应用订阅推送（app/notify/service.py）。放到后台，接收端慢不拖着收尾。被驳回的走
|
||
# 「实例结束」（next-round X6）：发起人要知道结果，订阅「实例失败」的人盯的是故障；手动停止的不推，同以前
|
||
from app.notify.service import notify_run, spawn
|
||
|
||
spawn(notify_run(run_id, "run.failed" if status == "failed" else "run.finished"))
|
||
if status == "succeeded":
|
||
# 口径账本（docs/design/next-round/02-口径账本.md 做法 2）：探索运行成功结束、封存完，拿这次执行的 SQL 和已发布口径
|
||
# 比对。和通知同一个位置、同样放到后台：比对是 CPU 活，在线程里限时限并发，不拖着收尾；结果追加在封存之后
|
||
from app.ledger import match as ledger_match
|
||
|
||
if ledger_match.wanted(run_class, origin_kind):
|
||
ledger_match.schedule(run_id)
|
||
if origin_kind == "recall" and status in ("succeeded", "failed", "cancelled"):
|
||
# 口径召回运行结束、封存完（ledger/recall.py）：在后台收尾——数变了开 recall 待办，没变的在原答案上记账
|
||
from app.ledger import recall as ledger_recall
|
||
|
||
ledger_recall.schedule_settle(run_id)
|
||
|
||
async def _seal(self, run_id: str) -> None:
|
||
"""对到此为止的全部事件算清单哈希，记下封到了哪一条。"""
|
||
from app.core.artifact_store import manifest_hash as _manifest
|
||
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if run is None:
|
||
return
|
||
rows = [tuple(r) for r in 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(rows)
|
||
run.manifest_seq = rows[-1][0] if rows else 0
|
||
await session.commit()
|
||
|
||
|
||
_TERMINAL = (EventType.RUN_FINISHED, EventType.RUN_FAILED, EventType.RUN_CANCELLED)
|
||
|
||
|
||
async def verify_manifest(run_id: str) -> dict[str, Any]:
|
||
"""重算封存范围内的清单哈希，和封存时记下的对一下。
|
||
|
||
以前只写不验：README 说"事后修改流水将无法对齐"，可仓库里没有一行代码去对。
|
||
封存范围之后追加的事件不参与核对。没有 manifest_seq 的是加这一列之前封存的
|
||
老运行——当时的哈希是在 run.finished 落库之前算的，按那时的口径还原范围。
|
||
"""
|
||
from app.core.artifact_store import manifest_hash as _manifest
|
||
|
||
async with SessionLocal() as session:
|
||
run = await session.get(Run, run_id)
|
||
if run is None:
|
||
raise KeyError("运行记录不存在")
|
||
if not run.manifest_hash:
|
||
return {"sealed": False, "ok": None,
|
||
"message": "这次运行尚未封存（尚未结束，或正停在人工审批）"}
|
||
rows = [tuple(r) for r in await session.execute(
|
||
select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data)
|
||
.where(RunEvent.run_id == run_id)
|
||
.order_by(RunEvent.seq)
|
||
)]
|
||
expected, sealed_at = run.manifest_hash, run.manifest_seq
|
||
|
||
legacy = sealed_at is None
|
||
if legacy:
|
||
last = next((r for r in reversed(rows) if r[1] in _TERMINAL), None)
|
||
if last is None:
|
||
sealed_at = rows[-1][0] if rows else 0
|
||
elif last[1] == EventType.RUN_FINISHED:
|
||
sealed_at = last[0] - 1 # 老口径：run.finished 本身不在清单里
|
||
else:
|
||
sealed_at = last[0] # run.failed / run.cancelled 在封存前就落库了
|
||
covered = [r for r in rows if r[0] <= sealed_at]
|
||
ok = _manifest(covered) == expected
|
||
return {
|
||
"sealed": True, "ok": ok, "events": len(covered), "sealed_at": sealed_at, "legacy": legacy,
|
||
"message": "事件记录与封存时一致" if ok else "事件记录与封存时不一致：封存后有事件被修改、删除或插入",
|
||
}
|
||
|
||
|
||
def _safe(value: Any, limit: int = 4000) -> Any:
|
||
"""把任意对象裁剪成能安全放进事件负载的形状。"""
|
||
return _clip(value, limit)[0]
|
||
|
||
|
||
def _clip(value: Any, limit: int = 4000) -> tuple[Any, bool]:
|
||
"""同 _safe，另外说出有没有真的截掉东西。"""
|
||
if isinstance(value, str):
|
||
return value[:limit], len(value) > limit
|
||
if isinstance(value, (dict, list)):
|
||
items = list(value.items()) if isinstance(value, dict) else list(enumerate(value))
|
||
cut = len(items) > 50
|
||
kept: list[tuple[Any, Any]] = []
|
||
for key, item in items[:50]:
|
||
clipped, lost = _clip(item, limit // 2)
|
||
kept.append((key, clipped))
|
||
cut = cut or lost
|
||
return (dict(kept) if isinstance(value, dict) else [v for _, v in kept]), cut
|
||
if isinstance(value, (int, float, bool)) or value is None:
|
||
return value, False
|
||
text = str(value)
|
||
return text[:limit], len(text) > limit
|
||
|
||
|
||
_STATUS_TEXT = {
|
||
"queued": "仍在排队", "running": "正在运行", "interrupted": "停在断点上",
|
||
"succeeded": "已完成", "failed": "已失败", "cancelled": "已取消",
|
||
}
|
||
|
||
|
||
async def _resume_refused(session: Any, run: Run) -> ValueError:
|
||
"""恢复被拒时抛什么：被驳回的（cancelled + end_reason=rejected，以及加 end_reason 之前记成 failed 的驳回）说「被驳回」、
|
||
带 RUN_REJECTED，和继续运行同一个错误码（以前照状态说「已取消的运行请重新发起」、对存量的还叫人点「继续运行」，
|
||
可继续运行也会被拒）；别的照状态说（_cannot_resume）。"""
|
||
if (run.status == "cancelled" and run.end_reason == END_REJECTED) or (
|
||
run.status == "failed" and await _rejected_before(session, run)):
|
||
return ContinueRefused(REJECTED_RESUME, code=RUN_REJECTED)
|
||
return ValueError(_cannot_resume(run.status))
|
||
|
||
|
||
def _cannot_resume(status: str) -> str:
|
||
hint = {
|
||
"running": "运行仍在进行，无需恢复。",
|
||
"queued": "运行尚未开始。",
|
||
"succeeded": "如需再次运行，请重新发起。",
|
||
"failed": "失败的运行可点「继续运行」从出错的节点继续。",
|
||
"cancelled": "已取消的运行请重新发起。",
|
||
}.get(status, "")
|
||
return f"这次运行{_STATUS_TEXT.get(status, status)}，没有待处理的审批，无法恢复。{hint}"
|
||
|
||
|
||
def _cannot_continue(status: str, has_pending: bool) -> str:
|
||
"""接着跑被拒时说清楚为什么、下一步做什么。
|
||
|
||
以前一律套一句「在等待审批，只有失败的运行能接着跑」：服务重启挂起的也被说成
|
||
在等审批，还让人去一张不存在的审批卡上处理；已取消的只说不行，不说该怎么办。
|
||
"""
|
||
if status == "interrupted" and has_pending:
|
||
return "这次运行正在等待审批，无需继续运行。在审批卡上批准或驳回后，运行会自动继续。"
|
||
return {
|
||
"cancelled": "这次运行已取消，无法继续运行。如需继续，请重新发起运行。",
|
||
"succeeded": "这次运行已完成，无需继续运行：只有失败的运行或被服务重启打断的运行"
|
||
"可以继续运行。如需再次运行，请重新发起。",
|
||
"running": "这次运行正在进行，无需继续运行。",
|
||
"queued": "这次运行仍在排队，尚未开始，无需继续运行。",
|
||
}.get(status, f"这次运行{_STATUS_TEXT.get(status, status)}，"
|
||
"只有失败的运行或被服务重启打断的运行可以继续运行。")
|
||
|
||
|
||
def _carried(run: Run) -> dict[str, Any]:
|
||
"""第一次发起时定下的运行参数。恢复、接着跑都沿用，老运行没记的按缺省。"""
|
||
return {
|
||
"memory_scope": run.memory_scope or "default",
|
||
"collection": run.collection or "default",
|
||
"approval_default": run.approval_default,
|
||
"tool_trust": run.tool_trust,
|
||
"agent_limits": run.agent_limits,
|
||
# 固定的数据版本照运行记录来，不重新读数据源的当前指针：中途重传过，续跑查的仍是发起时那一版
|
||
"data_versions": run.data_versions,
|
||
# 运行选项同理：照发起时的快照，不重新读设置
|
||
"options": run.options or {},
|
||
# 并发名额按来源归类（app/cluster/slots.py）：恢复、接着跑的那一段和第一段同一类
|
||
"origin_kind": run.origin_kind,
|
||
}
|
||
|
||
|
||
async def _data_versions(session: Any, spec: GraphSpec) -> dict[str, Any]:
|
||
"""发起时固定这次运行用到的上传表格版本：{源 id: {"snapshot": 当前快照 id, "name": 源名}}。
|
||
|
||
和 tool_trust、agent_limits 一样在发起这一刻定下、存进运行记录：续跑、审批恢复沿用它，
|
||
不再读数据源的当前指针——运行中途有人重传，这次运行查的仍是发起时那一版。数据源工具按它
|
||
解析源（tools/datasource.py），回收按它保留快照（data/table_versions.gc）。
|
||
|
||
只算这张图实际用到的源：调用工具节点、Agent、协作成员绑的 db_query__ / db_schema__，连同
|
||
子工作流里绑的。没用到的源不固定，否则它们的旧版本会因为这次运行白白留着。手工源不分版本，
|
||
不进来。工具名里有模板（{{ }}）时，发起时说不准它会落到哪个源，就把所有启用的上传源都固定上
|
||
（多固定几个只是多留几个旧版本；漏掉一个，这次运行就可能前后查到两版数据）。
|
||
|
||
发起时算不到的（继续运行时改了节点配置、没固定版本的子工作流中途被改过、运行中途新建的源）
|
||
由数据源工具第一次用到时补上（tools/datasource.py 的 _pin_late，记进运行记录并发一条日志事件）。
|
||
没用到任何上传源时返回空 dict 而不是 None：None 留给升级前发起的运行（照旧用当前版本，不补固定）。
|
||
|
||
记录表（origin=records，docs/design/normal-user-apps.md N8）同样在这里固定高水位：{"seq": 高水位, …}
|
||
（records/views.pin_entry）。这次运行看哪些行由发起处另外记上（records/views.scope_entries）：按需视图的缓存键
|
||
只要高水位。
|
||
"""
|
||
from app.db.models import DataSource
|
||
from app.records.views import pin_entry
|
||
|
||
names, anything = await _bound_sources(session, spec)
|
||
if not names and not anything:
|
||
return {}
|
||
query = select(DataSource.id, DataSource.name, DataSource.current_snapshot_id).where(
|
||
DataSource.enabled.is_(True), DataSource.origin == "upload",
|
||
DataSource.current_snapshot_id.is_not(None),
|
||
).order_by(DataSource.name)
|
||
if not anything:
|
||
query = query.where(DataSource.name.in_(sorted(names)))
|
||
rows = (await session.execute(query)).all()
|
||
out: dict[str, Any] = {sid: {"snapshot": snap, "name": name} for sid, name, snap in rows}
|
||
records = select(DataSource).where(DataSource.enabled.is_(True), DataSource.origin == "records") \
|
||
.order_by(DataSource.name)
|
||
if not anything:
|
||
records = records.where(DataSource.name.in_(sorted(names)))
|
||
for source in (await session.execute(records)).scalars():
|
||
if (entry := await pin_entry(session, source)) is not None:
|
||
out[source.id] = entry
|
||
return out
|
||
|
||
|
||
async def _bound_sources(session: Any, spec: GraphSpec) -> tuple[set[str], bool]:
|
||
"""图里（含子工作流）绑定的数据源工具指向的源名，以及有没有带模板、发起时定不下来的工具名。
|
||
|
||
绑定的取法和发起前检查同一个（engine/launch.py 的 tool_bindings），子工作流照运行时的取法
|
||
读出来接着找：固定了版本的读那一版，没固定的读工作流现在的图。读不到、图不成形的跳过——
|
||
那个节点运行时自己会报错。同一个工作流（同一版）只读一次，互相引用也不会转圈。
|
||
"""
|
||
# 在函数里导入：launch 在模块级导入了本模块（它要 run_manager）
|
||
from app.engine.launch import node_setting, tool_bindings
|
||
from app.db.models import WorkflowVersion
|
||
from app.tools.datasource import source_of_tool
|
||
|
||
names: set[str] = set()
|
||
anything = False
|
||
seen: set[tuple[str, str]] = set()
|
||
queue: list[GraphSpec] = [spec]
|
||
while queue:
|
||
graph = queue.pop(0)
|
||
for _, tool in tool_bindings(graph):
|
||
if "{{" in tool:
|
||
anything = True
|
||
elif (source := source_of_tool(tool)) is not None:
|
||
names.add(source)
|
||
for node in graph.nodes:
|
||
if node.type != NodeType.SUBGRAPH:
|
||
continue
|
||
# 和 NodeContext.cfg 一样：节点上没写就取图级 defaults（tool_bindings 读工具也是这个规则）
|
||
wid, pinned = (node_setting(graph, node, k) for k in ("workflow_id", "workflow_version"))
|
||
key = (str(wid or ""), str(pinned or ""))
|
||
if not wid or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
try:
|
||
# 只取图：Workflow 的 versions 是 selectin，整行读会把所有历史版本一起读出来
|
||
query = (select(WorkflowVersion.graph).where(
|
||
WorkflowVersion.workflow_id == wid, WorkflowVersion.version == int(pinned))
|
||
if pinned else select(Workflow.graph).where(Workflow.id == wid))
|
||
sub = (await session.execute(query)).scalar_one_or_none()
|
||
if sub:
|
||
queue.append(GraphSpec.model_validate(sub))
|
||
except Exception: # noqa: BLE001 - 读不到、不成形：子工作流节点运行时会说清楚
|
||
continue
|
||
return names, anything
|
||
|
||
|
||
async def _agent_limits(session: Any) -> dict[str, Any]:
|
||
from app.api.settings import run_defaults
|
||
from app.engine.guards import run_limits
|
||
|
||
return run_limits(await run_defaults(session), settings.max_agent_steps)
|
||
|
||
|
||
def _trust_event(tool_trust: dict[str, str] | None) -> Any:
|
||
from app.tools.trust import FORMAL_MARK, snapshot_levels
|
||
|
||
if tool_trust is None:
|
||
return None
|
||
return "formal" if tool_trust.get(FORMAL_MARK) else snapshot_levels(tool_trust)
|
||
|
||
|
||
def _rejected_via(state: dict[str, Any]) -> dict[str, Any]:
|
||
"""这次运行途经的驳回出口（engine/compiler._rejected_exit 记在 trail 上的 outcome=rejected）：{rejected_via: [{node_id,
|
||
label, kind, decision, actor, note, to}]}，按先后。终态事件带上它，运行照它的路径结束（成功结束也一样）照实说「途经
|
||
驳回」；没有途经的不带这个键，终态事件和以前一字不差。读的是收尾时的状态：节点出错、被驳回结束的那几种收尾拿不到
|
||
状态（final_state 为空），不带。子工作流里走了驳回出口的那几步也在这里：子图跑过的步骤交回父图的 trail（节点 id 带
|
||
「子工作流节点 id/」前缀、名字写「子工作流 › x」，engine/nodes/multi.run_subgraph）。"""
|
||
via = [{k: t.get(k) for k in ("node_id", "label", "kind", "decision", "actor", "note", "to")}
|
||
for t in state.get("trail") or [] if isinstance(t, dict) and t.get("outcome") == "rejected"]
|
||
return {"rejected_via": via} if via else {}
|
||
|
||
|
||
def rejected_data(exc: NodeRejected, spec: GraphSpec) -> dict[str, Any]:
|
||
"""被驳回的 run.cancelled 带什么：原因（rejected）、在哪一步（节点 id 和图上的名字）、哪种关卡、结论（驳回 / 超时
|
||
驳回）、谁、备注。进封存：谁在哪一步驳回、为什么，事后查得到。"""
|
||
node = spec.node_map().get(exc.node_id)
|
||
return {"reason": END_REJECTED, "node_id": exc.node_id, "label": _label_of(exc, node),
|
||
"kind": exc.kind, "decision": exc.decision, "actor": exc.actor, "note": exc.note}
|
||
|
||
|
||
async def _rejected_before(session: Any, run: Run) -> bool:
|
||
"""加 end_reason 之前被人驳回、记成 failed 的运行：失败的那一步上最近一条处理完的待办是驳回（和实例页认驳回的是
|
||
同一份判断，apps/resolve.legacy_rejected）。"""
|
||
from app.apps.resolve import legacy_rejected
|
||
|
||
return await legacy_rejected(session, run)
|
||
|
||
|
||
def _failure(exc: BaseException, spec: GraphSpec) -> tuple[str, tuple[EventType, dict[str, Any]]]:
|
||
"""失败的首行报错（给人看）和 run.failed 的负载（带上定位和原始异常）。"""
|
||
if isinstance(exc, NodeError):
|
||
# NodeError 的 message 本来就是写给人看的（"人工驳回：…"），
|
||
# 再套一层类型名只会让界面上的错误更难读
|
||
message = str(exc)
|
||
detail = raw_detail(exc.__cause__) if exc.__cause__ is not None else None
|
||
node_id: str | None = exc.node_id
|
||
else:
|
||
message = (f"运行出错：{describe_exception(exc)}。技术细节已记录在服务日志中；"
|
||
"可点「继续运行」重试，如仍失败，请将运行编号反馈给维护者")
|
||
detail = raw_detail(exc)
|
||
node_id = None
|
||
data: dict[str, Any] = {"error": message}
|
||
if detail:
|
||
data["detail"] = detail
|
||
if node_id:
|
||
node = spec.node_map().get(node_id)
|
||
data.update(node_id=node_id, label=_label_of(exc, node))
|
||
return message, (EventType.RUN_FAILED, data)
|
||
|
||
|
||
def _label_of(exc: BaseException, node: Any) -> str:
|
||
"""出错、被驳回的那一步在图上叫什么。子工作流里的那一步（节点 id「子工作流节点 id/…」）不在这张图上，名字由
|
||
multi.run_subgraph 接好记在异常上（「子工作流 › 那一步」）。"""
|
||
if node is not None:
|
||
return node.title
|
||
return str(getattr(exc, "label", None) or getattr(exc, "node_id", "") or "")
|
||
|
||
|
||
async def _approval_default(*, run_class: str) -> str:
|
||
"""工具审批策略的全局默认，取自设置里的「危险工具默认需要人工确认」。
|
||
|
||
正式运行不吃"关"。它跑的是封存的发布版，审批行为只该由那一版里写明的配置
|
||
决定，不能随一个随时能改的偏好变；受管门禁要求危险工具至少人工确认，也不能被
|
||
全局开关悄悄降掉。不能按工作流的 status 判断是不是受管：画布上再存一次，status
|
||
就退回 draft，而 published_version 仍指着受管的那一版。
|
||
"""
|
||
if run_class == "formal":
|
||
return "dangerous"
|
||
from app.api.settings import run_defaults, tool_approval_default
|
||
|
||
async with SessionLocal() as session:
|
||
return tool_approval_default(await run_defaults(session))
|
||
|
||
|
||
def _replayed(snapshot: Any) -> set[str]:
|
||
"""从断点驱动时要（重新）执行的节点：还没跑的（next），加上停在 interrupt 上、或者
|
||
失败了的。恢复过一次的节点手里已经有答复的写入，不在 next 里，只能从 tasks 上认。"""
|
||
again = {str(x) for x in getattr(snapshot, "next", None) or ()}
|
||
again.update(str(t.name) for t in getattr(snapshot, "tasks", None) or ()
|
||
if getattr(t, "interrupts", None) or getattr(t, "error", None))
|
||
return again
|
||
|
||
|
||
async def _stale_replay(run_id: str, spec: GraphSpec, snapshot: Any) -> str | None:
|
||
"""旧版重放协议写下的断点，按现在的代码重放会不会把做过的工具再做一遍。会的话返回节点名。
|
||
|
||
有风险的是：断点之后，旧版代码在这个节点里已经执行过工具、或者已经批过一次。它们
|
||
的 task 缓存排在旧的位置上，新代码多插了记录 task，位置全错开了。停在第一次审批上、
|
||
之前什么都没做的，重放只会多发一条 human.requested，照常放行；人工审批节点同理。
|
||
|
||
「断点之后」按 checkpoint 的写入时刻算：同一个 superstep 里恢复不会写新的 checkpoint，
|
||
所以升级后恢复过一次的运行，断点仍是旧版写的，而那之后的工作是新版做的——
|
||
哪一段是哪一版，看那一段 run.started 里记的 replay_protocol。
|
||
|
||
只多豁免一种（应用平台 A4 第 9 条）：调用工具节点在断点之后的每一条 tool.start 都带 action_key——这是写入动作，
|
||
发送前一定先按 action_key 查生效记录（app/process/actions.py），已生效的交回回执、已发出的转核实，重复执行由数据库
|
||
挡住，task 的位置错开也不会重发。一条 tool.start 都没有的不算（说不清它是不是写入节点），照旧拒绝。
|
||
"""
|
||
if protocol_of(getattr(snapshot, "metadata", None)) >= PROTOCOL:
|
||
return None
|
||
nodes = spec.node_map()
|
||
suspects = [n for n in _replayed(snapshot) if n in nodes and nodes[n].type != NodeType.HUMAN]
|
||
if not suspects:
|
||
return None
|
||
try:
|
||
since = datetime.fromisoformat(str(snapshot.created_at)).timestamp()
|
||
except (AttributeError, TypeError, ValueError):
|
||
since = 0.0
|
||
async with SessionLocal() as session:
|
||
rows = (await session.execute(
|
||
select(RunEvent.type, RunEvent.node_id, RunEvent.ts, RunEvent.data)
|
||
.where(RunEvent.run_id == run_id,
|
||
RunEvent.type.in_([EventType.RUN_STARTED, EventType.TOOL_START,
|
||
EventType.HUMAN_RESOLVED]))
|
||
.order_by(RunEvent.seq)
|
||
)).all()
|
||
protocol = 1
|
||
done: dict[str, list[tuple[str, dict[str, Any]]]] = {}
|
||
for kind, owner, ts, data in rows:
|
||
if kind == EventType.RUN_STARTED:
|
||
protocol = protocol_of({PROTOCOL_KEY: (data or {}).get("replay_protocol")})
|
||
continue
|
||
if protocol >= PROTOCOL or (ts or 0) < since:
|
||
continue
|
||
for node_id in suspects:
|
||
if owner == node_id or (owner or "").startswith(f"{node_id}/"):
|
||
done.setdefault(node_id, []).append((kind, data or {}))
|
||
for node_id, events in done.items():
|
||
starts = [data for kind, data in events if kind == EventType.TOOL_START]
|
||
if nodes[node_id].type == NodeType.TOOL and starts and all(d.get("action_key") for d in starts):
|
||
continue
|
||
return nodes[node_id].title or node_id
|
||
return None
|
||
|
||
|
||
async def _claim(
|
||
session: Any, run_id: str, expected: tuple[str, ...], refusal: Callable[[str], str],
|
||
**values: Any,
|
||
) -> None:
|
||
"""只在状态仍是 expected 时把运行改成 running；已经被别人改过就回滚、说明原因。"""
|
||
claimed = await session.execute(
|
||
update(Run).where(Run.id == run_id, Run.status.in_(expected))
|
||
.values(status="running", error=None, owner=identity.owner(), **values)
|
||
)
|
||
if claimed.rowcount != 1:
|
||
await session.rollback()
|
||
status = await session.scalar(select(Run.status).where(Run.id == run_id))
|
||
raise ValueError(refusal(status or ""))
|
||
|
||
|
||
async def _stage(session: Any, event: RunEventModel, run: Run | None = None) -> None:
|
||
"""把一条事件写进这个 session，随它一起提交。run 已经在 session 里时直接改它的 last_seq。
|
||
|
||
落库的 id 用事件自己的：广播出去的、交回给调用方的和库里的是同一个 id，回头按 id 找得到。
|
||
"""
|
||
session.add(RunEvent(id=event.id, run_id=event.run_id, seq=event.seq, type=str(event.type),
|
||
node_id=event.node_id, ts=event.ts, data=event.data))
|
||
if run is not None:
|
||
run.last_seq = event.seq
|
||
else:
|
||
# 只往大里改：另开会话记的事件（note）和执行任务同时在写时，取号和提交的先后可以不一样，号小的后提交，
|
||
# 无条件写会把 last_seq 写小（后果见 _seq_floor）
|
||
await session.execute(
|
||
update(Run).where(Run.id == event.run_id)
|
||
.values(last_seq=case((Run.last_seq > event.seq, Run.last_seq), else_=event.seq))
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
|
||
|
||
async def _seq_floor(session: Any, run_id: str, last_seq: int | None) -> int:
|
||
"""接着一次运行往下编号的起点：last_seq 和库里实际最大的序号取大的。
|
||
|
||
last_seq 按说就是最后一条，但有的写入会把它写小：按需裁判记判定（api/evidence.py 的 _append_judged）
|
||
的条件更新不带 last_seq 条件，它取了号还没写、封存后追加（engine/run_append.py）抢先提交了，它再提交时
|
||
把 last_seq 写回自己那个更小的号。同一进程里序号计数只增不减，不碍事；进程重启后计数从 0 起，只按
|
||
last_seq 接号就会撞上已有的序号。清单按序号排序算哈希，撞了号封存核对就时对时不对。
|
||
"""
|
||
stored = await session.scalar(select(func.max(RunEvent.seq)).where(RunEvent.run_id == run_id))
|
||
return max(last_seq or 0, stored or 0)
|
||
|
||
|
||
async def _close_approvals(session: Any, run_id: str, actor: str | None, now: datetime) -> int:
|
||
"""运行已经结束，它还没处理完的审批一并关掉，对应的待办（连同核实待办）跟着关闭。返回关掉了几条审批。"""
|
||
from app.process.tasks import close_for_run
|
||
|
||
rows = list((await session.execute(
|
||
select(Approval).where(Approval.run_id == run_id,
|
||
Approval.status.in_(["pending", "answered"]))
|
||
)).scalars())
|
||
for approval in rows:
|
||
approval.status = "cancelled"
|
||
approval.resolved_at = now
|
||
approval.resolved_by = actor
|
||
await close_for_run(session, run_id, now)
|
||
return len(rows)
|
||
|
||
|
||
_USAGE_KEYS = ("input_tokens", "output_tokens", "cost_usd", "calls")
|
||
#: 审批质询的用量里计进运行用量的键：unpriced_calls 是其中金额估不出来的调用次数（模型不在价格目录里）
|
||
_INQUIRY_KEYS = (*_USAGE_KEYS, "unpriced_calls")
|
||
|
||
|
||
async def _usage_of_events(session: Any, run_id: str) -> dict[str, Any]:
|
||
"""按事件加总的用量。
|
||
|
||
llm.end 逐次加总：每次模型调用都有一条，不管节点最后成没成。审批质询不在图里执行、不发 llm.end，
|
||
它的用量写在 approval.inquiry 事件数据的 usage 里，单独加总放在 inquiry 下（没有质询就没有这个键）：
|
||
llm.end 这本账要和状态里的用量比大小（_spent），质询花的两本里都没有，混进去比会被状态那本盖掉。
|
||
|
||
封存之后追加的质询（数据带 post_seal）不算：运行用量是这次运行自己花的，结束时就定下了、写进了终态事件；
|
||
事后的批注另有自己的账，同按需裁判的判定。继续运行之后它落进新的封存范围，照样不算——不然同一条质询
|
||
计不计，全看之后有没有继续运行。
|
||
"""
|
||
rows = await session.execute(
|
||
select(RunEvent.type, RunEvent.data).where(
|
||
RunEvent.run_id == run_id, RunEvent.type.in_([EventType.LLM_END, EventType.APPROVAL_INQUIRY]))
|
||
)
|
||
from app.providers.catalog import unpriced_call
|
||
|
||
total: dict[str, Any] = dict.fromkeys(_USAGE_KEYS, 0)
|
||
unpriced = 0
|
||
inquiry: dict[str, Any] = {}
|
||
for kind, data in rows:
|
||
if kind == EventType.APPROVAL_INQUIRY:
|
||
spent = (data or {}).get("usage")
|
||
if isinstance(spent, dict) and not (data or {}).get("post_seal"):
|
||
for key in _INQUIRY_KEYS:
|
||
value = spent.get(key)
|
||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
inquiry[key] = inquiry.get(key, 0) + value
|
||
continue
|
||
for key in _USAGE_KEYS:
|
||
value = (data or {}).get(key, 1 if key == "calls" else 0)
|
||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
total[key] += value
|
||
unpriced += unpriced_call(data)
|
||
total["cost_usd"] = round(total["cost_usd"], 6)
|
||
if unpriced:
|
||
# 有几次调用的金额估不出来（模型不在价格目录里）：cost_usd 只是估得出的那部分，界面据此不写 $0
|
||
total["unpriced_calls"] = unpriced
|
||
if inquiry:
|
||
if "cost_usd" in inquiry:
|
||
inquiry["cost_usd"] = round(inquiry["cost_usd"], 6)
|
||
total["inquiry"] = inquiry
|
||
return total
|
||
|
||
|
||
def _with_inquiry(usage: dict[str, Any], inquiry: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""把审批质询的用量计进运行用量，明细另放在 inquiry 下（界面要说清其中多少是质询花的）。
|
||
|
||
usage 里已经计过一回的（上次停下时记下的 inquiry）先减掉再加：停过两次、两次都问过的运行，
|
||
每次质询只计一遍。
|
||
"""
|
||
out = dict(usage)
|
||
counted = out.pop("inquiry", None) or {}
|
||
for key in _INQUIRY_KEYS:
|
||
delta = (inquiry or {}).get(key, 0) - counted.get(key, 0)
|
||
if delta:
|
||
out[key] = (out.get(key) or 0) + delta
|
||
if isinstance(out.get("cost_usd"), float):
|
||
out["cost_usd"] = round(out["cost_usd"], 6)
|
||
if "input_tokens" in out or "output_tokens" in out:
|
||
out["total_tokens"] = int(out.get("input_tokens") or 0) + int(out.get("output_tokens") or 0)
|
||
if inquiry:
|
||
out["inquiry"] = dict(inquiry)
|
||
return out
|
||
|
||
|
||
def _spent(state_usage: dict[str, Any], events_usage: dict[str, Any]) -> dict[str, Any]:
|
||
"""两本账取大的那本。新引擎每次调用都发 llm.end，事件那本不会更少；
|
||
加这条事件之前的老运行（agent 节点不发 llm.end）状态那本更全。"""
|
||
out = dict(state_usage)
|
||
for key in _USAGE_KEYS:
|
||
mine, theirs = out.get(key) or 0, events_usage.get(key) or 0
|
||
if isinstance(mine, (int, float)) and theirs > mine:
|
||
out[key] = theirs
|
||
if events_usage.get("unpriced_calls"):
|
||
out["unpriced_calls"] = events_usage["unpriced_calls"]
|
||
if "input_tokens" in out or "output_tokens" in out:
|
||
out["total_tokens"] = int(out.get("input_tokens") or 0) + int(out.get("output_tokens") or 0)
|
||
return out
|
||
|
||
|
||
async def _timing(session: Any, run: Run, elapsed_ms: int, now: datetime) -> dict[str, int]:
|
||
"""wall：从第一次开始到现在的墙钟；active：各段执行时长之和；wait：等人审批的总时长。
|
||
|
||
等待按事件算：每段从 run.interrupted 到下一条 run.resumed。服务重启挂起的那段
|
||
没有 run.interrupted，不算等审批；失败后到接着跑之间也不算。
|
||
"""
|
||
prev = run.usage or {}
|
||
before = prev.get("active_ms", prev.get("duration_ms", 0))
|
||
active = int(before or 0) + elapsed_ms
|
||
|
||
rows = await session.execute(
|
||
select(RunEvent.type, RunEvent.ts)
|
||
.where(RunEvent.run_id == run.id,
|
||
RunEvent.type.in_([EventType.RUN_INTERRUPTED, EventType.RUN_RESUMED]))
|
||
.order_by(RunEvent.seq)
|
||
)
|
||
waited, since = 0.0, None
|
||
for kind, ts in rows:
|
||
if kind == EventType.RUN_INTERRUPTED:
|
||
since = ts if since is None else since
|
||
elif since is not None:
|
||
waited += max(0.0, ts - since)
|
||
since = None
|
||
if since is not None:
|
||
waited += max(0.0, now.timestamp() - since)
|
||
|
||
wall = int((now - run.started_at).total_seconds() * 1000) if run.started_at else active
|
||
return {"wall_ms": max(wall, active), "active_ms": active, "wait_ms": int(waited * 1000)}
|
||
|
||
|
||
run_manager = RunManager()
|
||
|
||
|
||
def _fenced_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver:
|
||
"""断点库的写也要过 cluster/fence：本进程续不上租约之后，手上的执行不再写断点（接手的进程会从断点接着跑，两边都写就分叉了）。
|
||
|
||
包成协程、在协程里查（复核 R-b8）：LangGraph 在节点的完成回调里调 aput_writes 拿协程交给后台任务——同步包装在回调里就抛了，
|
||
每次隔离 asyncio 都记一条「Exception in callback … FuturesDict.on_done」；在协程里抛，异常留在写断点的那个任务里，由 LangGraph
|
||
照常收走。"""
|
||
for name in ("aput", "aput_writes"):
|
||
original = getattr(saver, name, None)
|
||
if original is None:
|
||
continue
|
||
|
||
async def guarded(*args: Any, __original: Any = original, **kwargs: Any) -> Any:
|
||
fence.check()
|
||
return await __original(*args, **kwargs)
|
||
|
||
setattr(saver, name, guarded)
|
||
return saver
|
||
|
||
|
||
def _stop_fenced() -> None:
|
||
"""隔离了：停掉本进程在执行的运行（停的时候不写终态，见 _drive 的 fenced）。"""
|
||
for task in list(run_manager._tasks.values()):
|
||
if not task.done():
|
||
task.cancel()
|
||
|
||
|
||
fence.on_trip(_stop_fenced)
|
||
|
||
|
||
def _register_requests() -> None:
|
||
"""别的进程发来的：停止本进程在执行的运行、往它里面记一条事件（app/cluster/requests.py）。"""
|
||
from app.cluster import requests
|
||
|
||
@requests.handle("stop", "run")
|
||
async def _stop(run_id: str, actor: str | None, _payload: dict[str, Any]) -> None:
|
||
# 不在本进程执行的也照停止的规矩办（cancel 自己看属主）：在别的活着的进程里执行就转给它，属主不在了就接手、放弃，停在
|
||
# 断点上的收成 cancelled，已经结束的什么都不做。隔离之后转到新身份名下的停止请求就是这样落下的（复核 R-b10）
|
||
await run_manager.cancel(run_id, actor=actor)
|
||
|
||
@requests.handle("note", "run")
|
||
async def _note(run_id: str, _actor: str | None, payload: dict[str, Any]) -> None:
|
||
kind, node_id, data = payload.get("type") or EventType.LOG, payload.get("node_id"), dict(payload.get("data") or {})
|
||
if run_manager.is_active(run_id):
|
||
await run_manager._emit(run_id, kind, node_id=node_id, data=data)
|
||
return
|
||
# 收到时运行已经停下了：停在审批上的接在库里最后一条后面记（同 process/events.record），别的不记
|
||
from app.engine.run_append import append_events
|
||
|
||
if await append_events(run_id, [(kind, node_id, data)], require="interrupted") is None:
|
||
logger.info("运行 %s 已经不在执行，别的进程要记的一条 %s 没有记", run_id, kind)
|
||
|
||
|
||
_register_requests()
|