FlowWorker¶

FlowWorker（plaita.server.flow_worker，需 server extra）把 FlowExecution.run_distributed 与 ExecutionStorage / FlowStorage / EventBus 串起来：加载流程、持久化 checkpoint、处理 start/resume 任务。

把它当成 suspend/resume 编排器，而不是 Temporal/Cadence 级「容错工作流引擎」。API 名里的 Distributed 指「可跨进程挂起/恢复」。

可靠性边界（必读）¶

选型含义：

• 适合：审批回调、HTTP 回调、延迟唤醒等「挂起等待外部事件」、可接受重复投递（幂等 resume）的场景。
• 不适合：把「恰好一次」「自动故障转移」「金融级幂等」当默认承诺的场景——副作用仍须幂等。
• 崩溃恢复的如实语义（2026-10 更新，#23）：worker 崩溃后 pending 里的 start 任务被重投时，会创建全新执行从头重跑（新 execution_id，即从首节点起全部节点重跑——首节点必须幂等）——每步落盘的 checkpoint 不会被 start 任务消费；旧执行会停留在 running 状态，需要运维侧用 scripts/reap_zombie_executions.py 巡检清理（租约在则跳过、条件写落盘——单看 last_update_time 会误杀长节点）。存活持有者不会被重派双跑：start 消息被 XCLAIM 重派后，重派者因租约被原持有者持有而拿不到租约、消息被 ack 释放（#23）；只有持有者真死（租约过期）后重投才会从头重跑。resume 任务的重投是安全的：终态执行会被幂等短路（原样返回，不再推进，也不会被改写状态）。事件/超时 resume 的中间态重投（#258）：resume 在流内推进并落过 checkpoint 后（PERSIST_EVERY_N_STEPS=1），瞬态 StatePersistError 或进程崩溃会让同一条 resume_type=event/timeout 消息重投；此时盘上是 running + $LAST_NODE 非 pending 挂起点、订阅已注销，策略层会判协议错误。worker 在 ack 前做状态感知：仍可续跑就把载体归一化为 continue（从 checkpoint 续跑，已完成节点不重放；run() 兜底先入队新副本再 ack 原条，入队失败/读盘瞬断则不 ack、消息留 pending）；真协议错配（如 suspended + continue）与陈旧 resume 打在仍有活跃订阅的挂起点上仍按 #33/#221 拒绝并 ack。

CLI：--consumer-group、--consumer-name、--claim-min-idle-ms（默认 60000）、--lease-ttl-seconds（默认 120）、--max-deliveries（默认 5）、--dlq-key。--queue-name 为 Stream 键名（与旧 List 不兼容）。

配置代际与运行期配置（A2R31 / B2R31-01/02/03）¶

worker 启动时经 plaita.runtime_config.load_runtime_config("worker") 一次性解析 queue/redis/group/dlq/预算/安全开关，并计算三个互不替代的代际：

• config_family（cf1-…）：queue/group/dlq/schema/namespace 的硬兼容键；
• config_generation（cfg1-…）：全部语义字段的可观测哈希（不含 log level/metrics port）；
• budget_generation（bg1-…）：max_deliveries/lease TTL/载体年龄/DLQ 上限。

启动日志 event=config.effective 与 registry metadata / 容量块都带这三个值； GET /api/cluster/config 与 GET /api/cluster/instances 可核对混跑。

消费规则：

1. 首投与 reclaim 同判：两端都非 legacy 且 config_family 不同 → 不执行； reclaim 用 XRANGE 预读，不 XCLAIM、不烧 delivery，消息留 pending + event=config.family_mismatch。
2. budget_generation 不同 → 不把新预算作用旧 PEL：留 pending + event=config.budget_stale，要求 drain 或按源预算宽限。
3. 重投/交接/克隆/DLQ replay 继承源信封代际；缺失按 legacy，绝不盖当前值。
4. 历史消息（无字段）始终可执行；畸形字段 WARNING + legacy，不按当前值覆盖。

租户状态纪元：console 写入状态时同时刷新 plaita:tenant_status:epoch（来自 PLAITA_DEPLOYMENT_EPOCH）。worker 的 tenant_is_active(..., expected_epoch=...) 在 epoch 缺失/更旧时拒绝非 default 租户、default 保持 #237 fail-open；Redis epoch 更新时拒绝本 worker。运行中 Redis-only 恢复由 console 周期对账（默认 60s）或 POST /api/cluster/tenant-status/reconcile 纠正；PLAITA_TENANT_STATUS_EPOCH_STRICT=0 是回滚阀（启动 WARNING）。

非法值口径：identity/security/resource 类在进程入口聚合后 fail-fast； optional 类 WARNING + 安全默认；安全门未知值 fail-closed。空串 identity （queue/redis/dlq/group/instance id/delay queue/epoch）一律拒绝，不再写空键。 详见 运维 Runbook 的非法值矩阵。

长步骤与消息回收（claim_min_idle_ms / XCLAIM）¶

一条任务从被 XREADGROUP 读到 XACK 前一直留在 consumer group 的 pending 列表。单个步骤执行超过 --claim-min-idle-ms（默认 60000ms）后，这条 「仍在处理中」的消息就对其他 consumer 的 XCLAIM 回收可见——回收本身不是 故障，双跑由执行租约拦截（start 与 resume 同一套租约，#23）：

1. 抢到消息的 worker resume 时拿不到租约 → ExecutionLeaseError。冲突分支 先核实持有者存活再决定 ack（#50，2026-10-08）：租约值嵌有持有者的 注册表 instance id，回收方查 plaita:registry:flow_worker:{instance}—— 在册且 status=running（活持有者，正常竞争）→ ack 释放（否则对端每 claim_min_idle_ms XCLAIM 一次烧一次 delivery，超限即假死信）；不在册 （持有者已死、租约键是 ≤TTL 的尾巴）→ 不 ack 留 pending，等租约过期后 下一轮回收从 checkpoint 续跑（代价 ≤TTL+60s，远小于 zombie reap）；在册 但处于 draining/stopping/stopped/starting（S18-02A）→ 同样不 ack： 这些状态不保证收尾，ack 会释放执行的唯一重投载体。信号缺失 （旧格式租约 / Redis 瞬断）按「存活未知」退化 ack；未启用注册表 （--no-registry）例外（B65-03，2026-10-20）：该形态下存活信号静态 不可得，把「存活未知」当「活持有者」会在持有者被冻结（SIGSTOP/GC/ 调度饥饿，租约仍有效）时 ack 掉唯一消息——冻结者恢复后 lease_lost 退出，执行停在 running 且无载体；故该分支一律不 ack，留 pending 等租约过期后 reclaim（租约在时死信守卫拒绝死信，不会回归假死信风暴）； PLAITA_DISABLE_HOLDER_LIVENESS=1 整体旁路回到无条件 ack； 决议载体例外（A46-04/A47-F2，2026-10-19）：resume_type 为 event/timeout/cancel 的消息不看存活信号、一律不 ack——suspend 提交窗口 [订阅注册, 租约释放) 内到达的它们不是重复载体，而是挂起执行 唯一的唤醒来源（#33）；持有者落盘 suspended 后不会再读它。消息留 pending， 租约释放后由 reclaim（或 #242 死信守卫重入队副本）推进；代价是最多 max_deliveries × claim_min_idle_ms 的延迟与 XCLAIM churn。 continue/retry/start 无推进语义，仍按上面的存活判据 ack；
2. 持租约的活 worker 由看门狗每 lease TTL/3（默认 120s → 40s）续租，步骤 执行期间租约不会过期——XCLAIM 真正接手的只有已死 worker 的消息；续租成功 时顺带把在跑节点的进度写进 node_timings（长节点期间唯一的活性心跳， 见下节「在跑节点心跳」）；
3. 退化路径：PLAITA_DISABLE_LEASE_WATCHDOG=1 且单步超过 lease TTL 时， 租约可能在步骤中途过期、接管者拿到更新的 fence 世代——旧 worker 的 下一次落盘被 fencing CAS 拒绝、步界续租失败自爆（均 ExecutionLeaseError 且不 ack），状态不会被双写，但当前步副作用可能重复，业务侧仍须幂等。

