跳到主要内容

数据截至 (上游 commit 97b9856e49f2)

人在环中:暂停/恢复的实现与状态机

30 秒导读: PySpur 的招牌特性是让一条工作流在跑到一半时停下来等人——审批、修改、放行,然后从断点继续。本章讲透这套暂停/恢复机制:一个节点怎么用异常把整条流"冻住",执行器怎么接住这个异常、把下游节点标成"待定"而不是"取消/失败",以及恢复时怎么带着人给的新输入从断点续跑,不重算已完成的部分。

本章聚焦 pause/resume 状态机这一块。工作流的通用拓扑调度、异步并发和失败传播见 02-executor.md;任务状态怎么落库的持久化细节见 05(LLM 与 Agent 一章之外的持久化,引到 05)。这里只讲"人在环中"这条最绕的支线。

1. 这是什么(零基础也能懂)​

  • 一句话定义: Human-in-the-Loop(人在环中,简称 HITL)= 让工作流在某个节点主动暂停,把控制权交回给人,等人做完决定再从那里往下跑。

  • 解决什么问题 / 给谁用: 假设你搭了一条自动化流水线——AI 读一封客诉邮件、起草一份退款方案、然后执行退款。你大概率不想让最后这步全自动:退款前必须有个人看一眼、点"批准"。HITL 就是在流水线中间插一个"栅栏",跑到这里必须有人放行。

  • 它能做什么:

    • 在任意位置插入一个 HumanInterventionNode(人工干预节点),工作流跑到这里就停。
    • 停下来时,把当前的数据原样透传保存下来,等人来看。
    • 人可以选三种动作:批准 / 拒绝 / 覆盖(改掉数据再放行)。
    • 恢复后,不重跑已经算完的节点,只跑被这道栅栏挡住的下游。

  • 一句话直觉/类比: 把工作流想成一条流水线,HumanInterventionNode 就是流水线上的一道闸门。闸门一落,闸门后面所有工位都进入"待命(PENDING)"——注意是待命,不是报废。人一按放行键,闸门抬起,后面的工位从待命状态接着干,而闸门前面已经装配好的部件原样保留、不返工。

"待命而非报废"是本章的核心暗线。 后面你会反复看到代码在做同一件事:凡是暂停节点的下游,状态一律往 PENDING 掰,绝不留成 CANCELED 或 FAILED——因为报废了就没法恢复了。

2. 顶层全景(它大概怎么转)​

一次"暂停 → 人操作 → 恢复"的完整生命周期,横跨三个角色:触发暂停的节点、接住暂停的执行器、驱动恢复的 API。

怎么读下面这张图: 从上往下是时间顺序;左侧是"暂停半程",右侧是"恢复半程";中间那条虚线是人的介入点(工作流在这里真正停住,等 HTTP 请求)。

        暂停半程                                  恢复半程
  ┌──────────────────────┐               ┌──────────────────────────┐
  │ HumanInterventionNode │               │  POST /process_pause_    │
  │      .run()          │               │      action/{run_id}     │
  │  raise PauseError    │               │  (APPROVE/DECLINE/       │
  │  (node_id,msg,output)│               │        OVERRIDE)         │
  └──────────┬───────────┘               └────────────┬─────────────┘
             │ 异常冒泡                                │
             ▼                                        ▼
  ┌──────────────────────┐               ┌──────────────────────────┐
  │ _execute_node 捕获    │               │ _update_paused_task:     │
  │ except PauseError    │               │  PAUSED → COMPLETED      │
  │  → _handle_pause_    │               │  存人给的 inputs 为输出   │
  │      exception       │               └────────────┬─────────────┘
  └──────────┬───────────┘                            ▼
             ▼                             ┌──────────────────────────┐
  ┌──────────────────────┐               │ 重建 executor,带上       │
  │ 该节点 → PAUSED       │               │  resumed_node_ids=       │
  │ 下游 → PENDING(BFS)   │               │   [paused_node_id]       │
  │ run → PAUSED         │               │ precomputed_outputs=     │
  └──────────┬───────────┘               │   已完成节点的输出        │
             ▼                             └────────────┬─────────────┘
       ┌ ─ ─ ─ ─ ─ ┐                                   ▼
        工作流停住,      ← 人的介入点 →      ┌──────────────────────────┐
        等 HTTP 恢复                          │ executor.run(只跑下游)    │
       └ ─ ─ ─ ─ ─ ┘                          │ 跳过已完成 / 已恢复节点   │
                                              │ run → COMPLETED          │
                                              └──────────────────────────┘

