Add concat_flatten + merge_all reducers; activate 0035/0036 fixtures (#88)

* Add concat_flatten + merge_all reducers; activate 0035/0036 fixtures

Bumps the spec submodule v0.26.1 → v0.27.1 and lands the remaining
two proposals of the v0.10.0 batch.

Proposal 0036 (fan-out collection reducers):

- concat_flatten — the list-of-lists analog of append. A fan-out
  subgraph emitting list[X] per instance lands list[list[X]] at the
  parent target_field; this reducer flattens one level onto prior.
- merge_all — the list-of-mappings analog of merge. A fan-out
  subgraph emitting dict[str, X] per instance lands list[dict];
  this reducer folds the sequence into prior with shallow
  last-write-wins per key.

Both are strict like their single-level counterparts: they raise
TypeError on a bad update-element shape, which the engine wraps as
ReducerError per graph-engine §4. Exported from openarmature.graph
and registered in the conformance adapter's REDUCERS map. No
fan-out config-validation change was needed — OA never restricted
target_field to append (it only checks field declaration).

Proposal 0035 (Langfuse graph-topology fixtures):

- Un-defers 031/032/033 in the Langfuse conformance harness. Spec
  v0.27.1 patched the two fixture-vs-impl ambiguities raised in the
  clarify-subgraph-name-semantics coord thread (fixture 031's
  outer_out step 2 → 3 per the §6 shared counter; fixture 033's
  detached-trace inner namespace → the wrapper node name). The
  Option A subgraph_identity wiring already on main satisfies both.
- Ports the OTel harness's _patch_unsupported_directives into the
  Langfuse harness so the topology fixtures' update_pure_from_state
  inner nodes (computed values the structural assertions don't
  inspect) run as no-ops.

Conformance adapter: parse bare `list` / `dict` field types (the
0036 fixtures use unconstrained container types so the reducer,
not the typed-state layer, is the gatekeeper for the
list-of-lists / list-of-mappings shape).

conformance.toml: spec_pin → v0.27.1, [proposals."0036"] added.
All of 0031-0036 stay not-yet until the v0.10.0 release PR flips
them to implemented since = 0.10.0 together.

Docs: state-and-reducers + fan-out concept pages document the five
built-in reducers and the per-instance collection-shape recipe;
the AGENTS.md non-obvious-shape note is reframed from a custom-
reducer workaround to the now-real built-ins.

12 new reducer unit tests (success + strict-raise paths for both).
937 passed, 130 skipped; lint / format / pyright / manifest clean.

* Address PR 88 review: self-contained reducer snippets + accurate shapes

- concat_flatten / merge_all doc snippets: add the missing
  `from typing import Annotated` / `from pydantic import Field`
  imports so each recipe is self-contained.
- Reword the per-instance collection-shape sentence: a mapping
  per-instance value lands list[dict] (merge_all), not list[list[X]]
  — name the two non-flat shapes separately instead of lumping the
  mapping case under "list-of-lists".

AGENTS.md regenerated from docs/agent/non-obvious-shapes.md.

Author: chris-colinsky

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,35 @@ 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 pydantic import Field
+
+from openarmature.graph import State, concat_flatten
+
+class PipelineState(State):
+    attributed_candidates: Annotated[list[ClaimCandidate], concat_flatten] = Field(default_factory=list)
+```
+
+`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.
+
+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:
 
-class _ConcatFlatten(Reducer):
-    name = "concat_flatten"
+```python
+from typing import Annotated
 
-    def __call__(self, prior: list[Any], update: list[list[Any]]) -> list[Any]:
-        return [*prior, *(item for sublist in update for item in sublist)]
+from pydantic import Field
 
-concat_flatten = _ConcatFlatten()
+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 two non-flat shapes emerge only when the per-instance value is itself a container: a `list[X]` per instance lands `list[list[X]]` (use `concat_flatten`), and a `dict[str, X]` per instance lands `list[dict]` (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.
 

openarmature-spec

@@ -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 = [

pyproject.toml

@@ -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,35 @@ 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 pydantic import Field
+
+from openarmature.graph import State, concat_flatten
+
+class PipelineState(State):
+    attributed_candidates: Annotated[list[ClaimCandidate], concat_flatten] = Field(default_factory=list)
+```
+
+`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.
+
+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:
 
-class _ConcatFlatten(Reducer):
-    name = "concat_flatten"
+```python
+from typing import Annotated
 
-    def __call__(self, prior: list[Any], update: list[list[Any]]) -> list[Any]:
-        return [*prior, *(item for sublist in update for item in sublist)]
+from pydantic import Field
 
-concat_flatten = _ConcatFlatten()
+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 two non-flat shapes emerge only when the per-instance value is itself a container: a `list[X]` per instance lands `list[list[X]]` (use `concat_flatten`), and a `dict[str, X]` per instance lands `list[dict]` (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/AGENTS.md

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

src/openarmature/__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",
 ]