因此调小 claim_min_idle_ms 只加快「死 worker 消息」的回收，不会中断活 worker 的执行；调大到超过最长步骤耗时，可减少 lease_conflicts 噪音。 对 start 同样成立：无租约时代的「数小时 run 消费 60s 后被空闲 worker 重派、整条 flow 从头双跑」已被租约窗口根除（#23）；持有者死亡后重派消息 照常取得租约接管——start 的接管点是从头重跑（checkpoint 由后续 resume 消费），租约只保证「同一时刻至多一个推进者」。

死信守卫：超过 --max-deliveries 的消息进 DLQ 前，先查消息体 execution_id 的执行状态与 resume 租约（键 {ns}:execution:lease:{id}，ns 按消息体 tenant_id 路由：default/空 = plaita，其余 = plaita:{tenant_id}）—— 执行已终态直接放行；租约仍在（活 worker 正处理长步骤）则跳过死信、消息留 pending；执行仍非终态但节点重试计数已达预算（见下节）则终态化 error 后 放行（不再重入队——重入队会让 delivery 归 1 再耗尽，无限循环）；租约/状态 查询失败同样保守跳过；非终态且租约空（持有者已死）重入队一份 delivery 归 1 的恢复消息后放行。

死信路径收敛（#252）：run() 兜底异常分支 （ExecutionStateLoadError / StatePersistError 等瞬态失败的落点）、 _reclaim_one 的超限处置、超 max_schema_version 的信封拒收（首投与回收 同判）以及畸形 payload 处置，走同一 dead_letter_guard 判定（worker 侧经 队列层 dead_letter_if_allowed 单一入口）：守卫拒绝/抛异常时不 ack，消息 留 pending 等下一轮处置并推进 dlq_guard_skipped；守卫为 None（默认）恒 放行，存量行为不变。DLQ 信封固定带 schema_version（畸形消息另带 raw_fields），审计与重放不靠猜。

超版本拒收不触发死信守卫的「克隆恢复」（A49-01）：守卫在「执行非终态 + 租约空」时会重入队一份恢复副本再放行原条死信，但该恢复路径对版本拒收不 成立——副本继承同一个 schema_version（A44-02），本 build 仍然无法消费， 下一轮读到副本又会拒收并再次克隆，形成流内恒 1 条、DLQ 每轮 +1 的自持热环 （回滚 / 生产者先行 / 伪造信封）。队列在拒收时给任务打 schema_rejected 标记， FlowWorker 的守卫对该标记不克隆、不终态化，只放行死信一次（留证 + 告警）； 执行保持原状（suspended 仍可被 event/cancel/timeout 唤醒），运维升级后可经 DLQ 重放入口恢复载体。租约仍在（活 worker 正处理）时仍拒绝死信，消息留 pending 等新 build。非版本拒收（如 max_deliveries）的恢复克隆语义不变。

DLQ replay 契约（A53-S2）：人工回灌统一走 plaita/server/dlq_replay.py （console 端点 POST /api/queues/dlq/{dlq_id}/replay，平台管理员专属）， 不要直接 enqueue_task：

• 以 (dlq_key, dlq_id) 写一次性标记（plaita:dlq:replay:{key}:{id}，7 天）； 第二次回灌返回 already_replayed，不发第二条消息；
• schema_version 继承源信封（缺失 = v1）；超过本 build 理解范围时在入队前 具名拒绝 schema_too_new，不制造「入队即再死信」的副本；
• 无 execution_id 的 start 载体铸造确定性 id（uuid5(dlq_key, dlq_id)）并补 dedup_key：标记过期/force 重灌仍收敛到同一条执行；
• 执行墓碑命中 → execution_deleted（不复活已删除执行）；缺 tenant_id 必须由 平台显式指定，payload 租户与显式租户冲突 → tenant_mismatch；
• DLQ 原条目不删除；每次尝试写 plaita:dlq:replay:audit。保留上限 PLAITA_DLQ_MAX_LEN 是有损 FIFO：裁剪会丢弃未处理证据，dlq_trimmed / dlq_trim_failures 指标与 WARNING 可见。

版本边界对首投与回收同判：_reclaim_one 在 XCLAIM 之后、返回 消费循环之前做与 xreadgroup(">") 相同的 schema_version > max_schema_version 校验。消费者在 schema 检查/死信 ack 前崩溃时，同一条消息不会因「重启后走 reclaim」就从死信留证变成旧语义硬跑。

worker 作为生产者的版本继承：worker 的重投/交接/归一化副本 （start 幂等命中的 resume、节点可重试失败、亲和交接、死信守卫恢复、超配额 延迟、#258 载体归一化）继承源消息的 schema_version，而不是无条件盖 当前 build 的写版本。混跑池里新实例写出的恢复副本仍能被旧实例接管；只有新 生产消息（console BFF / 调度 / event_filter）才写当前 TASK_SCHEMA_VERSION。 因此 worker 自身的灰度不再要求「整池原子升级」，但这些新生产者服务仍须按 「先升消费者、再升生产者」的顺序发布（见 Runbook）。

--max-deliveries 计的是处理尝试次数：xreadgroup(">") 交付的首投 （delivery_count=1）此前零处理尝试，绝不会被预判死信——--max-deliveries 1 仍允许这一次尝试，只有该次尝试失败/未 ack 后（worker 兜底分支或 reclaim） 才进 DLQ。

部署步骤与故障手册见 运维 Runbook；副作用设计见 幂等 Resume。

claim 前的本机盘预检（#49）¶

问题（2026-10-08 实测）：多 worker 池里，「本机盘够不够」这件事此前只有 flow 内部的 preflight 节点在管（跑在领取任务的宿主上：os.statvfs(repo) → free_gib < 阈值 → 返回 retry-later）。而 worker 是「谁先 XREADGROUP 谁拿走」，于是一夜出现 11 次连续白跑：低盘机先抢到任务 → flow preflight 判 盘不足 → retry-later 回评 + 重排；同时刻富盘机完全空闲（队列 XLEN=0）。 更糟的是 keeper 的 retry-later 退避会升档——瞬时低盘被放大成 ≈6h 停机。

修法：把「盘够不够」的判断提到 claim 之前，由 worker 自己做：

