feat(api): read a thread back with its activities, not just its messages

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 |

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}`,

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));

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

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.