Add concat_flatten + merge_all reducers; activate 0035/0036 fixtures

conformance.toml

@@ -29,7 +29,7 @@
 
 [manifest]
 implementation = "openarmature-python"
-spec_pin = "v0.26.0"
+spec_pin = "v0.27.1"
 
 # Status values:
 #   implemented   — shipped behavior matches the proposal's contract
@@ -150,9 +150,9 @@ status = "textual-only"
 since = "0.9.0"
 note = "Drain snapshot semantic and timeout-input validation already implemented as part of the proposal 0010 impl PR (v0.9.0); no additional module-level work needed."
 
-# Spec v0.23.0-v0.26.1 batch (proposals 0031, 0032, 0033, 0034, 0035).
-# All five have impl work landing across the v0.10.0 release cycle;
-# status stays `not-yet` until the release PR flips them to
+# Spec v0.23.0-v0.27.1 batch (proposals 0031, 0032, 0033, 0034, 0035,
+# 0036). All six have impl work landing across the v0.10.0 release
+# cycle; status stays `not-yet` until the release PR flips them to
 # `implemented` with `since = "0.10.0"`. The pinned spec submodule
 # advances ahead of the impl status because newer fixtures need to be
 # visible to the conformance harness as each PR lands.
@@ -170,3 +170,6 @@ status = "not-yet"
 
 [proposals."0035"]
 status = "not-yet"
+
+[proposals."0036"]
+status = "not-yet"

docs/agent/non-obvious-shapes.md

@@ -196,24 +196,28 @@ attributed_candidates.0  Input should be a valid dictionary or
   input_type=list]
 ```
 
-The right fix is a flattening reducer. Until OA ships the spec-blessed built-ins (proposal 0036 — `concat_flatten` for the list-of-lists case, `merge_all` for the dict-of-mappings case — accepted in spec v0.27.0 but not yet absorbed into the python impl), use a small custom reducer:
+The fix is the `concat_flatten` built-in reducer (proposal 0036) — the list-of-lists analog of `append`. Declare it on the parent's collection field:
 
 ```python
-from openarmature.graph import Reducer
+from typing import Annotated
+from openarmature.graph import State, concat_flatten
 
-class _ConcatFlatten(Reducer):
-    name = "concat_flatten"
+class PipelineState(State):
+    attributed_candidates: Annotated[list[ClaimCandidate], concat_flatten] = Field(default_factory=list)
+```
 
-    def __call__(self, prior: list[Any], update: list[list[Any]]) -> list[Any]:
-        return [*prior, *(item for sublist in update for item in sublist)]
+`concat_flatten` folds the per-instance lists into one flat list (`[*prior, *(item for sublist in update for item in sublist)]`), strict like `append` — it raises `ReducerError` if any element of the update isn't itself a list.
 
-concat_flatten = _ConcatFlatten()
+The dict-shaped analog is `merge_all` (also proposal 0036): when each fan-out instance contributes a `dict[str, X]`, the parent's `target_field` receives `list[dict]`, which plain `merge` can't consume. `merge_all` folds the sequence of mappings into the prior with shallow last-write-wins per key:
+
+```python
+from openarmature.graph import State, merge_all
 
 class PipelineState(State):
