plaita.server.flow_worker

分布式流程工作器 FlowWorker（需 server extra）。串联 ExecutionStorage / FlowStorage / EventBus，负责执行流程、持久化状态、处理恢复任务。

flow_worker

CodeNodeRegistrationError

Bases: RuntimeError

worker 启动期 CodeNode 注册被具名拒绝（A56-01，fail-closed）。

StatePersistError

Bases: RuntimeError

执行状态落盘失败（save_execution_state 返回 False）——瞬态那一支。

#80 起本类只表示「可重试的落盘失败」（Redis 瞬断 ⇒ 返回 False）。 「不可序列化」是永久失败、已从本类分家为 plaita.storage.base.StateSerializationError（存储层直接抛，不返回 False）， 由调用点转成可观测终态——别把两者再合回一个类型：那正是 #80 的病灶。 （本条注释原写『网络瞬断/序列化失败均如此』，已随契约分家同步。）

历史：终态/挂起/步进 落盘失败若被静默放行，消息会被 ack——节点副作用已发生而执行状态永远 停在旧值（僵尸执行）。此处把 False 升级为异常，让 RedisFlowWorker.run 的兜底路径不 ack 消息、走 at-least-once 重投。

刻意不是 ValueError 子类：run() 把 ValueError 当畸形消息 poison ack， 继承它会让存储瞬态错误仍被丢弃（ReviewFix D1 同款教训）。

ServiceDispatchError

Bases: RuntimeError

挂起服务任务投递失败（有 redis 客户端时不再吞）。

挂起时先 save(suspended) 再 rpush 服务队列；rpush 失败若被吞，执行已 suspended、消息被 ack → delay/approval 任务永无人接，永久挂起。抛出后 消息走重投、下轮 resume 重新执行挂起节点再派发（重复注册的订阅由 EventFilter 终态 GC / TTL 兜底）。suspended 状态保留，不翻 error。

ExecutionIdConflictError

Bases: ValueError

预铸 execution_id 已属于另一条 flow：拒绝启动，绝不覆盖盘上状态。

台账第三十一轮 V1（worker P2 批）：盘上 execution_id=X 属于 flow A 时， 一条 start_flow(flow_id=B, execution_id=X) 的载体（畸形/误重投/ID 复用）会让 start 新建 context={} 的 B running 行覆写 X——A 的 checkpoint/状态永久丢失。现在 start 载体查盘发现 flow_id 不一致即抛本 异常，不落任何行、不动 A 的状态。

刻意继承 ValueError：run() 把 ValueError 当畸形消息 poison ack—— 同一 execution_id 换 flow 重投永远不会自愈，重投只烧 delivery。 状态保留 + 具名日志是这条路径的可观测面。

FlowHashMismatchError

Bases: ValueError

flow 定义自启动后已变更（hash 不匹配），执行无法安全续跑（波次二任务②）。

resume_flow 在终态化 error 之后抛出本异常：run() 把它当 ValueError poison ack（终态已可观测，重投只会命中 already_terminal 短路）。刻意 用独立子类而非裸 ValueError——resume 的通用 except 需要区分「指纹 不匹配（已终态化，原样上抛）」与「引擎内部其他 ValueError（维持现状 终态化路径）」。

NodeFailureTerminalizedError

Bases: ValueError

节点失败已终态化 error（非瞬态/预算耗尽）——消息必须 poison ack。

为什么要有这个独立异常（plaita#73，2026-10-10 实证）：

终态化路径原本抛裸 RuntimeError。RuntimeError 不是 ValueError，于是一路落到 run() 的兜底 except Exception ——那条分支不 ack、留 pending 等超时回收 ⇒ 消息被反复重投 ⇒ resume_flow 见 status=error 只放行 resume_type=retry ⇒ 再次撞同一确定性失败 ⇒ 再次终态化 + 再次 RuntimeError…… 状态早已 error，消息却永远在转（实测某日同一执行重投 598 次、 沙箱实例持续占位 5.66 实例小时）。

更糟的是这条路径不碰重试计数键：INCR 位于 _node_failure_retry_decision 内、在 _is_retryable_node_failure 判据之后，确定性失败（exited 1 / AgsError: sync_in / 超时）直接 return None ⇒ 计数键恒为 1（实测全库 74 个键全为 1） ⇒ 「预算耗尽」分支从未触发，重投无界。

语义与既有的 FlowHashMismatchError 同族：raise 前状态已终态化， run() 按 ValueError poison ack（终态已可观测，重投只会命中 already_terminal 短路，纯属浪费）。刻意用独立子类而非裸 ValueError，以便 resume 的通用 except 区分「已终态化」与 「引擎内部其他 ValueError（维持现状终态化路径）」。

TenantQuotaDeferred

Bases: RuntimeError

start 消息超出租户并发配额（plaita#28）：延迟，不是失败。

消费入口抛出——由 run() 的专门分支重投（见 _defer_over_quota_start）： 入队一份同体新副本到 stream 尾部（让位给更早入队的其他租户消息）+ ack 原条。刻意不「留 pending」：XCLAIM 回收每轮让 deliveries +1，烧到 max_deliveries 会误死信——配额已满不是消息的错，绝不能因排队而死信。 刻意不是 ValueError 子类（防 poison ack 语义误伤）。

命中前提：消息不带名额（非 console 入口生产方）。console start 在入口 即占名额（try_admit_tenant_start），其消息抵达 worker 时已是成员，闸 幂等放行——worker 闸只拦「绕过入口的第三方生产方」的洪峰。

有界（防重投空转）：副本带 quota_defers 计数，达 PLAITA_TENANT_QUOTA_DEFER_MAX（默认 64）后带 quota_bypass 放行 ——宁可短暂突破软配额，不可让消息在队列里无限空转烧 CPU。

TaskNotForThisWorker

Bases: RuntimeError

任务与本机不亲和（repo/run_dir 指向别的机器的绝对路径）。

消费入口抛出——由 run() 的专门分支交接（见 _handover_non_affine）： ack 原消息 + 重入队一份同体新副本（delivery 归 1）。不终态化、不死信： 任务本身没毛病，只是不该在本机跑。

为何要「重入队」而非「留 pending」：留 pending 的消息会被对端 worker 按 claim_min_idle_ms 反复 XCLAIM，而 XCLAIM 每次让 Redis 的 deliveries +1——两 worker 环境下 60s 一轮，5 轮即触顶 max_deliveries → 误死信（2026-10-06 双机实测：非亲和任务被让出 5 次后进 DLQ）。 重入队新副本把计数清零，交接无损耗。

但有界（2026-10-11 无限交接风暴）：重入队使 times_delivered 恒为 1 ⇒ max_deliveries 永不触发 ⇒ 若判据不可满足，交接没有终止条件。 故交接次数由 worker 侧按 execution_id 计数约束（AFFINITY_HANDOVER_MAX）， 达上限即终态化 error（可观测），不再交接——见 _handover_non_affine / _terminalize_handover_exhausted。

ResumeProtocolError

ResumeProtocolError(message: str, execution_id: Optional[str] = None)

Bases: RuntimeError

resume 协议错误（#33）：执行状态与 resume_type 不匹配，执行保持原状。

策略层守卫 ResumeGuardError（continue/retry 试图绕过 pending 挂起节点、 resume 打在无挂起/非挂起 checkpoint 上）经 _chain_has_resume_guard_error 判出后，resume_flow 不终态化、把执行留在原状态（suspended/running）， 以本异常上抛。裸 ResumeError 不在豁免面内（节点 resume() 自身 抛错是执行失败，走终态化 error 的现状路径，见该判据 docstring）。 run() 主循环按 「消费失败」处理：消息走 ack（执行状态原样可查、事件订阅仍在，真正的 决议路径 event/cancel/timeout 随时可入），重投只会重复命中同一守卫， 无意义。例外（plaita#258）：event/timeout 载体在 running 中间态 已错过挂起点（订阅已注销）时，ack 前先状态感知并归一化 continue 重投——见 _handle_resume_protocol_error；直接 ack 会丢掉唯一载体。 刻意不是 ValueError 子类（防 poison ack 语义误伤）。

源代码位于： plaita/server/flow_worker.py

def __init__(self, message: str, execution_id: Optional[str] = None):
    super().__init__(message)
    self.execution_id = execution_id

NodeExecutionRetryableError

NodeExecutionRetryableError(message: str, execution_id: Optional[str] = None, attempt: int = 0, delivery_count: Optional[int] = None)

Bases: RuntimeError

节点执行可重试失败（波次二任务①）：执行未终态化，消息等重投。

节点异常经 _is_retryable_node_failure 判为瞬态且重试预算未耗尽时， 处理函数跳过终态化改抛本异常——磁盘 state 停在最后成功步 checkpoint， run() 据此退避后显式重投一份新副本（_requeue_retryable_failure： 按 attempt 指数退避后先入队、后 ack 原条，plaita#91），新副本从 checkpoint 重跑失败节点；start 载体且盘上已有 checkpoint 时重投副本 转换为 resume 载体（不整条重放、不覆写 checkpoint）。

2026-10-09 变更（plaita#52 真因）：此前是「不 ack、留 pending 等 claim_min_idle_ms 回收」。该语义在消息已被竞争者 ack 释放时不成立 （ExecutionLeaseError 分支为避免烧 delivery 会显式 ack），实测导致执行 既不重试也不终态、卡在 running 58 分钟。现改为显式重投；仅当重投本身 失败（Redis 抖动等）才退化为留 pending 交 reclaim。2026-10-16（#91）： 显式重投补上指数退避（旧实现无节奏，5 次预算 ~40ms 烧光）与 start 载体的 resume 转换。

刻意不是 ValueError 子类（同 StatePersistError 的理由）：ValueError 会被 run() 当畸形消息 poison ack。

属性：

名称	类型	描述
execution_id		所属执行（日志与守卫后备判据用）。
attempt		已放行的重试次数（重试计数键 INCR 后的值）。
delivery_count		消息当前投递次数（仅可观测；预算判定用重试计数键， 见 _record_node_retry 的说明）。

源代码位于： plaita/server/flow_worker.py

def __init__(self, message: str, execution_id: Optional[str] = None,
             attempt: int = 0, delivery_count: Optional[int] = None):
    super().__init__(message)
    self.execution_id = execution_id
    self.attempt = attempt
    self.delivery_count = delivery_count

FlowWorker

FlowWorker(execution_storage: ExecutionStorage, flow_storage: FlowStorage, event_bus: EventBus = None, cache_size: int = 100, cache_ttl: int = 300, callback_handlers: Optional[list] = None, execution_lease: Optional[ExecutionLease] = None, lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS)

流程工作器：把 run_distributed 与 ExecutionStorage / FlowStorage / EventBus 串起来。

这是 suspend/resume 编排器，不是具备至少一次投递或崩溃安全的工作流引擎。 可靠性边界见 docs-site distributed/flow-worker.md 与类常量 PERSIST_EVERY_N_STEPS / RedisFlowWorker.run 的队列语义说明。

初始化流程工作器

参数：

名称	类型	描述	默认
execution_storage	ExecutionStorage	执行状态存储实例	必需
flow_storage	FlowStorage	流程定义存储实例	必需
event_bus	EventBus	事件总线实例	None
cache_size	int	缓存大小，默认100	100
cache_ttl	int	缓存过期时间(秒)，默认300秒	300
callback_handlers	Optional[list]	分布式执行期间贯穿所有步骤的回调列表	None
execution_lease	Optional[ExecutionLease]	resume 租约（默认 NullExecutionLease）；RedisFlowWorker 注入 Redis 实现	None
lease_ttl_seconds	int	resume 租约 TTL（秒），推进过程中会 renew	DEFAULT_LEASE_TTL_SECONDS

源代码位于： plaita/server/flow_worker.py

def __init__(
    self,
    execution_storage: ExecutionStorage,
    flow_storage: FlowStorage,
    event_bus: EventBus = None,
    cache_size: int = 100,
    cache_ttl: int = 300,
    callback_handlers: Optional[list] = None,
    execution_lease: Optional[ExecutionLease] = None,
    lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS,
):
    """
    初始化流程工作器

    Args:
        execution_storage: 执行状态存储实例
        flow_storage: 流程定义存储实例
        event_bus: 事件总线实例
        cache_size: 缓存大小，默认100
        cache_ttl: 缓存过期时间(秒)，默认300秒
        callback_handlers: 分布式执行期间贯穿所有步骤的回调列表
        execution_lease: resume 租约（默认 NullExecutionLease）；RedisFlowWorker 注入 Redis 实现
        lease_ttl_seconds: resume 租约 TTL（秒），推进过程中会 renew
    """
    self.execution_storage = execution_storage
    self.flow_storage = flow_storage
    self.event_bus = event_bus
    self.callback_handlers = list(callback_handlers) if callback_handlers else []
    # 节点耗时采集：**按执行**一个采集器（并发处理多个执行时不串场），只监听
    # on_node_start/on_node_end；落盘时由 _persist_state_or_raise 写入
    # ExecutionState.node_timings。
    self._node_timings: Dict[str, NodeTimingCallback] = {}
    # run 级事件时间线（plaita#25）：**按执行**一个采集器，与 _node_timings
    # 同一对钩子、同一落盘收口（_collect_execution_timeline）。它同时是本进程
    # 向 SSE 频道 ``plaita:execution:events:{id}`` 发布事件的**唯一发布者**
    # （修复前该频道全仓零发布者，订阅者除 initial_state 外永远静默）。
    self._timelines: Dict[str, ExecutionTimelineCallback] = {}
    # 执行状态写的**每执行互斥锁**（见 _state_write_lock）：同一 execution_id
    # 的所有状态写（推进路径 _persist_state_or_raise ↔ 看门狗心跳
    # _publish_node_progress）必须串行，否则心跳的「读-改-写」会把并发终态写
    # 整行回滚（2026-10-10 评审复现：completed 被改成 running + 旧 checkpoint）。
    # WeakValueDictionary：没有线程再持有该锁时自动回收（长跑 worker 不泄漏）。
    self._state_write_locks: "weakref.WeakValueDictionary[str, threading.RLock]" = (
        weakref.WeakValueDictionary()
    )
    self._state_write_locks_guard = threading.Lock()
    # 沙箱生命周期回调：**按执行**缓存（同一执行的多个 step 必须复用同一个
    # 实例，否则 agent 节点在早先 step 产生的 workspace 快照到 flow 结束那一步
    # 已经丢了）；终态落盘时按持久化上下文释放并回收（_release_sandboxes）。
    self._sandbox_callbacks: Dict[str, Any] = {}
    # 优雅停机（无损升级）：draining = 不再领新任务、等 in-flight 收尾；
    # 超时由 _force_stop_after_drain 兜底退出（消息留 pending 待接管）。
    self._draining = threading.Event()
    self._drain_timer: Optional[threading.Timer] = None
    self._drain_started_at: Optional[float] = None
    # 注意：_draining 在 _drain_event 里惰性兜底，__new__ 构造的骨架 worker 也安全
    # 初始化流程定义缓存，使用TTL缓存
    self.flow_definition_cache = TTLCache(maxsize=cache_size, ttl=cache_ttl)
    # 原始定义指纹缓存（plaita#36）：与 flow_definition_cache 同窗口，
    # 在 get_flow_definition 里随解析结果一起写入。
    self.flow_definition_hash_cache = TTLCache(maxsize=cache_size, ttl=cache_ttl)
    self.execution_lease = execution_lease or NullExecutionLease()
    self.lease_ttl_seconds = lease_ttl_seconds
    # 取消监听（波次③：步内可中断）：基类先建好登记表与停止位，内存 worker
    # 不启动线程（_cancel_requested 无 redis 客户端时恒为 False，登记为空转）。
    self._cancel_watch_lock = threading.Lock()
    self._cancel_watch: Dict[str, Tuple[Any, str]] = {}
    self._cancel_thread: Optional[threading.Thread] = None
    self._cancel_stop = threading.Event()
    self._cancel_poll_seconds: Optional[float] = None
    # A69-04：年龄闸「两源都不可判」的 fail-closed 计数（/metrics 暴露为
    # carrier_age_unknown_total；只增不减，运维据此判断是否有生产者缺 timestamp）。
    self._carrier_age_unknown_total = 0
    # A69-04：启动期保留期配置校验（每进程一次；可疑配置只 WARNING 不阻断）。
    validate_tombstone_retention()

get_flow_definition

get_flow_definition(flow_id: str, version: Optional[str] = None) -> Flow

获取流程定义，支持按ID和版本获取

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
version	Optional[str]	流程版本，如果不指定则获取最新版本	None

返回：

类型	描述
Flow	Dict[str, Any]: 流程定义

引发：

类型	描述
ValueError	如果找不到流程定义或版本不匹配

源代码位于： plaita/server/flow_worker.py

def get_flow_definition(self, flow_id: str, version: Optional[str] = None) -> Flow:
    """
    获取流程定义，支持按ID和版本获取

    Args:
        flow_id: 流程ID
        version: 流程版本，如果不指定则获取最新版本

    Returns:
        Dict[str, Any]: 流程定义

    Raises:
        ValueError: 如果找不到流程定义或版本不匹配
    """
    # 生成缓存键（含租户段：同名流程可共存于不同租户 namespace）
    cache_key = f"{current_tenant()}:{flow_id}:{version or 'latest'}"

    # 尝试从缓存获取
    if cache_key in self.flow_definition_cache:
        logger.info("从缓存获取流程定义: %s", cache_key)
        return self.flow_definition_cache[cache_key]

    # 缓存未命中，从存储获取
    logger.info("从存储获取流程定义: %s, 版本: %s", flow_id, version or 'latest')
    flow_definition = self.flow_storage.get_flow(flow_id, version)

    if not flow_definition:
        # Delegate diagnostic details to the storage layer via an optional
        # diagnose() method — FlowWorker must not peek at storage internals
        # (e.g. Redis keys / namespaces) directly.
        if hasattr(self.flow_storage, "diagnose_missing_flow"):
            try:
                self.flow_storage.diagnose_missing_flow(flow_id)
            except Exception as exc:
                logger.warning("flow_storage.diagnose_missing_flow(%s) failed: %s", flow_id, _scrub_exception_text(exc))

        error_msg = f"找不到流程定义: {flow_id}"
        logger.error(error_msg)
        raise ValueError(error_msg)

    # 解析流程定义
    try:
        logger.info("成功获取流程定义: %s, 数据: %s", flow_id, flow_definition.get('name', 'unknown'))
        flow = Flow.model_validate(flow_definition)
    except Exception as e:
        error_msg = f"解析流程定义失败: {e}"
        logger.error(error_msg)
        raise ValueError(error_msg)

    self.flow_definition_cache[cache_key] = flow
    # 指纹随定义一起缓存（plaita#36）：指纹只取 model_validate **之前**的
    # 原始定义，与缓存里的 Flow 同窗口，避免 start/resume 各自重算走叉。
    self._flow_definition_hash_cache()[cache_key] = compute_flow_definition_hash(
        flow_definition
    )

    return flow

get_flow_definition_hash

get_flow_definition_hash(flow_id: str, version: Optional[str] = None) -> str

当前存储定义的指纹（plaita#36：原始定义 JSON 的 sha256）。

与 get_flow_definition 共用缓存窗口：定义加载时就算好（同一次 flow_storage.get_flow），调用方（start/resume）不自己从 Flow 对象反推——那正是版本漂移的入口。缓存未命中（定义缓存被清 / 加载器被 测试替身替换）时回源到同一个 FlowStorage 直读原始定义，绝不回退 到「解析后 dump」口径。

源代码位于： plaita/server/flow_worker.py

def get_flow_definition_hash(self, flow_id: str, version: Optional[str] = None) -> str:
    """当前存储定义的指纹（plaita#36：原始定义 JSON 的 sha256）。

    与 ``get_flow_definition`` 共用缓存窗口：定义加载时就算好（同一次
    ``flow_storage.get_flow``），调用方（start/resume）不自己从 ``Flow``
    对象反推——那正是版本漂移的入口。缓存未命中（定义缓存被清 / 加载器被
    测试替身替换）时回源到**同一个 FlowStorage** 直读原始定义，绝不回退
    到「解析后 dump」口径。
    """
    cache_key = f"{current_tenant()}:{flow_id}:{version or 'latest'}"
    cache = self._flow_definition_hash_cache()
    cached = cache.get(cache_key)
    if cached is not None:
        return cached
    flow_definition = self.flow_storage.get_flow(flow_id, version)
    if not flow_definition:
        raise ValueError(f"找不到流程定义: {flow_id}")
    digest = compute_flow_definition_hash(flow_definition)
    cache[cache_key] = digest
    return digest

start_flow

start_flow(flow_id: str, params: Dict[str, Any], version: Optional[str] = None, execution_id: Optional[str] = None, dedup_key: Optional[str] = None, delivery_count: Optional[int] = None, enqueued_at: Optional[str] = None, schema_version: Optional[int] = None) -> Dict[str, Any]

启动流程执行

