fix(nodes): RedisNode 把连接缓存在节点实例上——循环内 target 重求值被静默丢弃（写错实例），且缓存的 client 使同一 flow 第二次 start/resume 的 flow_hash 序列化直接抛错

Start in plaita/node/redis.py: the redis_client field is cached on the node and execute() only builds the connection once, so later target values are ignored. Then check _compute_flow_hash in plaita/server/flow_worker.py, which fails when the cached client is present. Done when a test in tests/unit/test_reviewfix_c_redis.py runs execute() twice with two different redis:// URLs and gets two constructions with the right host/db, and a second flow hash computes without error.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

RedisNode 把 redis.Redis 连接缓存在节点实例的 pydantic 字段 redis_client 上（redis.py:26），而 target 在每次 execute() 里都重新求值（redis.py:32）。两者合在一起产生两个后果：

1. 静默写错实例：连接只在首次 execute() 建一次（redis.py:44），已建连时该次 target 解析出的 hostname / port / db / username / password 五个值全部被丢弃，后续调用一律复用首次的连接。
2. 同一 Flow 对象第二次执行硬失败：缓存的 client 是 pydantic 模型上路由不到 JSON 的任意类型，flow.model_dump(mode="json") 直接抛 PydanticSerializationError；FlowWorker 的 start（flow_worker.py:1320）与 resume（flow_worker.py:1505）都在执行节点之前调 _compute_flow_hash(flow)（flow_worker.py:1052-1057），于是第二次 start_flow / resume 抛 RuntimeError。

1. 节点字段与缓存分支

plaita/node/redis.py:26
    redis_client: Optional[redis.Redis] = None

plaita/node/redis.py:31-44
    def execute(self, execution):
        redis_url = urllib.parse.urlparse(execution.evaluate(self.target))
        username = redis_url.username
        password = redis_url.password
        hostname = redis_url.hostname
        port = redis_url.port
        db = 0
        if len(redis_url.path) > 1:
            db = redis_url.path[1:]
        assert username is not None, "redis node target must have username"
        assert password is not None, "redis node target must have password"
        assert hostname is not None, "redis node target must have hostname"
        assert port is not None, "redis node target must have port"
        if not self.redis_client:

plaita/node/redis.py:49-54
            self.redis_client = redis.Redis(
                host=hostname, port=port, db=db, username=username, password=password,
                socket_connect_timeout=5, socket_timeout=30, retry_on_timeout=True,
            )
        argument_list = re.split(r"\s+", execution.evaluate(self.arguments).strip())
        return self.redis_client.execute_command(execution.evaluate(self.command), *argument_list)

validate() 是空实现，没有任何守卫（redis.py:28-29）：

plaita/node/redis.py:28-29
    def validate(self):
        pass

2. 为什么同一个节点实例会被跨轮 / 跨次复用

LoopNode.execute 每轮把同一个 self.child_flow 交给新子执行体，get_child_execution 只新建 FlowExecution、不复制 flow：

plaita/node/loop.py:211-213
        for item in collection:
            item_execution = execution.get_child_execution()
            last_result = item_execution.run_compatible(self.child_flow, False, item=item, index=index)

plaita/core/executor.py:299-304
    def get_child_execution(self):
        # 用 child() 不传额外 handler: 子级通过 inherit_handlers 继承父级各一份,
        # 避免把父级 handler 当"新增 handler"再传一次导致双发。
        child = FlowExecution(self, callback_manager=self.callback_manager.child())
        child.mode = self.mode
        return child

plaita/core/ 内没有任何对 Flow / 节点的复制（反证：grep -rn --include='*.py' "model_copy" plaita/core/ plaita/node/ → 零命中；deepcopy 命中只有 plaita/core/context.py:300（flow.global_context）与 plaita/core/state.py:129-130、plaita/node/loop.py:155（上下文），均不复制 flow 或节点）。

FlowWorker 的流程定义 TTLCache 命中即返回同一个 Flow 对象：

plaita/server/flow_worker.py:1008-1010
        if cache_key in self.flow_definition_cache:
            logger.info("从缓存获取流程定义: %s", cache_key)
            return self.flow_definition_cache[cache_key]

