fix(core): $PARENT 惰性快照对 fire-and-forget 分支不是快照——分支首读晚于父推进时读到父流程未来态（非确定性），父侧原地写 $NODE 还会让分支 RuntimeError 崩掉且只记 warning

Start with _LazyRootSnapshot and update_node_result in plaita/core/state.py, then _execute_background_branches in plaita/node/concurrent.py, where background branches are submitted. Run tests/unit/test_clean_c1_state.py and tests/unit/test_concurrent.py to confirm the 20 passing baseline. Done means a background branch reading $PARENT.$NODE after the parent writes a later node does not see that node, and a concurrent snapshot-vs-write stress test raises no RuntimeError.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

$PARENT 的"快照"是惰性的：setup_flow 只把父 ExecutionContext 的 live CheckpointState 包进 _LazyRootSnapshot，真正的物化推迟到子侧首次读取。这个设计成立的前提（父在子流程存续期内被引擎阻塞）只对 InlineFlow 同步子流程成立；对 Parallel 的 fire-and-forget 后台分支不成立——父流程在分支提交后立刻继续推进并写 $NODE，而后台分支稍后的首读会把这个"未来态"读进"快照"。父侧 update_node_result 又是对同一 dict 的原地 mutate，与分支线程的 _snapshot_value 遍历并发时会直接抛 RuntimeError。

plaita/core/context.py:302（自称快照）：

        # $PARENT is a snapshot (checkpoint-safe; not a live state)。

plaita/core/context.py:313-315（实际传入的是 live 源的惰性包装，不是快照值）：

        self._state.setup_flow(
            input_value=_coerce_input_value(args, kwargs),
            parent_context=_LazyRootSnapshot(self.parent.context) if self.parent else {},

plaita/core/context.py:238-240（parent.context 就是父执行的可变状态本体）：

    @property
    def context(self) -> CheckpointState:
        return self._state

plaita/core/state.py:69 与 :96-103（"快照"= 首读时才从 live 源物化）：

class _LazyRootSnapshot(Mapping):

    def __getitem__(self, key: Any) -> Any:
        try:
            return self._cache[key]
        except KeyError:
            pass
        value = _snapshot_value(self._source[key])  # 源缺键 → KeyError 原样传播
        self._cache[key] = value
        return value

plaita/core/state.py:79-83（冻结语义的前置条件被逐字限定为"父被阻塞"，恰是后台分支不满足的条件）：

    冻结语义为何成立：子流程经 ``InlineFlow`` 同步执行，父执行在其存续期
    内被引擎阻塞（非重入守卫），父状态不会变化——「首读时物化」与「setup
    时全量快照」等价。跨进程边界（checkpoint 落盘/恢复）走
    ``to_checkpoint_dict`` / ``from_checkpoint_dict``，经 ``_snapshot_value``
    完全物化为 plain dict，惰性对象不出进程。

plaita/core/state.py:55-56（物化就是遍历 live 容器，无锁、无 copy-on-write）：

    if isinstance(value, dict):
        return {k: _snapshot_value(v) for k, v in value.items()}

plaita/core/state.py:516-521（父侧原地 mutate 同一个 dict 对象）：

    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

plaita/core/runner.py:240（父流程每执行完一个节点就写一次，因此"未来态"是常态而非边界情况）：

        self.context.update_node_result(node, result)

后台分支确实不等父流程：plaita/node/concurrent.py:110 的缺省是空列表，:255-256 把不在 joinBranches 里的分支一律归为后台，:220 提交后立即返回 join 结果：

    join_branches: List[str] = Field(default_factory=list)

        join_branches = [b for b in branches if b.name in self.join_branches]
        background_branches = [b for b in branches if b.name not in self.join_branches]
        return join_branches, background_branches

        self._execute_background_branches(background_branches, executor, execution)
        return self._execute_join_branches(join_branches, executor, execution)

plaita/node/concurrent.py:278-283（submit 后不取结果）与 :306-311（分支崩了只记 warning + errors 列表，不进流程返回值）：

        for branch in background_branches:
            fut = executor.submit(self.exec_branch, branch, execution)
            state["futures"].append(fut)
            fut.add_done_callback(
                self._make_background_done_callback(branch, errors, state)
            )

            logger.warning(
                "parallel background branch %r raised: %s", branch.name, exc,
                exc_info=True,
            )
            with _BG_LOCK:
                errors.append({"branch": branch.name, "error": str(exc)})

唯一可能挡住该路径的守卫在 plaita/node/concurrent.py:214-218，它只在 Parallel 本身已跑在 plaita 共享池 worker 线程上时才降级为同步内联执行；顶层 Parallel（FlowExecution 主循环线程）不满足该条件：

        if mode == THREAD and in_plaita_pool_thread() and (join_branches or background_branches):
            # 本节点已跑在共享线程池的 worker 上: 再向同一池提交并阻塞 join,
            # 分支数 >= max_workers 时所有 worker 互相等待, 永久死锁。降级为
            # 串行内联执行 (正确性优先, 仅嵌套场景放弃并发)。
            return self._execute_nested_inline(join_branches, background_branches, execution)

文档把"其余分支 fire-and-forget"当作正常用法对外承诺（docs-site/docs/nodes/builtin.md:199，片段摘录）：

`joinBranches` 列出的分支会汇聚结果（`{branchName: result}`），其余作为后台分支 fire-and-forget。

docs-site/docs/guide/expressions.md:33 也把 $PARENT.x 标注为"父流程上下文（子流程中可用）"，没有"仅同步子流程"的限制：

| `$PARENT.x` | 父流程上下文（子流程中可用） | `None` |

既有测试只钉住"首读之后不再漂移"，并在 docstring 里自认等价性依赖 InlineFlow 阻塞语义——tests/unit/test_clean_c1_state.py:117-124：

        """子侧首读后的 $PARENT 快照不再随父 update_node_result 漂移。

        惰性快照契约（_LazyRootSnapshot）：每个根键在子侧首次读取时物化冻结。
        引擎里父执行在 InlineFlow 子流程存续期被阻塞（非重入），「setup 后
        首读前」父状态不可能变化，故首读物化 ≡ setup 时全量快照；本测试钉
        首读之后的隔离（此前浅拷贝连首读后都不隔离——父原地 mutate 直接
        漂进子侧持有的引用）。
        """

实测一：快照读到父流程未来态（语义错误）

$ /tmp/plaita-audit-venv/bin/python /tmp/repro_pcf_f2.py
=== A) _LazyRootSnapshot is a LIVE view, not a snapshot ===
snapshot still exposes later: {'before': 'at-branch-creation', 'later': 42}
A-verdict: 'later' visible = True

构造方式是 st.update_node_result("before", ...) → snap = _LazyRootSnapshot(st) → st.update_node_result("later", 42)；父在包装对象创建之后写入的 later 仍出现在 snap["$NODE"] 里。这一段的可运行版本（repro_pcf_f2.py:14-20）：

st = CheckpointState(prefix="$", input_name="INPUT", parent_name="PARENT",
                     node_name="NODE", global_name="GLOBAL", env_name="ENV")
st.update_node_result("before", "at-branch-creation")
snap = _LazyRootSnapshot(st)            # created NOW (conceptually: branch start)
st.update_node_result("later", 42)      # parent advances AFTERWARD
print("snapshot still exposes later:", snap["$NODE"])
print("A-verdict: 'later' visible =", "later" in snap["$NODE"])

是否漂移取决于分支首读与父推进的时序（分支首读晚于父推进则必漂），所以是非确定性数据竞争，不是稳定的错误值——这会让依赖 $PARENT 的分支结果在同一 flow 上时对时错。

实测二：父侧原地写与分支物化并发 → RuntimeError

$ /tmp/plaita-audit-venv/bin/python /tmp/repro_pcf_f2.py
=== B) concurrent in-place mutation while snapshotting ===
reader exceptions: ['RuntimeError: dictionary changed size during iteration']

本机每次实跑，读线程都命中 RuntimeError；该异常发生在子分支内部，按 concurrent.py:306-311 只会是一条 logger.warning 加一条 _BG_STATE[...]["errors"] 记录，流程返回值不受影响。

实测三：端到端复现（@flow 生成形态的并行后台分支）

flow 形状：start → Parallel(branches=[bg], joinBranches 缺省) → 写 $NODE.later=42 → End，后台分支的子流程 sleep 0.4s 后求值 $PARENT.$NODE.later：

$ /tmp/plaita-audit-venv/bin/python /tmp/repro_pcf_f2c.py
DEBUG:plaita.node.concurrent:branch bg executed: 42
parent result: 42
BG_STATE: {'9224ac91f26c4e96a642f795d86565ec': {'futures': [], 'errors': [], 'done': 1}}

后台分支在父流程写入 later 之后才执行，却成功读到了 42——快照语义在生产调用形态下被打破，且 _BG_STATE 里 errors 为空，没有任何信号。

反证：没有后台分支场景的测试覆盖

$ grep -rn '$PARENT' --include='*.py' tests/ | wc -l
      44
$ grep -rn '$PARENT' --include='*.py' tests/unit/test_clean_c1_state.py | wc -l
      17
$ grep -rn '$PARENT' --include='*.py' tests/unit/test_concurrent.py
tests/unit/test_concurrent.py:160:                                    End(id="end", resultType="success", output="$PARENT.$INPUT.input"),

44 处命中里，17 处在 test_clean_c1_state.py（纯单元级），其余分散在 state/context 单元测试与 codeflow 编译期断言；唯一出现在 Parallel 分支内的就是 test_concurrent.py:160。

后台分支失败记录的读取口 get_background_errors() 也没有生产消费方：

$ grep -rn "get_background_errors" --include='*.py' .
./tests/unit/test_parallel_background_observability.py:84:        errors = par.get_background_errors(execution)
./tests/unit/test_parallel_background_observability.py:88:            errors = par.get_background_errors(execution)
./tests/unit/test_parallel_background_observability.py:125:        assert par.get_background_errors(execution) == []
./plaita/node/concurrent.py:335:    def get_background_errors(self, execution) -> list:

