From e4e7dd5c6eac30113e0f38eef7a069f1817a2cef Mon Sep 17 00:00:00 2001 From: Kinkoolino-Hermes <297364961+Kinkoolino-Hermes@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:46:20 +0200 Subject: [PATCH] fix(desktop): preserve full transcripts for running sessions --- .../hooks/use-session-actions.test.tsx | 421 +++++++++++++++++- .../hooks/use-session-actions/index.ts | 161 +++++-- .../hooks/use-session-actions/utils.test.ts | 179 +++++++- .../hooks/use-session-actions/utils.ts | 236 +++++++++- 4 files changed, 942 insertions(+), 55 deletions(-) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx index e0d2d76a91..6cc4e5c7cf 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx @@ -612,6 +612,8 @@ function ResumeHarness({ sessionStateByRuntimeIdRef?: MutableRefObject> }) { const ref = (value: T): MutableRefObject => ({ current: value }) + const runtimeMapRef = runtimeIdByStoredSessionIdRef ?? ref(new Map()) + const stateMapRef = sessionStateByRuntimeIdRef ?? ref(new Map()) const actions = useSessionActions({ activeSessionId: null, @@ -624,13 +626,16 @@ function ResumeHarness({ navigate: vi.fn() as never, requestGateway, resetViewSync: vi.fn(), - runtimeIdByStoredSessionIdRef: runtimeIdByStoredSessionIdRef ?? ref(new Map()), + runtimeIdByStoredSessionIdRef: runtimeMapRef, selectedStoredSessionId, selectedStoredSessionIdRef: ref(selectedStoredSessionId), - sessionStateByRuntimeIdRef: sessionStateByRuntimeIdRef ?? ref(new Map()), + sessionStateByRuntimeIdRef: stateMapRef, syncSessionStateToView: vi.fn(), updateSessionState: (sessionId, updater) => { - const next = updater({} as ClientSessionState) + const current = stateMapRef.current.get(sessionId) ?? ({} as ClientSessionState) + const next = updater(current) + + stateMapRef.current.set(sessionId, next) onStateUpdate?.(sessionId, next) return next @@ -767,12 +772,16 @@ describe('resumeSession failure recovery', () => { expect($messages.get().map(message => message.id)).toContain('user-optimistic') }) - it('restores the in-flight turn and queued user prompt after a full renderer restart', async () => { + it('keeps the complete transcript with the live tail after a full renderer restart', async () => { const storedMessages = [ - { content: 'earlier question', role: 'user', timestamp: 1 }, - { content: 'earlier answer', role: 'assistant', timestamp: 2 } + { content: 'older question removed by compression', role: 'user', timestamp: 1 }, + { content: 'older answer removed by compression', role: 'assistant', timestamp: 2 }, + { content: 'recent question', role: 'user', timestamp: 3 }, + { content: 'recent answer', role: 'assistant', timestamp: 4 } ] + const compressedRuntimeMessages = storedMessages.slice(-2) + vi.mocked(getLatestSessionMessages).mockResolvedValue({ messages: storedMessages, session_id: 'stored-1' } as never) const requestGateway = vi.fn(async (method: string) => { @@ -781,8 +790,8 @@ describe('resumeSession failure recovery', () => { session_id: 'runtime-1', session_key: 'stored-1', resumed: 'stored-1', - message_count: storedMessages.length, - messages: storedMessages, + message_count: compressedRuntimeMessages.length, + messages: compressedRuntimeMessages, running: true, inflight: { user: 'current prompt', @@ -810,11 +819,99 @@ describe('resumeSession failure recovery', () => { await resume!('stored-1', true) const renderedMessages = JSON.stringify(resumedState?.messages) + expect(renderedMessages).toContain('older question removed by compression') expect(renderedMessages).toContain('current prompt') expect(renderedMessages).toContain('partial answer') expect(renderedMessages).toContain('newest prompt') }) + it('preserves a runtime-cache delta that arrives while cold resume waits for REST', async () => { + const persisted = deferred>>() + + const sessionStateByRuntimeIdRef: MutableRefObject> = { + current: new Map() + } + + const compressedRuntimeMessages = [ + { content: 'recent question', role: 'user', timestamp: 3 }, + { content: 'recent answer', role: 'assistant', timestamp: 4 } + ] + + vi.mocked(getLatestSessionMessages).mockReturnValue(persisted.promise) + + const requestGateway = vi.fn(async (method: string) => { + if (method === 'session.resume') { + return { + session_id: 'runtime-1', + session_key: 'stored-1', + resumed: 'stored-1', + message_count: compressedRuntimeMessages.length, + messages: compressedRuntimeMessages, + running: true, + inflight: { + user: 'current prompt', + assistant: 'partial A', + streaming: true + }, + info: {} + } as never + } + + return {} as never + }) + + let resumedState: ClientSessionState | undefined + let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise) | null = null + + render( + (resume = ready)} + onStateUpdate={(_sessionId, state) => (resumedState = state)} + requestGateway={requestGateway} + sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef} + /> + ) + await waitFor(() => expect(resume).not.toBeNull()) + + const resumePromise = resume!('stored-1', true) + + await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.resume', expect.anything())) + + const runtimeState = clientState('stored-1') + runtimeState.messages = [ + { + id: 'assistant-stream-live-cold', + role: 'assistant', + parts: [{ type: 'text', text: ' + delta B' }], + pending: true + } + ] + runtimeState.streamId = 'assistant-stream-live-cold' + sessionStateByRuntimeIdRef.current.set('runtime-1', runtimeState) + + await act(async () => { + persisted.resolve({ + messages: [ + { content: 'older question removed by compression', role: 'user', timestamp: 1 }, + { content: 'older answer removed by compression', role: 'assistant', timestamp: 2 }, + ...compressedRuntimeMessages + ], + session_id: 'stored-1' + } as never) + await resumePromise + }) + + const renderedText = JSON.stringify(resumedState?.messages) + + const streamingAssistantRows = resumedState?.messages.filter(message => message.id.startsWith('assistant-stream-')) + + expect(renderedText).toContain('older question removed by compression') + expect(renderedText).toContain('partial A') + expect(renderedText).toContain('delta B') + expect(streamingAssistantRows).toHaveLength(1) + expect(streamingAssistantRows?.[0].id).toBe('assistant-stream-live-cold') + }) + it('uses the continuation projection when resume rotates an equal-length stored transcript', async () => { const parentMessages = [ { content: 'question before compression', role: 'user', timestamp: 1 }, @@ -1681,6 +1778,314 @@ describe('resumeSession warm-cache mapping integrity', () => { expect(renderedMessages).toContain('still here after wake too') }) + it('keeps the complete persisted transcript when activating a compressed running session', async () => { + const runtimeIdByStoredSessionIdRef: MutableRefObject> = { + current: new Map([['stored-A', 'rt-A']]) + } + + const state = clientState('stored-A') + state.messages = [ + { + id: 'runtime-user', + role: 'user', + parts: [{ type: 'text', text: 'recent prompt' }] + }, + { + id: 'runtime-assistant', + role: 'assistant', + parts: [{ type: 'text', text: 'recent answer' }] + } + ] + + const sessionStateByRuntimeIdRef: MutableRefObject> = { + current: new Map([['rt-A', state]]) + } + + const persistedMessages = [ + { content: 'older prompt that compression removed', role: 'user', timestamp: 1 }, + { content: 'older answer that compression removed', role: 'assistant', timestamp: 2 }, + { content: 'recent prompt', role: 'user', timestamp: 3 }, + { content: 'recent answer', role: 'assistant', timestamp: 4 } + ] + + const compressedRuntimeMessages = persistedMessages.slice(2) + + vi.mocked(getLatestSessionMessages).mockResolvedValue({ + messages: persistedMessages, + session_id: 'stored-A' + } as never) + + const requestGateway = vi.fn(async (method: string) => { + if (method === 'session.activate') { + return { + session_id: 'rt-A', + session_key: 'stored-A', + resumed: 'stored-A', + message_count: compressedRuntimeMessages.length, + messages: compressedRuntimeMessages, + running: true, + inflight: { + user: 'current prompt', + assistant: 'partial answer', + streaming: true + }, + info: {} + } as never + } + + return {} as never + }) + + let resumedState: ClientSessionState | undefined + let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise) | null = null + + render( + (resume = ready)} + onStateUpdate={(_sessionId, next) => (resumedState = next)} + requestGateway={requestGateway} + runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef} + sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef} + /> + ) + await waitFor(() => expect(resume).not.toBeNull()) + await resume!('stored-A', true) + + const renderedMessages = resumedState?.messages ?? [] + const renderedText = JSON.stringify(renderedMessages) + + expect(renderedText).toContain('older prompt that compression removed') + expect(renderedText).toContain('older answer that compression removed') + expect(renderedText).toContain('recent prompt') + expect(renderedText).toContain('recent answer') + expect(renderedText).toContain('partial answer') + expect(renderedMessages.filter(message => JSON.stringify(message).includes('current prompt'))).toHaveLength(1) + }) + + it('preserves live cache updates that arrive while the persisted transcript is loading', async () => { + const runtimeIdByStoredSessionIdRef: MutableRefObject> = { + current: new Map([['stored-A', 'rt-A']]) + } + + const state = clientState('stored-A') + state.messages = [ + { + id: 'runtime-user', + role: 'user', + parts: [{ type: 'text', text: 'recent prompt' }], + timestamp: 3 + }, + { + id: 'runtime-assistant', + role: 'assistant', + parts: [{ type: 'text', text: 'recent answer' }], + timestamp: 4 + }, + { + id: 'user-inflight-rt-A', + role: 'user', + parts: [{ type: 'text', text: 'current prompt' }] + }, + { + id: 'assistant-stream-live-123', + role: 'assistant', + parts: [{ type: 'text', text: 'partial A' }], + pending: true + } + ] + + const sessionStateByRuntimeIdRef: MutableRefObject> = { + current: new Map([['rt-A', state]]) + } + + const persisted = deferred>>() + + const compressedRuntimeMessages = [ + { content: 'recent prompt', role: 'user', timestamp: 3 }, + { content: 'recent answer', role: 'assistant', timestamp: 4 } + ] + + vi.mocked(getLatestSessionMessages).mockReturnValue(persisted.promise) + + const requestGateway = vi.fn(async (method: string) => { + if (method === 'session.activate') { + return { + session_id: 'rt-A', + session_key: 'stored-A', + resumed: 'stored-A', + message_count: compressedRuntimeMessages.length, + messages: compressedRuntimeMessages, + running: true, + inflight: { + user: 'current prompt', + assistant: 'partial A', + streaming: true + }, + info: {} + } as never + } + + return {} as never + }) + + let resumedState: ClientSessionState | undefined + let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise) | null = null + + render( + (resume = ready)} + onStateUpdate={(_sessionId, next) => (resumedState = next)} + requestGateway={requestGateway} + runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef} + sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef} + /> + ) + await waitFor(() => expect(resume).not.toBeNull()) + + const resumePromise = resume!('stored-A', true) + + await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.activate', expect.anything())) + + const liveState = sessionStateByRuntimeIdRef.current.get('rt-A')! + + const liveMessages = liveState.messages.map(message => + message.id === 'assistant-stream-live-123' + ? { ...message, parts: [{ type: 'text' as const, text: 'partial A + delta B' }] } + : message + ) + + sessionStateByRuntimeIdRef.current.set('rt-A', { + ...liveState, + messages: [ + ...liveMessages, + { + id: 'user-racing', + role: 'user', + parts: [{ type: 'text', text: 'racing prompt' }] + } + ] + }) + + await act(async () => { + persisted.resolve({ + messages: [ + { content: 'older prompt', role: 'user', timestamp: 1 }, + { content: 'older answer', role: 'assistant', timestamp: 2 }, + ...compressedRuntimeMessages + ], + session_id: 'stored-A' + } as never) + await resumePromise + }) + + const renderedText = JSON.stringify(resumedState?.messages) + + expect(renderedText).toContain('older prompt') + expect(renderedText).toContain('partial A + delta B') + expect(renderedText).toContain('racing prompt') + + const streamingAssistantRows = resumedState?.messages.filter(message => message.id.startsWith('assistant-stream-')) + + expect(streamingAssistantRows).toHaveLength(1) + expect(streamingAssistantRows?.[0].id).toBe('assistant-stream-live-123') + }) + + it('does not duplicate an in-flight user prompt already present in the persisted suffix', async () => { + const runtimeIdByStoredSessionIdRef: MutableRefObject> = { + current: new Map([['stored-A', 'rt-A']]) + } + + const state = clientState('stored-A') + state.messages = [ + { + id: 'runtime-user', + role: 'user', + parts: [{ type: 'text', text: 'earlier prompt' }] + }, + { + id: 'runtime-assistant', + role: 'assistant', + parts: [{ type: 'text', text: 'earlier answer' }] + }, + { + id: 'user-optimistic', + role: 'user', + parts: [{ type: 'text', text: 'current prompt' }] + }, + { + id: 'assistant-stream-rt-A', + role: 'assistant', + parts: [{ type: 'text', text: 'partial answer' }], + pending: true + } + ] + + const sessionStateByRuntimeIdRef: MutableRefObject> = { + current: new Map([['rt-A', state]]) + } + + const compressedRuntimeMessages = [ + { content: 'earlier prompt', role: 'user', timestamp: 1 }, + { content: 'earlier answer', role: 'assistant', timestamp: 2 } + ] + + const persistedMessages = [ + { content: 'older prompt removed by compression', role: 'user', timestamp: -1 }, + { content: 'older answer removed by compression', role: 'assistant', timestamp: 0 }, + ...compressedRuntimeMessages, + { content: 'current prompt', role: 'user', timestamp: 3 } + ] + + vi.mocked(getLatestSessionMessages).mockResolvedValue({ + messages: persistedMessages, + session_id: 'stored-A' + } as never) + + const requestGateway = vi.fn(async (method: string) => { + if (method === 'session.activate') { + return { + session_id: 'rt-A', + session_key: 'stored-A', + resumed: 'stored-A', + message_count: compressedRuntimeMessages.length, + messages: compressedRuntimeMessages, + running: true, + inflight: { + user: 'current prompt', + assistant: 'partial answer', + streaming: true + }, + info: {} + } as never + } + + return {} as never + }) + + let resumedState: ClientSessionState | undefined + let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise) | null = null + + render( + (resume = ready)} + onStateUpdate={(_sessionId, next) => (resumedState = next)} + requestGateway={requestGateway} + runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef} + sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef} + /> + ) + await waitFor(() => expect(resume).not.toBeNull()) + await resume!('stored-A', true) + + const currentPromptRows = (resumedState?.messages ?? []).filter(message => + JSON.stringify(message).includes('current prompt') + ) + + expect(currentPromptRows).toHaveLength(1) + expect(JSON.stringify(resumedState?.messages)).toContain('partial answer') +>>>>>>> bfb973e25c (fix(desktop): preserve full transcripts for running sessions) + }) + it('keeps a warm runtime and optimistic turn on a transient activation timeout', async () => { const runtimeIdByStoredSessionIdRef: MutableRefObject> = { current: new Map([['stored-A', 'rt-A']]) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts index 88deda8dcb..1971ac1ce8 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts @@ -82,10 +82,13 @@ import { applyStoredSessionPreviewRuntimeInfo, type BranchMessage, chatMessageArraysEquivalent, + dedupeInflightUserAgainstTranscript, isSessionGoneError, + overlayConcurrentMessageChanges, patchSessionWorkspace, preserveLocalPendingTurnMessages, reconcileResumeMessages, + removeRepresentedLocalLiveProjection, resolveResumedBusy, resolveSessionProfile, resolveStoredSession, @@ -137,17 +140,27 @@ function applyStoredUsage(stored: { input_tokens?: number | null; output_tokens? setCurrentUsage(current => ({ ...current, input, output, total: input + output })) } +function reconcileAuthoritativeChatMessages( + authoritativeMessages: ChatMessage[], + previousMessages: ChatMessage[], + liveProjection?: Pick +): ChatMessage[] { + const withLiveProjection = liveProjection + ? appendLiveSessionProjection(authoritativeMessages, liveProjection) + : authoritativeMessages + + const reconciled = reconcileResumeMessages(withLiveProjection, previousMessages) + const withPendingTurn = preserveLocalPendingTurnMessages(reconciled, previousMessages) + + return preserveLocalAssistantErrors(withPendingTurn, previousMessages) +} + function reconcileAuthoritativeMessages( authoritativeMessages: SessionResumeResponse['messages'], previousMessages: ChatMessage[], liveProjection?: Pick ): ChatMessage[] { - const authoritative = toChatMessages(authoritativeMessages) - const withLiveProjection = liveProjection ? appendLiveSessionProjection(authoritative, liveProjection) : authoritative - const reconciled = reconcileResumeMessages(withLiveProjection, previousMessages) - const withPendingTurn = preserveLocalPendingTurnMessages(reconciled, previousMessages) - - return preserveLocalAssistantErrors(withPendingTurn, previousMessages) + return reconcileAuthoritativeChatMessages(toChatMessages(authoritativeMessages), previousMessages, liveProjection) } // `session.create` params from the current profile + sticky-UI model/effort/fast, @@ -839,12 +852,12 @@ export function useSessionActions({ Boolean(sessionStateByRuntimeIdRef.current.get(cachedRuntimeId)?.busy) ) - // While idle, the persisted REST transcript is the display - // authority: session.activate returns the runtime's compressed - // context projection, not necessarily the complete conversation. - // During a live turn, keep the runtime/cache projection so an - // accepted but not-yet-persisted prompt or stream is never lost. - if (!running && persistedTranscriptPromise) { + // The persisted REST transcript is the display authority: a live + // runtime may carry only the agent's compressed context projection, + // which is intentionally smaller than the user-visible conversation. + // Reconcile its in-flight/queued tail onto the complete transcript + // instead of replacing durable history while the turn is running. + if (persistedTranscriptPromise) { const persisted = await persistedTranscriptPromise if (!isCurrentResume()) { @@ -869,10 +882,34 @@ export function useSessionActions({ persistedMatchesActivatedSession && (persisted.messages.length || !activatedMessages.length) ) { - activatedMessages = reconcileAuthoritativeMessages(persisted.messages, activatedMessages) + const persistedMessages = toChatMessages(persisted.messages) + const runtimeMessages = toChatMessages(activated.messages) + const previousMessages = removeRepresentedLocalLiveProjection(cachedViewState.messages, activated) + + const liveProjection = dedupeInflightUserAgainstTranscript( + persistedMessages, + runtimeMessages, + activated + ) + + activatedMessages = reconcileAuthoritativeChatMessages( + persistedMessages, + previousMessages, + liveProjection + ) } } + const currentMessages = sessionStateByRuntimeIdRef.current.get(cachedRuntimeId)?.messages + + if (currentMessages) { + activatedMessages = overlayConcurrentMessageChanges( + activatedMessages, + cachedViewState.messages, + currentMessages + ) + } + const activatedState = updateSessionState( cachedRuntimeId, state => ({ @@ -967,6 +1004,7 @@ export function useSessionActions({ let prefetchApplied = false let prefetchedStoredSessionId: string | null = null + let prefetchedTranscriptMessages: ChatMessage[] | null = null // REST transcript prefetch and the gateway resume RPC are independent // — run them concurrently so a big session's wall time is @@ -975,6 +1013,8 @@ export function useSessionActions({ // Watch windows skip the prefetch — lazy resume attaches the live mirror. const prefetchPromise = watchWindow ? null : getLatestSessionMessages(storedSessionId, sessionProfile) + let resumeRuntimeBaselineMessages: ChatMessage[] = [] + const resumePromise = requestGateway('session.resume', { session_id: storedSessionId, cols: 96, @@ -988,6 +1028,11 @@ export function useSessionActions({ // background while the prefetch above paints the transcript. ...(watchWindow ? { lazy: true } : { omit_messages: true }), ...(sessionProfile ? { profile: sessionProfile } : {}) + }).then(resumed => { + resumeRuntimeBaselineMessages = + sessionStateByRuntimeIdRef.current.get(resumed.session_id)?.messages ?? resumeRuntimeBaselineMessages + + return resumed }) // The rejection is consumed by the `await` below; this guard only @@ -1018,7 +1063,8 @@ export function useSessionActions({ ? preserveLocalPendingTurnMessages($messages.get(), resumeStartMessages) : $messages.get() - localSnapshot = reconcileAuthoritativeMessages(prefetchedResult.messages, previousMessages) + prefetchedTranscriptMessages = toChatMessages(prefetchedResult.messages) + localSnapshot = reconcileAuthoritativeChatMessages(prefetchedTranscriptMessages, previousMessages) prefetchApplied = true prefetchedStoredSessionId = prefetchedResult.session_id || storedSessionId } @@ -1037,26 +1083,63 @@ export function useSessionActions({ const hasLiveProjection = Boolean(resumed.inflight || resumed.queued) - const preferredMessages = - prefetchApplied && prefetchMatchesResumedSession && !hasLiveProjection - ? localSnapshot - : (() => { - const previousMessages = resumedSameSelectedSession - ? preserveLocalPendingTurnMessages(currentMessages, resumeStartMessages) - : currentMessages + const preferredMessages = (() => { + if (prefetchApplied && prefetchMatchesResumedSession) { + if (hasLiveProjection && prefetchedTranscriptMessages) { + const runtimeMessages = toChatMessages(resumed.messages) + const previousMessages = removeRepresentedLocalLiveProjection(currentMessages, resumed) - // Omitted, not empty — same trap as the activate path above. - // The REST prefetch IS the transcript here; the resume payload - // only contributes the live tail, so graft rather than rebuild. - // (Without a usable prefetch there is nothing better to stand - // on, so the projection alone remains the degraded fallback.) - const resumedMessages = - resumed.messages_omitted && prefetchApplied && prefetchMatchesResumedSession - ? appendLiveSessionProjection(localSnapshot, resumed) - : reconcileAuthoritativeMessages(resumed.messages, previousMessages, resumed) + // Omitted-messages resumes stay safe here: `resumed.messages` + // is empty, so `runtimeMessages` has no anchor and the dedupe + // helper returns the projection unchanged, while the REST + // prefetch below remains the authoritative transcript — the + // same "graft, don't rebuild" outcome the pre-restructure + // messages_omitted branch produced. + const liveProjection = dedupeInflightUserAgainstTranscript( + prefetchedTranscriptMessages, + runtimeMessages, + resumed + ) - return chatMessageArraysEquivalent(currentMessages, resumedMessages) ? currentMessages : resumedMessages - })() + const resumedMessages = reconcileAuthoritativeChatMessages( + prefetchedTranscriptMessages, + previousMessages, + liveProjection + ) + + const withConcurrentChanges = overlayConcurrentMessageChanges( + resumedMessages, + localSnapshot, + currentMessages + ) + + return chatMessageArraysEquivalent(currentMessages, withConcurrentChanges) + ? currentMessages + : withConcurrentChanges + } + + if (!hasLiveProjection) { + return localSnapshot + } + } + + const previousMessages = resumedSameSelectedSession + ? preserveLocalPendingTurnMessages(currentMessages, resumeStartMessages) + : currentMessages + + const resumedMessages = reconcileAuthoritativeMessages(resumed.messages, previousMessages, resumed) + + return chatMessageArraysEquivalent(currentMessages, resumedMessages) ? currentMessages : resumedMessages + })() + + const currentRuntimeMessages = + sessionStateByRuntimeIdRef.current.get(resumed.session_id)?.messages ?? resumeRuntimeBaselineMessages + + const preferredWithRuntimeChanges = overlayConcurrentMessageChanges( + preferredMessages, + resumeRuntimeBaselineMessages, + currentRuntimeMessages + ) // #70449: same stale-snapshot guard as the warm path — a turn that // started while the resume RPC was in flight has already marked the @@ -1071,20 +1154,16 @@ export function useSessionActions({ // (persisted by use-session-state-cache while the turn streamed; // survives renderer/app death) back onto the restored transcript. The // backend's own inflight projection is already inside - // `preferredMessages` (appendLiveSessionProjection), so this merge only - // adds the locally recorded structure — tool calls, sealed interim - // rows — that the backend's text-only snapshot cannot carry. A no-op - // returns `preferredMessages` by reference, keeping the fast path - // below intact. - const inFlightRecovery = recoverInFlightTurnJournal(storedSessionId, preferredMessages, { + // `preferredWithRuntimeChanges`, so this merge only adds the locally + // recorded structure that the backend's text-only snapshot cannot carry. + const inFlightRecovery = recoverInFlightTurnJournal(storedSessionId, preferredWithRuntimeChanges, { keepPending: resumedRunning }) recoveredInFlightTail = inFlightRecovery.applied - // Prefetch-hit fast path: `preferredMessages` IS the live `$messages` - // array (already error-merged when `localSnapshot` was built), so reuse - // the ref instead of rebuilding a throwaway transcript+Map every switch. + // Prefetch-hit fast path: reuse the live array when neither runtime + // changes nor in-flight recovery changed the reconciled transcript. const messagesForView = inFlightRecovery.messages === currentMessages ? currentMessages diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/utils.test.ts b/apps/desktop/src/app/session/hooks/use-session-actions/utils.test.ts index 1e2667e937..bfe95ae015 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/utils.test.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/utils.test.ts @@ -13,7 +13,7 @@ import { setSelectedStoredSessionId, workspaceCwdBelongsToSelectedSession } from '@/store/session' -import type { SessionInfo } from '@/types/hermes' +import type { SessionInfo, SessionResumeResponse } from '@/types/hermes' import { appendLiveSessionProjection, @@ -22,9 +22,12 @@ import { chatMessageArraysEquivalent, chatMessagesEquivalent, chatPartsEquivalent, + dedupeInflightUserAgainstTranscript, isSessionGoneError, + overlayConcurrentMessageChanges, preserveLocalPendingTurnMessages, reconcileResumeMessages, + removeRepresentedLocalLiveProjection, resolveResumedBusy, sessionMatchesStoredId, sessionShouldHaveTranscript, @@ -1447,3 +1450,177 @@ describe('resolveResumedBusy', () => { expect(resolveResumedBusy(true, true)).toBe(true) }) }) + +const runningProjection = (user: string): SessionResumeResponse => + ({ + session_id: 'runtime-1', + session_key: 'stored-1', + resumed: 'stored-1', + message_count: 2, + messages: [], + running: true, + inflight: { user, assistant: 'partial answer', streaming: true } + }) as SessionResumeResponse + +describe('dedupeInflightUserAgainstTranscript', () => { + it('retains the in-flight user source only when it already exists after the runtime anchor', () => { + const runtime = [ + msg('runtime-user', 'user', 'earlier prompt', { timestamp: 1 }), + msg('runtime-assistant', 'assistant', 'earlier answer', { timestamp: 2 }) + ] + + const persisted = [...runtime, msg('persisted-current', 'user', 'current prompt', { timestamp: 3 })] + + const deduped = dedupeInflightUserAgainstTranscript(persisted, runtime, runningProjection('current prompt')) + + expect(deduped.inflight?.user).toBe('current prompt') + expect(deduped.inflight?.assistant).toBe('partial answer') + }) + + it('preserves the assistant boundary before a queued turn when the persisted in-flight user has no delta', () => { + const runtime = [ + msg('runtime-user', 'user', 'earlier prompt', { timestamp: 1 }), + msg('runtime-assistant', 'assistant', 'earlier answer', { timestamp: 2 }) + ] + + const persisted = [...runtime, msg('persisted-current', 'user', 'current prompt', { timestamp: 3 })] + + const projection = { + ...runningProjection('current prompt'), + inflight: { user: 'current prompt', assistant: '', streaming: false }, + queued: { user: 'queued prompt' } + } + + const deduped = dedupeInflightUserAgainstTranscript(persisted, runtime, projection) + const restored = appendLiveSessionProjection(persisted, deduped) + + expect(restored.map(message => message.role)).toEqual(['user', 'assistant', 'user', 'assistant', 'user']) + expect(restored.slice(-2).map(message => message.id)).toEqual([ + 'assistant-stream-runtime-1', + 'user-queued-runtime-1' + ]) + }) + + it('preserves an intentionally repeated prompt when the match is before the runtime anchor', () => { + const runtime = [ + msg('runtime-user', 'user', 'repeat this', { timestamp: 1 }), + msg('runtime-assistant', 'assistant', 'finished answer', { timestamp: 2 }) + ] + + const projection = runningProjection('repeat this') + const unchanged = dedupeInflightUserAgainstTranscript(runtime, runtime, projection) + + expect(unchanged).toBe(projection) + expect(unchanged.inflight?.user).toBe('repeat this') + }) + + it('preserves a repeated in-flight prompt when the persisted match already has an answer', () => { + const runtime = [ + msg('runtime-user', 'user', 'earlier prompt', { timestamp: 1 }), + msg('runtime-assistant', 'assistant', 'earlier answer', { timestamp: 2 }) + ] + + const persisted = [ + ...runtime, + msg('persisted-repeat', 'user', 'repeat this', { timestamp: 3 }), + msg('persisted-repeat-answer', 'assistant', 'finished repeat answer', { timestamp: 4 }) + ] + + const projection = runningProjection('repeat this') + const unchanged = dedupeInflightUserAgainstTranscript(persisted, runtime, projection) + + expect(unchanged).toBe(projection) + expect(unchanged.inflight?.user).toBe('repeat this') + }) +}) + +describe('removeRepresentedLocalLiveProjection', () => { + it('removes only matched synthetic rows from the open local tail', () => { + const previous = [ + msg('user-old-optimistic', 'user', 'current prompt'), + msg('assistant-complete', 'assistant', 'finished answer'), + msg('user-current', 'user', 'current prompt'), + msg('assistant-stream-current', 'assistant', 'partial answer', { pending: true }), + msg('user-queued-runtime', 'user', 'queued prompt'), + msg('user-racing', 'user', 'new racing prompt') + ] + + const projection = { + ...runningProjection('current prompt'), + queued: { user: 'queued prompt' } + } + + const remaining = removeRepresentedLocalLiveProjection(previous, projection) + + expect(remaining.map(message => message.id)).toEqual(['user-old-optimistic', 'assistant-complete', 'user-racing']) + }) + + it('preserves an ambiguous text-identical local race prompt without a matching stream boundary', () => { + const previous = [ + msg('runtime-assistant', 'assistant', 'finished answer'), + msg('user-racing', 'user', 'repeat this') + ] + + const projection = runningProjection('repeat this') + + expect(removeRepresentedLocalLiveProjection(previous, projection)).toBe(previous) + }) + + it('does not consume a generic racing user as the activation-owned queued row', () => { + const previous = [ + msg('runtime-assistant', 'assistant', 'finished answer'), + msg('user-current', 'user', 'current prompt'), + msg('assistant-stream-current', 'assistant', 'partial answer', { pending: true }), + msg('user-racing', 'user', 'repeat this') + ] + + const projection = { + ...runningProjection('current prompt'), + queued: { user: 'repeat this' } + } + + const remaining = removeRepresentedLocalLiveProjection(previous, projection) + + expect(remaining.map(message => message.id)).toEqual(['runtime-assistant', 'user-racing']) + }) +}) + +describe('overlayConcurrentMessageChanges', () => { + it('does not replace an authoritative row with an unchanged baseline cache row', () => { + const baseline = [msg('shared-assistant', 'assistant', 'stale cached answer')] + const authoritative = [msg('shared-assistant', 'assistant', 'completed persisted answer')] + + const overlaid = overlayConcurrentMessageChanges(authoritative, baseline, baseline) + + expect(overlaid).toBe(authoritative) + expect(overlaid[0].parts).toEqual([{ type: 'text', text: 'completed persisted answer' }]) + }) + + it('replaces an activation stream placeholder and appends rows created after the baseline', () => { + const baseline = [msg('assistant-stream-runtime', 'assistant', 'partial A', { pending: true })] + const authoritative = [msg('assistant-stream-activation', 'assistant', 'partial A', { pending: true })] + + const current = [ + msg('assistant-stream-runtime', 'assistant', 'partial A + delta B', { pending: true }), + msg('user-racing', 'user', 'racing prompt') + ] + + const overlaid = overlayConcurrentMessageChanges(authoritative, baseline, current) + + expect(overlaid.map(message => message.id)).toEqual(['assistant-stream-runtime', 'user-racing']) + expect(overlaid[0].parts).toEqual([{ type: 'text', text: 'partial A + delta B' }]) + }) + + it('merges an activation prefix with a baseline-new runtime delta chunk', () => { + const authoritative = [msg('assistant-stream-activation', 'assistant', 'partial A', { pending: true })] + const current = [msg('assistant-stream-runtime', 'assistant', ' + delta B', { pending: true })] + + const overlaid = overlayConcurrentMessageChanges(authoritative, [], current) + + expect(overlaid.map(message => message.id)).toEqual(['assistant-stream-runtime']) + expect(overlaid[0].parts).toEqual([ + { type: 'text', text: 'partial A' }, + { type: 'text', text: ' + delta B' } + ]) + }) +}) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts b/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts index cfa9477f9b..df7049b61b 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts @@ -683,10 +683,17 @@ export function preserveLocalPendingTurnMessages( * memory. Stable ids let repeated activate/resume hydration reconcile instead * of growing duplicate rows. */ -export function appendLiveSessionProjection( - messages: ChatMessage[], - projection: Pick -): ChatMessage[] { +const safelyPersistedInflightUser = Symbol('safelyPersistedInflightUser') + +type LiveSessionProjection = Pick & { + [safelyPersistedInflightUser]?: true +} + +type ReconciledSessionResumeResponse = SessionResumeResponse & { + [safelyPersistedInflightUser]?: true +} + +export function appendLiveSessionProjection(messages: ChatMessage[], projection: LiveSessionProjection): ChatMessage[] { const inflightUser = projection.inflight?.user?.trim() ?? '' const inflightAssistant = projection.inflight?.assistant ?? '' const inflightStreaming = Boolean(projection.inflight?.streaming) @@ -762,7 +769,8 @@ export function appendLiveSessionProjection( message => textWithoutReferenceLines(chatMessageText(message)) === textWithoutReferenceLines(text) ) - const inflightUserAlreadyPersisted = Boolean(inflightUser) && persistedInLatestRun(inflightUser) + const inflightUserAlreadyPersisted = + projection[safelyPersistedInflightUser] === true || (Boolean(inflightUser) && persistedInLatestRun(inflightUser)) if (inflightUser && !inflightUserAlreadyPersisted) { projected.push({ @@ -894,6 +902,224 @@ export function appendLiveSessionProjection( return projected.length ? [...messages, ...projected] : messages } +function normalizedMessageText(message: ChatMessage): string { + return chatMessageText(message).replace(/\s+/g, ' ').trim() +} + +function transcriptAnchorMatches(a: ChatMessage, b: ChatMessage): boolean { + if (a.role !== b.role) { + return false + } + + const aText = normalizedMessageText(a) + const bText = normalizedMessageText(b) + + if (a.timestamp !== undefined && b.timestamp !== undefined) { + return a.timestamp === b.timestamp && aText === bText + } + + return Boolean(aText) && aText === bText +} + +/** + * Mark only an already-materialized `inflight.user` for visual suppression. + * + * A running gateway returns two independent truths: its compressed runtime + * history plus the current in-flight turn, while REST may already have flushed + * that user row into the complete persisted transcript. Global text dedupe is + * unsafe because users may intentionally submit the same prompt twice. Instead, + * find the last runtime message inside the persisted transcript and inspect only + * the newer persisted suffix. + * + * Keep `inflight.user` intact because it also carries turn structure: a queued + * prompt needs its assistant boundary even when the persisted user has no + * assistant delta yet. The private marker lets the renderer suppress only that + * duplicate bubble. If the histories have no safe common anchor, keep the + * projection unchanged — a duplicate is recoverable, but dropping a real + * accepted prompt is not. + */ +export function dedupeInflightUserAgainstTranscript( + persistedMessages: ChatMessage[], + runtimeMessages: ChatMessage[], + projection: SessionResumeResponse +): ReconciledSessionResumeResponse { + const inflightUser = projection.inflight?.user?.replace(/\s+/g, ' ').trim() ?? '' + + if (!inflightUser) { + return projection + } + + let suffixStart = 0 + + if (runtimeMessages.length) { + const runtimeAnchor = runtimeMessages[runtimeMessages.length - 1] + let persistedAnchorIndex = -1 + + for (let index = persistedMessages.length - 1; index >= 0; index -= 1) { + if (transcriptAnchorMatches(persistedMessages[index], runtimeAnchor)) { + persistedAnchorIndex = index + + break + } + } + + if (persistedAnchorIndex < 0) { + return projection + } + + suffixStart = persistedAnchorIndex + 1 + } + + const persistedTail = persistedMessages.slice(suffixStart) + const lastPersistedMessage = persistedTail[persistedTail.length - 1] + + const persistedUserPresent = + lastPersistedMessage?.role === 'user' && normalizedMessageText(lastPersistedMessage) === inflightUser + + if (!persistedUserPresent) { + return projection + } + + return { ...projection, [safelyPersistedInflightUser]: true } +} + +/** + * Drop only synthetic local tail rows that the activation snapshot replaces. + * Unmatched optimistic rows survive so a submit racing with activation is not + * lost; completed transcript rows before the open tail are never considered. + */ +export function removeRepresentedLocalLiveProjection( + previousMessages: ChatMessage[], + projection: Pick +): ChatMessage[] { + const inflightUser = projection.inflight?.user?.replace(/\s+/g, ' ').trim() ?? '' + const inflightAssistant = projection.inflight?.assistant?.replace(/\s+/g, ' ').trim() ?? '' + const queuedUser = projection.queued?.user?.replace(/\s+/g, ' ').trim() ?? '' + + const hasAssistantProjection = Boolean( + projection.inflight?.assistant || projection.inflight?.streaming || (inflightUser && queuedUser) + ) + + if (!inflightUser || !hasAssistantProjection) { + return previousMessages + } + + let openTailStart = 0 + + for (let index = previousMessages.length - 1; index >= 0; index -= 1) { + const message = previousMessages[index] + + if (message.role === 'assistant' && !message.pending) { + openTailStart = index + 1 + + break + } + } + + const inflightUserIndex = previousMessages.findIndex( + (message, index) => + index >= openTailStart && + message.role === 'user' && + message.id.startsWith('user-') && + normalizedMessageText(message) === inflightUser + ) + + const assistantIndex = inflightUserIndex + 1 + const assistant = previousMessages[assistantIndex] + + const assistantMatches = + inflightUserIndex >= openTailStart && + assistant?.role === 'assistant' && + assistant.id.startsWith('assistant-stream-') && + normalizedMessageText(assistant) === inflightAssistant + + if (!assistantMatches) { + return previousMessages + } + + let queuedUserIndex = -1 + + if (queuedUser) { + queuedUserIndex = previousMessages.findIndex( + (message, index) => + index > assistantIndex && + message.role === 'user' && + message.id.startsWith('user-queued-') && + normalizedMessageText(message) === queuedUser + ) + } + + return previousMessages.filter( + (_message, index) => index !== inflightUserIndex && index !== assistantIndex && index !== queuedUserIndex + ) +} + +/** + * Overlay messages that changed while activation waited on REST. Existing ids + * replace the older activation row; only rows added or changed since the warm + * cache baseline are appended. This is identity-based, never text-based. + */ +export function overlayConcurrentMessageChanges( + nextMessages: ChatMessage[], + baselineMessages: ChatMessage[], + currentMessages: ChatMessage[] +): ChatMessage[] { + const baselineById = new Map(baselineMessages.map(message => [message.id, message])) + const nextIndexById = new Map(nextMessages.map((message, index) => [message.id, index])) + let changed = false + const overlaid = [...nextMessages] + + let activationStreamIndex = overlaid.findIndex( + message => + message.role === 'assistant' && message.id.startsWith('assistant-stream-') && !baselineById.has(message.id) + ) + + for (const current of currentMessages) { + const baseline = baselineById.get(current.id) + const changedSinceBaseline = !baseline || !chatMessagesEquivalent(baseline, current) + + if (!changedSinceBaseline) { + continue + } + + const nextIndex = nextIndexById.get(current.id) + + if (nextIndex !== undefined) { + if (!chatMessagesEquivalent(overlaid[nextIndex], current)) { + overlaid[nextIndex] = current + changed = true + } + + continue + } + + if (activationStreamIndex >= 0 && current.role === 'assistant' && current.id.startsWith('assistant-stream-')) { + const activationStream = overlaid[activationStreamIndex] + const activationText = chatMessageText(activationStream) + const currentText = chatMessageText(current) + + const replacement = + activationText && !currentText.startsWith(activationText) + ? { ...current, parts: [...activationStream.parts, ...current.parts] } + : current + + nextIndexById.delete(activationStream.id) + nextIndexById.set(current.id, activationStreamIndex) + overlaid[activationStreamIndex] = replacement + activationStreamIndex = -1 + changed = true + + continue + } + + nextIndexById.set(current.id, overlaid.length) + overlaid.push(current) + changed = true + } + + return changed ? overlaid : nextMessages +} + export interface BranchMessage { content: string role: ChatMessage['role']