python -m plaita.server.flow_worker \
  --min-free-disk-gib 20 \          # 或 PLAITA_WORKER_MIN_FREE_DISK_GIB=20
  --disk-guard-path /home/user/repo # 或 PLAITA_WORKER_DISK_GUARD_PATH（默认 .）

语义与边界：

• 阈值 0（默认）= 关闭，与 PLAITA_WORKER_DENY_REPOS 同口径——存量部署 零行为变化，是否启用由运营侧显式配置；
• 低于守线时根本不发 XREADGROUP（也不 XCLAIM 回收别人的 pending）： 不产生 claim、不烧 delivery、不写 retry-later 回评，任务自然落到池内 富盘 worker 手里；
• 空转重探默认 15s（disk_guard_poll_seconds），休眠切成 ≤0.2s 小片， SIGTERM/request_drain() 可即时打断；盘回线自动恢复领取，无需重启；
• 预检路径要与 flow 内 preflight 节点的判据对齐（recursive 侧为 RECURSIVE_MIN_FREE_DISK_GIB，默认 20）：worker 侧阈值低于 flow 侧时仍会 白跑一轮，worker 侧高于 flow 侧则会过早停领（池内可能整体闲置）；
• 探测失败（路径不存在 / 无权限）放行：预检自身故障不得让整池停摆， 宁可回到「可能白跑一次」的旧行为；
• 该闸只管领取：已在跑（或本进程 pending 里）的任务照常跑完/重投， draining、租约、死信守卫等语义一律不变。

配套的运营动作见 运维 Runbook · 按宿主资源路由。

节点级有界重试（2026-10 二波）¶

分布式路径上节点执行失败（LLM/HTTP 网络抖一次）不再直接废掉整个执行：

• 判别：run_distributed 把异常归一化为 FlowErrorException（原始异常在 __cause__）。链中出现 NodeExecutionError（节点执行异常）→ 可重试； 但其 __cause__ 里的原始节点异常按 node-F2 同一分类复核（A50-03）： 1001 / 非校验 ValueError 是确定性错误，不重试（否则 abort 包装层会把 core 的「执行 1 次」重新判成可重试，同一错误仍被 worker 重投 5 次）；#219 输出 校验异常仍可重试；1002/1003 与未知码维持可重试； 链中出现超时（NodeTimeoutError/FlowTimeoutError，含执行级 ExecutionDeadlineError——它是 FlowTimeoutError 子类，#267）或取消 （FlowCancelledException）→ 维持现状终态化 error；deadline 到期不会重投 （_error_fields 保留 code=-1 / flow_error，可从执行状态收尸）； 恢复守卫类 ResumeGuardError——策略层在分发到节点 resume() 之前 逐个标记出来的那几个位点（continue/retry 想绕过 pending 挂起节点、resume 打在 无挂起/非挂起 checkpoint 上、给出非 continue 的 resume 意图却没有 checkpoint）， 语义是「调用方协议与执行状态不匹配」而非「执行失败」（#33）→ 不终态化： 执行保持原状并抛 ResumeProtocolError，run() 对其 ack（重投只会重复命中同一 守卫。这类错误天然由 at-least-once 重投产生——重复投递的 event resume 打在 已推进的 checkpoint 上同族——把挂起执行终态化成 error 会永久切断 event/cancel/timeout 的唤醒路径）。裸 ResumeError 不在豁免面内（含未标记的 Unsupported resume type for EventNode）： strategies._handle_resume 把 current_node.resume() 抛出的任何异常 （事件数据畸形、节点恢复逻辑失败…）包成的 ResumeError 是执行自身失败， 静默保持原状只会变成一个「无 error 记录、永远等不到决议」的哑执行——它走 下面的现状路径（可重试判据 → 终态化 error + poison ack），可观测、可人工 retry；其余图错误（NodeNotFoundError 等）→ 维持现状终态化 error。 超时不重试是刻意的：确定性信号重试=再烧一次全款。
• 载体：at-least-once 消息重投本身。重试时执行不终态化（磁盘 state 停在最后成功步 checkpoint——失败节点不写 context），run() 显式重投新副本 （先入队再 ack；重投前按重试次数指数退避：min(30s, 1s×2^(n-1))，即 1s/2s/4s/8s…封顶 30s——plaita#91：旧实现「入队即重投」无任何节奏，5 次预算 ~40ms 烧光，持续数秒的上游抖动本可自愈却被直接打成 error 终态。退避睡在消费 线程内，单 worker 会推迟同队列其他任务 ≤30s；队列级延迟投递（ZSET/延迟 Stream）是演进项。cap 30s < claim_min_idle_ms（默认 60s），睡窗内原条 pending 不会被对端 XCLAIM 抢走）。resume 从 checkpoint 自然重跑失败节点。 start 载体的重投带 resume 语义（plaita#91）：盘上已有同一 execution_id 的状态行时，重投副本转换为 {"type": "resume", "resume_type": "continue"} 载体（与 start 幂等命中 _handle_start_dedup_hit 的转换同形状，非空 params 随载体透传），从 checkpoint 续跑——不从首节点整条重放、不用 context={} 覆写盘上 checkpoint；盘上无行（首持久化前崩溃）、flow_id 与 盘上行不一致或读盘失败时才原样重投 start。flow_id 不一致的 start 载体 由 start_flow 查盘后具名拒绝（ExecutionIdConflictError，台账第三十一轮 V1）：同一 execution_id 已属于另一条 flow 时既不能转 resume（会被流程 ID 校验打成 poison ack），也不能按首投新建 context={} 行（会永久覆写原 flow 的 checkpoint）——拒绝并保持原状态，消息按畸形载体 poison ack。重投路径不依赖 pending 回收： 消息可能已被竞争者的 ExecutionLeaseError 分支 ack 释放，只留 pending 会导致重投永不发生（2026-10-09 plaita#52 实证）。 重投失败（Redis 抖动）才退化为不 ack 留 pending；两种情况都不计 poison。
• 预算：重试计数键 {ns}:execution:noderetry:{id}:{node_id}（INCR + 滑动 7 天 EX，租户路由与租约键同规则），默认预算 = --max-deliveries（5）。耗尽 → 终态化 error（error.node_retries 记录重试次数、error.failed_node 记录失败 节点 id），消息走 DLQ。刻意不用消息 delivery_count 判预算：回收路径上报 的是 XCLAIM 前的投递数（少计 1），且达限消息在队列层就地死信、不进处理函数 ——按它判预算永不触发。 计数语义是「当前节点的连续失败次数」，键也必须按节点维度隔离 （plaita#123 真因，2026-10-10）：此前键只含 execution_id（执行维度），而成功 推进时把整个执行的计数清零 ⇒ 在「前序节点稳定成功 + 某节点稳定失败」的 flow 上，前序节点每轮成功都把失败节点的计数抹平 ⇒ 永远到不了预算 ⇒ 永不终态化 ⇒ 无限重投（实测 2026-10-10 13:0x 本机 Redis db1：keeper-watch 累积 667 个悬停 running 执行、最老 40 小时。该数字为当时快照、随时间增长， 勿当常量；复现需在同期环境用 SCAN plaita:execution:* 采集）。 现版：键带失败节点 id（取自异常链 NodeExecutionError.node，不能用 checkpoint 的 $LAST_NODE——失败节点不写 context，那是上一个成功节点），成功推进只清该 节点的键。取不到节点 id 时退化为执行维度键（保守，不比修复前差）。 存量扁平旧键（无节点维度）不迁移、不被新逻辑读取，靠 7 天 TTL 过期；代价是 升级瞬间在途执行多拿一轮预算。
• 兜底总上限（plaita#123 评审建议）：{ns}:execution:noderefetch:{id} （INCR + 7 天 EX，不含任何会漂移的段）——只按 execution_id 累计「本执行 一共放行过多少次节点重试」，达 NODE_RETRY_TOTAL_MAX（默认 100）即终态化 error。 节点维度预算依赖「键里的节点段稳定」，任何键空间漂移（节点 id 动态变化、定义被改） 都会让单节点预算失效而退回无界重投；该计数不看键名、只看次数，是「预算机制 整体失效」时的最后一道闸。正常执行永远碰不到（100 ≫ 单节点预算 5 × 合理节点数）。
• 与 G1 retry 唤醒的组合（43828aa）：预算耗尽终态化的执行仍可经人工 resume_type=retry 唤醒（error 态断点续跑）——唤醒放行即清零该执行下所有节点 的计数键，人工唤醒后拿全新预算；flow 定义指纹校验先于唤醒，定义被改时执行保持 error（修复定义后仍可再 retry）。
• 唤醒/失败次数是有界的（plaita#73，2026-10-10）：只清预算不限次数的循环是 自毁的（唤醒 → 清零预算 → 跑 5 次 → 又终态化 → 再唤醒，实测某执行一天被 唤醒 5 轮、每轮烧 5 次沙箱）。两道上限都落在 FlowWorker 类常量上， 达限时 resume_flow 对 retry 幂等拒绝（返回 already_terminal=True + g1_wakeups_exhausted / deterministic_failure_exhausted，不抛异常、不改 状态），消息层重投随之短路（仅对 error 态执行，见下节 running 执行的 retry）：
• {ns}:execution:g1wakeups:{id}（G1_MAX_WAKEUPS，默认 2）：同一执行被 retry 唤醒的次数。人工与自动共用这个额度——达限后运维点「从断点重试」 不再生效（BFF 会返回 409 并给出要删的键，见 ops-runbook）；
• {ns}:execution:nofail:{id}（DETERMINISTIC_FAILURE_MAX，默认 12）： 连续确定性失败次数（exited 1 / sync_in / 超时…）。任一节点成功推进 即清零，跨节点的偶发失败不会累计判死；G1 唤醒不清零（否则永不收敛）；
• 人工解封：确认失败原因已修好后 DEL 对应计数键再 retry——键名与步骤见 运维 Runbook · 唤醒预算达限。
• 引擎 execution_id 必须与 worker 一致（plaita#123 配套）：worker 的 resume 与步进两处 run_distributed 显式传 execution_id=execution_id。不传时引擎侧 execution_id 与 worker 脱钩——恢复分支走 context.context = saved_context 整体 替换 checkpoint，若 checkpoint 缺 $EXECUTION_ID（resume 常见），引擎 execution_id 退化为空串 ⇒ result.execution_id 回空、重试计数键沦为空 id （运维无法从键反查执行）。注意这不会导致无限重投：旧代码 _node_failure_retry_decision 对空串直接 return None（连键都不写、直接终态化）， 属可观测性缺陷而非重投根因——重投根因见下方「节点维度预算」。
• 节点维度预算（plaita#123 真因，2026-10-10）：计数键 {ns}:execution:noderetry:{id}:{node} 按节点隔离，成功推进只清该节点 的键。此前键只含 execution_id（执行维度），而成功推进清整个执行的键 ⇒ 在 「前序节点稳定成功 + 某节点稳定失败」的 flow 上，同一轮内前序节点成功先清零、 失败节点随后 INCR 归 1（strategies.py："Execute one node per call"，一次调用 只推进一个节点），计数永远到不了预算 ⇒ 永不终态化 ⇒ 无限重投 （实测 2026-10-10 13:0x 本机 Redis db1：keeper-watch 累积 667 个悬停 running 执行、最老 40 小时；该数字为当时快照、随时间增长，勿当常量； 复现需在同期环境用 SCAN plaita:execution:* 采集）。 另设兜底总上限 {ns}:execution:noderefetch:{id}（NODE_RETRY_TOTAL_MAX，默认 100），不看键名只看次数，防未来任何键空间漂移再次退化为无界重投。

