fix(core): 状态快照无环/深度保护 + 节点结果按引用入库——$NODE（或任何整表根）自引用时 generator/distributed 直接 RecursionError（normal 反而成功），且 $NODE['a'] is $NODE 暴露写入别名

Start with _snapshot_value in plaita/core/state.py (around line 46) and its caller to_checkpoint_dict (around line 422). Reproduce with a flow s -> a(output="$NODE") -> e in normal, generator and distributed modes, then look at update_node_result (around line 516), which writes into the live dict. Done when all three modes finish without RecursionError and the aliasing checks in tests/unit/test_clean_c1_state.py pass. The cycle policy (reuse the object or raise an explicit error) needs a maintainer decision first.

优先级 P2 · 依赖：无

基线 HEAD = 65f48f2

Summary

把整个 $NODE 表作为 assignment 的输出（output: "$NODE"）时，状态表被写回自身、变成自引用容器；快照出口
_snapshot_value 是无 memo、无环检测、无深度上限的朴素递归，于是 generator / distributed
两个模式直接 RecursionError，而 normal 反而「成功」。
（通用形态是「任何被回指的容器」，$NODE 是实测触发路径。）

证据 1（写入别名）：update_node_result 把节点结果按引用塞进 $NODE

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

self.get(key, {}) 取到的是 $NODE 当前live dict，result 也原样存入，无拷贝、无类型/值域判断。
调用方：plaita/core/runner.py:240

        self.context.update_node_result(node, result)

（plaita/core/context.py:341-342 只是转发 self._state.update_node_result(node.id, result)。）

证据 2（根查找返回 live 对象，故 output: "$NODE" 回指状态表本身）

plaita/core/expression_parser.py:383-387

        def thunk(frame: Frame) -> Any:
            context, registry, prefix = frame
            # Root lookup: KeyError preserved for missing keys (matches old engine)
            try:
                obj = context[root_key]

此处的 context 是 CheckpointState（context.py:330 把 self._state 传进表达式引擎），它的
__getitem__（state.py:332，$NODE 走 _key_to_field() 命中的字段，返回 getattr 的存储对象）
不做拷贝，所以 output: "$NODE" 取到的就是 $NODE 的 live dict。
plaita/node/assignment.py:43-45 的校验只断言非空，第二个断言恒真，不拦值域：

    def validate(self):
        assert self.output_type is not None, "assignment node required outputType."
        assert self.output is not None or len(self.upstream_output) >= 0, "Assignment Node required output."

文档 docs-site/docs/guide/expressions.md:29 只把 $NODE.<node> 形态列为合法引用，未定义「整个 $NODE 作为节点输出」的语义：

| `$NODE.assign` / `$NODE.assign.field` | 节点 `assign` 的输出 / 其字段 | `None` |

证据 3（快照出口无环/深度保护）

plaita/core/state.py:46-66

def _snapshot_value(value: Any) -> Any:
    """递归物化 dict/list/tuple 容器为私有快照（C1-2，2026-10 评审修复包 C1）。

    历史上 checkpoint/context 出口（``to_checkpoint_dict``、``$PARENT`` 快照）
    是浅拷贝——``$NODE`` 等嵌套容器按引用共享，消费方（分布式宿主异步序列化、
    子流程 ``$PARENT``）拿到的是 live 别名：写穿透回运行态、后续执行让已"落盘"
    的数据漂移。快照只在容器边界重建；JSON 不可表达的非容器对象（与历史行为
    一致）仍按引用共享——它们本就无法进入 checkpoint 序列化。
    """
    if isinstance(value, dict):
        return {k: _snapshot_value(v) for k, v in value.items()}
    if isinstance(value, _LazyRootSnapshot):
        return value.materialize()
    if isinstance(value, list):
        return [_snapshot_value(v) for v in value]
    if isinstance(value, tuple):
        return tuple(_snapshot_value(v) for v in value)
    if isinstance(value, (set, frozenset)):
        # 集合元素必可哈希（不可变），浅重建即快照
        return value.copy()
    return value

