From e14f91f9f904d19fb83d9b64ade656c8a3803413 Mon Sep 17 00:00:00 2001 From: Aayush <21987529+aayushbtw@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:53:59 +0400 Subject: [PATCH 1/5] fix(ai-client): keep the replay's answer when a rejoin starts with encrypted reasoning --- .changeset/rejoin-keeps-replay-message.md | 5 + packages/ai-client/src/chat-client.ts | 27 ++-- .../chat-client-reasoning-rejoin.test.ts | 115 ++++++++++++++++++ 3 files changed, 138 insertions(+), 9 deletions(-) create mode 100644 .changeset/rejoin-keeps-replay-message.md diff --git a/.changeset/rejoin-keeps-replay-message.md b/.changeset/rejoin-keeps-replay-message.md new file mode 100644 index 0000000000..2f3f531896 --- /dev/null +++ b/.changeset/rejoin-keeps-replay-message.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-client': patch +--- + +Rejoining a run no longer loses the answer when the replay starts with encrypted-only reasoning, for example from Azure OpenAI Responses. The client now drops only the assistant message it hydrated before the replay, not the message the replay has already started. diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 3085d2f526..7db869bd7d 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -2155,6 +2155,12 @@ export class ChatClient< const connectTimer = setTimeout(() => { if (!attached) controller.abort() }, REJOIN_CONNECT_DEADLINE_MS) + // Captured before the replay: chunks before the first rebuild trigger + // (reasoning start with metadata) can already append the replay's own + // assistant message, which must survive the drop. + const hydrated = this.processor.getMessages().at(-1) + const hydratedAssistantId = + hydrated?.role === 'assistant' ? hydrated.id : undefined try { for await (const chunk of joinRun(runId, controller.signal)) { if (controller.signal.aborted) break @@ -2164,7 +2170,9 @@ export class ChatClient< } if (!rebuilt && rebuildsAssistantMessage(chunk)) { rebuilt = true - this.dropTrailingInFlightAssistant() + if (hydratedAssistantId) { + this.dropHydratedAssistant(hydratedAssistantId) + } } this.processIncomingChunk(chunk) } @@ -2222,16 +2230,17 @@ export class ChatClient< } /** - * Drop a hydrated, still-in-flight assistant turn so a resume replay can - * rebuild it cleanly. Only touches a trailing assistant message (the shape a - * reload-mid-stream leaves); a thread whose last turn is a user message (run - * never produced, or already settled) is left untouched. + * Drop the hydrated, still-in-flight assistant turn so a resume replay can + * rebuild it cleanly. `id` is the trailing assistant message from before the + * replay (the shape a reload-mid-stream leaves); a thread whose last turn was + * a user message (run never produced, or already settled) has none to drop. */ - private dropTrailingInFlightAssistant(): void { + private dropHydratedAssistant(id: string): void { const messages = this.processor.getMessages() - const last = messages[messages.length - 1] - if (last && last.role === 'assistant') { - this.processor.setMessages(messages.slice(0, -1)) + if (messages.some((message) => message.id === id)) { + this.processor.setMessages( + messages.filter((message) => message.id !== id), + ) } } diff --git a/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts b/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts index e6ef5351fc..ac714c008a 100644 --- a/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts +++ b/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts @@ -151,3 +151,118 @@ it.each(firstEvents)( } }, ) + +const hydratedTails: Array<{ name: string; tail: Array }> = [ + { name: 'a user turn', tail: [] }, + { + name: 'an in-flight assistant', + tail: [ + { + id: 'partial', + role: 'assistant', + parts: [{ type: 'text', content: 'Hel' }], + }, + ], + }, +] + +it.each(hydratedTails)( + 'keeps the answer after encrypted-only reasoning when hydrate ends with $name', + async ({ tail }) => { + const metadata = { tanstack: { model: 'example-model' } } + const replay: Array = [ + { + type: EventType.RUN_STARTED, + runId: 'r1', + threadId: 't1', + timestamp: 1, + }, + { + type: EventType.REASONING_START, + messageId: 'reasoning', + metadata, + timestamp: 2, + }, + { + type: EventType.REASONING_MESSAGE_START, + messageId: 'reasoning', + role: 'reasoning', + metadata, + timestamp: 3, + }, + { + type: EventType.REASONING_MESSAGE_END, + messageId: 'reasoning', + timestamp: 4, + }, + { type: EventType.REASONING_END, messageId: 'reasoning', timestamp: 5 }, + { + type: EventType.REASONING_ENCRYPTED_VALUE, + subtype: 'message', + entityId: 'reasoning', + encryptedValue: 'signature', + timestamp: 6, + }, + { + type: EventType.TEXT_MESSAGE_START, + messageId: 'answer', + role: 'assistant', + timestamp: 7, + }, + { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'answer', + delta: 'Hello from Acme.', + timestamp: 8, + }, + { type: EventType.TEXT_MESSAGE_END, messageId: 'answer', timestamp: 9 }, + { + type: EventType.RUN_FINISHED, + runId: 'r1', + threadId: 't1', + timestamp: 10, + }, + ] + const connection: ResumableConnectConnectionAdapter = { + async hydrate() { + return { + messages: [ + { + id: 'u1', + role: 'user', + parts: [{ type: 'text', content: 'Say hello.' }], + }, + ...tail, + ], + activeRun: { runId: 'r1' }, + interrupts: null, + } + }, + async *joinRun() { + yield* replay + }, + async *connect() {}, + } + const client = new ChatClient({ + connection, + threadId: 't1', + persistence: true, + }) + try { + client.attach() + await vi.waitFor(() => { + expect(client.getMessages()).not.toHaveLength(0) + expect(client.getIsLoading()).toBe(false) + }) + const assistants = client + .getMessages() + .filter((message) => message.role === 'assistant') + expect(assistants).toHaveLength(1) + expect(assistants[0]!.parts).toContainEqual( + expect.objectContaining({ type: 'text', content: 'Hello from Acme.' }), + ) + } finally { + client.dispose() + } + }, +) From a5e1bf18cc00bd312bc934624ca8e139bd867cc9 Mon Sep 17 00:00:00 2001 From: Aayush <21987529+aayushbtw@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:14:03 +0400 Subject: [PATCH 2/5] test(ai-client): rejoin tests send reasoning start metadata like chat() One table covers readable and encrypted-only reasoning against a hydrate that ends on a user turn or an in-flight partial. Each case checks one assistant message with the thinking part, the tool call and the answer. --- .../chat-client-reasoning-rejoin.test.ts | 225 +++++++----------- 1 file changed, 80 insertions(+), 145 deletions(-) diff --git a/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts b/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts index ac714c008a..7d6e98217a 100644 --- a/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts +++ b/packages/ai-client/tests/chat-client-reasoning-rejoin.test.ts @@ -4,26 +4,60 @@ import { ChatClient } from '../src/chat-client' import type { ResumableConnectConnectionAdapter } from '../src/connection-adapters' import type { StreamChunk, UIMessage } from '@tanstack/ai/client' -const firstEvents: Array<{ +// chat() stamps this on every adapter chunk. On REASONING_MESSAGE_START it +// makes the processor create the assistant message before any rebuild trigger. +const callMetadata = { + tanstack: { + source: { provider: 'openai', api: 'responses', model: 'example-model' }, + }, +} + +const reasonings: Array<{ name: string chunks: Array - thinking?: string + thinking: { content: string; signature?: string } }> = [ { - name: 'reasoning content', + name: 'readable reasoning', chunks: [ + { type: EventType.REASONING_START, messageId: 'thinking', timestamp: 2 }, + { + type: EventType.REASONING_MESSAGE_START, + messageId: 'thinking', + role: 'reasoning', + timestamp: 2, + }, { type: EventType.REASONING_MESSAGE_CONTENT, messageId: 'thinking', delta: 'Inspect the files.', timestamp: 2, }, + { + type: EventType.REASONING_MESSAGE_END, + messageId: 'thinking', + timestamp: 2, + }, + { type: EventType.REASONING_END, messageId: 'thinking', timestamp: 2 }, ], - thinking: 'Inspect the files.', + thinking: { content: 'Inspect the files.' }, }, { - name: 'encrypted reasoning', + name: 'encrypted-only reasoning', chunks: [ + { type: EventType.REASONING_START, messageId: 'thinking', timestamp: 2 }, + { + type: EventType.REASONING_MESSAGE_START, + messageId: 'thinking', + role: 'reasoning', + timestamp: 2, + }, + { + type: EventType.REASONING_MESSAGE_END, + messageId: 'thinking', + timestamp: 2, + }, + { type: EventType.REASONING_END, messageId: 'thinking', timestamp: 2 }, { type: EventType.REASONING_ENCRYPTED_VALUE, subtype: 'message', @@ -32,13 +66,31 @@ const firstEvents: Array<{ timestamp: 2, }, ], - thinking: '', + thinking: { content: '', signature: 'signature' }, }, ] -it.each(firstEvents)( - 'keeps activity when replay starts with $name', - async ({ chunks, thinking }) => { +const hydratedTails: Array<{ name: string; tail: Array }> = [ + { name: 'a user turn', tail: [] }, + { + name: 'an in-flight assistant', + tail: [ + { + id: 'partial', + role: 'assistant', + parts: [{ type: 'thinking', content: 'Inspect' }], + }, + ], + }, +] + +const cases = reasonings.flatMap((reasoning) => + hydratedTails.map((hydrated) => ({ reasoning, hydrated })), +) + +it.each(cases)( + 'keeps the thread when replay starts with $reasoning.name and hydrate ends with $hydrated.name', + async ({ reasoning, hydrated }) => { const replay: Array = [ { type: EventType.RUN_STARTED, @@ -46,7 +98,7 @@ it.each(firstEvents)( threadId: 't1', timestamp: 1, }, - ...chunks, + ...reasoning.chunks, { type: EventType.TOOL_CALL_START, toolCallId: 'read', @@ -87,6 +139,12 @@ it.each(firstEvents)( timestamp: 10, }, ] + // chat() runs the tool itself, so the tool result is not an adapter chunk. + const stamped = replay.map((chunk) => + chunk.type === EventType.TOOL_CALL_RESULT + ? chunk + : { ...chunk, metadata: callMetadata }, + ) let sent: Array = [] const connection: ResumableConnectConnectionAdapter = { async hydrate() { @@ -97,18 +155,14 @@ it.each(firstEvents)( role: 'user', parts: [{ type: 'text', content: 'Build an app.' }], }, - { - id: 'partial', - role: 'assistant', - parts: [{ type: 'thinking', content: 'Inspect' }], - }, + ...hydrated.tail, ], activeRun: { runId: 'r1' }, interrupts: null, } }, async *joinRun() { - yield* replay + yield* stamped }, async *connect(messages) { sent = messages as Array @@ -122,144 +176,25 @@ it.each(firstEvents)( try { client.attach() await vi.waitFor(() => { - expect( - client.getMessages().flatMap((message) => message.parts), - ).toContainEqual( - expect.objectContaining({ type: 'text', content: 'Built the app.' }), - ) + expect(client.getMessages()).not.toHaveLength(0) + expect(client.getIsLoading()).toBe(false) }) await client.sendMessage('Change the title.') - const parts = sent.flatMap((message) => message.parts) - if (thinking !== undefined) { - expect(parts).toContainEqual( - expect.objectContaining({ type: 'thinking', content: thinking }), - ) - } - expect(parts).toContainEqual( + const assistants = sent.filter((message) => message.role === 'assistant') + expect(assistants).toHaveLength(1) + expect(assistants[0]!.id).not.toBe('partial') + expect(assistants[0]!.parts).toContainEqual( + expect.objectContaining({ type: 'thinking', ...reasoning.thinking }), + ) + expect(assistants[0]!.parts).toContainEqual( expect.objectContaining({ type: 'tool-call', id: 'read', output: 'file contents', }), ) - expect(parts).toContainEqual( - expect.objectContaining({ type: 'text', content: 'Built the app.' }), - ) - expect(sent.some((message) => message.id === 'partial')).toBe(false) - } finally { - client.dispose() - } - }, -) - -const hydratedTails: Array<{ name: string; tail: Array }> = [ - { name: 'a user turn', tail: [] }, - { - name: 'an in-flight assistant', - tail: [ - { - id: 'partial', - role: 'assistant', - parts: [{ type: 'text', content: 'Hel' }], - }, - ], - }, -] - -it.each(hydratedTails)( - 'keeps the answer after encrypted-only reasoning when hydrate ends with $name', - async ({ tail }) => { - const metadata = { tanstack: { model: 'example-model' } } - const replay: Array = [ - { - type: EventType.RUN_STARTED, - runId: 'r1', - threadId: 't1', - timestamp: 1, - }, - { - type: EventType.REASONING_START, - messageId: 'reasoning', - metadata, - timestamp: 2, - }, - { - type: EventType.REASONING_MESSAGE_START, - messageId: 'reasoning', - role: 'reasoning', - metadata, - timestamp: 3, - }, - { - type: EventType.REASONING_MESSAGE_END, - messageId: 'reasoning', - timestamp: 4, - }, - { type: EventType.REASONING_END, messageId: 'reasoning', timestamp: 5 }, - { - type: EventType.REASONING_ENCRYPTED_VALUE, - subtype: 'message', - entityId: 'reasoning', - encryptedValue: 'signature', - timestamp: 6, - }, - { - type: EventType.TEXT_MESSAGE_START, - messageId: 'answer', - role: 'assistant', - timestamp: 7, - }, - { - type: EventType.TEXT_MESSAGE_CONTENT, - messageId: 'answer', - delta: 'Hello from Acme.', - timestamp: 8, - }, - { type: EventType.TEXT_MESSAGE_END, messageId: 'answer', timestamp: 9 }, - { - type: EventType.RUN_FINISHED, - runId: 'r1', - threadId: 't1', - timestamp: 10, - }, - ] - const connection: ResumableConnectConnectionAdapter = { - async hydrate() { - return { - messages: [ - { - id: 'u1', - role: 'user', - parts: [{ type: 'text', content: 'Say hello.' }], - }, - ...tail, - ], - activeRun: { runId: 'r1' }, - interrupts: null, - } - }, - async *joinRun() { - yield* replay - }, - async *connect() {}, - } - const client = new ChatClient({ - connection, - threadId: 't1', - persistence: true, - }) - try { - client.attach() - await vi.waitFor(() => { - expect(client.getMessages()).not.toHaveLength(0) - expect(client.getIsLoading()).toBe(false) - }) - const assistants = client - .getMessages() - .filter((message) => message.role === 'assistant') - expect(assistants).toHaveLength(1) expect(assistants[0]!.parts).toContainEqual( - expect.objectContaining({ type: 'text', content: 'Hello from Acme.' }), + expect.objectContaining({ type: 'text', content: 'Built the app.' }), ) } finally { client.dispose() From e0d87e90d71e035aa1171801e2007ad40f9bba97 Mon Sep 17 00:00:00 2001 From: Aayush <21987529+aayushbtw@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:14:16 +0400 Subject: [PATCH 3/5] refactor(ai-client): drop the hydrated assistant in one pass --- packages/ai-client/src/chat-client.ts | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 7db869bd7d..7bce935951 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -2237,11 +2237,8 @@ export class ChatClient< */ private dropHydratedAssistant(id: string): void { const messages = this.processor.getMessages() - if (messages.some((message) => message.id === id)) { - this.processor.setMessages( - messages.filter((message) => message.id !== id), - ) - } + const kept = messages.filter((message) => message.id !== id) + if (kept.length < messages.length) this.processor.setMessages(kept) } private processIncomingChunk(chunk: StreamChunk): void { From 280bca9259c06338b48d9e99b61850b5f82dcb2b Mon Sep 17 00:00:00 2001 From: Aayush <21987529+aayushbtw@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:14:16 +0400 Subject: [PATCH 4/5] chore(ai-client): changeset names any reasoning-first rejoin --- .changeset/rejoin-keeps-replay-message.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/rejoin-keeps-replay-message.md b/.changeset/rejoin-keeps-replay-message.md index 2f3f531896..78aefdfcf4 100644 --- a/.changeset/rejoin-keeps-replay-message.md +++ b/.changeset/rejoin-keeps-replay-message.md @@ -2,4 +2,4 @@ '@tanstack/ai-client': patch --- -Rejoining a run no longer loses the answer when the replay starts with encrypted-only reasoning, for example from Azure OpenAI Responses. The client now drops only the assistant message it hydrated before the replay, not the message the replay has already started. +Rejoining a run no longer loses the answer when the replay starts with reasoning. A reasoning start event that carries metadata, which `chat()` adds to every adapter event, already creates the replay's assistant message. The client used to drop that message at the first rebuild event. It now drops only the assistant message it hydrated before the replay. From 0056aa31e9b5f6087697daad1f888c2dfc08d2ad Mon Sep 17 00:00:00 2001 From: Aayush <21987529+aayushbtw@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:16:24 +0400 Subject: [PATCH 5/5] test(e2e): a reload mid run that reasons first keeps its answer --- testing/e2e/README.md | 2 +- .../src/routes/api.join-run-client-tool.ts | 46 ++++++++++++++++++- .../e2e/src/routes/join-run-client-tool.tsx | 9 ++++ .../e2e/tests/join-run-client-tool.spec.ts | 27 +++++++++++ 4 files changed, 82 insertions(+), 2 deletions(-) diff --git a/testing/e2e/README.md b/testing/e2e/README.md index e28cb81663..99442940a3 100644 --- a/testing/e2e/README.md +++ b/testing/e2e/README.md @@ -97,7 +97,7 @@ Notes: | -------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- | | `tests/delivery-durability.spec.ts` | Transport layer: offset-tagged log, `Last-Event-ID` reconnect, second-tab join, SSE + NDJSON | | `tests/persistence-durability.spec.ts` | Client layer: a browser refresh restores the conversation and any pending interrupt | -| `tests/join-run-client-tool.spec.ts` | Mid-run reload: `joinRun` replay that ends on a client tool drains the continuation (issue #1058) | +| `tests/join-run-client-tool.spec.ts` | Mid-run reload: `joinRun` replay that ends on a client tool drains the continuation (issue #1058); a replay that starts with reasoning keeps its answer | | `tests/sandbox-durability.spec.ts` | Sandbox instances: a second run resumes the persisted sandbox | | `tests/durable-takeover.spec.ts` | Takeover with log alignment, detach-on-disconnect, out-of-band cancel in both bands, cancel-vs-disconnect divergence, and the superseded-driver epoch fence | diff --git a/testing/e2e/src/routes/api.join-run-client-tool.ts b/testing/e2e/src/routes/api.join-run-client-tool.ts index 81b243ca02..757938ed55 100644 --- a/testing/e2e/src/routes/api.join-run-client-tool.ts +++ b/testing/e2e/src/routes/api.join-run-client-tool.ts @@ -30,6 +30,7 @@ import type { const JOIN_OK = 'JOIN_OK the lookup is 42' const LATER_OK = 'LATER_OK the follow-up ran' +const REASON_OK = 'REASON_OK the answer after reasoning' const DISCONNECT_FALLBACK_MS = 3000 const SETTLE_AFTER_DISCONNECT_MS = 1500 @@ -51,6 +52,8 @@ const parentInterruptByThread = new Map< // User text for the #1429 scenario: the first run pauses on the client tool, // and the continuation run (not the first run) is held open. const CONTINUATION_SCENARIO = 'continue' +// User text for a held run that reasons before it answers. +const REASONING_SCENARIO = 'reason' async function* recordInterrupt( stream: AsyncIterable, @@ -154,7 +157,13 @@ const adapter: AnyTextAdapter = { ) const userText = lastUserText(options.messages) const reply = - userText === 'later' ? LATER_OK : hasToolResult ? JOIN_OK : null + userText === 'later' + ? LATER_OK + : userText === REASONING_SCENARIO + ? REASON_OK + : hasToolResult + ? JOIN_OK + : null yield { type: EventType.RUN_STARTED, @@ -165,6 +174,41 @@ const adapter: AnyTextAdapter = { } if (reply !== null) { + if (userText === REASONING_SCENARIO) { + const reasoningId = `${runId}-reasoning` + yield { + type: EventType.REASONING_START, + messageId: reasoningId, + model, + timestamp: Date.now(), + } + yield { + type: EventType.REASONING_MESSAGE_START, + messageId: reasoningId, + role: 'reasoning', + model, + timestamp: Date.now(), + } + yield { + type: EventType.REASONING_MESSAGE_CONTENT, + messageId: reasoningId, + delta: 'Check the order.', + model, + timestamp: Date.now(), + } + yield { + type: EventType.REASONING_MESSAGE_END, + messageId: reasoningId, + model, + timestamp: Date.now(), + } + yield { + type: EventType.REASONING_END, + messageId: reasoningId, + model, + timestamp: Date.now(), + } + } yield { type: EventType.TEXT_MESSAGE_START, messageId, diff --git a/testing/e2e/src/routes/join-run-client-tool.tsx b/testing/e2e/src/routes/join-run-client-tool.tsx index f8ac438b26..318e3df163 100644 --- a/testing/e2e/src/routes/join-run-client-tool.tsx +++ b/testing/e2e/src/routes/join-run-client-tool.tsx @@ -63,6 +63,14 @@ function JoinRunClientToolPage() { ), ) .join('') + const assistantThinking = messages + .filter((message) => message.role === 'assistant') + .flatMap((message) => + message.parts.flatMap((part) => + part.type === 'thinking' ? [part.content] : [], + ), + ) + .join('') const toolPending = messages.some((message) => message.parts.some((part) => { @@ -88,6 +96,7 @@ function JoinRunClientToolPage() {
{hydrated ?
: null}
{assistantText || 'none'}
+
{assistantThinking || 'none'}
{toolPending ? 'true' : 'false'}
diff --git a/testing/e2e/tests/join-run-client-tool.spec.ts b/testing/e2e/tests/join-run-client-tool.spec.ts index 4b29acb319..7f99c4bd8a 100644 --- a/testing/e2e/tests/join-run-client-tool.spec.ts +++ b/testing/e2e/tests/join-run-client-tool.spec.ts @@ -94,3 +94,30 @@ test.describe('joinRun client-tool continuation (issue #1058)', () => { }) }) }) + +test.describe('joinRun replay that starts with reasoning', () => { + test('a reload mid run keeps the thinking and the answer', async ({ + page, + }) => { + const threadId = `join-run-reasoning-${crypto.randomUUID()}` + await page.goto( + `/join-run-client-tool?threadId=${encodeURIComponent(threadId)}`, + ) + await expect(page.getByTestId('hydration-marker')).toBeAttached() + + await sendMessage(page, 'reason') + // The run is held open after its answer, so the remount sees `activeRun`. + await expect(page.getByTestId('assistant-text')).toContainText('REASON_OK') + + await page.reload() + await expect(page.getByTestId('hydration-marker')).toBeAttached() + + // Hydrate returns no messages, so both can only come from the replay. + await expect(page.getByTestId('assistant-text')).toContainText('REASON_OK') + await expect(page.getByTestId('assistant-thinking')).toHaveText( + 'Check the order.', + ) + await expect(page.getByTestId('assistant-message')).toHaveCount(1) + await expect(page.getByTestId('loading-indicator')).toHaveCount(0) + }) +})