AgentLab/backend/app/engine/runner.py

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 = "服务重启时这次运行仍在排队，没有可恢复的进度，请重新发起运行"
#: 本进程隔离（续不上租约）之后没能重新登记，这一段执行没有开始（cluster/fence.ready 等不到）
NOT_REREGISTERED_QUEUED = "本进程隔离后没能重新登记，这次运行没有开始执行，请重新发起运行"
NOT_REREGISTERED_RESUMABLE = "本进程隔离后没能重新登记，这次运行没有接着执行，已挂起，可从断点继续运行"


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:
            await self._not_reregistered(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 _not_reregistered(self, run_id: str) -> None:
        """隔离之后新身份一直没登记上、这一段执行不开始了（fence.ready 等不到，复核 R-b15）：照实收尾，不留给别的进程按「服务
        重启时还在排队」去标。排队中的（一步都没执行过）收成失败；已经执行过、停在断点上又要接着执行的收成中断（断点完好，可以
        继续运行）。条件更新：状态、属主还是刚才看到的才改（别的进程先接手了就不动）。"""
        from app.cluster.claims import same

        logger.error("运行 %s：本进程隔离之后一直没重新登记上，这一段执行不开始", run_id)
        try:
            event: RunEventModel | None = None
            async with SessionLocal() as session:
                seen = (await session.execute(select(Run.status, Run.owner).where(Run.id == run_id))).first()
                if seen is None or seen.status not in ("queued", "running"):
                    return
                if seen.status == "queued":
                    values: dict[str, Any] = {"status": "failed", "error": NOT_REREGISTERED_QUEUED}
                else:
                    values = {"status": "interrupted", "error": NOT_REREGISTERED_RESUMABLE}
                claimed = await session.execute(update(Run).where(
                    Run.id == run_id, Run.status == seen.status, same(Run.owner, seen.owner))
                    .values(owner=None, **values).execution_options(synchronize_session=False))
                if claimed.rowcount != 1:
                    await session.rollback()
                    return
                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.RUN_FAILED, data={"error": NOT_REREGISTERED_QUEUED})
                         if seen.status == "queued" else
                         self._event(run_id, EventType.LOG, data={"level": "warn", "code": "not_reregistered",
                                                                  "message": NOT_REREGISTERED_RESUMABLE}))
                await _stage(session, event, run)
                await session.commit()
            await bus.publish(event)
            await bus.close(run_id)
        except Exception:  # noqa: BLE001 - 收不了尾的留给别的进程按「属主不在了」处理
            logger.exception("运行 %s：收尾时出错，留给别的进程按属主不在了处理", run_id)
        finally:
            self._tasks.pop(run_id, None)
            self._stopped_by.pop(run_id, 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()