-    attributed_candidates: Annotated[list[ClaimCandidate], concat_flatten] = ...
+    keyed_results: Annotated[dict[str, Result], merge_all] = Field(default_factory=dict)
 ```
 
-Single-record-per-instance fan-outs (`collect_field: str`, parent field `Annotated[list[X], append]`) don't hit this — the engine still wraps each instance's value as one element, but `append` flattens it correctly since each element is already an `X`. The list-of-lists shape only emerges when the per-instance value is itself a list.
+Single-record-per-instance fan-outs (`collect_field: str`, parent field `Annotated[list[X], append]`) don't hit this — the engine still wraps each instance's value as one element, but `append` flattens it correctly since each element is already an `X`. The list-of-lists shape only emerges when the per-instance value is itself a list (use `concat_flatten`) or a mapping (use `merge_all`).
 
 If a parent field is populated by BOTH direct node writes AND fan-out collection, that's an architectural ambiguity worth fixing upstream — split into two fields, or pick one path.
 

docs/concepts/fan-out.md

@@ -100,6 +100,29 @@ containing:
 - `on_empty="noop"` for an empty items_field → all the above with empty
   lists; `count_field` set to 0.
 
+### Choosing the `target_field` reducer
+
+The engine writes `target_field` as a list with one entry per
+successful instance: `[instance_0_value, instance_1_value, …]`. The
+reducer you declare on the parent field decides how that list folds
+into prior state:
+
+- Each instance emits a single value (`collect_field: X`) →
+  declare `append` on `Annotated[list[X], append]`. Each instance's
+  value is already an `X`; `append` concatenates cleanly.
+- Each instance emits a `list[X]` (0..N records per instance) → the
+  engine lands `list[list[X]]`. Declare `concat_flatten` instead —
+  it flattens one level so the parent field stays `list[X]`. Plain
+  `append` would leave the nesting and fail Pydantic validation.
+- Each instance emits a `dict[str, X]` → the engine lands
+  `list[dict]`. Declare `merge_all`, which folds the mappings into
+  the parent dict with last-write-wins per key. Plain `merge` can't
+  consume a `list[dict]`.
+
+`concat_flatten` and `merge_all` are strict — they raise
+`ReducerError` if an update element isn't the expected list/mapping
+shape. See [state and reducers](state-and-reducers.md#five-built-in-reducers).
+
 ## Empty fan-outs
 
 If `items_field` is set and the parent list is empty (or `count`

docs/concepts/state-and-reducers.md

@@ -107,18 +107,30 @@ engine applies the merge consistently. If two nodes write the same
 field and the merge strategy is wrong, the fix is one line on the
 schema, not surgery across call sites.
 
-## Three built-in reducers
+## Five built-in reducers
 
-| Reducer           | Semantics                                | Typical use                           |
-| ----------------- | ---------------------------------------- | ------------------------------------- |
-| `last_write_wins` | `partial` replaces `prior` *(default)*   | Scalars owned by a single node        |
-| `append`          | `[*prior, *partial]` for list fields     | Traces, message history, accumulators |
-| `merge`           | `{**prior, **partial}` (shallow)         | Metadata bags, namespaced state       |
+| Reducer           | Semantics                                       | Typical use                           |
+| ----------------- | ----------------------------------------------- | ------------------------------------- |
+| `last_write_wins` | `partial` replaces `prior` *(default)*          | Scalars owned by a single node        |
+| `append`          | `[*prior, *partial]` for list fields            | Traces, message history, accumulators |
+| `merge`           | `{**prior, **partial}` (shallow)                | Metadata bags, namespaced state       |
+| `concat_flatten`  | `[*prior, *(x for sub in partial for x in sub)]` | Fan-out collecting `list[X]` per instance |
+| `merge_all`       | fold `list[dict]` into `prior` (last-write-wins) | Fan-out collecting `dict[str, X]` per instance |
 
 ```python
-from openarmature.graph import append, last_write_wins, merge
+from openarmature.graph import append, concat_flatten, last_write_wins, merge, merge_all
 ```
 
+`concat_flatten` and `merge_all` exist for the fan-out collection
+shapes: when a fan-out subgraph emits `list[X]` per instance, the
+parent's `target_field` receives `list[list[X]]` (which `append`
+would leave nested); when it emits `dict[str, X]`, the parent
+receives `list[dict]` (which `merge` can't consume). Both are
+strict like their single-level counterparts — they raise
+`ReducerError` when an update element isn't the expected
+list/mapping shape. See the [fan-out](fan-out.md) page for the
+full pattern.
+
 You can write your own. A reducer is any named callable matching the
 `(prior, partial) -> new` contract.
 

pyproject.toml

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

src/openarmature/AGENTS.md

@@ -1,6 +1,6 @@
 # OpenArmature — Agent documentation
 
-*This is the agent guide bundled with the openarmature Python package, version 0.9.0 (spec v0.26.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.9.0 (spec v0.27.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`.*
 
 ## 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.26.1. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._
+_Sourced from openarmature-spec v0.27.1. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._
 
 ### Capability: `graph-engine`
 
@@ -46,8 +46,31 @@ engine constant, not a reserved node name, so a user node may happen to be named
 
 **Reducer.** A function that merges a node's partial update into the prior state for a given field. Each state
 field has exactly one reducer. The default reducer is _last-write-wins_ (the new value replaces the old).
