Fix nested fan-out collapse under concurrency

A fan-out nested inside an outer fan-out instance keyed its
per-fan-out tracking entry by namespace and node name only, so the
shared progress dict collided across concurrent outer instances. The
second instance found the first's entry already complete and rolled
its result forward, so every outer instance returned the first
instance's inner result (silently wrong output) and the inner
subgraph ran only once.

Carry the enclosing fan-out instance lineage on the tracking key, in
the in-memory dict and through the checkpoint projection, lookup,
cleanup, and restore. Top-level and subgraph- or branch-nested
fan-outs have an empty lineage, so their behavior, including resume,
is unchanged. A fan-out nested inside an outer instance re-runs
rather than skipping on resume, since the record format carries no
lineage; tracked as a follow-up.

Author: chris-colinsky

CHANGELOG.md

@@ -12,6 +12,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 - **Langfuse per-branch dispatch-span observation** (observability §4.3 / §8.4.2, proposals 0042 / 0044). The Langfuse observer now synthesizes a per-branch Span observation under a `parallel_branches` dispatcher node, so each branch's inner observations nest under their own branch span (a three-level dispatcher / per-branch-span / inner-nodes tree) instead of parenting directly under the dispatcher. The per-branch observation carries the OA-emitted `branch_name` alongside the caller baseline metadata and any per-branch augmentation, and the Generation observation now carries `branch_name` too. The OTel observer already produced this shape (proposal 0044 shipped OTel-only in v0.11.0); this brings the Langfuse mapping into line. Callable branches (proposal 0075) are unchanged.
 - **Augmentation no longer lands on a shared-parent fan-out / parallel-branches node** (observability §3.4, proposal 0045). A key set via `set_invocation_metadata` inside a fan-out instance or a parallel-branches branch was incorrectly applied to the shared fan-out / dispatcher NODE span (the fork point) when the augmenting context executed at that node's own namespace, in addition to the per-instance / per-branch dispatch span where it belongs. Both the OTel and Langfuse observers now skip the shared-parent node in that case, matching the behavior already applied to strict-ancestor shared parents. The per-instance / per-branch dispatch spans and the lineage ancestors that carry the augmentation are unaffected.
 - **Langfuse fan-out-instance dispatch nested below the top level** (observability §5.4, proposal 0013). The Langfuse observer's per-instance dispatch synthesis and parent resolution are now prefix-general, so a fan-out node sitting inside a serial subgraph wrapper (rather than at the top namespace level) gets its per-instance dispatch observation synthesized and its inner observations parented under it. This matches the OTel observer, which already resolved across every namespace prefix.
+- **Nested fan-out no longer collapses under concurrency** (engine). A fan-out nested inside an outer fan-out instance shared a single per-fan-out tracking entry across all outer instances, because that entry was keyed by namespace plus node name only. With concurrent outer instances the second instance found the first's entry already marked complete and rolled its result forward, so every outer instance returned the first instance's inner result (silently wrong output) and the inner subgraph ran only once. The tracking key now carries the enclosing fan-out instance lineage, so each outer instance gets its own inner fan-out progress and correct per-instance results. Top-level and subgraph- or branch-nested fan-outs are unaffected (their enclosing lineage is empty). Resume of a fan-out nested inside an outer fan-out instance does not yet round-trip per-outer-instance progress, so it re-runs rather than skipping on resume; tracked as a follow-up.
 
 ## [0.15.0] — 2026-06-22
 

src/openarmature/graph/compiled.py