出口链路：plaita/core/context.py:367-372

    def to_dict(self) -> Dict[str, Any]:
        from plaita.core.state import validate_checkpoint
        data = self._state.to_checkpoint_dict()
        for warning in validate_checkpoint(data, self.express_prefix):
            logger.warning(warning)
        return data

→ plaita/core/state.py:422

        return {k: _snapshot_value(self[k]) for k in self}

每个步进结果都过这个出口：plaita/core/strategies.py:190-193（generator 每步）

            yield _create_lazy_output(
                current, result, branch, context.to_dict(),
                is_end=reached_end, execution_id=context.execution_id,
            )

plaita/core/strategies.py:347（distributed 每步）

        return _create_lazy_output(current_node, result, branch, context.to_dict(), execution_id=context.execution_id)

plaita/core/strategies.py:395（resume 步）

        return _create_lazy_output(current_node, result, None, context.to_dict(), is_suspend=False, execution_id=context.execution_id)

复现（已实跑）

流程：s(start) → a(assignment, output="$NODE") → e(end, output="$NODE.a")。

$ /tmp/plaita-audit-venv/bin/python /tmp/repro_pcf_f5.py
=== normal ===
normal -> OK: {'s': None, 'a': {...}, 'e': {...}}
   $NODE['a'] is $NODE ? True
=== generator ===
generator -> RecursionError maximum recursion depth exceeded
=== distributed ===
distributed -> FlowErrorException maximum recursion depth exceeded
   cause: RecursionError maximum recursion depth exceeded
=== deep nesting (no cycle, just depth) via to_dict ===
deep 2500 -> RecursionError maximum recursion depth exceeded

distributed 的真实调用链（实跑 traceback 去重后的关键帧；末两行为归一化再抛）：

plaita/core/executor.py:460 in run_distributed
plaita/core/strategies.py:273 in execute
plaita/core/strategies.py:347 in _execute_current_node
plaita/core/context.py:369 in to_dict
plaita/core/state.py:422 in to_checkpoint_dict
plaita/core/state.py:56 in _snapshot_value
plaita/core/state.py:57 in _snapshot_value   → RecursionError
plaita/core/executor.py:462 in run_distributed
plaita/core/_error_normalization.py:69 in raise_distributed_error  → FlowErrorException

generator 的 traceback 则是 _snapshot_value 在 state.py:56 上递归重复直至触顶（setrecursionlimit(300) 下显示为 [Previous line repeated 285 more times]）。

别名在后续步进中真实写穿（已实跑，同一 $NODE 被两步回指）

流程 s → a(output="$NODE") → b(output="$NODE.a") → e，normal 模式取 $NODE：

keys: ['a', 'b', 'e', 's']
$NODE['a'] is $NODE      -> True
$NODE['b'] is $NODE['a'] -> True
'b' in $NODE['a']        -> True

即 a 步记录的「节点结果」在 b 步之后凭空多了 b 键——已记录的节点结果是被引擎继续原地改写的 live 容器。

反证：既有测试完全没有覆盖

$ grep -rn "RecursionError" --include="*.py" tests/
（零命中，exit=1）

$ grep -rln "to_checkpoint" --include="*.py" tests/
tests/unit/test_clean_c1_state.py
tests/unit/test_state.py
tests/unit/test_state_mutations.py

test_clean_c1_state.py 只覆盖写入隔离（test_to_checkpoint_dict_no_write_through、test_snapshot_covers_list_tuple_dict_shapes 等），无自引用、无深嵌套用例。

影响

