fix(expr): 内建 mutable 函数就地改写 live 执行上下文——循环条件之外的赋值/节点参数求值即破坏 $INPUT/$GLOBAL，并随 checkpoint 落盘

Start by running reproduction 1 from the issue to see $INPUT.items change after $F.pop is evaluated. The mutators are in plaita/core/expression.py (around lines 224-247, registered near 349-389), and the evaluation path is ExecutionContext._evaluate_string in plaita/core/context.py. Read the loop-condition deepcopy in plaita/node/loop.py lines 60-70 and its test TestLoopConditionIsolation. Done when $INPUT/$GLOBAL stay unchanged after evaluation, the flow in reproduction 2 returns [1, 2, 3], and the existing tests still pass. Choosing between copy-on-write and evaluator-level isolation is a maintainer decision.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

表达式引擎的内建「就地改写」函数（$F.pop / $F.set / $F.clear / $F.setListItem …）收到的
参数是 live 执行上下文里的对象引用，不是副本。只要这类函数出现在循环条件之外的任何求值点
（assignment 节点的 output、end / http / code / child 等节点的参数求值），它就地改写的
就是 $INPUT / $GLOBAL / $NODE 本体；改写结果随后被 to_dict() / to_checkpoint_dict()
序列化进 checkpoint，恢复后仍然是脏的。求值「看起来」是纯函数调用，实际上带副作用，且没有任何
引擎侧拦截。

1. plaita/core/expression.py:224 / :230-231 / :239-241 / :247 —— 这 9 个函数直接对入参执行
   就地操作（pop / __setitem__ / update / clear），返回值是「被弹出的元素」或 None，
   而不是新容器：

def _fn_pop(a, b): return a.pop(b)
def _fn_delListItem(a, b): return a.pop(b)
def _fn_setListItem(a, b, c): return a.__setitem__(b, c)
def _fn_set(a, b, c): return a.update({b: c})
def _fn_delete(a, b): return a.pop(b)
def _fn_clear(a): return a.clear()
def _fn_delDictValue(a, b): return a.pop(b)
def _fn_setDictValue(a, b, c): return a.update({b: c})
def _fn_clearDict(a): return a.clear()

2. plaita/core/expression.py:349-355 与 :380-389 —— 它们被打上 has_side_effects=True 注册，
   但这只是元数据：注册表里没有任何逻辑因此改变调用方式：

    _side_effect: list[tuple[str, Callable, str]] = [
        ("pop", _fn_pop, "Pop item at index (mutates list)"),
        ("delListItem", _fn_delListItem, "Delete item at index (mutates list)"),
        ("setListItem", _fn_setListItem, "Set item at index (mutates list)"),
    ]
    for name, func, desc in _side_effect:
        reg.register(name, func, FunctionCategory.ARRAY, has_side_effects=True, description=desc)

3. plaita/core/context.py:328-331 —— execution.evaluate(...) 把 self._state
   （live CheckpointState，见 plaita/core/context.py:238-240 的 context 属性直接返回 self._state）
   原样交给求值器：

    def _evaluate_string(self, value: Any) -> Any:
        return _resolve_expression_string(
            self._evaluator, self._state, self.express_prefix, self.parent, value,
        )

4. plaita/core/state.py:332-345 —— CheckpointState.__getitem__ 返回的是 live 对象引用，
   不做拷贝，所以 $INPUT.items 取到的就是上下文里那个 list 本身：

    def __getitem__(self, key: str) -> Any:
        if key == "EXPRESS_PREFIX":
            if "EXPRESS_PREFIX" in self._present:
                return self.prefix
            raise KeyError(key)
        mapping = self._key_to_field()
        if key in mapping:
            if key in self._present:
                return getattr(self, mapping[key])
            raise KeyError(key)
        extras = self._extras
        if key in extras:
            return extras[key]
        raise KeyError(key)

5. plaita/core/state.py:516-521 —— 节点结果入库同样不拷贝，$NODE.<id> 里的容器也共享同一批对象：

    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

6. 求值点遍布各节点，都会拿到 live 上下文（仅 loop.py 的条件路径例外，见第 7 条）：

