feat: Paper II Proposition 30 store-independence + ctx.fork() (P0-3)

Implements pair confluence enforcement from Paper II Proposition 30:
- writes(term): static analysis of store write-sets for all 11 constructs
- check_store_independence(): verifies writes(f) ∩ writes(g) = ∅
- StoreConflictError with descriptive messages referencing Prop. 30
- AsyncPar now checks store-independence before parallel execution
- AsyncPar/Par now use ctx.fork() for isolated contexts per branch
- Context.fork() creates independent trace + memory copies
- Context.merge_trace() combines child traces into parent
- 20 passing tests + 79 prior = 99 total

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

lambdagent/__init__.py

@@ -94,6 +94,8 @@ from .effects import (
     effect_leq, max_effect,
     parse_effect_annotation, infer_effect_for_term,
 )
+# Paper II: Store Independence Analysis
+from .store_analysis import StoreConflictError
 # Phase 1: P0 Engineering Improvements
 from .cancellation import CancellationToken, CancelledError, NullCancellationToken
 from .retry import RetryPolicy, CircuitBreaker, CircuitOpenError, with_retry, with_retry_sync
@@ -185,6 +187,8 @@ __all__ = [
     "serial", "parallel", "iterate",
     "effect_leq", "max_effect",
     "parse_effect_annotation", "infer_effect_for_term",
+    # Paper II: 存储独立性分析
+    "StoreConflictError",
     # 辅助设施
     "Dataset",
     "from_config", "build_agent", "describe_config",

lambdagent/extensions.py

@@ -37,13 +37,17 @@ class Par(Term):
         ctx = ctx or Context()
         if len(self.agents) <= 1:
             return tuple(a.apply(input, ctx) for a in self.agents)
-        # True parallel via thread pool
+        # True parallel via thread pool — each branch gets forked context (Paper II Prop. 30)
         results = [None] * len(self.agents)
+        forked_ctxs = [ctx.fork() for _ in self.agents]
         with ThreadPoolExecutor(max_workers=len(self.agents)) as pool:
-            futures = {pool.submit(a.apply, input, ctx): i for i, a in enumerate(self.agents)}
+            futures = {pool.submit(a.apply, input, forked_ctxs[i]): i for i, a in enumerate(self.agents)}
             for future in as_completed(futures):
                 idx = futures[future]
                 results[idx] = future.result()
+        # Merge traces back
+        for fork_ctx in forked_ctxs:
+            ctx.merge_trace(fork_ctx)
         return tuple(results)
 
     def __or__(self, other: Term) -> Par:

lambdagent/multiagent.py

@@ -612,6 +612,9 @@ class AsyncPar(Term):
     Lambda 语义:
         AsyncPar(f, g) = λx. let (r₁, r₂) = concurrent(f(x), g(x)) in (r₁, r₂)
 
+    Paper II Proposition 30 (Pair Confluence):
+        writes(f) ∩ writes(g) = ∅ → 结果与调度策略无关
+
     与 Par 的区别:
         Par      = 顺序执行（假并行）
         AsyncPar = 线程池并发（真并行）
@@ -622,41 +625,55 @@ class AsyncPar(Term):
     """
 
     def __init__(self, *agents: Term, max_workers: Optional[int] = None,
-                 timeout: Optional[float] = 120.0):
+                 timeout: Optional[float] = 120.0,
+                 check_store_independence: bool = True):
         """
         Args:
             agents: 要并行执行的 Agent
             max_workers: 线程池大小（默认=Agent 数量）
             timeout: 总超时时间（秒）
+            check_store_independence: 是否在执行前检查存储独立性 (Paper II Prop. 30)
         """
         names = " ∥ ".join(a._name for a in agents)
         super().__init__(f"AsyncPar({names})")
         self.agents = list(agents)
         self.max_workers = max_workers or len(agents)
         self.timeout = timeout
+        self._check_store_independence = check_store_independence
 
     def apply(self, input: Any, ctx: Context | None = None) -> tuple:
         """
         并发执行所有 Agent。
 
-        每个 Agent 在独立线程中运行，共享 Context（trace 是线程安全的列表追加）。
+        Paper II Proposition 30:
+            1. 检查 writes(f) ∩ writes(g) = ∅（存储独立性）
+            2. 每个分支 fork 独立的 Context（防止竞态条件）
+            3. 执行后合并 trace 到父 Context
+
         返回顺序与 agents 列表一致。
         """
         ctx = ctx or Context()
         t0 = time.time()
+
+        # Paper II Prop. 30: 存储独立性检查
+        if self._check_store_independence:
+            from .store_analysis import check_store_independence
+            check_store_independence(self.agents)
+
         results = [None] * len(self.agents)
         errors = [None] * len(self.agents)
+        forked_ctxs = [ctx.fork() for _ in self.agents]  # Paper II: 独立上下文
 
-        def _run_agent(idx: int, agent: Term) -> Tuple[int, Any]:
+        def _run_agent(idx: int, agent: Term, fork_ctx: Context) -> Tuple[int, Any]:
             try:
-                result = agent.apply(input, ctx)
+                result = agent.apply(input, fork_ctx)
                 return (idx, result, None)
             except Exception as e:
                 return (idx, None, e)
 
         with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
             futures = {
-                executor.submit(_run_agent, i, agent): i
+                executor.submit(_run_agent, i, agent, forked_ctxs[i]): i
                 for i, agent in enumerate(self.agents)
             }
             for future in as_completed(futures, timeout=self.timeout):
@@ -664,6 +681,10 @@ class AsyncPar(Term):
                 results[idx] = result
                 errors[idx] = error
 
+        # 合并子 Context 的 trace 到父 Context
+        for fork_ctx in forked_ctxs:
+            ctx.merge_trace(fork_ctx)
+
         # 检查错误
         for i, err in enumerate(errors):
             if err is not None:

lambdagent/store_analysis.py

@@ -0,0 +1,214 @@
+"""
+lambdagent.store_analysis — Paper II Proposition 30 存储独立性分析
+
+实现论文 II 命题 30 的前提条件检查:
+    Pair confluence: writes(f) ∩ writes(g) = ∅ → schedule-independent
+
+核心功能:
+  - writes(term): 静态分析 Term 可能写入的存储键集合
+  - reads(term): 静态分析 Term 可能读取的存储键集合
+  - check_store_independence(terms): 检查多个 Term 的写入集合不相交
+  - StoreConflictError: 存储冲突错误
+"""
+
+from __future__ import annotations
+
+from typing import Any, FrozenSet, List, Set
+
+from .core import Term, LambdagentError
+
+
+# ============================================================
+# 异常
+# ============================================================
+
+class StoreConflictError(LambdagentError):
+    """
+    存储冲突错误 — Paper II Proposition 30 违反。
+
+    当 writes(f) ∩ writes(g) ≠ ∅ 时，Pair(f, g) 的结果
+    依赖于调度策略，不满足合流性。
+    """
+
+    def __init__(self, agent_a: str, agent_b: str,
+                 conflicting_keys: FrozenSet[str]):
+        self.agent_a = agent_a
+        self.agent_b = agent_b
+        self.conflicting_keys = conflicting_keys
+        keys_str = ", ".join(sorted(conflicting_keys))
+        super().__init__(
+            f"Pair confluence violation (Paper II Prop. 30): "
+            f"'{agent_a}' and '{agent_b}' both write to: {{{keys_str}}}. "
+            f"Parallel execution is schedule-dependent. "
+            f"Fix: ensure writes({agent_a}) ∩ writes({agent_b}) = ∅, "
+            f"or use sequential Compose instead of Pair/Par."
+        )
+
+
+# ============================================================
+# 写入集合分析: writes(term)
+# ============================================================
+
+def writes(term: Term) -> FrozenSet[str]:
+    """
+    静态分析 Term 可能写入的存储键集合。
+
+    Paper II Proposition 30:
+        writes(f) ∩ writes(g) = ∅ is required for schedule-independence
+
+    规则:
+        writes(Lam)      = ∅ (LLM 调用不写入存储)
+        writes(Tool)     = ∅ (工具调用不直接写入存储，除非显式标注)
+        writes(Compose)  = ∪ writes(stage_i)
+        writes(Pair)     = writes(first) ∪ writes(second)
+        writes(Par)      = ∪ writes(agent_i)
+        writes(If)       = writes(then_) ∪ writes(else_)
+        writes(Loop)     = writes(body)
+        writes(Memory)   = {all keys in Memory.store} ∪ writes(inner)
+        writes(Guard)    = writes(inner)
+        writes(Route)    = ∪ writes(route_i)
+        writes(SharedMemoryAgent) = {all possible keys}
+    """
+    from .primitives import Lam, Compose, If, Loop, Pair, Fst, Snd, Tool
+    from .extensions import Par, Route, Memory, Guard
+
+    result: Set[str] = set()
+
+    if isinstance(term, (Lam, Tool, Fst, Snd)):
+        # 纯函数或外部调用 — 不写入 Lambda 存储（除非有显式标注）
+        if hasattr(term, '_writes'):
+            return frozenset(term._writes)
+        return frozenset()
+
+    elif isinstance(term, Compose):
+        for stage in term.stages:
+            result.update(writes(stage))
+
+    elif isinstance(term, Pair):
+        result.update(writes(term.first))
+        result.update(writes(term.second))
+
+    elif isinstance(term, Par):
+        for agent in term.agents:
+            result.update(writes(agent))
+
+    elif isinstance(term, If):
+        result.update(writes(term.then_))
+        result.update(writes(term.else_))
+        if isinstance(term.cond, Term):
+            result.update(writes(term.cond))
+
+    elif isinstance(term, Loop):
+        result.update(writes(term.body))
+
+    elif isinstance(term, Memory):
+        # Memory 可以写入其 store 的所有键
+        if term.store:
+            result.update(term.store.keys())
+        result.update(writes(term.agent))
+
+    elif isinstance(term, Guard):
+        result.update(writes(term.agent))
+
+    elif isinstance(term, Route):
+        result.update(writes(term.classifier))
+        for route_agent in term.routes.values():
+            result.update(writes(route_agent))
+
+    else:
+        # 多智能体扩展: 尝试分析
+        try:
+            from .multiagent import AsyncPar, _SharedMemoryAgent, GroupChat, Handoff
+            if isinstance(term, AsyncPar):
+                for agent in term.agents:
+                    result.update(writes(agent))
+            elif isinstance(term, _SharedMemoryAgent):
+                # SharedMemory agent 写入所有可能的键
+                all_keys = set(term.shared._store.keys())
+                result.update(all_keys)
+                result.add("__shared_memory__")  # sentinel
+                result.update(writes(term.agent))
+            elif isinstance(term, GroupChat):
+                for agent in term.agent_list:
+                    result.update(writes(agent))
+            elif isinstance(term, Handoff):
+                for agent in term.registry.values():
+                    result.update(writes(agent))
+        except ImportError:
+            pass
+
+    # 检查是否有显式 _writes 标注
+    if hasattr(term, '_writes'):
+        result.update(term._writes)
+
+    return frozenset(result)
+
+
+def reads(term: Term) -> FrozenSet[str]:
+    """
+    静态分析 Term 可能读取的存储键集合。
+
+    类似 writes() 但分析读取操作。
+    """
+    from .primitives import Lam, Compose, If, Loop, Pair, Fst, Snd, Tool
+    from .extensions import Par, Route, Memory, Guard
+
+    result: Set[str] = set()
+
+    if isinstance(term, (Lam, Tool, Fst, Snd)):
+        return frozenset()
+
+    elif isinstance(term, Compose):
+        for stage in term.stages:
+            result.update(reads(stage))
+
+    elif isinstance(term, Pair):
+        result.update(reads(term.first))
+        result.update(reads(term.second))
+
+    elif isinstance(term, Par):
+        for agent in term.agents:
+            result.update(reads(agent))
+
+    elif isinstance(term, Memory):
+        if term.store:
+            result.update(term.store.keys())
+        result.update(reads(term.agent))
+
+    elif isinstance(term, Guard):
+        result.update(reads(term.agent))
+
+    # 检查显式 _reads 标注
+    if hasattr(term, '_reads'):
+        result.update(term._reads)
+
+    return frozenset(result)
+
+
+# ============================================================
+# 存储独立性检查 (Paper II Proposition 30)
+# ============================================================
+
+def check_store_independence(terms: List[Term]) -> None:
+    """
+    检查多个 Term 的写入集合两两不相交。
+
+    Paper II Proposition 30:
+        ∀ i < j: writes(terms[i]) ∩ writes(terms[j]) = ∅
+
+    Raises:
+        StoreConflictError: 存在写入冲突
+    """
+    write_sets = [(t, writes(t)) for t in terms]
+
+    for i in range(len(write_sets)):
+        for j in range(i + 1, len(write_sets)):
+            term_a, writes_a = write_sets[i]
+            term_b, writes_b = write_sets[j]
+            conflict = writes_a & writes_b
+            if conflict:
+                raise StoreConflictError(
+                    term_a._name,
+                    term_b._name,
+                    conflict,
+                )

lambdagent/tests/test_store_independence.py

@@ -0,0 +1,222 @@
+"""
+Tests for Paper II Proposition 30 — store-independence enforcement.
+
+Tests cover:
+  1. writes() analysis for all constructs
+  2. Store-independence checking
+  3. ctx.fork() isolation
+  4. AsyncPar with store-independence enforcement
+  5. Par with forked contexts
+"""
+
+import pytest
+from lambdagent.store_analysis import (
+    writes, reads, check_store_independence, StoreConflictError,
+)
+from lambdagent.core import Context
+from lambdagent.primitives import Lam, Compose, If, Loop, Pair, Tool
+from lambdagent.extensions import Par, Route, Memory, Guard
+from lambdagent.multiagent import AsyncPar
+
+
+# ============================================================
+# 1. writes() Analysis
+# ============================================================
+
+class TestWritesAnalysis:
+
+    def test_lam_no_writes(self):
+        """Lam doesn't write to store"""
+        lam = Lam("test", "prompt")
+        assert writes(lam) == frozenset()
+
+    def test_tool_no_writes(self):
+        """Tool doesn't write to store"""
+        tool = Tool("t", lambda x: x)
+        assert writes(tool) == frozenset()
+
+    def test_memory_writes_keys(self):
+        """Memory writes all keys in its store"""
+        mem = Memory(Tool("t", lambda x: x), store={"key1": "v1", "key2": "v2"})
+        w = writes(mem)
+        assert "key1" in w
+        assert "key2" in w
+
+    def test_compose_unions_writes(self):
+        """Compose writes = union of all stage writes"""
+        m1 = Memory(Tool("t1", lambda x: x), store={"a": 1})
+        m2 = Memory(Tool("t2", lambda x: x), store={"b": 2})
+        comp = Compose(m1, m2)
+        w = writes(comp)
+        assert "a" in w
+        assert "b" in w
+
+    def test_guard_propagates_writes(self):
+        """Guard propagates inner agent writes"""
+        inner = Memory(Tool("t", lambda x: x), store={"key": "v"})
+        guard = Guard(inner, lambda x: True)
+        assert "key" in writes(guard)
+
+    def test_pair_unions_writes(self):
+        """Pair writes = first writes ∪ second writes"""
+        m1 = Memory(Tool("t1", lambda x: x), store={"a": 1})
+        m2 = Memory(Tool("t2", lambda x: x), store={"b": 2})
+        pair = Pair(m1, m2)
+        w = writes(pair)
+        assert "a" in w
+        assert "b" in w
+
+    def test_explicit_writes_annotation(self):
+        """Term with _writes attribute"""
+        tool = Tool("t", lambda x: x)
+        tool._writes = {"custom_key"}
+        assert "custom_key" in writes(tool)
+
+
+# ============================================================
+# 2. Store Independence Checking
+# ============================================================
+
+class TestStoreIndependence:
+
+    def test_independent_agents_pass(self):
+        """Agents with no writes → passes"""
+        agents = [
+            Lam("a", "prompt1"),
+            Lam("b", "prompt2"),
+            Tool("c", lambda x: x),
+        ]
+        check_store_independence(agents)  # Should not raise
+
+    def test_disjoint_writes_pass(self):
+        """Agents writing to different keys → passes"""
+        agents = [
+            Memory(Tool("t1", lambda x: x), store={"a": 1}),
+            Memory(Tool("t2", lambda x: x), store={"b": 2}),
+        ]
+        check_store_independence(agents)  # Should not raise
+
+    def test_overlapping_writes_fail(self):
+        """Agents writing to same key → fails"""
+        agents = [
+            Memory(Tool("t1", lambda x: x), store={"shared": 1}),
+            Memory(Tool("t2", lambda x: x), store={"shared": 2}),
+        ]
+        with pytest.raises(StoreConflictError) as exc_info:
+            check_store_independence(agents)
+        assert "shared" in str(exc_info.value)
+
+    def test_error_message_quality(self):
+        """Error message includes agent names and conflicting keys"""
+        m1 = Memory(Tool("agent_A", lambda x: x), store={"key": 1})
+        m2 = Memory(Tool("agent_B", lambda x: x), store={"key": 2})
+        with pytest.raises(StoreConflictError) as exc_info:
+            check_store_independence([m1, m2])
+        err = str(exc_info.value)
+        assert "Pair confluence" in err
+        assert "Prop. 30" in err
+
+
+# ============================================================
+# 3. ctx.fork() Isolation
+# ============================================================
+
+class TestContextFork:
+
+    def test_fork_independent_trace(self):
+        """Forked context has independent trace"""
+        parent = Context()
+        parent.log("parent_op", "id1", "in", "out", 10.0)
+        child = parent.fork()
+        child.log("child_op", "id2", "in", "out", 5.0)
+        assert len(parent.trace) == 1
+        assert len(child.trace) == 1
+
+    def test_fork_independent_memory(self):
+        """Forked context has independent memory"""
+        parent = Context(memory={"key": "original"})
+        child = parent.fork()
+        child.memory["key"] = "modified"
+        assert parent.memory["key"] == "original"
+        assert child.memory["key"] == "modified"
+
+    def test_fork_shares_bindings(self):
+        """Forked context copies bindings (read-only sharing)"""
+        parent = Context(bindings={"x": 42})
+        child = parent.fork()
+        assert child.bindings["x"] == 42
+
+    def test_merge_trace(self):
+        """merge_trace() combines child trace into parent"""
+        parent = Context()
+        child = parent.fork()
+        child.log("child_op", "id1", "in", "out", 5.0)
+        parent.merge_trace(child)
+        assert len(parent.trace) == 1
+        assert parent.trace[0].term_name == "child_op"
+
+
+# ============================================================
+# 4. AsyncPar with Store Independence
+# ============================================================
+
+class TestAsyncParStoreIndependence:
+
+    def test_asyncpar_passes_with_pure_agents(self):
+        """AsyncPar with pure agents (no writes) passes store check"""
+        agents = [
+            Tool("t1", lambda x: f"result1: {x}"),
+            Tool("t2", lambda x: f"result2: {x}"),
+        ]
+        ap = AsyncPar(*agents)
+        result = ap.apply("test", Context())
+        assert len(result) == 2
+
+    def test_asyncpar_fails_with_conflicting_writes(self):
+        """AsyncPar detects store conflicts"""
+        m1 = Memory(Tool("t1", lambda x: x), store={"shared": 1})
+        m2 = Memory(Tool("t2", lambda x: x), store={"shared": 2})
+        ap = AsyncPar(m1, m2)
+        with pytest.raises(StoreConflictError):
+            ap.apply("test", Context())
+
+    def test_asyncpar_skip_check(self):
+        """AsyncPar with check_store_independence=False skips check"""
+        m1 = Memory(Tool("t1", lambda x: x), store={"shared": 1})
+        m2 = Memory(Tool("t2", lambda x: x), store={"shared": 2})
+        ap = AsyncPar(m1, m2, check_store_independence=False)
+        # Should not raise StoreConflictError (may still have race conditions)
+        result = ap.apply("test", Context())
+        assert len(result) == 2
+
+    def test_asyncpar_forked_contexts(self):
+        """AsyncPar uses forked contexts — traces are merged after"""
+        agents = [
+            Tool("t1", lambda x: f"r1:{x}"),
+            Tool("t2", lambda x: f"r2:{x}"),
+        ]
+        ap = AsyncPar(*agents)
+        ctx = Context()
+        result = ap.apply("input", ctx)
+        # Each agent produces a trace entry + AsyncPar itself
+        assert len(ctx.trace) >= 2  # at least agent traces + asyncpar log
+
+
+# ============================================================
+# 5. Par with Forked Contexts
+# ============================================================
+
+class TestParForkedContexts:
+
+    def test_par_forked_contexts(self):
+        """Par uses forked contexts for thread safety"""
+        agents = [
+            Tool("t1", lambda x: f"r1:{x}"),
+            Tool("t2", lambda x: f"r2:{x}"),
+        ]
+        par = Par(*agents)
+        ctx = Context()
+        result = par.apply("input", ctx)
+        assert len(result) == 2
+        # Traces from forked contexts should be merged back
+        assert len(ctx.trace) >= 2