/agentsop-state-reducer

name: agentsop-state-reducer
description: |
  Tool skill for declaring reducers on LangGraph state keys so parallel writes
  merge instead of crashing. Activates whenever a coder agent designs a
  StateGraph with parallel branches, fan-out via Send, multi-agent topologies,
  or whenever a run raises `InvalidUpdateError: At key '<k>': Can receive only
  one value per step`. Encodes the rule "every state key is single-writer or
  has a reducer — nothing in between."
version: 0.1.0

State Reducer · Tool Skill

> Scope: a single decision — for each state key, declare a reducer or > guarantee single-writer. Out of scope: checkpointers, HITL, supervisor vs. > swarm — see `langgraph-sop` for those.

---

1. 何时激活 (Activation Rules)

Activate when **any** trigger fires:

• The task involves a LangGraph `StateGraph`, `TypedDict`/`BaseModel` schema,

or `Annotated[..., <reducer>]` typing.

• The graph has ≥1 of: parallel branches via static fan-out, `Send` API,

multiple agents writing shared state, or a supervisor pattern where workers return concurrently.

• The run raises `langgraph.errors.InvalidUpdateError` — message looks like

`At key 'messages': Can receive only one value per step. Use an Annotated key to handle multiple values.`

• The user pastes a state schema and asks "why does this crash on parallel?"

or "do I need a reducer here?".

• The user is migrating a linear chain to a fan-out / map-reduce shape.

Do **not** activate if every key is written by exactly one node per superstep (see §6: over-reducing single-writer keys is an anti-pattern).

---

2. 核心心智模型 (Core Mental Model)

**LangGraph state is either single-writer or has a reducer. Nothing in between.**

When a node returns `{"k": v}`, LangGraph must decide how to merge `v` into existing `state["k"]`. There are exactly two legal regimes:

1. **Single-writer / "set" semantics** (default, no `Annotated`). At most one node writes the key per superstep. The new value *replaces* the old. Two concurrent writers → `InvalidUpdateError`. 2. **Reducer / "merge" semantics** (`Annotated[T, reducer_fn]`). Any number of writers may write per superstep; LangGraph folds them via `reducer_fn(current, new)`.

The reducer is *commutative-enough* algebra that lets the engine schedule parallel writes without you reasoning about interleavings. Three canonical reducers cover ~90% of real graphs:

| Reducer | Type | Behaviour | |---|---|---| | `add_messages` (from `langgraph.graph.message`) | `list[BaseMessage]` | Append; **dedupe-and-update by message `id`** (in-place edit when IDs match) | | `operator.add` | `list`, `int`, `float`, `str` | List concat / numeric sum | | Custom `(curr, new) -> merged` | anything | Domain-specific merge (keep-latest, dedupe-by-id, LLM-summarise) |

> "Reducers are mandatory, not optional, for parallel execution." > — `[cheatsheet/gotchas]`

The `add_messages` quirk that beginners miss: it is **not** plain append. If a new message shares an `id` with one in state, it overwrites in place. This is what makes HITL `interrupt() + edit-state` work — a human can edit the last AI turn and resume.

The decision is per-key, not per-schema. A schema can mix freely: one key single-writer, another with `add_messages`, another with `operator.add`.

---

3. SOP 工作流 (Agentic Protocol)

Run top-down for every new or modified state schema.

Step 1 · Enumerate writers per key

For each key `k` in the state schema, list every node that returns `k` in its update dict. Be paranoid: include nodes spawned via `Send`, subgraphs whose output schema overlaps the parent, and any conditional branches.

Step 2 · Classify each key

• **Single-writer-per-superstep** → leave un-annotated. Plain `k: T`.
• **Multi-writer in the same superstep** (parallel branches, fan-out workers,

swarm handoffs) → **must** declare a reducer. Plain `k: T` will crash.

• **Sequential multi-writer** (different supersteps) — a reducer is

*optional*: without one, each later write overwrites; with one, each write folds in. Choose based on intent.

> Decision gate: if you cannot answer "which nodes write this key?" in one > sentence per key, stop and redraw the graph before continuing.

Step 3 · Pick the reducer

Use the OP table in §4. Default ladder:

1. List of `BaseMessage` → `add_messages`. 2. Append-only list of anything else → `operator.add`. 3. Counter / accumulator → `operator.add`. 4. Anything weirder → write a custom reducer (§4 OP-5).

Step 4 · Probe-test the parallel update

Before shipping, write a unit test that invokes the graph along the parallel path with deterministic node outputs and asserts state matches expectation. If the reducer is wrong (e.g., `operator.add` on dicts), this is where you find out — not in production.

def test_parallel_merge():
    out = graph.invoke({"items": []})
    assert sorted(out["items"]) == ["a", "b"]  # both workers' contributions present

Step 5 · Document the contract

In a docstring on the TypedDict / Pydantic model, note for each non-default key *why* it has a reducer. Future maintainers will thank you when they refactor.

---

4. 操作模型 (Operation Models)

Each op: **Trigger → Action → Output**.

OP-1 · Declare `add_messages` for a chat history key

• **Trigger**: state has a `messages: list[BaseMessage]` field; ≥1 of (HITL

edit-state, multi-agent that all append turns, ReAct loop).

• **Action**:

  from typing import Annotated, TypedDict
  from langchain_core.messages import AnyMessage
  from langgraph.graph.message import add_messages

  class S(TypedDict):
      messages: Annotated[list[AnyMessage], add_messages]

• **Output**: Concurrent message writes append; same-id writes overwrite

in-place (enables HITL state edits).

OP-2 · Declare `operator.add` for an accumulating list

• **Trigger**: parallel workers each emit zero or more items into a shared

list (e.g., one chunk per worker in a map step).

• **Action**: