From 1622e62c2bc8895866f15ec35c43ada7b8fe6f2e Mon Sep 17 00:00:00 2001 From: umutcagand Date: Mon, 24 Aug 2026 00:08:08 +0300 Subject: [PATCH 1/3] fix(cli): preserve state after interrupted runs --- cli/src/hooks/helpers/send-message.ts | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/cli/src/hooks/helpers/send-message.ts b/cli/src/hooks/helpers/send-message.ts index f5432dbc97..6debd1e80b 100644 --- a/cli/src/hooks/helpers/send-message.ts +++ b/cli/src/hooks/helpers/send-message.ts @@ -272,6 +272,7 @@ export const setupStreamingContext = (params: { setMessages: (updater: (messages: ChatMessage[]) => ChatMessage[]) => void streamRefs: StreamController abortController?: AbortController + onAbort?: () => void setStreamStatus: (status: StreamStatus) => void setCanProcessQueue: (can: boolean) => void isQueuePausedRef?: MutableRefObject @@ -303,13 +304,16 @@ export const setupStreamingContext = (params: { const abortController = params.abortController ?? new AbortController() abortController.signal.addEventListener('abort', () => { + try { + params.onAbort?.() + } catch { + // Checkpoint callbacks are best-effort; never skip abort cleanup. + } + // Abort means the user stopped streaming; update UI with an interruption notice. - // Release the chain lock immediately so new messages can be sent directly instead - // of being queued. The minor trade-off is that if the user sends a new message - // before client.run() resolves, it may use stale previousRunStateRef. This is - // acceptable because: (1) the user explicitly cancelled, and (2) client.run() - // will update previousRunStateRef when it eventually resolves, so subsequent - // runs will have the full state. + // The owner checkpoints its latest SDK snapshot synchronously through onAbort, + // while generation guards prevent late results from an older run from replacing + // state selected by a newer run. streamRefs.setters.setWasAbortedByUser(true) setIsRetrying(false) timerController.stop('aborted') From 3b5e2f921801883da172ae6fa4259fc02213c974 Mon Sep 17 00:00:00 2001 From: umutcagand Date: Mon, 24 Aug 2026 00:08:09 +0300 Subject: [PATCH 2/3] fix(cli): preserve state after interrupted runs --- cli/src/hooks/use-send-message.ts | 76 +++++++++++++++++++++---------- 1 file changed, 52 insertions(+), 24 deletions(-) diff --git a/cli/src/hooks/use-send-message.ts b/cli/src/hooks/use-send-message.ts index 39065317ee..5eeb811f14 100644 --- a/cli/src/hooks/use-send-message.ts +++ b/cli/src/hooks/use-send-message.ts @@ -158,6 +158,9 @@ export const useSendMessage = ({ const previousRunStateRef = useRef( useChatStore.getState().runState, ) + // Incremented for every send so a late result from an interrupted run cannot + // overwrite the state selected by the newer run that replaced it. + const runGenerationRef = useRef(0) // Memoize stream controller to maintain referential stability across renders const streamRefsRef = useRef resolveCurrentChatDir() === runChatDir + const runIsCurrent = () => + runGenerationRef.current === runGeneration && runChatIsCurrent() let latestRunStateSnapshot: RunState = previousRunStateRef.current ?? { traceSessionId: randomUUID(), output: { @@ -506,6 +515,11 @@ export const useSendMessage = ({ setMessages, streamRefs, abortController, + onAbort: () => { + if (runGenerationRef.current !== runGeneration) return + previousRunStateRef.current = latestRunStateSnapshot + setRunState(latestRunStateSnapshot) + }, setStreamStatus, setCanProcessQueue, isQueuePausedRef, @@ -551,7 +565,7 @@ export const useSendMessage = ({ ) const eventHandlerState = createEventHandlerState({ - isActive: () => !abortController.signal.aborted && runChatIsCurrent(), + isActive: () => !abortController.signal.aborted && runIsCurrent(), streamRefs, setStreamingAgents, setStreamStatus, @@ -596,7 +610,7 @@ export const useSendMessage = ({ // conversation, and checkpointing them into this run's directory // would overwrite that chat's transcript with foreign (possibly // empty) state — the chat would then be hidden from /history. - if (abortController.signal.aborted || !runChatIsCurrent()) { + if (abortController.signal.aborted || !runIsCurrent()) { return } // Persist asynchronously and coalescing: the periodic snapshot @@ -642,7 +656,7 @@ export const useSendMessage = ({ // context, and previousRunStateRef/setRunState would leak this run's // agent state into the other chat. (A plain Esc interrupt keeps the // same chat, so the interrupted turn is still saved as before.) - if (runChatIsCurrent()) { + if (runIsCurrent()) { // Finalize: persist state and mark complete previousRunStateRef.current = runState setRunState(runState) @@ -657,29 +671,31 @@ export const useSendMessage = ({ // traps is several times slower. saveChatState(runState, useChatStore.getState().messages, runChatDir) } - handleRunCompletion({ - runState, - actualCredits, - agentMode, - timerController, - updater, - aiMessageId, - wasAbortedByUser: abortController.signal.aborted, - hasReceivedContent: hasReceivedContentRef.current, - setStreamStatus, - setCanProcessQueue, - updateChainInProgress, - setHasReceivedPlanResponse, - resumeQueue, - isProcessingQueueRef, - isQueuePausedRef, - }) + if (runIsCurrent()) { + handleRunCompletion({ + runState, + actualCredits, + agentMode, + timerController, + updater, + aiMessageId, + wasAbortedByUser: abortController.signal.aborted, + hasReceivedContent: hasReceivedContentRef.current, + setStreamStatus, + setCanProcessQueue, + updateChainInProgress, + setHasReceivedPlanResponse, + resumeQueue, + isProcessingQueueRef, + isQueuePausedRef, + }) + } } catch (error) { // If this run was aborted, the abort handler already handled cleanup. // Don't run error handling to avoid interfering with any new run that // may have started. Uses per-run abortController.signal (not shared // streamRefs) so a newer run's reset() can't clear this flag. - if (!abortController.signal.aborted) { + if (!abortController.signal.aborted && runIsCurrent()) { handleRunError({ error, timerController, @@ -692,11 +708,18 @@ export const useSendMessage = ({ isQueuePausedRef, hasReceivedContent: hasReceivedContentRef.current, }) + // Keep the latest successful SDK snapshot available to the next + // message in this process, not only on disk. Without this, a failed + // or expired turn is followed by a fresh run with stale history. + if (runIsCurrent()) { + previousRunStateRef.current = latestRunStateSnapshot + setRunState(latestRunStateSnapshot) + } // Persist the last checkpoint plus the error banner so a restart // after a failed run still shows this turn. Settle async checkpoints // first so a stale write can't clobber this one. Skipped after a // mid-run chat switch — the store's messages belong to the new chat. - if (runChatIsCurrent()) { + if (runIsCurrent()) { await settleCheckpointSave() saveChatState( latestRunStateSnapshot, @@ -704,8 +727,13 @@ export const useSendMessage = ({ runChatDir, ) } - } else { + } else if (abortController.signal.aborted) { logger.debug({ error }, '[send-message] Ignoring error after abort') + } else { + logger.debug( + { error }, + '[send-message] Ignoring error after run superseded', + ) } } finally { // Stop exit-flushing this run's checkpoint; the final state (or last @@ -717,7 +745,7 @@ export const useSendMessage = ({ // interfering with any new run that may have started after the abort. // Uses per-run abortController.signal (not shared streamRefs) so a newer // run's reset() can't clear this flag. - if (!abortController.signal.aborted) { + if (!abortController.signal.aborted && runIsCurrent()) { if (isChainInProgressRef.current) { logger.warn( {}, From e858eefb8c31d214d7fce00fbdce0530a4917122 Mon Sep 17 00:00:00 2001 From: umutcagand Date: Mon, 24 Aug 2026 00:08:10 +0300 Subject: [PATCH 3/3] fix(cli): preserve state after interrupted runs --- .../helpers/__tests__/send-message.test.ts | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/cli/src/hooks/helpers/__tests__/send-message.test.ts b/cli/src/hooks/helpers/__tests__/send-message.test.ts index 7e23893249..d07654c389 100644 --- a/cli/src/hooks/helpers/__tests__/send-message.test.ts +++ b/cli/src/hooks/helpers/__tests__/send-message.test.ts @@ -289,6 +289,32 @@ describe('setupStreamingContext', () => { expect(abortController).toBe(ownedAbortController) }) + test('invokes the owner checkpoint callback before abort cleanup', () => { + let messages = createBaseMessages() + const streamRefs = createStreamController() + const timerController = createMockTimerController() + const onAbort = mock(() => {}) + + const { abortController } = setupStreamingContext({ + aiMessageId: 'ai-1', + timerController, + setMessages: (fn: any) => { + messages = fn(messages) + }, + streamRefs, + onAbort, + setStreamStatus: () => {}, + setCanProcessQueue: () => {}, + updateChainInProgress: () => {}, + setIsRetrying: () => {}, + setStreamingAgents: () => {}, + }) + + abortController.abort() + + expect(onAbort).toHaveBeenCalledTimes(1) + }) + test('setupStreamingContext resets streamRefs and starts timer', () => { let messages = createBaseMessages() const streamRefs = createStreamController()