挂起执行的 continue/retry 幂等短路（#33）¶

suspended 执行收到 resume_type=continue（挂起双写窗口 crash 后重投的原消息/ start 重派竞速/运维误发）或 retry 时，策略层 pending 守卫必拒（continue 不允许 绕过挂起节点）。resume_flow 在租约之前幂等短路：返回 already_suspended=True，不推进、不改状态、不取租约——挂起执行只能被真正的 决议路径（event/cancel/timeout）唤醒。此前的行为是守卫 ResumeError 被通用 except 终态化成 error，一次重复投递就把可恢复的挂起执行永久打封（终态短路从此 拒绝一切 resume 类型，retry 也不可入，死局）。兜底路径（绕过入口短路的竞态窗口） 由 ResumeProtocolError 豁免承接：只有策略层守卫 ResumeGuardError 不终态化、 执行保持原状、消息 ack（豁免面刻意窄，见上节「判别」——节点 resume() 自身抛错 仍终态化 error）。

租约冲突窗口内的决议载体不 ack（A46-04 / A47-F2，2026-10-19）：suspend 提交窗口 [订阅注册, 租约释放) 内到达的 event/timeout/cancel 消息，即使 执行推进租约被在册 worker 持有，也不得按「重复载体」ack——持有者可能正在 落盘 suspended，释放租约后不会再读这条消息，而它是挂起执行唯一的唤醒来源。 worker 对这些载体一律不 ack（信号缺失也不退化 ack），消息留 pending，租约释放 后由 reclaim / #242 死信守卫副本推进。continue/retry 不携带推进语义，保持 原有的「持有者在册即 ack」语义；#258 的协议错配状态感知（ResumeProtocolError 分支）是另一个窗口，两套判据互不影响。

取消意图优先（台账第三十四轮 ④）：短路前先查取消标志键 （{ns}:execution:cancel:{id}）。挂起前控制面可能已在 running 阶段写下取消 意图；若此处照常早退，消息被 ack、盘上保持 suspended、flag 仍在——独立的 cancel 载体一旦丢失，执行会带着取消意图长期挂起。命中 flag 时不短路： 转正常路径取租约、在租约内复核并用 resume 入口既有的取消检查点终态化 cancelled（保留取消点前 checkpoint）。无取消意图时「不取租约」契约不变—— 只有确需终态化写盘时才取租约。

running 执行的 retry 与分支未命中终态化（#53）¶

工单背景：值守实测两例条件假分支后执行停 running 88 分钟（无 error、无终态）， 且对 running 执行 POST /resume {"resume_type":"retry"} 返回「已加入队列」却 一行不推进（node_timings / last_update_time 双双不动）= 假受理。