缓存默认 cache_size=100 / cache_ttl=300（flow_worker.py:504-505），且既有测试直接断言两次取到同一对象（tests/unit/test_flow_worker_scenarios.py:130）：

        assert flow1 is flow2  # 应该是同一个对象（来自缓存）

3. 根因：节点上的 client 不可 JSON 序列化

plaita/server/flow_worker.py:1052-1057
        payload = json.dumps(
            flow.model_dump(mode="json"),
            sort_keys=True,
            ensure_ascii=False,
            default=str,
        )

异常发生在 model_dump(mode="json") 内部，所以 default=str 救不了。两处调用点都在节点执行之前：

plaita/server/flow_worker.py:1320
                flow_hash=self._compute_flow_hash(flow),

plaita/server/flow_worker.py:1505
        current_flow_hash = self._compute_flow_hash(flow)

4. 复现

环境：/tmp/plaita-audit-venv/bin/python（Python 3.12，editable 安装 plaita[dev,all] + plaita-nodes）。

(a) loop 遍历两个不同 Redis URL —— 第二次写入被静默丢弃

$ /tmp/plaita-audit-venv/bin/python /tmp/pnb-repro/t5b_redis_loop.py
flow result: wrote to h1:0
Redis constructed: 1 [{'host': 'h1', 'port': 6379, 'db': '0', 'username': 'u1', 'password': 'p1', 'socket_connect_timeout': 5, 'socket_timeout': 30, 'retry_on_timeout': True}]

子流程中 target="$INPUT.item.url"，两次迭代分别是 redis://u1:p1@h1:6379/0 与 redis://u2:p2@h2:6379/1；输出只有 h1:0，Redis 只构造 1 次。

(b) 同一实例连续 execute() 两次

$ /tmp/plaita-audit-venv/bin/python /tmp/pnb-repro/t5_redis.py
result1: h1:0 result2: h1:0
Redis constructed 1 times: [{'host': 'h1', 'port': 6379, 'db': '0', 'username': 'u1', 'password': 'p1', 'socket_connect_timeout': 5, 'socket_timeout': 30, 'retry_on_timeout': True}]
node.redis_client is same object: <__main__.FakeRedis object at 0x10da88b00>

(c) 引擎路径：同一 Flow 对象 run() 两次

$ /tmp/plaita-audit-venv/bin/python /tmp/pnb-repro/t7_engine_twice.py
run1: wrote h1:0
run2: wrote h1:0
constructed: 1 [('h1', '0')]

(d) worker 跨次执行 —— 硬失败

$ /tmp/plaita-audit-venv/bin/python /tmp/pnb-repro/t6_worker_double.py
run1: {'id': 'end', 'type': 'end', 'name': None, 'result': 'ok', 'branch': '', 'context': {...}, 'is_end': True, 'is_suspend': False, 'execution_id': 'bda35019856c46bf86298e57b2cfbc97'}
run2 raised: RuntimeError 执行流程出错: Unable to serialize unknown type: <class '__main__.FakeRedis'>

第 1 次成功，第 2 次（同一 flow_id/version，命中 300s 定义缓存）抛 RuntimeError。

(e) 根因最小复现（真实 redis.Redis）

$ /tmp/plaita-audit-venv/bin/python /tmp/pnb-repro/t5e_hashfn.py
1st hash: 62902ab22875a305
2nd hash raised: PydanticSerializationError Unable to serialize unknown type: <class 'redis.client.Redis'>

5. 触发前提

RedisNode 不在默认注册表内：_BUILTIN_NODES（plaita/node/init.py:37-62）不含它，pyproject.toml:94-99 的 plaita.nodes entry points 只有 delay / redis_queue / kafka_queue / http_callback / approval 及 plaita-nodes 的节点（实测 get_default_registry().list_types() 45 项，'redis' in types 为 False）。文档也明说需自行注册：

docs-site/docs/nodes/builtin.md:47
    - **`calculate`** / **`redis`**（需 `redis` extra）：存在于源码但不在 `_BUILTIN_NODES`，自行 `register`。

即部署方显式 get_default_registry().register(RedisNode) 后上述路径才可达。

影响

