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..e0290ccf1 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", () => { + it("mounts the background 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/__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/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 66% rename from src/features/chat/hooks/useReleasedQueuedMessageDrain.test.tsx rename to src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx index ea6d11060..e06363ce0 100644 --- a/src/features/chat/hooks/useReleasedQueuedMessageDrain.test.tsx +++ b/src/features/chat/hooks/useBackgroundQueuedMessageDrain.test.tsx @@ -5,14 +5,30 @@ 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 { 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"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import type { QueuedMessageRecord } from "@/features/chat/stores/chatStore"; -import { useReleasedQueuedMessageDrain } from "./useReleasedQueuedMessageDrain"; +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", () => ({ @@ -27,7 +43,7 @@ function DrainHarness({ sessionId?: string; ownerReady?: boolean; } = {}) { - useReleasedQueuedMessageDrain(sessionId, ownerReady); + useBackgroundQueuedMessageDrain(sessionId, ownerReady); return null; } @@ -70,9 +86,26 @@ 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(); + resetBackgroundQueueDrainStateForTesting(); + resetReclaimedQueueReconciliationForTesting(); + vi.spyOn(queuePersistence, "loadPersistedMessageQueues").mockResolvedValue( + {}, + ); mocks.sendQueuedPromptToExistingSessionInBackground.mockResolvedValue( undefined, ); @@ -688,6 +721,32 @@ describe("useReleasedQueuedMessageDrain", () => { }, }); }); + 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 () => { @@ -737,6 +796,164 @@ describe("useReleasedQueuedMessageDrain", () => { }); }); + 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("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({ + 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({ @@ -757,20 +974,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 +999,156 @@ 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("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({ + 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 60% rename from src/features/chat/hooks/useReleasedQueuedMessageDrain.ts rename to src/features/chat/hooks/useBackgroundQueuedMessageDrain.ts index c6cb0219e..7138c185a 100644 --- a/src/features/chat/hooks/useReleasedQueuedMessageDrain.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 { @@ -11,6 +12,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, @@ -18,6 +24,11 @@ import { } from "@/features/chat/stores/chatStore"; 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(); @@ -33,10 +44,95 @@ 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 + * 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; 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" }, + sessionId: string, +): boolean { + if (isBerdctlCrossSessionQueuedMessage(record)) { + return false; + } + if (record.releasedFromDeferred) { + return true; + } + 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. + */ function reconcileContentionWaiters(scopedSessionId?: string): void { const sessionStore = useChatSessionStore.getState(); const queue = useChatStore.getState().queuedMessageBySession; @@ -72,11 +168,12 @@ 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; + if (isReclaimedQueueReconciliationPending(sessionId)) return; const sessionExists = Boolean( useChatSessionStore.getState().getSession(sessionId), ); @@ -106,7 +203,7 @@ function drainReleasedQueuedMessage(sessionId: string, ownerId: string): void { queuedMessage, sessionStore.getSession(sessionId), ) || - !queuedMessage.releasedFromDeferred || + !isBackgroundDrainableHead(queuedMessage, sessionId) || !isQueuedSessionReady(runtime) ) { return; @@ -161,9 +258,9 @@ function drainReleasedQueuedMessage(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( - `[released-queue] failed to send queued prompt for session ${sessionId}`, + `[background-queue] failed to send queued prompt for session ${sessionId}`, error, ); const current = @@ -176,6 +273,17 @@ function drainReleasedQueuedMessage(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(() => { @@ -186,10 +294,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,20 +314,21 @@ function getOwnedSessionIds( ); } -function drainReadyReleasedMessages(scopedSessionId?: string): void { +function drainReadyQueuedMessages(scopedSessionId?: string): void { const ownerId = ownerIdFor(scopedSessionId); if (!activeOwners.has(ownerId)) return; reconcileContentionWaiters(scopedSessionId); const { queuedMessageBySession } = useChatStore.getState(); + noteRestoredRecords(queuedMessageBySession); for (const sessionId of getOwnedSessionIds( queuedMessageBySession, scopedSessionId, )) { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } } -export function useReleasedQueuedMessageDrain( +export function useBackgroundQueuedMessageDrain( scopedSessionId?: string, ownerReady = true, ): void { @@ -227,7 +336,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 +345,30 @@ export function useReleasedQueuedMessageDrain( (!previousState.hasLoadedSnapshot || state.openSessions !== previousState.openSessions) ) { - drainReadyReleasedMessages(); + if ( + !previousState.hasLoadedSnapshot || + !requestReclaimedQueueReconciliation( + previousState.openSessions, + state.openSessions, + ) + ) { + drainReadyQueuedMessages(); + } } }); + // 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( + () => { + liftRestoredExclusionsForOwnedSessions(); + drainReadyQueuedMessages(scopedSessionId); + }, + ); + const unsubscribeReclaimedQueues = subscribeReclaimedQueueReconciliation( + () => drainReadyQueuedMessages(scopedSessionId), + ); const unsubscribeSessionStore = useChatSessionStore.subscribe( (state, previousState) => { reconcileContentionWaiters(scopedSessionId); @@ -258,17 +388,18 @@ export function useReleasedQueuedMessageDrain( ), ) ) { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } } if (state.hasHydratedSessions && !previousState.hasHydratedSessions) { - drainReadyReleasedMessages(scopedSessionId); + drainReadyQueuedMessages(scopedSessionId); } }, ); const unsubscribeChatStore = useChatStore.subscribe( (state, previousState) => { reconcileContentionWaiters(scopedSessionId); + noteRestoredRecords(state.queuedMessageBySession); for (const sessionId of getOwnedSessionIds( state.queuedMessageBySession, scopedSessionId, @@ -276,7 +407,7 @@ export function useReleasedQueuedMessageDrain( const queuedMessage = state.queuedMessageBySession[sessionId]?.[0]; if ( queuedMessage?.kind !== "transport-ready" || - !queuedMessage.releasedFromDeferred + !isBackgroundDrainableHead(queuedMessage, sessionId) ) { continue; } @@ -294,13 +425,15 @@ export function useReleasedQueuedMessageDrain( !currentBlocked && (previousBlocked || becameTransportReady) ) { - drainReleasedQueuedMessage(sessionId, ownerId); + drainQueuedMessage(sessionId, ownerId); } } }, ); return () => { unsubscribeWindowStore?.(); + unsubscribeReclaimedQueues(); + 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/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/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) { - + 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/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}}?", 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}}?",