Fix parallel-branches branch middleware to run in subgraph state space

CHANGELOG.md

@@ -17,6 +17,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 - **Failure-isolation events report the originating cause's category at non-node placements** (proposal 0065, pipeline-utilities §6.3). When `FailureIsolationMiddleware` runs as instance middleware (§9.7), branch middleware (§11.7), or parent-node middleware on a fan-out / parallel-branches node, the graph engine has already wrapped the originating error as a `node_exception` carrier before the middleware catches it. `FailureIsolatedEvent.caught_exception.category` now resolves through that carrier (and any nested carriers) to the nearest categorized originating cause and reports its category instead of the masking `node_exception`, so the reported category agrees with what the §6.1 retry classifier acted on. For example, an instance whose retries exhaust on `provider_unavailable` now surfaces `provider_unavailable` rather than `node_exception`. The `message` tracks the resolved cause for category/message coherence. Node-level placement was already faithful and is unchanged, and catch/degrade behavior is unchanged at every site (only the event's reported cause changes). The wrapped-instance/branch lineage SHOULD (`fan_out_index` / `branch_name`) is deferred to a follow-up, since it needs the engine to surface per-instance identity to the wrapping-site middleware.
 - **Observer privacy flag `disable_llm_payload` renamed to `disable_provider_payload`** (proposal 0059, observability §5.5.4, spec v0.54.0). The observer-level flag on both bundled observers (`OTelObserver` and `LangfuseObserver`) is renamed, and its scope broadens from LLM-completion payload to any provider-call payload (LLM completion today; embedding and rerank when those land). This is a breaking change to both observer constructors: config passing `disable_llm_payload=True` (or `False`) updates to `disable_provider_payload=...` with no other change. The default stays `True` (payload suppressed), and the gating behavior for `LlmCompletionEvent` / `LlmFailedEvent` rendering is unchanged at every existing site. The rename is the only part of proposal 0059 adopted this cycle: the retrieval-provider capability itself (the `EmbeddingProvider` protocol, the `EmbeddingEvent` / `EmbeddingFailedEvent` typed variants, and the embedding span / observation mapping) is not yet implemented and rides as `not-yet` in `conformance.toml`. The §5.5.4 rename touches existing LLM-payload gating, so it lands with the pin. Pinned spec advances v0.53.0 → v0.54.0.
 
