diff --git a/docs/CONSUMING.md b/docs/CONSUMING.md index 051e5c2..15b1067 100644 --- a/docs/CONSUMING.md +++ b/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 diff --git a/examples/agui-events/README.md b/examples/agui-events/README.md index 41d6677..24326e1 100644 --- a/examples/agui-events/README.md +++ b/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. diff --git a/examples/agui-events/web/src/chat.tsx b/examples/agui-events/web/src/chat.tsx index ed61131..72924aa 100644 --- a/examples/agui-events/web/src/chat.tsx +++ b/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; /** * 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 | undefined)?.[ - TOOL_STATE - ] ?? {}) as Snapshot), - }, + state: applyDelta(turn.state, event.delta as Operation[]), })); }, }); diff --git a/src/mcp_agent/streaming.py b/src/mcp_agent/streaming.py index f66b9e5..6acdd43 100644 --- a/src/mcp_agent/streaming.py +++ b/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,9 +237,8 @@ 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 @@ -244,7 +246,10 @@ async def stream_turn( 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), diff --git a/src/mcp_agent_api/events.py b/src/mcp_agent_api/events.py index f6a2eb8..98a1304 100644 --- a/src/mcp_agent_api/events.py +++ b/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 diff --git a/src/mcp_agent_api/routes.py b/src/mcp_agent_api/routes.py index de6c180..9209292 100644 --- a/src/mcp_agent_api/routes.py +++ b/src/mcp_agent_api/routes.py @@ -567,12 +567,12 @@ async def thread_snapshot(thread_id: str) -> list[Message]: "One turn, as Server-Sent Events. Each frame is a `data:` " "line carrying an AG-UI event: `RUN_STARTED` first (with " "both ids), then `TEXT_MESSAGE_*` for the answer, " - "`TOOL_CALL_*` per tool, `STATE_SNAPSHOT`, and " + "`TOOL_CALL_*` per tool, `STATE_DELTA`, and " "`ACTIVITY_SNAPSHOT` for what AG-UI has no vocabulary for " "— where a tool's arguments came from and which `ui://` " - "view renders its result. Every `STATE_SNAPSHOT` carries " - "its metadata under `toolState`, leaving the rest of the " - "state object to the client. A `MESSAGES_SNAPSHOT` closes " + "view renders its result. Every `STATE_DELTA` operation " + "names a path under `toolState`, so the rest of the state " + "object stays the client's own. A `MESSAGES_SNAPSHOT` closes " "the run with the thread as the server holds it — history " "is the server's, so reconcile against this rather than " "trusting a local copy. Ends with `RUN_FINISHED`, or " @@ -671,7 +671,7 @@ async def read_thread(thread_id: str) -> dict[str, Any]: tools={tool.name: tool for tool in agent.tools}, ) ], - # The same shape the turn's STATE_SNAPSHOT carries, so a restored + # The same shape the turn's STATE_DELTA carries, so a restored # thread knows what is in state without a second convention. "state": state_metadata(values.get(TOOL_STATE_KEY)), } @@ -723,7 +723,7 @@ async def read_turns(thread_id: str) -> dict[str, Any]: "turn": turn.n, "question": turn.question, "checkpointId": turn.checkpoint_id, - # The same shape every STATE_SNAPSHOT carries, so a turn's + # The same shape every STATE_DELTA carries, so a turn's # state needs no second convention to read. "state": state_metadata(turn.state), } diff --git a/tests/mcp_agent/test_streaming.py b/tests/mcp_agent/test_streaming.py index 491cc97..a0f35c5 100644 --- a/tests/mcp_agent/test_streaming.py +++ b/tests/mcp_agent/test_streaming.py @@ -359,6 +359,36 @@ async def test_state_changes_are_a_running_total_not_the_latest_write(): ] +async def test_a_later_turns_running_total_starts_from_the_thread(): + """ "The whole of it" has to mean the thread's state, not this turn's writes. + + The update channel names only what the node that ran wrote, so a total + seeded empty would announce a second turn's one key as though the thread + held nothing else — and every consumer would have to merge to undo it. + """ + agent, _ = with_session_state( + StreamingScriptedModel( + script=[ + _tool_call("first", "c1"), + AIMessage(content="done"), + _tool_call("second", "c2"), + AIMessage(content="done"), + ] + ), + [ + _publisher_named("first", "a/geometry"), + _publisher_named("second", "b/geometry"), + ], + InMemorySaver(), + ) + + assert [event async for event in stream_turn(agent, "one", "t2")] + events = [event async for event in stream_turn(agent, "two", "t2")] + + changes = [event.state for event in events if isinstance(event, StateChanged)] + assert [sorted(state) for state in changes] == [["a/geometry", "b/geometry"]] + + async def test_a_state_change_carries_the_value_a_display_needs(): """The shape in a receipt line is read off the stored value, so the state a consumer holds has to include it — metadata alone could not describe it.""" diff --git a/tests/mcp_agent_api/test_events.py b/tests/mcp_agent_api/test_events.py index e02df69..aa94847 100644 --- a/tests/mcp_agent_api/test_events.py +++ b/tests/mcp_agent_api/test_events.py @@ -27,6 +27,7 @@ TOOLS_WITHHELD, agui_events, state_metadata, + state_patch, ) from tests.mcp_agent.test_streaming import ( AOI, @@ -81,6 +82,34 @@ def _types(events: list) -> list[str]: return [event.type.value for event in events] +def _applied(events: list, state: dict[str, Any] | None = None) -> dict[str, Any]: + """Every ``STATE_DELTA`` applied in order, the way a client applies them. + + Twelve lines because that is all our patches need: one ``add`` of the whole + namespace, then ``add`` and ``remove`` of single keys under it. Asserting on + the result rather than on the operations is what makes these tests about the + state a client ends up holding. + """ + document = dict(state or {}) + for event in events: + if event.type.value != "STATE_DELTA": + continue + for operation in event.delta: + head, _, tail = operation["path"].lstrip("/").partition("/") + key = tail.replace("~1", "/").replace("~0", "~") + if not tail: + document[head] = operation["value"] + elif operation["op"] == "remove": + document[head].pop(key, None) + else: + document[head][key] = operation["value"] + return document + + +def _deltas(events: list) -> list: + return [event for event in events if event.type.value == "STATE_DELTA"] + + def _activities(events: list) -> dict[str, Any]: return { event.activity_type: event.content @@ -248,50 +277,122 @@ async def test_the_answer_is_labelled_with_the_id_the_thread_will_keep(): assert {event.message_id for event in opened} <= stored -async def test_the_state_object_leaves_room_for_the_client(): - """AG-UI's `state` is shared with the client, not ours to own. Our metadata - sits under one key so a client's own state can sit beside it, and so a - STATE_DELTA has a stable path to patch.""" +def test_a_state_key_is_escaped_before_it_becomes_a_pointer(): + """State keys are ``toolset/name`` and ``/`` is the pointer's own separator. + Unescaped, ``gazet/candidates`` would address a member of a ``gazet`` object + that does not exist, and the whole patch would be rejected.""" + operations = state_patch({}, {"gazet/candidates": {"bytes": 1}, "a~b": {}}) + + assert [operation["path"] for operation in operations] == [ + f"/{STATE_NAMESPACE}/gazet~1candidates", + f"/{STATE_NAMESPACE}/a~0b", + ] + + +def test_state_that_did_not_move_says_nothing(): + """A tool writing one key of six should not re-announce the other five, and + a turn whose state is unchanged should send no event at all.""" + before = {"a/one": {"bytes": 1}, "b/two": {"bytes": 2}} + + assert state_patch(before, before) == [] + assert state_patch(before, {**before, "b/two": {"bytes": 9}}) == [ + {"op": "add", "path": f"/{STATE_NAMESPACE}/b~1two", "value": {"bytes": 9}} + ] + + +def test_a_key_that_left_state_is_removed_from_the_client(): + """Eviction is the only way a key leaves. Left in place it would name a + value the state route now answers 404 for.""" + operations = state_patch({"a/one": {"bytes": 1}}, {}) + + assert operations == [{"op": "remove", "path": f"/{STATE_NAMESPACE}/a~1one"}] + + +async def test_the_client_keeps_its_own_state_through_a_whole_run(): + """AG-UI's ``state`` is shared with the client, not ours to own. Every + operation names a path inside our namespace, so a client's own keys survive + a run untouched — which a snapshot, replacing the object, cannot promise.""" events = await _events() - snapshot = [e for e in events if e.type.value == "STATE_SNAPSHOT"][-1].snapshot - assert list(snapshot) == [STATE_NAMESPACE] - assert STATE_KEY in snapshot[STATE_NAMESPACE] + mine = {"selectedLayer": "era5", "theme": "dark"} + after = _applied(events, {**mine, "unrelated": {"deep": [1, 2]}}) + assert {key: after[key] for key in mine} == mine + assert after["unrelated"] == {"deep": [1, 2]} + assert STATE_KEY in after[STATE_NAMESPACE] + + +async def test_a_runs_first_delta_establishes_the_namespace_whole(): + """One ``add`` of the whole namespace, which cannot fail on an object and is + what gives the per-key operations after it something to address. It is also + the resynchronisation point: a client that lost track gets the map back at + the top of every run.""" + events = await _events() + + first = _deltas(events)[0].delta + assert [operation["op"] for operation in first] == ["add"] + assert first[0]["path"] == f"/{STATE_NAMESPACE}" + assert STATE_KEY in first[0]["value"] + + +async def test_a_second_runs_first_delta_names_the_whole_thread(): + """The namespace goes out whole at the top of each run, so it has to *be* + whole. A run announcing only what this turn wrote would tell a client the + thread had lost everything an earlier turn published.""" + agent = _agent( + [ + _tool_call("search", "c1"), + _tool_call("clip", "c2"), + AIMessage(content="done"), + # The second turn calls only the consumer, so it writes nothing of + # its own and everything it announces came from the first. + _tool_call("clip", "c3"), + AIMessage(content="done"), + ] + ) + await _events(agent) + events = [ + event + async for event in agui_events( + stream_turn(agent, "again", "t1"), thread_id="t1", run_id="r2" + ) + ] + + assert STATE_KEY in _deltas(events)[0].delta[0]["value"] -async def test_state_snapshots_carry_metadata_and_never_the_value(): +async def test_state_deltas_carry_metadata_and_never_the_value(): events = await _events() - snapshots = [e for e in events if e.type.value == "STATE_SNAPSHOT"] - assert snapshots, "a tool published, so state changed" - entry = snapshots[-1].snapshot[STATE_NAMESPACE][STATE_KEY] + deltas = _deltas(events) + assert deltas, "a tool published, so state changed" + entry = _applied(events)[STATE_NAMESPACE][STATE_KEY] assert entry["kind"] == "geojson.AreaOfInterest" assert entry["tool"] == "search" assert entry["bytes"] > 0 assert "value" not in entry - assert "FeatureCollection" not in json.dumps(snapshots[-1].snapshot) + assert "FeatureCollection" not in json.dumps([e.delta for e in deltas]) -async def test_the_last_state_snapshot_is_the_merged_one(): +async def test_the_last_state_delta_is_the_merged_one(): """Write order is assigned by the state reducer, so an entry taken from a - mid-turn update has no ``seq`` — it is omitted rather than sent as null, - and the snapshot at the end of the turn carries the real one.""" + mid-turn update has no ``seq`` — it is omitted rather than sent as null, and + the delta closing the turn carries the real one.""" events = await _events() - snapshots = [e for e in events if e.type.value == "STATE_SNAPSHOT"] - assert "seq" not in snapshots[0].snapshot[STATE_NAMESPACE][STATE_KEY] - assert snapshots[-1].snapshot[STATE_NAMESPACE][STATE_KEY]["seq"] == 1 - assert _types(events).index("STATE_SNAPSHOT") < _types(events).index( + deltas = _deltas(events) + assert "seq" not in deltas[0].delta[0]["value"][STATE_KEY] + assert _applied(events)[STATE_NAMESPACE][STATE_KEY]["seq"] == 1 + assert _types(events).index("STATE_DELTA") < _types(events).index( "TEXT_MESSAGE_START" ) -async def test_the_final_snapshot_lands_before_the_run_closes(): +async def test_the_final_delta_lands_before_the_run_closes(): events = await _events() kinds = _types(events) assert kinds[-1] == "RUN_FINISHED" - assert kinds[-2] == "STATE_SNAPSHOT" + assert kinds[-2] == "STATE_DELTA" async def test_a_view_is_announced_with_its_uri_not_its_bundle(): @@ -373,7 +474,7 @@ async def test_a_view_waits_for_the_state_its_own_tool_just_wrote(): for position, event in enumerate(events) if getattr(event, "activity_type", None) == MCP_VIEW ) - assert kinds[:view].count("STATE_SNAPSHOT") == 1 + assert kinds[:view].count("STATE_DELTA") == 1 # Still before the answer opens, which is the rule that matters to a client. assert "TEXT_MESSAGE_START" not in kinds[:view]