From 76151d4b501e98d39535b95fc795ed75f8a9dbb5 Mon Sep 17 00:00:00 2001 From: Colby Aregullin Date: Tue, 25 Aug 2026 12:25:12 -0600 Subject: [PATCH 1/2] fix(chat): stop stranding queued sends rejected before commit A queued send that is rejected pre-commit was retried exactly once, and that retry bailed out if the session was not ready when the timer fired. During draft promotion the session is not ready yet, so the single retry burned on the readiness check. Pre-commit rejection is silent, leaving no store transition to re-trigger the drain, so the message sat in the queue forever with no visible failure. Retries now back off exponentially up to 30s and re-arm when the session is still not ready instead of abandoning the record. Any later readiness transition resets the backoff. LAWS/CHAT.md requires the queue to resume dispatching once the session becomes ready. Separately, queues are keyed by session id, and a session that is still creating only has a client-local draft id. Persisting its queue restores, on next launch, a record bound to an id no backend session will ever have, which can never drain. Quitting mid-creation now drops the message rather than stranding it. Co-Authored-By: Claude Fable 5 --- .../hooks/__tests__/useMessageQueue.test.ts | 42 +++++++----- src/features/chat/hooks/useMessageQueue.ts | 58 +++++++++++----- .../chat/stores/queuePersistence.test.ts | 67 +++++++++++++++++++ src/features/chat/stores/queuePersistence.ts | 12 +++- 4 files changed, 148 insertions(+), 31 deletions(-) diff --git a/src/features/chat/hooks/__tests__/useMessageQueue.test.ts b/src/features/chat/hooks/__tests__/useMessageQueue.test.ts index 7936a1f28..9f537990a 100644 --- a/src/features/chat/hooks/__tests__/useMessageQueue.test.ts +++ b/src/features/chat/hooks/__tests__/useMessageQueue.test.ts @@ -1279,7 +1279,10 @@ describe("useMessageQueue", () => { expect(useChatStore.getState().queuedMessageBySession.s1).toBeUndefined(); }); - it("retries a retained pre-commit failure once while the session stays ready", async () => { + it("keeps retrying a retained pre-commit failure with backoff while the session stays ready", async () => { + // LAWS/CHAT.md: the queue must resume sending when the session is ready. + // Pre-commit rejections are silent and leave no store transition behind, + // so abandoning the record after a fixed retry count strands it forever. vi.useFakeTimers(); const sendMessage = vi.fn().mockResolvedValue(false); useChatStore.getState().enqueueTransportReadyMessage("s1", { @@ -1297,12 +1300,21 @@ describe("useMessageQueue", () => { expect(sendMessage).toHaveBeenCalledTimes(2); await act(async () => { - await vi.advanceTimersByTimeAsync(60_000); + await vi.advanceTimersByTimeAsync(2_000); }); - expect(sendMessage).toHaveBeenCalledTimes(2); - expect( - useChatStore.getState().queuedMessageBySession.s1?.[0]?.payload, - ).toMatchObject({ text: "queued" }); + expect(sendMessage).toHaveBeenCalledTimes(3); + + await act(async () => { + await vi.advanceTimersByTimeAsync(4_000); + }); + expect(sendMessage).toHaveBeenCalledTimes(4); + + sendMessage.mockResolvedValue(true); + await act(async () => { + await vi.advanceTimersByTimeAsync(8_000); + }); + expect(sendMessage).toHaveBeenCalledTimes(5); + expect(useChatStore.getState().queuedMessageBySession.s1).toBeUndefined(); vi.useRealTimers(); }); @@ -1428,7 +1440,7 @@ describe("useMessageQueue", () => { }); }); - it("restores one automatic retry on every later readiness transition", async () => { + it("resets the retry backoff on every later readiness transition", async () => { vi.useFakeTimers(); const sendMessage = vi.fn().mockReturnValue(false); useChatStore.getState().enqueueTransportReadyMessage("s1", { @@ -1441,23 +1453,23 @@ describe("useMessageQueue", () => { await act(async () => { await vi.advanceTimersByTimeAsync(1_000); }); - expect(sendMessage).toHaveBeenCalledTimes(2); + await act(async () => { + await vi.advanceTimersByTimeAsync(2_000); + }); + expect(sendMessage).toHaveBeenCalledTimes(3); act(() => { useChatStore.getState().setChatState("s1", "streaming"); useChatStore.getState().setChatState("s1", "idle"); }); - expect(sendMessage).toHaveBeenCalledTimes(3); - - await act(async () => { - await vi.advanceTimersByTimeAsync(1_000); - }); expect(sendMessage).toHaveBeenCalledTimes(4); + // The idle transition cleared the backoff, so the next automatic retry + // fires at the initial one-second delay again. await act(async () => { - await vi.advanceTimersByTimeAsync(60_000); + await vi.advanceTimersByTimeAsync(1_000); }); - expect(sendMessage).toHaveBeenCalledTimes(4); + expect(sendMessage).toHaveBeenCalledTimes(5); expect( useChatStore.getState().queuedMessageBySession.s1?.[0]?.payload, ).toMatchObject({ text: "queued" }); diff --git a/src/features/chat/hooks/useMessageQueue.ts b/src/features/chat/hooks/useMessageQueue.ts index cffcdf307..255821351 100644 --- a/src/features/chat/hooks/useMessageQueue.ts +++ b/src/features/chat/hooks/useMessageQueue.ts @@ -40,6 +40,12 @@ interface QueueAttemptLease { // the replacement owner from overlapping the still-live attempt. const queueAttemptLeaseBySession = new Map(); +// LAWS/CHAT.md: the queue must resume sending when the session becomes ready. +// Rejected attempts back off but never abandon the record — a rejection can be +// silent (pre-commit ownership/readiness races around draft promotion) with no +// follow-up store transition to re-trigger the drain. +const MAX_AUTO_RETRY_DELAY_MS = 30_000; + function getQueuedMessageKey( queuedMessage: QueuedMessageRecord | null, ): string | null { @@ -96,9 +102,10 @@ export function useMessageQueue( const dispatchReleasePayloadRef = useRef< QueuedMessageRecord["payload"] | null >(null); - const automaticallyRetriedPayloadRef = useRef< - QueuedMessageRecord["payload"] | null - >(null); + const autoRetryRef = useRef<{ + payload: QueuedMessageRecord["payload"]; + attempts: number; + } | null>(null); const suppressNextRenderIdleCycleRef = useRef(false); const dismissedRecordIdRef = useRef(null); const queuedMessageKey = useMemo( @@ -277,31 +284,52 @@ export function useMessageQueue( retryPayload, ); } - if ( - retryTimerRef.current === null && - automaticallyRetriedPayloadRef.current !== retryPayload - ) { + if (autoRetryRef.current?.payload !== retryPayload) { + autoRetryRef.current = { payload: retryPayload, attempts: 0 }; + } + const scheduleAutoRetry = () => { + if (retryTimerRef.current !== null) return; + const attempts = autoRetryRef.current?.attempts ?? 0; + const delayMs = Math.min( + 1_000 * 2 ** attempts, + MAX_AUTO_RETRY_DELAY_MS, + ); retryTimerRef.current = setTimeout(() => { retryTimerRef.current = null; const state = useChatStore.getState(); const runtime = state.getSessionRuntime(sessionId); const retryHead = state.queuedMessageBySession[sessionId]?.[0]; if ( - !isQueuedSessionReady(runtime, isPreparationReadyRef.current) || getQueuedMessageKey(retryHead) !== key || retryHead?.payload !== retryPayload ) { return; } - automaticallyRetriedPayloadRef.current = retryPayload; + if ( + !isQueuedSessionReady(runtime, isPreparationReadyRef.current) + ) { + // Readiness can return without a transition this hook + // observes; keep the retry armed instead of abandoning the + // record. + lastAttemptRef.current = null; + scheduleAutoRetry(); + return; + } + if (autoRetryRef.current?.payload === retryPayload) { + autoRetryRef.current = { + payload: retryPayload, + attempts: autoRetryRef.current.attempts + 1, + }; + } idleCycleRef.current += 1; tryDrainQueuedMessage(retryHead); - }, 1_000); - } + }, delayMs); + }; + scheduleAutoRetry(); return; } - automaticallyRetriedPayloadRef.current = null; + autoRetryRef.current = null; lastAttemptRef.current = null; useChatStore .getState() @@ -375,7 +403,7 @@ export function useMessageQueue( } if (editedCurrentRecord || advancedToNextRecord) { - automaticallyRetriedPayloadRef.current = null; + autoRetryRef.current = null; if (retryTimerRef.current !== null) { clearTimeout(retryTimerRef.current); retryTimerRef.current = null; @@ -383,7 +411,7 @@ export function useMessageQueue( } if (becameIdle || becameReadyWhileIdle) { - automaticallyRetriedPayloadRef.current = null; + autoRetryRef.current = null; if (retryTimerRef.current !== null) { clearTimeout(retryTimerRef.current); retryTimerRef.current = null; @@ -434,7 +462,7 @@ export function useMessageQueue( } if (queuedMessageKey !== lastAttemptRef.current?.key) { lastAttemptRef.current = null; - automaticallyRetriedPayloadRef.current = null; + autoRetryRef.current = null; if (retryTimerRef.current !== null) { clearTimeout(retryTimerRef.current); retryTimerRef.current = null; diff --git a/src/features/chat/stores/queuePersistence.test.ts b/src/features/chat/stores/queuePersistence.test.ts index f7b723a0b..012ed5796 100644 --- a/src/features/chat/stores/queuePersistence.test.ts +++ b/src/features/chat/stores/queuePersistence.test.ts @@ -9,12 +9,14 @@ import { loadPersistedMessageQueues, persistMessageQueues, } from "./queuePersistence"; +import { useChatSessionStore, type ChatSession } from "./chatSessionStore"; describe("queuePersistence", () => { beforeEach(() => { mockInvoke.mockReset(); window.localStorage.clear(); window.__TAURI_INTERNALS__ = {}; + useChatSessionStore.setState({ sessions: [] }); }); it("loads inline image attachments from native persistence when localStorage is over quota", async () => { @@ -265,6 +267,71 @@ describe("queuePersistence", () => { }); }); + it("drops queue writes for sessions whose creation has not settled", async () => { + mockInvoke.mockResolvedValue(undefined); + useChatSessionStore.setState({ + sessions: [ + { id: "draft-1", creationState: "pending" } as unknown as ChatSession, + { id: "draft-2", creationState: "failed" } as unknown as ChatSession, + ], + }); + + persistMessageQueues( + { + "draft-1": [ + { + kind: "transport-ready", + recordId: "pending-record", + payload: admitSystemInheritedQueuedMessage({ text: "pending" }), + }, + ], + "draft-2": [ + { + kind: "transport-ready", + recordId: "failed-record", + payload: admitSystemInheritedQueuedMessage({ text: "failed" }), + }, + ], + }, + ["draft-1", "draft-2"], + ); + + await vi.waitFor(() => + expect(mockInvoke).toHaveBeenCalledWith("persist_message_queue_updates", { + serializedUpdates: JSON.stringify({ "draft-1": null, "draft-2": null }), + }), + ); + expect( + window.localStorage.getItem("goose:chat-message-queues:v1"), + ).toBeNull(); + }); + + it("persists queues again once the session id belongs to a settled session", async () => { + mockInvoke.mockResolvedValue(undefined); + useChatSessionStore.setState({ + sessions: [{ id: "backend-1" } as unknown as ChatSession], + }); + + persistMessageQueues( + { + "backend-1": [ + { + kind: "transport-ready", + recordId: "promoted-record", + payload: admitSystemInheritedQueuedMessage({ text: "promoted" }), + }, + ], + }, + ["backend-1"], + ); + + await vi.waitFor(() => + expect(mockInvoke).toHaveBeenCalledWith("persist_message_queue_updates", { + serializedUpdates: expect.stringContaining("promoted-record"), + }), + ); + }); + it("writes only changed sessions through native read-modify-write persistence", async () => { mockInvoke.mockResolvedValue(undefined); persistMessageQueues( diff --git a/src/features/chat/stores/queuePersistence.ts b/src/features/chat/stores/queuePersistence.ts index 6d13ac0fa..ca6e328d5 100644 --- a/src/features/chat/stores/queuePersistence.ts +++ b/src/features/chat/stores/queuePersistence.ts @@ -1,4 +1,5 @@ import { invoke } from "@tauri-apps/api/core"; +import { useChatSessionStore } from "./chatSessionStore"; import type { QueuedMessagePayload, QueuedMessageRecord } from "./chatStore"; import { isAdmittedQueuedMessagePayload, @@ -186,6 +187,13 @@ export function loadCachedMessageQueues(): PersistedQueues { } } +// A session still creating (or failed to create) only exists under a +// client-local draft id that no restart can resolve, so persisting its queue +// would strand unreachable records; quitting mid-creation drops the message. +function isSessionPersistable(sessionId: string): boolean { + return !useChatSessionStore.getState().getSession(sessionId)?.creationState; +} + export function persistMessageQueues( queues: PersistedQueues, changedSessionIds: string[], @@ -194,7 +202,9 @@ export function persistMessageQueues( const updates = Object.fromEntries( changedSessionIds.map((sessionId) => [ sessionId, - queues[sessionId]?.length ? queues[sessionId] : null, + queues[sessionId]?.length && isSessionPersistable(sessionId) + ? queues[sessionId] + : null, ]), ); if (window.__TAURI_INTERNALS__) { From 57e06f671c77ca2f437dc5871bec97de90c7132b Mon Sep 17 00:00:00 2001 From: Colby Aregullin Date: Tue, 25 Aug 2026 12:43:49 -0600 Subject: [PATCH 2/2] fix(chat): bound queue retries to transient rejections Retrying until the session is ready is correct for a readiness or ownership race, but the previous commit applied it to every pre-commit rejection, including permanent ones. Auto-compaction returns false whenever compaction fails, and each failure appends an error notification and sets the session back to idle. That idle edge reads as a readiness transition, which both reset the backoff and re-triggered the drain, so a persistently failing send looped as fast as compaction could fail and appended an error notification every pass. A test reproducing twelve such cycles dispatched 73 times before this change. Rejections are now bounded to five per payload, counted at the single drain choke point so every trigger is covered rather than just the retry timer. The counter deliberately survives readiness edges, since the transition a failed send causes must not grant a fresh budget; a user edit or a new head still resets it. The record stays queued and visible, so the send is retryable by hand. Waiting for readiness stays unbounded, because abandoning it is the original stranding bug, but it now backs off to the 30s cap instead of re-arming at a flat one second forever: an unready session woke 960 times in 16 minutes. Co-Authored-By: Claude Fable 5 --- .../hooks/__tests__/useMessageQueue.test.ts | 87 +++++++++++++++++++ src/features/chat/hooks/useMessageQueue.ts | 58 ++++++++++++- 2 files changed, 144 insertions(+), 1 deletion(-) diff --git a/src/features/chat/hooks/__tests__/useMessageQueue.test.ts b/src/features/chat/hooks/__tests__/useMessageQueue.test.ts index 9f537990a..b0fde2442 100644 --- a/src/features/chat/hooks/__tests__/useMessageQueue.test.ts +++ b/src/features/chat/hooks/__tests__/useMessageQueue.test.ts @@ -1318,6 +1318,93 @@ describe("useMessageQueue", () => { vi.useRealTimers(); }); + it("stops automatic retries for a persistently rejected payload despite readiness churn", async () => { + // A failed auto-compaction returns false, appends an error notification, + // and sets the session back to idle. That idle edge re-triggers the drain, + // so without a ceiling the send loops as fast as compaction can fail and + // grows the transcript every pass. + vi.useFakeTimers(); + const sendMessage = vi.fn().mockResolvedValue(false); + useChatStore.getState().enqueueTransportReadyMessage("s1", { + persona: { kind: "inherit" }, + text: "queued", + }); + + renderHook(() => useMessageQueue("s1", "idle", sendMessage)); + await act(async () => Promise.resolve()); + + for (let pass = 0; pass < 12; pass += 1) { + await act(async () => { + useChatStore.getState().setChatState("s1", "compacting"); + useChatStore.getState().setChatState("s1", "idle"); + await vi.advanceTimersByTimeAsync(60_000); + }); + } + + expect(sendMessage).toHaveBeenCalledTimes(5); + expect( + useChatStore.getState().queuedMessageBySession.s1?.[0]?.payload, + ).toMatchObject({ text: "queued" }); + vi.useRealTimers(); + }); + + it("backs off the readiness wait instead of re-arming every second", async () => { + vi.useFakeTimers(); + const sendMessage = vi.fn().mockResolvedValue(false); + useChatStore.getState().enqueueTransportReadyMessage("s1", { + persona: { kind: "inherit" }, + text: "queued", + }); + + const { rerender } = renderHook( + ({ ready }: { ready: boolean }) => + useMessageQueue( + "s1", + ready ? "idle" : "thinking", + sendMessage, + false, + false, + ready, + ), + { initialProps: { ready: true } }, + ); + await act(async () => Promise.resolve()); + expect(sendMessage).toHaveBeenCalledOnce(); + + // Capture re-arm delays only; installing this earlier would intercept the + // scheduling that produces the first attempt. + const delays: number[] = []; + const fakeSetTimeout = globalThis.setTimeout; + globalThis.setTimeout = (( + fn: Parameters[0], + ms?: number, + ...rest: unknown[] + ) => { + if (typeof ms === "number" && ms >= 1_000) delays.push(ms); + return ( + fakeSetTimeout as unknown as ( + ...args: unknown[] + ) => ReturnType + )(fn, ms, ...rest); + }) as unknown as typeof globalThis.setTimeout; + + try { + rerender({ ready: false }); + await act(async () => { + await vi.advanceTimersByTimeAsync(600_000); + }); + } finally { + globalThis.setTimeout = fakeSetTimeout; + } + + expect(sendMessage).toHaveBeenCalledOnce(); + expect(delays.slice(0, 5)).toEqual([2_000, 4_000, 8_000, 16_000, 30_000]); + expect(Math.max(...delays)).toBe(30_000); + // Flat one-second re-arming would be 600 wakeups over the same window. + expect(delays.length).toBeLessThan(40); + vi.useRealTimers(); + }); + it("waits for preparation readiness before a timed retry", async () => { vi.useFakeTimers(); const sendMessage = vi.fn().mockResolvedValue(false); diff --git a/src/features/chat/hooks/useMessageQueue.ts b/src/features/chat/hooks/useMessageQueue.ts index 255821351..974d54ed0 100644 --- a/src/features/chat/hooks/useMessageQueue.ts +++ b/src/features/chat/hooks/useMessageQueue.ts @@ -46,6 +46,14 @@ const queueAttemptLeaseBySession = new Map(); // follow-up store transition to re-trigger the drain. const MAX_AUTO_RETRY_DELAY_MS = 30_000; +// Waiting for readiness is unbounded; actually dispatching and being rejected +// is not. A persistent pre-commit failure is not a race: auto-compaction +// returns false on every failed compaction and appends an error notification +// per failure, and the failure itself sets the session back to idle, which +// reads as a readiness edge and re-triggers the drain. Without a ceiling that +// spins as fast as compaction can fail, growing the transcript every pass. +const MAX_CONSECUTIVE_REJECTIONS = 5; + function getQueuedMessageKey( queuedMessage: QueuedMessageRecord | null, ): string | null { @@ -106,6 +114,12 @@ export function useMessageQueue( payload: QueuedMessageRecord["payload"]; attempts: number; } | null>(null); + // Deliberately survives readiness transitions, unlike autoRetryRef: the + // transition a failed send causes must not hand it a fresh retry budget. + const consecutiveRejectionsRef = useRef<{ + payload: QueuedMessageRecord["payload"]; + count: number; + } | null>(null); const suppressNextRenderIdleCycleRef = useRef(false); const dismissedRecordIdRef = useRef(null); const queuedMessageKey = useMemo( @@ -159,6 +173,16 @@ export function useMessageQueue( return false; } + // Single choke point for the rejection ceiling, so every drain trigger + // (retry timer, readiness edge, lease release, store subscription) is + // covered rather than just the timer. + if ( + consecutiveRejectionsRef.current?.payload === payload && + consecutiveRejectionsRef.current.count >= MAX_CONSECUTIVE_REJECTIONS + ) { + return false; + } + const alreadyAttemptedThisIdleCycle = lastAttemptRef.current?.key === key && lastAttemptRef.current.payload === payload && @@ -287,6 +311,24 @@ export function useMessageQueue( if (autoRetryRef.current?.payload !== retryPayload) { autoRetryRef.current = { payload: retryPayload, attempts: 0 }; } + const rejections = + consecutiveRejectionsRef.current?.payload === retryPayload + ? consecutiveRejectionsRef.current.count + 1 + : 1; + consecutiveRejectionsRef.current = { + payload: retryPayload, + count: rejections, + }; + if (rejections >= MAX_CONSECUTIVE_REJECTIONS) { + // Stop automatically. The record stays queued and showInComposer + // was forced true above, so it is visible and the user can resend. + autoRetryRef.current = null; + if (retryTimerRef.current !== null) { + clearTimeout(retryTimerRef.current); + retryTimerRef.current = null; + } + return; + } const scheduleAutoRetry = () => { if (retryTimerRef.current !== null) return; const attempts = autoRetryRef.current?.attempts ?? 0; @@ -310,8 +352,16 @@ export function useMessageQueue( ) { // Readiness can return without a transition this hook // observes; keep the retry armed instead of abandoning the - // record. + // record. No dispatch was attempted, so this costs no rejection + // budget — but it must still back off, or an unready session + // polls at a flat one second forever. lastAttemptRef.current = null; + if (autoRetryRef.current?.payload === retryPayload) { + autoRetryRef.current = { + payload: retryPayload, + attempts: autoRetryRef.current.attempts + 1, + }; + } scheduleAutoRetry(); return; } @@ -330,6 +380,7 @@ export function useMessageQueue( } autoRetryRef.current = null; + consecutiveRejectionsRef.current = null; lastAttemptRef.current = null; useChatStore .getState() @@ -403,7 +454,11 @@ export function useMessageQueue( } if (editedCurrentRecord || advancedToNextRecord) { + // A user edit or a new head is a materially different send, so the + // rejection budget resets here. Readiness edges below deliberately do + // not reset it. autoRetryRef.current = null; + consecutiveRejectionsRef.current = null; if (retryTimerRef.current !== null) { clearTimeout(retryTimerRef.current); retryTimerRef.current = null; @@ -463,6 +518,7 @@ export function useMessageQueue( if (queuedMessageKey !== lastAttemptRef.current?.key) { lastAttemptRef.current = null; autoRetryRef.current = null; + consecutiveRejectionsRef.current = null; if (retryTimerRef.current !== null) { clearTimeout(retryTimerRef.current); retryTimerRef.current = null;