fix(cli): resolve message queue race conditions and add recovery mechanisms

- Fix React concurrent mode race condition by reading from ref before setState
- Remove buggy startStreaming line that conflated busy/paused states
- Add resetEarlyReturnState helper for DRY queue state resets
- Set isChainInProgressRef synchronously to prevent race conditions
- Add 5-minute watchdog timer to recover from stuck queue locks
- Simplify queuePaused to only reflect user-initiated pause state
- Add unit tests for early return queue state reset scenarios
- Delete unused use-committed-value.ts file
- Update tests to use real resetEarlyReturnState instead of mock

Author: brandonkachen

cli/src/hooks/helpers/__tests__/send-message.test.ts

@@ -28,7 +28,7 @@ ensureEnv()
 
 const { useChatStore } = await import('../../../state/chat-store')
 const { createStreamController } = await import('../../stream-state')
-const { setupStreamingContext, handleRunError, finalizeQueueState } = await import(
+const { setupStreamingContext, handleRunError, finalizeQueueState, resetEarlyReturnState } = await import(
   '../send-message'
 )
 const { createBatchedMessageUpdater } = await import(
@@ -113,7 +113,7 @@ describe('setupStreamingContext', () => {
       // Verify stream status reset
       expect(streamStatus).toBe('idle')
 
-      // Verify queue processing enabled (no isQueuePausedRef)
+      // Verify queue processing enabled (no pause ref)
       expect(canProcessQueue).toBe(true)
 
       // Verify chain in progress reset
@@ -170,7 +170,7 @@ describe('setupStreamingContext', () => {
       // Trigger abort
       abortController.abort()
 
-      // When queue is paused, canProcessQueue should be false
+      // When queue was paused before streaming, canProcessQueue should be false
       expect(canProcessQueue).toBe(false)
     })
 
@@ -374,7 +374,7 @@ describe('finalizeQueueState', () => {
       isQueuePausedRef,
     })
 
-    // When queue is paused, canProcessQueue should be false
+    // When queue was paused before streaming, canProcessQueue should be false
     expect(canProcessQueue).toBe(false)
   })
 })
@@ -583,7 +583,7 @@ describe('handleRunError', () => {
       isQueuePausedRef,
     })
 
-    // When queue is paused, canProcessQueue should be false
+    // When queue was paused before streaming, canProcessQueue should be false
     expect(canProcessQueue).toBe(false)
   })
 
@@ -718,3 +718,248 @@ describe('handleRunError', () => {
     expect(timerController.stopCalls).toContain('error')
   })
 })
