feat(api)!: patch the shared state object rather than replace it

docs/CONSUMING.md

@@ -831,14 +831,14 @@ their screen. Ids line up — the question keeps the client's `id`, the answer
 carries the id the thread will store — so a client reconciles in place rather
 than rebuilding its list.
 
-**The read routes are what the stream deliberately leaves out.** A
-`STATE_SNAPSHOT` carries `{kind, tool, bytes}` per key and never the payload, so
+**The read routes are what the stream deliberately leaves out.** The state
+channel carries `{kind, tool, bytes}` per key and never the payload, so
 a client that has decided it wants the 38 kB geometry comes to `/state/{key}` for
 it. The key is qualified by its publishing toolset (`dataset-search/geometry`)
 and that slash is part of the key, not a path separator.
 
-**Per-turn state needs no new storage.** The state channel is cumulative — a
-snapshot names every key the thread holds — so neither "which keys did *this*
+**Per-turn state needs no new storage.** The state channel is cumulative — the
+patches take a client to every key the thread holds — so neither "which keys did *this*
 turn add" nor "what did this turn run on" is answerable from the stream. Both
 are answerable from what LangGraph already keeps: an immutable checkpoint per
 super-step, every past value retained. `/turns` and `?turn=N` read that back.
@@ -907,14 +907,24 @@ enough to link a key in a state panel back to the call that wrote it without any
 bookkeeping of your own — the example UI's cross-highlighting is that mapping and
 nothing else.
 
-`STATE_SNAPSHOT` carries metadata only — `kind`, `tool`, `bytes`, and `seq` once
+`STATE_DELTA` carries metadata only — `kind`, `tool`, `bytes`, and `seq` once
 known — never the stored value, which a frontend fetches when it actually wants
-to draw it. It sits under `toolState` inside AG-UI's state object, so read
-`snapshot.toolState`; the rest of that object is the client's, to hold whatever
-state of its own it wants to keep there. `seq` is assigned when a write is merged, so mid-turn snapshots omit
-it and the snapshot closing the turn carries it. Snapshots are also **partial
-mid-turn**: one names what its node wrote, so a client merges them into what it
-holds rather than replacing.
+to draw it. It sits under `toolState` inside AG-UI's state object.
+
+**Patched rather than snapshotted, and that is the point.** A `STATE_SNAPSHOT`
+replaces the whole `state` object, and that object is shared: everything the
+client keeps in it would go every time a tool wrote. Every operation sent here
+names a path under `toolState` instead, so the client's own keys are never
+touched. Each run opens with one `add` of the whole namespace — the
+resynchronisation point, carrying the thread's state and not merely this turn's
+writes — and what follows is one `add` or `remove` per key that moved. State
+that has not moved sends no event at all.
+
+Two details a client meets. **Escaping**: a state key is `toolset/name` and `/`
+is JSON Pointer's own separator, so `gazet/candidates` arrives as
+`/toolState/gazet~1candidates` (RFC 6901, `~` as `~0`). **`seq`**: assigned when
+a write is merged, so mid-turn entries omit it and the delta closing the turn
+adds it.
 
 Four protocol rules shape the event order, each checked against `@ag-ui/client`'s
 own verifier rather than read off the specification: nothing may precede

examples/agui-events/README.md

@@ -79,14 +79,16 @@ reader can only demonstrate that a format is self-consistent.
 | `TEXT_MESSAGE_START` / `CONTENT` / `END` | the answer, streamed |
 | `TOOL_CALL_START` / `ARGS` / `END` / `RESULT` | the tool-call lifecycle |
 | `MESSAGES_SNAPSHOT` | the thread as the server holds it, closing each run |
-| `STATE_SNAPSHOT` | the state channel |
+| `STATE_DELTA` | the state channel |
 | `ACTIVITY_SNAPSHOT` | a first-class message role in AG-UI, which is what receipts ride |
 
 Those are the only event types this server emits, out of the 33 AG-UI defines.
-No `STATE_DELTA`, no `STEP_*`, `REASONING_*`, `CUSTOM` or `RAW`. Snapshots only,
-each complete in itself, because a delta for an unknown `messageId` is dropped
-silently by the client and would make the wire depend on a patch having
-applied.
+No `STEP_*`, `REASONING_*`, `CUSTOM` or `RAW`. Every *message* event is a
+snapshot, complete in itself, because an activity delta for an unknown
+`messageId` is dropped silently by the client and would make the wire depend on
+a patch having applied. State is the exception, and deliberately: a state
+snapshot replaces the whole shared `state` object, including the keys the client
+keeps in it, so state moves by patch — see below.
 
 ### What a consumer has to know that the protocol does not tell it
 
@@ -100,7 +102,7 @@ implementation. This is the one place where reading the docs is unavoidable, and
 it is deliberate: `getCapabilities()` is **not** implemented, so the vocabulary
 is documented rather than advertised.
 