各部件一句话职责:

部件干什么在哪个文件
HumanInterventionNode.run透传输入并 raise PauseErrorbackend/pyspur/nodes/logic/human_intervention.py:86
PauseError承载 node_id / message / output 的暂停信号异常human_intervention.py:10
PauseAction恢复时人可选的动作枚举(APPROVE/DECLINE/OVERRIDE)human_intervention.py:25
_handle_pause_exception执行器接住暂停,编排三件事:标节点、标下游、标 runexecution/workflow_executor.py:360
get_blocked_nodesBFS 沿"被依赖"方向找出所有下游节点workflow_executor.py:174
is_downstream_of_pause判断某节点是否在某个暂停节点的下游workflow_executor.py:210
_fix_canceled_tasks_after_pause把误判成 CANCELED 的下游改回 PENDINGworkflow_executor.py:395
add_resumed_node_id登记"这个节点已恢复,别再当暂停/别再新建任务"workflow_executor.py:949
process_pause_action恢复流程的编排入口(API)api/workflow_run.py:1137

主线走一遍(高层): 执行器像往常一样并发跑各节点(见 02) → 某节点是 HumanInterventionNode,它 raise PauseError → 执行器捕获,把该节点标 PAUSED、把它的整条下游标 PENDING、把整个 run 标 PAUSED,然后把 PauseError 再抛给 API 层 → 工作流实际停住,等人。人通过 HTTP 提交一个动作 → API 重建执行器,告诉它"这个节点已经恢复了",并把之前算好的节点输出当作 precomputed_outputs 塞进去,只跑被挡住的下游 → run 标 COMPLETED。

3. 触发:一个节点如何"喊停"​

3.1 它要解决的小问题​

节点是工作流里最小的执行单位(见 03-node-system.md)。普通节点算完返回一个输出。但"人工干预节点"要做的事很反常:它不返回,而是要让整条工作流停下来。问题是——一个小小的节点,怎么有能力冻结它上面的整个引擎?

3.2 思路/直觉:用异常当"暂停信号"​

答案很朴素:抛异常。异常天然会沿调用栈往上冒泡,一路冒到执行器那层。PySpur 定义了一个专门的异常类型 PauseError,它不代表"出错",而代表"请在这里暂停"。执行器只要单独识别这一种异常、区别于真正的失败异常来处理,就能实现暂停。

PauseError 携带三样东西,恢复时都用得上:

字段含义
node_id是哪个节点喊的停(定位下游要用)
message给人看的提示文案(如"请审批退款")
output暂停这一刻透传下来的数据,原样保存
# 示意,非源码:PauseError 的角色
class PauseError(Exception):
    def __init__(self, node_id, message, output):
        self.node_id = node_id   # 谁喊的停
        self.message = message   # 给人看的话
        self.output = output     # 停这一刻的数据,留着给人看/给下游用

真实定义见 human_intervention.py:10 的 PauseError——注意它的 __str__ 拼成 Workflow paused at node {node_id}: {message},这行文案后面会被 API 当作暂停提示读出来。

3.3 三种恢复动作:PauseAction​

暂停之后人能做什么,由 PauseAction 枚举定义(human_intervention.py:25):

动作语义
APPROVE批准,原样放行
DECLINE拒绝
OVERRIDE覆盖——人改掉数据后再放行

这个枚举被 ResumeRunRequestSchema.action 引用(schemas/run_schemas.py:67),是人提交恢复请求时必带的字段。