全程持有 execution_id 执行租约（先行落 running 行之前取得，处理 结束 finally 释放）：队列按 claim_min_idle_ms（默认 60s）把长任务 的 pending 消息 XCLAIM 重派给空闲 worker——无租约的 start 在多 worker 池下跑超过 60s 必被重派双跑（#23）。另一 worker 已持租约（消息重派/ 并发投递）时抛 ExecutionLeaseError，本消息不 ack 留待处理。

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
params	Dict[str, Any]	流程输入参数	必需
version	Optional[str]	流程版本，如果不指定则使用最新版本	None
execution_id	Optional[str]	预铸 id（BFF start 时铸造、随消息透传）——提交方 即刻拿 id 返给调用方，无需等 worker 消费；缺省就地铸造	None
dedup_key	Optional[str]	可选 start 幂等键（消息重投/双开收敛；未提供则行为 与存量完全一致，见 _claim_start_dedup）	None
delivery_count	Optional[int]	消息投递次数（可观测；见 _process_execution_result）	None
enqueued_at	Optional[str]	console 入队时间（消息体 timestamp，ISO 串）。用于 时间线回答「排队多久」（plaita#25）：缺失/不可解析 ⇒ queue_wait_ms=None（未采集），不折算成 0。	None
schema_version	Optional[int]	触发本次 start 的消息信封版本（A44-02）。仅用于 start 幂等命中后重入队的 resume 副本继承源版本；直接调用方 缺省 None = 按当前 build 写版本。	None

返回：

类型	描述
Dict[str, Any]	Dict[str, Any]: 流程执行结果

源代码位于： plaita/server/flow_worker.py

def start_flow(self, flow_id: str, params: Dict[str, Any], version: Optional[str] = None,
               execution_id: Optional[str] = None,
               dedup_key: Optional[str] = None,
               delivery_count: Optional[int] = None,
               enqueued_at: Optional[str] = None,
               schema_version: Optional[int] = None) -> Dict[str, Any]:
    """
    启动流程执行

    全程持有 ``execution_id`` 执行租约（先行落 running 行之前取得，处理
    结束 finally 释放）：队列按 ``claim_min_idle_ms``（默认 60s）把长任务
    的 pending 消息 XCLAIM 重派给空闲 worker——无租约的 start 在多 worker
    池下跑超过 60s 必被重派双跑（#23）。另一 worker 已持租约（消息重派/
    并发投递）时抛 ``ExecutionLeaseError``，本消息不 ack 留待处理。

    Args:
        flow_id: 流程ID
        params: 流程输入参数
        version: 流程版本，如果不指定则使用最新版本
        execution_id: 预铸 id（BFF start 时铸造、随消息透传）——提交方
            即刻拿 id 返给调用方，无需等 worker 消费；缺省就地铸造
        dedup_key: 可选 start 幂等键（消息重投/双开收敛；未提供则行为
            与存量完全一致，见 ``_claim_start_dedup``）
        delivery_count: 消息投递次数（可观测；见 _process_execution_result）
        enqueued_at: console 入队时间（消息体 ``timestamp``，ISO 串）。用于
            时间线回答「排队多久」（plaita#25）：缺失/不可解析 ⇒
            ``queue_wait_ms=None``（未采集），不折算成 0。
        schema_version: 触发本次 start 的消息信封版本（A44-02）。仅用于
            start 幂等命中后重入队的 resume 副本继承源版本；直接调用方
            缺省 ``None`` = 按当前 build 写版本。

    Returns:
        Dict[str, Any]: 流程执行结果
    """
    # 获取流程定义
    flow = self.get_flow_definition(flow_id, version)



    logger.info("开始执行流程: %s, 版本: %s", flow_id, version or '最新版本')

    try:
        # P0 可见性修复（keeper 迁移设计稿 §5.5 清单①，43828aa）：
        # execution_id 优先吃 BFF 预铸（随消息透传，提交方即刻可轮询），
        # 否则就地铸造；行 id 与 result.execution_id 天然一致。异常遗留
        # 的 running 行是 zombie，交 reaper/心跳年龄判定处置。
        # 位置前移到执行器构造之前：节点耗时采集器按执行登记，需要 id 已定。
        # preminted（plaita#234）：只有自带 id 的载体才可能命中盘上已有
        # 断点；就地铸造的 id 天然是首次执行，不必多读一次盘。
        preminted_execution_id = bool(
            isinstance(execution_id, str) and execution_id.strip()
        )
        execution_id = execution_id or uuid.uuid4().hex

        # A51-01：删除墓碑复核（五种纵深之一）。分派闸之外，直达
        # ``start_flow`` 的调用同样不得把已删执行重建；新铸造的 id 天然
        # 不可能有墓碑，只有预铸 id 才需要多读一次。命中即 ack 语义
        # （返回值供调用方确认），并释放同 id 的 dedup 映射。
        if preminted_execution_id:
            tombstone = self._load_execution_tombstone(
                execution_id, current_tenant()
            )
            if tombstone is not None:
                if dedup_key:
                    self._release_start_dedup(dedup_key, execution_id)
                logger.warning(
                    "执行 %s 已删除（墓碑 tenant=%s reason=%s deleted_at=%s），"
                    "start_flow 拒绝重建（flow=%s）",
                    execution_id,
                    tombstone.get("tenant"),
                    tombstone.get("reason"),
                    tombstone.get("deleted_at"),
                    flow_id,
                )
                return {
                    "execution_id": execution_id,
                    "status": "deleted",
                    "discarded": True,
                }

        # 创建流程执行器并执行流程
        # 复用同一个 FlowExecution 贯穿所有分布式步骤, 保留用户回调
        handlers = self._handlers_for(execution_id, enqueued_at=enqueued_at)
        # 时间线开篇（plaita#25）：记 run_start + 由 enqueued_at 算出排队时长。
        # 位置在**落 running 行之前**——排队时长的终点是「worker 开始处理」，
        # 不是「首节点开始跑」，否则排队时长会被流程定义加载/租约获取计入。
        _timeline = self._timelines.get(execution_id)
        if _timeline is not None:
            _timeline.start_run()
        execution = FlowExecution(
            event_bus=self.event_bus,
            callback_handlers=handlers,
        )
        execution.mode = ExecutionMode.DISTRIBUTED
        self._bind_observers(execution)

        # start 幂等键（波次二任务③，与 G1 预铸 id 组合）：在先行落行
        # 之前认领，映射值=预铸或就地铸造的 execution_id——重投/双开
        # 在**任何新行落盘之前**即被拦截，首节点副作用由此受重投保护。
        dedup_action, mapped_execution_id = self._claim_start_dedup(
            dedup_key, execution_id
        )
        if dedup_action == "hit":
            hit_result = self._handle_start_dedup_hit(
                flow_id, mapped_execution_id, params=params,
                schema_version=schema_version,
                carrier_execution_id=execution_id,
            )
            if hit_result is not None:
                # 本次没有真正启动：回收按本 id 装配的时间线采集器
                # （否则节流命中会在长跑 worker 里留一条无人认领的采集器）。
                self._timelines.pop(execution_id, None)
                if hit_result.get("release_dedup") and dedup_key:
                    self._release_start_dedup(
                        dedup_key, hit_result.get("execution_id") or mapped_execution_id
                    )
                return hit_result
            # 孤儿映射（首次执行从未落盘）：释放后重新认领，让本次启动
            # 继续受保护；并发后来者抢先认领则放弃设防（存量语义）
            if dedup_key:
                self._release_start_dedup(dedup_key, mapped_execution_id)
                dedup_action, mapped_execution_id = self._claim_start_dedup(
                    dedup_key, execution_id
                )
                if dedup_action == "hit":
                    logger.warning(
                        "start 幂等键孤儿释放后被并发认领，本次启动不设防: %s",
                        dedup_key,
                    )

        # 载体转换（plaita#234）：start 消息**不是只能从首节点起跑**。
        # 盘上已有该 execution_id 的断点时（worker 崩溃/滚动重启、
        # StatePersistError、ServiceDispatchError 后消息不 ack 被 XCLAIM
        # 原样回收——这条回收路径**不经过** #91 的
        # ``_requeue_body_for_retry``），历史实现会新建 `context={}` 的
        # 行无条件覆写盘上 checkpoint，再以**不带 saved_context** 的
        # ``run_distributed`` 从首节点整条重放：已完成节点的外部副作用
        # 最多重复 max_deliveries 次，且 $LAST_NODE/$NODE/$INPUT 被永久
        # 抹成 {}，此后连正确的 event/continue resume 也失去续跑现场。
        #
        # 修法：把 start 载体**转成 resume 语义**（判定口径与 #91 一致，
        # 但入口收在 start_flow，覆盖「重投不经 _requeue_body_for_retry」
        # 的那条路），复用 resume_flow 已有的全部保护：终态/挂起幂等短路
        # （不重跑）、flow_hash 校验、取消检查、租约，以及 #233 的挂起
        # 补派。此时尚未取得 start 租约（两者共用同一 execution_id 租约
        # 键），转调不会自撞锁。
        if preminted_execution_id:
            existing_state = self._existing_start_checkpoint(flow_id, execution_id)
            if existing_state is not None:
                logger.warning(
                    "start 载体命中盘上已有断点（execution_id=%s, status=%s, "
                    "checkpoint=%s）：转 resume 语义，绝不整条重放/覆写 checkpoint",
                    execution_id,
                    existing_state.status,
                    "有" if existing_state.context else "空",
                )
                # 时间线采集器**保留**：resume 路径的 _handlers_for 会复用
                # 同一采集器，从而保住本消息的 enqueued_at/run_start（#25）。
                return self.resume_flow(
                    flow_id,
                    execution_id,
                    "continue",
                    delivery_count=delivery_count,
                    params=params,
                )

        # 先落 running 行再执行（P0 可见性）：首节点期间 /api/executions
        # 查得到此行、cancel 有锚。flow_hash：对**存储中的原始定义**计算
        # 指纹（plaita#36 口径 v2）随行落盘——resume 时与当前定义比对，防
        # 运行中改定义后续跑走错分支；原始定义口径保证同一份存储定义在
        # 不同引擎版本下算同一个指纹。
        state = ExecutionState(
            execution_id=execution_id,
            flow_id=flow_id,
            flow_version=version,
            flow_hash=self.get_flow_definition_hash(flow_id, version),
            flow_hash_algo=FLOW_HASH_ALGO,
            engine_version=_engine_version(),
            tenant_id=current_tenant(),
            context={},
            status="running",
            start_time=datetime.now().isoformat(),
            invoker="worker"
        )

        # start 入口取消检查点（与 resume 入口对称）：BFF cancel 可发生在
        # 「已入队、未消费」窗口，更关键的是 start 消息 at-least-once 重投
        # 时（在途失败→未 ack→重投）同 id 重跑整条 flow——不查标志则
        # 已取消意图永远等不到步界（E2E drill 3 实证：超时 flow 空转
        # N×900s）。命中即终态化，不再执行。
        if self._cancel_requested(execution_id):
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_at_start_entry")
            self._finalize_observers()
            logger.info("执行 %s 在 start 入口命中取消标志，终态化", execution_id)
            return {
                "execution_id": execution_id,
                "status": "cancelled",
                "cancelled_at_start": True,
            }

        # 租约（#23 扩展 A′）：**在落 running 行之前 acquire**，持有整个
        # start 处理窗口。A′（2026-10-06）只把租约对齐到 run_distributed
        # 之前，仍在先行落行**之后**——「落行 → acquire」间可被别的 worker
        # 抢先 XCLAIM 同一 start 消息（>claim_min_idle_ms 即可），抢占者
        # 先 acquire 成租约，原持有者 acquire 失败礼貌退出，但其 running
        # 行已被抢占者覆写。移动到落行前同时消除 A′ 的三个后果：
        # ①多 worker 抢到同一 start 消息时无闸可拦，双方同时跑同 id（双跑）；
        # ②死信守卫按「租约是否被持有」判活，start 执行恒"租约空"→ 长步
        #   误判持有者已死 → 误死信 + 重入队，长任务被反复打断；
        # ③抢占者白烧 5 次 delivery 配额。
        # 对齐 resume：acquire → 看门狗续租 → finally 释放。acquire 失败
        # 即抛 ExecutionLeaseError（run() 走不 ack/重投语义），抢占者礼貌退出。
        # holder 嵌入注册表 instance id（#50）：其他 worker 在租约冲突时
        # 可反解持有者并核其在册与否，区分「活持有者」与「死后 TTL 尾巴」。
        holder = new_holder_token(
            prefix="start", instance_id=getattr(self, "instance_id", None)
        )
        lease_value, fence_token = self._acquire_lease(execution_id, holder)
        if lease_value is None:
            raise ExecutionLeaseError(
                f"execution {execution_id} is leased by another worker; "
                "refuse concurrent start"
            )
        fence_token_reset = None
        if fence_token is not None:
            fence_token_reset = set_current_fence_token(fence_token)

        # 落 running 行（持租约窗口内）：与本执行此后所有写同 fence 世代，
        # 不会被后续接管者的 fenced CAS 拒绝序列排斥。
        self._persist_state_or_raise(execution_id, state, "start")

        # 取消监听（波次③）：登记整个推进窗口——取消标志键命中即在途节点被
        # execution.cancel() 中止（code 沙箱当场 killpg；协作节点如 agentrun
        # 消费 cancel_event 收手）。租约看门狗同窗口。
        self._register_cancel_watch(execution_id, execution)
        self._register_lease_watch(
            execution_id, lease_value, execution, fence_token=fence_token
        )
        # 标记「本次处理是否正常跑完」：只有正常跑完才说明该 execution
        # 这一轮真的结束了；从 run_distributed 里抛异常（尤其
        # ExecutionLeaseError 的 ack 释放路径）时执行对象可能仍在推进，
        # 见 finally 里的条件释放。
        completed_normally = False
        try:
            result = execution.run_distributed(flow, params=params, execution_id=execution_id)

            # 处理执行结果
            final_result = self._process_execution_result(
                flow, result, state, execution, delivery_count=delivery_count,
                lease_execution_id=execution_id,
            )

            completed_normally = True
            return final_result
        finally:
            self._unregister_cancel_watch(execution_id)
            self._unregister_lease_watch(execution_id, lease_value)
            if fence_token_reset is not None:
                reset_current_fence_token(fence_token_reset)
            # release 对整个租约 value 串 compare（fencing 档 = {holder}:{gen}）
            #
            # ⚠️ **释放租约 ≠ 执行已结束**（2026-10-10 二波，生产实证）：
            # 「执行被他人持租约 → ack 释放」这条路径让本消息处理走到
            # finally，但**该 execution 对象仍在沙箱里推进长节点**
            # （实测 exec 339c3262：19:20:31 复用沙箱跑 impl，19:21:25 因
            # 「他人持租约」被 ack 释放 ⇒ 此后心跳停 80 分钟 ⇒ keeper 判
            # zombie 收尸）。一旦 release，「无租约」成事实：看门狗登记已被
            # 上面 unregister 摘除，也从未走过 renew 失败分支 ⇒ **没有任何
            # 组件再发布它的进度** ⇒ 健康 run 必然被判死。
            #
            # 因此：**仅当执行确已停止推进时才真正 release**；否则交给
            # `_lease_orphan_watch` 继续观测（只发心跳，不续租、不重复
            # cancel），由它在该执行结束或重新登记时摘除。
            if not completed_normally and self._execution_still_advancing(execution):
                # tenant 由上下文取（该作用域无 tenant_id 形参，见上方
                # `tenant_id=current_tenant()` 的同款用法）。
                self._register_lease_orphan(
                    execution_id, execution, current_tenant(), fence_token
                )
            else:
                self.execution_lease.release(execution_id, lease_value)

    except ExecutionIdConflictError:
        # 冲突的 start 未落任何行（查盘在落 running 行之前）。若幂等映射
        # 是**本次**认领的，释放它——否则 7 天内同 dedup_key 会被指向
        # 另一条 flow 的 execution_id，后续修正后的 start 也走不到这里。
        # 原样上抛：run() 按 ValueError poison ack（换 flow 重投不会自愈）。
        if dedup_key and locals().get("dedup_action") == "claimed":
            self._release_start_dedup(
                dedup_key, mapped_execution_id or execution_id
            )
        raise

    except (NodeExecutionRetryableError, ServiceDispatchError, ExecutionLeaseError):
        # 节点级可重试失败（波次二任务①）/ 幂等命中后重入队 resume 失败
        # （波次二任务③）/ **租约被他人持有**（A′ + #23，租约覆盖整个
        # start 窗口含先行落行）：原样上抛给 run() 按不 ack 语义处理——
        # ExecutionLeaseError 尤其关键：run() 的
        # `except ExecutionLeaseError` 分支据此 note_lease_conflict 并把消息
        # 留在 pending 等租约过期后 reclaim；若被下方通用 except 包成
        # RuntimeError，run() 会误 ack 消息 → 抢占者把别人的活执行 ack 掉，
        # 且真实持有者崩溃后无人 reclaim（执行永久失联）。绝不能包。
        raise

    except Exception as e:
        logger.error("执行流程出错: %s", _scrub_exception_text(e))
        # 波次③步内取消：start 路径首个节点在途命中取消监听 →
        # 引擎自 run_distributed 归一化抛出（挂 __cause__）或协作节点自杀
        # 抛普通异常。两条判据任一命中即终态化 cancelled，绝不包成
        # RuntimeError 让执行停在 running（取消是控制面意图）。
        if (
            _chain_has_cancellation(e) or self._cancel_requested(execution_id)
        ) and "state" in locals() and state is not None:
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_in_start_node")
            self._finalize_observers()
            logger.info(
                "执行 %s 在 start 首节点执行期响应取消，终态化 cancelled",
                execution_id,
            )
            return {"execution_id": execution_id, "status": "cancelled"}
        # plaita#25 P2：错误路径落盘。确定性失败此前只打日志 + 实时发布事件
        # ⇒ 状态行停在 start 时落的 running，**落盘时间线只有 run_start**：
        # 失败节点（runner 已补发 on_node_end(error)）与收口事件都只存在于
        # 内存/SSE 里，数小时后复盘时没有第二来源。这里走**唯一落盘收口**
        # 把这次尝试的事件补进 state.timeline，并在流里补一条
        # run_end(final=False, status=error)。
        #
        # 状态行**刻意不动**：错误路径的 running 行是既有契约（可重投、
        # 可 cancel、zombie 可见——见
        # tests/unit/test_g1_resume_retry.py::test_running_row_survives_start_failure），
        # 也保证 ack/重投与死信守卫的判据逐字不变。
        # best-effort：落盘失败只告警，下面的 RuntimeError 照旧上抛 ⇒
        # 消息仍不 ack 走 at-least-once 重投。
        self._record_failed_start_attempt(execution_id, locals().get("state"))
        self._finalize_observers()
        raise RuntimeError(f"执行流程出错: {e}")

resume_flow

