From 269bbe4b0aba35ff2a5383909fb98d13bf11e32f Mon Sep 17 00:00:00 2001 From: kolaworld Date: Sat, 15 Aug 2026 14:09:18 -0400 Subject: [PATCH 1/3] fix(ai-persistence): ignore empty TEXT_MESSAGE_START message ids An empty start id was stored and blocked the TOOL_CALL_START parentMessageId fallback. --- .changeset/fix-empty-text-message-start-id.md | 5 ++ packages/ai-persistence/src/middleware.ts | 2 +- .../tests/with-persistence.test.ts | 53 +++++++++++++++++++ 3 files changed, 59 insertions(+), 1 deletion(-) create mode 100644 .changeset/fix-empty-text-message-start-id.md diff --git a/.changeset/fix-empty-text-message-start-id.md b/.changeset/fix-empty-text-message-start-id.md new file mode 100644 index 0000000000..950847214f --- /dev/null +++ b/.changeset/fix-empty-text-message-start-id.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-persistence': patch +--- + +Ignore empty `TEXT_MESSAGE_START` message IDs so tool-call `parentMessageId` can be used. diff --git a/packages/ai-persistence/src/middleware.ts b/packages/ai-persistence/src/middleware.ts index dbeeea2928..9bb6ad8b55 100644 --- a/packages/ai-persistence/src/middleware.ts +++ b/packages/ai-persistence/src/middleware.ts @@ -1541,7 +1541,7 @@ export function withPersistence( // bubble in place. if (ctx.phase === 'modelStream') { const s = runState.get(ctx) - if (s && chunk.type === 'TEXT_MESSAGE_START') { + if (s && chunk.type === 'TEXT_MESSAGE_START' && chunk.messageId !== '') { s.streamingMessageId = chunk.messageId s.streamingMessageCreatedAt = new Date() s.streamingText = '' diff --git a/packages/ai-persistence/tests/with-persistence.test.ts b/packages/ai-persistence/tests/with-persistence.test.ts index 338431c927..c60f9b444a 100644 --- a/packages/ai-persistence/tests/with-persistence.test.ts +++ b/packages/ai-persistence/tests/with-persistence.test.ts @@ -219,6 +219,59 @@ describe('withPersistence (state-only)', () => { ]) }) + it('does not let an empty TEXT_MESSAGE_START id replace parentMessageId', async () => { + const persistence = memoryPersistence() + const adapter = { + kind: 'text', + name: 'mock', + model: 'test-model', + '~types': {}, + chatStream: () => + (async function* () { + yield ev.runStarted() + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: '', + timestamp: 1, + } + yield { + type: EventType.TOOL_CALL_START, + toolCallId: 'call_1', + toolCallName: 'search', + toolName: 'search', + parentMessageId: 'stream-assistant', + timestamp: 1, + } + yield ev.text('Half a stor') + throw new Error('crash mid-stream') + })(), + structuredOutput: async () => ({ data: {}, rawText: '{}' }), + } as unknown as AnyTextAdapter + + await expect( + collect( + chat({ + adapter, + messages: [{ role: 'user', content: 'hi' }], + runId: 'r1', + threadId: 't1', + middleware: [ + withPersistence(persistence, { snapshotStreaming: true }), + ], + }) as AsyncIterable, + ), + ).rejects.toThrow('crash mid-stream') + + expect(await persistence.stores.messages!.loadThread('t1')).toEqual([ + { role: 'user', content: 'hi' }, + expect.objectContaining({ + role: 'assistant', + content: 'Half a stor', + id: 'stream-assistant', + }), + ]) + }) + it('stamps the terminal assistant turn with its stream messageId', async () => { const persistence = memoryPersistence() const { adapter } = mockAdapter([ From 4eebc698538c3784ab18d996f9b21c430ad2d2a0 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:55:20 +0000 Subject: [PATCH 2/3] ci: apply automated fixes --- packages/ai-persistence/src/middleware.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/ai-persistence/src/middleware.ts b/packages/ai-persistence/src/middleware.ts index 9bb6ad8b55..d01b62efe3 100644 --- a/packages/ai-persistence/src/middleware.ts +++ b/packages/ai-persistence/src/middleware.ts @@ -1541,7 +1541,11 @@ export function withPersistence( // bubble in place. if (ctx.phase === 'modelStream') { const s = runState.get(ctx) - if (s && chunk.type === 'TEXT_MESSAGE_START' && chunk.messageId !== '') { + if ( + s && + chunk.type === 'TEXT_MESSAGE_START' && + chunk.messageId !== '' + ) { s.streamingMessageId = chunk.messageId s.streamingMessageCreatedAt = new Date() s.streamingText = '' From 58183f0f99969042339faecc848312760c533348 Mon Sep 17 00:00:00 2001 From: Tom Beckenham <34339192+tombeckenham@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:42:10 +1000 Subject: [PATCH 3/3] fix(ai-persistence): keep per-turn stream state resets for empty-id starts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gate only the streamingMessageId capture on a non-empty (and string-typed) messageId instead of skipping the whole TEXT_MESSAGE_START branch. Skipping the branch also skipped the per-turn reset of streamingText and streamingMessageCreatedAt, so in a multi-turn agent run with empty-id starts a crash-window snapshot persisted turn 1's text concatenated with turn 2's, stamped with turn 1's tool-call id — two persisted messages sharing one id. Co-Authored-By: Claude Fable 5 --- packages/ai-persistence/src/middleware.ts | 15 ++-- .../tests/with-persistence.test.ts | 88 +++++++++++++++++++ 2 files changed, 97 insertions(+), 6 deletions(-) diff --git a/packages/ai-persistence/src/middleware.ts b/packages/ai-persistence/src/middleware.ts index d01b62efe3..91cabab815 100644 --- a/packages/ai-persistence/src/middleware.ts +++ b/packages/ai-persistence/src/middleware.ts @@ -1541,12 +1541,15 @@ export function withPersistence( // bubble in place. if (ctx.phase === 'modelStream') { const s = runState.get(ctx) - if ( - s && - chunk.type === 'TEXT_MESSAGE_START' && - chunk.messageId !== '' - ) { - s.streamingMessageId = chunk.messageId + if (s && chunk.type === 'TEXT_MESSAGE_START') { + // An empty/malformed messageId means "no identity" (matching the + // engine's convention), leaving room for the TOOL_CALL_START + // parentMessageId fallback below — but the per-turn accumulator + // still resets so snapshots never mix text across turns. + s.streamingMessageId = + typeof chunk.messageId === 'string' && chunk.messageId !== '' + ? chunk.messageId + : undefined s.streamingMessageCreatedAt = new Date() s.streamingText = '' } else if ( diff --git a/packages/ai-persistence/tests/with-persistence.test.ts b/packages/ai-persistence/tests/with-persistence.test.ts index c60f9b444a..b70a6d7513 100644 --- a/packages/ai-persistence/tests/with-persistence.test.ts +++ b/packages/ai-persistence/tests/with-persistence.test.ts @@ -268,10 +268,98 @@ describe('withPersistence (state-only)', () => { role: 'assistant', content: 'Half a stor', id: 'stream-assistant', + createdAt: expect.any(Date), }), ]) }) + it('resets streaming state on an empty-id TEXT_MESSAGE_START between turns', async () => { + const persistence = memoryPersistence() + let call = 0 + const adapter = { + kind: 'text', + name: 'mock', + model: 'test-model', + '~types': {}, + chatStream: () => { + call++ + if (call === 1) { + return (async function* () { + yield ev.runStarted() + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: '', + timestamp: 1, + } + yield ev.text('Let me search.') + yield { + type: EventType.TOOL_CALL_START, + toolCallId: 'call_1', + toolCallName: 'search', + toolName: 'search', + parentMessageId: 'assistant-turn-1', + timestamp: 1, + } + yield { + type: EventType.TOOL_CALL_ARGS, + toolCallId: 'call_1', + delta: '{}', + timestamp: 1, + } + yield { + type: EventType.RUN_FINISHED, + runId: 'r1', + threadId: 't1', + finishReason: 'tool_calls', + timestamp: 1, + } + })() + } + return (async function* () { + yield ev.runStarted() + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: '', + timestamp: 1, + } + yield ev.text('The answer is 42.') + throw new Error('crash mid-stream') + })() + }, + structuredOutput: async () => ({ data: {}, rawText: '{}' }), + } as unknown as AnyTextAdapter + + await expect( + collect( + chat({ + adapter, + messages: [{ role: 'user', content: 'search' }], + tools: [serverSearchTool()], + runId: 'r1', + threadId: 't1', + middleware: [ + withPersistence(persistence, { + snapshotStreaming: true, + snapshotIntervalMs: 0, + }), + ], + }) as AsyncIterable, + ), + ).rejects.toThrow('crash mid-stream') + + // The empty-id start on turn 2 still resets the per-turn accumulator: the + // crash-window snapshot holds only turn-2 text, and it must not inherit + // turn 1's tool-call id — two persisted messages may never share an id. + const thread = await persistence.stores.messages!.loadThread('t1') + expect(findAssistantToolCall(thread, 'call_1')?.id).toBe('assistant-turn-1') + const terminal = thread.at(-1) + expect(terminal).toMatchObject({ + role: 'assistant', + content: 'The answer is 42.', + }) + expect(terminal).not.toHaveProperty('id') + }) + it('stamps the terminal assistant turn with its stream messageId', async () => { const persistence = memoryPersistence() const { adapter } = mockAdapter([