• 生产（worker / console 发布的 JSON 流程）：部署方显式注册了 redis 节点、流程里有 redis 节点、且在 cache_ttl（默认 300s）内第二次触发同一 flow_id+version 时，第二次 start_flow 在落 running 行之前就抛 RuntimeError: ... Unable to serialize unknown type: <class 'redis.client.Redis'>。消息不会 ack，重投后仍失败——同一流程变成「第一次能跑、之后一直跑不起来」的死局。resume 侧同理（flow_worker.py:1505）：挂起的执行续跑时直接进入错误路径，而不是按设计做 flow_hash 一致性校验。
• 引擎 / loop 子流程（库嵌入与嵌入式执行）：在单次 run() 内，只要 redis 节点的 target 是表达式（$INPUT.* / $NODE.*），循环第二轮起 target 重求值结果被丢弃，读写仍然打到第一轮的主机/端口/db/账号。多环境或多租户批量循环（例如遍历一组不同实例的 URL）会把后续所有写入投到第一个实例上——这是跨实例写错，且返回结果看起来完全正常，无异常、无日志、无指标。
• 同一 Flow 对象被复用执行两次（flow.run() 两次、或任何缓存 Flow 的宿主）也走「静默写错实例」分支（t7）。
• 不修的后果：一类流程第一次执行成功、之后永久失败且原因不可见（表现为序列化错误，而非指向节点缓存），另一类流程静默把数据写到错误的 Redis 实例。

建议

1. 不要把连接缓存在节点模型上。最小修法是删掉 redis_client 字段，改为每次 execute() 局部建连（或存到 execution 级 / 进程级按解析后的 URL 作 key 的连接池），并复用同一份超时参数（socket_connect_timeout=5 / socket_timeout=30 / retry_on_timeout=True）。
2. 若要保留缓存，必须以解析后的连接四元组 (host, port, db, username) 为 key缓存，并在 target 求值结果变化时重建；同时把它移出 pydantic 模型字段（例如放在 PrivateAttr / 模块级 WeakKeyDictionary），避免进入 model_dump。
3. 验收条件：
   • 新增单测：同一个 RedisNode 实例连续 execute() 两次，target 分别为 redis://u1:p1@h1:6379/0 与 redis://u2:p2@h2:6379/1，断言 redis.Redis 被构造 2 次、第二次收到的 host/port/db 为 h2/6379/1，且返回值为 h2:1（现有 tests/unit/test_reviewfix_c_redis.py:124-150 只覆盖单次构造的超时参数，不覆盖复用分支，需补）。
   • 新增端到端测试：loop 子流程遍历两个不同 redis URL，断言两次写入分别打到 h1 与 h2（复现脚本 t5b 应从 Redis constructed: 1 变为 2，结果从 h1:0 变为第二次的 h2:1）。
   • 新增回归测试：注册 RedisNode 的 flow 连续 start_flow 两次（或对同一 Flow 对象连续 _compute_flow_hash 两次），断言不再抛 PydanticSerializationError / RuntimeError（复现脚本 t5e 应从「2nd hash raised」变为正常返回 hash，t6 的 run2 应成功）。

边界

• 本条只处理 RedisNode 在节点实例上缓存连接这一个根因，以及它引出的两个后果（写错实例 / flow_hash 序列化抛错）。不包含：redis 节点是否应进入 _BUILTIN_NODES 的注册策略讨论、FlowWorker 定义缓存是否应返回副本的整体设计、以及其它节点是否存在同类实例级缓存。
• 与 #36 相邻但不同根：#36 讲的是 flow_hash 基于解析后 model_dump 导致跨 pydantic/plaita 版本 hash 漂移（同一份定义在不同版本算出不同值）；本条是同一版本下 model_dump(mode='json') 因为节点实例上挂了不可序列化对象而直接抛异常，属序列化失败而非指纹漂移。修复 #36（例如改用规范化 IR 计算指纹）确实会顺带绕开本条在 worker 路径的表现，但不会修掉 loop 路径的写错实例，故不作为重复单。
• 本条的 loop / 引擎路径在部署方显式 register(RedisNode) 后即可达；不注册时不可达，此时不表现为缺陷。worker 路径需额外满足「300s 定义缓存窗口内第二次触发同一 flow」这一前置条件。