fix(storage): 写侧序列化失败与 Redis 瞬断混为同一个 False——上下文含非 JSON 值时 save 永久 False，worker 不 ack 反复重投、副作用重放且执行永不终态（memory 后端同一 state 恒 True）

Start in plaita/storage/redis.py, save_execution_state (around lines 281-316), where serialization and the Redis pipeline share one except that returns False, and in serialize_state in plaita/storage/base.py. Reproduce with the fakeredis snippet in section 9 and compare with MemoryExecutionStorage. Done means a non-JSON context value raises a distinct, non-ValueError error or saves successfully, transient ConnectionError still returns False, and tests/unit/test_worker_save_guard.py passes. The worker-side handling needs a design decision first.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

RedisExecutionStorage.save_execution_state 把「状态序列化失败」与「Redis 瞬断」包在同一个
except Exception → return False 里：只要 state.context 里出现过一个非 JSON 值，此后
每一次落盘都失败，而 worker 把这个 False 一律当作瞬态故障处理（不 ack、等重投），
于是该执行反复从 checkpoint 重跑同一节点、副作用重复发生，直到投递次数耗尽进 DLQ——执行
既跑不完也拿不到任何终态。同一个 state 在 memory 后端恒返回 True。

1. 写侧序列化器只做 json.dumps，失败即 re-raise（plaita/storage/base.py:142-146）：

        try:
            return json.dumps(state)
        except Exception as e:
            logger.error("Failed to serialize state: %s", e)
            raise

2. 症结：序列化与 Redis 写共用一个 try、共用同一个 False（plaita/storage/redis.py:296-316）：

        try:
            serialized = self.serialize_state(state.model_dump())
            ttl = (
                execution_state_ttl_seconds()
                if state.status in TERMINAL_EXECUTION_STATUSES
                else 0
            )
            pipe = self.client.pipeline()
            if ttl > 0:
                pipe.set(key, serialized, ex=ttl)
            else:
                pipe.set(key, serialized)
            pipe.zadd(
                self._execution_index_key(),
                {execution_id: execution_start_time_score(state)},
            )
            pipe.execute()
            return True
        except Exception as e:
            logger.error("Failed to save execution state %s: %s", execution_id, e)
            return False

serialize_state 是 try 内第一条语句。它抛出的 TypeError（永久）与 pipe.execute()
抛出的 ConnectionError（瞬态）落到同一分支、返回同一个 False，调用方无法区分「重投可
恢复」与「重投永远不可能成功」。

3. context 是无校验的 Dict[str, Any]，节点返回值原样落进去（plaita/storage/base.py:54）：

    context: Dict[str, Any]

（plaita/core/state.py:516-521）：

    def update_node_result(self, node_id: str, result: Any) -> None:
        """Record a node's result under ``$NODE`` (creating the map if absent)."""
        key = f"{self.prefix}{self.node_name}"
        node_results = self.get(key, {})
        node_results[node_id] = result
        self[key] = node_results

因此某个节点 run() 返回 set / datetime / bytes / 自定义对象时，该值不做任何强制转换
就随 state.context 进入 json.dumps。

4. 同一 state 在 memory 后端恒 True（plaita/storage/memory.py:15-18）：

    def save_execution_state(self, execution_id: str, state: ExecutionState) -> bool:
        """保存流程执行状态"""
        self.execution_states[execution_id] = state.model_dump()
        return True

memory 后端不经过 JSON 序列化，非 JSON 值原样进 dict——同一份 context 换后端行为分叉。

5. False 被提升为异常（plaita/server/flow_worker.py:871-882）：

        saved = self.execution_storage.save_execution_state(execution_id, state)
        if not saved:
            logger.error(
                "保存执行状态失败 (%s): execution_id=%s, status=%s——消息将不 ack 等待重投",
                phase,
                execution_id,
                getattr(state, "status", "?"),
            )
            raise StatePersistError(
                f"保存执行状态失败 ({phase}): execution_id={execution_id}, "
                f"status={getattr(state, 'status', '?')}"
            )

6. 异常类自述了这条混淆：把 False 一律当瞬态（plaita/server/flow_worker.py:405-415）：

