From 096f448c3524a350cae0f376adf26772480f8ce2 Mon Sep 17 00:00:00 2001 From: tulsi Date: Fri, 14 Aug 2026 15:46:09 -0400 Subject: [PATCH 1/3] Send queued messages in the background when the chat is not open Previously an ordinary queued message only drained via useMessageQueue, which is mounted by the open ChatView. Leaving the chat unmounted the drain, so a queued message sat in the queue until the chat was reopened, even after the session became idle and ready. Extend the global released-queue drain into a general background queue drain that also claims ordinary queued heads when no foreground chat owns the session's queue: - Add a foreground queue ownership registry; useMessageQueue registers itself per session while mounted (and not read-only), and the background drain defers to any registered owner. - Rename useReleasedQueuedMessageDrain / ReleasedQueuedMessageDrain to useBackgroundQueuedMessageDrain / BackgroundQueuedMessageDrain and let it drain ordinary (non-berdctl, non-restored) heads for unowned sessions, including reacting to ownership release when the user leaves a chat. - Extract the berdctl cross-session origin check into a shared helper so the chat drains and the berdctl runtime use one definition. Restored-from-persistence heads and berdctl cross-session sends keep their existing behavior (session activation and the dedicated berdctl drain respectively). Session dispatch leases and queued-attempt ownership assertions still guarantee at most one active send attempt per message (LAWS/CHAT.md), and the queue now resumes sending when the session becomes ready regardless of which view is open. --- src/app/SessionWindowApp.tsx | 4 +- src/app/main.berdctl.test.ts | 6 +- .../berdctl/commands/runtime/sessionSend.ts | 12 +- ... useBackgroundQueuedMessageDrain.test.tsx} | 139 ++++++++++++++++-- ....ts => useBackgroundQueuedMessageDrain.ts} | 65 ++++++-- src/features/chat/hooks/useMessageQueue.ts | 25 ++-- .../chat/lib/foregroundQueueOwnership.ts | 53 +++++++ src/features/chat/lib/queuedMessageOrigin.ts | 19 +++ .../chat/ui/BackgroundQueuedMessageDrain.tsx | 12 ++ .../chat/ui/ReleasedQueuedMessageDrain.tsx | 12 -- src/main.tsx | 4 +- 11 files changed, 279 insertions(+), 72 deletions(-) rename src/features/chat/hooks/{useReleasedQueuedMessageDrain.test.tsx => useBackgroundQueuedMessageDrain.test.tsx} (86%) rename src/features/chat/hooks/{useReleasedQueuedMessageDrain.ts => useBackgroundQueuedMessageDrain.ts} (80%) create mode 100644 src/features/chat/lib/foregroundQueueOwnership.ts create mode 100644 src/features/chat/lib/queuedMessageOrigin.ts create mode 100644 src/features/chat/ui/BackgroundQueuedMessageDrain.tsx delete mode 100644 src/features/chat/ui/ReleasedQueuedMessageDrain.tsx diff --git a/src/app/SessionWindowApp.tsx b/src/app/SessionWindowApp.tsx index c20ec6d89..07969d42a 100644 --- a/src/app/SessionWindowApp.tsx +++ b/src/app/SessionWindowApp.tsx @@ -37,7 +37,7 @@ import { } from "@/features/chat/stores/sessionWindowStore"; import { useBerdctlQueuedMessageDrain } from "@/features/berdctl/bridge/useBerdctlQueuedMessageDrain"; import { ChatView } from "@/features/chat/ui/ChatView"; -import { ReleasedQueuedMessageDrain } from "@/features/chat/ui/ReleasedQueuedMessageDrain"; +import { BackgroundQueuedMessageDrain } from "@/features/chat/ui/BackgroundQueuedMessageDrain"; import { useWorkspaceNameRequestQueue } from "@/features/chat/hooks/useWorkspaceNameRequestQueue"; import { ProjectWorkspaceStartupNameDialog } from "@/features/projects/ui/ProjectWorkspaceStartupNameDialog"; import { Button } from "@/shared/ui/button"; @@ -437,7 +437,7 @@ export function SessionWindowApp({ return ( <> - diff --git a/src/app/main.berdctl.test.ts b/src/app/main.berdctl.test.ts index 390f21bf1..b6e913875 100644 --- a/src/app/main.berdctl.test.ts +++ b/src/app/main.berdctl.test.ts @@ -101,15 +101,15 @@ describe("main entrypoint berdctl bridge loading", () => { const sessionBranch = mainSource.slice(sessionBranchStart, mainBranchStart); const mainBranch = mainSource.slice(mainBranchStart); expect(sessionBranch).not.toContain(""); - expect(mainBranch).toContain(""); + expect(mainBranch).toContain(""); expect(mainBranch).toContain(""); }); it("mounts the released queued-message drain unconditionally", () => { expect(mainSource).toContain( - 'import { ReleasedQueuedMessageDrain } from "@/features/chat/ui/ReleasedQueuedMessageDrain"', + 'import { BackgroundQueuedMessageDrain } from "@/features/chat/ui/BackgroundQueuedMessageDrain"', ); - expect(mainSource).not.toContain("OptionalReleasedQueuedMessageDrain"); + expect(mainSource).not.toContain("OptionalBackgroundQueuedMessageDrain"); }); it("reports a failed dynamic bridge import without boot-failing the app", () => { diff --git a/src/features/berdctl/commands/runtime/sessionSend.ts b/src/features/berdctl/commands/runtime/sessionSend.ts index c45aeaac5..1b0fa803a 100644 --- a/src/features/berdctl/commands/runtime/sessionSend.ts +++ b/src/features/berdctl/commands/runtime/sessionSend.ts @@ -14,9 +14,9 @@ export { SessionDispatchUnresolvedError, } from "@/features/chat/lib/queuedSessionSend"; import { formatIncludedWorkspacesPrompt } from "@/features/chat/lib/workspaceAttachments"; -import type { QueuedMessageRecord } from "@/features/chat/stores/chatStore"; import type { MessageMetadata } from "@/shared/types/messages"; import type { ChatSendOptions } from "@/features/chat/types"; +export { isBerdctlCrossSessionQueuedMessage } from "@/features/chat/lib/queuedMessageOrigin"; export const BERDCTL_CROSS_SESSION_ORIGIN = "berdctl_cross_session" satisfies NonNullable; @@ -32,16 +32,6 @@ export function berdctlCrossSessionSendOptions(): ChatSendOptions { }; } -export function isBerdctlCrossSessionQueuedMessage( - message: QueuedMessageRecord | undefined, -): boolean { - return ( - message?.kind === "transport-ready" && - message.payload.sendOptions?.userMessageMetadata?.origin === - BERDCTL_CROSS_SESSION_ORIGIN - ); -} - export async function sendPromptToExistingSessionInBackground( sessionId: string, prompt: string, diff --git a/src/features/chat/hooks/useReleasedQueuedMessageDrain.test.tsx b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx similarity index 86% rename from src/features/chat/hooks/useReleasedQueuedMessageDrain.test.tsx rename to src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx index ea6d11060..c5914ac7a 100644 --- a/src/features/chat/hooks/useReleasedQueuedMessageDrain.test.tsx +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx @@ -5,11 +5,15 @@ import { SessionDispatchContentionError } from "@/features/chat/lib/sessionDispa import type { SessionDispatchReleaseWaiter } from "@/features/chat/lib/sessionTargetCoordinator"; import { QueuedMessageOwnershipLostError } from "@/features/chat/lib/preCommitSendRejection"; import { QueuedSessionNotReadyError } from "@/features/chat/lib/queuedMessageReadiness"; +import { + registerForegroundQueueOwner, + resetForegroundQueueOwnershipForTesting, +} from "@/features/chat/lib/foregroundQueueOwnership"; import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; import { useChatStore } from "@/features/chat/stores/chatStore"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import type { QueuedMessageRecord } from "@/features/chat/stores/chatStore"; -import { useReleasedQueuedMessageDrain } from "./useReleasedQueuedMessageDrain"; +import { useBackgroundQueuedMessageDrain } from "./useBackgroundQueuedMessageDrain"; const mocks = vi.hoisted(() => ({ sendQueuedPromptToExistingSessionInBackground: vi.fn(), @@ -27,7 +31,7 @@ function DrainHarness({ sessionId?: string; ownerReady?: boolean; } = {}) { - useReleasedQueuedMessageDrain(sessionId, ownerReady); + useBackgroundQueuedMessageDrain(sessionId, ownerReady); return null; } @@ -70,9 +74,21 @@ function releasedRecord(): QueuedMessageRecord & { kind: "transport-ready" } { }; } -describe("useReleasedQueuedMessageDrain", () => { +function ordinaryRecord(): QueuedMessageRecord & { kind: "transport-ready" } { + return { + kind: "transport-ready", + recordId: "ordinary-record", + payload: { + text: "ordinary prompt", + persona: { kind: "inherit" }, + }, + }; +} + +describe("useBackgroundQueuedMessageDrain", () => { beforeEach(() => { vi.clearAllMocks(); + resetForegroundQueueOwnershipForTesting(); mocks.sendQueuedPromptToExistingSessionInBackground.mockResolvedValue( undefined, ); @@ -757,20 +773,10 @@ describe("useReleasedQueuedMessageDrain", () => { }); }); - it("ignores ordinary and Berdctl-origin transport-ready records", () => { + it("ignores Berdctl-origin transport-ready records", () => { useChatStore.setState({ queuedMessageBySession: { - ordinary: [ - { - kind: "transport-ready", - recordId: "ordinary-record", - payload: { - persona: { kind: "inherit" }, - text: "ordinary", - }, - }, - ], - berdctl: [ + "session-1": [ { kind: "transport-ready", recordId: "berdctl-record", @@ -792,4 +798,107 @@ describe("useReleasedQueuedMessageDrain", () => { mocks.sendQueuedPromptToExistingSessionInBackground, ).not.toHaveBeenCalled(); }); + + it("drains an ordinary queued head when no foreground chat owns the session", async () => { + useChatStore.setState({ + queuedMessageBySession: { "session-1": [ordinaryRecord()] }, + }); + + render(); + + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + }); + + it("defers an ordinary queued head to a mounted foreground owner", () => { + const releaseOwner = registerForegroundQueueOwner("session-1"); + useChatStore.setState({ + queuedMessageBySession: { "session-1": [ordinaryRecord()] }, + }); + + render(); + + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + releaseOwner(); + }); + + it("does not drain an ordinary head restored from persistence", () => { + useChatStore.setState({ + queuedMessageBySession: { + "session-1": [{ ...ordinaryRecord(), restored: true }], + }, + }); + + render(); + + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + }); + + it("drains an ordinary queued head after the foreground owner unmounts", async () => { + const releaseOwner = registerForegroundQueueOwner("session-1"); + useChatStore.setState({ + queuedMessageBySession: { "session-1": [ordinaryRecord()] }, + }); + + render(); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + + act(() => releaseOwner()); + + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + }); + + it("keeps deferring while any foreground owner remains registered", async () => { + const releaseFirst = registerForegroundQueueOwner("session-1"); + const releaseSecond = registerForegroundQueueOwner("session-1"); + useChatStore.setState({ + queuedMessageBySession: { "session-1": [ordinaryRecord()] }, + }); + + render(); + act(() => releaseFirst()); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + + act(() => releaseSecond()); + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + }); + + it("waits for an unowned ordinary head's active run to settle before draining", async () => { + useChatStore.setState({ + queuedMessageBySession: { "session-1": [ordinaryRecord()] }, + }); + useChatStore.getState().setActiveRunId("session-1", "run-1"); + + render(); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + + act(() => useChatStore.getState().setActiveRunId("session-1", null)); + + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + }); }); diff --git a/src/features/chat/hooks/useReleasedQueuedMessageDrain.ts b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts similarity index 80% rename from src/features/chat/hooks/useReleasedQueuedMessageDrain.ts rename to src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts index c6cb0219e..f0387edcb 100644 --- a/src/features/chat/hooks/useReleasedQueuedMessageDrain.ts +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts @@ -11,6 +11,11 @@ import { becameQueuedMessageTargetAttemptable, isQueuedMessageTargetAttemptable, } from "@/features/chat/lib/queuedMessageAttemptOwnership"; +import { + hasForegroundQueueOwner, + subscribeForegroundQueueOwnership, +} from "@/features/chat/lib/foregroundQueueOwnership"; +import { isBerdctlCrossSessionQueuedMessage } from "@/features/chat/lib/queuedMessageOrigin"; import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; import { type QueuedMessageRecord, @@ -37,6 +42,30 @@ function ownerIdFor(scopedSessionId?: string): string { return scopedSessionId ? `session:${scopedSessionId}` : "global"; } +/** + * The background drain claims a queued head when either + * - it was released from a deferred workspace send (the foreground queue never + * claims released heads), or + * - no mounted foreground chat currently owns the session's queue, so nothing + * else will send it (the user queued a message and left the chat). + * + * berdctl cross-session sends are excluded; they have a dedicated drain. + * Ordinary heads restored from persistence are excluded so an app relaunch + * does not fire stale prompts without the user reopening the chat. + */ +function isBackgroundDrainableHead( + record: QueuedMessageRecord & { kind: "transport-ready" }, + sessionId: string, +): boolean { + if (isBerdctlCrossSessionQueuedMessage(record)) { + return false; + } + if (record.releasedFromDeferred) { + return true; + } + return !record.restored && !hasForegroundQueueOwner(sessionId); +} + function reconcileContentionWaiters(scopedSessionId?: string): void { const sessionStore = useChatSessionStore.getState(); const queue = useChatStore.getState().queuedMessageBySession; @@ -72,10 +101,10 @@ function scheduleContentionResume( } waiter.resumeScheduled = true; contentionWaiters.delete(sessionId); - queueMicrotask(() => drainReleasedQueuedMessage(sessionId, waiter.ownerId)); + queueMicrotask(() => drainQueuedMessage(sessionId, waiter.ownerId)); } -function drainReleasedQueuedMessage(sessionId: string, ownerId: string): void { +function drainQueuedMessage(sessionId: string, ownerId: string): void { if (!activeOwners.has(ownerId)) return; const sessionExists = Boolean( useChatSessionStore.getState().getSession(sessionId), @@ -106,7 +135,7 @@ function drainReleasedQueuedMessage(sessionId: string, ownerId: string): void { queuedMessage, sessionStore.getSession(sessionId), ) || - !queuedMessage.releasedFromDeferred || + !isBackgroundDrainableHead(queuedMessage, sessionId) || !isQueuedSessionReady(runtime) ) { return; @@ -163,7 +192,7 @@ function drainReleasedQueuedMessage(sessionId: string, ownerId: string): void { if (error instanceof PreCommitSendRejectedError) return; const message = i18n.t("chat:queue.releasedSendFailed"); console.error( - `[released-queue] failed to send queued prompt for session ${sessionId}`, + `[background-queue] failed to send queued prompt for session ${sessionId}`, error, ); const current = @@ -186,10 +215,10 @@ function drainReleasedQueuedMessage(sessionId: string, ownerId: string): void { waiter.attemptSettled = true; scheduleContentionResume(sessionId, waiter); } else { - queueMicrotask(() => drainReleasedQueuedMessage(sessionId, ownerId)); + queueMicrotask(() => drainQueuedMessage(sessionId, ownerId)); } } else { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } }); } @@ -206,7 +235,7 @@ function getOwnedSessionIds( ); } -function drainReadyReleasedMessages(scopedSessionId?: string): void { +function drainReadyQueuedMessages(scopedSessionId?: string): void { const ownerId = ownerIdFor(scopedSessionId); if (!activeOwners.has(ownerId)) return; reconcileContentionWaiters(scopedSessionId); @@ -215,11 +244,11 @@ function drainReadyReleasedMessages(scopedSessionId?: string): void { queuedMessageBySession, scopedSessionId, )) { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } } -export function useReleasedQueuedMessageDrain( +export function useBackgroundQueuedMessageDrain( scopedSessionId?: string, ownerReady = true, ): void { @@ -227,7 +256,7 @@ export function useReleasedQueuedMessageDrain( if (!ownerReady) return; const ownerId = ownerIdFor(scopedSessionId); activeOwners.add(ownerId); - drainReadyReleasedMessages(scopedSessionId); + drainReadyQueuedMessages(scopedSessionId); const unsubscribeWindowStore = scopedSessionId ? undefined : useSessionWindowStore.subscribe((state, previousState) => { @@ -236,9 +265,14 @@ export function useReleasedQueuedMessageDrain( (!previousState.hasLoadedSnapshot || state.openSessions !== previousState.openSessions) ) { - drainReadyReleasedMessages(); + drainReadyQueuedMessages(); } }); + // When a foreground chat releases queue ownership (the user left the + // chat), any ready queued head it never sent becomes ours to send. + const unsubscribeForegroundOwnership = subscribeForegroundQueueOwnership( + () => drainReadyQueuedMessages(scopedSessionId), + ); const unsubscribeSessionStore = useChatSessionStore.subscribe( (state, previousState) => { reconcileContentionWaiters(scopedSessionId); @@ -258,11 +292,11 @@ export function useReleasedQueuedMessageDrain( ), ) ) { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } } if (state.hasHydratedSessions && !previousState.hasHydratedSessions) { - drainReadyReleasedMessages(scopedSessionId); + drainReadyQueuedMessages(scopedSessionId); } }, ); @@ -276,7 +310,7 @@ export function useReleasedQueuedMessageDrain( const queuedMessage = state.queuedMessageBySession[sessionId]?.[0]; if ( queuedMessage?.kind !== "transport-ready" || - !queuedMessage.releasedFromDeferred + !isBackgroundDrainableHead(queuedMessage, sessionId) ) { continue; } @@ -294,13 +328,14 @@ export function useReleasedQueuedMessageDrain( !currentBlocked && (previousBlocked || becameTransportReady) ) { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } } }, ); return () => { unsubscribeWindowStore?.(); + unsubscribeForegroundOwnership(); unsubscribeSessionStore(); unsubscribeChatStore(); activeOwners.delete(ownerId); diff --git a/src/features/chat/hooks/useMessageQueue.ts b/src/features/chat/hooks/useMessageQueue.ts index 6b5b2f389..cffcdf307 100644 --- a/src/features/chat/hooks/useMessageQueue.ts +++ b/src/features/chat/hooks/useMessageQueue.ts @@ -17,6 +17,8 @@ import { acquireSessionDispatchTarget, type SessionDispatchTargetLease, } from "../lib/sessionTargetCoordinator"; +import { registerForegroundQueueOwner } from "../lib/foregroundQueueOwnership"; +import { isBerdctlCrossSessionQueuedMessage } from "../lib/queuedMessageOrigin"; import type { QueuedMessageRecord } from "../stores/chatStore"; import type { ChatSendOptions } from "../types"; import type { SessionExecutionTarget } from "../lib/sessionExecutionTarget"; @@ -46,23 +48,15 @@ function getQueuedMessageKey( : null; } -function isBerdctlCrossSessionQueuedMessage( - queuedMessage: QueuedMessageRecord | null, -): boolean { - return ( - queuedMessage?.kind === "transport-ready" && - queuedMessage.payload.sendOptions?.userMessageMetadata?.origin === - "berdctl_cross_session" - ); -} - /** * Single-slot message queue that holds one pending message while the agent is * busy and auto-sends it when the chat transitions back to idle. * * State lives in the Zustand store (keyed by session) so it survives tab - * switches — users can queue a follow-up, navigate away, and come back to - * find it sent. + * switches. While mounted, this hook registers as the session's foreground + * queue owner; when the user navigates away and it unmounts, the background + * drain (`useBackgroundQueuedMessageDrain`) takes over so queued messages + * still send without the chat being open. * * A direct Zustand store subscription ensures the drain fires even when the * webview is backgrounded and React defers re-renders (e.g. rAF paused, @@ -112,6 +106,13 @@ export function useMessageQueue( [queuedRecord], ); + // Claim foreground ownership of this session's queue so the background + // drain defers to this hook while the chat is open and interactive. + useEffect(() => { + if (readOnly) return; + return registerForegroundQueueOwner(sessionId); + }, [readOnly, sessionId]); + // --- Background-safe store subscription --- // When the webview is hidden/minimized, React may not schedule re-renders // for Zustand selector updates because requestAnimationFrame is paused. diff --git a/src/features/chat/lib/foregroundQueueOwnership.ts b/src/features/chat/lib/foregroundQueueOwnership.ts new file mode 100644 index 000000000..c90d3437f --- /dev/null +++ b/src/features/chat/lib/foregroundQueueOwnership.ts @@ -0,0 +1,53 @@ +/** + * Registry of sessions whose queue is currently owned by a mounted foreground + * drain (`useMessageQueue` inside an interactive ChatView). The background + * queue drain defers to a registered foreground owner so exactly one drain + * claims each queued head by design; the session dispatch lease and queued + * message ownership assertions remain the safety net for mount races. + */ + +const ownerCountsBySession = new Map(); +const listeners = new Set<() => void>(); + +function notifyOwnershipChanged(): void { + for (const listener of [...listeners]) { + listener(); + } +} + +export function registerForegroundQueueOwner(sessionId: string): () => void { + ownerCountsBySession.set( + sessionId, + (ownerCountsBySession.get(sessionId) ?? 0) + 1, + ); + notifyOwnershipChanged(); + let released = false; + return () => { + if (released) return; + released = true; + const count = ownerCountsBySession.get(sessionId) ?? 0; + if (count <= 1) { + ownerCountsBySession.delete(sessionId); + } else { + ownerCountsBySession.set(sessionId, count - 1); + } + notifyOwnershipChanged(); + }; +} + +export function hasForegroundQueueOwner(sessionId: string): boolean { + return (ownerCountsBySession.get(sessionId) ?? 0) > 0; +} + +export function subscribeForegroundQueueOwnership( + listener: () => void, +): () => void { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; +} + +export function resetForegroundQueueOwnershipForTesting(): void { + ownerCountsBySession.clear(); +} diff --git a/src/features/chat/lib/queuedMessageOrigin.ts b/src/features/chat/lib/queuedMessageOrigin.ts new file mode 100644 index 000000000..14bd73a6e --- /dev/null +++ b/src/features/chat/lib/queuedMessageOrigin.ts @@ -0,0 +1,19 @@ +import type { QueuedMessageRecord } from "../stores/chatStore"; +import type { MessageMetadata } from "@/shared/types/messages"; + +const BERDCTL_CROSS_SESSION_ORIGIN = + "berdctl_cross_session" satisfies NonNullable; + +/** + * berdctl cross-session sends have their own dedicated drain + * (`useBerdctlQueuedMessageDrain`); the chat queue drains must not claim them. + */ +export function isBerdctlCrossSessionQueuedMessage( + record: QueuedMessageRecord | null | undefined, +): boolean { + return ( + record?.kind === "transport-ready" && + record.payload.sendOptions?.userMessageMetadata?.origin === + BERDCTL_CROSS_SESSION_ORIGIN + ); +} diff --git a/src/features/chat/ui/BackgroundQueuedMessageDrain.tsx b/src/features/chat/ui/BackgroundQueuedMessageDrain.tsx new file mode 100644 index 000000000..981f6fe3a --- /dev/null +++ b/src/features/chat/ui/BackgroundQueuedMessageDrain.tsx @@ -0,0 +1,12 @@ +import { useBackgroundQueuedMessageDrain } from "@/features/chat/hooks/useBackgroundQueuedMessageDrain"; + +export function BackgroundQueuedMessageDrain({ + sessionId, + ownerReady = true, +}: { + sessionId?: string; + ownerReady?: boolean; +} = {}) { + useBackgroundQueuedMessageDrain(sessionId, ownerReady); + return null; +} diff --git a/src/features/chat/ui/ReleasedQueuedMessageDrain.tsx b/src/features/chat/ui/ReleasedQueuedMessageDrain.tsx deleted file mode 100644 index 870533c3d..000000000 --- a/src/features/chat/ui/ReleasedQueuedMessageDrain.tsx +++ /dev/null @@ -1,12 +0,0 @@ -import { useReleasedQueuedMessageDrain } from "@/features/chat/hooks/useReleasedQueuedMessageDrain"; - -export function ReleasedQueuedMessageDrain({ - sessionId, - ownerReady = true, -}: { - sessionId?: string; - ownerReady?: boolean; -} = {}) { - useReleasedQueuedMessageDrain(sessionId, ownerReady); - return null; -} diff --git a/src/main.tsx b/src/main.tsx index ea0a2b480..61e0f5db3 100644 --- a/src/main.tsx +++ b/src/main.tsx @@ -15,7 +15,7 @@ import { App } from "@/app/App"; import { GitStateEvents } from "@/app/GitStateEvents"; import { LocalMediaCacheEvents } from "@/app/LocalMediaCacheEvents"; import { RendererTelemetry } from "@/app/RendererTelemetry"; -import { ReleasedQueuedMessageDrain } from "@/features/chat/ui/ReleasedQueuedMessageDrain"; +import { BackgroundQueuedMessageDrain } from "@/features/chat/ui/BackgroundQueuedMessageDrain"; import { UpdaterProvider } from "@/features/updates/hooks/useUpdater"; import { I18nProvider } from "@/shared/i18n"; import { initTelemetry, trackAppLaunched } from "@/shared/telemetry/client"; @@ -172,7 +172,7 @@ if (bootError) { - + From c82e1ff4a989ffc5749ae4f5d3161e6acf365bef Mon Sep 17 00:00:00 2001 From: tulsi Date: Fri, 14 Aug 2026 16:28:58 -0400 Subject: [PATCH 2/3] Harden background queue drain from review findings 1. Refresh reclaimed session queues from persistence before draining. When a session window closes, this renderer's in-memory queue copy is stale; draining from it could re-send a head the window already sent or dismissed (a duplicate user turn under LAWS/CHAT.md). Mirror the berdctl bridge's refreshReclaimedQueues so chat correctness does not depend on the optionally-loaded bridge: reload reclaimed queues from persistence (records arrive restored: true, which the drain refuses) before draining. 2. Track restored-head exclusions per record id. The restored flag is cleared by markQueuedMessagesReady during any session load, including loads the user did not initiate, which would let a relaunch-stale prompt fire. The drain now remembers restored heads it has seen and only lifts the exclusion when a foreground owner registers for the session - i.e., the user actually reopened the chat. 3. Surface background queued-send failures with an error toast. The background drain only claims unowned sessions, so a parked failure was invisible until the user reopened the chat. 4. Naming cleanup: rename the queue.releasedSendFailed i18n key to queue.backgroundSendFailed (drain-only usage) and update the stale 'released' test name in main.berdctl.test.ts. All fixes verified with discriminating tests (each fails against the pre-fix implementation). --- src/app/main.berdctl.test.ts | 2 +- .../useBackgroundQueuedMessageDrain.test.tsx | 183 +++++++++++++++++- .../hooks/useBackgroundQueuedMessageDrain.ts | 119 +++++++++++- src/shared/i18n/locales/en/chat.json | 3 +- 4 files changed, 299 insertions(+), 8 deletions(-) diff --git a/src/app/main.berdctl.test.ts b/src/app/main.berdctl.test.ts index b6e913875..e0290ccf1 100644 --- a/src/app/main.berdctl.test.ts +++ b/src/app/main.berdctl.test.ts @@ -105,7 +105,7 @@ describe("main entrypoint berdctl bridge loading", () => { expect(mainBranch).toContain(""); }); - it("mounts the released queued-message drain unconditionally", () => { + it("mounts the background queued-message drain unconditionally", () => { expect(mainSource).toContain( 'import { BackgroundQueuedMessageDrain } from "@/features/chat/ui/BackgroundQueuedMessageDrain"', ); diff --git a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx index c5914ac7a..b6442a792 100644 --- a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx @@ -11,12 +11,23 @@ import { } from "@/features/chat/lib/foregroundQueueOwnership"; import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; import { useChatStore } from "@/features/chat/stores/chatStore"; +import * as queuePersistence from "@/features/chat/stores/queuePersistence"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import type { QueuedMessageRecord } from "@/features/chat/stores/chatStore"; -import { useBackgroundQueuedMessageDrain } from "./useBackgroundQueuedMessageDrain"; +import { + resetBackgroundQueueDrainStateForTesting, + useBackgroundQueuedMessageDrain, +} from "./useBackgroundQueuedMessageDrain"; const mocks = vi.hoisted(() => ({ sendQueuedPromptToExistingSessionInBackground: vi.fn(), + toastError: vi.fn(), +})); + +vi.mock("sonner", () => ({ + toast: Object.assign(vi.fn(), { + error: (...args: unknown[]) => mocks.toastError(...args), + }), })); vi.mock("@/features/chat/lib/queuedSessionSend", () => ({ @@ -89,6 +100,10 @@ describe("useBackgroundQueuedMessageDrain", () => { beforeEach(() => { vi.clearAllMocks(); resetForegroundQueueOwnershipForTesting(); + resetBackgroundQueueDrainStateForTesting(); + vi.spyOn(queuePersistence, "loadPersistedMessageQueues").mockResolvedValue( + {}, + ); mocks.sendQueuedPromptToExistingSessionInBackground.mockResolvedValue( undefined, ); @@ -704,6 +719,32 @@ describe("useBackgroundQueuedMessageDrain", () => { }, }); }); + expect(mocks.toastError).toHaveBeenCalledWith( + "Session", + expect.objectContaining({ description: expect.any(String) }), + ); + }); + + it("does not toast when a send loses ownership pre-commit", async () => { + const released = releasedRecord(); + mocks.sendQueuedPromptToExistingSessionInBackground.mockRejectedValueOnce( + new QueuedMessageOwnershipLostError(), + ); + useChatStore.setState({ + queuedMessageBySession: { "session-1": [released] }, + }); + + render(); + + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + await act(async () => { + await Promise.resolve(); + }); + expect(mocks.toastError).not.toHaveBeenCalled(); }); it("scopes released draining to the renderer that owns each session", async () => { @@ -753,6 +794,97 @@ describe("useBackgroundQueuedMessageDrain", () => { }); }); + it("refreshes a reclaimed session's queue from persistence before draining", async () => { + // The session window already sent/dismissed the head; this renderer still + // holds a stale in-memory copy. Persistence is the source of truth. + const stale = { ...ordinaryRecord(), recordId: "stale-record" }; + useChatStore.setState({ + queuedMessageBySession: { "session-1": [stale] }, + }); + useSessionWindowStore.setState({ + openSessions: { "session-1": "window-1" }, + }); + vi.spyOn(queuePersistence, "loadPersistedMessageQueues").mockResolvedValue( + {}, + ); + + render(); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + + act(() => useSessionWindowStore.setState({ openSessions: {} })); + + await waitFor(() => + expect(queuePersistence.loadPersistedMessageQueues).toHaveBeenCalled(), + ); + await act(async () => { + await Promise.resolve(); + }); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + expect( + useChatStore.getState().queuedMessageBySession["session-1"], + ).toBeUndefined(); + }); + + it("does not fire a reclaimed head that persistence marks restored", async () => { + const stale = ordinaryRecord(); + useChatStore.setState({ + queuedMessageBySession: { "session-1": [stale] }, + }); + useSessionWindowStore.setState({ + openSessions: { "session-1": "window-1" }, + }); + vi.spyOn(queuePersistence, "loadPersistedMessageQueues").mockResolvedValue({ + "session-1": [{ ...ordinaryRecord(), restored: true }], + }); + + render(); + act(() => useSessionWindowStore.setState({ openSessions: {} })); + + await waitFor(() => + expect(queuePersistence.loadPersistedMessageQueues).toHaveBeenCalled(), + ); + await act(async () => { + await Promise.resolve(); + }); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + expect( + useChatStore.getState().queuedMessageBySession["session-1"]?.[0] + ?.restored, + ).toBe(true); + }); + + it("drains immediately when window changes reclaim no sessions", async () => { + useChatStore.setState({ + queuedMessageBySession: { "session-1": [ordinaryRecord()] }, + }); + useSessionWindowStore.setState({ + openSessions: { "other-session": "window-1" }, + }); + + render(); + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + + act(() => + useSessionWindowStore.setState({ + openSessions: { + "other-session": "window-1", + "main-session": "window-2", + }, + }), + ); + expect(queuePersistence.loadPersistedMessageQueues).not.toHaveBeenCalled(); + }); + it("waits for the authoritative window snapshot before global draining", async () => { const released = releasedRecord(); useChatStore.setState({ @@ -841,6 +973,55 @@ describe("useBackgroundQueuedMessageDrain", () => { ).not.toHaveBeenCalled(); }); + it("keeps excluding a restored head after an incidental load clears the flag", async () => { + const restored = { ...ordinaryRecord(), restored: true }; + useChatStore.setState({ + queuedMessageBySession: { "session-1": [restored] }, + }); + + render(); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + + // Simulate markQueuedMessagesReady from a session load the user did not + // initiate: the flag clears and a new head object appears. + act(() => useChatStore.getState().markQueuedMessagesReady("session-1")); + + await act(async () => { + await Promise.resolve(); + }); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + }); + + it("lifts a restored exclusion when the user opens the chat", async () => { + const restored = { ...ordinaryRecord(), restored: true }; + useChatStore.setState({ + queuedMessageBySession: { "session-1": [restored] }, + }); + + render(); + act(() => useChatStore.getState().markQueuedMessagesReady("session-1")); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + + // The user opens the chat (foreground owner registers), then leaves. + let releaseOwner: (() => void) | undefined; + act(() => { + releaseOwner = registerForegroundQueueOwner("session-1"); + }); + act(() => releaseOwner?.()); + + await waitFor(() => + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(), + ); + }); + it("drains an ordinary queued head after the foreground owner unmounts", async () => { const releaseOwner = registerForegroundQueueOwner("session-1"); useChatStore.setState({ diff --git a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts index f0387edcb..aaaa59a47 100644 --- a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts @@ -1,4 +1,5 @@ import { useEffect } from "react"; +import { toast } from "sonner"; import { i18n } from "@/shared/i18n"; import { @@ -21,12 +22,14 @@ import { type QueuedMessageRecord, useChatStore, } from "@/features/chat/stores/chatStore"; +import { loadPersistedMessageQueues } from "@/features/chat/stores/queuePersistence"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import { SessionDispatchContentionError } from "@/features/chat/lib/sessionDispatchAcquisition"; import { sendQueuedPromptToExistingSessionInBackground } from "@/features/chat/lib/queuedSessionSend"; const drainingSessionIds = new Set(); const activeOwners = new Set(); +let ownershipRefreshSequence = 0; type ContentionWaiter = { ownerId: string; record: QueuedMessageRecord; @@ -38,10 +41,58 @@ type ContentionWaiter = { const contentionWaiters = new Map(); +// Record ids observed with `restored: true` this app run, per session. The +// flag itself is cleared by `markQueuedMessagesReady` during any session +// load — including loads the user did not initiate (e.g. a berdctl +// cross-session send touching the session) — so the flag alone cannot +// guarantee "the user reopened this chat". A session's exclusions lift only +// when a foreground owner registers for it, which is the user actually +// opening the chat. +const restoredExclusionsBySession = new Map>(); + +function noteRestoredRecords( + queuedMessageBySession: Record, +): void { + for (const [sessionId, records] of Object.entries(queuedMessageBySession)) { + if (hasForegroundQueueOwner(sessionId)) continue; + for (const record of records) { + if (!record.restored) continue; + let exclusions = restoredExclusionsBySession.get(sessionId); + if (!exclusions) { + exclusions = new Set(); + restoredExclusionsBySession.set(sessionId, exclusions); + } + exclusions.add(record.recordId); + } + } +} + +function liftRestoredExclusionsForOwnedSessions(): void { + for (const sessionId of restoredExclusionsBySession.keys()) { + if (hasForegroundQueueOwner(sessionId)) { + restoredExclusionsBySession.delete(sessionId); + } + } +} + +function isRestoredExcluded( + sessionId: string, + record: QueuedMessageRecord & { kind: "transport-ready" }, +): boolean { + return ( + record.restored === true || + (restoredExclusionsBySession.get(sessionId)?.has(record.recordId) ?? false) + ); +} + function ownerIdFor(scopedSessionId?: string): string { return scopedSessionId ? `session:${scopedSessionId}` : "global"; } +export function resetBackgroundQueueDrainStateForTesting(): void { + restoredExclusionsBySession.clear(); +} + /** * The background drain claims a queued head when either * - it was released from a deferred workspace send (the foreground queue never @@ -51,7 +102,9 @@ function ownerIdFor(scopedSessionId?: string): string { * * berdctl cross-session sends are excluded; they have a dedicated drain. * Ordinary heads restored from persistence are excluded so an app relaunch - * does not fire stale prompts without the user reopening the chat. + * does not fire stale prompts without the user reopening the chat; the + * exclusion is tracked per record id so an incidental session load clearing + * the `restored` flag cannot lift it. */ function isBackgroundDrainableHead( record: QueuedMessageRecord & { kind: "transport-ready" }, @@ -63,7 +116,38 @@ function isBackgroundDrainableHead( if (record.releasedFromDeferred) { return true; } - return !record.restored && !hasForegroundQueueOwner(sessionId); + return ( + !isRestoredExcluded(sessionId, record) && + !hasForegroundQueueOwner(sessionId) + ); +} + +/** + * A session window keeps the authoritative queue while its session is open; + * this renderer's in-memory copy goes stale. When windows close, reload the + * reclaimed sessions' queues from persistence before draining so a stale + * head that the window already sent, edited, or dismissed cannot fire again. + * Reconciled records arrive `restored: true`, which the background drain + * refuses to claim. + */ +async function refreshReclaimedQueues( + previousOpenSessions: Record, + openSessions: Record, +): Promise { + const reclaimedSessionIds = Object.keys(previousOpenSessions).filter( + (sessionId) => !(sessionId in openSessions), + ); + if (reclaimedSessionIds.length === 0) { + drainReadyQueuedMessages(); + return; + } + const sequence = ++ownershipRefreshSequence; + const persistedQueues = await loadPersistedMessageQueues(); + if (sequence !== ownershipRefreshSequence) return; + useChatStore + .getState() + .reconcileQueuedMessages(persistedQueues, reclaimedSessionIds); + drainReadyQueuedMessages(); } function reconcileContentionWaiters(scopedSessionId?: string): void { @@ -190,7 +274,7 @@ function drainQueuedMessage(sessionId: string, ownerId: string): void { return; } if (error instanceof PreCommitSendRejectedError) return; - const message = i18n.t("chat:queue.releasedSendFailed"); + const message = i18n.t("chat:queue.backgroundSendFailed"); console.error( `[background-queue] failed to send queued prompt for session ${sessionId}`, error, @@ -205,6 +289,17 @@ function drainQueuedMessage(sessionId: string, ownerId: string): void { status: "failed", error: message, }); + // The user is not viewing this chat (the background drain only + // claims unowned sessions), so a parked failure would otherwise be + // invisible until they reopen it. Surface it where they are now. + const sessionTitle = useChatSessionStore + .getState() + .getSession(sessionId) + ?.title?.trim(); + toast.error( + sessionTitle || i18n.t("chat:queue.backgroundSendFailedTitle"), + { description: message }, + ); } }) .finally(() => { @@ -240,6 +335,7 @@ function drainReadyQueuedMessages(scopedSessionId?: string): void { if (!activeOwners.has(ownerId)) return; reconcileContentionWaiters(scopedSessionId); const { queuedMessageBySession } = useChatStore.getState(); + noteRestoredRecords(queuedMessageBySession); for (const sessionId of getOwnedSessionIds( queuedMessageBySession, scopedSessionId, @@ -265,13 +361,25 @@ export function useBackgroundQueuedMessageDrain( (!previousState.hasLoadedSnapshot || state.openSessions !== previousState.openSessions) ) { - drainReadyQueuedMessages(); + if (!previousState.hasLoadedSnapshot) { + drainReadyQueuedMessages(); + } else { + void refreshReclaimedQueues( + previousState.openSessions, + state.openSessions, + ); + } } }); // When a foreground chat releases queue ownership (the user left the // chat), any ready queued head it never sent becomes ours to send. + // Registration also lifts restored-head exclusions: the user reopening + // the chat is the signal that a persisted head is theirs to send again. const unsubscribeForegroundOwnership = subscribeForegroundQueueOwnership( - () => drainReadyQueuedMessages(scopedSessionId), + () => { + liftRestoredExclusionsForOwnedSessions(); + drainReadyQueuedMessages(scopedSessionId); + }, ); const unsubscribeSessionStore = useChatSessionStore.subscribe( (state, previousState) => { @@ -303,6 +411,7 @@ export function useBackgroundQueuedMessageDrain( const unsubscribeChatStore = useChatStore.subscribe( (state, previousState) => { reconcileContentionWaiters(scopedSessionId); + noteRestoredRecords(state.queuedMessageBySession); for (const sessionId of getOwnedSessionIds( state.queuedMessageBySession, scopedSessionId, diff --git a/src/shared/i18n/locales/en/chat.json b/src/shared/i18n/locales/en/chat.json index 9d3041648..a6d778329 100644 --- a/src/shared/i18n/locales/en/chat.json +++ b/src/shared/i18n/locales/en/chat.json @@ -411,7 +411,8 @@ "dismiss": "Dismiss queued message", "edit": "Edit queued message", "sendAnyway": "Send anyway", - "releasedSendFailed": "The queued message could not be sent. Retry it or edit the message.", + "backgroundSendFailed": "The queued message could not be sent. Retry it or edit the message.", + "backgroundSendFailedTitle": "Queued message failed", "configureWorktree": "Configure your new worktree?", "configureWorkspaces": "Configure new project workspaces?", "configureWorkspacePlan": "Configure {{worktreeLabel}}?", From ce14a6e9bb8d99371f7fc8e0fba34839ba465961 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Fri, 14 Aug 2026 19:02:10 -0500 Subject: [PATCH 3/3] Fix overlapping queue reclaim reconciliation --- .../useBerdctlQueuedMessageDrain.test.tsx | 2 + .../bridge/useBerdctlQueuedMessageDrain.ts | 43 ++++----- .../useBackgroundQueuedMessageDrain.test.tsx | 69 ++++++++++++++ .../hooks/useBackgroundQueuedMessageDrain.ts | 43 ++++----- .../chat/lib/reclaimedQueueReconciliation.ts | 92 +++++++++++++++++++ .../queueFailureLocaleParity.test.ts | 14 +++ src/shared/i18n/locales/es/chat.json | 2 + 7 files changed, 211 insertions(+), 54 deletions(-) create mode 100644 src/features/chat/lib/reclaimedQueueReconciliation.ts create mode 100644 src/shared/i18n/__tests__/queueFailureLocaleParity.test.ts diff --git a/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx b/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx index 8f2f85852..e0f2bceb7 100644 --- a/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx +++ b/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { SessionDispatchContentionError } from "@/features/chat/lib/sessionDispatchAcquisition"; import type { SessionDispatchReleaseWaiter } from "@/features/chat/lib/sessionTargetCoordinator"; import { QueuedMessageOwnershipLostError } from "@/features/chat/lib/preCommitSendRejection"; +import { resetReclaimedQueueReconciliationForTesting } from "@/features/chat/lib/reclaimedQueueReconciliation"; import { type QueuedMessageRecord, useChatStore, @@ -80,6 +81,7 @@ function resetChatStore(): void { describe("useBerdctlQueuedMessageDrain", () => { beforeEach(() => { vi.clearAllMocks(); + resetReclaimedQueueReconciliationForTesting(); mocks.sendPromptToExistingSessionInBackground.mockResolvedValue(undefined); mocks.sendQueuedPromptToExistingSessionInBackground.mockResolvedValue( undefined, diff --git a/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts b/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts index d1398be92..739e5dffc 100644 --- a/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts +++ b/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts @@ -15,13 +15,17 @@ import { useChatStore, } from "@/features/chat/stores/chatStore"; import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; -import { loadPersistedMessageQueues } from "@/features/chat/stores/queuePersistence"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import { isBerdctlCrossSessionQueuedMessage, sendPromptToExistingSessionInBackground, } from "@/features/berdctl/commands/runtime/sessionSend"; import { SessionDispatchContentionError } from "@/features/chat/lib/sessionDispatchAcquisition"; +import { + isReclaimedQueueReconciliationPending, + requestReclaimedQueueReconciliation, + subscribeReclaimedQueueReconciliation, +} from "@/features/chat/lib/reclaimedQueueReconciliation"; const drainingSessionIds = new Set(); const activeOwners = new Set(); @@ -35,7 +39,6 @@ type ContentionWaiter = { }; const contentionWaiters = new Map(); -let ownershipRefreshSequence = 0; function ownerIdFor(scopedSessionId?: string): string { return scopedSessionId ? `session:${scopedSessionId}` : "global"; @@ -63,28 +66,9 @@ function scheduleContentionResume( queueMicrotask(() => drainQueuedMessage(sessionId, waiter.ownerId)); } -async function refreshReclaimedQueues( - previousOpenSessions: Record, - openSessions: Record, -): Promise { - const reclaimedSessionIds = Object.keys(previousOpenSessions).filter( - (sessionId) => !(sessionId in openSessions), - ); - if (reclaimedSessionIds.length === 0) { - drainReadyQueuedMessages(); - return; - } - const sequence = ++ownershipRefreshSequence; - const persistedQueues = await loadPersistedMessageQueues(); - if (sequence !== ownershipRefreshSequence) return; - useChatStore - .getState() - .reconcileQueuedMessages(persistedQueues, reclaimedSessionIds); - drainReadyQueuedMessages(); -} - function drainQueuedMessage(queuedSessionId: string, ownerId: string): void { if (!activeOwners.has(ownerId)) return; + if (isReclaimedQueueReconciliationPending(queuedSessionId)) return; const sessionExists = Boolean( useChatSessionStore.getState().getSession(queuedSessionId), ); @@ -251,16 +235,20 @@ export function useBerdctlQueuedMessageDrain( (!previousState.hasLoadedSnapshot || state.openSessions !== previousState.openSessions) ) { - if (!previousState.hasLoadedSnapshot) { - drainReadyQueuedMessages(); - } else { - void refreshReclaimedQueues( + if ( + !previousState.hasLoadedSnapshot || + !requestReclaimedQueueReconciliation( previousState.openSessions, state.openSessions, - ); + ) + ) { + drainReadyQueuedMessages(); } } }); + const unsubscribeReclaimedQueues = subscribeReclaimedQueueReconciliation( + () => drainReadyQueuedMessages(queuedSessionId), + ); const unsubscribeSessionStore = useChatSessionStore.subscribe( (state, previousState) => { reconcileContentionWaiters(queuedSessionId); @@ -343,6 +331,7 @@ export function useBerdctlQueuedMessageDrain( ); return () => { unsubscribeWindowStore?.(); + unsubscribeReclaimedQueues(); unsubscribeSessionStore(); unsubscribeChatStore(); activeOwners.delete(ownerId); diff --git a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx index b6442a792..e06363ce0 100644 --- a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx @@ -9,6 +9,7 @@ import { registerForegroundQueueOwner, resetForegroundQueueOwnershipForTesting, } from "@/features/chat/lib/foregroundQueueOwnership"; +import { resetReclaimedQueueReconciliationForTesting } from "@/features/chat/lib/reclaimedQueueReconciliation"; import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; import { useChatStore } from "@/features/chat/stores/chatStore"; import * as queuePersistence from "@/features/chat/stores/queuePersistence"; @@ -101,6 +102,7 @@ describe("useBackgroundQueuedMessageDrain", () => { vi.clearAllMocks(); resetForegroundQueueOwnershipForTesting(); resetBackgroundQueueDrainStateForTesting(); + resetReclaimedQueueReconciliationForTesting(); vi.spyOn(queuePersistence, "loadPersistedMessageQueues").mockResolvedValue( {}, ); @@ -829,6 +831,73 @@ describe("useBackgroundQueuedMessageDrain", () => { ).toBeUndefined(); }); + it("serializes overlapping reclaims without draining either stale queue", async () => { + const staleA = { ...ordinaryRecord(), recordId: "stale-a" }; + const staleB = { ...ordinaryRecord(), recordId: "stale-b" }; + useChatStore.setState({ + queuedMessageBySession: { + "session-1": [staleA], + "detached-session": [staleB], + }, + }); + useSessionWindowStore.setState({ + openSessions: { + "session-1": "window-1", + "detached-session": "window-2", + }, + }); + let resolveFirst!: (queues: Record) => void; + let resolveSecond!: (queues: Record) => void; + vi.spyOn(queuePersistence, "loadPersistedMessageQueues") + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirst = resolve; + }), + ) + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveSecond = resolve; + }), + ); + + render(); + act(() => + useSessionWindowStore.setState({ + openSessions: { "detached-session": "window-2" }, + }), + ); + await waitFor(() => + expect(queuePersistence.loadPersistedMessageQueues).toHaveBeenCalledTimes( + 1, + ), + ); + act(() => useSessionWindowStore.setState({ openSessions: {} })); + + resolveFirst({}); + await waitFor(() => + expect(queuePersistence.loadPersistedMessageQueues).toHaveBeenCalledTimes( + 2, + ), + ); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + resolveSecond({}); + await waitFor(() => { + expect( + useChatStore.getState().queuedMessageBySession["session-1"], + ).toBeUndefined(); + expect( + useChatStore.getState().queuedMessageBySession["detached-session"], + ).toBeUndefined(); + }); + expect( + mocks.sendQueuedPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + }); + it("does not fire a reclaimed head that persistence marks restored", async () => { const stale = ordinaryRecord(); useChatStore.setState({ diff --git a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts index aaaa59a47..7138c185a 100644 --- a/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts @@ -22,14 +22,17 @@ import { type QueuedMessageRecord, useChatStore, } from "@/features/chat/stores/chatStore"; -import { loadPersistedMessageQueues } from "@/features/chat/stores/queuePersistence"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import { SessionDispatchContentionError } from "@/features/chat/lib/sessionDispatchAcquisition"; +import { + isReclaimedQueueReconciliationPending, + requestReclaimedQueueReconciliation, + subscribeReclaimedQueueReconciliation, +} from "@/features/chat/lib/reclaimedQueueReconciliation"; import { sendQueuedPromptToExistingSessionInBackground } from "@/features/chat/lib/queuedSessionSend"; const drainingSessionIds = new Set(); const activeOwners = new Set(); -let ownershipRefreshSequence = 0; type ContentionWaiter = { ownerId: string; record: QueuedMessageRecord; @@ -130,26 +133,6 @@ function isBackgroundDrainableHead( * Reconciled records arrive `restored: true`, which the background drain * refuses to claim. */ -async function refreshReclaimedQueues( - previousOpenSessions: Record, - openSessions: Record, -): Promise { - const reclaimedSessionIds = Object.keys(previousOpenSessions).filter( - (sessionId) => !(sessionId in openSessions), - ); - if (reclaimedSessionIds.length === 0) { - drainReadyQueuedMessages(); - return; - } - const sequence = ++ownershipRefreshSequence; - const persistedQueues = await loadPersistedMessageQueues(); - if (sequence !== ownershipRefreshSequence) return; - useChatStore - .getState() - .reconcileQueuedMessages(persistedQueues, reclaimedSessionIds); - drainReadyQueuedMessages(); -} - function reconcileContentionWaiters(scopedSessionId?: string): void { const sessionStore = useChatSessionStore.getState(); const queue = useChatStore.getState().queuedMessageBySession; @@ -190,6 +173,7 @@ function scheduleContentionResume( function drainQueuedMessage(sessionId: string, ownerId: string): void { if (!activeOwners.has(ownerId)) return; + if (isReclaimedQueueReconciliationPending(sessionId)) return; const sessionExists = Boolean( useChatSessionStore.getState().getSession(sessionId), ); @@ -361,13 +345,14 @@ export function useBackgroundQueuedMessageDrain( (!previousState.hasLoadedSnapshot || state.openSessions !== previousState.openSessions) ) { - if (!previousState.hasLoadedSnapshot) { - drainReadyQueuedMessages(); - } else { - void refreshReclaimedQueues( + if ( + !previousState.hasLoadedSnapshot || + !requestReclaimedQueueReconciliation( previousState.openSessions, state.openSessions, - ); + ) + ) { + drainReadyQueuedMessages(); } } }); @@ -381,6 +366,9 @@ export function useBackgroundQueuedMessageDrain( drainReadyQueuedMessages(scopedSessionId); }, ); + const unsubscribeReclaimedQueues = subscribeReclaimedQueueReconciliation( + () => drainReadyQueuedMessages(scopedSessionId), + ); const unsubscribeSessionStore = useChatSessionStore.subscribe( (state, previousState) => { reconcileContentionWaiters(scopedSessionId); @@ -444,6 +432,7 @@ export function useBackgroundQueuedMessageDrain( ); return () => { unsubscribeWindowStore?.(); + unsubscribeReclaimedQueues(); unsubscribeForegroundOwnership(); unsubscribeSessionStore(); unsubscribeChatStore(); diff --git a/src/features/chat/lib/reclaimedQueueReconciliation.ts b/src/features/chat/lib/reclaimedQueueReconciliation.ts new file mode 100644 index 000000000..db5116992 --- /dev/null +++ b/src/features/chat/lib/reclaimedQueueReconciliation.ts @@ -0,0 +1,92 @@ +import { useChatStore } from "@/features/chat/stores/chatStore"; +import { loadPersistedMessageQueues } from "@/features/chat/stores/queuePersistence"; +import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; + +const pendingSessionIds = new Set(); +const blockedSessionIds = new Set(); +const listeners = new Set<() => void>(); +let worker: Promise | undefined; +let lastObservedOpenSessions: Record | undefined; + +function notifyReconciled(): void { + for (const listener of [...listeners]) listener(); +} + +async function reconcilePendingQueues(): Promise { + while (pendingSessionIds.size > 0) { + const batch = [...pendingSessionIds]; + pendingSessionIds.clear(); + try { + const persistedQueues = await loadPersistedMessageQueues(); + const { openSessions } = useSessionWindowStore.getState(); + const stillReclaimed = batch.filter( + (sessionId) => !(sessionId in openSessions), + ); + if (stillReclaimed.length > 0) { + useChatStore + .getState() + .reconcileQueuedMessages(persistedQueues, stillReclaimed); + } + } finally { + for (const sessionId of batch) blockedSessionIds.delete(sessionId); + notifyReconciled(); + } + } +} + +/** + * Reconciles queues returning from detached session windows before any global + * drain may act on the main renderer's stale in-memory copy. Overlapping + * window closes accumulate behind one worker; each reclaimed session remains + * blocked until a persistence read covering it has settled. + */ +function ensureWorker(): void { + if (worker) return; + worker = Promise.resolve() + .then(reconcilePendingQueues) + .finally(() => { + worker = undefined; + if (pendingSessionIds.size > 0) ensureWorker(); + }); +} + +export function requestReclaimedQueueReconciliation( + previousOpenSessions: Record, + openSessions: Record, +): boolean { + if (lastObservedOpenSessions === openSessions) return false; + lastObservedOpenSessions = openSessions; + const reclaimedSessionIds = Object.keys(previousOpenSessions).filter( + (sessionId) => + !(sessionId in openSessions) && !blockedSessionIds.has(sessionId), + ); + if (reclaimedSessionIds.length === 0) return false; + + for (const sessionId of reclaimedSessionIds) { + blockedSessionIds.add(sessionId); + pendingSessionIds.add(sessionId); + } + ensureWorker(); + return true; +} + +export function isReclaimedQueueReconciliationPending( + sessionId: string, +): boolean { + return blockedSessionIds.has(sessionId); +} + +export function subscribeReclaimedQueueReconciliation( + listener: () => void, +): () => void { + listeners.add(listener); + return () => listeners.delete(listener); +} + +export function resetReclaimedQueueReconciliationForTesting(): void { + pendingSessionIds.clear(); + blockedSessionIds.clear(); + listeners.clear(); + worker = undefined; + lastObservedOpenSessions = undefined; +} diff --git a/src/shared/i18n/__tests__/queueFailureLocaleParity.test.ts b/src/shared/i18n/__tests__/queueFailureLocaleParity.test.ts new file mode 100644 index 000000000..54025f4de --- /dev/null +++ b/src/shared/i18n/__tests__/queueFailureLocaleParity.test.ts @@ -0,0 +1,14 @@ +import { describe, expect, it } from "vitest"; +import enChat from "../locales/en/chat.json"; +import esChat from "../locales/es/chat.json"; + +const keys = ["backgroundSendFailed", "backgroundSendFailedTitle"] as const; + +describe("background queue failure locale parity", () => { + it("provides translated Spanish copy for every new failure-toast string", () => { + for (const key of keys) { + expect(esChat.queue[key]).toBeTruthy(); + expect(esChat.queue[key]).not.toBe(enChat.queue[key]); + } + }); +}); diff --git a/src/shared/i18n/locales/es/chat.json b/src/shared/i18n/locales/es/chat.json index daa71bb18..12d76e8fd 100644 --- a/src/shared/i18n/locales/es/chat.json +++ b/src/shared/i18n/locales/es/chat.json @@ -410,6 +410,8 @@ "dismiss": "Descartar mensaje en cola", "edit": "Editar mensaje en cola", "sendAnyway": "Enviar de todos modos", + "backgroundSendFailed": "No se pudo enviar el mensaje en cola. Reinténtalo o edita el mensaje.", + "backgroundSendFailedTitle": "No se pudo enviar el mensaje en cola", "configureWorktree": "¿Configurar un worktree nuevo?", "configureWorkspaces": "¿Configurar espacios de trabajo nuevos para el proyecto?", "configureWorkspacePlan": "¿Configurar {{worktreeLabel}}?",