一个诚实的边界: 在本 commit 的执行器/恢复代码里,action 字段被记录下来,但恢复逻辑主要看的是人提交的 inputs(见 §5)。也就是说,APPROVE 和 OVERRIDE 在续跑时走的是同一条"用新 inputs 覆盖暂停节点输出"的路径,差别更多在语义/审计层面,而非分支出不同的执行逻辑。代码里没有对三种 action 做显式的分支处理。

3.4 真实实现:run() 只做两件事​

HumanInterventionNode.run 出奇地短(human_intervention.py:86)——它只做两件事:把输入原样 dump 成输出,然后立刻抛 PauseError。

# 真实源码 human_intervention.py:92-95(HumanInterventionNode.run)
output_dict = input.model_dump()                          # 输入原样透传
output = HumanInterventionNodeOutput(**output_dict)
raise PauseError(str(self.node_id), self.config.message, output)  # 喊停

为什么要透传输入? 因为暂停节点本身不做任何计算,它只是个"闸门"。它把上游给的数据原封不动往下传,这样恢复后下游节点仍能用 {{HumanInterventionNode_1.input_node.input_1}} 这样的模板路径引用到数据(节点输出的 Config.extra = "allow" 就是为了让这些透传字段能挂上去,见 human_intervention.py:54-56)。

block_only_dependent_nodes 配置(human_intervention.py:38): 节点还有个配置项,决定"暂停时到底挡住谁"——只挡直接依赖它的节点,还是挡所有下游。这个语义在 §4 的 get_blocked_nodes 里落地。

4. 执行器如何接住暂停(核心:下游标 PENDING 而非 CANCELED)​

这是本章工程含量最高的一块。执行器要在并发跑一堆节点的同时,精确地把"暂停"和"失败"区分开,并守住"下游可恢复"这条底线。

4.1 第一层:在节点执行处捕获​

执行器给每个节点起一个 async 任务跑 _execute_node(见 02)。在真正 await node_instance(...) 的地方,PauseError 被单独 except 出来,交给 _handle_pause_exception,然后返回 None(而不是继续把异常往外抛,那会污染下游)。

# 真实源码 workflow_executor.py:644-647(_execute_node 内)
except PauseError as e:
    self._handle_pause_exception(node_id, e)
    # Return None to prevent downstream execution
    return None

4.2 编排入口:_handle_pause_exception​

_handle_pause_exception(workflow_executor.py:360)是暂停处理的总指挥,依次做三件事,最后一次性 commit:

_handle_pause_exception(node_id, e)
      │
      ├─▶ _mark_node_as_paused(node_id, e.output)   该节点 → PAUSED,存 output
      │
      ├─▶ _mark_downstream_nodes_as_pending(node_id) 整条下游 → PENDING
      │
      └─▶ _update_run_status_to_paused()             整个 run → PAUSED
                    │
                    ▼
             db_session.commit()  一次性落库

  • 标暂停节点(_mark_node_as_paused, workflow_executor.py:298):把 PauseError.output 存进 self._outputs[node_id],并把任务状态更新为 TaskStatus.PAUSED。这就是恢复时能拿回"暂停那一刻数据"的来源。

  • 标 run(_update_run_status_to_paused, workflow_executor.py:339):把 RunModel.status 设成 RunStatus.PAUSED。注意它不立即 commit,留给 _handle_pause_exception 末尾统一提交——避免半截状态落库。

4.3 找下游:get_blocked_nodes 的 BFS​

要把"下游"全标成 PENDING,先得知道谁是下游。get_blocked_nodes(workflow_executor.py:174)做的就是这件事。

思路: 执行器手里存的 _dependencies 是"每个节点依赖谁"(A ← {B, C} 表示 A 依赖 B、C)。但我们要的是反方向——"谁依赖暂停节点"。所以先把依赖关系反转成一张"被依赖图"(dependents),再从暂停节点出发做 BFS(广度优先搜索),沿着"被依赖"的边一层层往下走,凡是能走到的都是下游。

# 真实源码 workflow_executor.py:189-206(get_blocked_nodes,已精简注释)
dependents = defaultdict(set)
for node_id, deps in self._dependencies.items():
    for dep_id in deps:
        dependents[dep_id].add(node_id)      # 反转:dep_id 被 node_id 依赖
