diff --git a/src/features/chat/hooks/__tests__/useMessageQueue.test.ts b/src/features/chat/hooks/__tests__/useMessageQueue.test.ts index 7936a1f28..b0fde2442 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,15 +1300,111 @@ describe("useMessageQueue", () => { expect(sendMessage).toHaveBeenCalledTimes(2); await act(async () => { - await vi.advanceTimersByTimeAsync(60_000); + await vi.advanceTimersByTimeAsync(2_000); }); - expect(sendMessage).toHaveBeenCalledTimes(2); + 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(); + }); + + 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); @@ -1428,7 +1527,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 +1540,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..974d54ed0 100644 --- a/src/features/chat/hooks/useMessageQueue.ts +++ b/src/features/chat/hooks/useMessageQueue.ts @@ -40,6 +40,20 @@ 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; + +// 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 { @@ -96,9 +110,16 @@ 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); + // 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( @@ -152,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 && @@ -277,31 +308,79 @@ export function useMessageQueue( retryPayload, ); } - if ( - retryTimerRef.current === null && - automaticallyRetriedPayloadRef.current !== retryPayload - ) { + 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; + 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. 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; + } + 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; + consecutiveRejectionsRef.current = null; lastAttemptRef.current = null; useChatStore .getState() @@ -375,7 +454,11 @@ export function useMessageQueue( } if (editedCurrentRecord || advancedToNextRecord) { - automaticallyRetriedPayloadRef.current = null; + // 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; @@ -383,7 +466,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 +517,8 @@ export function useMessageQueue( } if (queuedMessageKey !== lastAttemptRef.current?.key) { lastAttemptRef.current = null; - automaticallyRetriedPayloadRef.current = null; + autoRetryRef.current = null; + consecutiveRejectionsRef.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__) {