tests/unit/test_concurrent.py:145-169 是唯一在 Parallel 分支里用 $PARENT 的用例，但它 join_branches=["branch1", "branch2"]（全部 join，父被阻塞）且只读 $PARENT.$INPUT（分支存续期内父不写 $INPUT）。没有任何用例在 fire-and-forget 分支上读 $PARENT.$NODE，所以既有回归门不覆盖本缺陷：

$ /tmp/plaita-audit-venv/bin/python -m pytest tests/unit/test_clean_c1_state.py tests/unit/test_concurrent.py -q
20 passed, 6 subtests passed in 2.24s

影响

• 受影响面是"按文档写 flow"的普通作者：parallel 的文档口径就是"未列入 joinBranches 的分支 fire-and-forget"（默认值还是空列表，即默认全后台），所以 Parallel + 分支内读 $PARENT.* 是最自然的用法；这类 flow 在 thread 模式（默认）下即可踩到，不需要任何特殊部署形态。
• 读到的可能是父流程"未来"的数据：分支读 $PARENT.$NODE.x 拿到的是父在分支提交之后才写入的节点结果。表现为同一份输入、同一条 flow 的结果时对时错（取决于子流程启动/首读与父推进的调度时序），排查成本极高——错误数据看起来完全合法。
• 分支静默失败：并发原地写触发 RuntimeError: dictionary changed size during iteration 时，异常被 done_callback 吞成一条 warning，只写进模块级 _BG_STATE；读取该记录的 get_background_errors() 全仓只有测试调用（见下），也不进流程返回值——父流程照常结束，分支结果丢失与"分支正常返回 None"无法区分。
• 不修的后果：$PARENT 的"快照"承诺对默认并发形态持续为假，作者无法用任何写法把"父在分支创建时的状态"钉住，只能靠 sleep/自建拷贝规避。

建议

1. 让 $PARENT 的物化点回到父线程、分支提交之前：在 Parallel._execute_background_branches（plaita/node/concurrent.py:259-283）里，executor.submit 之前对每个后台分支的子执行上下文（execution.get_child_execution() + setup_flow）显式冻结 $PARENT；_LazyRootSnapshot 增加一个 freeze()/materialize() 供此处调用（已有 materialize()，见 plaita/core/state.py:111-113），保持"从未读取的根键零成本"的动机不变。
2. 最小止血（可与 1 并行落地）：把 plaita/core/state.py:516-521 的原地 mutate 改成 copy-on-write，例如 node_results = dict(self.get(key, {}) or {}) 后再赋值。单这一处即可消除实测二的 RuntimeError 竞态（读线程遍历的旧 dict 不再被改），但不修复实测一的未来态读取。
3. 无论采用哪种修法，都不要退回"setup 时全量 deepcopy 父状态"——plaita/core/state.py:71-77 记录的 while/for 热路径回归（300 轮 × ~1MB 达 7.7s）必须继续避免。

验收条件：

• 新增测试（tests/unit/test_clean_c1_state.py 或 tests/unit/test_concurrent.py）：父侧在后台分支提交之后调用 update_node_result，断言分支内读到的 $PARENT.$NODE 不含该新节点（快照语义），且父侧 live $NODE 正常包含它。
• 新增并发回归测试：一个线程反复物化 _LazyRootSnapshot($NODE)，另一个线程持续 update_node_result，断言全程零异常（当前必现 RuntimeError: dictionary changed size during iteration）。
• 复现脚本应当变成什么结果：实测三的 repro_pcf_f2c.py 中，后台分支求值 $PARENT.$NODE.later 必须取到分支创建时的值（该节点尚不存在，按 docs-site/docs/guide/expressions.md:33 的缺省口径应为 None），不得为 42；且分支不得出现在 _BG_STATE[...]["errors"] 中。
• 现有 tests/unit/test_clean_c1_state.py tests/unit/test_concurrent.py 保持全绿（当前 20 passed, 6 subtests passed）。

边界

• 本单只针对"$PARENT 惰性快照对非阻塞子执行（Parallel fire-and-forget 分支）不成立"这一个根因，不涉及 join 分支与 InlineFlow 同步子流程——那条路径父确实被阻塞，快照与 setup 时全量快照等价，保持现状。
• 不改 $PARENT / $NODE 的命名空间与缺省取值口径，不改 joinBranches 的缺省语义（后台分支仍是 fire-and-forget），也不顺手重构 _LazyRootSnapshot 的 __eq__/__deepcopy__ 兼容面。
• 不改 Parallel 的 in_plaita_pool_thread() 嵌套降级逻辑（:214-218），那是死锁防护，与本缺陷正交。
• 不承诺"分支内读到父快照后父再改也不影响分支已做的写入"这类双向语义；本单只要求读方向在分支创建时冻结。
• 已核对 gh issue list -R jeffkit/plaita --state all --limit 100：无同根既有单；最近邻 #16（map 子流程 end 递归限制 + if 块作用域隔离）讨论的是编译/作用域问题，与本单的快照时序无关。