$ grep -rn "execution.evaluate(" --include='*.py' plaita/node/
plaita/node/assignment.py:62:        evaluated = execution.evaluate(value)
plaita/node/child.py:51/58/97/104:  return child_execution.run_compatible(self.child_flow, lazy, execution.evaluate(self.input))
plaita/node/code.py:1010-1012:      language / code / input 三次求值
plaita/node/concurrent.py:158-159:   execution.evaluate(input_key) / execution.evaluate(pb.input)
plaita/node/end.py:65:            return execution.evaluate(self.output)
plaita/node/event_node.py:105:        resolved = execution.evaluate(self.event_type)
plaita/node/http.py:738:        return execution.evaluate(self.output)
plaita/node/loop.py:188/410/427:   collection / initial 求值
plaita/node/redis.py:32/53-54:       target / arguments / command 求值

7. plaita/node/loop.py:60-70 的注释自己承认了这个风险，但只在循环条件这一处做了隔离；
   has_side_effects 的实际用途全仓也只有这一个点和一个查询方法：

# 表达式引擎有 ``$F.set`` / ``$F.pop`` / ``$F.clear`` 等 mutate 函数（见
# core/expression.py，has_side_effects=True），条件引用它们时只能改到副本，
# 不污染原 context（契约钉在 tests/unit/test_loop.py::TestLoopConditionIsolation）。

反证「没有引擎侧拦截」—— has_side_effects 的全部非注册用法：

$ grep -rn "side_effect" --include='*.py' plaita/ | grep -v 注册/描述行
plaita/core/expression.py:139:    def side_effect_functions(self) -> List[FunctionDescriptor]:
plaita/node/loop.py:69:# core/expression.py，has_side_effects=True），条件引用它们时只能改到副本，

即：只有「一个查询方法」+「一处注释」。求值器在调用函数前不检查该标志，也不复制参数。

复现 1（实跑，/tmp/plaita-audit-venv/bin/python，Python 3.12，plaita editable 安装）——
live 上下文与 checkpoint 同时被改写：

$ cd /tmp && /tmp/plaita-audit-venv/bin/python - <<'EOF'
from plaita.core.context import ExecutionContext
from plaita.core.expression_parser import ExpressionParser
ctx = ExecutionContext()
class F:
    expose_env=None; global_context={}; flow_id='f1'
ctx.setup_flow(F(), ({},), {})
ctx.set_state('$INPUT', {'items':[1,2,3]})
p = ExpressionParser.for_prefix('$')
print('before:', ctx.get_state('$INPUT'))
print('pop result:', p.evaluate('{% $F.pop($INPUT.items, 0) %}', ctx.context))
print('after live:', ctx.get_state('$INPUT'))
print('checkpoint $INPUT:', ctx.to_dict()['$INPUT'])
EOF
--- 输出 ---
before: {'items': [1, 2, 3]}
pop result: 1
after live: {'items': [2, 3]}
checkpoint $INPUT: {'items': [2, 3]}

复现 1 补充（同一解释器）—— setListItem / clearDict 返回 None 但元素/字典已被改掉，
调用方拿不到任何「发生了写入」的信号：

setListItem returns: None
after setListItem live $INPUT: {'items': [99, 2, 3]}
clearDict returns: None
after clearDict $GLOBAL: {'cfg': {}}

复现 2（实跑）—— 端到端流程：assignment 节点的 output 只要求值，就把 $GLOBAL 吃掉了，
整条 flow 的返回值也随之变化：

$ cd /tmp && /tmp/plaita-audit-venv/bin/python - <<'EOF'
from plaita.core.executor import FlowExecution
from plaita.core.flow import Flow
flow = Flow.model_validate({'flow_id':'mutprobe2','name':'p','global_context':{'items':[1,2,3]},'nodes':[
  {'id':'s','type':'start','next':'a1'},
  {'id':'a1','type':'assignment','output':'{% $F.pop($GLOBAL.items, 0) %}','next':'end'},
  {'id':'end','type':'end','output':'$GLOBAL.items'}]})
print('flow result:', FlowExecution().run(flow, {}), ' (纯函数语义应为 [1,2,3])')
EOF
--- 输出 ---
flow result: [2, 3]  (纯函数语义应为 [1,2,3])