• 流程作者：在 assignment 的 output 里写整表根（实测 output: "$NODE"；任何把容器写回其自身可达位置的 output 同型）即可触发。本地 normal 跑得通还返回「成功」，$NODE['a'] is $NODE 为 True 时作者也看不出异常；一旦进 generator（FlowExecution.run(..., mode="generator") / lazy=True，公共 API）或上集群走 worker，立刻崩。
• 集群部署（生产可达）：plaita/server/flow_worker.py:1384 的 execution.run_distributed(...) 是 worker 的生产执行路径，每步都调 context.to_dict()。自引用状态下报错被 _error_normalization.raise_distributed_error 归一化成 FlowErrorException("maximum recursion depth exceeded")，节点代码、节点 id 一概不可见——「本地绿、上线红」且错误信息误导定位方向，一条 run 白跑。
• 深度不设限的第二类触发：即使没有环，to_checkpoint_dict() 对深嵌套值（实测 2500 层）同样 RecursionError。大 payload 摘要树、递归结构体一类流程数据在 checkpoint 出口会撞同一处，且撞在「落盘/序列化」这一跳，现场只剩一个裸 RecursionError。
• 别名写穿（与环独立的第二半）：update_node_result 按引用入库，使任何在 update_node_result 之前拿到 $NODE 或其某键容器的消费方（callback、宿主侧序列化前的读、节点内部缓存）观察到的是持续被后续步骤改写的容器；已「记录」的历史节点结果不再稳定。
• 不修的后果：一份在本地验证通过的合法 flow 会上线即崩，崩因被归一化掩盖；同时节点结果表的引用语义没有任何契约，下游无法安全缓存任何 $NODE 值。

建议

1. plaita/core/state.py:46 的 _snapshot_value 加环与深度保护：用 id() memo 集合检测已访问容器，命中环时按既定语义降级（复用已构建对象，或抛出带容器路径的明确异常）而不是递归到 RecursionError；同时加节点计数/深度上限，超限抛可捕获的显式异常（消息里带上路径前缀），避免裸递归错误穿透到归一化层。若要保住热路径成本，可改写为显式栈的迭代物化。
2. plaita/core/state.py:516-521 的 update_node_result：不要原地改写 self.get(key, {}) 返回的 live dict，改为取出后 copy 再写回；是否对 result 做快照按第 1 点的语义一并决定并在 docstring 写明「入库是否按值」。
3. 入口收敛（可选但便宜）：plaita/node/assignment.py:43-45 的 validate 或解析期对「output 指向不带字段段的整表根」给出明确报错或 warning——$NODE.<node> 才是文档承认的形态。
4. 验收条件：
   • 新增单测（建议放 tests/unit/test_clean_c1_state.py）构造 s → a(output="$NODE") → e 流程，断言 normal、generator、distributed 三模式均不抛 RecursionError；distributed 不产生 FlowErrorException("maximum recursion depth exceeded")。
   • 新增断言：to_checkpoint_dict() 对 2500 层嵌套值不再抛 RecursionError（或抛本仓自定义异常且消息含深度/路径信息）。
   • 别名回归：对 s → a(output="$NODE") → b(output="$NODE.a") → e 断言 $NODE['b'] is $NODE['a'] 为 False、'b' in $NODE['a'] 为 False（若判定按值入库不可行，则至少断言引擎给出明确拒绝/告警）。
   • 复现脚本 repro_pcf_f5.py 应当从「1 绿 2 红 + deep 红」变为四行全 OK。

边界

• 不重做 C1-2 快照设计：$PARENT 的 _LazyRootSnapshot 惰性物化是为热路径性能留的折中（见 state.py:69-88 注释），本单只补环/深度保护，不改惰性策略。
• 只处理「快照出口的环与深度」+「update_node_result 的按引用入库」这一处；不动 validate_checkpoint 的告警逻辑，也不改表达式引擎「根查找返回 live 对象」的既有语义（那是独立的只读包装议题）。
• 与 #16（map 子流程 end 递归限制 + if 块作用域）不同根：那是图/子流程调度的递归深度限制，本单是 Python 值快照的递归，二者触发点与修复位置均不同。
• 不覆盖 plaita-nodes；不改 normal 模式「允许整表根作为输出」的历史行为，除非第 3 点落地为硬校验。