resume_flow(flow_id: str, execution_id: str, resume_type: str, data: Optional[Dict[str, Any]] = None, delivery_count: Optional[int] = None, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

恢复流程执行。

取得 execution_id 租约后才推进；另一 worker 已持有租约时抛 ExecutionLeaseError（RedisFlowWorker 对此不 XACK，待租约过期后回收）。

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
execution_id	str	执行ID	必需
resume_type	str	恢复类型	必需
data	Optional[Dict[str, Any]]	恢复数据	None
delivery_count	Optional[int]	消息投递次数（可观测；见 _process_execution_result）	None
params	Optional[Dict[str, Any]]	可选入参（#122）。只在盘上 checkpoint 为空时被引擎用来 重新 setup_flow（start→resume 载体转换会撞上这种窗口）； checkpoint 非空（正常 resume）时不参与——续跑一律以盘上 context 为准，续跑语义零变化。	None

源代码位于： plaita/server/flow_worker.py

def resume_flow(self, flow_id: str, execution_id: str, resume_type: str,
                data: Optional[Dict[str, Any]] = None,
                delivery_count: Optional[int] = None,
                params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    恢复流程执行。

    取得 ``execution_id`` 租约后才推进；另一 worker 已持有租约时抛
    ``ExecutionLeaseError``（RedisFlowWorker 对此**不** XACK，待租约过期后回收）。

    Args:
        flow_id: 流程ID
        execution_id: 执行ID
        resume_type: 恢复类型
        data: 恢复数据
        delivery_count: 消息投递次数（可观测；见 _process_execution_result）
        params: 可选入参（#122）。只在盘上 checkpoint 为空时被引擎用来
            重新 ``setup_flow``（start→resume 载体转换会撞上这种窗口）；
            checkpoint 非空（正常 resume）时**不参与**——续跑一律以盘上
            context 为准，续跑语义零变化。
    """
    # 前置判定（租约之前）：终态/挂起幂等短路在此完成，保持 #33 的
    # 「suspended + continue 不取租约」契约。这里拿到的 state 只是
    # **预读快照**——取到租约后必须重读并以盘上状态为准（plaita#241）。
    state, early_result = self._load_and_guard_resume_state(
        flow_id, execution_id, resume_type
    )
    if early_result is not None:
        return early_result

    # 获取流程版本
    version = state.flow_version

    # 获取流程定义
    flow = self.get_flow_definition(flow_id, version)

    # flow 定义指纹（波次二任务②）：worker 的定义 TTLCache 有 300s 窗口、
    # console engine_sync 可直接覆盖 Redis 定义——挂起执行 resume 用
    # latest 版本定义时，若定义自启动后被改过，``_get_next_from_last``
    # 找不到节点/走错分支且无任何告警。指纹在 lease 内判定（与活
    # worker 的推进写串行化，防「他人持租约推进中却被误终态化」），
    # 不一致 → 终态化 error + raise ValueError（poison ack）——意图是
    # 可观测的终态而不是重投风暴。老状态 flow_hash 为 None → 跳过校验
    # （零回归）。指纹取**存储中的原始定义**（plaita#36 口径 v2）：同一份
    # 存储定义不受引擎版本/模型字段演进影响。
    current_flow_hash = self.get_flow_definition_hash(flow_id, version)

    # 解析流程定义
    logger.info("恢复流程执行: %s, 执行ID: %s, 恢复类型: %s", flow_id, execution_id, resume_type)

    # holder 嵌入注册表 instance id（#50），语义同 start 路径。
    holder = new_holder_token(
        prefix="resume", instance_id=getattr(self, "instance_id", None)
    )
    lease_value, fence_token = self._acquire_lease(execution_id, holder)
    if lease_value is None:
        raise ExecutionLeaseError(
            f"execution {execution_id} is leased by another worker; refuse concurrent resume"
        )

    fence_token_reset = None
    if fence_token is not None:
        # fencing 写侧守门（设计稿 §4.2）：本 resume 的所有状态落盘带
        # 世代号，FencedExecutionStorage 据此 CAS；finally 复位。
        fence_token_reset = set_current_fence_token(fence_token)

    try:
        # TOCTOU 复核（plaita#241）：上面的 load 发生在取租约**之前**。
        # 若窗口内另一 worker 已把执行推到终态并释放租约，本 worker 随后
        # 取得的是**新一代**租约——fencing 只拒绝旧世代写，不会拦它拿
        # 旧 state 继续 run_distributed（节点副作用重放、终态行被二次
        # 改写）。租约内重读并重跑前置判定；state/early_result 均以本次
        # 为准，不复用 acquire 前的快照。
        state, early_result = self._load_and_guard_resume_state(
            flow_id, execution_id, resume_type, lease_held=True
        )
        if early_result is not None:
            logger.info(
                "执行 %s 租约内复核发现状态已变更（status=%s）：拒绝本次 "
                "resume，不复用 acquire 前的旧 state（plaita#241）",
                execution_id, early_result.get("status"),
            )
            # 具名拒绝：区别于入口即终态的重复投递——本次是**赢了租约
            # 竞争、但盘上状态已由上一持有者推进**，测试/值守可据此
            # 识别 TOCTOU 窗口，而不是当成普通幂等短路。
            early_result["rejected_after_lease"] = True
            return early_result

        # plaita#247：#241 的复核只替换了 state，version/flow/current_flow_hash
        # 仍是 acquire 前快照算出的值。窗口内 flow_version 变更（同 flow_id 新
        # 版本 / 另一 start 载体回写）时，继续用旧 flow 执行并把结果写回新
        # version 的状态行 ⇒ 执行定义与状态身份不一致。复核后的 state 才是
        # 权威：版本变了就用新版本重算定义与指纹；hash 冲突仍走下方既有
        # mismatch 终态路径（绝不带着旧定义/旧指纹续跑）。
        if state.flow_version != version:
            logger.warning(
                "执行 %s 在 load→acquire 窗口内 flow_version 变更（%r → %r），"
                "按租约内复核的新版本重算 flow 定义与指纹（plaita#247）",
                execution_id, version, state.flow_version,
            )
            version = state.flow_version
            flow = self.get_flow_definition(flow_id, version)
            current_flow_hash = self.get_flow_definition_hash(flow_id, version)

        state_status = getattr(state, "status", "") or ""
        retry_wakeup = state_status == "error" and \
            ResumeType.coerce(resume_type) is ResumeType.RETRY
        if state_status == "running" and \
                ResumeType.coerce(resume_type) is ResumeType.RETRY:
            # running 执行没有失败节点可重跑（plaita#53 验收 2）：retry 在此的
            # 语义 = 从 checkpoint 派发后继节点（与 continue 同路），**必须真实
            # 重派**——禁止「受理后无动作」。显式留痕，值守从日志即可区分
            # 「真的推进了」与「白等」（租约被他人持有等不生效情形另见 run()
            # 的租约冲突日志）。
            logger.warning(
                "执行 %s 处于 running，resume_type=retry 按「从断点续跑」处理"
                "（running 执行无失败节点可重跑），本次将从 checkpoint 派发后继节点",
                execution_id,
            )

        # 陈旧 event/timeout 载体的归一化（plaita#258）：resume 在流内
        # 推进并落过 step_persist 后，若发生瞬态 StatePersistError / 进程
        # 崩溃，消息按 at-least-once 重投，但 resume_type 仍是 event/timeout
        # ———而挂起订阅已在第一次 resume 时注销、$LAST_NODE 已不是 pending
        # 挂起点。策略层会抛 ResumeGuardError（协议错误），旧 run() 分支据此
        # ack 唯一载体 ⇒ 执行永久停在 running。这里在**取得租约并重读状态
        # 之后**确认「执行仍活、但决议无处可落」，把本次意图改写为 continue
        # （剩余工作就是从 checkpoint 续跑，已完成节点不重放）。租约内
        # 复核保证：另一 worker 已推进到**新的挂起点**时 status=suspended，
        # 不会走本分支，陈旧 resume 仍由策略层的订阅绑定守卫拒绝（#221）。
        if resume_type in (ResumeType.EVENT, ResumeType.TIMEOUT, "event", "timeout"):
            stale_reason = self._resume_carrier_stale_reason(state, flow)
            if stale_reason:
                logger.warning(
                    "执行 %s 的 %s resume 载体已错过挂起点（%s）；订阅早已注销、"
                    "执行仍 running，本次归一化为 continue 从 checkpoint 续跑"
                    "（plaita#258）",
                    execution_id, resume_type, stale_reason,
                )
                resume_type = ResumeType.CONTINUE.value

        # XCLAIM 恢复路径取消检查点（设计稿 §3.1）：crash 后 cancel 消息
        # 或任一 resume 消息被重投，若取消标志在（控制面对运行中执行写的
        # 意图键），直接终态化而非继续推进。上一 checkpoint 已落盘，
        # cancelled 的 context 即取消点前一步的完整快照（§3.3）。
        if self._cancel_requested(execution_id):
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_at_resume_entry")
            logger.info("执行 %s 在 resume 入口命中取消标志，终态化", execution_id)
            return {
                "execution_id": execution_id,
                "status": "cancelled",
                "cancelled_at_resume": True,
            }

        # flow 定义指纹校验（波次二任务②）在 G1 retry 唤醒**之前**：
        # 定义被改时执行保持 error（hash 不匹配信息落盘），修复定义后
        # 仍可再 retry——G1 的「error 态可反复唤醒」语义不被破坏。
        # 取消是用户意图，已在上方优先放行。写盘走 _persist_state_or_raise
        # （带本 resume 的 fence 世代），随后 raise FlowHashMismatchError
        # → run() 按 ValueError poison ack（终态已可观测，重投无意义）。
        stored_flow_hash = getattr(state, "flow_hash", None)
        stored_algo = getattr(state, "flow_hash_algo", None)
        algo_changed = bool(stored_algo) and stored_algo != FLOW_HASH_ALGO
        allow_hash_change = bool(
            isinstance(data, dict) and data.get(ALLOW_FLOW_HASH_CHANGE_KEY)
        )
        hash_decision = classify_flow_hash_change(
            stored_hash=stored_flow_hash,
            current_hash=current_flow_hash,
            stored_algo=stored_algo,
            current_algo=FLOW_HASH_ALGO,
            allow_change=allow_hash_change,
        )
        if hash_decision in ("mismatch", "accepted"):
            if hash_decision == "accepted":
                # 升级后的人工裁决入口：显式承认「换算法/换定义」并续跑，
                # 必须留痕（WARNING + 状态里刷新指纹），不静默放行。
                logger.warning(
                    "执行 %s 的 flow 指纹变化被显式放行续跑（%s）：stored=%s(%s) current=%s(%s)",
                    execution_id,
                    ALLOW_FLOW_HASH_CHANGE_KEY,
                    stored_flow_hash[:12],
                    stored_algo or "unknown",
                    current_flow_hash[:12],
                    FLOW_HASH_ALGO,
                )
                state.flow_hash = current_flow_hash
                state.flow_hash_algo = FLOW_HASH_ALGO
                self._persist_state_or_raise(execution_id, state, "flow_hash_change_accepted")
            else:
                mismatch_msg = (
                    "flow 定义自启动后已变更，hash 不匹配，执行无法安全续跑 "
                    f"(execution_id={execution_id}, flow_id={flow_id}, "
                    f"stored_hash={stored_flow_hash[:12]}..., "
                    f"current_hash={current_flow_hash[:12]}..., "
                    f"stored_algo={stored_algo or 'unknown'}, current_algo={FLOW_HASH_ALGO})"
                )
                logger.error(mismatch_msg)
                state.status = "error"
                state.error = {
                    "message": mismatch_msg,
                    # plaita#93：本处终态化不由异常驱动（有专门的字段承载
                    # 细节），但落盘面同样要给下游一个机器可判别的身份：
                    # code 用引擎的通用 flow 级错误码（-500），error_type
                    # 是本处的 worker 级判据（不是节点业务码）。
                    "code": _FLOW_ERROR_CODE,
                    "error_type": "flow_hash_mismatch",
                    "stored_flow_hash": stored_flow_hash,
                    "current_flow_hash": current_flow_hash,
                    "stored_flow_hash_algo": stored_algo,
                    "current_flow_hash_algo": FLOW_HASH_ALGO,
                    # 升级疑似（算法标记变了且哈希也不同）：给运维明确下一步，
                    # 而不是只丢一句「hash 不匹配」。
                    "upgrade_suspected": algo_changed,
                    "hint": (
                        "引擎升级可能改变指纹算法；确认流程定义未变且可接受后，"
                        f"在 resume data 里带 {ALLOW_FLOW_HASH_CHANGE_KEY}: true 显式放行"
                    ) if algo_changed else None,
                }
                state.end_time = datetime.now().isoformat()
                self._persist_state_or_raise(execution_id, state, "flow_hash_mismatch")
                raise FlowHashMismatchError(mismatch_msg)
        elif hash_decision == "refresh":
            # 哈希相同 → 定义确实没变，只是算法标记变了（引擎升级）：续跑并刷新标记
            logger.info(
                "执行 %s 的 flow 指纹一致但算法标记升级（%s → %s），续跑并刷新标记",
                execution_id,
                stored_algo,
                FLOW_HASH_ALGO,
            )
            state.flow_hash_algo = FLOW_HASH_ALGO

        # 引擎版本跨 minor：只告警不拦截（硬门由 flow_hash 承担，这里给可观测性）
        stored_engine = getattr(state, "engine_version", None)
        current_engine = _engine_version()
        if stored_engine and _major_minor(stored_engine) != _major_minor(current_engine):
            logger.warning(
                "执行 %s 由引擎 %s 创建，当前 %s：跨 minor 续跑，"
                "若流程行为与预期不符请复核定义与节点实现",
                execution_id,
                stored_engine,
                current_engine,
            )

        # G1 retry 唤醒：error → running 翻转并先落盘（观察者/监控立即
        # 可见；若本 worker 接着硬死，行是 running 而非 error——再一轮
        # retry 仍可放行，zombie 判定也不误报「终态」）。checkpoint 原样，
        # saved_context 由下方 run_distributed 携带。
        # 与任务①组合语义：唤醒即清零节点重试计数键——预算耗尽终态化的
        # 执行经人工 retry 唤醒后拿全新预算，否则唤醒的执行第一次失败就
        # 立刻再耗尽，G1 形同虚设。翻转落盘成功后才清零（落盘失败抛
        # StatePersistError 走重投，唤醒未生效不清预算）。
        # 不传 node_id ⇒ 清**该执行下所有节点**的计数键（plaita#123 起键按
        # 节点隔离，唤醒是执行级意图，必须整体重来）。
        if retry_wakeup and not state.context:
            # plaita#240：retry 的能力前提是盘上有 checkpoint。此前先翻
            # running 再落盘、等 run_distributed 才由策略层抛
            # ResumeGuardError；协议错误分支虽「保持原状」，但原状已经被
            # 改成 running（error/end_time 也被清空）——非法 error→running
            # 迁移 + 失败证据丢失。必须在**任何状态变更之前**拒绝。
            # 保持 error 原状（不落盘）、不创建 FlowExecution、不触发节点。
            message = (
                "resume_type='retry' requires a saved checkpoint, but the "
                f"execution has none; execution {execution_id} remains error "
                "(status=error, error/end_time preserved)"
            )
            logger.warning(
                "执行 %s 收到 retry 但 checkpoint 为空，拒绝唤醒（保持 error 原状）",
                execution_id,
            )
            raise ResumeProtocolError(message, execution_id=execution_id)
        if retry_wakeup:
            state.status = "running"
            state.error = None
            state.end_time = None
            self._persist_state_or_raise(execution_id, state, "retry_wakeup")
            self._reset_node_retry_counter(execution_id)
            # 记一次唤醒（plaita#73 补丁）：唤醒会重置节点预算，必须计数
            # 才能让「唤醒→重置→耗尽→再唤醒」收敛。达上限的拒绝在上方
            # 入口处（`_g1_wakeup_budget_exhausted`）。
            self._record_g1_wakeup(execution_id)
            logger.info("执行 %s error 态经 retry 放行（重试计数已清零），从断点步进", execution_id)

        # 复用同一个 FlowExecution 贯穿恢复后的所有分布式步骤
        execution = FlowExecution(
            event_bus=self.event_bus,
            callback_handlers=self._handlers_for(execution_id),
        )
        execution.mode = ExecutionMode.DISTRIBUTED
        self._bind_observers(execution)
        # 登记看门狗（波次②）：持租约期间每 TTL/3 续租，防长步 > TTL
        # 被 XCLAIM 抢占双跑（fence 世代随登记带上，心跳写要过世代 CAS）
        self._register_lease_watch(
            execution_id, lease_value, execution, fence_token=fence_token
        )
        # 登记取消监听（波次③）：命中标志键即中止在途节点（与租约登记
        # 同窗口；finally 一并撤销）
        self._register_cancel_watch(execution_id, execution)

        # 直接使用 run_distributed 恢复执行
        # ``execution_id`` 必须显式透传（plaita#123 配套）：不传时引擎侧
        # 的 execution_id 与 worker 脱钩——resume 走 ``context.context =
        # saved_context`` 整体替换 checkpoint，若 checkpoint 缺
        # ``$EXECUTION_ID``（resume 常见），引擎 execution_id 为空串 ⇒
        # ``result.execution_id`` 回空、且重试计数键退化成
        # ``noderetry::{node}``（无执行 id），运维无法从键反查执行。
        # start 路径（本文件 ``start_flow``）早已是同款正确范式。
        #
        # ``params``（#122）：仅在 ``state.context`` 为空（假值）时被引擎
        # 用于 ``setup_flow``——即 start→resume 载体转换撞上的「首节点还
        # 没成功过」窗口；checkpoint 非空时引擎走真值分支，params 不参与
        # （续跑语义零变化，见 ``DistributedStrategy.execute``）。
        result = execution.run_distributed(
            flow,
            params=params,
            saved_context=state.context,
            resume_type=resume_type,
            resume_data=data,
            execution_id=execution_id,
        )

        # 处理执行结果
        final_result = self._process_execution_result(
            flow,
            result,
            state,
            execution,
            lease_execution_id=execution_id,
            lease_holder=lease_value,
            delivery_count=delivery_count,
        )

        return final_result

    except ExecutionLeaseError:
        raise
    except (StatePersistError, ServiceDispatchError, NodeExecutionRetryableError):
        # 落盘/派发失败：执行状态保持原样（is_end 前失败 → 仍 running，
        # 挂起派发失败 → 仍 suspended），消息不 ack 走 at-least-once 重投，
        # 下轮从 checkpoint / 挂起节点重入。绝不落 error 终态——重投消息
        # 会命中上方 already_terminal 短路被 ack，执行永久卡死。
        # NodeExecutionRetryableError（波次二任务①）同理：节点失败未
        # 终态化，重投后从 checkpoint 重跑失败节点；finally 释放租约，
        # 重投消息可重新取得租约。
        raise
    except FlowHashMismatchError:
        # flow 定义指纹不匹配（波次二任务②）：执行已在 raise 前终态化
        # error（可观测终态），原样上抛给 run() 按 ValueError poison
        # ack——重投只会命中 already_terminal 短路，终态化已完成，无意义。
        # 不落进下面的通用 except 二次终态化/包 RuntimeError。
        raise
    except ResumeProtocolError:
        # plaita#240：预检在**状态变更前**抛出的协议错误，必须原样上抛给
        # run()（ack、不终态化）——不能被下面通用 except 当作执行失败
        # 二次终态化 error。策略层守卫路径仍由 _chain_has_resume_guard_error
        # 在通用 except 内包成同类异常。
        raise
    except Exception as e:
        logger.error("恢复流程执行出错: %s", _scrub_exception_text(e))

        # 波次③步内取消：被恢复的节点在途命中取消监听 → 引擎抛
        # FlowCancelledException（归一化后挂 __cause__）或协作节点自杀抛
        # 普通异常。两条判据任一命中即终态化 cancelled，绝不写 error。
        if _chain_has_cancellation(e) or self._cancel_requested(execution_id):
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_in_node")
            self._finalize_observers()
            logger.info(
                "执行 %s 在恢复节点执行期响应取消，终态化 cancelled", execution_id
            )
            return {
                "execution_id": execution_id,
                "status": "cancelled",
            }

        # 挂起守卫豁免（#33）：**只**豁免策略层守卫 ``ResumeGuardError``
        # （continue/retry 试图绕过 pending 挂起节点、resume 打在无挂起/
        # 非挂起 checkpoint 上）——那是调用方协议与执行状态不匹配，不是
        # 执行自身失败。终态化 error 会把本可被 event/cancel/timeout 唤醒的
        # 挂起执行永久打封（终态短路从此拒绝一切 resume 类型，死局），且这类
        # 错误天然由 at-least-once 重投产生（重复投递的 event resume 打在已
        # 推进的 checkpoint 上同族），终态化会误杀健康执行。
        # 豁免为「保持原状 + 上抛」：suspended 执行保持 suspended，消息被
        # ack（重投也只会再次命中同一守卫，重投无意义）；running 执行保持
        # checkpoint 现状。
        # 裸 ``ResumeError``（节点 ``resume()`` 抛错被策略层包的那个，
        # 如事件数据畸形）**不豁免**：静默保持原状 = 无 error 记录、永远等
        # 不到决议的哑执行；交给下方现状路径（可重试判据 → 终态化 error）
        # 才可观测、可人工 retry。
        if _chain_has_resume_guard_error(e):
            logger.warning(
                "执行 %s (status=%s) 收到与挂起状态不匹配的 resume 请求，"
                "协议错误不终态化，执行保持原状: %s",
                execution_id, getattr(state, "status", "?"), _scrub_url_text(e),
            )
            self._finalize_observers()
            raise ResumeProtocolError(
                f"resume 协议错误（执行保持原状 status={getattr(state, 'status', '?')}）: {e}",
                execution_id=execution_id,
            ) from e

        # 节点级有界重试（波次二任务①）：同 _process_execution_result，
        # 瞬态节点失败不终态化、消息等重投；否则现状终态化 error。
        # 失败节点 id 取自异常链（plaita#123：计数键按节点隔离）；取不到
        # 时计数退化为执行维度，终态化 error 里也就没有按节点的次数。
        failed_node = _node_id_from_chain(e)
        retry_exc = self._node_failure_retry_decision(
            e, execution_id, delivery_count, failed_node
        )
        if retry_exc is not None:
            raise retry_exc from e

        # 更新执行状态为错误
        state.status = "error"
        retries = max(
            0, self._read_node_retry_counter(execution_id, failed_node) - 1
        )
        if retries > 0:
            state.error = {
                "message": f"节点执行失败（重试 {retries} 次后仍失败）: {e}",
                "node_retries": retries,
                # plaita#93：终态化那一刻把引擎已算出的 code / error_type 与
                # HTTP 明细摘要如实落盘（message 保留，老消费方零影响）。
                **_error_fields(e),
            }
        else:
            state.error = {"message": str(e), **_error_fields(e)}
        if failed_node:
            state.error["failed_node"] = failed_node
        state.end_time = datetime.now().isoformat()

        self._persist_state_or_raise(execution_id, state, "resume_error_handler")
        self._finalize_observers()

        # plaita#73：终态化**已完成**，必须 poison ack 终止重投循环。
        # 原先抛裸 RuntimeError → 落到 run() 兜底 except → 不 ack → 重投
        # → status=error 只放行 retry → 再撞同一确定性失败 → 无限循环
        # （且全程不碰重试计数键，预算永不耗尽）。
        raise NodeFailureTerminalizedError(
            f"恢复流程执行出错（已终态化 error，消息 poison ack）: {e}"
        )
    finally:
        self._unregister_lease_watch(execution_id, lease_value)
        self._unregister_cancel_watch(execution_id)
        if fence_token_reset is not None:
            reset_current_fence_token(fence_token_reset)
        # release 对整个租约 value 串 compare（fencing 档 = {holder}:{gen}）
        self.execution_lease.release(execution_id, lease_value)

RedisFlowWorker

RedisFlowWorker(redis_url: str, queue_name: str, execution_storage: ExecutionStorage, flow_storage: FlowStorage, event_bus: EventBus = None, cache_size: int = 100, cache_ttl: int = 300, enable_registry: bool = True, registry_ttl: int = DEFAULT_TTL, heartbeat_interval: int = DEFAULT_HEARTBEAT_INTERVAL, enable_redis_logging: bool = True, redis_client: Redis = None, callback_handlers=None, consumer_group: str = DEFAULT_CONSUMER_GROUP, consumer_name: Optional[str] = None, claim_min_idle_ms: int = DEFAULT_CLAIM_MIN_IDLE_MS, lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS, execution_lease: Optional[ExecutionLease] = None, max_deliveries: int = DEFAULT_MAX_DELIVERIES, dlq_key: Optional[str] = None, read_block_ms: int = 1000, concurrency: int = 1, min_free_disk_gib: Optional[float] = None, disk_guard_path: Optional[str] = None, disk_guard_poll_seconds: float = DEFAULT_DISK_GUARD_POLL_SECONDS, watchdog_interval_seconds: Optional[float] = None, cancel_poll_seconds: Optional[float] = None, residue_sweep_interval_seconds: Optional[float] = None, metrics_port: Optional[int] = None, metrics_host: Optional[str] = None, alert_webhook: Optional[str] = None)

Bases: RegistryMixin, ControlMixin, FlowWorker

基于 Redis Stream 队列的流程工作器（服务注册 / 心跳 / 远程控制）。

任务队列语义为 at-least-once：consumer group + XACK；处理成功前 崩溃则 pending 可被其他 consumer 在 claim_min_idle_ms 后回收重投。 resume 通过 Redis execution lease 保证同一 execution_id 最多一个 worker 推进。 控制面硬依赖 Redis。

源代码位于： plaita/server/flow_worker.py

def __init__(
    self,
    redis_url: str,
    queue_name: str,
    execution_storage: ExecutionStorage,
    flow_storage: FlowStorage,
    event_bus: EventBus = None,
    cache_size: int = 100,
    cache_ttl: int = 300,
    enable_registry: bool = True,
    registry_ttl: int = ServiceRegistry.DEFAULT_TTL,
    heartbeat_interval: int = ServiceRegistry.DEFAULT_HEARTBEAT_INTERVAL,
    enable_redis_logging: bool = True,
    redis_client: Redis = None,
    callback_handlers=None,
    consumer_group: str = DEFAULT_CONSUMER_GROUP,
    consumer_name: Optional[str] = None,
    claim_min_idle_ms: int = DEFAULT_CLAIM_MIN_IDLE_MS,
    lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS,
    execution_lease: Optional[ExecutionLease] = None,
    max_deliveries: int = DEFAULT_MAX_DELIVERIES,
    dlq_key: Optional[str] = None,
    read_block_ms: int = 1_000,
    concurrency: int = 1,
    min_free_disk_gib: Optional[float] = None,
    disk_guard_path: Optional[str] = None,
    disk_guard_poll_seconds: float = DEFAULT_DISK_GUARD_POLL_SECONDS,
    watchdog_interval_seconds: Optional[float] = None,
    cancel_poll_seconds: Optional[float] = None,
    residue_sweep_interval_seconds: Optional[float] = None,
    metrics_port: Optional[int] = None,
    metrics_host: Optional[str] = None,
    alert_webhook: Optional[str] = None,
):
    # A2R20-05：队列/租约共用客户端——connect 5s / read 30s 显式声明
    # （read 必须大于 XREADGROUP BLOCK 窗口，见 task_queue）。
    if redis_client is None:
        redis_client = make_redis_client(redis_url)
        pool_kwargs = redis_client.connection_pool.connection_kwargs
        logger.info(
            "Redis 超时预算: connect=%ss read=%ss（PLAITA_REDIS_* 可覆盖）",
            pool_kwargs.get("socket_connect_timeout"),
            pool_kwargs.get("socket_timeout"),
        )
    super().__init__(
        execution_storage,
        flow_storage,
        event_bus,
        cache_size,
        cache_ttl,
        callback_handlers=callback_handlers,
        execution_lease=execution_lease or TenantRoutingExecutionLease(redis_client),
        lease_ttl_seconds=lease_ttl_seconds,
    )
    self.redis_url = redis_url
    self.redis_client = redis_client
    self.queue_name = queue_name
    self._running = False
    # 消费阻塞窗口上限（毫秒）——见 run() 主循环内的分片说明
    self.read_block_ms = max(100, int(read_block_ms))
    # 并发消费线程数（2026-10-07）：>1 时 run() 起 N 个消费线程共享本实例，
    # 单机可同时处理 N 个任务（原为严格串行——一次 read 一条、处理完再读）。
    # 每线程各自持有一个独立 queue 对象（独立 Redis 连接），避免连接竞争；
    # 消息级 at-least-once 与执行级租约语义不变（并发下同一 execution 仍由
    # 租约串行化）。默认 1 = 零行为变化。
    self.concurrency = max(1, int(concurrency))
    # claim 前本机盘预检（#49）：多 worker 池下，低盘机器的 flow preflight
    # 要等「领到手」才判盘不足 → retry-later 白跑一轮（实测一夜 11 次），
    # 且把瞬时低盘放大成数小时退避停机；健康 worker 全程闲置。阈值 >0 时，
    # 本机可用盘低于它就不 XREADGROUP（空转按 disk_guard_poll_seconds
    # 重探，盘回线自动恢复，无需重启）。默认 0 = 关闭 = 零行为变化。
    self._min_free_disk_gib = (
        _min_free_disk_gib_from_env()
        if min_free_disk_gib is None
        else max(0.0, float(min_free_disk_gib))
    )
    self._disk_guard_path = disk_guard_path or _disk_guard_path_from_env()
    self._disk_guard_poll_seconds = max(0.1, float(disk_guard_poll_seconds))
    self._disk_guard_active = False
    self._active_task_count = 0
    self._active_count_lock = threading.Lock()
    self._log_handler = None
    self._consumer_group = consumer_group
    self._claim_min_idle_ms = claim_min_idle_ms
    self._consumer_name = consumer_name
    if int(max_deliveries) < 1:
        logger.warning(
            "max_deliveries=%s ≤0，钳制为 1（启动预检会拒绝该配置，B2R31-03）",
            max_deliveries,
        )
    self._max_deliveries = max(1, int(max_deliveries))
    self._dlq_key = dlq_key
    self._task_queue: Optional[RedisStreamTaskQueue] = None

    # B2R31-01/02/03：本进程的配置视图与三维代际。库内构造用非 strict
    # （非法值 WARNING + 安全默认，不阻断嵌入式使用）；CLI 入口另走
    # preflight 严格校验。显式构造参数优先于 env。
    resolved_dlq = dlq_key
    if resolved_dlq is None:
        try:
            resolved_dlq = resolve_dlq_key(queue_name, empty_policy="default")
        except ConfigError as exc:
            logger.warning(
                "DLQ 键解析失败，按默认键继续（配置预检会拒绝启动）: %s",
                _scrub_exception_text(exc),
            )
            resolved_dlq = f"{queue_name}:dlq"
    self._runtime_config = load_runtime_config(
        "worker",
        strict=False,
        overrides={
            "queue_name": queue_name,
            "redis_url": redis_url,
            "consumer_group": consumer_group,
            "dlq_key": resolved_dlq,
            "max_deliveries": max(1, int(max_deliveries)),
            "lease_ttl_seconds": max(1, int(lease_ttl_seconds)),
            "task_max_age_seconds": task_max_age_seconds(),
            # 与队列构造实际采用的钳制值同源，避免代际哈希与实际预算分叉。
            "dlq_max_len": _dlq_max_len_from_env(),
        },
    )
    self._config_generation: Optional[ConfigGeneration] = (
        self._runtime_config.generation
    )
    if self._config_generation is not None:
        self._config_family = self._config_generation.family
        self._config_generation_id = self._config_generation.generation
        self._budget_generation = self._config_generation.budget
        self._config_effective_at = self._config_generation.effective_at
    else:  # pragma: no cover - build_generation always returns an object
        self._config_family = LEGACY_GENERATION
        self._config_generation_id = LEGACY_GENERATION
        self._budget_generation = LEGACY_GENERATION
        self._config_effective_at = None
    self._tenant_status_epoch = self._runtime_config.tenant_status_epoch
    self._dlq_key = resolved_dlq
    log_config_effective(logger, self._runtime_config)

    # #26 观测导出：``/metrics`` 抓取端（0 = 不启动）与死信告警 webhook。
    # 队列计数器（enqueued/acked/dead_lettered…）是**进程内**状态，只有
    # worker 自己持有——console 无法代它导出，故端点必须跑在本进程里。
    self._metrics_port = (
        _metrics_port_from_env() if metrics_port is None else max(0, int(metrics_port))
    )
    self._metrics_host = metrics_host or _metrics_host_from_env()
    self._metrics_server: Optional[MetricsHttpServer] = None
    self._alert_webhook = (
        alert_webhook if alert_webhook is not None else _alert_webhook_from_env()
    )
    self._alerter: Optional[WebhookAlerter] = None

    # 启动期注册重试（A2R19-03）：注册失败时后台线程按有界退避重试；
    # 成功前 ``_registry_registered`` 保持 False（状态/就绪侧不宣称 ready）。
    self._registry_registered = False
    self._registry_retry_thread: Optional[threading.Thread] = None
    self._registry_retry_stop = threading.Event()
    # S18-02A：生命周期状态即时落盘失败计数（/metrics 可观测量）。
    self._registry_status_flush_failures = 0

    # 租约看门狗（波次② §4.1）：登记活跃 (execution_id → 租约值串/
    # FlowExecution/租户)，每 watchdog_interval_seconds（默认 TTL/3）
    # renew 一次，保证存活 worker 的租约永不空窗——XCLAIM 只会捡到
    # 真死 worker 的消息，抢占双跑被根除，fencing 是其下第二道保险。
    self._watchdog_interval_seconds = watchdog_interval_seconds
    self._watchdog_thread: Optional[threading.Thread] = None
    self._watchdog_stop = threading.Event()
    self._lease_watch_lock = threading.Lock()
    # execution_id -> (lease_value, execution, tenant_id)
    self._lease_watch: Dict[str, tuple] = {}
    self._lease_lost: Set[str] = set()
    # 失租后仍在跑的执行：只发布进度心跳，不续租、不重复 cancel。
    # 存在的原因见 `_watchdog_renew_once`：租约被别的消息处理释放后，
    # 执行对象仍会在沙箱里正常推进（实测 6caf8289），若停止发布心跳，
    # keeper 会按「末节点停滞 >1800s」判死一个健康 run。
    self._lease_orphan_watch: Dict[str, tuple] = {}

    # 取消监听（波次③ §3.3 步内中断）：登记活跃 (execution_id →
    # FlowExecution/租户)，每 cancel_poll_seconds（默认 1s）轮询取消标志键，
    # 命中即 execution.cancel() 中止在途节点。登记表/停止位已在基类建好，
    # 此处只接轮询间隔旋钮（单测可缩短）。
    if cancel_poll_seconds is not None:
        self._cancel_poll_seconds = cancel_poll_seconds

    # 队列残留回收（#43）：启动时扫一次 + 每 residue_sweep_interval_seconds
    # 兜底一次，清掉「已 ack 未 XDEL」的 Stream 残留条目（它们只增不减地
    # 计入 XLEN，让运维把队列读数读成假阳性）。非阻塞锁保证并发消费下
    # 同一时刻只有一个扫描在跑（幂等，重复扫描只是浪费一次往返）。
    self._residue_sweep_interval_seconds = (
        DEFAULT_RESIDUE_SWEEP_INTERVAL_SECONDS
        if residue_sweep_interval_seconds is None
        else float(residue_sweep_interval_seconds)
    )
    self._last_residue_sweep: Optional[float] = None
    self._residue_sweep_lock = threading.Lock()

    # 服务注册
    self._enable_registry = enable_registry
    if self._enable_registry:
        self.init_registry(
            redis_client=self.redis_client,
            service_type=self.SERVICE_TYPE,
            metadata={
                "queue_name": queue_name,
                # B75-01 §6：队列声明的契约版本。派发侧据此区分「新
                # worker 明确声明消费队列」与 pre-#158/自定义 worker 的
                # 缺失 metadata（unknown ≠ default）。
                "queue_contract_version": 2,
                "consumer_group": self._consumer_group,
                "dlq_key": self._dlq_key,
                # B2R31-01：代际与脱敏关键值进注册 metadata——控制面
                # （/api/cluster/instances、capacity、drift 预检）据此
                # 发现混跑；旧 build 忽略未知键。
                "config_family": self._config_family,
                "config_generation": self._config_generation_id,
                "budget_generation": self._budget_generation,
                "config_effective_at": self._config_effective_at,
                "max_deliveries": self._max_deliveries,
                "lease_ttl_seconds": lease_ttl_seconds,
                "task_max_age_seconds": self._runtime_config.task_max_age_seconds,
                "dlq_max_len": self._runtime_config.dlq_max_len,
                "deployment_epoch": self._runtime_config.deployment_epoch,
                "tenant_status_epoch": self._tenant_status_epoch,
                # plaita#158：本字典会被 registry.register() 原样 ``SETEX`` **持久化**
                # 进 ``plaita:registry:flow_worker:<instance_id>``——注册键名与
                # metadata 键语义不变（仍是 ``redis_url``），只有它的**值**是掩码
                # 形态，另加 host/port/db 三件套保住运维诊断价值。
                **_redis_registry_metadata(redis_url),
                "cache_size": cache_size,
                "cache_ttl": cache_ttl,
            },
            ttl=registry_ttl,
            heartbeat_interval=heartbeat_interval,
            instance_id=self._runtime_config.instance_id,
        )

        # 初始化控制监听
        self.init_control(
            redis_client=self.redis_client,
            instance_id=self._service_info.instance_id if self._service_info else "unknown"
        )

        # #42（W10-a）：把容量快照写进 metadata（并立即落一次注册键）。
        # 放在 init_control 之后：`_service_info` 已建好；此时 `_registry`
        # 尚未 start_heartbeat，直推一次让派发侧**首个心跳前**就能看到容量
        # （registry.register() 在 init_registry 内还没执行，本次推送顺带
        # 完成注册键写入——幂等，register() 之后再覆盖一次同样的值）。
        self._report_capacity()

    # Redis 日志处理器
    self._enable_redis_logging = enable_redis_logging
    if self._enable_redis_logging and self._service_info:
        self._log_handler = setup_redis_logging(
            redis_client=self.redis_client,
            service_type=self.SERVICE_TYPE,
            instance_id=self._service_info.instance_id,
            logger_name="plaita",
            level=logging.INFO
        )

registry_registered property

registry_registered: bool

本进程当前是否已在注册表中（重试成功前为 False）。

instance_id property

instance_id: Optional[str]

获取实例 ID

run

run(wait_for_inflight: bool = False)

从 Redis Stream consumer group 拉取 start / resume 任务。

成功处理后 XACK；崩溃或未 ack 的消息留在 pending，超时后可被 其他 consumer XCLAIM 回收（at-least-once，业务侧应幂等）。

wait_for_inflight=True（S18-01）：退出时 stop(wait=True)， 在看门狗/注销之前有界等待在途步收尾（库模式推荐）；CLI 默认 False， 以进程退出为硬中止边界。

源代码位于： plaita/server/flow_worker.py

def run(self, wait_for_inflight: bool = False):
    """
    从 Redis Stream consumer group 拉取 ``start`` / ``resume`` 任务。

    成功处理后 ``XACK``；崩溃或未 ack 的消息留在 pending，超时后可被
    其他 consumer ``XCLAIM`` 回收（at-least-once，业务侧应幂等）。

    ``wait_for_inflight=True``（S18-01）：退出时 ``stop(wait=True)``，
    在看门狗/注销之前有界等待在途步收尾（库模式推荐）；CLI 默认 False，
    以进程退出为硬中止边界。
    """
    # plaita#239 phase 2（owner 2026-10-18 裁决）：单租户部署迁移——
    # 服务入口**显式**声明 default 租户；写路径对隐式 default 已
    # fail-closed（TenantContextMissingError）。每条任务消息仍由
    # `_dispatch_task` 按其 tenant_id 覆盖（显式/derived）。
    set_current_tenant(DEFAULT_TENANT_ID, source=TENANT_SOURCE_EXPLICIT)
    self._running = True
    queue = self._get_task_queue()
    queue.ensure_group()

    # 队列残留回收（#43）：启动时先扫一次，之后由消费循环按间隔兜底
    self._sweep_residue_if_due(queue)

    # 租约看门狗（波次②）：持租约执行每 TTL/3 续租，防长步被 XCLAIM 抢占
    self._start_lease_watchdog()
    # 取消监听（波次③）：轮询取消标志键，命中即中止在途节点
    self._start_cancel_watcher()

    # 注册服务：失败不放弃——后台按有界退避重试，Redis 恢复后自愈（A2R19-03）
    if self._enable_registry:
        self._ensure_service_registration()
        # 启动控制监听
        self.start_control_listener()

    # #26 观测导出：/metrics 抓取端 + 死信告警 webhook（未配置 = 无操作）
    self._start_metrics_server()
    self._start_alerting()

    logger.info(
        "流程工作器已启动，监听 stream: %s (group=%s, consumer=%s, concurrency=%d)",
        self.queue_name,
        self._consumer_group,
        queue.consumer_name,
        self.concurrency,
    )

    # 并发消费（2026-10-07）：concurrency=1 时原地串行（零行为变化）；
    # >1 时起 N 个消费线程共享本实例。线程各自跑 _consume_loop，靠
    # XREADGROUP 的天然互斥（Redis 保证一条消息只投给一个 consumer 的一次读）
    # 分到不同任务；共享 queue 对象线程安全（redis-py 连接池）。
    threads: list = []
    try:
        if self.concurrency <= 1:
            self._consume_loop(queue)
        else:
            for i in range(self.concurrency):
                t = threading.Thread(
                    target=self._consume_loop,
                    args=(queue,),
                    name=f"flow-consume-{i}",
                    daemon=True,
                )
                t.start()
                threads.append(t)
            logger.info("已启动 %d 个并发消费线程", len(threads))
            # 主线程等任一消费线程退出（正常只在 stop() 后发生）
            for t in threads:
                while t.is_alive():
                    t.join(timeout=1.0)
                    if not self._running:
                        break
    finally:
        self.stop(wait=wait_for_inflight)

capacity_snapshot

capacity_snapshot() -> Dict[str, Any]

本机当前容量快照（纯函数式取值，不写注册表）。

free_disk_gib 为 None = 盘探针失败/未启用盘闸路径不可读——与

49 盘闸同口径「探不到就放行」，不写成 0（0 会被派发侧当成

「真的没盘」而拒单，把探针故障放大成停摆）。

源代码位于： plaita/server/flow_worker.py

def capacity_snapshot(self) -> Dict[str, Any]:
    """本机当前容量快照（纯函数式取值，不写注册表）。

    ``free_disk_gib`` 为 ``None`` = 盘探针失败/未启用盘闸路径不可读——与
    #49 盘闸同口径「探不到就放行」，**不**写成 0（0 会被派发侧当成
    「真的没盘」而拒单，把探针故障放大成停摆）。
    """
    block = build_capacity_block(
        free_disk_gib=_free_disk_gib(getattr(self, "_disk_guard_path", ".") or "."),
        min_free_disk_gib=getattr(self, "_min_free_disk_gib", 0.0) or 0.0,
        disk_blocked=bool(self.__dict__.get("_disk_guard_active")),
        active_tasks=getattr(self, "_active_task_count", 0) or 0,
        concurrency=getattr(self, "concurrency", 1) or 1,
        config_family=getattr(self, "_config_family", None),
        config_generation=getattr(self, "_config_generation_id", None),
        budget_generation=getattr(self, "_budget_generation", None),
        deployment_epoch=(
            self._runtime_config.deployment_epoch
            if getattr(self, "_runtime_config", None) is not None
            else None
        ),
    )
    return block

request_drain

request_drain(reason: str = 'manual') -> None

进入 draining（D1）：不再领新任务、注册表标 draining，并启动有界等待。

有界是关键：单个节点步可能跑几十分钟（如 HITL gate），等它做完等于把 停机窗口拉到不可接受。PLAITA_WORKER_DRAIN_TIMEOUT 是停止领取/ 停止等待的上界（advisory），不是「节点中止」的上界：超时后 _force_stop_after_drain 尽力 cancel 在途执行并让 run() 收尾，但不 承诺节点在 N 秒内停下。若步在进程退出前完成，仍允许 ack；若进程先 退出，消息留 pending 由其他 worker 经 XCLAIM 从步界检查点续跑 （那一步会重放，业务节点须幂等）。硬中断是 B18-01，未实现。

源代码位于： plaita/server/flow_worker.py

def request_drain(self, reason: str = "manual") -> None:
    """进入 draining（D1）：不再领新任务、注册表标 draining，并启动有界等待。

    有界是关键：单个节点步可能跑几十分钟（如 HITL gate），等它做完等于把
    停机窗口拉到不可接受。``PLAITA_WORKER_DRAIN_TIMEOUT`` 是**停止领取/
    停止等待**的上界（advisory），不是「节点中止」的上界：超时后
    ``_force_stop_after_drain`` 尽力 cancel 在途执行并让 run() 收尾，但不
    承诺节点在 N 秒内停下。若步在进程退出前完成，仍允许 ack；若进程先
    退出，消息留 pending 由其他 worker 经 XCLAIM 从**步界检查点**续跑
    （那一步会重放，业务节点须幂等）。硬中断是 B18-01，未实现。
    """
    with self._active_count_lock:
        if self._drain_event.is_set():
            return
        self._drain_event.set()
        self._drain_started_at = time.time()
    if self._enable_registry and not self._registry_is_closed():
        # 单一状态源：register() 之前（starting）也要让 drain 生效，
        # 否则 update_service_info 在 _registered_service is None 时直接
        # return，内存状态都没置上（S9 前置坑）。
        info = getattr(self, "_service_info", None)
        if info is not None:
            info.status = "draining"
        # 注册表可见：编排/K8s 就绪探针据此把这台从流量里摘掉
        self.update_registry_info(status="draining")
        # 立即落盘（S18-02A / A45-03）：不再等 10s 心跳，否则派发侧在
        # 这个窗口里仍按 running 选中本机（探针 S3）。
        self._flush_registry_status("draining")
    timeout = _drain_timeout_from_env()
    logger.info(
        "worker 进入 draining（%s）：不再领新任务，等待 %d 个在途任务收尾（上限 %.0fs）",
        reason,
        self._active_task_count,
        timeout,
    )
    if timeout <= 0:
        self._force_stop_after_drain(timeout)
        return
    timer = threading.Timer(timeout, self._force_stop_after_drain, args=(timeout,))
    timer.daemon = True
    self._drain_timer = timer
    timer.start()

wait_for_idle

wait_for_idle(timeout: Optional[float] = None) -> bool

等所有在途任务结束；超时返回 False（调用方决定是否强退）。

源代码位于： plaita/server/flow_worker.py

def wait_for_idle(self, timeout: Optional[float] = None) -> bool:
    """等所有在途任务结束；超时返回 False（调用方决定是否强退）。"""
    deadline = None if timeout is None else time.time() + max(0.0, timeout)
    while self._active_task_count > 0:
        if deadline is not None and time.time() >= deadline:
            return False
        time.sleep(0.05)
    return True

stop

stop(wait: bool = False)

停止流程工作器。

顺序对无损升级很重要：先 draining（摘流量：注册表标 draining + 停止 领新任务）→ 等 in-flight 收尾（有界）→ 再注销服务。反过来的话，注销 之后仍可能领到任务，编排会看到「已下线但仍消费」的矛盾状态。

wait=True（S18-01/B45-06）：在停看门狗/注销之前有界等待在途 任务收尾（预算沿用 PLAITA_WORKER_DRAIN_TIMEOUT 的剩余量），库模式 调用方应传 True；CLI 默认 False——进程退出即硬中止，daemon 消费线程 可能仍在跑，未完成步的消息留 pending 由 XCLAIM 重放。无论哪种模式， 本方法都会置 D2 标记，可中断退避据此不再 enqueue/ack。

源代码位于： plaita/server/flow_worker.py

def stop(self, wait: bool = False):
    """停止流程工作器。

    顺序对无损升级很重要：先 draining（摘流量：注册表标 draining + 停止
    领新任务）→ 等 in-flight 收尾（有界）→ 再注销服务。反过来的话，注销
    之后仍可能领到任务，编排会看到「已下线但仍消费」的矛盾状态。

    ``wait=True``（S18-01/B45-06）：在停看门狗/注销之前**有界**等待在途
    任务收尾（预算沿用 ``PLAITA_WORKER_DRAIN_TIMEOUT`` 的剩余量），库模式
    调用方应传 True；CLI 默认 False——进程退出即硬中止，daemon 消费线程
    可能仍在跑，未完成步的消息留 pending 由 XCLAIM 重放。无论哪种模式，
    本方法都会置 D2 标记，可中断退避据此不再 enqueue/ack。
    """
    if self._drain_timer is not None:
        self._drain_timer.cancel()
        self._drain_timer = None
    if self._running and not self._drain_event.is_set():
        # 直接 stop（非 drain 路径）也保证「先摘流量再退出」
        with self._active_count_lock:
            self._drain_event.set()
            self._drain_started_at = self._drain_started_at or time.time()
        if self._enable_registry and not self._registry_is_closed():
            info = getattr(self, "_service_info", None)
            if info is not None:
                info.status = "draining"
            self.update_registry_info(status="draining")
            self._flush_registry_status("draining")
    logger.info("正在停止流程工作器...")
    self._running = False
    # D2/D3：退避重投不再 enqueue/ack；未完成消息留 pending。
    self._force_stop_event.set()
    if wait:
        budget = _drain_timeout_from_env()
        started = getattr(self, "_drain_started_at", None)
        if started is not None:
            budget = max(0.0, budget - (time.time() - started))
        if not self.wait_for_idle(budget):
            logger.warning(
                "stop(wait=True) 在 %.0fs 内未等齐在途任务（active=%d）："
                "继续收尾退出；未完成步的消息留 pending，由 XCLAIM 重放",
                budget, self._active_task_count,
            )

    # 停止注册重试（A2R19-03）：先于 unregister，避免退出后又被重试线程写回
    self._stop_registry_retry()

    # 停止租约看门狗（波次②）
    self._stop_lease_watchdog()

    # 停止取消监听（波次③）
    self._stop_cancel_watcher()

    # 停止 #26 观测导出（抓取端线程 + 告警 webhook 后台线程）
    self._stop_metrics_server()
    self._stop_alerting()

    # 停止控制监听
    if self._enable_registry:
        self.stop_control_listener()

    # 注销服务（先标 stopping：编排侧能看到「正在下线」而非突然消失）
    if self._enable_registry:
        info = getattr(self, "_service_info", None)
        if info is not None:
            info.status = "stopping"
        self.update_registry_info(status="stopping")
        # S18-02A：stopping 也要**立即**落盘（不再等心跳）。
        self._flush_registry_status("stopping")
        # 写完 stopping 之后、注销之前置 closed：这一次 flush 是最后一次
        # 注册表写入；之后心跳/容量/迟到 bump 全部 no-op（A46-03）。
        self._mark_registry_closed()
        if not self.unregister_service():
            logger.error(
                "registry_unregister_failed: 服务注销未完全成功"
                "（instance=%s；stale 键由 stopped 墓碑/TTL 兜底）",
                getattr(self, "instance_id", None),
            )

    # 关闭日志处理器
    if self._log_handler:
        self._log_handler.close()

    logger.info("流程工作器已停止")

metrics_text

metrics_text() -> str

Prometheus 文本快照（/metrics 每次抓取时调用）。

队列 stats 读失败（Redis 瞬断）不整单 500：退化为只报 worker 自身 指标——抓不到队列读数本身也是信号，不该让整张看板空白。

源代码位于： plaita/server/flow_worker.py

def metrics_text(self) -> str:
    """Prometheus 文本快照（``/metrics`` 每次抓取时调用）。

    队列 stats 读失败（Redis 瞬断）不整单 500：退化为只报 worker 自身
    指标——抓不到队列读数本身也是信号，不该让整张看板空白。
    """
    try:
        stats = self._get_task_queue().stats()
    except Exception as exc:
        logger.warning("读取队列 stats 失败（/metrics 仅返回 worker 指标）: %s", _scrub_url_text(exc))
        stats = {"stream_key": self.queue_name}
    metrics = collect_queue_metrics(stats)
    service_info = getattr(self, "_service_info", None)
    heartbeat = None
    if service_info is not None and service_info.last_heartbeat:
        try:
            heartbeat = datetime.fromisoformat(service_info.last_heartbeat).timestamp()
        except ValueError:
            heartbeat = None
    metrics.extend(collect_worker_metrics(
        active_tasks=self._active_task_count,
        draining=self._drain_event.is_set(),
        running=self._running,
        heartbeat_timestamp=heartbeat,
        instance=self._resolve_consumer_name(),
        registered=service_info is not None,
    ))
    # S18-02A：draining/stopping 立即落盘失败必须可观测（不得静默）。
    metrics.append(Metric(
        "registry_status_flush_failed_total",
        getattr(self, "_registry_status_flush_failures", 0),
        {},
        "counter",
        "生命周期状态立即落盘失败次数（S18-02A）",
    ))
    # A69-04：载体年龄两源都不可判 → fail-closed 丢弃次数（可能意味着
    # 某个生产者既没写 timestamp、消息也不来自 Redis Stream ID）。
    metrics.append(Metric(
        "carrier_age_unknown_total",
        getattr(self, "_carrier_age_unknown_total", 0),
        {},
        "counter",
        "start/resume 载体年龄无法判定而 fail-closed 丢弃的次数（A69-04）",
    ))
    # A2R31-02：租户状态纪元拒绝次数（进程内；Redis-only 恢复后非 default
    # 租户被拒的直接信号）。
    from plaita.server.tenant_status import tenant_stale_epoch_total

    metrics.append(Metric(
        "tenant_status_stale_epoch_total",
        tenant_stale_epoch_total(),
        {},
        "counter",
        "租户状态不属于当前部署纪元而被拒绝的次数（A2R31-02）",
    ))
    # #42（W10-a）：容量指标与「派发侧看到的容量」同源（同一快照函数），
    # 否则看板与派发门会各报一套（口径分叉是最难查的一类事故）。
    try:
        capacity_block = self.capacity_snapshot()
    except Exception as exc:  # noqa: BLE001 — 容量探针故障不该打空整张看板
        logger.debug("容量快照失败（本次 /metrics 不含容量指标）: %s", _scrub_exception_text(exc))
        capacity_block = None
    if capacity_block is not None:
        metrics.extend(collect_capacity_metrics(
            capacity_block, instance=self._resolve_consumer_name()
        ))
    if self._alerter is not None:
        alert_stats = self._alerter.stats()
        for key, help_text in (
            ("sent", "已成功投递的告警数"),
            ("failed", "投递失败的告警数"),
            ("dropped", "告警队列满而丢弃的事件数"),
            ("pending", "告警队列中待投递的事件数"),
        ):
            metrics.append(Metric(
                f"alert_webhook_{key}_total" if key != "pending" else "alert_webhook_pending",
                alert_stats[key],
                {},
                "counter" if key != "pending" else "gauge",
                help_text,
            ))
    return render_prometheus(metrics)

get_flow_definition

get_flow_definition(flow_id: str, version: Optional[str] = None) -> Flow

获取流程定义，支持按ID和版本获取

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
version	Optional[str]	流程版本，如果不指定则获取最新版本	None

返回：

类型	描述
Flow	Dict[str, Any]: 流程定义

引发：

类型	描述
ValueError	如果找不到流程定义或版本不匹配

源代码位于： plaita/server/flow_worker.py

def get_flow_definition(self, flow_id: str, version: Optional[str] = None) -> Flow:
    """
    获取流程定义，支持按ID和版本获取

    Args:
        flow_id: 流程ID
        version: 流程版本，如果不指定则获取最新版本

    Returns:
        Dict[str, Any]: 流程定义

    Raises:
        ValueError: 如果找不到流程定义或版本不匹配
    """
    # 生成缓存键（含租户段：同名流程可共存于不同租户 namespace）
    cache_key = f"{current_tenant()}:{flow_id}:{version or 'latest'}"

    # 尝试从缓存获取
    if cache_key in self.flow_definition_cache:
        logger.info("从缓存获取流程定义: %s", cache_key)
        return self.flow_definition_cache[cache_key]

    # 缓存未命中，从存储获取
    logger.info("从存储获取流程定义: %s, 版本: %s", flow_id, version or 'latest')
    flow_definition = self.flow_storage.get_flow(flow_id, version)

    if not flow_definition:
        # Delegate diagnostic details to the storage layer via an optional
        # diagnose() method — FlowWorker must not peek at storage internals
        # (e.g. Redis keys / namespaces) directly.
        if hasattr(self.flow_storage, "diagnose_missing_flow"):
            try:
                self.flow_storage.diagnose_missing_flow(flow_id)
            except Exception as exc:
                logger.warning("flow_storage.diagnose_missing_flow(%s) failed: %s", flow_id, _scrub_exception_text(exc))

        error_msg = f"找不到流程定义: {flow_id}"
        logger.error(error_msg)
        raise ValueError(error_msg)

    # 解析流程定义
    try:
        logger.info("成功获取流程定义: %s, 数据: %s", flow_id, flow_definition.get('name', 'unknown'))
        flow = Flow.model_validate(flow_definition)
    except Exception as e:
        error_msg = f"解析流程定义失败: {e}"
        logger.error(error_msg)
        raise ValueError(error_msg)

    self.flow_definition_cache[cache_key] = flow
    # 指纹随定义一起缓存（plaita#36）：指纹只取 model_validate **之前**的
    # 原始定义，与缓存里的 Flow 同窗口，避免 start/resume 各自重算走叉。
    self._flow_definition_hash_cache()[cache_key] = compute_flow_definition_hash(
        flow_definition
    )

    return flow

get_flow_definition_hash

get_flow_definition_hash(flow_id: str, version: Optional[str] = None) -> str

当前存储定义的指纹（plaita#36：原始定义 JSON 的 sha256）。

与 get_flow_definition 共用缓存窗口：定义加载时就算好（同一次 flow_storage.get_flow），调用方（start/resume）不自己从 Flow 对象反推——那正是版本漂移的入口。缓存未命中（定义缓存被清 / 加载器被 测试替身替换）时回源到同一个 FlowStorage 直读原始定义，绝不回退 到「解析后 dump」口径。

源代码位于： plaita/server/flow_worker.py

def get_flow_definition_hash(self, flow_id: str, version: Optional[str] = None) -> str:
    """当前存储定义的指纹（plaita#36：原始定义 JSON 的 sha256）。

    与 ``get_flow_definition`` 共用缓存窗口：定义加载时就算好（同一次
    ``flow_storage.get_flow``），调用方（start/resume）不自己从 ``Flow``
    对象反推——那正是版本漂移的入口。缓存未命中（定义缓存被清 / 加载器被
    测试替身替换）时回源到**同一个 FlowStorage** 直读原始定义，绝不回退
    到「解析后 dump」口径。
    """
    cache_key = f"{current_tenant()}:{flow_id}:{version or 'latest'}"
    cache = self._flow_definition_hash_cache()
    cached = cache.get(cache_key)
    if cached is not None:
        return cached
    flow_definition = self.flow_storage.get_flow(flow_id, version)
    if not flow_definition:
        raise ValueError(f"找不到流程定义: {flow_id}")
    digest = compute_flow_definition_hash(flow_definition)
    cache[cache_key] = digest
    return digest

start_flow

start_flow(flow_id: str, params: Dict[str, Any], version: Optional[str] = None, execution_id: Optional[str] = None, dedup_key: Optional[str] = None, delivery_count: Optional[int] = None, enqueued_at: Optional[str] = None, schema_version: Optional[int] = None) -> Dict[str, Any]

启动流程执行

全程持有 execution_id 执行租约（先行落 running 行之前取得，处理 结束 finally 释放）：队列按 claim_min_idle_ms（默认 60s）把长任务 的 pending 消息 XCLAIM 重派给空闲 worker——无租约的 start 在多 worker 池下跑超过 60s 必被重派双跑（#23）。另一 worker 已持租约（消息重派/ 并发投递）时抛 ExecutionLeaseError，本消息不 ack 留待处理。

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
params	Dict[str, Any]	流程输入参数	必需
version	Optional[str]	流程版本，如果不指定则使用最新版本	None
execution_id	Optional[str]	预铸 id（BFF start 时铸造、随消息透传）——提交方 即刻拿 id 返给调用方，无需等 worker 消费；缺省就地铸造	None
dedup_key	Optional[str]	可选 start 幂等键（消息重投/双开收敛；未提供则行为 与存量完全一致，见 _claim_start_dedup）	None
delivery_count	Optional[int]	消息投递次数（可观测；见 _process_execution_result）	None
enqueued_at	Optional[str]	console 入队时间（消息体 timestamp，ISO 串）。用于 时间线回答「排队多久」（plaita#25）：缺失/不可解析 ⇒ queue_wait_ms=None（未采集），不折算成 0。	None
schema_version	Optional[int]	触发本次 start 的消息信封版本（A44-02）。仅用于 start 幂等命中后重入队的 resume 副本继承源版本；直接调用方 缺省 None = 按当前 build 写版本。	None

返回：

类型	描述
Dict[str, Any]	Dict[str, Any]: 流程执行结果

源代码位于： plaita/server/flow_worker.py

def start_flow(self, flow_id: str, params: Dict[str, Any], version: Optional[str] = None,
               execution_id: Optional[str] = None,
               dedup_key: Optional[str] = None,
               delivery_count: Optional[int] = None,
               enqueued_at: Optional[str] = None,
               schema_version: Optional[int] = None) -> Dict[str, Any]:
    """
    启动流程执行

    全程持有 ``execution_id`` 执行租约（先行落 running 行之前取得，处理
    结束 finally 释放）：队列按 ``claim_min_idle_ms``（默认 60s）把长任务
    的 pending 消息 XCLAIM 重派给空闲 worker——无租约的 start 在多 worker
    池下跑超过 60s 必被重派双跑（#23）。另一 worker 已持租约（消息重派/
    并发投递）时抛 ``ExecutionLeaseError``，本消息不 ack 留待处理。

    Args:
        flow_id: 流程ID
        params: 流程输入参数
        version: 流程版本，如果不指定则使用最新版本
        execution_id: 预铸 id（BFF start 时铸造、随消息透传）——提交方
            即刻拿 id 返给调用方，无需等 worker 消费；缺省就地铸造
        dedup_key: 可选 start 幂等键（消息重投/双开收敛；未提供则行为
            与存量完全一致，见 ``_claim_start_dedup``）
        delivery_count: 消息投递次数（可观测；见 _process_execution_result）
        enqueued_at: console 入队时间（消息体 ``timestamp``，ISO 串）。用于
            时间线回答「排队多久」（plaita#25）：缺失/不可解析 ⇒
            ``queue_wait_ms=None``（未采集），不折算成 0。
        schema_version: 触发本次 start 的消息信封版本（A44-02）。仅用于
            start 幂等命中后重入队的 resume 副本继承源版本；直接调用方
            缺省 ``None`` = 按当前 build 写版本。

    Returns:
        Dict[str, Any]: 流程执行结果
    """
    # 获取流程定义
    flow = self.get_flow_definition(flow_id, version)



    logger.info("开始执行流程: %s, 版本: %s", flow_id, version or '最新版本')

    try:
        # P0 可见性修复（keeper 迁移设计稿 §5.5 清单①，43828aa）：
        # execution_id 优先吃 BFF 预铸（随消息透传，提交方即刻可轮询），
        # 否则就地铸造；行 id 与 result.execution_id 天然一致。异常遗留
        # 的 running 行是 zombie，交 reaper/心跳年龄判定处置。
        # 位置前移到执行器构造之前：节点耗时采集器按执行登记，需要 id 已定。
        # preminted（plaita#234）：只有自带 id 的载体才可能命中盘上已有
        # 断点；就地铸造的 id 天然是首次执行，不必多读一次盘。
        preminted_execution_id = bool(
            isinstance(execution_id, str) and execution_id.strip()
        )
        execution_id = execution_id or uuid.uuid4().hex

        # A51-01：删除墓碑复核（五种纵深之一）。分派闸之外，直达
        # ``start_flow`` 的调用同样不得把已删执行重建；新铸造的 id 天然
        # 不可能有墓碑，只有预铸 id 才需要多读一次。命中即 ack 语义
        # （返回值供调用方确认），并释放同 id 的 dedup 映射。
        if preminted_execution_id:
            tombstone = self._load_execution_tombstone(
                execution_id, current_tenant()
            )
            if tombstone is not None:
                if dedup_key:
                    self._release_start_dedup(dedup_key, execution_id)
                logger.warning(
                    "执行 %s 已删除（墓碑 tenant=%s reason=%s deleted_at=%s），"
                    "start_flow 拒绝重建（flow=%s）",
                    execution_id,
                    tombstone.get("tenant"),
                    tombstone.get("reason"),
                    tombstone.get("deleted_at"),
                    flow_id,
                )
                return {
                    "execution_id": execution_id,
                    "status": "deleted",
                    "discarded": True,
                }

        # 创建流程执行器并执行流程
        # 复用同一个 FlowExecution 贯穿所有分布式步骤, 保留用户回调
        handlers = self._handlers_for(execution_id, enqueued_at=enqueued_at)
        # 时间线开篇（plaita#25）：记 run_start + 由 enqueued_at 算出排队时长。
        # 位置在**落 running 行之前**——排队时长的终点是「worker 开始处理」，
        # 不是「首节点开始跑」，否则排队时长会被流程定义加载/租约获取计入。
        _timeline = self._timelines.get(execution_id)
        if _timeline is not None:
            _timeline.start_run()
        execution = FlowExecution(
            event_bus=self.event_bus,
            callback_handlers=handlers,
        )
        execution.mode = ExecutionMode.DISTRIBUTED
        self._bind_observers(execution)

        # start 幂等键（波次二任务③，与 G1 预铸 id 组合）：在先行落行
        # 之前认领，映射值=预铸或就地铸造的 execution_id——重投/双开
        # 在**任何新行落盘之前**即被拦截，首节点副作用由此受重投保护。
        dedup_action, mapped_execution_id = self._claim_start_dedup(
            dedup_key, execution_id
        )
        if dedup_action == "hit":
            hit_result = self._handle_start_dedup_hit(
                flow_id, mapped_execution_id, params=params,
                schema_version=schema_version,
                carrier_execution_id=execution_id,
            )
            if hit_result is not None:
                # 本次没有真正启动：回收按本 id 装配的时间线采集器
                # （否则节流命中会在长跑 worker 里留一条无人认领的采集器）。
                self._timelines.pop(execution_id, None)
                if hit_result.get("release_dedup") and dedup_key:
                    self._release_start_dedup(
                        dedup_key, hit_result.get("execution_id") or mapped_execution_id
                    )
                return hit_result
            # 孤儿映射（首次执行从未落盘）：释放后重新认领，让本次启动
            # 继续受保护；并发后来者抢先认领则放弃设防（存量语义）
            if dedup_key:
                self._release_start_dedup(dedup_key, mapped_execution_id)
                dedup_action, mapped_execution_id = self._claim_start_dedup(
                    dedup_key, execution_id
                )
                if dedup_action == "hit":
                    logger.warning(
                        "start 幂等键孤儿释放后被并发认领，本次启动不设防: %s",
                        dedup_key,
                    )

        # 载体转换（plaita#234）：start 消息**不是只能从首节点起跑**。
        # 盘上已有该 execution_id 的断点时（worker 崩溃/滚动重启、
        # StatePersistError、ServiceDispatchError 后消息不 ack 被 XCLAIM
        # 原样回收——这条回收路径**不经过** #91 的
        # ``_requeue_body_for_retry``），历史实现会新建 `context={}` 的
        # 行无条件覆写盘上 checkpoint，再以**不带 saved_context** 的
        # ``run_distributed`` 从首节点整条重放：已完成节点的外部副作用
        # 最多重复 max_deliveries 次，且 $LAST_NODE/$NODE/$INPUT 被永久
        # 抹成 {}，此后连正确的 event/continue resume 也失去续跑现场。
        #
        # 修法：把 start 载体**转成 resume 语义**（判定口径与 #91 一致，
        # 但入口收在 start_flow，覆盖「重投不经 _requeue_body_for_retry」
        # 的那条路），复用 resume_flow 已有的全部保护：终态/挂起幂等短路
        # （不重跑）、flow_hash 校验、取消检查、租约，以及 #233 的挂起
        # 补派。此时尚未取得 start 租约（两者共用同一 execution_id 租约
        # 键），转调不会自撞锁。
        if preminted_execution_id:
            existing_state = self._existing_start_checkpoint(flow_id, execution_id)
            if existing_state is not None:
                logger.warning(
                    "start 载体命中盘上已有断点（execution_id=%s, status=%s, "
                    "checkpoint=%s）：转 resume 语义，绝不整条重放/覆写 checkpoint",
                    execution_id,
                    existing_state.status,
                    "有" if existing_state.context else "空",
                )
                # 时间线采集器**保留**：resume 路径的 _handlers_for 会复用
                # 同一采集器，从而保住本消息的 enqueued_at/run_start（#25）。
                return self.resume_flow(
                    flow_id,
                    execution_id,
                    "continue",
                    delivery_count=delivery_count,
                    params=params,
                )

        # 先落 running 行再执行（P0 可见性）：首节点期间 /api/executions
        # 查得到此行、cancel 有锚。flow_hash：对**存储中的原始定义**计算
        # 指纹（plaita#36 口径 v2）随行落盘——resume 时与当前定义比对，防
        # 运行中改定义后续跑走错分支；原始定义口径保证同一份存储定义在
        # 不同引擎版本下算同一个指纹。
        state = ExecutionState(
            execution_id=execution_id,
            flow_id=flow_id,
            flow_version=version,
            flow_hash=self.get_flow_definition_hash(flow_id, version),
            flow_hash_algo=FLOW_HASH_ALGO,
            engine_version=_engine_version(),
            tenant_id=current_tenant(),
            context={},
            status="running",
            start_time=datetime.now().isoformat(),
            invoker="worker"
        )

        # start 入口取消检查点（与 resume 入口对称）：BFF cancel 可发生在
        # 「已入队、未消费」窗口，更关键的是 start 消息 at-least-once 重投
        # 时（在途失败→未 ack→重投）同 id 重跑整条 flow——不查标志则
        # 已取消意图永远等不到步界（E2E drill 3 实证：超时 flow 空转
        # N×900s）。命中即终态化，不再执行。
        if self._cancel_requested(execution_id):
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_at_start_entry")
            self._finalize_observers()
            logger.info("执行 %s 在 start 入口命中取消标志，终态化", execution_id)
            return {
                "execution_id": execution_id,
                "status": "cancelled",
                "cancelled_at_start": True,
            }

        # 租约（#23 扩展 A′）：**在落 running 行之前 acquire**，持有整个
        # start 处理窗口。A′（2026-10-06）只把租约对齐到 run_distributed
        # 之前，仍在先行落行**之后**——「落行 → acquire」间可被别的 worker
        # 抢先 XCLAIM 同一 start 消息（>claim_min_idle_ms 即可），抢占者
        # 先 acquire 成租约，原持有者 acquire 失败礼貌退出，但其 running
        # 行已被抢占者覆写。移动到落行前同时消除 A′ 的三个后果：
        # ①多 worker 抢到同一 start 消息时无闸可拦，双方同时跑同 id（双跑）；
        # ②死信守卫按「租约是否被持有」判活，start 执行恒"租约空"→ 长步
        #   误判持有者已死 → 误死信 + 重入队，长任务被反复打断；
        # ③抢占者白烧 5 次 delivery 配额。
        # 对齐 resume：acquire → 看门狗续租 → finally 释放。acquire 失败
        # 即抛 ExecutionLeaseError（run() 走不 ack/重投语义），抢占者礼貌退出。
        # holder 嵌入注册表 instance id（#50）：其他 worker 在租约冲突时
        # 可反解持有者并核其在册与否，区分「活持有者」与「死后 TTL 尾巴」。
        holder = new_holder_token(
            prefix="start", instance_id=getattr(self, "instance_id", None)
        )
        lease_value, fence_token = self._acquire_lease(execution_id, holder)
        if lease_value is None:
            raise ExecutionLeaseError(
                f"execution {execution_id} is leased by another worker; "
                "refuse concurrent start"
            )
        fence_token_reset = None
        if fence_token is not None:
            fence_token_reset = set_current_fence_token(fence_token)

        # 落 running 行（持租约窗口内）：与本执行此后所有写同 fence 世代，
        # 不会被后续接管者的 fenced CAS 拒绝序列排斥。
        self._persist_state_or_raise(execution_id, state, "start")

        # 取消监听（波次③）：登记整个推进窗口——取消标志键命中即在途节点被
        # execution.cancel() 中止（code 沙箱当场 killpg；协作节点如 agentrun
        # 消费 cancel_event 收手）。租约看门狗同窗口。
        self._register_cancel_watch(execution_id, execution)
        self._register_lease_watch(
            execution_id, lease_value, execution, fence_token=fence_token
        )
        # 标记「本次处理是否正常跑完」：只有正常跑完才说明该 execution
        # 这一轮真的结束了；从 run_distributed 里抛异常（尤其
        # ExecutionLeaseError 的 ack 释放路径）时执行对象可能仍在推进，
        # 见 finally 里的条件释放。
        completed_normally = False
        try:
            result = execution.run_distributed(flow, params=params, execution_id=execution_id)

            # 处理执行结果
            final_result = self._process_execution_result(
                flow, result, state, execution, delivery_count=delivery_count,
                lease_execution_id=execution_id,
            )

            completed_normally = True
            return final_result
        finally:
            self._unregister_cancel_watch(execution_id)
            self._unregister_lease_watch(execution_id, lease_value)
            if fence_token_reset is not None:
                reset_current_fence_token(fence_token_reset)
            # release 对整个租约 value 串 compare（fencing 档 = {holder}:{gen}）
            #
            # ⚠️ **释放租约 ≠ 执行已结束**（2026-10-10 二波，生产实证）：
            # 「执行被他人持租约 → ack 释放」这条路径让本消息处理走到
            # finally，但**该 execution 对象仍在沙箱里推进长节点**
            # （实测 exec 339c3262：19:20:31 复用沙箱跑 impl，19:21:25 因
            # 「他人持租约」被 ack 释放 ⇒ 此后心跳停 80 分钟 ⇒ keeper 判
            # zombie 收尸）。一旦 release，「无租约」成事实：看门狗登记已被
            # 上面 unregister 摘除，也从未走过 renew 失败分支 ⇒ **没有任何
            # 组件再发布它的进度** ⇒ 健康 run 必然被判死。
            #
            # 因此：**仅当执行确已停止推进时才真正 release**；否则交给
            # `_lease_orphan_watch` 继续观测（只发心跳，不续租、不重复
            # cancel），由它在该执行结束或重新登记时摘除。
            if not completed_normally and self._execution_still_advancing(execution):
                # tenant 由上下文取（该作用域无 tenant_id 形参，见上方
                # `tenant_id=current_tenant()` 的同款用法）。
                self._register_lease_orphan(
                    execution_id, execution, current_tenant(), fence_token
                )
            else:
                self.execution_lease.release(execution_id, lease_value)

    except ExecutionIdConflictError:
        # 冲突的 start 未落任何行（查盘在落 running 行之前）。若幂等映射
        # 是**本次**认领的，释放它——否则 7 天内同 dedup_key 会被指向
        # 另一条 flow 的 execution_id，后续修正后的 start 也走不到这里。
        # 原样上抛：run() 按 ValueError poison ack（换 flow 重投不会自愈）。
        if dedup_key and locals().get("dedup_action") == "claimed":
            self._release_start_dedup(
                dedup_key, mapped_execution_id or execution_id
            )
        raise

    except (NodeExecutionRetryableError, ServiceDispatchError, ExecutionLeaseError):
        # 节点级可重试失败（波次二任务①）/ 幂等命中后重入队 resume 失败
        # （波次二任务③）/ **租约被他人持有**（A′ + #23，租约覆盖整个
        # start 窗口含先行落行）：原样上抛给 run() 按不 ack 语义处理——
        # ExecutionLeaseError 尤其关键：run() 的
        # `except ExecutionLeaseError` 分支据此 note_lease_conflict 并把消息
        # 留在 pending 等租约过期后 reclaim；若被下方通用 except 包成
        # RuntimeError，run() 会误 ack 消息 → 抢占者把别人的活执行 ack 掉，
        # 且真实持有者崩溃后无人 reclaim（执行永久失联）。绝不能包。
        raise

    except Exception as e:
        logger.error("执行流程出错: %s", _scrub_exception_text(e))
        # 波次③步内取消：start 路径首个节点在途命中取消监听 →
        # 引擎自 run_distributed 归一化抛出（挂 __cause__）或协作节点自杀
        # 抛普通异常。两条判据任一命中即终态化 cancelled，绝不包成
        # RuntimeError 让执行停在 running（取消是控制面意图）。
        if (
            _chain_has_cancellation(e) or self._cancel_requested(execution_id)
        ) and "state" in locals() and state is not None:
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_in_start_node")
            self._finalize_observers()
            logger.info(
                "执行 %s 在 start 首节点执行期响应取消，终态化 cancelled",
                execution_id,
            )
            return {"execution_id": execution_id, "status": "cancelled"}
        # plaita#25 P2：错误路径落盘。确定性失败此前只打日志 + 实时发布事件
        # ⇒ 状态行停在 start 时落的 running，**落盘时间线只有 run_start**：
        # 失败节点（runner 已补发 on_node_end(error)）与收口事件都只存在于
        # 内存/SSE 里，数小时后复盘时没有第二来源。这里走**唯一落盘收口**
        # 把这次尝试的事件补进 state.timeline，并在流里补一条
        # run_end(final=False, status=error)。
        #
        # 状态行**刻意不动**：错误路径的 running 行是既有契约（可重投、
        # 可 cancel、zombie 可见——见
        # tests/unit/test_g1_resume_retry.py::test_running_row_survives_start_failure），
        # 也保证 ack/重投与死信守卫的判据逐字不变。
        # best-effort：落盘失败只告警，下面的 RuntimeError 照旧上抛 ⇒
        # 消息仍不 ack 走 at-least-once 重投。
        self._record_failed_start_attempt(execution_id, locals().get("state"))
        self._finalize_observers()
        raise RuntimeError(f"执行流程出错: {e}")

resume_flow

resume_flow(flow_id: str, execution_id: str, resume_type: str, data: Optional[Dict[str, Any]] = None, delivery_count: Optional[int] = None, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

恢复流程执行。

取得 execution_id 租约后才推进；另一 worker 已持有租约时抛 ExecutionLeaseError（RedisFlowWorker 对此不 XACK，待租约过期后回收）。

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
execution_id	str	执行ID	必需
resume_type	str	恢复类型	必需
data	Optional[Dict[str, Any]]	恢复数据	None
delivery_count	Optional[int]	消息投递次数（可观测；见 _process_execution_result）	None
params	Optional[Dict[str, Any]]	可选入参（#122）。只在盘上 checkpoint 为空时被引擎用来 重新 setup_flow（start→resume 载体转换会撞上这种窗口）； checkpoint 非空（正常 resume）时不参与——续跑一律以盘上 context 为准，续跑语义零变化。	None

源代码位于： plaita/server/flow_worker.py

def resume_flow(self, flow_id: str, execution_id: str, resume_type: str,
                data: Optional[Dict[str, Any]] = None,
                delivery_count: Optional[int] = None,
                params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    恢复流程执行。

    取得 ``execution_id`` 租约后才推进；另一 worker 已持有租约时抛
    ``ExecutionLeaseError``（RedisFlowWorker 对此**不** XACK，待租约过期后回收）。

    Args:
        flow_id: 流程ID
        execution_id: 执行ID
        resume_type: 恢复类型
        data: 恢复数据
        delivery_count: 消息投递次数（可观测；见 _process_execution_result）
        params: 可选入参（#122）。只在盘上 checkpoint 为空时被引擎用来
            重新 ``setup_flow``（start→resume 载体转换会撞上这种窗口）；
            checkpoint 非空（正常 resume）时**不参与**——续跑一律以盘上
            context 为准，续跑语义零变化。
    """
    # 前置判定（租约之前）：终态/挂起幂等短路在此完成，保持 #33 的
    # 「suspended + continue 不取租约」契约。这里拿到的 state 只是
    # **预读快照**——取到租约后必须重读并以盘上状态为准（plaita#241）。
    state, early_result = self._load_and_guard_resume_state(
        flow_id, execution_id, resume_type
    )
    if early_result is not None:
        return early_result

    # 获取流程版本
    version = state.flow_version

    # 获取流程定义
    flow = self.get_flow_definition(flow_id, version)

    # flow 定义指纹（波次二任务②）：worker 的定义 TTLCache 有 300s 窗口、
    # console engine_sync 可直接覆盖 Redis 定义——挂起执行 resume 用
    # latest 版本定义时，若定义自启动后被改过，``_get_next_from_last``
    # 找不到节点/走错分支且无任何告警。指纹在 lease 内判定（与活
    # worker 的推进写串行化，防「他人持租约推进中却被误终态化」），
    # 不一致 → 终态化 error + raise ValueError（poison ack）——意图是
    # 可观测的终态而不是重投风暴。老状态 flow_hash 为 None → 跳过校验
    # （零回归）。指纹取**存储中的原始定义**（plaita#36 口径 v2）：同一份
    # 存储定义不受引擎版本/模型字段演进影响。
    current_flow_hash = self.get_flow_definition_hash(flow_id, version)

    # 解析流程定义
    logger.info("恢复流程执行: %s, 执行ID: %s, 恢复类型: %s", flow_id, execution_id, resume_type)

    # holder 嵌入注册表 instance id（#50），语义同 start 路径。
    holder = new_holder_token(
        prefix="resume", instance_id=getattr(self, "instance_id", None)
    )
    lease_value, fence_token = self._acquire_lease(execution_id, holder)
    if lease_value is None:
        raise ExecutionLeaseError(
            f"execution {execution_id} is leased by another worker; refuse concurrent resume"
        )

    fence_token_reset = None
    if fence_token is not None:
        # fencing 写侧守门（设计稿 §4.2）：本 resume 的所有状态落盘带
        # 世代号，FencedExecutionStorage 据此 CAS；finally 复位。
        fence_token_reset = set_current_fence_token(fence_token)

    try:
        # TOCTOU 复核（plaita#241）：上面的 load 发生在取租约**之前**。
        # 若窗口内另一 worker 已把执行推到终态并释放租约，本 worker 随后
        # 取得的是**新一代**租约——fencing 只拒绝旧世代写，不会拦它拿
        # 旧 state 继续 run_distributed（节点副作用重放、终态行被二次
        # 改写）。租约内重读并重跑前置判定；state/early_result 均以本次
        # 为准，不复用 acquire 前的快照。
        state, early_result = self._load_and_guard_resume_state(
            flow_id, execution_id, resume_type, lease_held=True
        )
        if early_result is not None:
            logger.info(
                "执行 %s 租约内复核发现状态已变更（status=%s）：拒绝本次 "
                "resume，不复用 acquire 前的旧 state（plaita#241）",
                execution_id, early_result.get("status"),
            )
            # 具名拒绝：区别于入口即终态的重复投递——本次是**赢了租约
            # 竞争、但盘上状态已由上一持有者推进**，测试/值守可据此
            # 识别 TOCTOU 窗口，而不是当成普通幂等短路。
            early_result["rejected_after_lease"] = True
            return early_result

        # plaita#247：#241 的复核只替换了 state，version/flow/current_flow_hash
        # 仍是 acquire 前快照算出的值。窗口内 flow_version 变更（同 flow_id 新
        # 版本 / 另一 start 载体回写）时，继续用旧 flow 执行并把结果写回新
        # version 的状态行 ⇒ 执行定义与状态身份不一致。复核后的 state 才是
        # 权威：版本变了就用新版本重算定义与指纹；hash 冲突仍走下方既有
        # mismatch 终态路径（绝不带着旧定义/旧指纹续跑）。
        if state.flow_version != version:
            logger.warning(
                "执行 %s 在 load→acquire 窗口内 flow_version 变更（%r → %r），"
                "按租约内复核的新版本重算 flow 定义与指纹（plaita#247）",
                execution_id, version, state.flow_version,
            )
            version = state.flow_version
            flow = self.get_flow_definition(flow_id, version)
            current_flow_hash = self.get_flow_definition_hash(flow_id, version)

        state_status = getattr(state, "status", "") or ""
        retry_wakeup = state_status == "error" and \
            ResumeType.coerce(resume_type) is ResumeType.RETRY
        if state_status == "running" and \
                ResumeType.coerce(resume_type) is ResumeType.RETRY:
            # running 执行没有失败节点可重跑（plaita#53 验收 2）：retry 在此的
            # 语义 = 从 checkpoint 派发后继节点（与 continue 同路），**必须真实
            # 重派**——禁止「受理后无动作」。显式留痕，值守从日志即可区分
            # 「真的推进了」与「白等」（租约被他人持有等不生效情形另见 run()
            # 的租约冲突日志）。
            logger.warning(
                "执行 %s 处于 running，resume_type=retry 按「从断点续跑」处理"
                "（running 执行无失败节点可重跑），本次将从 checkpoint 派发后继节点",
                execution_id,
            )

        # 陈旧 event/timeout 载体的归一化（plaita#258）：resume 在流内
        # 推进并落过 step_persist 后，若发生瞬态 StatePersistError / 进程
        # 崩溃，消息按 at-least-once 重投，但 resume_type 仍是 event/timeout
        # ———而挂起订阅已在第一次 resume 时注销、$LAST_NODE 已不是 pending
        # 挂起点。策略层会抛 ResumeGuardError（协议错误），旧 run() 分支据此
        # ack 唯一载体 ⇒ 执行永久停在 running。这里在**取得租约并重读状态
        # 之后**确认「执行仍活、但决议无处可落」，把本次意图改写为 continue
        # （剩余工作就是从 checkpoint 续跑，已完成节点不重放）。租约内
        # 复核保证：另一 worker 已推进到**新的挂起点**时 status=suspended，
        # 不会走本分支，陈旧 resume 仍由策略层的订阅绑定守卫拒绝（#221）。
        if resume_type in (ResumeType.EVENT, ResumeType.TIMEOUT, "event", "timeout"):
            stale_reason = self._resume_carrier_stale_reason(state, flow)
            if stale_reason:
                logger.warning(
                    "执行 %s 的 %s resume 载体已错过挂起点（%s）；订阅早已注销、"
                    "执行仍 running，本次归一化为 continue 从 checkpoint 续跑"
                    "（plaita#258）",
                    execution_id, resume_type, stale_reason,
                )
                resume_type = ResumeType.CONTINUE.value

        # XCLAIM 恢复路径取消检查点（设计稿 §3.1）：crash 后 cancel 消息
        # 或任一 resume 消息被重投，若取消标志在（控制面对运行中执行写的
        # 意图键），直接终态化而非继续推进。上一 checkpoint 已落盘，
        # cancelled 的 context 即取消点前一步的完整快照（§3.3）。
        if self._cancel_requested(execution_id):
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_at_resume_entry")
            logger.info("执行 %s 在 resume 入口命中取消标志，终态化", execution_id)
            return {
                "execution_id": execution_id,
                "status": "cancelled",
                "cancelled_at_resume": True,
            }

        # flow 定义指纹校验（波次二任务②）在 G1 retry 唤醒**之前**：
        # 定义被改时执行保持 error（hash 不匹配信息落盘），修复定义后
        # 仍可再 retry——G1 的「error 态可反复唤醒」语义不被破坏。
        # 取消是用户意图，已在上方优先放行。写盘走 _persist_state_or_raise
        # （带本 resume 的 fence 世代），随后 raise FlowHashMismatchError
        # → run() 按 ValueError poison ack（终态已可观测，重投无意义）。
        stored_flow_hash = getattr(state, "flow_hash", None)
        stored_algo = getattr(state, "flow_hash_algo", None)
        algo_changed = bool(stored_algo) and stored_algo != FLOW_HASH_ALGO
        allow_hash_change = bool(
            isinstance(data, dict) and data.get(ALLOW_FLOW_HASH_CHANGE_KEY)
        )
        hash_decision = classify_flow_hash_change(
            stored_hash=stored_flow_hash,
            current_hash=current_flow_hash,
            stored_algo=stored_algo,
            current_algo=FLOW_HASH_ALGO,
            allow_change=allow_hash_change,
        )
        if hash_decision in ("mismatch", "accepted"):
            if hash_decision == "accepted":
                # 升级后的人工裁决入口：显式承认「换算法/换定义」并续跑，
                # 必须留痕（WARNING + 状态里刷新指纹），不静默放行。
                logger.warning(
                    "执行 %s 的 flow 指纹变化被显式放行续跑（%s）：stored=%s(%s) current=%s(%s)",
                    execution_id,
                    ALLOW_FLOW_HASH_CHANGE_KEY,
                    stored_flow_hash[:12],
                    stored_algo or "unknown",
                    current_flow_hash[:12],
                    FLOW_HASH_ALGO,
                )
                state.flow_hash = current_flow_hash
                state.flow_hash_algo = FLOW_HASH_ALGO
                self._persist_state_or_raise(execution_id, state, "flow_hash_change_accepted")
            else:
                mismatch_msg = (
                    "flow 定义自启动后已变更，hash 不匹配，执行无法安全续跑 "
                    f"(execution_id={execution_id}, flow_id={flow_id}, "
                    f"stored_hash={stored_flow_hash[:12]}..., "
                    f"current_hash={current_flow_hash[:12]}..., "
                    f"stored_algo={stored_algo or 'unknown'}, current_algo={FLOW_HASH_ALGO})"
                )
                logger.error(mismatch_msg)
                state.status = "error"
                state.error = {
                    "message": mismatch_msg,
                    # plaita#93：本处终态化不由异常驱动（有专门的字段承载
                    # 细节），但落盘面同样要给下游一个机器可判别的身份：
                    # code 用引擎的通用 flow 级错误码（-500），error_type
                    # 是本处的 worker 级判据（不是节点业务码）。
                    "code": _FLOW_ERROR_CODE,
                    "error_type": "flow_hash_mismatch",
                    "stored_flow_hash": stored_flow_hash,
                    "current_flow_hash": current_flow_hash,
                    "stored_flow_hash_algo": stored_algo,
                    "current_flow_hash_algo": FLOW_HASH_ALGO,
                    # 升级疑似（算法标记变了且哈希也不同）：给运维明确下一步，
                    # 而不是只丢一句「hash 不匹配」。
                    "upgrade_suspected": algo_changed,
                    "hint": (
                        "引擎升级可能改变指纹算法；确认流程定义未变且可接受后，"
                        f"在 resume data 里带 {ALLOW_FLOW_HASH_CHANGE_KEY}: true 显式放行"
                    ) if algo_changed else None,
                }
                state.end_time = datetime.now().isoformat()
                self._persist_state_or_raise(execution_id, state, "flow_hash_mismatch")
                raise FlowHashMismatchError(mismatch_msg)
        elif hash_decision == "refresh":
            # 哈希相同 → 定义确实没变，只是算法标记变了（引擎升级）：续跑并刷新标记
            logger.info(
                "执行 %s 的 flow 指纹一致但算法标记升级（%s → %s），续跑并刷新标记",
                execution_id,
                stored_algo,
                FLOW_HASH_ALGO,
            )
            state.flow_hash_algo = FLOW_HASH_ALGO

        # 引擎版本跨 minor：只告警不拦截（硬门由 flow_hash 承担，这里给可观测性）
        stored_engine = getattr(state, "engine_version", None)
        current_engine = _engine_version()
        if stored_engine and _major_minor(stored_engine) != _major_minor(current_engine):
            logger.warning(
                "执行 %s 由引擎 %s 创建，当前 %s：跨 minor 续跑，"
                "若流程行为与预期不符请复核定义与节点实现",
                execution_id,
                stored_engine,
                current_engine,
            )

        # G1 retry 唤醒：error → running 翻转并先落盘（观察者/监控立即
        # 可见；若本 worker 接着硬死，行是 running 而非 error——再一轮
        # retry 仍可放行，zombie 判定也不误报「终态」）。checkpoint 原样，
        # saved_context 由下方 run_distributed 携带。
        # 与任务①组合语义：唤醒即清零节点重试计数键——预算耗尽终态化的
        # 执行经人工 retry 唤醒后拿全新预算，否则唤醒的执行第一次失败就
        # 立刻再耗尽，G1 形同虚设。翻转落盘成功后才清零（落盘失败抛
        # StatePersistError 走重投，唤醒未生效不清预算）。
        # 不传 node_id ⇒ 清**该执行下所有节点**的计数键（plaita#123 起键按
        # 节点隔离，唤醒是执行级意图，必须整体重来）。
        if retry_wakeup and not state.context:
            # plaita#240：retry 的能力前提是盘上有 checkpoint。此前先翻
            # running 再落盘、等 run_distributed 才由策略层抛
            # ResumeGuardError；协议错误分支虽「保持原状」，但原状已经被
            # 改成 running（error/end_time 也被清空）——非法 error→running
            # 迁移 + 失败证据丢失。必须在**任何状态变更之前**拒绝。
            # 保持 error 原状（不落盘）、不创建 FlowExecution、不触发节点。
            message = (
                "resume_type='retry' requires a saved checkpoint, but the "
                f"execution has none; execution {execution_id} remains error "
                "(status=error, error/end_time preserved)"
            )
            logger.warning(
                "执行 %s 收到 retry 但 checkpoint 为空，拒绝唤醒（保持 error 原状）",
                execution_id,
            )
            raise ResumeProtocolError(message, execution_id=execution_id)
        if retry_wakeup:
            state.status = "running"
            state.error = None
            state.end_time = None
            self._persist_state_or_raise(execution_id, state, "retry_wakeup")
            self._reset_node_retry_counter(execution_id)
            # 记一次唤醒（plaita#73 补丁）：唤醒会重置节点预算，必须计数
            # 才能让「唤醒→重置→耗尽→再唤醒」收敛。达上限的拒绝在上方
            # 入口处（`_g1_wakeup_budget_exhausted`）。
            self._record_g1_wakeup(execution_id)
            logger.info("执行 %s error 态经 retry 放行（重试计数已清零），从断点步进", execution_id)

        # 复用同一个 FlowExecution 贯穿恢复后的所有分布式步骤
        execution = FlowExecution(
            event_bus=self.event_bus,
            callback_handlers=self._handlers_for(execution_id),
        )
        execution.mode = ExecutionMode.DISTRIBUTED
        self._bind_observers(execution)
        # 登记看门狗（波次②）：持租约期间每 TTL/3 续租，防长步 > TTL
        # 被 XCLAIM 抢占双跑（fence 世代随登记带上，心跳写要过世代 CAS）
        self._register_lease_watch(
            execution_id, lease_value, execution, fence_token=fence_token
        )
        # 登记取消监听（波次③）：命中标志键即中止在途节点（与租约登记
        # 同窗口；finally 一并撤销）
        self._register_cancel_watch(execution_id, execution)

        # 直接使用 run_distributed 恢复执行
        # ``execution_id`` 必须显式透传（plaita#123 配套）：不传时引擎侧
        # 的 execution_id 与 worker 脱钩——resume 走 ``context.context =
        # saved_context`` 整体替换 checkpoint，若 checkpoint 缺
        # ``$EXECUTION_ID``（resume 常见），引擎 execution_id 为空串 ⇒
        # ``result.execution_id`` 回空、且重试计数键退化成
        # ``noderetry::{node}``（无执行 id），运维无法从键反查执行。
        # start 路径（本文件 ``start_flow``）早已是同款正确范式。
        #
        # ``params``（#122）：仅在 ``state.context`` 为空（假值）时被引擎
        # 用于 ``setup_flow``——即 start→resume 载体转换撞上的「首节点还
        # 没成功过」窗口；checkpoint 非空时引擎走真值分支，params 不参与
        # （续跑语义零变化，见 ``DistributedStrategy.execute``）。
        result = execution.run_distributed(
            flow,
            params=params,
            saved_context=state.context,
            resume_type=resume_type,
            resume_data=data,
            execution_id=execution_id,
        )

        # 处理执行结果
        final_result = self._process_execution_result(
            flow,
            result,
            state,
            execution,
            lease_execution_id=execution_id,
            lease_holder=lease_value,
            delivery_count=delivery_count,
        )

        return final_result

    except ExecutionLeaseError:
        raise
    except (StatePersistError, ServiceDispatchError, NodeExecutionRetryableError):
        # 落盘/派发失败：执行状态保持原样（is_end 前失败 → 仍 running，
        # 挂起派发失败 → 仍 suspended），消息不 ack 走 at-least-once 重投，
        # 下轮从 checkpoint / 挂起节点重入。绝不落 error 终态——重投消息
        # 会命中上方 already_terminal 短路被 ack，执行永久卡死。
        # NodeExecutionRetryableError（波次二任务①）同理：节点失败未
        # 终态化，重投后从 checkpoint 重跑失败节点；finally 释放租约，
        # 重投消息可重新取得租约。
        raise
    except FlowHashMismatchError:
        # flow 定义指纹不匹配（波次二任务②）：执行已在 raise 前终态化
        # error（可观测终态），原样上抛给 run() 按 ValueError poison
        # ack——重投只会命中 already_terminal 短路，终态化已完成，无意义。
        # 不落进下面的通用 except 二次终态化/包 RuntimeError。
        raise
    except ResumeProtocolError:
        # plaita#240：预检在**状态变更前**抛出的协议错误，必须原样上抛给
        # run()（ack、不终态化）——不能被下面通用 except 当作执行失败
        # 二次终态化 error。策略层守卫路径仍由 _chain_has_resume_guard_error
        # 在通用 except 内包成同类异常。
        raise
    except Exception as e:
        logger.error("恢复流程执行出错: %s", _scrub_exception_text(e))

        # 波次③步内取消：被恢复的节点在途命中取消监听 → 引擎抛
        # FlowCancelledException（归一化后挂 __cause__）或协作节点自杀抛
        # 普通异常。两条判据任一命中即终态化 cancelled，绝不写 error。
        if _chain_has_cancellation(e) or self._cancel_requested(execution_id):
            state.status = "cancelled"
            state.end_time = datetime.now().isoformat()
            self._persist_state_or_raise(execution_id, state, "cancelled_in_node")
            self._finalize_observers()
            logger.info(
                "执行 %s 在恢复节点执行期响应取消，终态化 cancelled", execution_id
            )
            return {
                "execution_id": execution_id,
                "status": "cancelled",
            }

        # 挂起守卫豁免（#33）：**只**豁免策略层守卫 ``ResumeGuardError``
        # （continue/retry 试图绕过 pending 挂起节点、resume 打在无挂起/
        # 非挂起 checkpoint 上）——那是调用方协议与执行状态不匹配，不是
        # 执行自身失败。终态化 error 会把本可被 event/cancel/timeout 唤醒的
        # 挂起执行永久打封（终态短路从此拒绝一切 resume 类型，死局），且这类
        # 错误天然由 at-least-once 重投产生（重复投递的 event resume 打在已
        # 推进的 checkpoint 上同族），终态化会误杀健康执行。
        # 豁免为「保持原状 + 上抛」：suspended 执行保持 suspended，消息被
        # ack（重投也只会再次命中同一守卫，重投无意义）；running 执行保持
        # checkpoint 现状。
        # 裸 ``ResumeError``（节点 ``resume()`` 抛错被策略层包的那个，
        # 如事件数据畸形）**不豁免**：静默保持原状 = 无 error 记录、永远等
        # 不到决议的哑执行；交给下方现状路径（可重试判据 → 终态化 error）
        # 才可观测、可人工 retry。
        if _chain_has_resume_guard_error(e):
            logger.warning(
                "执行 %s (status=%s) 收到与挂起状态不匹配的 resume 请求，"
                "协议错误不终态化，执行保持原状: %s",
                execution_id, getattr(state, "status", "?"), _scrub_url_text(e),
            )
            self._finalize_observers()
            raise ResumeProtocolError(
                f"resume 协议错误（执行保持原状 status={getattr(state, 'status', '?')}）: {e}",
                execution_id=execution_id,
            ) from e

        # 节点级有界重试（波次二任务①）：同 _process_execution_result，
        # 瞬态节点失败不终态化、消息等重投；否则现状终态化 error。
        # 失败节点 id 取自异常链（plaita#123：计数键按节点隔离）；取不到
        # 时计数退化为执行维度，终态化 error 里也就没有按节点的次数。
        failed_node = _node_id_from_chain(e)
        retry_exc = self._node_failure_retry_decision(
            e, execution_id, delivery_count, failed_node
        )
        if retry_exc is not None:
            raise retry_exc from e

        # 更新执行状态为错误
        state.status = "error"
        retries = max(
            0, self._read_node_retry_counter(execution_id, failed_node) - 1
        )
        if retries > 0:
            state.error = {
                "message": f"节点执行失败（重试 {retries} 次后仍失败）: {e}",
                "node_retries": retries,
                # plaita#93：终态化那一刻把引擎已算出的 code / error_type 与
                # HTTP 明细摘要如实落盘（message 保留，老消费方零影响）。
                **_error_fields(e),
            }
        else:
            state.error = {"message": str(e), **_error_fields(e)}
        if failed_node:
            state.error["failed_node"] = failed_node
        state.end_time = datetime.now().isoformat()

        self._persist_state_or_raise(execution_id, state, "resume_error_handler")
        self._finalize_observers()

        # plaita#73：终态化**已完成**，必须 poison ack 终止重投循环。
        # 原先抛裸 RuntimeError → 落到 run() 兜底 except → 不 ack → 重投
        # → status=error 只放行 retry → 再撞同一确定性失败 → 无限循环
        # （且全程不碰重试计数键，预算永不耗尽）。
        raise NodeFailureTerminalizedError(
            f"恢复流程执行出错（已终态化 error，消息 poison ack）: {e}"
        )
    finally:
        self._unregister_lease_watch(execution_id, lease_value)
        self._unregister_cancel_watch(execution_id)
        if fence_token_reset is not None:
            reset_current_fence_token(fence_token_reset)
        # release 对整个租约 value 串 compare（fencing 档 = {holder}:{gen}）
        self.execution_lease.release(execution_id, lease_value)

init_control

init_control(redis_client: Redis, instance_id: str)

初始化控制监听

参数：

名称	类型	描述	默认
redis_client	Redis	Redis 客户端	必需
instance_id	str	实例 ID	必需

源代码位于： plaita/server/control.py

def init_control(
    self,
    redis_client: Redis,
    instance_id: str
):
    """
    初始化控制监听

    Args:
        redis_client: Redis 客户端
        instance_id: 实例 ID
    """
    self._control_listener = ControlListener(
        redis_client=redis_client,
        instance_id=instance_id,
        on_stop=self._on_stop_command,
        on_status=self._on_status_command
    )

start_control_listener

start_control_listener()

启动控制监听

源代码位于： plaita/server/control.py

def start_control_listener(self):
    """启动控制监听"""
    if self._control_listener:
        self._control_listener.start()

stop_control_listener

stop_control_listener()

停止控制监听

源代码位于： plaita/server/control.py

def stop_control_listener(self):
    """停止控制监听"""
    if self._control_listener:
        self._control_listener.stop()

init_registry

init_registry(redis_client: Redis, service_type: str, metadata: Optional[Dict[str, Any]] = None, ttl: int = DEFAULT_TTL, heartbeat_interval: int = DEFAULT_HEARTBEAT_INTERVAL, instance_id: Optional[str] = None)

初始化服务注册

参数：

名称	类型	描述	默认
redis_client	Redis	Redis 客户端	必需
service_type	str	服务类型	必需
metadata	Optional[Dict[str, Any]]	服务元数据	None
ttl	int	TTL（秒）	DEFAULT_TTL
heartbeat_interval	int	心跳间隔（秒）	DEFAULT_HEARTBEAT_INTERVAL

源代码位于： plaita/server/registry.py

def init_registry(
    self,
    redis_client: Redis,
    service_type: str,
    metadata: Optional[Dict[str, Any]] = None,
    ttl: int = ServiceRegistry.DEFAULT_TTL,
    heartbeat_interval: int = ServiceRegistry.DEFAULT_HEARTBEAT_INTERVAL,
    instance_id: Optional[str] = None,
):
    """
    初始化服务注册

    Args:
        redis_client: Redis 客户端
        service_type: 服务类型
        metadata: 服务元数据
        ttl: TTL（秒）
        heartbeat_interval: 心跳间隔（秒）
    """
    self._registry = ServiceRegistry(
        redis_client=redis_client,
        ttl=ttl,
        heartbeat_interval=heartbeat_interval
    )
    # 注销世代闸（S18-02A / A46-03）：注销成功后所有写入 no-op。
    # 惰性 helper 在 ``_registry_is_closed`` / ``_mark_registry_closed``，
    # 骨架 worker（不走 init_registry 的 __new__ 单测）同样安全。
    self._registry_closed = threading.Event()

    self._service_info = ServiceInfo(
        # B2R31-02：显式实例 id（service_manager 注入的 PLAITA_INSTANCE_ID）
        # 与注册键必须同值；未注入才生成 hostname-uuid。
        instance_id=instance_id or ServiceRegistry.generate_instance_id(),
        service_type=service_type,
        host=ServiceRegistry.get_host_address(),
        status="starting",
        metadata=metadata or {}
    )

flush_registry_status

flush_registry_status(retries: int = 3, backoff: float = 0.1) -> bool

立即把当前 ServiceInfo（含 status/active/capacity）落盘。

request_drain / stop 不再等 10s 心跳：内存置位后直接写一次。 Redis 瞬断时短退避重试（默认 3 次，0.1s/0.2s），仍失败返回 False （调用方负责 ERROR + 指标，不得静默）。

源代码位于： plaita/server/registry.py

def flush_registry_status(
    self, retries: int = 3, backoff: float = 0.1
) -> bool:
    """**立即**把当前 ``ServiceInfo``（含 status/active/capacity）落盘。

    ``request_drain`` / ``stop`` 不再等 10s 心跳：内存置位后直接写一次。
    Redis 瞬断时短退避重试（默认 3 次，0.1s/0.2s），仍失败返回 False
    （调用方负责 ERROR + 指标，不得静默）。
    """
    if self._registry_is_closed():
        return True  # 已注销：无键可写，不算失败
    registry = getattr(self, "_registry", None)
    service_info = getattr(self, "_service_info", None)
    if registry is None or service_info is None:
        return False
    attempts = max(1, int(retries))
    for attempt in range(1, attempts + 1):
        if registry.heartbeat(service_info):
            return True
        if attempt < attempts:
            time.sleep(max(0.0, float(backoff)) * attempt)
    return False

register_service

register_service() -> bool

注册服务并启动心跳。

源代码位于： plaita/server/registry.py

def register_service(self) -> bool:
    """注册服务并启动心跳。"""
    if self._registry_is_closed():
        logger.warning("服务注册已关闭（注销后不得复活），忽略本次注册")
        return False
    if not self._registry or not self._service_info:
        logger.warning("服务注册未初始化")
        return False

    success = self._registry.register(self._service_info)
    if success:
        self._registry.start_heartbeat(
            self._service_info,
            should_write=self._registry_heartbeat_allowed,
        )

    return success

unregister_service

unregister_service() -> bool

注销服务并停止心跳（幂等；入口先置 closed 封住迟到写入）。

源代码位于： plaita/server/registry.py

def unregister_service(self) -> bool:
    """注销服务并停止心跳（幂等；入口先置 closed 封住迟到写入）。"""
    if not self._registry or not self._service_info:
        return False

    self._mark_registry_closed()
    self._registry.stop_heartbeat()

    return self._registry.unregister(
        self._service_info.service_type,
        self._service_info.instance_id
    )

update_registry_info

update_registry_info(active_tasks: Optional[int] = None, metadata: Optional[Dict[str, Any]] = None, status: Optional[str] = None)

更新注册信息（status 用于 draining 等生命周期状态，None = 不改）。

直接改 _service_info（单一状态源）——register() 之前调用的 request_drain 也要能在内存中生效，否则 update_service_info 在 _registered_service is None 时直接 return（S9 前置坑）。 注销后（closed）任何更新都是 no-op。

源代码位于： plaita/server/registry.py

def update_registry_info(
    self,
    active_tasks: Optional[int] = None,
    metadata: Optional[Dict[str, Any]] = None,
    status: Optional[str] = None
):
    """更新注册信息（status 用于 draining 等生命周期状态，None = 不改）。

    直接改 ``_service_info``（单一状态源）——``register()`` 之前调用的
    ``request_drain`` 也要能在内存中生效，否则 ``update_service_info``
    在 ``_registered_service is None`` 时直接 return（S9 前置坑）。
    注销后（closed）任何更新都是 no-op。
    """
    if self._registry_is_closed():
        logger.debug("注册表已关闭，忽略状态更新 status=%s", status)
        return
    service_info = getattr(self, "_service_info", None)
    if service_info is not None:
        if active_tasks is not None:
            service_info.active_tasks = active_tasks
        if metadata:
            service_info.metadata.update(metadata)
        if status:
            service_info.status = status
    if self._registry:
        self._registry.update_service_info(
            active_tasks=active_tasks,
            metadata=metadata,
            status=status,
        )

sweep_paused_sandboxes

sweep_paused_sandboxes(max_age_secs: int = 6 * 3600) -> list

同步执行一次暂停清扫（测试/手工调用用）。

worker 启动路径请用 _paused_sweeper() + 后台线程（import 前置到主线程）。

源代码位于： plaita/server/flow_worker.py

def sweep_paused_sandboxes(max_age_secs: int = 6 * 3600) -> list:
    """同步执行一次暂停清扫（测试/手工调用用）。

    worker 启动路径请用 ``_paused_sweeper()`` + 后台线程（import 前置到主线程）。
    """
    run = _paused_sweeper(max_age_secs)
    return run() if run is not None else []

classify_flow_hash_change

classify_flow_hash_change(*, stored_hash: Optional[str], current_hash: str, stored_algo: Optional[str], current_algo: str, allow_change: bool) -> str

resume 时 flow 指纹变化的判定表（纯函数，便于测试与复用）。

返回： - no_guard ：老状态没有指纹，跳过校验（存量语义） - match ：指纹相同、算法标签也相同 - refresh ：指纹相同但算法标签升级——定义确实没变，续跑并刷新标签 - accepted ：指纹不同 + 算法标签不同 + 调用方显式放行（升级后的补救入口） - mismatch ：其余指纹不同——按定义变更处理，拒绝续跑并给可执行的提示

源代码位于： plaita/server/flow_worker.py

def classify_flow_hash_change(
    *,
    stored_hash: Optional[str],
    current_hash: str,
    stored_algo: Optional[str],
    current_algo: str,
    allow_change: bool,
) -> str:
    """resume 时 flow 指纹变化的判定表（纯函数，便于测试与复用）。

    返回：
    - ``no_guard``  ：老状态没有指纹，跳过校验（存量语义）
    - ``match``     ：指纹相同、算法标签也相同
    - ``refresh``   ：指纹相同但算法标签升级——定义确实没变，续跑并刷新标签
    - ``accepted``  ：指纹不同 + 算法标签不同 + 调用方显式放行（升级后的补救入口）
    - ``mismatch``  ：其余指纹不同——按定义变更处理，拒绝续跑并给可执行的提示
    """
    if not stored_hash:
        return "no_guard"
    if stored_hash == current_hash:
        return "match" if stored_algo == current_algo else "refresh"
    algo_changed = bool(stored_algo) and stored_algo != current_algo
    if algo_changed and allow_change:
        return "accepted"
    return "mismatch"

canonical_flow_definition_json

canonical_flow_definition_json(flow_definition: Mapping[str, Any]) -> str

把存储中的原始流程定义规范化为稳定 JSON 文本（指纹的输入面）。

plaita#36：指纹只取这一层——model_validate 之前的定义才是「作者写下的 语义」。解析后的 Flow.model_dump(mode="json") 会随模型字段演进（新增/ 删除字段、默认值的物化、别名规范化）改变形态，同一份存储定义在新旧引擎下 算出不同指纹，滚动升级窗口把在途 run 批量终态化。

规范化口径：sort_keys（dict 序无关）+ ensure_ascii=False + default=str（非 JSON 原生值降级为 str，与 sanitize_context_for_persist 同款兜底，绝不为「日志/指纹路径」抛）。

源代码位于： plaita/server/flow_worker.py

def canonical_flow_definition_json(flow_definition: Mapping[str, Any]) -> str:
    """把**存储中的原始流程定义**规范化为稳定 JSON 文本（指纹的输入面）。

    plaita#36：指纹只取这一层——``model_validate`` 之前的定义才是「作者写下的
    语义」。解析后的 ``Flow.model_dump(mode="json")`` 会随模型字段演进（新增/
    删除字段、默认值的物化、别名规范化）改变形态，同一份存储定义在新旧引擎下
    算出不同指纹，滚动升级窗口把在途 run 批量终态化。

    规范化口径：``sort_keys``（dict 序无关）+ ``ensure_ascii=False`` +
    ``default=str``（非 JSON 原生值降级为 str，与 ``sanitize_context_for_persist``
    同款兜底，绝不为「日志/指纹路径」抛）。
    """
    return json.dumps(
        flow_definition,
        sort_keys=True,
        ensure_ascii=False,
        default=str,
        separators=(",", ":"),
    )

compute_flow_definition_hash

compute_flow_definition_hash(flow_definition: Mapping[str, Any]) -> str

原始定义 JSON 的 sha256（跨进程/跨引擎版本确定）。

源代码位于： plaita/server/flow_worker.py

def compute_flow_definition_hash(flow_definition: Mapping[str, Any]) -> str:
    """原始定义 JSON 的 sha256（跨进程/跨引擎版本确定）。"""
    return hashlib.sha256(
        canonical_flow_definition_json(flow_definition).encode("utf-8")
    ).hexdigest()

sanitize_context_for_persist

sanitize_context_for_persist(context: Any) -> Optional[dict]

80 层 1：把不可 JSON 化的值降级为 str，返回副本。

绝不就地改原对象 —— 原 context 仍被其它路径引用（终态化是收尾动作， 不该反向改动流程语义）。

返回 None 表示「连兜底都失败」（如循环引用，json.dumps 会抛 ValueError: Circular reference detected）⇒ 由调用方走层 2（最小 error 状态）。

源代码位于： plaita/server/flow_worker.py

def sanitize_context_for_persist(context: Any) -> Optional[dict]:
    """#80 层 1：把不可 JSON 化的值**降级为 str**，返回**副本**。

    **绝不就地改原对象** —— 原 ``context`` 仍被其它路径引用（终态化是收尾动作，
    不该反向改动流程语义）。

    返回 ``None`` 表示「连兜底都失败」（如**循环引用**，``json.dumps`` 会抛
    ``ValueError: Circular reference detected``）⇒ 由调用方走层 2（最小 error 状态）。
    """
    def _default(o: Any) -> str:
        try:
            return str(o)
        except Exception:  # noqa: BLE001 —— __str__ 自己抛异常的对象
            return f"<unstr-able {type(o).__name__}>"

    try:
        return json.loads(json.dumps(context, default=_default))
    except Exception:  # noqa: BLE001 —— 循环引用等，交给层 2
        return None

main

main()

命令行入口程序

源代码位于： plaita/server/flow_worker.py

def main():
    """命令行入口程序"""
    # B2R31-03：身份/安全/资源类非法值在启动期一次聚合、fail-fast；可选调优
    # 类降级为 WARNING + 安全默认（不阻断，但绝不静默）。
    preflight_warnings = []
    try:
        preflight_warnings = preflight_security_and_limits()
        _runtime_cfg = load_runtime_config("worker", strict=True)
    except ConfigError as exc:
        print(f"配置预检失败，worker 拒绝启动：{exc}", file=sys.stderr)
        raise SystemExit(2)
    # CLI 默认给控制台日志：库内 logger 无 handler，原样跑运维看到的是
    # 零输出（2026-09 分布式评审 P2-6）。--quiet / PLAITA_LOG_LEVEL 可调。
    logging.basicConfig(
        level=os.environ.get("PLAITA_LOG_LEVEL", "INFO"),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    log_config_effective(logger, _runtime_cfg)
    # preflight 的 WARNING 发生在 logging.basicConfig 之前（NullHandler），
    # 这里补打一次，确保可选调优降级在启动日志可见（B2R31-03）。
    for warning in preflight_warnings:
        logger.warning("配置预检：%s", warning)
    parser = argparse.ArgumentParser(description="Plaita流程工作器")

    # Redis参数（统一 resolver：PLAITA_REDIS_URL > REDIS_URL > 默认）
    parser.add_argument("--redis-url",
                      default=resolve_redis_url("worker"),
                      help="Redis连接URL")
    parser.add_argument("--queue-name",
                      default=resolve_queue_name("worker"),
                      help="Redis Stream 键名（任务队列，需 Redis 5+）")
    parser.add_argument("--consumer-group",
                      default=resolve_consumer_group(),
                      help="Stream consumer group 名称")
    parser.add_argument("--consumer-name",
                      default=os.environ.get("PLAITA_CONSUMER_NAME"),
                      help="本 worker 的 consumer 名称（默认 instance_id 或 worker-pid）")
    parser.add_argument("--claim-min-idle-ms", type=int,
                      default=int(os.environ.get("PLAITA_CLAIM_MIN_IDLE_MS", str(DEFAULT_CLAIM_MIN_IDLE_MS))),
                      help="pending 消息最短空闲毫秒数后才可被其他 consumer 回收")
    parser.add_argument("--lease-ttl-seconds", type=int,
                      default=int(os.environ.get("PLAITA_LEASE_TTL_SECONDS", str(DEFAULT_LEASE_TTL_SECONDS))),
                      help="resume execution lease TTL（秒），推进中会 renew")
    parser.add_argument("--max-deliveries", type=int,
                      default=int(os.environ.get("PLAITA_MAX_DELIVERIES", str(DEFAULT_MAX_DELIVERIES))),
                      help="任务最大投递次数，超过后写入 DLQ 并 ack")
    parser.add_argument("--dlq-key",
                      default=resolve_dlq_key(resolve_queue_name("worker"), empty_policy="default"),
                      help="死信 Stream 键（默认 <queue-name>:dlq）")

    # 数据库参数
    parser.add_argument("--database-url", default="sqlite:///flow.db",
                      help="数据库连接URL")

    # 存储组件类型（execution/flow 仅 memory|redis；db 与同步 ABC 不兼容，见 factory）
    parser.add_argument("--execution-storage-type", choices=["memory", "redis"], default="redis",
                      help="执行状态存储类型（memory|redis；db 已下架）")
    parser.add_argument("--flow-storage-type", choices=["memory", "redis"], default="redis",
                      help="流程定义存储类型（memory|redis；db 已下架）")

    # 事件总线参数（默认启用：分布式挂起/恢复依赖订阅写入；
    # 历史 --use-event-bus 为 opt-in，导致默认部署订阅落内存、EventFilter 读 Redis）
    parser.add_argument("--event-bus-type", choices=["memory", "redis"], default="redis",
                      help="事件总线类型（生产用 redis；db/sqlalchemy 已标 experimental，见 factory）")
    parser.add_argument("--use-event-bus", action="store_true",
                      help=argparse.SUPPRESS)  # 已默认启用，保留兼容旧脚本
    parser.add_argument("--no-event-bus", action="store_true",
                      help="禁用事件总线（仅无挂起节点的纯同步场景）")

    # 缓存参数
    parser.add_argument("--cache-size", type=int, default=100,
                      help="缓存大小")
    parser.add_argument("--cache-ttl", type=int, default=300,
                      help="缓存TTL(秒)")

    # 服务注册参数
    parser.add_argument("--enable-registry", action="store_true", default=True,
                      help="启用服务注册")
    parser.add_argument("--no-registry", action="store_true",
                      help="禁用服务注册")
    parser.add_argument("--registry-ttl", type=int, default=30,
                      help="服务注册TTL(秒)")
    parser.add_argument("--read-block-ms", type=int, default=1_000,
                        help="XREADGROUP 阻塞窗口上限（毫秒）。默认 1000：让 SIGTERM "
                             "后的停机延迟 ≤1s；调大可略降 Redis 往返次数")
    parser.add_argument("--concurrency", type=int,
                        default=int(os.environ.get("PLAITA_WORKER_CONCURRENCY", "1")),
                        help="并发消费线程数（默认 1=串行）。>1 时单机可同时处理多个任务；"
                             "轻任务（TS/Python/Node）可放大，Rust 重构建任务建议保持 1。"
                             "也可用环境变量 PLAITA_WORKER_CONCURRENCY")
    parser.add_argument("--quiet", action="store_true",
                        help="关闭 INFO 级控制台日志（等价 PLAITA_LOG_LEVEL=WARNING）")
    parser.add_argument("--min-free-disk-gib", type=float,
                        default=_min_free_disk_gib_from_env(),
                        help="claim 前的本机可用盘守线（GiB）：低于它就不领任务，"
                             "空转重探、盘回线自动恢复（不产生白跑 claim）。"
                             "默认 0=关闭；跟随 PLAITA_WORKER_MIN_FREE_DISK_GIB，"
                             "应与 flow 内 preflight 节点的阈值对齐")
    parser.add_argument("--disk-guard-path", default=_disk_guard_path_from_env(),
                        help="盘预检路径（默认当前工作目录或 PLAITA_WORKER_DISK_GUARD_PATH）")
    parser.add_argument("--heartbeat-interval", type=int, default=10,
                      help="心跳间隔(秒)")
    parser.add_argument("--metrics-port", type=int,
                      default=_metrics_port_from_env(),
                      help="Prometheus /metrics 抓取端端口（默认跟随 PLAITA_METRICS_PORT；"
                           "0 = 不启动）。暴露 stream/pending/dlq/计数器/心跳")
    parser.add_argument("--metrics-host", default=_metrics_host_from_env(),
                      help="/metrics 抓取端监听地址（默认 0.0.0.0 或 PLAITA_METRICS_HOST）")
    parser.add_argument("--alert-webhook", default=_alert_webhook_from_env(),
                      help="死信告警 webhook URL（JSON POST）；默认跟随 PLAITA_ALERT_WEBHOOK")
    parser.add_argument("--langfuse", action="store_true", default=None,
                        help="启用 Langfuse 观测（需 pip install plaita[langfuse]；"
                             "凭据走 LANGFUSE_PUBLIC_KEY/SECRET_KEY/HOST 环境变量）。"
                             "也可用环境变量 PLAITA_WORKER_LANGFUSE=1 开启")

    args = parser.parse_args()
    if args.quiet:
        logging.getLogger().setLevel(logging.WARNING)

    # writefile 写入 jail（plaita#39）：writefile 节点默认任意路径可写，而部署入口
    # 此前从不设置 PLAITA_NODES_WORKSPACE_ROOT——机制在节点侧，门在运营侧缺省开着。
    # worker 启动即注入（未显式配置则 fail-closed 推导默认根）。
    apply_writefile_jail("flow-worker")

    callback_handlers = []
    if args.langfuse or os.environ.get("PLAITA_WORKER_LANGFUSE") == "1":
        try:
            from plaita.obs import LangfuseCallback

            callback_handlers.append(LangfuseCallback())
            logger.info("Langfuse 观测已启用（trace id = 运行时 $EXECUTION_ID）")
        except ImportError as e:
            logger.warning("Langfuse 观测未启用（缺依赖）: %s", _scrub_url_text(e))
        except Exception as e:  # noqa: BLE001 — SDK 初始化失败（如缺凭据）只降级不退出
            logger.warning("Langfuse 观测未启用: %s", _scrub_url_text(e))

    # 外部业务节点模块加载（与 console 的 PLAITA_CONSOLE_NODE_MODULES 约定对齐）：
    # PLAITA_NODE_PATH 冒号分隔追加 sys.path；PLAITA_NODE_MODULES 逗号分隔，
    # 逐个 import 并调用其 register_all()（或 register）。业务仓（如 mediaflow
    # 的 plaita_flows.nodes）由此在 console 拉起的 worker 内生效。
    import importlib

    for extra in [p for p in os.environ.get("PLAITA_NODE_PATH", "").split(os.pathsep) if p]:
        if extra not in sys.path:
            sys.path.insert(0, extra)
    for mod_path in [m.strip() for m in os.environ.get("PLAITA_NODE_MODULES", "").split(",") if m.strip()]:
        try:
            mod = importlib.import_module(mod_path)
            register = getattr(mod, "register_all") or getattr(mod, "register")
            register()
            logger.info("已加载外部节点模块: %s", mod_path)
        except Exception as e:
            logger.error("外部节点模块加载失败 %s: %s", mod_path, _scrub_exception_text(e))
            raise SystemExit(f"外部节点模块加载失败: {mod_path}")

    # CodeNode 按需注册：0.4.0 起移出默认注册表（其执行任意用户代码，需显式 opt-in）。
    # 生产流程（如 self-improve-v2）含 code 节点，worker 必须注册否则整单被丢弃
    # （「unRecognized node type: code」）。后端由 PLAITA_CODE_BACKEND 选择，默认
    # subprocess（无需 docker daemon）；声明值非法或配置后端不可用时**具名拒绝并
    # 停止启动**（A56-01 fail-closed，不自动降级；显式逃生开关见下方 helper）。
    # 后端白名单由 PLAITA_SANDBOX_ALLOWED_BACKENDS 配置（见 _register_code_node_for_worker）。
    if _code_node_enabled():
        _register_code_node_for_worker()

    # 沙箱暂停实例清扫（启动一次性、后台线程）：**import 已在主线程完成**
    # （`_paused_sweeper`）——线程内 import 会与主线程懒加载撞 import 锁并整体卡死
    # （2026-10-08 实测）。线程内只做网络调用。
    _sweeper = _paused_sweeper()
    if _sweeper is not None:
        threading.Thread(target=_sweeper, name="sandbox-sweep", daemon=True).start()

    # 处理注册开关
    enable_registry = args.enable_registry and not args.no_registry

    try:
        # 创建执行状态存储
        storage_kwargs = {
            "redis_url": args.redis_url,
            "database_url": args.database_url
        }

        execution_storage = create_storage_component(
            args.execution_storage_type,
            "execution",
            tenant_routing=True,
            **storage_kwargs
        )
        logger.info("已创建执行状态存储: %s类型（多租户路由）", args.execution_storage_type)

        # 创建流程定义存储
        flow_storage = create_storage_component(
            args.flow_storage_type,
            "flow",
            tenant_routing=True,
            **storage_kwargs
        )
        logger.info("已创建流程定义存储: %s类型（多租户路由）", args.flow_storage_type)

        # 创建事件总线（默认启用；--no-event-bus 显式关闭）
        event_bus = None
        if not args.no_event_bus:
            event_bus = create_event_bus(
                args.event_bus_type,
                **storage_kwargs
            )
            logger.info("已创建事件总线: %s类型", args.event_bus_type)
        else:
            logger.warning("已禁用事件总线（--no-event-bus）；含挂起节点的流程将无法恢复")

        # 创建Redis流程工作器
        worker = RedisFlowWorker(
            redis_url=args.redis_url,
            queue_name=args.queue_name,
            execution_storage=execution_storage,
            flow_storage=flow_storage,
            event_bus=event_bus,
            callback_handlers=callback_handlers or None,
            cache_size=args.cache_size,
            cache_ttl=args.cache_ttl,
            enable_registry=enable_registry,
            registry_ttl=args.registry_ttl,
            heartbeat_interval=args.heartbeat_interval,
            consumer_group=args.consumer_group,
            consumer_name=args.consumer_name or None,
            claim_min_idle_ms=args.claim_min_idle_ms,
            lease_ttl_seconds=args.lease_ttl_seconds,
            max_deliveries=args.max_deliveries,
            dlq_key=args.dlq_key or None,
            read_block_ms=args.read_block_ms,
            concurrency=args.concurrency,
            min_free_disk_gib=args.min_free_disk_gib,
            disk_guard_path=args.disk_guard_path,
            metrics_port=args.metrics_port,
            metrics_host=args.metrics_host,
            alert_webhook=args.alert_webhook or None,
        )

        # 注册信号处理器以支持优雅关闭。只置位、由 run() 主循环在任务边界
        # 自然退出；不再 sys.exit 腰斩在途任务（ReviewFix D3，语义见
        # _request_graceful_stop）。
        def signal_handler(signum, frame):
            _request_graceful_stop(worker, signum)

        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)

        # 启动工作器。历史上这里有 --debug-mode 分支硬编码 flow_id="event_flow_demo"
        # 直接读 Redis + 手动 lrem 队列消息, 是开发期临时脚本——已删除。需要类似
        # 调试请用 Redis CLI 或独立 dev 脚本, 不要留在生产 CLI 入口里。
        logger.info("流程工作器启动成功，监听队列: %s", args.queue_name)
        worker.run()

    except Exception as e:
        logger.error("流程工作器启动失败: %s", _scrub_url_text(e))
        sys.exit(1)