-**`STATE_SNAPSHOT` carries metadata, not state.** This is the sharpest
+**`STATE_DELTA` carries metadata, not state.** This is the sharpest
 difference. AG-UI's state channel is normally the agent's actual state, and
 clients render or patch it wholesale. Here each key carries only
 `{kind, tool, bytes, seq}` — never the value, because the values are exactly
@@ -109,8 +111,12 @@ what session state exists to keep out of the conversation. Fetching one is
 client showing "state" will show sizes and kinds and think it has everything.
 
 The metadata sits under a **`toolState`** key rather than at the root of the
-state object, so the rest of that object stays the client's own — read
-`snapshot.toolState`, not `snapshot`.
+state object, and every operation names a path inside it — `add` of the whole
+namespace to open a run, then `add` and `remove` of one key each. So whatever
+the client keeps in `state` beside `toolState` survives a run untouched, which
+is the reason this one channel is patched rather than snapshotted. `chat.tsx`
+applies them in twelve lines; the only trap is RFC 6901 escaping, since a state
+key is `toolset/name` and `/` is the pointer's own separator.
 
 **History is the server's.** AG-UI's convention is client-authoritative:
 `RunAgentInput.messages` is the conversation, and the client owns it.
@@ -186,7 +192,7 @@ activity *is* a message in AG-UI, so the library's own pipeline has already put
 each receipt where it belongs, and the server emitted it before the answer's
 text message opened so it cannot land after the answer it explains.
 
-**The heavy value is never on the wire.** `STATE_SNAPSHOT` carries
+**The heavy value is never on the wire.** The state channel carries
 `{kind, tool, bytes}` per key. The right-hand panel is built from that; clicking
 a key fetches `GET /threads/{id}/state/{key}` and shows the 39 kB geometry that
 the transcript never held.

examples/agui-events/web/src/chat.tsx

@@ -12,16 +12,46 @@ type StateEntry = {
   seq?: number;
 };
 
-/** Every key the thread holds, which is what `STATE_SNAPSHOT` carries. */
+/** Every key the thread holds, which is what the state channel describes. */
 type Snapshot = Record<string, StateEntry>;
 
 /**
  * Key inside AG-UI's `state` object holding session-state metadata. The rest
- * of that object belongs to the client, so this reads the one key rather than
- * treating the whole snapshot as the agent's.
+ * of that object belongs to the client, and every operation the server sends
+ * names a path inside this one key — so whatever a client keeps beside it
+ * survives a run untouched.
  */
 const TOOL_STATE = "toolState";
 
+/** One JSON Patch operation, as `STATE_DELTA` carries them. */
+type Operation = { op: string; path: string; value?: StateEntry };
+
+/**
+ * Our `toolState`, moved on by one delta.
+ *
+ * Short because the server only ever sends two shapes: `add` of the whole
+ * namespace, which opens every run and is the resynchronisation point, and
+ * `add`/`remove` of one key under it. RFC 6901 escaping has to be undone —
+ * state keys are `toolset/name`, and `/` is the pointer's own separator.
+ */
+function applyDelta(state: Snapshot, delta: Operation[]): Snapshot {
+  let next = state;
+  for (const { op, path, value } of delta) {
+    if (path === `/${TOOL_STATE}`) {
+      next = { ...((value ?? {}) as unknown as Snapshot) };
+      continue;
+    }
+    const key = path
+      .slice(`/${TOOL_STATE}/`.length)
+      .replace(/~1/g, "/")
+      .replace(/~0/g, "~");
+    next = { ...next };
+    if (op === "remove") delete next[key];
+    else if (value) next[key] = value;
+  }
+  return next;
+}
+
 /** Where a key came from, read off the `state.published` that announced it. */
 type Origin = { toolCallId: string; tool: string; activityId: string };
 
@@ -361,27 +391,19 @@ export function Chat() {
           setMessages([...messages]);
           patch((turn) => ({ ...turn, published: origins(messages, turn.from) }));
         },
-        // Session state arrives on AG-UI's standard `state` channel, under
-        // `toolState` — the rest of that object is the client's, so read the
-        // one key rather than the whole snapshot. Each entry carries
+        // Session state arrives on AG-UI's standard `state` channel as
+        // patches, every one of them under `toolState`. Each entry carries
         // `{kind, tool, bytes, seq}`; see the README.
         //
-        // Merged rather than assigned: a mid-turn snapshot is built from what
-        // that node wrote, so it names only those keys, and assigning it would
-        // drop the panel to four keys and back to six as the turn lands. The
-        // snapshot closing the turn *is* the whole merged state, and overwrites
-        // all of them with the `seq` the reducer finally gave each one. Nothing
-        // ever leaves session state, so a merge cannot keep a key alive past
-        // its deletion.
-        onStateSnapshotEvent: ({ event }) => {
+        // Applied rather than merged: the operations say what changed,
+        // including a key leaving, which a merge could not express. The one
+        // that opens a run replaces the namespace whole — that is the
+        // resynchronisation point, and it carries the thread's state, not just
+        // this turn's writes.
+        onStateDeltaEvent: ({ event }) => {
           patch((turn) => ({
             ...turn,
-            state: {
-              ...turn.state,
-              ...(((event.snapshot as Record<string, unknown> | undefined)?.[
-                TOOL_STATE
-              ] ?? {}) as Snapshot),
-            },
+            state: applyDelta(turn.state, event.delta as Operation[]),
           }));
         },
       });

src/mcp_agent/streaming.py

@@ -112,6 +112,9 @@ class StateChanged:
     total instead, which is what a surface rendering "what is in state" needs,
     and what makes an earlier value's shape available to describe a later tool's
     receipt.
+
+    The total starts from what the thread already held, so "the whole of it"
+    means the thread's state and not merely this turn's writes.
     """
 
     state: dict[str, StateEntry]
@@ -234,17 +237,19 @@ async def stream_turn(
     # The thread is read before the turn to know where this turn's messages
     # begin, exactly as run_turn does; it is cheap next to the model call.
     before = await agent.aget_state(cast(Any, merged))
-    existing: list[BaseMessage] = (getattr(before, "values", None) or {}).get(
-        "messages"
-    ) or []
+    was: dict[str, Any] = getattr(before, "values", None) or {}
+    existing: list[BaseMessage] = was.get("messages") or []
     seen = len(existing)
     if message_id is not None and any(
         getattr(message, "id", None) == message_id for message in existing
     ):
         message_id = None
 
     last_ai: BaseMessage | None = None
-    running: dict[str, StateEntry] = {}
+    # Seeded from the thread, not empty: a second turn writing one key would
+    # otherwise announce that key as though it were all the thread held, and
+    # every consumer would have to merge to undo it.
+    running: dict[str, StateEntry] = dict(was.get(TOOL_STATE_KEY) or {})
     async for mode, payload in agent.astream(
         cast(Any, {"messages": [HumanMessage(text, id=message_id)]}),
         cast(Any, merged),

src/mcp_agent_api/events.py

@@ -29,8 +29,13 @@
    before the answer's text message opens. Citations are the deliberate
    exception: they belong after the answer and are only known then.
 3. **Deltas fail silently.** An ``ACTIVITY_DELTA`` for an unknown ``messageId``
-   is dropped without error, and a patch that fails to apply is a console
-   warning. Only snapshots are emitted here, each complete in itself.
+   is dropped without error, and a state patch that fails to apply is a console
+   warning and no more. Every message event here is therefore a snapshot,
+   complete in itself. State is the one delta, because there the alternative is
+   worse: a snapshot replaces the whole ``state`` object, including the keys the
+   client keeps in it. It is made safe by construction — each run opens by
+   adding :data:`STATE_NAMESPACE` whole, which cannot fail on an object and
+   leaves a run's later per-key operations something to address.
 4. **Activity content must be a JSON object.** ``ActivityMessage.content`` is a
    mapping, so a bare list would not survive being read back — citations go out
    as ``{"ids": [...]}``, never as an array.
@@ -49,7 +54,7 @@
     RunErrorEvent,
     RunFinishedEvent,
     RunStartedEvent,
-    StateSnapshotEvent,
+    StateDeltaEvent,
     TextMessageContentEvent,
     TextMessageEndEvent,
     TextMessageStartEvent,
@@ -81,24 +86,57 @@
 MCP_VIEW = "mcp.view"
 ANSWER_CITATIONS = "answer.citations"
 
-#: Key inside AG-UI's ``state`` object under which every ``STATE_SNAPSHOT``
-#: here carries session-state metadata.
+#: Key inside AG-UI's ``state`` object under which session-state metadata is
+#: carried, and the only path this server ever writes.
 #:
 #: The object is the *client's* as much as ours — the protocol has it hold
-#: whatever a client and agent share, and ``STATE_DELTA`` patches it by JSON
-#: Pointer. Writing our map at the root leaves a client nowhere to keep its own
-#: keys, and overwrites any it sent. Under a namespace both fit, and a patch
-#: has a stable path to address.
+#: whatever a client and agent share. Writing our map at the root leaves a
+#: client nowhere to keep its own keys; under a namespace both fit.
 STATE_NAMESPACE = "toolState"
 
 
-def _state_snapshot(state: Mapping[str, StateEntry] | None) -> StateSnapshotEvent:
-    """One ``STATE_SNAPSHOT``, with the metadata under :data:`STATE_NAMESPACE`.
+def _pointer(key: str) -> str:
+    """A ``tool_state`` key as a JSON Pointer under :data:`STATE_NAMESPACE`.
 
-    The read routes serve :func:`state_metadata` unwrapped: there the body is
-    ours alone, and there is nothing to share it with.
+    Keys are ``toolset/name`` and ``/`` is the pointer's own separator, so an
+    unescaped ``gazet/candidates`` would address a ``candidates`` member of a
+    ``gazet`` object that does not exist. RFC 6901 escaping, ``~`` before ``/``
+    because the second escape introduces a ``~``.
     """
-    return StateSnapshotEvent(snapshot={STATE_NAMESPACE: state_metadata(state)})
+    return f"/{STATE_NAMESPACE}/" + key.replace("~", "~0").replace("/", "~1")
+
+
+def state_patch(
+    previous: Mapping[str, Any] | None, current: Mapping[str, Any]
+) -> list[dict[str, Any]]:
+    """JSON Patch operations taking a client's state from *previous* to *current*.
+
+    Every operation is confined to :data:`STATE_NAMESPACE`, which is what makes
+    the state object shareable: a client's own keys sit beside ours and are
+    never named, so nothing this server sends can disturb them.
+
+    ``previous`` is ``None`` before anything has been announced on a run, and
+    the whole namespace goes out as one ``add``. ``add`` on an object member
+    replaces or creates that member and leaves its siblings alone, so it both
+    establishes the namespace a later per-key operation needs and resynchronises
+    it wholesale — the one thing a delta stream needs to be able to do.
+
+    Nothing is emitted for state that has not moved: the empty list means the
+    caller sends no event at all.
+    """
+    if previous is None:
+        return [{"op": "add", "path": f"/{STATE_NAMESPACE}", "value": dict(current)}]
+    removed = [
+        {"op": "remove", "path": _pointer(key)}
+        for key in previous
+        if key not in current
+    ]
+    changed = [
+        {"op": "add", "path": _pointer(key), "value": entry}
+        for key, entry in current.items()
+        if previous.get(key) != entry
+    ]
+    return removed + changed
 
 
 def state_metadata(state: Mapping[str, StateEntry] | None) -> dict[str, Any]:
@@ -112,9 +150,9 @@ def state_metadata(state: Mapping[str, StateEntry] | None) -> dict[str, Any]:
     ``seq`` is omitted rather than sent as null when it is not yet known. It is
     assigned by the state reducer when the write is merged, so an entry taken
     from a mid-turn update does not carry one — and a client ordering by it
-    would be sorting nulls. The snapshot at the end of the turn is built from
-    the merged state and does carry it. ``kind`` is always present, because
-    there ``None`` is a fact: the value is untyped.
+    would be sorting nulls. The turn's closing update is built from the merged
+    state and does carry it. ``kind`` is always present, because there ``None``
+    is a fact: the value is untyped.
     """
     return {
         key: {
@@ -272,6 +310,18 @@ async def agui_events(
     deferred: list[tuple[str, ToolFinished, str]] = []
     sequence = 0
 
+    #: The metadata map this run has told the client about, or None before it
+    #: has said anything. What a delta is measured against.
+    announced: dict[str, Any] | None = None
+
+    def state_event(source: Mapping[str, StateEntry] | None) -> StateDeltaEvent | None:
+        """One ``STATE_DELTA``, or None when nothing about state has moved."""
+        nonlocal announced
+        current = state_metadata(source)
+        operations = state_patch(announced, current)
+        announced = current
+        return StateDeltaEvent(delta=operations) if operations else None
+
     def next_id(prefix: str) -> str:
         nonlocal sequence
         sequence += 1
@@ -344,7 +394,8 @@ def ready() -> list[BaseEvent]:
 
                 case StateChanged():
                     state = dict(event.state)
-                    yield _state_snapshot(state)
+                    if delta := state_event(state):
+                        yield delta
                     # The write that was missing is in now, so a view held
                     # back at its tool can be filled and sent.
                     for view in ready():
@@ -377,11 +428,13 @@ def ready() -> list[BaseEvent]:
                     if open_message is not None:
                         yield TextMessageEndEvent(message_id=open_message)
                         open_message = None
-                    # The merged state, which the mid-turn snapshots are not:
-                    # they are assembled from what each node wrote, before the
-                    # reducer has assigned write order.
+                    # The merged state, which the mid-turn deltas are not: they
+                    # are assembled from what each node wrote, before the
+                    # reducer has assigned write order. Usually this closing
+                    # delta says only that each entry now has its `seq`.
                     if event.result.sidecar:
-                        yield _state_snapshot(event.result.sidecar)
+                        if delta := state_event(event.result.sidecar):
+                            yield delta
                     if cited := citations_content(event.result.citations):
                         yield _activity(next_id("act"), ANSWER_CITATIONS, cited)
     except Exception as error:  # noqa: BLE001 - the client is owed a reason