queue = deque([paused_node_id]); visited = set()
while queue:
    current = queue.popleft(); visited.add(current)
    for dependent in dependents.get(current, set()):
        if dependent not in visited:
            blocked_nodes.add(dependent)     # 收集下游
            queue.append(dependent)

举例: A → B → C,且 A → D(A 后面分叉到 B 和 D)。若 B 是暂停节点,get_blocked_nodes(B) 从 B 出发,只能走到 C,得到 {C}。D 不受影响——它不在 B 的下游,应该正常跑完。这正是"只挡真正被挡住的"的精确性所在。

4.4 标下游:_mark_downstream_nodes_as_pending​

拿到下游集合后,_mark_downstream_nodes_as_pending(workflow_executor.py:314)把每个下游节点的任务标成 PENDING,并带上关键标志 is_downstream_of_pause=True;同时如果这些节点之前被误加进 _failed_nodes,还要移除出来。

# 真实源码 workflow_executor.py:325-335(节选)
for blocked_node_id in blocked_nodes:
    self.task_recorder.update_task(
        node_id=blocked_node_id,
        status=TaskStatus.PENDING,
        end_time=current_time,
        is_downstream_of_pause=True,   # 关键标志
    )
    if blocked_node_id in self._failed_nodes:
        self._failed_nodes.remove(blocked_node_id)  # 从失败集合里捞回来

4.5 为什么"PENDING 而非 CANCELED/FAILED"是贯穿逻辑​

这里要讲清本章最核心的设计。暂停会引发一连串误判:暂停节点返回了 None(见 §4.1),而执行器的通用调度逻辑里,一个依赖的输出是 None,通常意味着"这条路断了,把我标成 CANCELED"。如果放任不管,暂停节点的整条下游会被逐个标成 CANCELED——而 CANCELED 是不可恢复的终态,恢复时就没法再跑它们了。

所以整个执行器散布着同一个防御动作:一旦发现某节点的依赖处于 PAUSED,就把它标成 PENDING 而不是 CANCELED。这套防御分布在四处:

位置场景做法
workflow_executor.py:453_execute_node 开头,发现自己在暂停节点下游直接标 PENDING 返回 None,根本不进计算
workflow_executor.py:493-502依赖里有 PAUSED 节点标 PENDING 而非往下走
workflow_executor.py:504-530依赖输出为 None,但需先检查是否有 PAUSED 依赖有则 PENDING,无则才 CANCELED
workflow_executor.py:671-685UpstreamFailureError 兜底若在暂停节点下游,改标 PENDING 而非 CANCELED

最后还有一道兜底防线,连 TaskRecorder 层都加了保险:update_task 里只要传了 is_downstream_of_pause=True,而状态又是 FAILED/CANCELED,就强制掰回 PENDING 并清空 error(execution/task_recorder.py:172-174):

# 真实源码 task_recorder.py:171-174
# If task is downstream of a paused node, mark it as pending instead of failed/canceled
if is_downstream_of_pause and status in [TaskStatus.FAILED, TaskStatus.CANCELED]:
    status = TaskStatus.PENDING
    error = None  # Clear any error message

一句话:下游 = PENDING 这条不变式(invariant),被从 _execute_node 到 TaskRecorder 层层设防守住,因为它是"可恢复"的前提。

4.6 收尾纠错:_fix_canceled_tasks_after_pause​

尽管处处设防,并发环境下仍可能有节点赶在暂停被处理之前就把自己标成了 CANCELED(时序竞态)。所以在 _execute_workflow 全部任务 gather 完之后,还有一道事后纠错(workflow_executor.py:849-850)。

_fix_canceled_tasks_after_pause(workflow_executor.py:395)的做法:重新从数据库查出这个 run 的所有任务,通过 _get_tasks_to_update(workflow_executor.py:379)算出"哪些任务当前是 CANCELED、但其实在某个暂停节点的下游",然后把它们统统改回 PENDING。

# 真实源码 workflow_executor.py:388-393(_get_tasks_to_update 的判定)
return [
    task.node_id
    for task in run.tasks
    if task.status == TaskStatus.CANCELED and task.node_id in all_blocked_nodes
]