class StatePersistError(RuntimeError):
    """执行状态落盘失败（``save_execution_state`` 返回 False）。

    Redis 后端把一切异常（网络瞬断/序列化失败）吞成 False；终态/挂起/步进
    落盘失败若被静默放行，消息会被 ack——节点副作用已发生而执行状态永远
    停在旧值（僵尸执行）。此处把 False 升级为异常，让 ``RedisFlowWorker.run``
    的兜底路径不 ack 消息、走 at-least-once 重投。

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

7. StatePersistError 在两条路径被原样上抛、不终态化（plaita/server/flow_worker.py:1679-1687）：

        except (StatePersistError, ServiceDispatchError, NodeExecutionRetryableError):
            # 落盘/派发失败：执行状态保持原样（is_end 前失败 → 仍 running，
            # 挂起派发失败 → 仍 suspended），消息不 ack 走 at-least-once 重投，
            # 下轮从 checkpoint / 挂起节点重入。绝不落 error 终态——重投消息
            # 会命中上方 already_terminal 短路被 ack，执行永久卡死。
            # NodeExecutionRetryableError（波次二任务①）同理：节点失败未
            # 终态化，重投后从 checkpoint 重跑失败节点；finally 释放租约，
            # 重投消息可重新取得租约。
            raise

（plaita/server/flow_worker.py:1870-1874）：

                except StatePersistError:
                    # 步进落盘失败必须原样上抛：若落进下面的通用 except，
                    # 会把可重投的瞬态失败改写成 error 终态（消息被 ack 后
                    # 重投命中 already_terminal 短路），执行永久卡死。
                    raise

StatePersistError 是 RuntimeError（非 ValueError），所以它最终落进 run() 的兜底分支。

8. run() 兜底：不 ack、反复重投，超限进 DLQ（plaita/server/flow_worker.py:2820-2842）：

            except Exception as exc:
                # 兜底重投递路径（at-least-once）：不 ack，等超时回收。
                # 刻意放在 except ValueError 之后——存储层抛出的
                # ExecutionStateLoadError（读取瞬断/状态损坏，ReviewFix D1）
                # 由此处理：消息留在 pending 重投，超过 max_deliveries 才
                # 进 DLQ；绝不能像 ValueError 一样被当畸形消息 poison ack，
                # 否则 Redis 抖动一次 = 挂起执行永久失去恢复机会。
                queue.note_failed()
                if task.delivery_count >= queue.max_deliveries:
                    queue.dead_letter(
                        task,
                        reason=f"processing_failed:{type(exc).__name__}:{exc}"[:500],
                    )
                    acked = True
                else:
                    logger.error(
                        "任务处理失败 %s (delivery=%s/%s)，未 ack（将重投）: %s",
                        task.message_id,
                        task.delivery_count,
                        queue.max_deliveries,
                        exc,
                        exc_info=True,
                    )

重投上限默认 5（plaita/server/task_queue.py:19：DEFAULT_MAX_DELIVERIES = 5）。

9. 实测：两种完全不同的失败给出同一个 False：

序列化失败（context 含 set 与 datetime）——memory 成功、redis 失败：

$ cd /Users/jeff.jie/projects/infra4agent/plaita && /tmp/plaita-audit-venv/bin/python -c '
import datetime
from plaita.storage.memory import MemoryExecutionStorage
from plaita.storage.base import ExecutionState
from plaita.storage.redis import RedisExecutionStorage
import fakeredis
s = ExecutionState(execution_id="e1", context={"$NODE": {"n1": {"result": {1,2}}}, "dt": datetime.datetime.now()})
print("memory save:", MemoryExecutionStorage().save_execution_state("e1", s))
r = RedisExecutionStorage(client=fakeredis.FakeStrictRedis(decode_responses=True))
print("redis save:", r.save_execution_state("e1", s))
print("redis load:", r.load_execution_state("e1"))
print("keys:", r.client.keys("*"))
'
memory save: True
redis save: False
redis load: None
keys: []

开日志后可见两次 error 同源、同一次调用：

LOG ERROR: Failed to serialize state: Object of type set is not JSON serializable
LOG ERROR: Failed to save execution state e1: Object of type set is not JSON serializable
redis save -> False

对照：context 干净、pipeline() 抛 ConnectionError 的桩客户端（模拟 Redis 瞬断）：

LOG ERROR: Failed to save execution state e2: simulated Redis 瞬断 (ConnectionError)
redis save (JSON-clean state, Redis down) -> False

即「永久性序列化失败」与「瞬态网络失败」返回值完全一致（都是 False），
没有任何终态出口或异常类型可区分。

10. 同仓其他落盘路径已有 JSON 兜底，本路径没有（plaita/server/flow_worker.py:1052-1057）：

        payload = json.dumps(
            flow.model_dump(mode="json"),
            sort_keys=True,
            ensure_ascii=False,
            default=str,
        )

11. 读侧对同类问题已有正确解法，可作为修法参照（plaita/storage/redis.py:111-123）：

class ExecutionStateLoadError(RuntimeError):
    """执行状态**读取失败**（Redis 瞬断/超时或数据损坏无法反序列化）。

    与「键不存在 → load_execution_state 返回 None」严格区分：调用方
    （FlowWorker.run）对前者不 ack、走重投递/DLQ 路径，对后者才做
    poison ack。历史上两类情况都被吞成 None，Redis 抖动一次 = 挂起
    执行的 resume 消息被当毒丸丢弃，永久失去恢复机会（ReviewFix D1）。

    注：语义上属于存储层公共契约，理想位置是 plaita/storage/base.py；
    本次修复受改动文件白名单约束先定义在 Redis 后端模块，后续可上移。
    （刻意不继承 ValueError——FlowWorker.run 把 ValueError 当畸形消息
    poison ack，继承它会让瞬态错误仍被丢弃。）
    """

读侧专门造异常来区分「瞬态」与「不存在」；写侧没有对应的「永久失败」出口。

影响

• 生产可达（Redis / 集群档）：只要某次执行写入 state.context 的值里含非 JSON-native
  对象，该执行的每一次 checkpoint 落盘都失败。触发源是真实节点行为：in-process code 节点的
  Python run() 直接返回 set / datetime / bytes / dataclass / 自定义对象；或经
  PLAITA_CONSOLE_NODE_MODULES、entry_points 注册的业务节点（plaita-nodes）返回同类值。
  不需要恶意输入，一个 datetime 字段就够。
• 代价一：worker 侧没有任何终态化路径。 StatePersistError 被刻意排除在通用 except
  之外并原样上抛（flow_worker.py:1679-1687、:1870-1874），执行状态停留在最后一次成功
  写入的值（步进/终态失败 → 仍 running，挂起失败 → 仍 running），自己永远不会收敛到终态。
  唯一的出口是人工跑运维巡检脚本 scripts/reap_zombie_executions.py（默认 --idle-minutes 60，
  见 ops-runbook.md#僵尸执行巡检）——它把超期 running 强制写成 error(orphaned)，即把这条
  执行当孤儿丢弃（进度全丢），而不是让它跑完；脚本不跑就一直停在 running。
• 代价二：节点副作用按重投次数重复发生。 落盘失败的那一步没有进 checkpoint，重投后从上一
  个成功 checkpoint 重跑该节点（flow_worker.py:1680-1682 自述「下轮从 checkpoint / 挂起
  节点重入」）——HTTP 调用、git push、写文件、发消息、扣费等副作用每次都真实发生；默认
  max_deliveries=5 后消息进 DLQ 不再重投，但执行状态仍是 running，既没收尸也无法续跑。
• 代价三：跨后端不一致，开发绿、生产红。 同一份 state 在 memory 后端 save=True
  （本地开发、单测、MemoryExecutionStorage），切到 Redis 集群档即 save=False 并触发上述
  链路——问题只在生产部署形态暴露。
• 不修的后果：流程上下文里一旦落进过一个非 JSON 值，这条执行的落盘就永久失败，worker
  只会把产生该值的节点反复重跑到投递上限然后丢进 DLQ；执行跑不完，也拿不到有意义的终态
  （只能被人工巡检当孤儿丢进度），副作用按重投次数重复发生，且日志里只有一行
  「保存执行状态失败」，看不出是永久问题还是 Redis 抖动。

建议

• 核心修法：在写侧把「不可重试的序列化失败」与「可重试的 Redis 瞬断」分开，与读侧
  ExecutionStateLoadError 的思路对称。
  • 最小改法：在 RedisExecutionStorage.save_execution_state
    （plaita/storage/redis.py:281-316）把 self.serialize_state(state.model_dump())
    移出 Redis 操作的 try（或在 except 中按异常类型分派）：序列化失败不再
    return False，而是抛一个独立的、非 ValueError 的专用异常（如
    StateSerializationError，语义上属存储层公共契约，位置宜在
    plaita/storage/base.py，参照 ExecutionStateLoadError 的注释）。
  • 同一处还有第二个入口：FencedExecutionStorage.save_execution_state
    （plaita/storage/fenced.py:118-144）同样把 serialize_state 与 client.eval 包在
    一个 except 里 → 同样 return False，需按同一判据一并修正。
  • FlowWorker 侧：新异常不能复用 StatePersistError 的「不 ack 重投」路径（那是为瞬态
    设计的，见 test_worker_save_guard.py 钉住的契约）；应让它在不重投的前提下转成可观测
    终态（写 error 终态并 ack），使失败在状态里可判定，而不必依赖僵尸巡检兜底。
• 可选方向：给 serialize_state 加 default= 兜底（同 flow_worker.py:1056 的
  default=str），让非 JSON 值退化为字符串而不是整条落盘失败；但这会改变读回语义，必须显式
  记录（如写入标记或 warning），不能静默发生。
• 验收条件：
  1. 新增单测（建议放 tests/unit/test_worker_save_guard.py 或
     tests/integration/test_storage_redis.py）：构造
     ExecutionState(execution_id=..., context={"$NODE": {"n1": {"result": {1, 2}}}})，断言
     RedisExecutionStorage.save_execution_state 不再返回 False——要么抛出可捕获的
     专用异常，要么（若采纳兜底方案）返回 True 且 load_execution_state 能与写入值完成
     JSON 往返。
  2. 负例不得回退：模拟 Redis 瞬断（pipeline() / execute() 抛 ConnectionError）时仍返回
     False，FlowWorker 仍走「不 ack → 重投」；tests/unit/test_worker_save_guard.py 现有
     契约必须继续通过。
  3. 端到端：context 含 datetime 的分布式执行在落盘失败后不得无限重投——断言该执行最终进入
     可观测终态（error）而不是永远停在 running，且重投次数不超过 max_deliveries。
  4. 上文第 9 条复现命令的期望结果变为：序列化失败被显式区分（抛异常或成功落盘），memory 与
     redis 两个后端对同一 state 的行为不再分叉。

边界

• 不改 worker 对瞬态落盘失败的不 ack 重投契约（flow_worker.py:854-882、:1679-1687、
  :1870-1874，以及 tests/unit/test_worker_save_guard.py 钉住的行为）。那是 2026-10-02
  评审任务①针对「Redis 瞬断被静默吞掉」的正确修复，本条只补它遗漏的「永久失败」这一支。
• 不在节点返回值入口做 JSON 可序列化前置校验。那会改变引擎对节点返回值的既有容忍度
  （现有流程可能依赖任意 Python 对象经由 memory 后端/进程内传递），属于更大的契约变更，不在
  本条范围。
• 不顺手重构 storage 层的同步/异步接口，也不动 serialize_state 自身的「失败即
  re-raise」语义（plaita/storage/base.py:142-146 的行为已被
  tests/unit/test_storage_mutations.py 钉住）；本条只改「失败之后如何上报」。
• 与 #40 同域但不同根，不是重复：#40（fix(storage): sqlalchemy 后端 list_executions 丢 tenant_id/flow_hash；redis 后端列表 O(全命名空间) 无索引）修的是 list_executions 的查询
  字段与索引选择，本条修的是 save_execution_state 的失败分类；文件/函数、修法、断言三者
  互不覆盖，任一条修好都不会让另一条消失。
• 与 #50 / #52 同属「执行无法收敛」的表象，但不是重复：#52（执行终态滞后）要修的是重试耗尽后终态化这一步没有发生，#50 要修的是重投载体被租约冲突 ack 吃掉；本条的落盘每一次都失败（序列化永久错误），所以 #52 的「补一次 flush」与 #50 的「换重投载体」都不会让落盘成功。反过来说，本单修好后 #50 / #52 的症结依然存在，三者应各自独立修复。