From 2bc866a52ec17aec4f33ede48ff238c0fc2c40ff Mon Sep 17 00:00:00 2001 From: Ciaran Sweet Date: Mon, 17 Aug 2026 11:46:12 +0100 Subject: [PATCH 1/2] feat(api): read a thread back with its activities, not just its messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A restored thread showed what was said and nothing about what the agent did — no receipts, no publications, no views, no citations. Those are the things this runtime exists to make visible, so a reload lost the half that matters. They were never derived, only stored: capture writes the receipts and the captured-key map onto each tool message's artifact, and the checkpointer keeps them. So this reads them back rather than rebuilding anything. An activity is a message in AG-UI, so they travel in `messages` and sit beside the call they belong to with no correlation work — the property the live stream already relies on. The activity payload builders are now shared with the stream rather than reimplemented, so a receipt cannot read one way live and another after a reload. Verified against a live turn: every restored activity is byte-identical to the one the stream sent, `mcp.view` and its rendered data included. Two things degrade rather than guess. A turn whose checkpoints have been pruned has no state to describe its receipts against, so its view is not rebuilt. A `ui://` bundle comes from the deployment as it stands, so a tool removed since has no view rather than a dangling URI. Costs a second read per thread — turns_of walks the checkpoint history — which is the documented price of deriving turns from a structure that does not record them. Co-Authored-By: Claude Opus 5 --- docs/CONSUMING.md | 2 +- examples/agui-events/README.md | 17 ++- examples/agui-events/web/src/chat.tsx | 25 ++-- src/mcp_agent/streaming.py | 10 +- src/mcp_agent_api/events.py | 179 ++++++++++++-------------- src/mcp_agent_api/routes.py | 150 ++++++++++++++++++--- tests/mcp_agent_api/test_routes.py | 50 ++++++- 7 files changed, 293 insertions(+), 140 deletions(-) diff --git a/docs/CONSUMING.md b/docs/CONSUMING.md index 1fbbe93..051e5c2 100644 --- a/docs/CONSUMING.md +++ b/docs/CONSUMING.md @@ -808,7 +808,7 @@ stay. Documenting without re-serialising keeps both. | | | | --- | --- | | `POST /runs` | one turn, streamed as AG-UI SSE — the whole conversation is here | -| `GET /threads/{id}` | the thread's messages, so a page reload restores it | +| `GET /threads/{id}` | the thread's messages and activities, so a page reload restores it | | `GET /threads/{id}/turns` | its turns, and what session state held at the end of each | | `GET /threads/{id}/state/{key}` | one session-state value in full; `?turn=N` for the value as of then | | `GET /views/{toolset}/{view}` | the HTML for a `ui://` bundle a tool declared | diff --git a/examples/agui-events/README.md b/examples/agui-events/README.md index 7890eff..41d6677 100644 --- a/examples/agui-events/README.md +++ b/examples/agui-events/README.md @@ -147,11 +147,18 @@ one. Two routes rebuild it: `/threads/{id}` for the transcript and question, since the stream carries no turn boundary a reloaded client could have seen. -What does *not* come back is the annotation. Receipts, views and citations are -activity messages, and the server does not rebuild past turns' activities — so a -restored thread shows what was said and what is in session state, but not where -a tool's arguments came from, and the cross-highlighting is empty until the next -turn. +The activities come back too — receipts, publications, views and citations — +because an activity *is* a message in AG-UI, so `/threads/{id}` carries them +beside the calls they belong to. They are the same payloads the stream sent, +through the same builders, so a receipt cannot read one way live and another +after a reload. The cross-highlighting works on a restored thread with no +special case. + +Two things a reload cannot bring back, and the server degrades rather than +guesses: a turn whose checkpoints have been **pruned** has no state to describe +its receipts against, so its view is not rebuilt; and a `ui://` bundle comes +from the deployment as it stands, so a tool **removed since** has no view rather +than a dangling URI. **Views are a second protocol.** `mcp.view` names a `ui://` URI and carries the tool's structured content; the HTML comes from `GET /views/{toolset}/{view}`, diff --git a/examples/agui-events/web/src/chat.tsx b/examples/agui-events/web/src/chat.tsx index 6fea797..ed61131 100644 --- a/examples/agui-events/web/src/chat.tsx +++ b/examples/agui-events/web/src/chat.tsx @@ -236,15 +236,15 @@ export function Chat() { /** Rebuild the conversation from the thread id alone. * * Two routes, because the stream has no turn boundary a reloaded client - * could have seen: `/threads/{id}` is the transcript, `/threads/{id}/turns` - * is what state held at the end of each turn. They are joined on the - * question — turn *n* starts at the *n*th user message. + * could have seen: `/threads/{id}` is the transcript **and its activities**, + * `/threads/{id}/turns` is what state held at the end of each turn. They are + * joined on the question — turn *n* starts at the *n*th user message. * - * **Activities do not come back.** Receipts, views and citations are - * activity messages, and the server does not rebuild past turns' — so a - * restored thread shows what was said and what is in state, but not where a - * tool's arguments came from. `published` is empty for the same reason: it - * is read off `state.published`, which is an activity. + * The activities come back as messages, which is what an activity is in + * AG-UI, so `origins` folds them into the same `key -> origin` map the live + * client builds and the cross-highlighting works with no special case. Each + * turn is bounded by the next one's start: unbounded, turn 1 would claim + * every later turn's publications too. */ useEffect(() => { let cancelled = false; @@ -254,19 +254,20 @@ export function Chat() { const past = await readTurns(threadId).catch(() => null); if (cancelled) return; - const questions = thread.messages + const all = thread.messages as unknown as Message[]; + const starts = thread.messages .map((message, index) => ({ message, index })) .filter(({ message }) => message.role === "user"); - const restored: Turn[] = questions.map(({ message, index }, n) => ({ + const restored: Turn[] = starts.map(({ message, index }, n) => ({ n: n + 1, question: message.content || "", questionId: message.id, from: index, state: (past?.history[n]?.state ?? {}) as Snapshot, - published: {}, + published: origins(all.slice(0, starts[n + 1]?.index ?? all.length), index), })); - agent.setMessages(thread.messages as unknown as Message[]); + agent.setMessages(all); setMessages([...agent.messages]); setTurns(restored); setShowing(Math.max(restored.length - 1, 0)); diff --git a/src/mcp_agent/streaming.py b/src/mcp_agent/streaming.py index 546ddb2..f66b9e5 100644 --- a/src/mcp_agent/streaming.py +++ b/src/mcp_agent/streaming.py @@ -153,7 +153,13 @@ def _tool_calls(message: BaseMessage) -> list[ToolStarted]: ] -def _tool_result(message: BaseMessage) -> ToolFinished: +def tool_finished(message: BaseMessage) -> ToolFinished: + """One tool message as a :class:`ToolFinished`. + + Public because a thread readback builds the same thing from the same + message: the artifact carries the receipts and the captured-key map, so + what a tool did with session state is recoverable long after its run. + """ artifact = getattr(message, "artifact", None) identifier = getattr(message, "id", None) return ToolFinished( @@ -254,7 +260,7 @@ async def stream_turn( continue for message in update.get("messages") or []: if getattr(message, "type", None) == "tool": - yield _tool_result(message) + yield tool_finished(message) continue for started in _tool_calls(message): yield started diff --git a/src/mcp_agent_api/events.py b/src/mcp_agent_api/events.py index 5ae33d5..f6a2eb8 100644 --- a/src/mcp_agent_api/events.py +++ b/src/mcp_agent_api/events.py @@ -139,26 +139,6 @@ def _rough_size(value: Any) -> int | None: return None -def _received( - call: ToolStarted | None, - finished: ToolFinished, - state: Mapping[str, StateEntry] | None, -) -> dict[str, Any]: - """Receipts for one call, each with the line a simple client can print. - - ``step_input`` renders the arguments as the Chainlit host shows them, so the - rendered form on the wire and the rendered form in the bundled UI cannot - drift apart. The receipt's own fields go out beside it: a client branches on - ``via``, never on the string. - """ - arguments = call.arguments if call else {} - rendered = step_input(dict(arguments), finished, dict(state or {})) - return { - parameter: {**receipt, "display": str(rendered.get(parameter, ""))} - for parameter, receipt in finished.received.items() - } - - def _activity( message_id: str, activity_type: str, content: dict[str, Any] ) -> BaseEvent: @@ -167,53 +147,80 @@ def _activity( ) -def _view(message_id: str, finished: ToolFinished, uri: str) -> BaseEvent: - """A view activity, minus the data — see :func:`_filled`. +def consumed_content( + arguments: Mapping[str, Any], + finished: ToolFinished, + state: Mapping[str, StateEntry] | None, +) -> dict[str, Any] | None: + """The ``state.consumed`` payload, or ``None`` when nothing was supplied. - Built where the tool finishes, so its ``messageId`` keeps the run's - ordering, but not sent from there. A view is written against its tool's own - return, and :func:`~mcp_state.restore_structured` rebuilds that from the - artifact plus whatever capture moved into state — and this tool's own - writes reach state on the *next* event. Filling it here would hand a view - the fields small enough to have stayed behind and silently drop the ones - it exists to render. + Shared by the live stream and by a thread readback, so a receipt rendered + now and the same receipt rendered after a reload cannot say different + things. ``arguments`` are the model's, from the call this answers. """ - return _activity( - message_id, - MCP_VIEW, - { - "toolCallId": finished.id, - "tool": finished.name, - "uri": uri, - # Carried so the fill below can reach it; stripped before sending. - "_artifact": finished.artifact, - "display": f"view: {uri}", + if not finished.received: + return None + rendered = step_input(dict(arguments), finished, dict(state or {})) + return { + "toolCallId": finished.id, + "tool": finished.name, + "received": { + parameter: {**receipt, "display": str(rendered.get(parameter, ""))} + for parameter, receipt in finished.received.items() }, - ) + } + + +def published_content(finished: ToolFinished) -> dict[str, Any] | None: + """The ``state.published`` payload, or ``None`` when the tool wrote nothing.""" + if not finished.published: + return None + return { + "toolCallId": finished.id, + "tool": finished.name, + "published": dict(finished.published), + "display": "→ " + ", ".join(sorted(finished.published.values())), + } -def _filled( - view: BaseEvent, state: Mapping[str, StateEntry] | None -) -> BaseEvent | None: - """The same view activity with the data its bundle renders, or ``None``. +def view_content( + finished: ToolFinished, uri: str, state: Mapping[str, StateEntry] | None +) -> dict[str, Any] | None: + """The ``mcp.view`` payload, or ``None`` when there is nothing to render. - A view cannot draw a geometry it was not given, so this is the one place - the wire carries a payload rather than a description of one — the trade - the ``ui://`` contract makes, and why a view's own tool decides how much - it returns. + A view is written against its tool's own return, which + :func:`~mcp_state.restore_structured` rebuilds from the artifact plus + whatever capture moved into state. ``None`` is what a tool call that raised + leaves behind: no structured content, so no view — announcing one anyway + gives a client a panel it can only draw empty, beside the error explaining + why. - ``None`` when there is nothing to render, which is what a tool call that - raised leaves behind: no structured content, so no view. Announcing one - anyway gives a client a panel it can only draw empty, next to the error - that explains why. + **The state has to be the state after that tool's own writes landed.** They + reach the graph on the event *after* the tool finishes, which is why the + live stream holds a view back rather than filling it where it is built. """ - content = dict(getattr(view, "content", {}) or {}) - artifact = content.pop("_artifact", None) - data = restore_structured(artifact, dict(state or {})) + data = restore_structured(finished.artifact, dict(state or {})) if data is None: return None - content["data"] = data - return _activity(str(getattr(view, "message_id", "")), MCP_VIEW, content) + return { + "toolCallId": finished.id, + "tool": finished.name, + "uri": uri, + "display": f"view: {uri}", + "data": data, + } + + +def citations_content(citations: Sequence[str]) -> dict[str, Any] | None: + """The ``answer.citations`` payload, or ``None`` when the answer cited none. + + A mapping rather than a bare list: ``ActivityMessage.content`` is an object, + so an array would not survive being read back. + """ + if not citations: + return None + ids = list(citations) + return {"ids": ids, "display": "Sources: " + ", ".join(ids)} async def agui_events( @@ -259,9 +266,10 @@ async def agui_events( #: The open answer message, or None when none is open. Holding the id #: rather than a flag means every close has the id it needs. open_message: str | None = None - #: View activities waiting for the state their tool just wrote. See - #: :func:`_view` for why they cannot go out where they are built. - deferred: list[BaseEvent] = [] + #: Views waiting for the state their tool just wrote, as + #: ``(message id, tool, uri)``. See :func:`view_content` for why they + #: cannot go out where they are built. + deferred: list[tuple[str, ToolFinished, str]] = [] sequence = 0 def next_id(prefix: str) -> str: @@ -272,7 +280,11 @@ def next_id(prefix: str) -> str: def ready() -> list[BaseEvent]: """Take the deferred views, filled, dropping any with nothing to draw.""" taken, deferred[:] = list(deferred), [] - return [found for view in taken if (found := _filled(view, state)) is not None] + return [ + _activity(message_id, MCP_VIEW, content) + for message_id, finished, uri in taken + if (content := view_content(finished, uri, state)) is not None + ] try: if withheld: @@ -320,32 +332,15 @@ def ready() -> list[BaseEvent]: tool_call_id=event.id, content=event.content, ) - if event.received: - yield _activity( - next_id("act"), - STATE_CONSUMED, - { - "toolCallId": event.id, - "tool": event.name, - "received": _received( - calls.get(event.id), event, state - ), - }, - ) - if event.published: - yield _activity( - next_id("act"), - STATE_PUBLISHED, - { - "toolCallId": event.id, - "tool": event.name, - "published": dict(event.published), - "display": "→ " - + ", ".join(sorted(event.published.values())), - }, - ) + started = calls.get(event.id) + if consumed := consumed_content( + started.arguments if started else {}, event, state + ): + yield _activity(next_id("act"), STATE_CONSUMED, consumed) + if published := published_content(event): + yield _activity(next_id("act"), STATE_PUBLISHED, published) if uri := view_uri_for((tools or {}).get(event.name)): - deferred.append(_view(next_id("act"), event, uri)) + deferred.append((next_id("act"), event, uri)) case StateChanged(): state = dict(event.state) @@ -387,16 +382,8 @@ def ready() -> list[BaseEvent]: # reducer has assigned write order. if event.result.sidecar: yield _state_snapshot(event.result.sidecar) - if event.result.citations: - yield _activity( - next_id("act"), - ANSWER_CITATIONS, - { - "ids": list(event.result.citations), - "display": "Sources: " - + ", ".join(event.result.citations), - }, - ) + 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 # Close an open message first: RUN_ERROR while a text message is # still open is rejected by the client's verifier. diff --git a/src/mcp_agent_api/routes.py b/src/mcp_agent_api/routes.py index cb67776..2dca043 100644 --- a/src/mcp_agent_api/routes.py +++ b/src/mcp_agent_api/routes.py @@ -11,7 +11,10 @@ ``POST /runs`` One turn, streamed as Server-Sent Events. The whole conversation is here. ``GET /threads/{thread_id}`` - The thread's messages, so a page reload restores it. + The thread's messages *and its activities*, so a page reload restores + what the agent was seen to do and not just what was said. Receipts and the + captured-key map are on each tool message's artifact, which the checkpointer + keeps, so they are read back rather than re-derived. ``GET /threads/{thread_id}/turns`` The thread's turns, and what session state held at the end of each. The state channel is cumulative, so this is what says which keys a *particular* @@ -63,6 +66,7 @@ from typing import Any, Protocol, cast from ag_ui.core import ( + ActivityMessage, AssistantMessage, Event, FunctionCall, @@ -79,14 +83,26 @@ from langchain_core.tools import BaseTool from pydantic import BaseModel, ConfigDict, Field -from mcp_agent.host import view_bundles +from mcp_agent.host import view_bundles, view_uri_for from mcp_agent.main import ( + answer_citations, new_thread_id, resolve_credentials, user_credentials, ) -from mcp_agent.streaming import stream_turn -from mcp_agent_api.events import agui_events, state_metadata +from mcp_agent.streaming import stream_turn, tool_finished +from mcp_agent_api.events import ( + ANSWER_CITATIONS, + MCP_VIEW, + STATE_CONSUMED, + STATE_PUBLISHED, + agui_events, + citations_content, + consumed_content, + published_content, + state_metadata, + view_content, +) from mcp_agent_api.history import Turn, turns_of from mcp_state.state import TOOL_STATE_KEY, StateEntry from mcp_state.wiring import Unsatisfiable @@ -158,11 +174,12 @@ class RunRequest(BaseModel): containing the activity messages *this server* generated still validates. The protocol requires an ``id`` on every message, and so do we by using its - models. Nothing here reads it: history is the server's, only the trailing - user message's text is taken, and the id a client posts is discarded rather - than stored. It is required because the protocol says so — a client - assembling ``TEXT_MESSAGE_*`` deltas needs ids on the messages it receives — - and any string will do, a fresh uuid per message being the obvious choice. + models. The trailing user message's id is **kept**: it becomes the id the + thread stores for that question, so a client's own copy and anything it + reads back later are the same message. Every other id is ignored, because + every other message is. An id the thread already holds is ignored too — + see :func:`~mcp_agent.streaming.stream_turn`. A fresh uuid per message is + the obvious choice. """ model_config = ConfigDict(populate_by_name=True) @@ -323,21 +340,98 @@ def _as_message(message: BaseMessage, fallback_id: str) -> Message | None: return None -def thread_messages(history: Iterable[BaseMessage]) -> list[Message]: - """A thread's transcript as AG-UI messages, in order. - - Past turns' activities are not rebuilt. Doing so would mean re-deriving - every historical turn's receipts and re-resolving its view props at the - moment a page is trying to load; the state and view routes are how a client - recovers a past turn's geometry or view. +def _activities_for( + call: BaseMessage | None, + result: BaseMessage, + state: dict[str, StateEntry], + tools: Mapping[str, BaseTool], + position: int, +) -> list[Message]: + """One tool result's activities, rebuilt from what the thread already holds. + + Nothing here is re-derived. Capture wrote the receipts and the captured-key + map onto the tool message's artifact and the checkpointer kept them, so + these are the same payloads the stream sent, through the same builders — + which is what stops a reloaded receipt and a live one saying different + things. """ + finished = tool_finished(result) + arguments = next( + ( + dict(item.get("args") or {}) + for item in getattr(call, "tool_calls", None) or [] + if str(item.get("id") or "") == finished.id + ), + {}, + ) + built: list[tuple[str, dict[str, Any] | None]] = [ + (STATE_CONSUMED, consumed_content(arguments, finished, state)), + (STATE_PUBLISHED, published_content(finished)), + ] + # The URI comes from the deployment as it stands, not from the turn: a tool + # removed since has no bundle to point at, and a dangling `ui://` is worse + # than no view. + if uri := view_uri_for(tools.get(finished.name)): + built.append((MCP_VIEW, view_content(finished, uri, state))) return [ - converted - for position, message in enumerate(history) - if (converted := _as_message(message, f"msg_{position}")) is not None + ActivityMessage( + id=f"act_{position}_{index}", activity_type=kind, content=content + ) + for index, (kind, content) in enumerate(built) + if content is not None ] +def thread_messages( + history: Iterable[BaseMessage], + *, + turns: Sequence[Turn] = (), + tools: Mapping[str, BaseTool] | None = None, +) -> list[Message]: + """A thread's transcript as AG-UI messages, in order, activities included. + + An activity *is* a message in AG-UI, so a receipt read back sits beside the + call it belongs to with no correlation work — the same property the live + stream relies on. + + ``turns`` supplies the session state each turn ended with, because a value + a later turn overwrote would otherwise describe an earlier turn's call with + the wrong value. A turn the checkpointer has since pruned has no state, and + its views are simply not rebuilt: the alternative is drawing one from + whatever state happens to be current, which is worse than drawing none. + Omit ``turns`` and ``tools`` for the bare transcript. + + One approximation worth knowing: state is per *turn*, not per call, so two + tools writing the same key within one turn describe each other's value. The + checkpoints could tell them apart; the turn index cannot. + """ + by_tool = dict(tools or {}) + restored: list[Message] = [] + call: BaseMessage | None = None + turn = 0 + for position, message in enumerate(history): + kind = getattr(message, "type", None) + if kind == "human": + turn += 1 + if (converted := _as_message(message, f"msg_{position}")) is None: + continue + restored.append(converted) + if kind == "ai": + call = message + if cited := citations_content(answer_citations(message)): + restored.append( + ActivityMessage( + id=f"act_{position}_c", + activity_type=ANSWER_CITATIONS, + content=cited, + ) + ) + elif kind == "tool": + state = next((t.state for t in turns if t.n == turn), {}) + restored.extend(_activities_for(call, message, state, by_tool, position)) + return restored + + class ViewCache: """The deployment's ``ui://`` bundles, read once. @@ -539,13 +633,27 @@ async def frames() -> AsyncIterator[str]: @router.get("/threads/{thread_id}", responses={200: {"model": ThreadResponse}}) async def read_thread(thread_id: str) -> dict[str, Any]: - """The thread's messages, for a client restoring a conversation.""" + """The thread's messages and activities, for a client restoring it. + + Two reads, not one: the transcript, and the per-turn state a receipt or + a view has to be described against. That second walk is proportional to + the length of the conversation — the documented cost of deriving turns + from a structure that does not record them — and it is what makes the + difference between restoring a conversation and restoring what the + agent was seen to do. + """ + agent = built() values = await thread_values(thread_id) + past = await turns_of(agent.agent, thread_id) return { "threadId": thread_id, "messages": [ message.model_dump(by_alias=True, exclude_none=True) - for message in thread_messages(values["messages"]) + for message in thread_messages( + values["messages"], + turns=past.turns, + tools={tool.name: tool for tool in agent.tools}, + ) ], # The same shape the turn's STATE_SNAPSHOT carries, so a restored # thread knows what is in state without a second convention. diff --git a/tests/mcp_agent_api/test_routes.py b/tests/mcp_agent_api/test_routes.py index 393997f..3979ec5 100644 --- a/tests/mcp_agent_api/test_routes.py +++ b/tests/mcp_agent_api/test_routes.py @@ -399,12 +399,56 @@ async def test_a_thread_reads_back_as_messages(): roles = [message["role"] for message in thread["messages"]] assert roles[0] == "user" assert roles.count("tool") == 2 - # Past turns' activities are not rebuilt; the state and view routes are how - # a reloaded page recovers what they carried. - assert "activity" not in roles assert thread["state"][STATE_KEY]["kind"] == "geojson.AreaOfInterest" +async def test_a_reload_gets_the_activities_back_too(): + """The transcript alone says what was said. The activities are where a tool's + arguments came from and which view drew its result — the things this runtime + exists to make visible, and the reason a restored thread is worth having.""" + async with _client() as client: + live = await _run(client, threadId="t1") + thread = (await client.get("/threads/t1")).json() + + def keyed(pairs: Any) -> dict[tuple[str, str], Any]: + return { + (kind, content.get("toolCallId", "")): content for kind, content in pairs + } + + restored = keyed( + (message["activityType"], message["content"]) + for message in thread["messages"] + if message["role"] == "activity" + ) + assert {kind for kind, _ in restored} == {"state.consumed", "state.published"} + + # Identical to what the stream sent, because both go through the same + # builders — a receipt that read differently after a reload would be worse + # than one that was missing. + streamed = keyed( + (event["activityType"], event["content"]) + for event in live + if event["type"] == "ACTIVITY_SNAPSHOT" + ) + for key, content in restored.items(): + assert content == streamed[key], key + + +async def test_a_restored_activity_sits_beside_the_call_it_belongs_to(): + """An activity *is* a message in AG-UI, so position is the correlation — the + same property the live stream relies on.""" + async with _client() as client: + await _run(client, threadId="t1") + thread = (await client.get("/threads/t1")).json() + + roles = [message["role"] for message in thread["messages"]] + for index, message in enumerate(thread["messages"]): + if message["role"] != "activity": + continue + # Every activity here belongs to a tool result, and follows it. + assert "tool" in roles[:index] + + async def test_a_tool_call_survives_the_round_trip(): """AG-UI carries a call's arguments as a JSON string, not as an object.""" async with _client() as client: From 440a9da6e57a7d13e91a3ec7094a8fbfde8ad114 Mon Sep 17 00:00:00 2001 From: Ciaran Sweet Date: Mon, 17 Aug 2026 12:05:05 +0100 Subject: [PATCH 2/2] fix(api): keep activities out of a run's closing snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit They duplicated. A client applying a MESSAGES_SNAPSHOT keeps every local activity whether the snapshot names it or not — defaultApplyEvents exempts the role — so a snapshot can never correct a client's activities, only append a second copy of ones the stream just sent. Readback ids are per-position and the stream's are per-run, so nothing matches and every one duplicates at the end of the turn. Omitting turns and tools was not enough to prevent it: state.consumed and state.published need neither, so they came out anyway, rendered against no state at all. Now an explicit `activities` flag, off by default, and the read route is the one caller that asks for them. Co-Authored-By: Claude Opus 5 --- src/mcp_agent_api/routes.py | 28 ++++++++++++++++++++++------ tests/mcp_agent_api/test_routes.py | 14 ++++++++++++++ 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/src/mcp_agent_api/routes.py b/src/mcp_agent_api/routes.py index 2dca043..de6c180 100644 --- a/src/mcp_agent_api/routes.py +++ b/src/mcp_agent_api/routes.py @@ -385,21 +385,34 @@ def _activities_for( def thread_messages( history: Iterable[BaseMessage], *, + activities: bool = False, turns: Sequence[Turn] = (), tools: Mapping[str, BaseTool] | None = None, ) -> list[Message]: - """A thread's transcript as AG-UI messages, in order, activities included. + """A thread's transcript as AG-UI messages, in order. - An activity *is* a message in AG-UI, so a receipt read back sits beside the - call it belongs to with no correlation work — the same property the live - stream relies on. + With ``activities``, the receipts, publications, views and citations come + too. An activity *is* a message in AG-UI, so a receipt read back sits + beside the call it belongs to with no correlation work — the same property + the live stream relies on. + + ``activities`` is off by default, and a caller wanting them owes ``turns`` + and ``tools`` as well — without those the receipts are rendered against no + state and the views cannot be found, which is a worse answer than no + activity at all. + + **A run's own closing snapshot wants them off.** A client applying a + snapshot keeps every local activity regardless of whether the snapshot + names it — ``defaultApplyEvents`` exempts the role — so a snapshot can + never *correct* a client's activities, only append a second copy of ones it + already has. These ids are per-position and the stream's are per-run, so + they never match and every one would duplicate. ``turns`` supplies the session state each turn ended with, because a value a later turn overwrote would otherwise describe an earlier turn's call with the wrong value. A turn the checkpointer has since pruned has no state, and its views are simply not rebuilt: the alternative is drawing one from whatever state happens to be current, which is worse than drawing none. - Omit ``turns`` and ``tools`` for the bare transcript. One approximation worth knowing: state is per *turn*, not per call, so two tools writing the same key within one turn describe each other's value. The @@ -418,6 +431,8 @@ def thread_messages( restored.append(converted) if kind == "ai": call = message + if not activities: + continue if cited := citations_content(answer_citations(message)): restored.append( ActivityMessage( @@ -426,7 +441,7 @@ def thread_messages( content=cited, ) ) - elif kind == "tool": + elif kind == "tool" and activities: state = next((t.state for t in turns if t.n == turn), {}) restored.extend(_activities_for(call, message, state, by_tool, position)) return restored @@ -651,6 +666,7 @@ async def read_thread(thread_id: str) -> dict[str, Any]: message.model_dump(by_alias=True, exclude_none=True) for message in thread_messages( values["messages"], + activities=True, turns=past.turns, tools={tool.name: tool for tool in agent.tools}, ) diff --git a/tests/mcp_agent_api/test_routes.py b/tests/mcp_agent_api/test_routes.py index 3979ec5..d001907 100644 --- a/tests/mcp_agent_api/test_routes.py +++ b/tests/mcp_agent_api/test_routes.py @@ -434,6 +434,20 @@ def keyed(pairs: Any) -> dict[tuple[str, str], Any]: assert content == streamed[key], key +async def test_the_closing_snapshot_carries_no_activities(): + """A client keeps every local activity when it applies a snapshot — + `defaultApplyEvents` exempts the role — so a snapshot can never correct + them, only append a second copy of ones the stream just sent. These ids are + per-position and the stream's per-run, so every one would duplicate.""" + async with _client() as client: + events = await _run(client, threadId="t1") + + snapshot = [e for e in events if e["type"] == "MESSAGES_SNAPSHOT"][-1] + assert [m["role"] for m in snapshot["messages"]].count("activity") == 0 + # And the stream did send some, so this is not passing by producing none. + assert any(e["type"] == "ACTIVITY_SNAPSHOT" for e in events) + + async def test_a_restored_activity_sits_beside_the_call_it_belongs_to(): """An activity *is* a message in AG-UI, so position is the correlation — the same property the live stream relies on."""