这就是"防御 + 事后纠错"的双保险:实时逻辑尽量标对,标错了的最后再统一扫一遍捞回来。

4.7 谁是暂停下游:is_downstream_of_pause​

is_downstream_of_pause(workflow_executor.py:210)是上面那些防御动作在 _execute_node 开头调用的判定函数。它做两件值得注意的事:

  1. 恢复优先短路(workflow_executor.py:221-222): 如果这个节点 node_id in self._resumed_node_ids,直接返回 False——已恢复的节点不算暂停下游,该正常跑。这是 §5 恢复机制和暂停机制的接缝。
  2. 从 TaskRecorder 找当前仍 PAUSED 的节点(workflow_executor.py:228-231),且同样排除掉 _resumed_node_ids 里的,再用 get_blocked_nodes 判断当前节点是否落在某个暂停节点的下游。

5. 恢复:带着人的输入从断点续跑​

暂停半程解决了"怎么停住、怎么不把下游搞报废"。恢复半程要解决的是:怎么从断点接着跑,还不重算已经算完的节点。

5.1 入口:一个 HTTP 动作​

人通过 POST /process_pause_action/{run_id}/(路由在 api/workflow_management.py:73)提交一个 ResumeRunRequestSchema——里面带 inputs(人给的/改过的数据)、user_id、action(§3.3 那三种)。这个端点转手调 process_pause_action(api/workflow_run.py:1137),它是恢复流程的编排入口,依次做四步。

5.2 第一步:把暂停任务改成 COMPLETED​

_update_paused_task(workflow_run.py:986)找到那个 PAUSED 任务,把它翻成 COMPLETED,并把人提交的 inputs 存成这个节点的 outputs:

# 真实源码 workflow_run.py:1002-1005
paused_task.end_time = datetime.now(timezone.utc)
paused_task.status = TaskStatus.COMPLETED   # 暂停节点视为"已完成"
paused_task.error = None
paused_task.outputs = action_request.inputs # 人给的输入 = 这个节点的输出

这一步之后,还会删掉同一节点残留的 PENDING 任务(workflow_run.py:1007-1020),避免恢复时重复建任务。

5.3 第二步:重建执行器,带上 resumed_node_ids​

恢复不是在原来那个执行器实例上继续(那个进程可能早没了),而是重新 new 一个 WorkflowExecutor,并通过构造参数 resumed_node_ids 告诉它:"这个暂停节点已经恢复了。"

# 真实源码 workflow_run.py:1069-1074(_setup_workflow_executor)
executor = WorkflowExecutor(
    workflow=workflow_definition,
    task_recorder=task_recorder,
    context=context,
    resumed_node_ids=[paused_task.node_id],  # 告诉执行器:这个节点已恢复
)

resumed_node_ids 是恢复机制的钥匙。 回到 §4.7:is_downstream_of_pause 一开头就检查 _resumed_node_ids,命中就返回 False。所以把暂停节点登记进这个集合,等于关掉了针对它的暂停拦截——它的下游不再被视为"暂停下游",可以正常往下跑了。构造器里 self._resumed_node_ids = set(resumed_node_ids or [])(workflow_executor.py:67)。

add_resumed_node_id(workflow_executor.py:949)则是运行期动态往这个集合里加节点用的,恢复任务里会用到(见下)。

5.4 第三步:把已完成节点的输出灌回去​

恢复时绝不能重跑已经算完的节点(浪费算力,还可能有副作用如再次发邮件)。做法是把它们的历史输出当作 precomputed_outputs(预计算输出) 灌进执行器——见到有预计算输出的节点,执行器直接拿来用、不再执行(这套通用机制见 02-executor.md)。

_update_executor_outputs(workflow_run.py:1079)先把 run 里存的历史 outputs 反序列化回执行器的 executor.outputs,并特别地为暂停节点重建一个扁平结构的 HumanInterventionNodeOutput,把人给的 inputs 合并进去(workflow_run.py:1109-1134),这样下游模板才能用 {{HumanInterventionNode_1.input_1}} 直接取到人填的值。

