Implement fan-out degrade contribution (0066) (#161)

Adopt proposal 0066: a degraded fan-out instance is a success whose
contribution is its degraded_update, read in subgraph-field-name space.

Make the per-instance partial uniformly subgraph-keyed across the
extract, accumulator, resume, and fan-in paths, so a degrade's
subgraph-space degraded_update composes through the same fan-in. This
fixes a latent bug where an instance degraded_update's extra_outputs
values were looked up by parent field name and silently dropped.

A static degraded_update on an instance FailureIsolationMiddleware that
omits collect_field is now a compile-time error
(FanOutDegradedUpdateMissingCollectField); a callable degraded_update
that omits it yields a graceful null slot rather than raising.

Advance the pinned spec to v0.56.0, mark 0066 implemented, wire fixture
065 (four cases), and sync the spec-version declarations.

Author: chris-colinsky

CHANGELOG.md

@@ -16,7 +16,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 - **`RetryMiddleware` now takes a `RetryConfig` record** instead of individual constructor kwargs (proposal 0050 prep). The four retry settings (`max_attempts` / `classifier` / `backoff` / `on_retry`, each optional) move onto a frozen `RetryConfig`; construct as `RetryMiddleware(RetryConfig(max_attempts=...))`, while bare `RetryMiddleware()` still applies the defaults. This is a breaking change to the `RetryMiddleware` constructor. The record is the same shape the upcoming call-level `complete(retry=...)` parameter will accept, so one retry config serves both the per-node and per-call layers. `None` fields resolve to the canonical defaults (`default_classifier` / `exponential_jitter_backoff`) at use, preserving the prior behavior.
 - **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. `conformance.toml` marks proposal 0065 `implemented`, and conformance fixture 064 (three cases: the §9.7 instance and §11.7 branch sites plus an uncategorized cause) passes.
 - **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.55.1 across the v0.14.0 cycle**, in two steps: v0.54.0 (proposal 0059, the observer-flag rename above) and v0.55.1 (proposal 0065 above; the v0.55.1 patch also carries an observability §11 span-links text reconciliation that narrows an *Out of scope* bullet, with no python-observable change). `conformance.toml` records 0065 as `implemented` and 0059 as `not-yet` (only its cross-spec flag rename was adopted).
+- **Fan-out failure-isolation degrade contribution implemented** (proposal 0066, pipeline-utilities §9.3 / §9.8 / §11.7, spec v0.56.0). When `FailureIsolationMiddleware` degrades a fan-out instance, that instance is a success whose contribution is its `degraded_update`, read in subgraph-field-name space and never merged onto the failed instance's pre-failure state. This also fixes a latent bug: an instance `degraded_update`'s `extra_outputs` values were previously looked up by the parent field name and silently dropped (`collect_field` was unaffected). A static `degraded_update` that omits the node's `collect_field` is now a compile-time error (`FanOutDegradedUpdateMissingCollectField`); a callable `degraded_update` that omits it yields a graceful null slot rather than raising, preserving one collection slot per item. The parallel-branches counterpart (a branch `degraded_update` omitting a projected `outputs` field skips that field) was already correct as of the parallel-branches fix above and is now pinned by fixture 065. Success-path and resume behavior for correctly-configured fan-outs is unchanged.
+- **Pinned spec advances v0.53.0 → v0.56.0 across the v0.14.0 cycle**, in three steps: v0.54.0 (proposal 0059, the observer-flag rename above), v0.55.1 (proposal 0065 above; the v0.55.1 patch also carries an observability §11 span-links text reconciliation that narrows an *Out of scope* bullet, with no python-observable change), and v0.56.0 (proposal 0066, the fan-out degrade contribution above). `conformance.toml` records 0065 and 0066 as `implemented` and 0059 as `not-yet` (only its cross-spec flag rename was adopted).
 
 ### Fixed
 

conformance.toml

@@ -32,7 +32,7 @@
 
 [manifest]
 implementation = "openarmature-python"
-spec_pin = "v0.55.1"
+spec_pin = "v0.56.0"
 
 # Status values:
 #   implemented   — shipped behavior matches the proposal's contract
@@ -621,3 +621,17 @@ status = "not-yet"
 [proposals."0065"]
 status = "implemented"
 since = "0.14.0"
+
+# Spec v0.56.0 (proposal 0066).  Fan-out failure-isolation degrade
+# contribution (pipeline-utilities §9.3 / §9.8 / §11.7).  A degraded
+# fan-out instance is a success whose contribution IS its degraded_update
+# (subgraph-space, read by subgraph field name; ``extra_outputs`` keyed by
+# subgraph field too, fixing a latent drop).  A static degraded_update
+# omitting collect_field is a compile error
+# (``fan_out_degraded_update_missing_collect_field``); a callable that omits
+# it yields a graceful null slot (no raise).  The §11.7 parallel-branches
+# counterpart skips an uncovered projected field.  Fixture 065 (four
+# cases) passes.
+[proposals."0066"]
+status = "implemented"
+since = "0.14.0"

openarmature-spec

@@ -63,7 +63,7 @@ Specification = "https://github.com/LunarCommand/openarmature-spec"
 openarmature = "openarmature.cli:main"
 
 [tool.openarmature]
-spec_version = "0.55.1"
+spec_version = "0.56.0"
 
 [dependency-groups]
 dev = [

pyproject.toml

@@ -1,6 +1,6 @@
 # OpenArmature — Agent documentation
 
-*This is the agent guide bundled with the openarmature Python package, version 0.13.0 (spec v0.55.1). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.*
+*This is the agent guide bundled with the openarmature Python package, version 0.13.0 (spec v0.56.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.*
 
 ## TL;DR
 
@@ -10,7 +10,7 @@ OpenArmature is a workflow framework for LLM pipelines and tool-calling agents:
 
 ## Capability contracts
 
-_Sourced from openarmature-spec v0.55.1. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md` verbatim — including additions from accepted proposals that this Python implementation may not yet ship. For per-proposal implementation status (implemented / partial / textual-only / not-yet), see the `conformance.toml` manifest at the repo root. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._
+_Sourced from openarmature-spec v0.56.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md` verbatim — including additions from accepted proposals that this Python implementation may not yet ship. For per-proposal implementation status (implemented / partial / textual-only / not-yet), see the `conformance.toml` manifest at the repo root. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._
 
 ### Capability: `graph-engine`
 

src/openarmature/AGENTS.md

@@ -25,7 +25,7 @@
 """
 
 __version__ = "0.13.0"
-__spec_version__ = "0.55.1"
+__spec_version__ = "0.56.0"
 # Proposal 0052 (spec observability §5.1 / §8.4.1): canonical
 # package-registry name for this implementation. Surfaces on every
 # OTel invocation span as ``openarmature.implementation.name`` and on

src/openarmature/__init__.py

@@ -18,6 +18,7 @@
     DanglingEdge,
     EdgeException,
     FanOutCountModeAmbiguous,
+    FanOutDegradedUpdateMissingCollectField,
     FanOutEmpty,
     FanOutFieldNotList,
     FanOutInvalidConcurrency,
@@ -84,6 +85,7 @@
     "FailureIsolationMiddleware",
     "FanOutConfig",
     "FanOutCountModeAmbiguous",
+    "FanOutDegradedUpdateMissingCollectField",
     "FanOutEmpty",
     "FanOutFieldNotList",
     "FanOutInvalidConcurrency",

src/openarmature/graph/__init__.py

@@ -24,6 +24,7 @@
     ConflictingReducers,
     DanglingEdge,
     FanOutCountModeAmbiguous,
+    FanOutDegradedUpdateMissingCollectField,
     FanOutFieldNotList,
     MappingReferencesUndeclaredField,
     MultipleOutgoingEdges,
@@ -32,7 +33,7 @@
     UnreachableNode,
 )
 from .fan_out import ConcurrencyResolver, CountResolver, FanOutConfig, FanOutNode
-from .middleware import Middleware
+from .middleware import FailureIsolationMiddleware, Middleware
 from .nodes import FunctionNode, Node
 from .parallel_branches import BranchSpec, ParallelBranchesNode
 from .projection import FieldNameMatching, ProjectionStrategy
@@ -248,6 +249,22 @@ def add_fan_out_node[ChildT: State](
                     direction="fan_out.extra_outputs", side="subgraph", field_name=sub_f
                 )
 
+        # Materialize instance_middleware once so the degraded_update check
+        # below and the FanOutConfig build don't both consume a one-shot
+        # iterable.
+        instance_middleware = tuple(instance_middleware or ())
+        # §9.8: a degraded fan-out instance contributes its degraded_update
+        # as the instance result, so a static (mapping) degraded_update on an
+        # instance FailureIsolationMiddleware must cover collect_field. A
+        # callable degraded_update is exempt — its output isn't knowable at
+        # construction; an omitted collect_field yields a runtime null slot.
+        for mw in instance_middleware:
+            if not isinstance(mw, FailureIsolationMiddleware):
+                continue
+            degraded = mw.degraded_update
+            if isinstance(degraded, Mapping) and collect_field not in cast("Mapping[str, Any]", degraded):
+                raise FanOutDegradedUpdateMissingCollectField(node_name=name, collect_field=collect_field)
+
         cfg = FanOutConfig(
             subgraph=subgraph,
             collect_field=collect_field,

src/openarmature/graph/builder.py

@@ -123,6 +123,26 @@ def __init__(self, node_name: str, field_name: str) -> None:
         self.field_name = field_name
 
 
+class FanOutDegradedUpdateMissingCollectField(CompileError):
+    """Raised when a fan-out instance ``FailureIsolationMiddleware`` has a
+    static (mapping) ``degraded_update`` that omits the node's
+    ``collect_field``. A degraded instance contributes its degraded_update
+    as the instance result, so the collected field has to be present. A
+    callable ``degraded_update`` is exempt: its output is not known at
+    construction time, and an omitted collect_field yields a null slot at
+    runtime instead of a failure."""
+
+    category = "fan_out_degraded_update_missing_collect_field"
+
+    def __init__(self, node_name: str, collect_field: str) -> None:
+        super().__init__(
+            f"fan-out node {node_name!r}: a static degraded_update on an instance "
+            f"FailureIsolationMiddleware must include collect_field {collect_field!r}"
+        )
+        self.node_name = node_name
+        self.collect_field = collect_field
+
+
 class ParallelBranchesNoBranches(CompileError):
     """Raised at registration when a parallel-branches node's
     ``branches`` mapping is empty. Per pipeline-utilities §11.9

src/openarmature/graph/errors.py

@@ -336,10 +336,13 @@ async def innermost(s: ChildT) -> Mapping[str, Any]:
             # (§10.11.1) depends on this ordering.
             tracked.result = partial.get(cfg.collect_field)
             tracked.result_is_error = False
+            # ``partial`` is subgraph-space (success or degrade); read each
+            # extra_outputs value by its subgraph field name and store the
+            # accumulator entry under the parent field name.
             tracked.extra_outputs = {
-                parent_field: partial[parent_field]
-                for parent_field in cfg.extra_outputs
-                if parent_field in partial
+                parent_field: partial[sub_field]
+                for parent_field, sub_field in cfg.extra_outputs.items()
+                if sub_field in partial
             }
             tracked.state = "completed"
 
@@ -576,11 +579,16 @@ def _extract_instance_partial(cfg: FanOutConfig, final_state: Any) -> Mapping[st
     """Extract collect_field + extra_outputs values from a finished
     instance's state. Returned as the per-instance partial that flows
     up the instance_middleware chain."""
+    # Per §9.3 the per-instance partial is subgraph-space: collect_field
+    # and every extra_outputs SOURCE field are keyed by their subgraph
+    # field name (the same shape a degrade's degraded_update carries), so
+    # the success and degrade paths compose through one fan-in. The §9.4
+    # projection to parent field names happens in the fan-in.
     partial: dict[str, Any] = {
         cfg.collect_field: getattr(final_state, cfg.collect_field),
     }
-    for parent_field, sub_field in cfg.extra_outputs.items():
-        partial[parent_field] = getattr(final_state, sub_field)
+    for sub_field in cfg.extra_outputs.values():
+        partial[sub_field] = getattr(final_state, sub_field)
     return partial
 
 
@@ -617,10 +625,13 @@ def _rolled_forward_partial(cfg: FanOutConfig, tracked: _FanOutInstanceState) ->
     verbatim — same shape as :func:`_extract_instance_partial` would
     have produced on the original run, sourced from the per-instance
     tracked state instead of a freshly-computed inner state."""
+    # Reconstruct the subgraph-space partial: collect_field plus each
+    # extra_outputs SOURCE field keyed by its subgraph name, sourced from
+    # the parent-keyed accumulator entry.
     partial: dict[str, Any] = {cfg.collect_field: tracked.result}
-    for parent_field in cfg.extra_outputs:
+    for parent_field, sub_field in cfg.extra_outputs.items():
         if parent_field in tracked.extra_outputs:
-            partial[parent_field] = tracked.extra_outputs[parent_field]
+            partial[sub_field] = tracked.extra_outputs[parent_field]
     return partial
 
 
@@ -764,11 +775,15 @@ def _fan_in_fail_fast(
     the fail_fast policy. All ``results`` succeeded (otherwise gather
     would have raised), so the count is just ``len(results)``. Spec
     §9.3 + §9.4: instance-index order."""
+    # §9.4 projection: read each instance's subgraph-space partial by
+    # subgraph field name and collect into the parent field. ``.get`` keeps
+    # an omitted collect_field (a callable degrade that doesn't set it, §9.3)
+    # a graceful null slot rather than a raise.
     partial: dict[str, Any] = {
-        cfg.target_field: [r[cfg.collect_field] for r in results],
+        cfg.target_field: [r.get(cfg.collect_field) for r in results],
     }
-    for parent_field in cfg.extra_outputs:
-        partial[parent_field] = [r[parent_field] for r in results]
+    for parent_field, sub_field in cfg.extra_outputs.items():
+        partial[parent_field] = [r.get(sub_field) for r in results]
     if cfg.count_field is not None:
         partial[cfg.count_field] = len(results)
     return partial
@@ -806,10 +821,10 @@ def _fan_in_collect(
             successes.append(r)
 
     partial: dict[str, Any] = {
-        cfg.target_field: [s[cfg.collect_field] for s in successes],
+        cfg.target_field: [s.get(cfg.collect_field) for s in successes],
     }
-    for parent_field in cfg.extra_outputs:
-        partial[parent_field] = [s[parent_field] for s in successes]
+    for parent_field, sub_field in cfg.extra_outputs.items():
+        partial[parent_field] = [s.get(sub_field) for s in successes]
     if cfg.errors_field is not None:
         partial[cfg.errors_field] = error_records
     if cfg.count_field is not None: