diff --git a/packages/client/core/src/conversation-store.ts b/packages/client/core/src/conversation-store.ts index 5d398140e..8ec749625 100644 --- a/packages/client/core/src/conversation-store.ts +++ b/packages/client/core/src/conversation-store.ts @@ -2,7 +2,7 @@ import type { AgentEvent, SessionId } from '@linkcode/schema'; import type { Unsubscribe } from '@linkcode/transport'; import { noop } from 'foxact/noop'; import type { LinkCodeClient, SequencedAgentEvent } from './client'; -import type { Conversation, ConversationSeed } from './conversation'; +import type { Conversation, ConversationBuilder, ConversationSeed } from './conversation'; import { createConversationBuilder } from './conversation'; /** A `useSyncExternalStore`-shaped incremental projection of one session's conversation. @@ -29,52 +29,80 @@ const EMPTY_CONVERSATION: Conversation = { pendingQuestionIds: [], }; -/** - * Whether the seed's transcript snapshot can be assumed to contain this event — the only license - * the `uptoSeq` cut has to drop it as "already in the snapshot". Providers flush transcripts by - * whole item, so coverage is checked per provider identity, or by counted value for user prompts - * whose host and provider ids cannot converge. A chunk of a message the snapshot never saw (the - * in-flight reply — claude-code writes the row only when the message completes) must survive a - * mid-turn reseed, or the streamed text vanishes at a chunk boundary (CODE-272). Everything - * outside the switch (interactive requests and resolutions, status, stop, errors, usage …) is - * ephemeral: it never appears in `history.read`, so cutting it would erase it outright — a pending - * permission-request would vanish and strand the turn (CODE-35). - */ -function coveredBySeed( +type UserMessageEvent = Extract; +interface SeedUserMessageQueue { + messages: UserMessageEvent[]; + nextIndex: number; +} + +function takeSeedUserMessage( + messagesByContent: Map, + content: UserMessageEvent['content'], +): UserMessageEvent | undefined { + const key = JSON.stringify(content); + const queue = messagesByContent.get(key); + if (!queue) return undefined; + const message = queue.messages[queue.nextIndex]; + queue.nextIndex += 1; + if (queue.nextIndex === queue.messages.length) messagesByContent.delete(key); + return message; +} + +/** Fold a pre-cut event only when the transcript snapshot does not already cover it. */ +function foldPreCutEvent( + builder: ConversationBuilder, event: AgentEvent, + receivedAt: number | undefined, seedMessageIds: ReadonlySet, seedToolIds: ReadonlySet, - seedUserMessageCounts: Map, -): boolean { + seedUserMessages: Map, +): void { switch (event.type) { case 'agent-message': case 'agent-message-chunk': case 'agent-thought': - case 'agent-thought-chunk': - return seedMessageIds.has(event.messageId); + case 'agent-thought-chunk': { + if (!seedMessageIds.has(event.messageId)) builder.advance(event, receivedAt); + break; + } case 'user-message': { - // Host and provider ids cannot converge, so consume one matching seed row by value. Counting - // preserves repeated prompts while an unflushed queued prompt remains visible past the cut. - const key = JSON.stringify(event.content); - const remaining = seedUserMessageCounts.get(key) ?? 0; - if (remaining === 0) return false; - if (remaining === 1) seedUserMessageCounts.delete(key); - else seedUserMessageCounts.set(key, remaining - 1); - return true; + // Host and provider ids cannot converge, so consume matching seed rows by value. Some + // histories omit images; use the full live echo to enrich that seed row in place. + if (takeSeedUserMessage(seedUserMessages, event.content)) break; + if (event.content.some((block) => block.type === 'image')) { + const seedMessage = takeSeedUserMessage( + seedUserMessages, + event.content.filter((block) => block.type !== 'image'), + ); + if (seedMessage) { + builder.advance({ + ...event, + messageId: seedMessage.messageId, + branchCursor: seedMessage.branchCursor, + }); + break; + } + } + builder.advance(event, receivedAt); + break; + } + case 'tool-call': { + if (!seedToolIds.has(event.toolCall.toolCallId)) builder.advance(event, receivedAt); + break; + } + case 'tool-call-content-chunk': { + if (!seedToolIds.has(event.toolCallId)) builder.advance(event, receivedAt); + break; } - case 'tool-call': - return seedToolIds.has(event.toolCall.toolCallId); - case 'tool-call-content-chunk': - return seedToolIds.has(event.toolCallId); default: - return false; + builder.advance(event, receivedAt); } } /** * Project a session's conversation from a transcript seed plus the live event buffer: the seed * folds once, then `getSnapshot` lazily advances by unconsumed events, skipping events inside the - * `uptoSeq` cut that the snapshot verifiably covers (see {@link coveredBySeed}). The sync is idempotent and monotone with a stable snapshot identity + * `uptoSeq` cut that the snapshot verifiably covers (see {@link foldPreCutEvent}). The sync is idempotent and monotone with a stable snapshot identity * between events — the `useSyncExternalStore` getSnapshot contract. A store is bound to one * (session, seed) pair; create a fresh one when either changes. */ @@ -92,7 +120,7 @@ export function createConversationStore( // Identities the snapshot actually holds, for the per-event coverage check of the cut. const seedMessageIds = new Set(); const seedToolIds = new Set(); - const seedUserMessageCounts = new Map(); + const seedUserMessages = new Map(); if (seed) { for (const { event } of seed.events) { switch (event.type) { @@ -104,7 +132,9 @@ export function createConversationStore( break; case 'user-message': { const key = JSON.stringify(event.content); - seedUserMessageCounts.set(key, (seedUserMessageCounts.get(key) ?? 0) + 1); + const queue = seedUserMessages.get(key); + if (queue) queue.messages.push(event); + else seedUserMessages.set(key, { messages: [event], nextIndex: 0 }); break; } case 'tool-call': @@ -131,11 +161,10 @@ export function createConversationStore( const events = client.eventsSnapshot(sessionId); for (let i = firstIndexAfter(events, consumedSeq); i < events.length; i += 1) { const { event, seq, receivedAt } = events[i]; - if ( - seq > uptoSeq || - !coveredBySeed(event, seedMessageIds, seedToolIds, seedUserMessageCounts) - ) { + if (seq > uptoSeq) { builder.advance(event, receivedAt); + } else { + foldPreCutEvent(builder, event, receivedAt, seedMessageIds, seedToolIds, seedUserMessages); } } // Snap to the counter even when the buffer lags it (cleared by a stop): those events are diff --git a/packages/client/core/tests/integration/conversation-store.test.ts b/packages/client/core/tests/integration/conversation-store.test.ts index 2db199374..b63f18a65 100644 --- a/packages/client/core/tests/integration/conversation-store.test.ts +++ b/packages/client/core/tests/integration/conversation-store.test.ts @@ -73,6 +73,54 @@ describe('createConversationStore', () => { close(); }); + it('enriches a lossy seeded prompt with its live image instead of appending a duplicate', async () => { + const { client, send, close } = await harness(); + const livePrompt: AgentEvent = { + type: 'user-message', + messageId: 'host-prompt' as MessageId, + content: [ + { type: 'text', text: 'describe this image' }, + { type: 'image', data: 'cG5n', mimeType: 'image/png' }, + ], + branchCursor: 'live-cursor', + }; + const reply: AgentEvent = { + type: 'agent-message', + messageId: 'reply' as MessageId, + content: [{ type: 'text', text: 'It is a test image.' }], + }; + send(livePrompt); + send(reply); + await tick(); + + const store = createConversationStore(client, sessionId, { + events: [ + // Some provider histories retain the prompt text but omit its image blocks. + { + event: { + type: 'user-message', + messageId: 'provider-prompt' as MessageId, + content: [{ type: 'text', text: 'describe this image' }], + branchCursor: 'provider-cursor', + }, + }, + { event: reply }, + ], + uptoSeq: 2, + }); + + const messages = store.getSnapshot().items.filter((item) => item.kind === 'message'); + expect(messages).toHaveLength(2); + expect(messages[0]).toMatchObject({ + id: 'provider-prompt', + role: 'user', + blocks: livePrompt.content, + branchCursor: 'provider-cursor', + }); + expect(messages[1]).toMatchObject({ id: 'reply', role: 'assistant' }); + close(); + }); + it('consumes only one matching seed row for repeated prompt content', async () => { const { client, send, close } = await harness(); send(userText('repeat', 'host-1'));