• 分支未命中必须显式失败，绝不悬死 running：if/switch 全部分支未命中 且无 default 时，plaita/core/strategies.py 的 _get_next_from_last （distributed）与 _advance_one（normal/generator）抛 FlowExecutionException → worker 把执行终态化为 error（error 里点名节点 id）， 不再带着 $NODE 中间态静默收尾、也不停在 running。if 无 else 的两种形态都 被这一道拦下：else_next=""（分支未命中）与 else_next="false"（if 节点的 占位默认值指向不存在的节点 → NodeNotFoundError）。唯一逃生口是节点显式 errorHandler.strategy=continue（降级为 warning + $NODE 收尾，与 0.5.x 历史 行为一致）。升级核对：线上 worker 必须含该防御——不含该防御的版本上，条件 假分支无后继就是「不派发 / 不落终态 / 永久 running」（工单「未证实」项）。
• running + retry = 断点续派：running 执行没有「失败节点」可重跑， retry 与 continue 同路，从 checkpoint 派发后继节点，并留一行 WARNING 说明「按从断点续跑处理」。两道唤醒预算幂等拒绝（g1wakeups / nofail） 只对 status=error 生效（plaita#53）：计数键寿命长于状态行（终态化写盘 失败、状态被回滚等都会留下「running + 计数达限」的组合），若把 running 也按「已终态」拒绝，返回的 already_terminal=True 就是谎报终态 + 一行不 推进——即工单里的假受理。
• resume 被丢弃必须可见：continue/retry resume 在推进租约被在册 worker 持有时按既有语义 ack 释放（#50/#52：留 pending 会烧 delivery 并产生 假死信），但会额外打 WARNING「本次 resume 未生效（不产生任何节点活动）」 并给出下一步（确认卡死后人工 cancel 再按需重开）——值守据此区分「已救活」 与「白等」。每次点「从断点重试」都返回 200 却毫无推进时，先 grep 这行日志。 event/timeout/cancel 决议载体不在此列：它们一律不 ack、留 pending 重投（A46-04/A47-F2，见「挂起执行的 continue/retry 幂等短路」一节）。

在跑节点心跳与 keeper 活性判据¶

执行状态的落盘只发生在节点边界（_persist_state_or_raise 的 step_persist / 终态等路径），而 sandbox_agent 这类单个长节点（实测 35 分钟）期间一次落盘 都不发生 → node_timings 停在进入该节点之前。keeper 的活性判据①「有 started_at、无 ended_at 的节点 = 活证据」于是读不到它，落到判据③「末节点 ended_at 停滞超 1800s」→ 健康 run 被误判 zombie 并 cancel（2026-10-10

27/#31/#32/#55/#62 五单事故）。¶

修法：看门狗续租成功后顺带落一次在跑节点进度（_publish_node_progress，默认 周期 = lease TTL/3 ≈ 40s）——只合并 node_timings（在跑节点写 ended_at: ""）并刷新 last_update_time，不推进流程、不改 context。 写侧的两道约束（2026-10-10 评审）：

• 与推进写串行：心跳是「load → 合并 → save」的整行写回，与推进线程的 _persist_state_or_raise 是同一条状态行上的两个写者。若终态写落在心跳的 load 与 save 之间，心跳会把它整行回滚成 running + 旧 checkpoint + 新 last_update_time——既不会被 keeper 回收（open node + 新鲜 last_update_time 正是它的「活着」判据），后续 resume 还会从旧 checkpoint 重放节点副作用。两者现持同一把每执行互斥锁（_state_write_lock）。
• 跨进程过 fencing：心跳跑在看门狗线程里，那里没有本执行的 fence 世代 ContextVar、也没有租户上下文——不带就会（a）写错租户 namespace（多租户下 心跳整条失效）、（b）退化成 FencedExecutionStorage 的裸写分支、绕开世代 CAS（租约已被新世代接管时旧心跳仍能改写新世代的行）。现在 fence 世代与租户 随租约登记一起带上，世代不符即拒绝写（ExecutionLeaseError，日志留痕）。

keeper 侧必须给判据①配年龄界：心跳把在跑节点写进 node_timings 后， 「open node」既可能是活 worker（每 ~40s 刷新一次 last_update_time），也可能是 被杀的 worker 留下的最后一份心跳——单看「有 started_at、无 ended_at」 会把后者永远读成「活着」（keeper 判据① 原样没有年龄上限）。读法应为 「open node 且 last_update_time 在 N×心跳周期内（建议 N=2~3）」，再叠加 租约键 / worker 注册表在册与否兜底；否则僵尸 run 既不进 reap 路径、也没人 收尾。

运行中改定义（flow 定义指纹，2026-10 二波）¶

worker 的流程定义 TTLCache 有 300s 窗口、console engine_sync 可直接覆盖 Redis 定义——挂起执行 resume 用 latest 版本定义时，若定义自启动后被改，图遍历会 找不到节点/走错分支且无告警。现版语义：

• start_flow 对实际加载执行的 Flow 计算指纹 （sha256 of model_dump(mode="json") + sort_keys 规范化 JSON）写入 ExecutionState.flow_hash；
• resume_flow 取得租约后比对：不一致 → 终态化 error（message 写明「flow 定义自启动后已变更，hash 不匹配，执行无法安全续跑」，附前后指纹）+ 消息 poison ack——故意选可观测的终态而不是重投风暴（定义被改是确定性不一致， 重投 N 次结果相同）；
• 老状态 flow_hash 为 None（旧版本 worker 写入）→ 跳过校验，零回归；
• 在租约内判定是为了与活 worker 的推进写串行化：不会把他人正持有租约推进中 的执行误终态化。

运维含义：改定义前应确认没有 running/suspended 的在途执行（console 按 flow_id 查询非终态执行）；确需强升时接受在途执行被终态化 error、由上游重提。

start 幂等键 dedup_key（2026-10 二波）¶

start 消息重投（worker 崩溃/保存失败）历史上会新建 execution_id 从头重跑， 首节点副作用双份。现版支持调用方显式声明幂等键：

