From 095e3c34463fc4b2ae1ff601fbeb5d4c1e22a4f3 Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Wed, 15 Jul 2026 19:10:05 +0530 Subject: [PATCH] fix(chat): don't wipe fresh streaming/tool-timeline on a completed hydrate snapshot (#4903) --- app/src/store/chatRuntimeSlice.test.ts | 97 ++++++++++++++++++++++++++ app/src/store/chatRuntimeSlice.ts | 30 ++++++-- 2 files changed, 120 insertions(+), 7 deletions(-) diff --git a/app/src/store/chatRuntimeSlice.test.ts b/app/src/store/chatRuntimeSlice.test.ts index 05c97de39..769a8baaa 100644 --- a/app/src/store/chatRuntimeSlice.test.ts +++ b/app/src/store/chatRuntimeSlice.test.ts @@ -17,6 +17,7 @@ import chatRuntimeReducer, { resetSessionTokenUsage, setPendingApprovalForThread, setQueueStatusForThread, + setStreamingAssistantForThread, setToolTimelineForThread, setWorkflowProposalForThread, } from './chatRuntimeSlice'; @@ -759,6 +760,102 @@ describe('hydrateRuntimeFromSnapshot — workflow proposal race guard', () => { }); }); +// Regression: a `completed` snapshot rehydration must not clobber streaming +// narration / a tool timeline that is fresher than the snapshot itself. This +// happens when the socket-disconnect reconciliation path (ChatRuntimeProvider) +// deliberately preserves `streamingAssistantByThread` across `endInferenceTurn` +// so a partial reply stays visible while the socket reconnects, and a +// `fetchAndHydrateTurnState` rehydration then lands with a `completed` +// snapshot for that same (now-settled) turn. Only `interrupted` (a genuine +// crash — nothing fresher can exist) should unconditionally clobber; a +// `completed` snapshot should only fill in when there is no live state to +// lose (cold boot / new window). +describe('hydrateRuntimeFromSnapshot — streaming/timeline race guard', () => { + function makeTimelineSnapshot( + threadId: string, + lifecycle: PersistedTurnState['lifecycle'] + ): PersistedTurnState { + return { + threadId, + requestId: 'req-snapshot', + lifecycle, + iteration: 3, + maxIterations: 10, + streamingText: '', + thinking: '', + toolTimeline: [{ id: 'c-snapshot', name: 'web_search', round: 1, status: 'success' }], + startedAt: '2026-06-23T00:00:00Z', + updatedAt: '2026-06-23T00:00:00Z', + }; + } + + it('does not wipe a fresher streaming/tool-timeline lane on a completed snapshot', () => { + const store = makeStore(); + // The live driver already has state for this thread — e.g. the + // disconnect-reconciliation path kept the streaming bubble around while + // the socket reconnects. + store.dispatch( + setStreamingAssistantForThread({ + threadId: 't-settled', + streaming: { requestId: 'req-live', content: 'partial reply so far', thinking: '' }, + }) + ); + store.dispatch( + setToolTimelineForThread({ + threadId: 't-settled', + entries: [{ id: 'c-live', name: 'read_file', round: 1, status: 'success' }], + }) + ); + + store.dispatch( + hydrateRuntimeFromSnapshot({ snapshot: makeTimelineSnapshot('t-settled', 'completed') }) + ); + + const state = store.getState().chatRuntime; + // The fresher live streaming bubble survives untouched… + expect(state.streamingAssistantByThread['t-settled']?.content).toBe('partial reply so far'); + // …and so does the live tool timeline, rather than being replaced by the + // (behind) snapshot's single row. + expect(state.toolTimelineByThread['t-settled'].map(e => e.id)).toEqual(['c-live']); + }); + + it('clears the streaming/tool-timeline lanes on an interrupted snapshot (crash cleanup)', () => { + const store = makeStore(); + store.dispatch( + setStreamingAssistantForThread({ + threadId: 't-crashed', + streaming: { requestId: 'req-stale', content: 'stale partial reply', thinking: '' }, + }) + ); + store.dispatch( + setToolTimelineForThread({ + threadId: 't-crashed', + entries: [{ id: 'c-stale', name: 'read_file', round: 1, status: 'running' }], + }) + ); + + store.dispatch( + hydrateRuntimeFromSnapshot({ snapshot: makeTimelineSnapshot('t-crashed', 'interrupted') }) + ); + + const state = store.getState().chatRuntime; + expect(state.streamingAssistantByThread['t-crashed']).toBeUndefined(); + expect(state.toolTimelineByThread['t-crashed'].map(e => e.id)).toEqual(['c-snapshot']); + }); + + it('still hydrates the timeline from a completed snapshot on cold boot (no prior live state)', () => { + const store = makeStore(); + + store.dispatch( + hydrateRuntimeFromSnapshot({ snapshot: makeTimelineSnapshot('t-cold-boot', 'completed') }) + ); + + const state = store.getState().chatRuntime; + expect(state.streamingAssistantByThread['t-cold-boot']).toBeUndefined(); + expect(state.toolTimelineByThread['t-cold-boot'].map(e => e.id)).toEqual(['c-snapshot']); + }); +}); + describe('hydrateRuntimeFromSnapshot — persisted tool result output', () => { it('maps the persisted output onto parent and sub-agent rows as result', () => { const store = makeStore(); diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index 5ca4e0689..d0cad6d80 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -1989,13 +1989,29 @@ const chatRuntimeSlice = createSlice({ // is still carried so "View processing" replays the full reasoning. if (snapshot.lifecycle === 'interrupted' || snapshot.lifecycle === 'completed') { delete state.inferenceStatusByThread[threadId]; - delete state.streamingAssistantByThread[threadId]; - // Settle any in-flight rows so their agent names stop pulsing - // (no-op for an already-completed snapshot whose rows are terminal). - state.toolTimelineByThread[threadId] = preserveLiveSubagentProse( - state.toolTimelineByThread[threadId], - snapshot.toolTimeline.map(toolTimelineFromPersisted).map(settleOrphanedTimelineEntry) - ); + + // A `completed` snapshot can still lag behind live state this session + // already has for the thread: the socket-disconnect reconciliation + // path (`ChatRuntimeProvider`) deliberately *keeps* + // `streamingAssistantByThread` set across `endInferenceTurn` so a + // partial reply stays visible while the socket reconnects, and a + // `fetchAndHydrateTurnState` rehydration can land moments later. The + // same applies to `toolTimelineByThread`, which the live event + // stream keeps richer/fresher than a flush-boundary persisted copy. + // Only let the snapshot clobber those lanes when it is unambiguously + // the authority: an `interrupted` snapshot (the process that was + // streaming is gone — nothing fresher can exist) or there is no live + // streaming state for this thread to lose (cold boot / new window). + const hasFresherLiveStream = Boolean(state.streamingAssistantByThread[threadId]); + if (snapshot.lifecycle === 'interrupted' || !hasFresherLiveStream) { + delete state.streamingAssistantByThread[threadId]; + // Settle any in-flight rows so their agent names stop pulsing + // (no-op for an already-completed snapshot whose rows are terminal). + state.toolTimelineByThread[threadId] = preserveLiveSubagentProse( + state.toolTimelineByThread[threadId], + snapshot.toolTimeline.map(toolTimelineFromPersisted).map(settleOrphanedTimelineEntry) + ); + } state.processingByThread[threadId] = snapshot.transcript ?? []; return; }