复现 3（实跑）—— checkpoint 落盘确认：用 lazy 逐节点输出（strategies.py 里就是
context.to_dict() 的产物）观察每个节点结束时的 checkpoint 快照，assignment 求值之后
落盘的 $GLOBAL 已经是 [2, 3]：

$ cd /tmp && /tmp/plaita-audit-venv/bin/python - <<'EOF'
from plaita.core.executor import FlowExecution
from plaita.core.flow import Flow
flow = Flow.model_validate({'flow_id':'mutprobe7','name':'p','global_context':{'items':[1,2,3]},'nodes':[
  {'id':'s','type':'start','next':'a1'},
  {'id':'a1','type':'assignment','output':'{% $F.pop($GLOBAL.items, 0) %}','next':'end'},
  {'id':'end','type':'end','output':'$GLOBAL.items'}]})
fe = FlowExecution()
for out in fe.run_compatible(flow, True, {}):
    d = out if isinstance(out, dict) else getattr(out, '__dict__', {})
    ctx = d.get('context') if isinstance(d, dict) else None
    if ctx: print(ctx.get('$GLOBAL'))
EOF
--- 输出 ---
{'items': [1, 2, 3], 'flow_id': 'mutprobe7'}
{'items': [2, 3], 'flow_id': 'mutprobe7'}      # assignment 求值之后
{'items': [2, 3], 'flow_id': 'mutprobe7'}

反证「只有循环条件被保护」：现有测试只钉了循环条件这一处的隔离契约，没有任何测试断言
「求值后原容器不变」：

$ grep -rn "F.set\|F.pop\|F.clear\|setListItem" tests/
tests/unit/test_expression.py      # 只断言 has_side_effects 元数据
tests/unit/test_io.py
tests/unit/test_loop.py:133-140    # TestLoopConditionIsolation
tests/unit/test_mfix_l_loop_context.py:198-225  # 循环条件隔离（$F.set 不得污染 $GLOBAL）

影响

• 谁踩到：任何写了 mutate 函数但没意识到它就地改写上下文的流程作者，以及用 @flow / IR
  生成的流程。$GLOBAL 是流程级共享状态、$INPUT 是入口数据，表达式里出现
  {% $F.pop($GLOBAL.items, 0) %} 这种「取第一个元素」的直觉写法就会命中——从函数名与返回值
  看，它与纯函数的 $F.getListItem / $F.slice 长得几乎一样，没有任何报错或告警提示副作用。
• 触发点不止循环条件：assignment（output）、end / http（output）、code
  （language / code / input）、child（input）、concurrent、redis、event_node 的
  参数求值都会拿到 live 上下文。循环条件有 deepcopy 隔离（loop.py:60-70），其他点全部裸奔。
• 代价，按严重度排序：
  1. 数据错/丢失：$INPUT / $GLOBAL 被静默改写。同一个容器被多个节点续用时会累积漂移；
     同一个 $GLOBAL.items 在第二次求值 $F.pop 时弹出的是「已经少了一个元素」的列表。
  2. 随 checkpoint 落盘并跨恢复存活：改写后的容器被 to_dict() 序列化（复现 1 与复现 3），
     执行中断 / 挂起 / 恢复 / worker 重启后脏数据仍在，重试不会自愈。
  3. 不可观测：setListItem / clearDict / set 返回 None；pop 返回的是被弹出的元素
     （看起来像一次正常的取值），作者拿不到「上下文被改了」的任何信号。
  4. 并发写风险：plaita/core/expression.py:358-367 的 _register_dict docstring 自己写明这类
     函数「not thread-safe —— 并行节点执行 / 跨 async 任务共享上下文时可能损坏数据」，
     而当前引擎并未替调用者加锁或复制（docstring 把责任推给调用者，但调用者是求值器本身）。
• 生产可达：可达，无需特殊部署档位。任何在非循环条件位置写 mutate 表达式的流程即可命中
  （复现 2 是 3 个节点、一条 assignment 的最小流程）。