• 消息体可选 dedup_key（console BFF POST /executions 请求体 additive 字段 同名透传）：worker 在首节点执行之前以 SET NX EX 7d 原子认领 {ns}:start-dedup:{key}，映射值 = G1 预铸的 execution_id（BFF start 时 铸造随消息透传）或就地铸造的 id——认领在先行落 running 行之前，重投/ 双开在任何新行落盘前即被拦截；
• 命中 → 读映射的执行状态，绝不二次 start：已终态 → 返回 already 形状； running（崩溃/重试搁浅）→ 重入队一份 resume 消息接续执行再 ack start； suspended → 只返回现状形状（挂起执行自有 delay/approval 的 resume 链路）；
• 孤儿映射（认领后首次执行从未落盘，如 crash 在 claim 与先行落行之间）→ 释放后重新认领、按新启动继续（首节点可能重跑——首节点须幂等仍是既有约定）； G1 先行落行后该窗口已收窄到极小；命中 running 的重入队 resume 对 context={} 的先行行同样正确（无 last_node_id → 从首节点步进）； 重入队的 resume 若与原 start 持有者并发，由执行租约串行化（#23）；
• 不传 dedup_key 则行为与存量完全一致；键必须调用方显式提供，worker 不做 body hash 自动键——同参数定时任务（cron 每小时跑同一 flow）会被误判 为重复启动而永不执行。键按租户隔离，7 天过期。
• 调度服务是唯一的自动加键方（#23 引入，#118 修正基准）：fire_schedule （cron 循环与 console「立即触发」共用）入队的键：
• cron → sched:{schedule_id}:due:{due_ms}，due_ms = 本次到期时点 next_run_at（epoch 毫秒）。同一个 due 无论被触发几次（状态回写失败 + 30s 互斥锁过期重触发、消息 XCLAIM 重派）都得到同一个键，重派/重投收敛到 同一 execution，不产生平行执行；不同 due 的键必然不同，下一周期不会被误吞。 旧版键取「触发墙钟秒」，同一个 due 的两次触发落在不同秒即成两个键，双跑。
• manual → sched:{schedule_id}:manual:{ns}-{seq}，每次调用唯一，与 cron 的 键空间不相交——手动「立即触发」是显式要求再跑一次，不得与 cron 同秒互相吞并。
• cron 且既无 due_at 也无可用 next_run_at 时降级为 sched:{schedule_id}:fired:{墙钟秒} 并 warning（调度循环会先自愈 next_run_at，该路径只在定义缺失且调用方未传 due 时可达）。 手动 POST /executions 仍不自动加键。

职责¶

• 从 FlowStorage 加载流程定义（带 TTL 缓存）
• 用 ExecutionStorage 保存/读取执行状态
• 推进 Distributed 流程：跑到挂起节点，或从断点恢复
• 与 EventFilter / 外延服务配合：外部事件入队后由 worker resume
• 贯穿所有步骤地保留用户回调（须复用同一 FlowExecution 实例）

构造¶

from plaita.server.flow_worker import FlowWorker
from plaita.storage.redis import RedisExecutionStorage, RedisFlowStorage  # 需 redis extra
from plaita.event.redis import RedisEventBus

worker = FlowWorker(
    execution_storage=RedisExecutionStorage(client=redis_client),
    flow_storage=RedisFlowStorage(client=redis_client),
    event_bus=RedisEventBus(client=redis_client),
    cache_size=100,
    cache_ttl=300,
    callback_handlers=[MyCallback()],
)

流程定义缓存¶

get_flow_definition(flow_id, version) 先查 TTLCache，未命中再查 FlowStorage 并回填，避免反复反序列化流程 JSON。缓存键为 flow_id:version（version 缺省时用 latest）。

推进与恢复¶