-Implementations MUST provide at least: `last_write_wins`, `append` (for list-typed fields), and `merge`
-(for mapping-typed fields). Users MAY register custom reducers per field.
+Implementations MUST provide at least: `last_write_wins`, `append` (for list-typed fields), `merge`
+(for mapping-typed fields), `concat_flatten` (for list-typed fields whose updates are lists of lists —
+e.g., fan-out target fields collecting list-emitting per-instance values), and `merge_all` (for
+mapping-typed fields whose updates are lists of mappings — e.g., fan-out target fields collecting
+dict-emitting per-instance values). Users MAY register custom reducers per field.
+
+**`concat_flatten` semantics.** `concat_flatten(prior, update)` returns the concatenation of `prior` with the
+one-level flattening of `update`. Both `prior` and `update` MUST be lists, and every element of `update` MUST
+itself be a list. Violations raise `ReducerError` per §4 (the engine MUST surface the offending field, the
+reducer name, and a root-cause naming the non-list value). Empty `update` is a no-op (returns `prior`
+unchanged). Empty sub-lists inside `update` contribute zero elements (the one-to-many fan-out case where an
+instance legitimately produces zero records). Implementations MUST NOT auto-detect whether `update` is a list
+of lists vs. a flat list — `concat_flatten` is strictly the two-level reducer; callers with mixed-shape
+requirements MUST register a custom reducer rather than rely on shape-dependent behavior.
+
+**`merge_all` semantics.** `merge_all(prior, update)` folds the sequence of mappings in `update` into `prior`,
+applying the same shallow merge semantics as `merge` (later writes win on key conflict; non-conflicting keys
+from `prior` are preserved). For `update = [d_1, d_2, ..., d_n]`, the result is equivalent to applying `merge`
+N times sequentially: `merge(merge(...merge(merge(prior, d_1), d_2)...), d_n)`, so within `update`
+last-write-wins applies across all N dicts (e.g., if `d_2` and `d_n` both set key `k`, `d_n`'s value wins).
+`prior` MUST be a mapping, `update` MUST be a list, and every element of `update` MUST itself be a mapping.
+Violations raise `ReducerError` per §4. Empty `update` is a no-op (returns `prior` unchanged). Empty mappings
+inside `update` contribute zero keys. Implementations MUST NOT auto-detect whether `update` is a list of
+mappings vs. a single mapping — `merge_all` is strictly the list-of-mappings reducer; callers needing both
+behaviors on the same field MUST register a custom reducer rather than rely on shape-dependent behavior.
 
 **Subgraph.** A compiled graph used as a node inside another graph. A subgraph executes against its own state
 schema and produces a partial update that is merged into the parent's state. The merge uses the same reducer
@@ -1035,24 +1058,28 @@ attributed_candidates.0  Input should be a valid dictionary or
   input_type=list]
 ```
 
-The right fix is a flattening reducer. Until OA ships the spec-blessed built-ins (proposal 0036 — `concat_flatten` for the list-of-lists case, `merge_all` for the dict-of-mappings case — accepted in spec v0.27.0 but not yet absorbed into the python impl), use a small custom reducer:
+The fix is the `concat_flatten` built-in reducer (proposal 0036) — the list-of-lists analog of `append`. Declare it on the parent's collection field:
 
 ```python
-from openarmature.graph import Reducer
+from typing import Annotated
+from openarmature.graph import State, concat_flatten
 
-class _ConcatFlatten(Reducer):
-    name = "concat_flatten"
+class PipelineState(State):
+    attributed_candidates: Annotated[list[ClaimCandidate], concat_flatten] = Field(default_factory=list)
+```
 
-    def __call__(self, prior: list[Any], update: list[list[Any]]) -> list[Any]:
-        return [*prior, *(item for sublist in update for item in sublist)]
+`concat_flatten` folds the per-instance lists into one flat list (`[*prior, *(item for sublist in update for item in sublist)]`), strict like `append` — it raises `ReducerError` if any element of the update isn't itself a list.
 
-concat_flatten = _ConcatFlatten()
+The dict-shaped analog is `merge_all` (also proposal 0036): when each fan-out instance contributes a `dict[str, X]`, the parent's `target_field` receives `list[dict]`, which plain `merge` can't consume. `merge_all` folds the sequence of mappings into the prior with shallow last-write-wins per key:
+
+```python
+from openarmature.graph import State, merge_all
 
 class PipelineState(State):
-    attributed_candidates: Annotated[list[ClaimCandidate], concat_flatten] = ...
+    keyed_results: Annotated[dict[str, Result], merge_all] = Field(default_factory=dict)
 ```
 
-Single-record-per-instance fan-outs (`collect_field: str`, parent field `Annotated[list[X], append]`) don't hit this — the engine still wraps each instance's value as one element, but `append` flattens it correctly since each element is already an `X`. The list-of-lists shape only emerges when the per-instance value is itself a list.
+Single-record-per-instance fan-outs (`collect_field: str`, parent field `Annotated[list[X], append]`) don't hit this — the engine still wraps each instance's value as one element, but `append` flattens it correctly since each element is already an `X`. The list-of-lists shape only emerges when the per-instance value is itself a list (use `concat_flatten`) or a mapping (use `merge_all`).
 
 If a parent field is populated by BOTH direct node writes AND fan-out collection, that's an architectural ambiguity worth fixing upstream — split into two fields, or pick one path.
 

src/openarmature/__init__.py

@@ -25,4 +25,4 @@
 """
 
 __version__ = "0.9.0"
-__spec_version__ = "0.26.1"
+__spec_version__ = "0.27.1"

src/openarmature/graph/__init__.py

@@ -51,7 +51,7 @@
 from .observer import DrainSummary, Observer, RemoveHandle, SubscribedObserver
 from .parallel_branches import BranchSpec, ParallelBranchesNode
 from .projection import ExplicitMapping, FieldNameMatching, ProjectionStrategy
-from .reducers import Reducer, append, last_write_wins, merge
+from .reducers import Reducer, append, concat_flatten, last_write_wins, merge, merge_all
 from .state import State
 from .subgraph import SubgraphNode
 
@@ -106,9 +106,11 @@
     "TimingRecord",
     "UnreachableNode",
     "append",
+    "concat_flatten",
     "default_classifier",
     "deterministic_backoff",
     "exponential_jitter_backoff",
     "last_write_wins",
     "merge",
+    "merge_all",
 ]