@@ -316,7 +316,13 @@ def _find_innermost_fan_out_instance_state(
     # fan-out's full key is (namespace_before_fan_out, fan_out_name)
     # where namespace_before_fan_out + (fan_out_name,) == prefix.
     for split in range(len(prefix), 0, -1):
-        key = (prefix[: split - 1], prefix[split - 1])
+        # The fan-out at prefix[:split] registered its tracking entry keyed by
+        # its ENCLOSING fan-out instance lineage (the non-None fan_out_index chain
+        # up to its own level, prefix depth split-1). Reconstruct it from the
+        # current chain so a fan-out nested inside an outer instance routes to the
+        # right outer instance's entry.
+        lineage = tuple(i for i in context.fan_out_index_chain[: split - 1] if i is not None)
+        key = (prefix[: split - 1], prefix[split - 1], lineage)
         if key in state_dict:
             exec_state = state_dict[key]
             idx = context.fan_out_index
@@ -326,7 +332,7 @@ def _find_innermost_fan_out_instance_state(
 
 
 def _project_fan_out_progress(
-    state_dict: Mapping[tuple[tuple[str, ...], str], _FanOutExecutionState],
+    state_dict: Mapping[tuple[tuple[str, ...], str, tuple[int, ...]], _FanOutExecutionState],
 ) -> tuple[FanOutProgress, ...]:
     """Project the engine-internal mutable per-fan-out state into the
     frozen :class:`FanOutProgress` shape on a saved record.
@@ -343,7 +349,16 @@ def _project_fan_out_progress(
     byte-identically, which matters for backends that hash records.
     """
     out: list[FanOutProgress] = []
-    for (namespace, name), exec_state in sorted(state_dict.items()):
+    # The key's third element is the enclosing fan-out instance lineage; it is
+    # NOT projected onto the record (top-level / subgraph-nested fan-outs have an
+    # empty lineage, and nested-fan-out resume is a tracked limitation). One
+    # consequence: a fan-out nested inside an outer fan-out instance emits one
+    # record PER outer instance, all sharing (namespace, fan_out_node_name);
+    # _restore_fan_out_progress_state is last-wins on that collision (the nested
+    # fan-out re-runs on resume regardless). Sorting includes the lineage so
+    # those same-namespace entries still order deterministically (preserving the
+    # byte-identical-record guarantee above).
+    for (namespace, name, _lineage), exec_state in sorted(state_dict.items()):
         instances = tuple(
             FanOutInstanceProgress(
                 state=inst.state,
@@ -366,7 +381,7 @@ def _project_fan_out_progress(
 
 def _restore_fan_out_progress_state(
     saved: Sequence[FanOutProgress],
-) -> dict[tuple[tuple[str, ...], str], _FanOutExecutionState]:
+) -> dict[tuple[tuple[str, ...], str, tuple[int, ...]], _FanOutExecutionState]:
     """Inverse projection of :func:`_project_fan_out_progress`. On resume
     the loaded record's frozen ``fan_out_progress`` tuple gets unpacked
     into the mutable per-fan-out tracking dict that ``FanOutNode``
@@ -387,7 +402,7 @@ def _restore_fan_out_progress_state(
     the engine's canonical error-record shape, and a heuristic would
     misclassify them.
     """
-    out: dict[tuple[tuple[str, ...], str], _FanOutExecutionState] = {}
+    out: dict[tuple[tuple[str, ...], str, tuple[int, ...]], _FanOutExecutionState] = {}
     for fp in saved:
         instances: list[_FanOutInstanceState] = []
         for inst in fp.instances:
@@ -400,7 +415,18 @@ def _restore_fan_out_progress_state(
                     completed_inner_positions=list(inst.completed_inner_positions),
                 )
             )
-        key = (fp.namespace, fp.fan_out_node_name)
+        # The enclosing fan-out instance lineage defaults to empty: the saved
+        # record carries no lineage, which is correct for top-level and
+        # subgraph/branch-nested fan-outs (all empty). A fan-out nested inside an
+        # outer fan-out instance does not round-trip its per-outer-instance
+        # progress through the current record format (it would need the lineage
+        # on the record): its in-memory keys carry the lineage, so the restored
+        # empty-lineage entry never matches and the nested fan-out RE-RUNS on
+        # resume. (Before the lineage fix it would instead skip, rolling forward
+        # the collapsed/wrong shared entry -- so re-running is the safer of two
+        # never-correct behaviors, and matches the spec's inner-subgraph re-entry
+        # model.) A full fix needs the lineage on the record; tracked separately.
+        key = (fp.namespace, fp.fan_out_node_name, ())
         out[key] = _FanOutExecutionState(
             fan_out_node_name=fp.fan_out_node_name,
             namespace=fp.namespace,
@@ -1987,7 +2013,14 @@ async def innermost(s: Any) -> Mapping[str, Any]:
         # raised, so subsequent saves in this invocation do not carry
         # stale fan-out progress and a retry middleware on the fan-out
         # node sees a fresh tracked state on the second attempt.
-        fan_out_progress_key = (context.namespace_prefix, current)
+        # Match the lineage-aware key FanOutNode.run registers (namespace, node
+        # name, enclosing fan-out instance lineage) so a nested fan-out's cleanup
+        # pops its own outer-instance entry, not a sibling's.
+        fan_out_progress_key = (
+            context.namespace_prefix,
+            current,
+            tuple(i for i in context.fan_out_index_chain if i is not None),
+        )
         try:
             try:
                 try:

src/openarmature/graph/fan_out.py

@@ -208,7 +208,18 @@ async def run_with_context(
         # shared dict. Resume threads a pre-restored entry through
         # ``context.fan_out_progress_state``; first-run constructs a
         # fresh one with all instances ``not_started``.
-        key = (context.namespace_prefix, self.name)
+        #
+        # The key carries the ENCLOSING fan-out instance lineage (the non-None
+        # fan_out_index chain), not just the namespace + node name. A fan-out
+        # nested inside an outer fan-out instance has the same namespace for every
+        # outer instance, so without the lineage the shared dict collides across
+        # concurrent outer instances and the second instance rolls forward the
+        # first's "completed" results (silently wrong results). Subgraph / branch
+        # nesting and top-level fan-outs contribute no fan-out index, so the
+        # lineage is empty there -- matching the resume restore (which defaults it
+        # to empty), so top-level / subgraph-nested resume is unaffected.
+        enclosing_fan_out_lineage = tuple(i for i in context.fan_out_index_chain if i is not None)
+        key = (context.namespace_prefix, self.name, enclosing_fan_out_lineage)
         exec_state = context.fan_out_progress_state.get(key)
         if exec_state is None:
             exec_state = _FanOutExecutionState(

src/openarmature/graph/observer.py

@@ -525,8 +525,11 @@ class _InvocationContext:
     # of the fan-out so concurrent saves see consistent sibling state.
     # ``_maybe_save_checkpoint`` projects this into the frozen
     # ``FanOutProgress`` shape on the saved CheckpointRecord.
-    fan_out_progress_state: dict[tuple[tuple[str, ...], str], _FanOutExecutionState] = field(
-        default_factory=dict[tuple[tuple[str, ...], str], _FanOutExecutionState]
+    # Keyed by (namespace, fan_out_node_name, enclosing_fan_out_instance_lineage)
+    # -- the lineage (non-None outer fan_out_index chain) disambiguates a fan-out
+    # nested inside an outer fan-out instance across concurrent outer instances.
+    fan_out_progress_state: dict[tuple[tuple[str, ...], str, tuple[int, ...]], _FanOutExecutionState] = field(
+        default_factory=dict[tuple[tuple[str, ...], str, tuple[int, ...]], _FanOutExecutionState]
     )
     # Per spec §6 Drain (proposal 0010): shared mutable counters that
     # the worker reads at drain-cancel time to report undelivered events

tests/unit/test_checkpoint.py

@@ -745,6 +745,118 @@ async def test_nested_fan_out_records_outermost_schema_version() -> None:
         )
 
 
+# ---------------------------------------------------------------------------
+# Nested fan-out (a fan-out inside an outer fan-out INSTANCE): the per-fan-out
+# tracking key carries the enclosing instance lineage. Verify the save/restore
+# round-trip projects the inner fan-out's progress and resume rolls the outer
+# instances forward to the correct per-instance results.
+# ---------------------------------------------------------------------------
+
+
+class _RNestLeafState(State):
+    tag: str = ""
+    seed: str = ""
+    out: str = ""
+
+
+class _RNestMidState(State):
+    tag: str = ""
+    seeds: list[str] = Field(default_factory=list[str])
+    collected: list[str] = Field(default_factory=list[str])
+
+
+class _RNestOuterState(State):
+    products: list[str] = Field(default_factory=list[str])
+    seeds: list[str] = Field(default_factory=list[str])
+    results: list[Any] = Field(default_factory=list[Any])
+
+
+async def test_nested_fan_out_in_instance_resume_round_trips() -> None:
+    """A fan-out nested inside an outer fan-out instance round-trips through a
+    checkpoint save/restore: the projection emits the inner fan-out's
+    per-outer-instance progress, restore rebuilds the lineage-keyed dict, and
+    resuming from the completed record rolls the outer instances forward to the
+    correct per-instance results without re-running the inner leaf."""
+    leaf_calls = 0
+
+    async def leaf(state: _RNestLeafState) -> dict[str, str]:
+        nonlocal leaf_calls
+        leaf_calls += 1
+        return {"out": f"{state.tag}-{state.seed}"}
+
+    leaf_g = (
+        GraphBuilder(_RNestLeafState).add_node("ask", leaf).add_edge("ask", END).set_entry("ask").compile()
+    )
+    mid_g = (
+        GraphBuilder(_RNestMidState)
+        .add_fan_out_node(
+            "inner_fan",
+            subgraph=leaf_g,
+            items_field="seeds",
+            item_field="seed",
+            inputs={"tag": "tag"},
+            collect_field="out",
+            target_field="collected",
+        )
+        .add_edge("inner_fan", END)
+        .set_entry("inner_fan")
+        .compile()
+    )
+    cp = InMemoryCheckpointer()
+    captured_ids: list[str] = []
+    captured_records: list[CheckpointRecord] = []
+    original_save = cp.save
+
+    async def capture_save(invocation_id: str, record: CheckpointRecord) -> None:
+        captured_ids.append(invocation_id)
+        captured_records.append(record)
+        await original_save(invocation_id, record)
+
+    cp.save = capture_save  # type: ignore[method-assign]
+
+    outer_g = (
+        GraphBuilder(_RNestOuterState)
+        .add_fan_out_node(
+            "outer_fan",
+            subgraph=mid_g,
+            items_field="products",
+            item_field="tag",
+            inputs={"seeds": "seeds"},
+            collect_field="collected",
+            target_field="results",
+        )
+        .add_edge("outer_fan", END)
+        .set_entry("outer_fan")
+        .with_checkpointer(cp)
+        .compile()
+    )
+    correct = [("A-x", "A-y"), ("B-x", "B-y")]
+    final = await outer_g.invoke(_RNestOuterState(products=["A", "B"], seeds=["x", "y"]))
+    assert sorted(tuple(sorted(sub)) for sub in final.results) == correct
+    assert leaf_calls == 4
+    # The projection emitted the INNER fan-out's progress (a record whose
+    # fan_out_node_name is the inner node), so the lineage-keyed nested entry
+    # round-trips through the record format without colliding the projection's
+    # key destructuring.
+    assert any(
+        fp.fan_out_node_name == "inner_fan" for rec in captured_records for fp in rec.fan_out_progress
+    ), "expected the inner fan-out's progress in a saved record"
+
+    # Resume from the completed invocation: the outer instances are tracked
+    # ``completed`` so they roll forward (no inner re-run) to the same results,
+    # exercising _restore_fan_out_progress_state on a record set that includes
+    # the nested fan-out's (collapsed-lineage) entries.
+    resume_id = captured_ids[-1]
+    leaf_calls_before_resume = leaf_calls
+    resumed = await outer_g.invoke(
+        _RNestOuterState(products=["A", "B"], seeds=["x", "y"]), resume_invocation=resume_id
+    )
+    assert sorted(tuple(sorted(sub)) for sub in resumed.results) == correct
+    assert leaf_calls == leaf_calls_before_resume, (
+        "completed outer instances roll forward on resume; the inner leaf must not re-run"
+    )
+
+
 # ---------------------------------------------------------------------------
 # Resume re-entry into subgraph: parent_states populated on inner-node saves
 # ---------------------------------------------------------------------------

tests/unit/test_fan_out.py

@@ -273,6 +273,90 @@ async def compute(state: WorkerState) -> Mapping[str, Any]:
     assert final.results == [11, 12, 13]
 
 
+# ---------------------------------------------------------------------------
+# nested fan-out (a fan-out inside an outer fan-out instance)
+# ---------------------------------------------------------------------------
+
+
+class _NestedLeafState(State):
+    tag: str = ""
+    seed: str = ""
+    out: str = ""
+
+
+class _NestedMidState(State):
+    tag: str = ""
+    seeds: list[str] = Field(default_factory=list[str])
+    collected: Annotated[list[str], append] = Field(default_factory=list[str])
+
+
+class _NestedOuterState(State):
+    products: list[str] = Field(default_factory=list[str])
+    seeds: list[str] = Field(default_factory=list[str])
+    results: Annotated[list[Any], append] = Field(default_factory=list[Any])
+
+
+async def test_nested_fan_out_distinct_per_outer_instance_under_concurrency() -> None:
+    """A fan-out nested inside an outer fan-out instance runs its inner
+    subgraph once per (outer, inner) pair and returns the right per-outer
+    result, even with the outer instances running concurrently."""
+    # Regression: the per-fan-out tracking entry was keyed by (namespace, node
+    # name) only, so the inner fan-out's entry collided across outer instances.
+    # With concurrent outer instances the second found the first's entry already
+    # marked complete and rolled its result forward, so every outer instance
+    # returned the first's inner result and the inner subgraph ran only once.
+    leaf_calls = 0
+
+    async def leaf(state: _NestedLeafState) -> Mapping[str, Any]:
+        nonlocal leaf_calls
+        await asyncio.sleep(0)  # yield so the concurrent outer instances interleave
+        leaf_calls += 1
+        return {"out": f"{state.tag}-{state.seed}"}
+
+    leaf_builder: GraphBuilder[_NestedLeafState] = GraphBuilder(_NestedLeafState)
+    leaf_builder.set_entry("ask")
+    leaf_builder.add_node("ask", leaf)
+    leaf_builder.add_edge("ask", END)
+    leaf_graph = leaf_builder.compile()
+
+    mid_builder: GraphBuilder[_NestedMidState] = GraphBuilder(_NestedMidState)
+    mid_builder.set_entry("inner_fan")
+    mid_builder.add_fan_out_node(
+        "inner_fan",
+        subgraph=leaf_graph,
+        items_field="seeds",
+        item_field="seed",
+        inputs={"tag": "tag"},
+        collect_field="out",
+        target_field="collected",
+    )
+    mid_builder.add_edge("inner_fan", END)
+    mid_graph = mid_builder.compile()
+
+    outer_builder: GraphBuilder[_NestedOuterState] = GraphBuilder(_NestedOuterState)
+    outer_builder.set_entry("outer_fan")
+    outer_builder.add_fan_out_node(
+        "outer_fan",
+        subgraph=mid_graph,
+        items_field="products",
+        item_field="tag",
+        inputs={"seeds": "seeds"},
+        collect_field="collected",
+        target_field="results",
+    )
+    outer_builder.add_edge("outer_fan", END)
+    outer_graph = outer_builder.compile()
+
+    final = await outer_graph.invoke(_NestedOuterState(products=["A", "B"], seeds=["x", "y"]))
+    await outer_graph.drain()
+    # Each outer instance collected its OWN inner results; the collapse bug gave
+    # [["A-x", "A-y"], ["A-x", "A-y"]] (the second outer reused the first's).
+    got = sorted(tuple(sorted(sub)) for sub in final.results)
+    assert got == [("A-x", "A-y"), ("B-x", "B-y")]
+    # The inner leaf ran once per (outer, inner) pair, not once total.
+    assert leaf_calls == 4
+
+
 # ---------------------------------------------------------------------------
 # concurrency
 # ---------------------------------------------------------------------------