5.5 第四步:只跑被挡住的下游​

真正的续跑在 _create_resume_workflow_task 生成的后台任务 resume_workflow_task 里(workflow_run.py:1219)。它的逻辑正好和暂停半程对称:

  1. 算出要跑的节点 = 暂停节点的下游(get_blocked_nodes,workflow_run.py:1224),把它们从 PENDING 拉成 RUNNING。明确不包含暂停节点本身——它已经"完成"了(workflow_run.py:1242-1245)。
  2. 登记已恢复/已完成节点:把暂停节点 add_resumed_node_id(workflow_run.py:1249);再把所有 COMPLETED 任务也 add_resumed_node_id(workflow_run.py:1253-1257)——防止执行器给这些节点重新建任务。
  3. 组装 precomputed:把 executor.outputs 里非空的输出全 dump 进 precomputed,并确保暂停节点用的是"合并了人输入"的那份输出(workflow_run.py:1260-1271)。
  4. 续跑:executor.run(node_ids=下游, precomputed_outputs=precomputed)——只跑下游、复用一切已算好的(workflow_run.py:1274-1278)。
  5. 跑完把 run 标 COMPLETED;出错则标 FAILED(workflow_run.py:1289-1294)。
# 真实源码 workflow_run.py:1274-1278(续跑调用)
outputs = await executor.run(
    input={},                        # 初始输入首跑时已给过
    node_ids=list(nodes_to_run),     # 只跑被挡住的下游
    precomputed_outputs=precomputed, # 复用已完成节点 + 人的输入
)

5.6 完整状态机​

把两个半程接起来,单个节点在整个生命周期里的状态流转如下(暂停节点走上支,它的下游节点走下支):

            首跑                暂停被处理           人提交动作            续跑
 暂停节点:  RUNNING ──raise──▶ PAUSED ──────────▶ COMPLETED ────────▶ (视为已完成,不再跑)
                              (存 output)         (存人的 inputs)

 下游节点:  (未及运行/         PENDING  ───────────────────────────▶ RUNNING ──▶ COMPLETED
            或误判CANCELED)    ▲                                    (被 get_blocked_nodes
                              │ _fix_canceled_tasks_after_pause      选中并拉起)
                              │ / task_recorder 兜底 掰回来

 run 整体:  RUNNING ────────▶ PAUSED ──────────▶ RUNNING ─────────▶ COMPLETED(或 FAILED)

TaskStatus 全集见 models/task_model.py:19(PENDING/RUNNING/COMPLETED/FAILED/CANCELED/PAUSED);RunStatus 见 models/run_model.py:22,含 PENDING/RUNNING/COMPLETED/FAILED/PAUSED,以及 run_model.py:28 的 CANCELED——其注释写着 "Added for canceling workflows awaiting human approval"(为"取消一个正等人审批的工作流"而加)。也就是说,RunStatus 比正常路径上会走到的几个状态多一个"人主动取消"的终态;本章的暂停/恢复主线只用到 PAUSED → RUNNING → COMPLETED/FAILED,而 CANCELED 对应的是人不想恢复、直接把这条 run 作废的场景。

6. 巧妙之处(可借鉴的技术)​

  • 用异常做控制流信号,而非仅报错。 PauseError 借异常的天然冒泡能力,让一个叶子节点无需任何回调/信号总线就能冻住整个引擎;执行器只要单独 except 这一种类型即可(workflow_executor.py:644)。干净、零耦合。

  • "不可恢复终态"的严防死守。 把 CANCELED/FAILED 视为不可逆,于是围绕"暂停下游必须留 PENDING"布下多层防御 + 一层事后纠错 + 一层 TaskRecorder 兜底(task_recorder.py:172)。这是"想清楚哪些状态不能进,然后处处拦截"的典范。

  • 恢复靠"复用输出 + 标记已恢复"而非"记录进度指针"。 没有 program counter,没有序列化的执行栈。恢复时只需两个集合:resumed_node_ids(哪些别再拦)+ precomputed_outputs(哪些别再算)。状态全在数据库里的任务表,执行器是无状态可重建的——这让恢复能在一个全新进程里发生。

  • BFS 反转依赖图求下游(get_blocked_nodes, workflow_executor.py:174):同一个函数在暂停半程(找谁要 PENDING)和恢复半程(找谁要拉起 RUNNING)两头复用,对称利落。

