Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fix-empty-text-message-start-id.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@tanstack/ai-persistence': patch
---

Ignore empty `TEXT_MESSAGE_START` message IDs so tool-call `parentMessageId` can be used.
6 changes: 5 additions & 1 deletion packages/ai-persistence/src/middleware.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1541,7 +1541,11 @@ export function withPersistence<TStores extends ChatTranscriptStores>(
// 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 = ''
Expand Down
53 changes: 53 additions & 0 deletions packages/ai-persistence/tests/with-persistence.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<StreamChunk>,
),
).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([
Expand Down
Loading