+### Fixed
+
+- **Parallel-branches branch middleware now runs in the branch subgraph's state space** (pipeline-utilities §11.7). Branch middleware wraps the branch's subgraph invocation, so a middleware that short-circuits with a subgraph-space partial update (notably `FailureIsolationMiddleware`'s `degraded_update`) is now projected to the parent through the branch's `outputs` mapping, exactly like a real subgraph result. Previously the `outputs` projection ran inside the middleware chain, so a branch-level `degraded_update` written in the subgraph's fields reached the parent state unprojected and tripped extra-field validation. The bug was latent because the only bundled branch middleware exercised until now was `RetryMiddleware`, which re-invokes the chain rather than returning a cross-space update; it surfaces with failure isolation at a branch placement. A `degraded_update` that does not cover a projected `outputs` field contributes nothing for that field (the parent keeps its prior value) rather than raising, consistent with the §11.4 buffer-then-merge model for partial contributions. The success path, fan-out instance middleware (which already operated in subgraph space), and node-level placement are unchanged.
+
 ## [0.13.0] — 2026-06-09
 
 LLM provider hardening release. The pinned spec advances from v0.46.0 to v0.53.0, absorbing four implemented proposals. Proposal 0049 introduces the first spec-normatively-typed observer event variant, `LlmCompletionEvent`, dispatched on every successful LLM provider call; proposal 0058 adds the failure-side counterpart, `LlmFailedEvent`; proposal 0057 extends the completion variant with eight request-side fields. The bundled `OpenAIProvider` retires its sentinel-namespace `NodeEvent` emission for LLM calls entirely, and the OTel and Langfuse observers now drive their LLM span / Generation from the typed events with back-dated timestamps so durations reflect the adapter boundary. Proposal 0047 closes implicit prefix-cache wire-byte stability: `Response.usage` gains cache-stat fields, the OTel observer emits `openarmature.llm.cache_read` attributes, and the OpenAI Chat Completions request body is byte-stable across equivalent inputs regardless of dict insertion order. Custom observers that filtered LLM calls by sentinel namespace MUST migrate to `isinstance` discrimination; `LLM_NAMESPACE` and `LlmEventPayload` remain as a documented compatibility surface.

src/openarmature/graph/parallel_branches.py

@@ -166,16 +166,42 @@ async def run_branch(branch_name: str, spec: BranchSpec[Any]) -> Mapping[str, An
 
                 async def innermost(s: Any) -> Mapping[str, Any]:
                     final_branch_state = await spec.subgraph._invoke(s, child_context)  # noqa: SLF001
-                    # Per §11.4 projection out: only fields named
-                    # in ``outputs`` contribute back to parent
-                    # state; unnamed subgraph fields are discarded.
+                    # Branch middleware wraps the subgraph invocation
+                    # (§11.7), so the chain operates in the branch
+                    # subgraph's state space. Surface the ``outputs``
+                    # source fields keyed by their subgraph names (via
+                    # getattr, preserving field-value identity) so a
+                    # middleware that short-circuits with a subgraph-space
+                    # partial update — FailureIsolation's degraded_update —
+                    # composes in the same space. The §11.4 projection to
+                    # parent fields runs below, OUTSIDE the chain.
                     return {
-                        parent_field: getattr(final_branch_state, sub_field)
-                        for parent_field, sub_field in spec.outputs.items()
+                        sub_field: getattr(final_branch_state, sub_field)
+                        for sub_field in spec.outputs.values()
                     }
 
                 chain: ChainCall = compose_chain(spec.middleware, innermost)
-                return await chain(initial)
+                branch_partial = await chain(initial)
+                # Per §11.4 projection out: map each ``outputs`` sub-field
+                # (read from the subgraph-space partial the chain produced
+                # — the real subgraph result on success, or a
+                # degraded_update on isolation) to its parent field.
+                # Unnamed subgraph fields are discarded.
+                #
+                # Skip a sub-field the partial doesn't carry: a branch
+                # contributes only the parent fields it supplies and the
+                # §11.4 buffer-then-merge model already merges heterogeneous
+                # partial contributions, so an omitted field leaves the
+                # parent to its prior / sibling-branch value. On the success
+                # path ``innermost`` always supplies every sub-field; the
+                # subset case is a degraded_update that doesn't cover a
+                # projected field, where a hard miss would defeat the point
+                # of failure isolation.
+                return {
+                    parent_field: branch_partial[sub_field]
+                    for parent_field, sub_field in spec.outputs.items()
+                    if sub_field in branch_partial
+                }
             finally:
                 _reset_branch_name(token)
 

tests/unit/test_parallel_branches.py

@@ -39,6 +39,12 @@
     append,
     merge,
 )
+from openarmature.graph.middleware import (
+    FailureIsolationMiddleware,
+    RetryConfig,
+    RetryMiddleware,
+    deterministic_backoff,
+)
 
 # ---------------------------------------------------------------------------
 # Shared schemas + helpers
@@ -217,6 +223,151 @@ async def test_three_heterogeneous_branches_merge_to_parent() -> None:
     assert final.gamma_result == 3
 
 
+# ---------------------------------------------------------------------------
+# Branch middleware — state space (§11.7)
+# ---------------------------------------------------------------------------
+
+
+async def test_branch_middleware_degraded_update_projects_through_outputs() -> None:
+    # Regression: branch middleware wraps the subgraph invocation (§11.7),
+    # so the chain operates in the branch subgraph's state space. A
+    # middleware that short-circuits with a subgraph-space partial update —
+    # here FailureIsolation's degraded_update writing the subgraph field
+    # ``b_out`` — MUST project to the parent through the branch's
+    # ``outputs`` mapping, exactly like a real subgraph result. Before the
+    # fix the ``outputs`` projection ran INSIDE the middleware chain, so the
+    # degraded_update reached the parent as ``b_out`` and tripped
+    # extra-field validation (ParentState has no ``b_out``).
+    isolation = FailureIsolationMiddleware(
+        degraded_update={"b_out": 99},
+        event_name="beta_isolated",
+    )
+    compiled = (
+        GraphBuilder(ParentState)
+        .set_entry("dispatcher")
+        .add_parallel_branches_node(
+            "dispatcher",
+            branches={
+                "beta": BranchSpec(
+                    subgraph=_build_beta_raises("boom"),
+                    outputs={"beta_result": "b_out"},
+                    middleware=(isolation,),
+                ),
+            },
+        )
+        .add_edge("dispatcher", END)
+        .compile()
+    )
+    final = await compiled.invoke(ParentState())
+    await compiled.drain()
+    # The branch failed; FailureIsolation degraded it in subgraph space
+    # (b_out=99); ``outputs`` projected b_out -> parent beta_result.
+    assert final.beta_result == 99
+
+
+async def test_branch_middleware_success_path_projects_subgraph_output() -> None:
+    # Guards the other side of the fix: with branch middleware present but
+    # the branch SUCCEEDING, the real subgraph output (not the degraded
+    # value) must still project through ``outputs``. Confirms moving the
+    # projection outside the middleware chain left the success path intact.
+    isolation = FailureIsolationMiddleware(
+        degraded_update={"a_out": 99},
+        event_name="alpha_isolated",
+    )
+    compiled = (
+        GraphBuilder(ParentState)
+        .set_entry("dispatcher")
+        .add_parallel_branches_node(
+            "dispatcher",
+            branches={
+                "alpha": BranchSpec(
+                    subgraph=_build_alpha_succeeds(),  # returns a_out=1
+                    outputs={"alpha_result": "a_out"},
+                    middleware=(isolation,),
+                ),
+            },
+        )
+        .add_edge("dispatcher", END)
+        .compile()
+    )
+    final = await compiled.invoke(ParentState())
+    await compiled.drain()
+    assert final.alpha_result == 1  # real subgraph output, not the degraded 99
+
+
+async def test_branch_middleware_degraded_update_omitting_field_skips_contribution() -> None:
+    # Leniency: a degraded_update that does not cover a projected
+    # ``outputs`` sub-field contributes nothing for that field rather than
+    # raising. The §11.4 buffer-then-merge model already merges partial
+    # contributions, so the parent field keeps its prior value. Here the
+    # branch degrades with an EMPTY update, so ``beta_result`` is never
+    # contributed and stays at its ParentState default (0). A hard miss
+    # would defeat the point of failure isolation (the resilience primitive
+    # would itself crash the invocation).
+    isolation = FailureIsolationMiddleware(
+        degraded_update={},
+        event_name="beta_isolated",
+    )
+    compiled = (
+        GraphBuilder(ParentState)
+        .set_entry("dispatcher")
+        .add_parallel_branches_node(
+            "dispatcher",
+            branches={
+                "beta": BranchSpec(
+                    subgraph=_build_beta_raises("boom"),
+                    outputs={"beta_result": "b_out"},
+                    middleware=(isolation,),
+                ),
+            },
+        )
+        .add_edge("dispatcher", END)
+        .compile()
+    )
+    final = await compiled.invoke(ParentState())
+    await compiled.drain()
+    assert final.beta_result == 0  # never contributed; parent default retained
+
+
+async def test_branch_middleware_isolation_wraps_retry_degrades_after_exhaustion() -> None:
+    # Fixture-064-Case-1-shaped at a branch: middleware [failure_isolation,
+    # retry] (outer-to-inner). The branch's node fails on every attempt;
+    # retry exhausts its two attempts and re-raises; failure_isolation
+    # catches the exhausted exception and degrades in subgraph space, which
+    # projects to the parent. Exercises the state-space fix through a real
+    # multi-middleware chain rather than a single frame.
+    isolation = FailureIsolationMiddleware(
+        degraded_update={"b_out": 99},
+        event_name="beta_isolated",
+    )
+    retry = RetryMiddleware(
+        RetryConfig(
+            max_attempts=2,
+            classifier=lambda _exc, _state: True,
+            backoff=deterministic_backoff(0),
+        )
+    )
+    compiled = (
+        GraphBuilder(ParentState)
+        .set_entry("dispatcher")
+        .add_parallel_branches_node(
+            "dispatcher",
+            branches={
+                "beta": BranchSpec(
+                    subgraph=_build_beta_raises("boom"),
+                    outputs={"beta_result": "b_out"},
+                    middleware=(isolation, retry),
+                ),
+            },
+        )
+        .add_edge("dispatcher", END)
+        .compile()
+    )
+    final = await compiled.invoke(ParentState())
+    await compiled.drain()
+    assert final.beta_result == 99
+
+
 # ---------------------------------------------------------------------------
 # fail_fast policy
 # ---------------------------------------------------------------------------