7. 边界与局限(诚实)​

  • 三种 PauseAction 在续跑逻辑上未真正分叉。 如 §3.3 所述,action(APPROVE/DECLINE/OVERRIDE)被存下来用于审计,但本 commit 的恢复代码并没有据此走出不同分支——DECLINE 并不会自动中止下游、OVERRIDE 与 APPROVE 走同一条"用 inputs 覆盖输出"的路。真正的差异化行为得靠工作流设计者在下游节点里自己读这些值来实现。

  • 恢复是"重建执行器"而非"复活原进程"。 依赖任务状态完整落库。若数据库里的任务/输出记录不一致,恢复会踩空(代码里 _get_and_validate_paused_run 会校验必须有 PAUSED 任务,workflow_run.py:945)。

  • 恢复在 FastAPI BackgroundTasks 里跑(workflow_run.py:1180),失败只是把 run 标 FAILED 并 log(workflow_run.py:1291-1294),没有自动重试。

  • 暂停粒度到"节点的下游"。 靠 block_only_dependent_nodes 配置和 get_blocked_nodes 的 BFS 决定挡谁;并行的兄弟分支(如 §4.3 里的 D)不受影响、会继续跑完。

  • 任务/run 状态如何持久化的细节(TaskRecorder 的落库、字段结构)本章只用到接口,完整机制引到 05。

8. 代码地图(导航索引)​

主题文件路径符号名
暂停信号异常backend/pyspur/nodes/logic/human_intervention.py:10PauseError
恢复动作枚举backend/pyspur/nodes/logic/human_intervention.py:25PauseAction
触发暂停的节点backend/pyspur/nodes/logic/human_intervention.py:86HumanInterventionNode.run
节点执行处捕获暂停backend/pyspur/execution/workflow_executor.py:644_execute_node (except PauseError)
暂停处理总指挥backend/pyspur/execution/workflow_executor.py:360_handle_pause_exception
标暂停节点backend/pyspur/execution/workflow_executor.py:298_mark_node_as_paused
标下游为 PENDINGbackend/pyspur/execution/workflow_executor.py:314_mark_downstream_nodes_as_pending
BFS 求下游backend/pyspur/execution/workflow_executor.py:174get_blocked_nodes
判断是否暂停下游backend/pyspur/execution/workflow_executor.py:210is_downstream_of_pause
事后纠错 CANCELED→PENDINGbackend/pyspur/execution/workflow_executor.py:395_fix_canceled_tasks_after_pause
挑出误判 CANCELED 的任务backend/pyspur/execution/workflow_executor.py:379_get_tasks_to_update
标 run 为 PAUSEDbackend/pyspur/execution/workflow_executor.py:339_update_run_status_to_paused
登记已恢复节点backend/pyspur/execution/workflow_executor.py:949add_resumed_node_id
TaskRecorder 兜底掰回 PENDINGbackend/pyspur/execution/task_recorder.py:172update_task (is_downstream_of_pause)
恢复流程编排入口backend/pyspur/api/workflow_run.py:1137process_pause_action
暂停任务转 COMPLETEDbackend/pyspur/api/workflow_run.py:986_update_paused_task
重建执行器(带 resumed)backend/pyspur/api/workflow_run.py:1026_setup_workflow_executor
灌回历史输出backend/pyspur/api/workflow_run.py:1079_update_executor_outputs
续跑后台任务backend/pyspur/api/workflow_run.py:1219resume_workflow_task
恢复 HTTP 端点backend/pyspur/api/workflow_management.py:73take_pause_action
TaskStatus 枚举backend/pyspur/models/task_model.py:19TaskStatus
RunStatus 枚举(含 CANCELED)backend/pyspur/models/run_model.py:22RunStatus