flowchart TD
    Start["队列任务 start/resume"] --> Load["加载流程定义
(缓存/存储)"]
    Load --> LoadState["从 ExecutionStorage 读 context"]
    LoadState --> Run["FlowExecution.run_distributed"]
    Run --> Suspend{"is_suspend?"}
    Suspend -- 是 --> Save["立即保存 context"]
    Save --> Wait["等待下一事件入队"]
    Suspend -- 否, is_end --> Done["标记 completed
立即保存"]
    Suspend -- 否, 还有节点 --> Maybe["每步落中间态
PERSIST_EVERY_N_STEPS=1"]
    Maybe --> Run

状态模型¶

ExecutionState（plaita.storage.base）记录一次执行的完整状态：

存储后端¶

生产路径（FlowWorker CLI / create_storage_component）仅支持同步契约后端：

ExecutionStorage 接口为同步方法：save_execution_state / load_execution_state / delete_execution_state / list_executions，并提供 serialize_state / deserialize_state（JSON）。

Redis 后端的 delete_execution_state 是删除即吊销：同一 pipeline 内删除状态键、列表索引成员、{ns}:execution:fence:{id} 当前世代键、{ns}:execution:lease:{id} 租约键，以及该执行的计数键（nofail / g1wakeups / noderefetch / noderetry（含 :{node} 子键）/ affinityhandover）。在途 worker 下一次落盘会因 fence 世代失配抛 ExecutionLeaseError，不会把已删执行的状态写回；计数键同删保证同一 execution_id 重建后预算从零开始。世代计数器 {ns}:execution:fencegen:{id} 刻意不删、无 TTL：try_acquire_fenced 只对它 INCR，删除后新 acquire 的世代严格大于旧世代，旧 worker 的迟到 save/delete 才会被 CAS 拒绝（A2R13 ABA 第三格）。FencedExecutionStorage.delete_execution_state 在调用方持世代号时先做 fence CAS——旧世代不能删除新世代已推进的状态。

删除墓碑（A51-01/A52-01/A69-04）：同一 pipeline（fenced 档同一段 Lua）还会写平台根键 plaita:exec_tombstone:{execution_id}，值含 tenant / deleted_at / reason。删除 API 返回后，任何仍在队列/PEL 里的 start 载体都不再被 start_flow 的「盘上无状态 = 全新执行」分支重建：worker 在派发闸（配额之前）、start_flow 直达调用、start 幂等命中、节点失败重投与死信守卫克隆前复核墓碑（同租户才命中；跨租户同 id 放行）。新 execution_id 不受影响；同 id 重放被 ack 丢弃并释放 dedup 映射。墓碑 TTL 独立默认 30 天：max(30d, PLAITA_EXECUTION_STATE_TTL_DAYS) 只作「不短于 state TTL」的下限，STATE_TTL=0 不再产生永久墓碑；PLAITA_EXECUTION_TOMBSTONE_TTL_SECONDS<=0 视为非法回退默认 + WARNING（不再用 0 隐式表达永久），并 clamp 到 PLAITA_TASK_MAX_AGE_SECONDS。二次删除用 SET NX + 只延长 TTL，保留首次 deleted_at（升级前的存量永久墓碑不自动缩短）。墓碑过期后的历史载体由年龄闸丢弃：年龄取 payload timestamp 与 Redis Stream ID 时间的最大值（未来时间戳或重投后的新 Stream ID 都不能洗白源年龄），两者都不可判时 fail-closed（WARNING + /metrics carrier_age_unknown_total）；PLAITA_TASK_MAX_AGE_SECONDS 默认 30 天，0 关闭（启动 WARNING）。重投/交接/克隆副本保留源年龄（取 payload timestamp 与源 Stream ID 时间中更早者；payload 畸形/超前时以源 Stream ID 为准）；DLQ replay 缺失 timestamp 时用 dead_lettered_at 或 DLQ Stream ID，绝不刷新成 now。租户清理（_purge_tenant_redis）不触碰平台根墓碑，因此租户删除+同名重建后旧 start 仍不会执行。残留（B52-02 horizon）：攻击者在删除之后同 id 投递并压在队列/PEL 里超过墓碑 TTL，墓碑过期后仍可能重建——需租户/队列世代裁决，不在本条闭合范围。

恢复路径的墓碑/世代语义（A2R25-04/05）：墓碑同样拦 resume 载体——删除后从备份只恢复状态键不会撤销删除。resume_flow 在 load 之后、取租约之前（以及租约内复核）命中墓碑即具名返回 deleted，绝不 run_distributed；派发闸对 resume 与 start 同口径丢弃。fenced save/delete 的 CAS 还要求 fence 与 fencegen 都等于本次世代：只恢复旧 state+fence 而 fencegen 仍为新值时，陈旧 worker 抛 ExecutionLeaseError（WARNING 留痕），不会重新获得写/删权。

恢复后的列表可见性（A2R25-07）：load_execution_state 读到状态键后会检查索引成员，缺失即补回（恢复/外部写入只 SET 状态键的自愈路径，WARNING 留痕）；需要全量兜底时用 RedisExecutionStorage.rebuild_execution_index() 或 plaita-console/scripts/reindex_executions.py --tenant <slug> 重建索引（O(N) SCAN，仅在恢复现场手工调用）。

plaita.storage.sqlalchemy 仍存在，但其方法为 async def，与上述同步契约及 Worker 调用方式不兼容。公开路径已拒绝 db 作为 execution/flow 存储；详见根目录 MIGRATION.md「Storage：db 执行/流程存储从公开路径下架」。

与 ServiceManager 协作¶

FlowWorker 负责执行流程，ServiceManager 负责监听外部触发源（定时器、队列、HTTP 回调、审批系统）并 publish 事件。二者通过 EventBus 解耦。完整闭环见 外延服务。

下一步¶

• 扩展节点 —— 在流程里声明等待
• 外延服务 —— 在流程外触发恢复
• API: plaita.server.flow_worker

checkpoint 成本警示（2026-09 性能实测）¶

$NODE 保留每个节点的结果，且每步 checkpoint 对全量上下文重新 JSON 序列化——落盘成本随步数 二次方增长，随 payload 大小线性放大：

节点执行本身仅 2-13 ms/步：大 payload 下落盘开销是节点执行的 17-24 倍。 缓解：让大 payload 只在被消费时进入上下文（避免透传 1MB+ 的中间结果存进 $NODE）；调大 PERSIST_EVERY_N_STEPS；长流程拆分为 child/子执行。 增量式 checkpoint（每节点结果独立 key）在路线图上，当前版本请按上述方式规避。

启动期服务注册失败的重试（A2R19-03）¶

服务注册（plaita:registry:flow_worker:<instance> + 心跳线程）与队列消费走的是 两条独立恢复路径：队列读对瞬时 Redis 故障退避重试、恢复后自动继续；而 register_service() 只在 run() 启动时调用一次，失败（内部吞异常、返回 False）后不会重来。结果是「worker 启动瞬间 Redis 不可达 → Redis 恢复后队列 能消费，但注册表永远为空」——/api/services、容量快照、亲和派发 （target_instance_id）与以注册表为判据的就绪探针都看不到该进程。

现在启动注册失败会进入后台有界退避重试（1s → 2s → … → 30s 封顶，直到 成功或 stop()），成功后由 register_service() 顺带启动心跳线程； stop() 先停重试线程再注销，避免退出后又被写回。失败每次仍由注册表打 ERROR，退避封顶保证 Redis 全断时不刷屏；进程内 registry_registered 属性在成功前保持 False（成功前不宣称 ready）。测试见 tests/unit/test_issue_a2r19_registry_retry.py。

注册表状态机与派发资格（S18-02A）¶

注册键的 status 是唯一派发白名单，不是普通字段：

• 只有 running 参与派发：starting/draining/stopping/stopped 与未知值 在容量判定之前就被拒，即使上报了完美容量；running 缺容量/陈旧容量仍按 红线放行（不因上报故障停摆）。
• 启动期 drain 保留（S9）：register() 只把 starting 迁到 running； 若 SIGTERM 早于 run()（已是 draining），注册不会覆盖回 running。 终态 stopped/closed 拒绝注册（不复活）。
• 即时落盘（A45-03）：request_drain() / stop() 在内存置位后立即写一次 注册键（短退避重试 2 次），不再等 10s 心跳；失败打 ERROR 并计入 registry_status_flush_failed_total 指标。
• closed 世代闸（A46-03）：stop() 写完 stopping、注销前置 closed； 之后心跳、容量上报、_bump_active(±1)、注册重试全部不再写 Redis （在跑数仍更新内存，供 metrics/join 使用）。
• 注销可观测（A46-01）：先写 stopped 墓碑（TTL 5s）再 delete；delete 失败时键仍是 stopped（不可派）；get 失败仍继续尝试 delete；任一步失败 stop() 日志含 registry_unregister_failed。
• 目标不可用回退（A46-02）：非目标 worker 交接前复核目标注册键；目标 不存在或 status != running → 剥掉消息体 target_instance_id 按广播重投 （dispatch-target-unavailable:broadcast），避免集群每台机器各交接一次后 AffinityHandoverExhausted；Redis 读失败保留 target（fail-open）。

测试：tests/unit/test_r50_registry_state_machine.py、 tests/unit/test_issue42_dispatch_target.py。

drain 状态机与 ack 策略（S18-01，2026-10-19）¶

不变量（与代码逐条对应）：

• I1 post-read gate：D1 之后，任何「drain 前已发出、drain 后才返回」的 read/reclaim 结果都不 dispatch、不 ack、不 bump，消息留在 PEL 由其他 worker 在 claim_min_idle_ms 后 XCLAIM（drain-read-gate WARNING）。
• I2 advisory timeout：PLAITA_WORKER_DRAIN_TIMEOUT 是停止领取/停止等待的 上界，不是节点中止的上界。_force_stop_after_drain 无论 active 是否为 0 都置 _running=False（旧实现的 active<=0 早退会让已发出的 read 继续 dispatch）。
• I3 未完成不 ack：只有 _dispatch_task 已正常返回的消息才 ack；进程在步 返回前退出 → 消息留 pending，由 XCLAIM 重放。若步在 D2 后、进程退出前完成， 允许 ack（工作已完成）。
• I4 D3 后不再产生新步：stop() 置 D2 标记后，退避/消费线程不得再 enqueue/ack。
• I5 可中断退避：节点可重试失败的指数退避按 0.2s 分片，每片检查 D2 标记； 只有强停打断（仅 D1 仍让在途步完整跑完）。打断后不 enqueue、不 ack， 消息留 pending，日志 retry-backoff-aborted-by-drain。

交接（亲和闸 TaskNotForThisWorker）同样是 enqueue→ack：先写同体新副本， 成功后才 ack 原条；enqueue 失败 → 原条留 pending（绝不先 ack 造成唯一载体消失）； ack 失败 → 新副本已在队列、原条仍 pending，重复由租约/终态短路吸收。

库模式：run(wait_for_inflight=True) / stop(wait=True) 会在停看门狗与 注销之前有界等待在途任务收尾；CLI 默认不等待，以进程退出为硬中止边界 （daemon 消费线程可能仍在跑，未完成步的消息留 pending）。

停机与运维参数（2026-09）¶

• 优雅停机：SIGTERM/SIGINT 后 worker 在当前任务完成后退出。消费阻塞窗口 上限由 --read-block-ms（默认 1000ms）控制——XREADGROUP 的 BLOCK 无法被 信号中断，窗口越大停机延迟越长；默认值保证停机延迟 ≲1s。drain 的完整状态机、 advisory 超时与 ack 策略见上节 drain 状态机与 ack 策略。
• CLI 日志：python -m plaita.server.flow_worker 默认输出 INFO 级控制台 日志（PLAITA_LOG_LEVEL 可调，--quiet 关闭）。
• 本机盘守线（#49）：--min-free-disk-gib（或 PLAITA_WORKER_MIN_FREE_DISK_GIB）

  0 时，claim 前先查 --disk-guard-path（或 PLAITA_WORKER_DISK_GUARD_PATH， 默认工作目录）的可用盘，低于守线就不领任务。默认 0 = 关闭； 详见 claim 前的本机盘预检。

• Langfuse 观测：--langfuse（或 PLAITA_WORKER_LANGFUSE=1）启用 LangfuseCallback （需 pip install plaita[langfuse]，凭据走 LANGFUSE_* 环境变量）。trace id = 运行时 execution_id（随 checkpoint 持久化，跨进程 resume 续写同一 trace）； 依赖缺失或 SDK 初始化失败只告警降级，不影响执行。
• writefile 写入 jail（2026-10）：worker 启动即注入 writefile 节点的 PLAITA_NODES_WORKSPACE_ROOT（未显式配置则 fail-closed 推导默认根，/ 不算 边界；候选落在引擎自身 checkout 内则上溯到其父目录 = 部署根，跨仓 run 的产物 才写得出去，见 plaita#51）； 显式放行任意路径用 PLAITA_ALLOW_UNRESTRICTED_WRITES=1（仅单机信任 部署）。见 运维 Runbook · writefile 写入 jail。
• HTTP 节点 SSRF 防护（plaita#31）：PLAITA_HTTP_BLOCK_PRIVATE=1 让运营者强制 所有 http 节点屏蔽回环/内网/链路本地目标（节点声明的 blockPrivateNetworks 只能更严不能更松）；策略激活时建连解析经同一策略校验并只连校验过的 IP（关闭 DNS rebinding TOCTOU）。响应体默认上限 10MiB（PLAITA_HTTP_MAX_RESPONSE_BYTES）。 见 运维 Runbook · HTTP 节点 SSRF 防护。
• code 沙箱后端白名单（plaita#22）：worker 注册 code 节点时施加后端白名单 （PLAITA_SANDBOX_ALLOWED_BACKENDS，未配置则 docker ∪ 生效后端；生效后端取 PLAITA_CODE_BACKEND，默认 subprocess）。流程 JSON 声明的 sandbox_backend 落在白名单外即解析期拒绝——否则流程作者可逐节点把后端降级为 unsafe （进程内 raw exec，宿主任意代码执行）。配置后端不可用时具名拒绝注册并停止 启动（A56-01 fail-closed），不自动降级、不扩白名单；需要可用性降级须显式设置 PLAITA_ALLOW_SUBPROCESS_FALLBACK=1 且白名单含 subprocess。见 运维 Runbook · code 沙箱后端白名单。
• code 语言白名单（plaita#29）：worker 同时施加语言白名单 （PLAITA_SANDBOX_ALLOWED_LANGUAGES，未配置则只放行 python）。language: "js" 此前绕开整个档位体系（无隔离 / 无超时 / 无取消），故改为须显式放行；放行后 js 仍受 后端白名单与「语言×档位」约束（restricted 没有 js 实现）。见 运维 Runbook · code 语言白名单。
• 指标与告警（plaita#26）：PLAITA_METRICS_PORT（或 --metrics-port）> 0 时 本进程起 Prometheus /metrics 抓取端，导出队列 stream_length/pending/ dlq_length、全部进程内计数器与 worker 存活/心跳；死信经 RedisStreamTaskQueue(on_dead_letter=...) 钩子外发，PLAITA_ALERT_WEBHOOK 配置后即为 JSON POST（有界队列 + 后台线程，best-effort，不反压消费）。见 运维 Runbook · 指标与告警。
• event_filter 的 --redis-url 默认取 PLAITA_REDIS_URL 环境变量（与 flow_worker 一致）。
• 残留订阅 GC：EventFilter 匹配到已终态（completed/error）执行的订阅时， 就地注销该订阅而不是入队注定失败的 resume——error 路径泄漏的订阅不再 需要等 7 天 TTL。

Redis 日志 handler 的客户端队列上限（plaita#84）¶

RedisStreamHandler（enable_redis_logging=True 时默认挂载）把日志先放进客户端队列 再批量写 Redis Stream。该队列是有界的，边界与 Stream 侧的 max_len 无关：

Stream 侧 max_len（默认 DEFAULT_MAX_LEN = 10000，approximate=True）只约束 Redis 里的 Stream 长度，不约束 worker 进程内存；进程内存由上面的客户端队列上限约束。

事件回扫失败轮：游标不推进、失败不计 replayed（plaita#153）¶

EventReconciler.scan_once 回放任一事件失败时整轮作废：

TimeoutCallbackError 的公开入口是 plaita.event（实现处 plaita.event.timeout 是内部路径）：

from plaita.event import TimeoutCallbackError   # 公开入口

R1 的两条已知边界（本单未修，勿当成已修）：

• T1 — 回调抛 BaseException 仍中断本轮、饿死其余回调：KeyboardInterrupt / SystemExit 不被本链路的 except Exception 捕获，因此会跳过本轮剩余回调与后续 订阅。这是刻意不 catch（进程级信号不应被聚合成「某订阅的回调失败」）。 base 修订 b46f78e 行为完全相同——既有边界，非本单引入；本单不修， 也不为此加 except BaseException。
• T2 — 回调永不返回仍饿死同轮其余回调：本单没有引入 per-callback 超时预算 （无 asyncio.wait_for / timeout=），一个 await 永不结束的回调会把同轮后续回调 与后续订阅永久挂住。要收敛须另开单设计超时预算，本单不修。

r5 变动性质的如实说明（verifier C2/C3）：plaita/server/event_reconcile.py 在 r5（fc8fccf）里的改动是 docstring/注释 + 钉子测试，零可执行变更； plaita/server/log_handler.py 的失败回灌行 pending = entries[start:]（决定「只回灌 未提交批次」的那一行）未被 r5 改动——r5 对该文件其余部位只是 list → deque 的 语义等价机械适配（代码注释已声明「丢弃方向/计数/分批提交语义零变化」）。即 C2/C3 是措辞更正 + 测试，不是行为修复；上表的行为契约由 r4 的执行代码承担。

租户上下文显式声明（plaita#239 phase 2）¶

写路径（credentials/storage 的 save / delete）在租户来源为 default （当前进程/线程从未显式或派生设置）时fail-closed，抛具名 TenantContextMissingError，不再静默写 default 租户。

• 单租户部署：服务入口显式声明一次 set_current_tenant(DEFAULT_TENANT_ID, source="explicit")。FlowWorker.run() 与 console create_app() / lifespan 已内置该迁移行；自定义启动器/嵌入方 必须保留。
• 多租户：任务消息按 tenant_id 显式 set；经 #237 传播链（线程池 / asyncio 桥 / 进程 pickle）到达的上下文来源为 derived，同样放行。
• 「未设置」不再等价于「有意 default」；缺失上下文会被拒绝并从日志/探针 write_probe_snapshot() 的 (site, "default") 计数观测到。