• 不修的后果：表达式求值语义「看起来纯、实际带写副作用」这一条会长期作为静默数据损坏源存在；
  checkpoint 会把脏状态固化下来，恢复/重试都无法回到正确状态。

建议

• 修复方向（具体到文件/函数）：
  • 方案 A（推荐，语义最干净）：在 plaita/core/expression.py 把 9 个 mutate 函数改成
    copy-on-write 语义——先复制入参再操作并返回新对象：
    _fn_pop(a, b) → 复制 a、pop 后返回「新容器」或按现有契约返回被弹元素；
    _fn_setListItem / _fn_set / _fn_setDictValue 同样先 copy 再写。这样与
    _fn_append / _fn_extend / _fn_insert（已经返回新列表，见 expression.py:221-223）一致，
    也消除了 docstring 里的线程安全告警前提。
  • 方案 B（最小改动，保留原地语义）：在求值侧统一隔离——ExecutionContext._evaluate_string
    （plaita/core/context.py:328-331）在把 self._state 交给求值器前，对表达式静态引用到的根键
    做与 plaita/node/loop.py:139-158 相同的「根键 deepcopy 替换」，把 loop 的局部补丁提升为
    引擎级契约；loop 自己的隔离逻辑随之可以删除，避免两处语义分叉。
  • 无论选哪个方案，has_side_effects=True 都应当从「元数据」变成「被求值器消费的契约」：
    求值器在调用带该标志的函数时必须保证参数是副本，而不是只提供一个 side_effect_functions()
    查询方法（expression.py:139-141）供人自查。
• 最小修法（一处）：在 _evaluate_string 里对 mutate 类调用做一次入参浅/深拷贝即可掐掉
  「assignment 节点 output 吃掉 $GLOBAL」这条最短路径（复现 2）；copy 的粒度按容器层级
  选择即可，不必引入通用快照框架。
• 验收条件（可执行、可断言）：
  • 新增单测（可放在 tests/unit/test_loop.py 的 TestLoopConditionIsolation 附近，或新建
    TestMutatorContextIsolation）：对 $INPUT.items 求值 {% $F.pop($INPUT.items, 0) %}
    之后，断言 execution.context 与 to_dict() 里的 $INPUT.items 仍是 [1, 2, 3]；
    对 $F.clearDict($GLOBAL.cfg) / $F.setListItem($INPUT.items, 0, 99) 做同样的不变式断言。
  • 端到端回归：复现 2 的 flow 返回值必须变回 [1, 2, 3]（当前是 [2, 3]）；
    复现 3 中每一步落盘的 $GLOBAL 必须恒为 {'items': [1, 2, 3]}。
  • 反向用例不得被误伤：$F.pop 作为「取值」的返回值语义（复现 1 里返回 1）、
    tests/unit/test_mfix_l_loop_context.py:198-225 的循环条件隔离契约、
    以及 tests/unit/test_expression.py 里关于 has_side_effects 元数据的断言都必须继续通过。

边界

• 本单只处理内建 mutable 函数就地改写 live 执行上下文这一条根因（对象引用共享 + 求值器不隔离）。
  不改表达式解析/编译架构，不引入通用配额或快照系统，不改 $F.pop 等函数「返回被弹出元素」的
  返回值契约（方案 A 若改变返回值需在本单内明确并同步测试），也不顺手重构 loop.py 的热路径优化。
• 与相邻 issue 的差异（查重结论）：gh issue list -R jeffkit/plaita --state all --limit 100
  共 33 条，逐条比对后无同根条目。#23（多 worker 下 start 消息被 XCLAIM 重派导致整条 flow
  双跑）与 #36（flow_hash 漂移把在途 run 批量终态化）都在 worker/队列调度层，最终表现为
  「重复执行」而非「容器被就地改写」；#30 / #41 在子进程 env/资源限制层；
  本单的写入发生在 Python 求值器内部、对同一进程里的 live 上下文对象生效，修复面不重叠，故非重复。
• 不顺手扩大范围：循环条件已有的隔离补丁（loop.py）当前工作正常，本单不要求重写它；
  若采用方案 B，把该逻辑上提是消除重复语义的必要步骤，但不构成对 loop.py 其他行为的改动要求。