Adopt immutable TanStack client snapshots in the SDK

platform/src/routes/docs/-content/sdk/headless.md

@@ -56,6 +56,8 @@ chat.dispose()
 
 ## Reading the transcript
 
+`messages` is a readonly snapshot. Treat its messages, parts, and metadata as immutable, and use session actions to change the conversation. Derive display values with methods such as `map`, `filter`, or `toSorted` instead of mutating the transcript.
+
 The core exports the part helpers the widget itself renders with.
 
 - `isSandboxTool`, `readSandboxFileWrite`, `readSandboxCommandRun`, `readSandboxArtifact`, `collectSandboxActivity`.

sdk/src/core/messages.ts

@@ -10,7 +10,7 @@ export function isSettledToolCall(part: { state: string; output?: unknown }): bo
 // A busy stream can be silent for a while (server-side reasoning, follow-ups after tool
 // results), so a "Thinking…" affordance shows until the newest assistant part visibly
 // makes progress.
-export function lastPartInProgress(messages: UIMessage[]): boolean {
+export function lastPartInProgress(messages: readonly UIMessage[]): boolean {
   const lastMessage = messages.at(-1)
   const lastPart = lastMessage?.role === "assistant" ? lastMessage.parts.at(-1) : undefined
   return (
@@ -25,7 +25,10 @@ export function lastPartInProgress(messages: UIMessage[]): boolean {
 // the message goes unanswered, and the redelivered call can re-execute a side-effecting
 // tool — so those windows count as busy too. Questionnaires and calls to tools this mount
 // never implemented stay interactive: the pre-send settle resolves them.
-export function hasPendingToolRun(messages: UIMessage[], toolNames: ReadonlySet<string>): boolean {
+export function hasPendingToolRun(
+  messages: readonly UIMessage[],
+  toolNames: ReadonlySet<string>,
+): boolean {
   return messages.some((message) =>
     message.parts.some(
       (part) =>

sdk/src/core/sandbox.ts

@@ -123,7 +123,7 @@ export function readSandboxCommandRun(part: ToolCallPart): SandboxCommandRun | u
  * The whole conversation's sandbox work, for the panel. A path written more than once appears
  * once, holding its latest content, because the panel answers "what is in the sandbox now".
  */
-export function collectSandboxActivity(messages: UIMessage[]): SandboxActivity {
+export function collectSandboxActivity(messages: readonly UIMessage[]): SandboxActivity {
   const files = new Map<string, SandboxFileWrite>()
   const commands: SandboxCommandRun[] = []
   for (const message of messages) {

sdk/src/core/session-persistence.test.ts

@@ -1714,31 +1714,58 @@ test.each(["before", "after"] as const)(
   },
 )
 
-test("a session restarted like React Strict Mode keeps loading history and sending", async () => {
-  let sends = 0
-  stubSessionFetch((input: string | URL, init?: RequestInit) => {
-    const path = new URL(input).pathname
-    if (path.endsWith("/config")) return sessionResponse({ attachments: true })
-    if (path.endsWith("/threads") && init?.method === "POST")
-      return sessionResponse({ ...thread, title: "", version: 0 })
-    if (path.endsWith("/chat")) return Promise.resolve(acceptedStream(String(++sends)))
-    return sessionResponse(page())
-  })
-  const chat = createAstralBeamChat({ fetchAstralBeamToken: chatTestToken, threadId: "new" }, true)
-  try {
-    chat.start()
-    chat.dispose()
-    chat.start()
-    await vi.waitFor(() => expect(chat.getState().auth.status).toBe("ready"))
-    for (const text of ["first", "second"]) {
-      expect(await chat.sendMessage(text)).toMatchObject({ phase: "accepted" })
-      expect(chat.getState().history).toBe("idle")
+test.each(["new", "pending"])(
+  "a %s session restarted like React Strict Mode keeps loading history and sending",
+  async (initial) => {
+    let sends = 0
+    let reads = 0
+    let cancelled = false
+    stubSessionFetch((input: string | URL, init?: RequestInit) => {
+      const path = new URL(input).pathname
+      if (path.endsWith("/config")) return sessionResponse({ attachments: true })
+      if (path.endsWith("/threads") && init?.method === "POST")
+        return sessionResponse({ ...thread, title: "", version: 0 })
+      if (path.endsWith("/chat")) return Promise.resolve(acceptedStream(String(++sends)))
+      if (path.endsWith("/messages") && ++reads === 1 && initial === "pending") {
+        const signal = init!.signal!
+        return new Promise((_resolve, reject) =>
+          signal.addEventListener(
+            "abort",
+            () => {
+              cancelled = true
+              reject(new DOMException("History request cancelled", "AbortError"))
+            },
+            { once: true },
+          ),
+        )
+      }
+      return sessionResponse(page())
+    })
+    const chat = createAstralBeamChat(
+      { fetchAstralBeamToken: chatTestToken, threadId: initial === "new" ? "new" : thread.id },
+      true,
+    )
+    try {
+      chat.start()
+      if (initial !== "new") await vi.waitUntil(() => reads === 1)
+      chat.dispose()
+      expect(cancelled).toBe(initial === "pending")
+      chat.start()
+      await vi.waitFor(() => {
+        expect(chat.getState().auth.status).toBe("ready")
+        expect(chat.getState().history).toBe("idle")
+        expect(chat.getState().thread?.id).toBe(initial === "new" ? undefined : thread.id)
+      })
+      for (const text of ["first", "second"]) {
+        expect(await chat.sendMessage(text)).toMatchObject({ phase: "accepted" })
+        expect(chat.getState().history).toBe("idle")
+      }
+      expect(sends).toBe(2)
+    } finally {
+      chat.dispose()
     }
-    expect(sends).toBe(2)
-  } finally {
-    chat.dispose()
-  }
-})
+  },
+)
 
 test("a send before the configured conversation opens returns to its caller", async () => {
   const sends = vi.fn()
@@ -1756,14 +1783,23 @@ test("a send before the configured conversation opens returns to its caller", as
   expect(chat.getState().submission).toBeUndefined()
 })
 
-test.each(["history", "thread"] as const)(
-  "Stop returns a send still waiting for its %s without dispatching it",
+test.each(["history", "thread", "restarted history"] as const)(
+  "a send waiting for %s remains undispatched until its wait completes or Stop refuses it",
   async (wait) => {
+    const restarting = wait === "restarted history"
+    const waitingForHistory = wait !== "thread"
     const sends = vi.fn()
-    let release: (() => void) | undefined
-    const hold = (response: Response) =>
-      new Promise<Response>((resolve) => {
-        release = () => resolve(response)
+    const release: Array<() => void> = []
+    const hold = (response: Response, signal?: AbortSignal | null) =>
+      new Promise<Response>((resolve, reject) => {
+        release.push(() => resolve(response))
+        signal?.addEventListener(
+          "abort",
+          () => reject(new DOMException("Cancelled", "AbortError")),
+          {
+            once: true,
+          },
+        )
       })
     let reads = 0
     stubSessionFetch((input: string | URL, init?: RequestInit) => {
@@ -1776,23 +1812,38 @@ test.each(["history", "thread"] as const)(
       if (path.endsWith("/threads") && init?.method === "POST")
         return hold(Response.json({ ...thread, title: "", version: 0 }))
       const response = Response.json(page())
-      return wait === "history" && ++reads === 2 ? hold(response) : Promise.resolve(response)
+      return waitingForHistory && (++reads === 2 || (restarting && reads === 3))
+        ? hold(response, restarting ? init?.signal : undefined)
+        : Promise.resolve(response)
     })
-    const chat = createTestSession(wait === "history" ? { threadId: thread.id } : {})
+    const chat = createTestSession(waitingForHistory ? { threadId: thread.id } : {})
     await vi.waitFor(() => expect(chat.getState().auth.status).toBe("ready"))
-    if (wait === "history") {
+    if (waitingForHistory) {
       await vi.waitUntil(() => chat.getState().thread)
       void chat.refreshThread()
     }
-    expect(chat.getState().history).toBe(wait === "history" ? "refreshing" : "idle")
-    const outcome = chat.sendMessage("Stop me")
-    await vi.waitFor(() => expect(release).toBeDefined())
-    chat.stop()
-    expect(chat.getState()).toMatchObject({ status: "ready", submission: { phase: "refused" } })
-    release!()
-    expect(await outcome).toEqual(expect.objectContaining({ phase: "refused", content: "Stop me" }))
-    expect((await outcome).error).toBeUndefined()
+    if (restarting) {
+      await vi.waitUntil(() => release.length === 1)
+      chat.dispose()
+      chat.start()
+    }
+    expect(chat.getState().history).toBe(waitingForHistory ? "refreshing" : "idle")
+    const outcome = chat.sendMessage("Wait for me")
+    await vi.waitFor(() => expect(release).toHaveLength(restarting ? 2 : 1))
+    await new Promise((resolve) => setTimeout(resolve, 0))
     expect(sends).not.toHaveBeenCalled()
+    if (!restarting) chat.stop()
+    expect(chat.getState()).toMatchObject({
+      status: restarting ? "submitted" : "ready",
+      submission: { phase: restarting ? "submitting" : "refused" },
+    })
+    release.at(-1)!()
+    expect(await outcome).toMatchObject({
+      phase: restarting ? "accepted" : "refused",
+      content: "Wait for me",
+    })
+    expect((await outcome).error).toBeUndefined()
+    expect(sends).toHaveBeenCalledTimes(restarting ? 1 : 0)
   },
 )
 

sdk/src/core/session.test.ts

@@ -171,17 +171,33 @@ test("the reply after a server tool round keeps its first text delta", async ()
   const chat = createTestSession({
     streamCallbacks: { onFinish },
   })
-  const statuses: string[] = []
-  chat.subscribe(() => statuses.push(chat.getState().status))
+  const snapshots: ReturnType<typeof chat.getState>[] = []
+  chat.subscribe(() => snapshots.push(chat.getState()))
   await chat.sendMessage("Write a file")
   expect(onFinish.mock.calls.at(-1)?.[0].parts.at(-1)).toEqual({
     type: "text",
     content: "I wrote it.",
   })
   // The tool round's intermediate RUN_FINISHED must not make the widget look idle mid-turn.
+  const statuses = snapshots.map((snapshot) => snapshot.status)
   expect(
     statuses.slice(statuses.indexOf("streaming"), statuses.lastIndexOf("streaming")),
   ).not.toContain("ready")
+  const firstDelta = snapshots.find((snapshot) =>
+    snapshot.messages.some((message) =>
+      message.parts.some((part) => part.type === "text" && part.content === "I"),
+    ),
+  )!
+  const reply = snapshots.findLast((snapshot) => snapshot.messages.at(-1)?.id === "reply")!
+  expect(firstDelta.messages.at(-1)?.parts).toContainEqual({ type: "text", content: "I" })
+  expect(firstDelta.messages.find((message) => message.id === "call-turn")).toBeDefined()
+  expect(reply.messages.find((message) => message.id === "call-turn")).toBe(
+    firstDelta.messages.find((message) => message.id === "call-turn"),
+  )
+  expect(reply.status).toBe("ready")
+  expect(reply.messages).toBe(
+    snapshots.findLast((snapshot) => snapshot.status === "streaming")?.messages,
+  )
 })
 
 // A React host rebuilds its tool objects every render, so publishing a fresh `agentTools` for an