+
+/**
+ * Tests for early return queue state reset in sendMessage.
+ * These test the resetEarlyReturnState helper used across multiple early return paths:
+ * - prepareUserMessage exception
+ * - validation failure (success: false)
+ * - validation exception
+ */
+describe('resetEarlyReturnState', () => {
+  describe('prepareUserMessage exception path', () => {
+    test('resets chain in progress to false', () => {
+      let chainInProgress = true
+
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: () => {},
+      })
+
+      expect(chainInProgress).toBe(false)
+    })
+
+    test('sets canProcessQueue to true when queue is not paused', () => {
+      let canProcessQueue = false
+      const isQueuePausedRef = { current: false }
+
+      resetEarlyReturnState({
+        updateChainInProgress: () => {},
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isQueuePausedRef,
+      })
+
+      expect(canProcessQueue).toBe(true)
+    })
+
+    test('sets canProcessQueue to false when queue is paused', () => {
+      let canProcessQueue = true
+      const isQueuePausedRef = { current: true }
+
+      resetEarlyReturnState({
+        updateChainInProgress: () => {},
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isQueuePausedRef,
+      })
+
+      expect(canProcessQueue).toBe(false)
+    })
+
+    test('resets isProcessingQueueRef to false', () => {
+      const isProcessingQueueRef = { current: true }
+
+      resetEarlyReturnState({
+        updateChainInProgress: () => {},
+        setCanProcessQueue: () => {},
+        isProcessingQueueRef,
+      })
+
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+
+    test('handles missing isProcessingQueueRef gracefully', () => {
+      // Should not throw when isProcessingQueueRef is undefined
+      expect(() => {
+        resetEarlyReturnState({
+          updateChainInProgress: () => {},
+          setCanProcessQueue: () => {},
+        })
+      }).not.toThrow()
+    })
+
+    test('handles missing isQueuePausedRef gracefully (defaults to canProcessQueue=true)', () => {
+      let canProcessQueue = false
+
+      resetEarlyReturnState({
+        updateChainInProgress: () => {},
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        // No isQueuePausedRef - should default to !undefined = true
+      })
+
+      expect(canProcessQueue).toBe(true)
+    })
+  })
+
+  describe('validation failure path (success: false)', () => {
+    test('resets all queue state correctly when processing queued message', () => {
+      let chainInProgress = true
+      let canProcessQueue = false
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: false }
+
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      expect(chainInProgress).toBe(false)
+      expect(canProcessQueue).toBe(true)
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+
+    test('respects queue paused state after validation failure', () => {
+      let chainInProgress = true
+      let canProcessQueue = true
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: true }
+
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      expect(chainInProgress).toBe(false)
+      expect(canProcessQueue).toBe(false) // Queue was paused, should stay paused
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+  })
+
+  describe('validation exception path', () => {
+    test('resets all queue state correctly when validation throws', () => {
+      let chainInProgress = true
+      let canProcessQueue = false
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: false }
+
+      // Simulating what happens after catching validation exception
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      expect(chainInProgress).toBe(false)
+      expect(canProcessQueue).toBe(true)
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+
+    test('preserves queue pause state when validation throws', () => {
+      let canProcessQueue = true
+      const isQueuePausedRef = { current: true }
+      const isProcessingQueueRef = { current: true }
+
+      resetEarlyReturnState({
+        updateChainInProgress: () => {},
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      // Queue was explicitly paused before, should remain paused after error
+      expect(canProcessQueue).toBe(false)
+      // But processing lock should be released to allow manual resume
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+  })
+
+  describe('complete early return scenarios', () => {
+    test('queue can process next message after prepareUserMessage exception', () => {
+      // Scenario: Message was being processed from queue, prepareUserMessage throws
+      let chainInProgress = true
+      let canProcessQueue = false
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: false }
+
+      // After exception, reset is called
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      // Queue should be able to process next message
+      expect(chainInProgress).toBe(false)
+      expect(canProcessQueue).toBe(true)
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+
+    test('queue can process next message after validation returns success=false', () => {
+      // Scenario: Message was being processed, validation returns failure
+      let chainInProgress = true
+      let canProcessQueue = false
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: false }
+
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      // All locks released, queue can continue
+      expect(chainInProgress).toBe(false)
+      expect(canProcessQueue).toBe(true)
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+
+    test('queue can process next message after validation throws exception', () => {
+      // Scenario: Message was being processed, validation throws
+      let chainInProgress = true
+      let canProcessQueue = false
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: false }
+
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      // All locks released, queue can continue
+      expect(chainInProgress).toBe(false)
+      expect(canProcessQueue).toBe(true)
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+
+    test('queue remains blocked after error if user had paused it', () => {
+      // Scenario: User paused queue, then an error occurred
+      // Queue should remain paused after error recovery
+      let chainInProgress = true
+      let canProcessQueue = true
+      const isProcessingQueueRef = { current: true }
+      const isQueuePausedRef = { current: true } // User explicitly paused
+
+      resetEarlyReturnState({
+        updateChainInProgress: (value) => { chainInProgress = value },
+        setCanProcessQueue: (can) => { canProcessQueue = can },
+        isProcessingQueueRef,
+        isQueuePausedRef,
+      })
+
+      // Chain is no longer in progress
+      expect(chainInProgress).toBe(false)
+      // But queue should remain blocked because user paused it
+      expect(canProcessQueue).toBe(false)
+      // Processing lock is released though
+      expect(isProcessingQueueRef.current).toBe(false)
+    })
+  })
+})

cli/src/hooks/helpers/send-message.ts

@@ -35,6 +35,29 @@ import type { StreamStatus } from '../use-message-queue'
 import type { MessageContent, RunState } from '@codebuff/sdk'
 import type { MutableRefObject, SetStateAction } from 'react'
 
+/** Resets queue state on early return (before streaming starts). */
+export type ResetEarlyReturnStateParams = {
+  setCanProcessQueue: (can: boolean) => void
+  updateChainInProgress: (value: boolean) => void
+  isProcessingQueueRef?: MutableRefObject<boolean>
+  isQueuePausedRef?: MutableRefObject<boolean>
+}
+
+export const resetEarlyReturnState = (params: ResetEarlyReturnStateParams): void => {
+  const {
+    setCanProcessQueue,
+    updateChainInProgress,
+    isProcessingQueueRef,
+    isQueuePausedRef,
+  } = params
+
+  updateChainInProgress(false)
+  setCanProcessQueue(!isQueuePausedRef?.current)
+  if (isProcessingQueueRef) {
+    isProcessingQueueRef.current = false
+  }
+}
+
 /** Resets queue state after streaming completes, aborts, or errors. */
 export type FinalizeQueueStateParams = {
   setStreamStatus: (status: StreamStatus) => void
@@ -164,7 +187,7 @@ export const prepareUserMessage = async (params: {
       next = postUserMessage(next)
     }
     if (next.length > 100) {
-      return next.slice(-100)
+      next = next.slice(-100)
     }
     return next
   })