diff --git a/apps/desktop/src/lib/inflight-turn-journal.test.ts b/apps/desktop/src/lib/inflight-turn-journal.test.ts index 206a12228894..e53857f7d237 100644 --- a/apps/desktop/src/lib/inflight-turn-journal.test.ts +++ b/apps/desktop/src/lib/inflight-turn-journal.test.ts @@ -183,7 +183,7 @@ describe('recoverInFlightTurnJournal', () => { ]) const base = [user('u0', 'earlier turn'), assistant('a0', 'earlier reply')] - const result = recoverInFlightTurnJournal('stored-1', base) + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: true }) expect(result.applied).toBe(true) expect(result.messages.map(m => m.id)).toEqual(['u0', 'a0', 'u1', 'assistant-stream-1']) @@ -283,6 +283,76 @@ describe('recoverInFlightTurnJournal', () => { expect(merged.id).toBe('assistant-stream-rt9') expect(merged.parts[1]).toMatchObject({ type: 'text', text: 'a much longer locally journaled partial answer' }) }) + + // ── Scrambled-transcript regression (duplicate trailing answers) ─────────── + // The journal can outlive the turn it recorded (reclaim/reconnect/restart + // races skip the settle that would clear it). On resume the fold then + // re-appends content that the committed transcript ALREADY holds, rendering + // the same answers twice at the end of the conversation. Reported on the + // desktop as "the answer was already there, but it was inputted again". + + it('does not re-append committed answers when the journaled user row never persisted', () => { + // A resume projection can journal a `user-inflight-*` row that was never + // written to the DB (and may even belong to a different conversation). + // Because no base user matches it, the fold used to treat the whole tail + // as unknown and append it — duplicating the assistant answers below. + journalEntry([ + user('user-inflight-a3c2beb1', 'a stray user bubble that never persisted'), + assistant('assistant-stream-1', 'the committed answer') + ]) + + const base = [user('db-u1', 'the real prompt'), assistant('db-a1', 'the committed answer')] + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: false }) + + expect(result.caughtUp).toBe(true) + expect(result.applied).toBe(false) + expect(result.messages).toBe(base) + expect(result.messages.map(m => m.id)).toEqual(['db-u1', 'db-a1']) + // The stale entry is cleared so the next resume stays clean. + expect(readInFlightTurnJournal('stored-1')).toBeNull() + }) + + it('does not re-append committed answers when the journal tail has no user row', () => { + // A tail captured after a partial hydrate can end on assistant rows with + // no user prompt before them. The old code appended them verbatim, so the + // transcript ended with a duplicate of an answer that was already settled. + journalEntry([assistant('assistant-stream-1', 'the committed answer')]) + + const base = [user('db-u1', 'the real prompt'), assistant('db-a1', 'the committed answer')] + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: false }) + + expect(result.caughtUp).toBe(true) + expect(result.messages).toBe(base) + expect(readInFlightTurnJournal('stored-1')).toBeNull() + }) + + it('keeps appending a genuinely unknown turn (crash recovery still works)', () => { + // The staleness check must not swallow a tail the base never saw: that is + // the crash-recovery path the journal exists for. + journalEntry([user('u1', 'the live prompt'), assistant('assistant-stream-1', 'partial answer', { pending: true })]) + + const base = [user('db-u0', 'an earlier turn'), assistant('db-a0', 'earlier reply')] + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: true }) + + expect(result.applied).toBe(true) + expect(result.caughtUp).toBe(false) + expect(result.messages.map(m => m.id)).toEqual(['db-u0', 'db-a0', 'u1', 'assistant-stream-1']) + expect(readInFlightTurnJournal('stored-1')).not.toBeNull() + }) + + it('does not resurrect the journal streamId on a not-running resume (journal self-clear)', () => { + // The fold used to carry the stale entry's streamId onto the resumed state + // even when the backend reported the session idle. persistInFlightTurnState + // then re-wrote the journal instead of clearing it, so the same stale tail + // was folded again on every open — the scramble never healed. + journalEntry([user('u1', 'do the thing'), assistant('assistant-stream-1', 'partial answer', { pending: true })]) + + const base = [user('db-u0', 'an earlier turn'), assistant('db-a0', 'earlier reply')] + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: false }) + + expect(result.applied).toBe(true) + expect(result.streamId).toBeNull() + }) }) describe('mergeInFlightMessages', () => { diff --git a/apps/desktop/src/lib/inflight-turn-journal.ts b/apps/desktop/src/lib/inflight-turn-journal.ts index 27912970ff67..0ce7789c8429 100644 --- a/apps/desktop/src/lib/inflight-turn-journal.ts +++ b/apps/desktop/src/lib/inflight-turn-journal.ts @@ -376,6 +376,32 @@ function withoutBaseIds(rows: ChatMessage[], baseMessages: ChatMessage[]): ChatM return rows.filter(row => !baseIds.has(row.id)) } +/** Whether every recoverable assistant row in the journal tail already exists + * as committed text in the base transcript. When true, the journal outlived + * the turn it recorded and appending it would re-render the same answers at + * the end of the transcript (the "scrambled conversation" regression). */ +function journalTailAlreadyCommitted(tailAssistants: ChatMessage[], baseMessages: ChatMessage[]): boolean { + const recoverable = tailAssistants.filter(assistantHasRecoverableContent) + + if (recoverable.length === 0) { + return false + } + + const baseTexts = new Set( + baseMessages + .filter(message => message.role === 'assistant' && !message.hidden) + .map(message => normalizedText(chatMessageText(message))) + ) + + return recoverable.every(message => { + const text = normalizedText(chatMessageText(message)) + + // Error-only rows carry no text to verify against — keep the conservative + // append path rather than risk dropping a recoverable failure. + return text.length > 0 && baseTexts.has(text) + }) +} + export function mergeInFlightMessages( baseMessages: ChatMessage[], tailMessages: ChatMessage[], @@ -402,15 +428,27 @@ export function mergeInFlightMessages( const matchingUserIndex = tailUser ? baseMessages.findLastIndex(message => userMessagesMatch(message, tailUser)) : -1 if (matchingUserIndex < 0) { - // Base doesn't know this turn at all (user row was never persisted): - // append the whole tail. + // No base user matches the tail's user row (a projected user-inflight row + // that never persisted, or a tail captured without its user prompt). If the + // tail's answers are already committed in the transcript, the journal is + // stale — appending it would re-render the same replies at the end of the + // conversation. Otherwise, the base never saw this turn at all: append the + // whole tail (the crash-recovery path the journal exists for). + if (journalTailAlreadyCommitted(tailAssistants, baseMessages)) { + return { ...noop, caughtUp: true } + } + const streamId = lastJournalRow?.id ?? null return { applied: true, caughtUp: false, messages: [...baseMessages, ...withoutBaseIds(tail, baseMessages)], - streamId, + // Only a genuinely running turn keeps a live stream target. On an idle + // resume, carrying the stale streamId would keep the journal entry alive + // (persistInFlightTurnState only clears when streamId is null) and the + // same tail would be folded again on every open. + streamId: options.keepPending ? streamId : null, turnStartedAt: null } } @@ -464,7 +502,16 @@ export function mergeInFlightMessages( ...baseMessages.slice(projectionIndex + 1) ] - return { applied: true, caughtUp: false, messages, streamId: merged.id, turnStartedAt: null } + return { + applied: true, + caughtUp: false, + messages, + // Same idle-resume rule as the append path: only a running turn keeps the + // stream target alive, so an idle resume clears the journal instead of + // re-folding the same tail on every open. + streamId: options.keepPending ? merged.id : null, + turnStartedAt: null + } } const persistTimers = new Map>() @@ -570,7 +617,11 @@ export function recoverInFlightTurnJournal( return { ...recovered, - streamId: recovered.applied ? (recovered.streamId ?? snapshot.streamId) : null, + // Never resurrect a stale stream target on an idle resume: with + // keepPending=false the session is not running, so the recovered rows are + // settled history and the journal must clear on the next state update — + // otherwise the same stale tail is folded again on every open. + streamId: recovered.applied ? (recovered.streamId ?? (options.keepPending ? snapshot.streamId : null)) : null, turnStartedAt: recovered.applied ? snapshot.turnStartedAt : null } } diff --git a/contributors/emails/nformenton@Nicolass-MacBook-Air.local b/contributors/emails/nformenton@Nicolass-MacBook-Air.local new file mode 100644 index 000000000000..9b9caaa5824c --- /dev/null +++ b/contributors/emails/nformenton@Nicolass-MacBook-Air.local @@ -0,0 +1,2 @@ +Nicolas-Formenton +# PR #84137 / #84021 (desktop fixes from a second machine)