diff --git a/apps/web/src/components/BackgroundQueuedMessages.test.tsx b/apps/web/src/components/BackgroundQueuedMessages.test.tsx new file mode 100644 index 000000000000..f81b2058847a --- /dev/null +++ b/apps/web/src/components/BackgroundQueuedMessages.test.tsx @@ -0,0 +1,213 @@ +import { act, create, type ReactTestRenderer } from "react-test-renderer"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import { EnvironmentId, ThreadId, ProviderInstanceId } from "@t3tools/contracts"; +import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { useComposerDraftStore } from "../composerDraftStore"; +import { useQueuedMessageStore } from "../queuedMessageStore"; +import { BackgroundQueuedMessages } from "./BackgroundQueuedMessages"; + +const io = vi.hoisted(() => ({ + threads: new Map(), + prepare: vi.fn(), + start: vi.fn(), + connected: true, + shellPresent: true, + config: {}, +})); +vi.mock("../state/entities", () => ({ + useThread: (ref: { threadId: string }) => io.threads.get(ref.threadId), + useServerConfigs: () => new Map([[EnvironmentId.make("env"), io.config]]), + useThreadShell: () => (io.shellPresent ? {} : null), +})); +vi.mock("../state/threads", () => ({ useEnvironmentThread: () => ({ status: "live" }) })); +vi.mock("../state/environments", () => ({ + useEnvironments: () => ({ + environments: [ + { + environmentId: EnvironmentId.make("env"), + connection: { phase: io.connected ? "connected" : "reconnecting" }, + }, + ], + }), +})); +vi.mock("../hooks/useSettings", () => ({ useClientSettingsHydrated: () => true })); +vi.mock("../lib/sendBackgroundQueuedMessage", () => ({ + sendBackgroundQueuedMessage: async (...args: unknown[]) => io.start(...args), +})); + +const ref = scopeThreadRef(EnvironmentId.make("env"), ThreadId.make("a")); +const key = scopedThreadKey(ref); +let root: ReactTestRenderer | undefined; +const render = async (activeThreadKey: string | null) => { + await act(async () => { + const element = ; + if (root) root.update(element); + else root = create(element); + }); +}; +function enqueue(prompt = "follow up") { + return useQueuedMessageStore.getState().enqueue(key, { + prompt, + images: [], + files: [], + terminalContexts: [], + previewAnnotations: [], + reviewComments: [], + submissionIntent: "foreground", + queuedAfterToolActivityId: null, + createdAt: "2026-09-22T00:00:00Z", + sendOptions: { + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, + runtimeMode: "full-access", + interactionMode: "default", + promptEffort: null, + }, + }); +} +beforeEach(() => { + (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + useQueuedMessageStore.setState({ + queuesByThreadKey: {}, + backgroundSendsByThreadKey: {}, + drainGeneration: 0, + }); + io.connected = true; + io.shellPresent = true; + io.threads.set("a", { session: { status: "running" }, activities: [] }); + io.start.mockReset().mockImplementation(async (_ref, message, _options, canSend) => { + if (!canSend()) return false; + return Boolean(useQueuedMessageStore.getState().take(key, message.id, null)); + }); +}); +afterEach(async () => { + if (root) await act(() => root!.unmount()); + root = undefined; +}); +describe("background queued messages", () => { + it("sends a completed thread's queue after navigating away, without selecting it again", async () => { + enqueue(); + await render(key); + await render("another-thread"); + expect(io.start).not.toHaveBeenCalled(); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render("another-thread"); + expect(io.start).toHaveBeenCalledOnce(); + expect(io.start.mock.calls[0]?.[0]).toEqual(ref); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toBeUndefined(); + }); + it("leaves selected and disconnected threads to wait", async () => { + enqueue(); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(key); + expect(io.start).not.toHaveBeenCalled(); + io.connected = false; + await render(null); + expect(io.start).not.toHaveBeenCalled(); + io.connected = true; + io.shellPresent = true; + await render(null); + expect(io.start).toHaveBeenCalledOnce(); + }); + it("advances after a text-only turn even when there is no new tool activity", async () => { + enqueue("first"); + enqueue("second"); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(null); + expect(io.start).toHaveBeenCalledOnce(); + io.threads.set("a", { session: { status: "running" }, activities: [] }); + await render(null); + expect(io.start).toHaveBeenCalledOnce(); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(null); + expect(io.start).toHaveBeenCalledTimes(2); + }); + it("waits for checkpoint rewind to finish", async () => { + enqueue(); + useComposerDraftStore.setState({ rewindingThreadKeys: new Set([key]) }); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(null); + expect(io.start).not.toHaveBeenCalled(); + await act(() => { + useComposerDraftStore.setState({ rewindingThreadKeys: new Set() }); + }); + expect(io.start).toHaveBeenCalledOnce(); + }); + it.each(["approval.requested", "user-input.requested"])( + "waits for %s to resolve", + async (kind) => { + enqueue(); + io.threads.set("a", { + session: { status: "ready" }, + activities: [ + { + id: "request", + kind, + createdAt: "2026-09-22T00:00:00Z", + payload: { + requestId: "req-1", + questions: [ + { + id: "q1", + header: "name", + question: "name?", + options: [], + allowCustomAnswer: true, + multiSelect: false, + }, + ], + }, + }, + ], + }); + await render(null); + expect(io.start).not.toHaveBeenCalled(); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(null); + expect(io.start).toHaveBeenCalledOnce(); + }, + ); + it("does not miss a boundary that arrives while dispatch is awaiting its receipt", async () => { + enqueue("first"); + enqueue("second"); + let release!: () => void; + const receipt = new Promise((resolve) => { + release = resolve; + }); + io.start.mockImplementationOnce(async (_ref, message) => { + useQueuedMessageStore.getState().take(key, message.id, null); + await receipt; + return true; + }); + io.threads.set("a", { + session: { status: "ready" }, + activities: [], + latestTurn: { turnId: "old" }, + }); + await render(null); + io.threads.set("a", { + session: { status: "ready" }, + activities: [], + latestTurn: { turnId: "new" }, + }); + await render(null); + await act(async () => release()); + expect(io.start).toHaveBeenCalledTimes(2); + }); + it("retries when the shell arrives after the detail", async () => { + enqueue(); + io.shellPresent = false; + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(null); + expect(io.start).not.toHaveBeenCalled(); + io.shellPresent = true; + await render(null); + expect(io.start).toHaveBeenCalledOnce(); + }); + it("does not send a held message", async () => { + const message = enqueue(); + useQueuedMessageStore.getState().holdAtFront(key, message); + io.threads.set("a", { session: { status: "ready" }, activities: [] }); + await render(null); + expect(io.start).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/web/src/components/BackgroundQueuedMessages.tsx b/apps/web/src/components/BackgroundQueuedMessages.tsx new file mode 100644 index 000000000000..587ce80e0818 --- /dev/null +++ b/apps/web/src/components/BackgroundQueuedMessages.tsx @@ -0,0 +1,139 @@ +import { useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; +import { useComposerDraftStore } from "../composerDraftStore"; +import { useShallow } from "zustand/react/shallow"; +import { parseScopedThreadKey, scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { useParams } from "@tanstack/react-router"; +import { resolveThreadRouteTarget } from "../threadRoutes"; +import { derivePendingRequests } from "@t3tools/client-runtime/pending-requests"; +import { + useQueuedMessageStore, + useQueuedMessages, + isQueuedMessageDue, + latestCompletedToolActivityId, +} from "../queuedMessageStore"; +import { useThread, useThreadShell, useServerConfigs } from "../state/entities"; +import { useEnvironmentThread } from "../state/threads"; +import { useEnvironments } from "../state/environments"; +import { useClientSettingsHydrated } from "../hooks/useSettings"; +import { derivePhase } from "../session-logic"; +import { sendBackgroundQueuedMessage } from "../lib/sendBackgroundQueuedMessage"; + +export function BackgroundQueueCoordinator() { + const target = useParams({ strict: false, select: resolveThreadRouteTarget }); + return ( + + ); +} + +/** Subscribe only to threads with queued work, independently of the selected route. */ +export function BackgroundQueuedMessages({ activeThreadKey }: { activeThreadKey: string | null }) { + const keys = useQueuedMessageStore( + useShallow((state) => [ + ...new Set([ + ...Object.keys(state.queuesByThreadKey), + ...Object.keys(state.backgroundSendsByThreadKey), + ]), + ]), + ); + return keys.map((threadKey) => ( + + )); +} + +function BackgroundThreadQueue({ threadKey, active }: { threadKey: string; active: boolean }) { + const ref = useMemo(() => parseScopedThreadKey(threadKey), [threadKey]); + const thread = useThread(ref); + const shell = useThreadShell(ref); + const sending = useQueuedMessageStore( + (state) => state.backgroundSendsByThreadKey[threadKey] !== undefined, + ); + const detail = useEnvironmentThread(ref?.environmentId ?? null, ref?.threadId ?? null); + const { environments } = useEnvironments(); + const configs = useServerConfigs(); + const hydrated = useClientSettingsHydrated(); + const message = useQueuedMessages(threadKey)[0]; + const rewinding = useComposerDraftStore((state) => state.rewindingThreadKeys.has(threadKey)); + const config = ref ? configs.get(ref.environmentId) : undefined; + const phase = derivePhase(thread?.session ?? null); + const latestToolActivityId = latestCompletedToolActivityId(thread?.activities ?? []); + const pending = derivePendingRequests(thread?.activities ?? []); + const connected = environments.some( + (environment) => + environment.environmentId === ref?.environmentId && + environment.connection.phase === "connected", + ); + const blocked = + (active && !sending) || + rewinding || + !connected || + !hydrated || + !thread || + !shell || + detail.status !== "live" || + pending.approvals.length > 0 || + pending.userInputs.length > 0; + const busy = useRef(false); + const [completedAttempt, setCompletedAttempt] = useState(null); + const lastSentBoundary = useRef(null); + const boundary = JSON.stringify([phase, latestToolActivityId, thread?.latestTurn?.turnId]); + const live = useRef({ blocked, phase, latestToolActivityId, message, config, boundary }); + const mounted = useRef(false); + useLayoutEffect(() => { + live.current = { blocked, phase, latestToolActivityId, message, config, boundary }; + }, [blocked, phase, latestToolActivityId, message, config, boundary]); + useEffect(() => { + mounted.current = true; + return () => { + mounted.current = false; + }; + }, []); + useEffect(() => { + if (lastSentBoundary.current !== boundary) lastSentBoundary.current = null; + if ( + !ref || + !config || + !message?.sendOptions || + blocked || + busy.current || + completedAttempt === live.current || + lastSentBoundary.current === boundary + ) + return; + const canSend = () => { + const current = live.current; + return ( + mounted.current && + !current.blocked && + isQueuedMessageDue({ + message, + phase: current.phase, + latestToolActivityId: current.latestToolActivityId, + }) + ); + }; + if (!canSend()) return; + busy.current = true; + const attempt = live.current; + void sendBackgroundQueuedMessage( + ref, + message, + message.sendOptions, + canSend, + () => live.current.latestToolActivityId, + ) + .then((sent) => { + if (sent) lastSentBoundary.current = boundary; + }) + .finally(() => { + busy.current = false; + if (mounted.current) setCompletedAttempt(attempt); + }); + }, [ref, message, blocked, boundary, config, completedAttempt]); + return null; +} diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index b97411e7a44b..52868c0a4ecf 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -3201,7 +3201,7 @@ export default function ChatView(props: ChatViewProps) { localDispatchStartedAt, latestUserMessageAt, isPreparingWorktree: isLocallyPreparingWorktree, - isSendBusy, + isSendBusy: isLocalSendBusy, backgroundSubmissionPending, } = useLocalDispatchState({ activeThread, @@ -3211,6 +3211,11 @@ export default function ChatView(props: ChatViewProps) { activePendingUserInput: activePendingUserInput?.requestId ?? null, threadError, }); + const isBackgroundQueueSending = useQueuedMessageStore( + (state) => + activeThreadKey !== null && state.backgroundSendsByThreadKey[activeThreadKey] !== undefined, + ); + const isSendBusy = isLocalSendBusy || isBackgroundQueueSending; const optimisticCompactionMessage = optimisticUserMessages.at(-1); const pendingCompactionMessage = isSendBusy && @@ -7651,6 +7656,19 @@ export default function ChatView(props: ChatViewProps) { previewAnnotations: [...composerPreviewAnnotations], reviewComments: [...composerReviewComments], submissionIntent, + sendOptions: { + modelSelection: ctxSelectedModelSelection, + runtimeMode, + interactionMode: sendInteractionMode, + promptEffort: resolvePromptInjectedEffort( + getProviderModelCapabilities( + ctxSelectedProviderModels, + ctxSelectedModel, + ctxSelectedProvider, + ), + ctxSelectedPromptEffort, + ), + }, queuedAfterToolActivityId: latestCompletedToolActivityId(threadActivities), createdAt: new Date().toISOString(), }); diff --git a/apps/web/src/lib/sendBackgroundQueuedMessage.test.ts b/apps/web/src/lib/sendBackgroundQueuedMessage.test.ts new file mode 100644 index 000000000000..aca97f8e0f8a --- /dev/null +++ b/apps/web/src/lib/sendBackgroundQueuedMessage.test.ts @@ -0,0 +1,381 @@ +import { beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import { createElement } from "react"; +import { act, create, type ReactTestRenderer } from "react-test-renderer"; +import { BackgroundQueuedMessages } from "../components/BackgroundQueuedMessages"; +import { EnvironmentId, ThreadId, ProviderInstanceId } from "@t3tools/contracts"; +import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { useQueuedMessageStore, type QueuedMessageSendOptions } from "../queuedMessageStore"; +import { sendBackgroundQueuedMessage } from "./sendBackgroundQueuedMessage"; + +const io = vi.hoisted(() => ({ + run: vi.fn(), + upload: vi.fn(), + release: vi.fn(), + toast: vi.fn(), + config: {}, + threads: new Map(), +})); +vi.mock("@t3tools/client-runtime/state/runtime", async (load) => ({ + ...(await load()), + runAtomCommand: (...args: unknown[]) => io.run(...args), +})); +vi.mock("../rpc/atomRegistry", () => ({ + appAtomRegistry: { get: () => new Map([["env-a", io.config]]) }, +})); +vi.mock("../state/server", () => ({ environmentServerConfigsAtom: {} })); +vi.mock("../state/threads", () => ({ + useEnvironmentThread: () => ({ status: "live" }), + threadEnvironment: { + updateMetadata: "metadata", + setRuntimeMode: "runtime", + setInteractionMode: "interaction", + startTurn: "start", + }, +})); +vi.mock("../state/environments", () => ({ + useEnvironments: () => ({ + environments: [{ environmentId: "env-a", connection: { phase: "connected" } }], + }), +})); +vi.mock("../hooks/useSettings", () => ({ useClientSettingsHydrated: () => true })); +vi.mock("../state/entities", () => ({ + useThread: (ref: { threadId: string }) => io.threads.get(ref.threadId), + useServerConfigs: () => new Map([["env-a", io.config]]), + useThreadShell: () => ({}), + readThreadShell: () => ({ runtimeMode: "full-access", interactionMode: "default" }), +})); +vi.mock("../components/ui/toast", () => ({ + toastManager: { add: (...args: unknown[]) => io.toast(...args) }, +})); +vi.mock("./attachmentUploadQueue", () => ({ + startAttachmentUpload: vi.fn(), + awaitAttachmentUploads: (...args: unknown[]) => io.upload(...args), + getUploadedAttachments: () => [ + { type: "image", id: "uploaded-image", name: "test.png", mimeType: "image/png", sizeBytes: 4 }, + ], + releaseDraftAttachments: (...args: unknown[]) => io.release(...args), +})); +const ref = scopeThreadRef(EnvironmentId.make("env-a"), ThreadId.make("thread-a")); +const key = scopedThreadKey(ref); +const options: QueuedMessageSendOptions = { + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, + runtimeMode: "full-access", + interactionMode: "default", + promptEffort: null, +}; +function enqueue() { + return useQueuedMessageStore.getState().enqueue(key, { + prompt: "follow up", + images: [], + files: [], + terminalContexts: [], + previewAnnotations: [], + reviewComments: [], + submissionIntent: "foreground", + queuedAfterToolActivityId: null, + createdAt: "2026-09-22T00:00:00Z", + sendOptions: options, + }); +} +const startCalls = () => io.run.mock.calls.filter((call) => call[1] === "start"); +beforeEach(() => { + useQueuedMessageStore.setState({ + queuesByThreadKey: {}, + backgroundSendsByThreadKey: {}, + drainGeneration: 0, + }); + io.run.mockReset().mockResolvedValue({ _tag: "Success", value: undefined }); + io.upload.mockReset().mockResolvedValue(undefined); + io.release.mockReset(); + io.toast.mockReset(); + io.config = { + providers: [ + { instanceId: "codex", driver: "codex", enabled: true, installed: true, status: "ready" }, + ], + environment: { capabilities: { attachmentUploads: true, inlineMessageContext: true } }, + }; +}); +describe("background queued send", () => { + it("runs the mounted processor through preparation and dispatch after navigating away", async () => { + (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + enqueue(); + io.threads.set("thread-a", { session: { status: "running" }, activities: [] }); + let root!: ReactTestRenderer; + await act(() => { + root = create(createElement(BackgroundQueuedMessages, { activeThreadKey: key })); + }); + try { + await act(() => + root.update(createElement(BackgroundQueuedMessages, { activeThreadKey: "thread-b" })), + ); + expect(startCalls()).toHaveLength(0); + io.threads.set("thread-a", { session: { status: "ready" }, activities: [] }); + await act(() => + root.update(createElement(BackgroundQueuedMessages, { activeThreadKey: "thread-b" })), + ); + expect(startCalls()).toHaveLength(1); + expect(startCalls()[0]?.[2]).toMatchObject({ + environmentId: "env-a", + input: { threadId: "thread-a", message: { text: "follow up" } }, + }); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toBeUndefined(); + } finally { + await act(() => root.unmount()); + } + }); + it("claims the queued intent before any persistent thread mutation", async () => { + const message = enqueue(); + io.run.mockImplementation(async () => { + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toBeUndefined(); + return { _tag: "Success", value: undefined }; + }); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ).toBe(true); + }); + it("dispatches to the original environment/thread with the queued model and modes", async () => { + const message = enqueue(); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => "tool-2", + ), + ).toBe(true); + expect(startCalls()).toHaveLength(1); + expect(startCalls()[0]?.[2]).toMatchObject({ + environmentId: "env-a", + input: { + threadId: "thread-a", + modelSelection: options.modelSelection, + runtimeMode: options.runtimeMode, + interactionMode: options.interactionMode, + message: { text: "follow up", attachments: [] }, + }, + }); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toBeUndefined(); + }); + it("Stop after claiming an intent prevents the turn and retains it for manual retry", async () => { + const message = enqueue(); + io.run.mockImplementation(async (_registry, command) => { + if (command === "metadata") useQueuedMessageStore.getState().drain(key); + return { _tag: "Success", value: undefined }; + }); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ).toBe(false); + expect(startCalls()).toHaveLength(0); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toEqual([ + { ...message, holdUntilUserAction: true }, + ]); + expect(useQueuedMessageStore.getState().backgroundSendsByThreadKey[key]).toBeUndefined(); + }); + it("rechecks approvals and connection gates after async preparation", async () => { + const message = enqueue(); + let allowed = true; + io.run.mockImplementation(async () => { + allowed = false; + return { _tag: "Success", value: undefined }; + }); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => allowed, + () => null, + ), + ).toBe(false); + expect(startCalls()).toHaveLength(0); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toHaveLength(1); + }); + it("holds failures at the head without losing the following message", async () => { + const message = enqueue(); + const second = enqueue(); + io.run.mockRejectedValue(new Error("offline")); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ).toBe(false); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toEqual([ + { ...message, holdUntilUserAction: true }, + second, + ]); + expect(io.toast).toHaveBeenCalledOnce(); + }); + it("does not send a message held while preparation was pending", async () => { + const message = enqueue(); + io.run.mockImplementation(async () => { + useQueuedMessageStore.getState().holdAtFront(key, message); + return { _tag: "Success", value: undefined }; + }); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ).toBe(false); + expect(startCalls()).toHaveLength(0); + }); + it("preserves uploaded image references and review context", async () => { + const message = enqueue(); + message.images.push({ + type: "image", + id: "local-image", + name: "test.png", + mimeType: "image/png", + sizeBytes: 4, + previewUrl: "blob:test", + file: new File(["test"], "test.png", { type: "image/png" }), + }); + message.reviewComments.push({ + id: "review", + sectionId: "s", + sectionTitle: "Review", + filePath: "a.ts", + startIndex: 0, + endIndex: 0, + rangeLabel: "L1", + text: "check this branch", + diff: "", + }); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ).toBe(true); + expect(startCalls()[0]?.[2]).toMatchObject({ + input: { + message: { + attachments: [{ id: "uploaded-image" }], + context: { + records: expect.arrayContaining([ + expect.objectContaining({ + kind: "review-comment", + contextId: "review-comment_review", + text: "check this branch", + }), + expect.objectContaining({ + kind: "image", + contextId: "image_local-image", + attachmentId: "uploaded-image", + }), + ]), + }, + }, + }, + }); + expect(io.release).toHaveBeenCalledWith(message.images); + }); + it("keeps an uploading message cancellable and never starts a turn after Stop", async () => { + const message = enqueue(); + message.images.push({ + type: "image", + id: "local-image", + name: "test.png", + mimeType: "image/png", + sizeBytes: 4, + previewUrl: "blob:test", + file: new File(["test"], "test.png", { type: "image/png" }), + }); + let release!: () => void; + io.upload.mockImplementation( + () => + new Promise((resolve) => { + release = resolve; + }), + ); + const sending = sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ); + expect(useQueuedMessageStore.getState().drain(key)).toEqual([message]); + release(); + expect(await sending).toBe(false); + expect(startCalls()).toHaveLength(0); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toBeUndefined(); + }); + it("revalidates the current provider catalog after settings persistence", async () => { + const message = enqueue(); + const provider = { + instanceId: "codex", + driver: "antigravity", + enabled: true, + installed: true, + status: "ready", + auth: { status: "authenticated" }, + models: [{ slug: "gpt-5" }], + }; + io.config = { + providers: [provider], + environment: { capabilities: { attachmentUploads: true, inlineMessageContext: true } }, + }; + io.run.mockImplementation(async () => { + provider.models = [{ slug: "replacement-model" }]; + return { _tag: "Success", value: undefined }; + }); + expect( + await sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ).toBe(false); + expect(startCalls()).toHaveLength(0); + expect(useQueuedMessageStore.getState().queuesByThreadKey[key]).toEqual([ + { ...message, holdUntilUserAction: true }, + ]); + }); + it("has only one dispatch winner if two preparations race", async () => { + const message = enqueue(); + const results = await Promise.all([ + sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + sendBackgroundQueuedMessage( + ref, + message, + options, + () => true, + () => null, + ), + ]); + expect(results.filter(Boolean)).toHaveLength(1); + expect(startCalls()).toHaveLength(1); + }); +}); diff --git a/apps/web/src/lib/sendBackgroundQueuedMessage.ts b/apps/web/src/lib/sendBackgroundQueuedMessage.ts new file mode 100644 index 000000000000..fb9660b52601 --- /dev/null +++ b/apps/web/src/lib/sendBackgroundQueuedMessage.ts @@ -0,0 +1,272 @@ +import type { ScopedThreadRef } from "@t3tools/contracts"; +import { scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { runAtomCommand, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import { applyClaudePromptEffortPrefix } from "@t3tools/shared/model"; +import { serializeLegacyContextMessage } from "@t3tools/shared/composerContextLegacySend"; +import { appAtomRegistry } from "../rpc/atomRegistry"; +import { environmentServerConfigsAtom } from "../state/server"; +import { threadEnvironment } from "../state/threads"; +import { readThreadShell } from "../state/entities"; +import { + useQueuedMessageStore, + type QueuedComposerMessage, + type QueuedMessageSendOptions, +} from "../queuedMessageStore"; +import { + deriveComposerSendState, + getAntigravitySendBlockReason, + readFileAsDataUrl, +} from "../components/ChatView.logic"; +import { getComposerSubmissionValidationMessage } from "../components/chat/composerSubmission"; +import { ATTACHMENT_ONLY_BOOTSTRAP_PROMPT } from "../components/chat/composerPromptHistory"; +import { fileAttachmentCapabilityBlockReason } from "../components/chat/composerAttachmentFiles"; +import { toastManager } from "../components/ui/toast"; +import { buildMessageContext, terminalContextReference } from "./composerContextRecords"; +import { removeInlineContextReference } from "./composerContextReferences"; +import { + awaitAttachmentUploads, + getUploadedAttachments, + releaseDraftAttachments, + startAttachmentUpload, +} from "./attachmentUploadQueue"; +import { newMessageId } from "./utils"; + +/** Uploads stay cancellable. Taking the intent commits it before persistent thread updates. */ +export async function sendBackgroundQueuedMessage( + ref: ScopedThreadRef, + message: QueuedComposerMessage, + options: QueuedMessageSendOptions, + canSend: () => boolean, + latestToolActivityId: () => string | null, +): Promise { + const key = scopedThreadKey(ref); + const stillQueued = () => + useQueuedMessageStore + .getState() + .queuesByThreadKey[key]?.some((entry) => entry.id === message.id) === true; + const readConfig = () => appAtomRegistry.get(environmentServerConfigsAtom).get(ref.environmentId); + const readyToSend = () => { + const state = useQueuedMessageStore.getState(); + const head = state.queuesByThreadKey[key]?.[0]; + const ownsIntent = taken + ? state.backgroundSendsByThreadKey[key]?.cancelled === false + : head?.id === message.id; + const provider = readConfig()?.providers.find( + (entry) => entry.instanceId === options.modelSelection.instanceId, + ); + const blockReason = getAntigravitySendBlockReason(provider, options.modelSelection.model); + if (blockReason) throw new Error(blockReason); + return ( + canSend() && + ownsIntent && + !(head?.id === message.id && head.holdUntilUserAction) && + provider?.enabled === true && + provider.installed && + provider.availability !== "unavailable" && + provider.status === "ready" + ); + }; + const attachments = [...message.images, ...message.files]; + let taken = false; + try { + const config = readConfig(); + const provider = config?.providers.find( + (entry) => entry.instanceId === options.modelSelection?.instanceId, + ); + if ( + !canSend() || + !stillQueued() || + !config || + !provider?.enabled || + !provider.installed || + provider.availability === "unavailable" || + provider.status !== "ready" + ) + return false; + const providerBlockReason = getAntigravitySendBlockReason( + provider, + options.modelSelection?.model ?? "", + ); + if (providerBlockReason) throw new Error(providerBlockReason); + const { sendableTerminalContexts, hasSendableContent } = deriveComposerSendState({ + prompt: message.prompt, + imageCount: attachments.length, + terminalContexts: message.terminalContexts, + elementContextCount: message.previewAnnotations.length + message.reviewComments.length, + }); + if (!hasSendableContent) { + useQueuedMessageStore.getState().remove(key, message.id); + return false; + } + const prompt = message.terminalContexts + .filter((context) => !sendableTerminalContexts.includes(context)) + .reduce( + (text, context) => + removeInlineContextReference(text, terminalContextReference(context).contextId).prompt, + message.prompt, + ) + .trim(); + const text = applyClaudePromptEffortPrefix( + prompt || ATTACHMENT_ONLY_BOOTSTRAP_PROMPT, + options.promptEffort, + ); + const validation = getComposerSubmissionValidationMessage({ + prompt: message.prompt, + providerInput: text, + submissionTarget: "provider-turn", + }); + if (validation) throw new Error(validation); + const checkFiles = () => { + const current = readConfig(); + const reason = fileAttachmentCapabilityBlockReason({ + files: message.files, + attachmentUploadsCapabilityKnown: current !== undefined, + supportsAttachmentUploads: current?.environment.capabilities.attachmentUploads === true, + maxFileAttachmentBytes: + current?.environment.capabilities.fileAttachments?.maxUploadBytes ?? null, + }); + if (reason) throw new Error(reason); + }; + checkFiles(); + const upload = config.environment.capabilities.attachmentUploads === true; + if (upload && attachments.length > 0) { + for (const attachment of attachments) + startAttachmentUpload({ + environmentId: ref.environmentId, + image: attachment, + draftTarget: ref, + }); + await awaitAttachmentUploads(attachments.map((attachment) => attachment.id)); + } + const wireAttachments = await Promise.all( + attachments.map(async (attachment) => { + if (upload) { + const uploaded = getUploadedAttachments({ + environmentId: ref.environmentId, + images: [attachment], + })?.[0]; + if (!uploaded) throw new Error("Retry or remove failed uploads before sending."); + return uploaded; + } + if (attachment.type !== "image") + throw new Error("This server does not support file attachments."); + return { + type: "image" as const, + id: attachment.id, + name: attachment.name, + mimeType: attachment.mimeType, + sizeBytes: attachment.sizeBytes, + dataUrl: await readFileAsDataUrl(attachment.file), + ...(attachment.source ? { source: attachment.source } : {}), + }; + }), + ); + checkFiles(); + if (!readyToSend()) return false; + const shell = readThreadShell(ref); + if (!shell) return false; + // Claim before mutating settings. A queued Cancel can only win before this + // point; Stop can still cancel the owned send through its in-flight marker. + if (!useQueuedMessageStore.getState().take(key, message.id, latestToolActivityId(), true)) + return false; + taken = true; + const stopIfBlocked = () => { + if (readyToSend()) return false; + useQueuedMessageStore.getState().holdAtFront(key, message); + return true; + }; + const createdAt = new Date().toISOString(); + if (options.modelSelection) { + const result = await runAtomCommand( + appAtomRegistry, + threadEnvironment.updateMetadata, + { + environmentId: ref.environmentId, + input: { threadId: ref.threadId, modelSelection: options.modelSelection }, + }, + { reportFailure: false }, + ); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + } + if (stopIfBlocked()) return false; + if (options.runtimeMode && options.runtimeMode !== shell.runtimeMode) { + const result = await runAtomCommand( + appAtomRegistry, + threadEnvironment.setRuntimeMode, + { + environmentId: ref.environmentId, + input: { threadId: ref.threadId, runtimeMode: options.runtimeMode, createdAt }, + }, + { reportFailure: false }, + ); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + } + if (stopIfBlocked()) return false; + if (options.interactionMode && options.interactionMode !== shell.interactionMode) { + const result = await runAtomCommand( + appAtomRegistry, + threadEnvironment.setInteractionMode, + { + environmentId: ref.environmentId, + input: { threadId: ref.threadId, interactionMode: options.interactionMode, createdAt }, + }, + { reportFailure: false }, + ); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + } + checkFiles(); + if (stopIfBlocked()) return false; + const context = buildMessageContext({ + terminalContexts: sendableTerminalContexts, + previewAnnotations: message.previewAnnotations, + reviewComments: message.reviewComments, + attachments: attachments.map((attachment, index) => ({ + attachment, + attachmentId: wireAttachments[index]?.id ?? attachment.id, + })), + }); + const inlineContext = readConfig()?.environment.capabilities.inlineMessageContext === true; + const result = await runAtomCommand( + appAtomRegistry, + threadEnvironment.startTurn, + { + environmentId: ref.environmentId, + input: { + threadId: ref.threadId, + message: { + messageId: newMessageId(), + role: "user", + text: + context && !inlineContext + ? serializeLegacyContextMessage({ text, records: context.records }) + : text, + attachments: wireAttachments, + ...(context && inlineContext ? { context } : {}), + }, + modelSelection: options.modelSelection, + runtimeMode: options.runtimeMode, + interactionMode: options.interactionMode, + createdAt, + }, + }, + { reportFailure: false }, + ); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + if (upload) releaseDraftAttachments(attachments); + return true; + } catch (error) { + // A cancelled preparation must not resurrect a message the user removed. + if (taken || stillQueued()) { + useQueuedMessageStore.getState().holdAtFront(key, message); + toastManager.add({ + type: "error", + title: "Queued message not sent", + description: + error instanceof Error ? error.message : "Open the thread to retry the queued message.", + }); + } + return false; + } finally { + if (taken) useQueuedMessageStore.getState().finishBackgroundSend(key); + } +} diff --git a/apps/web/src/queuedMessageStore.test.ts b/apps/web/src/queuedMessageStore.test.ts index 33869d89da62..beaa864911ba 100644 --- a/apps/web/src/queuedMessageStore.test.ts +++ b/apps/web/src/queuedMessageStore.test.ts @@ -23,7 +23,11 @@ function makeMessage(prompt: string): Omit { describe("queuedMessageStore", () => { beforeEach(() => { - useQueuedMessageStore.setState({ queuesByThreadKey: {}, drainGeneration: 0 }); + useQueuedMessageStore.setState({ + queuesByThreadKey: {}, + backgroundSendsByThreadKey: {}, + drainGeneration: 0, + }); }); it("keeps messages in submission order per thread", () => { @@ -60,6 +64,21 @@ describe("queuedMessageStore", () => { ).toBe(false); }); + it("owns a background send across navigation and scopes Stop to its thread", () => { + const store = useQueuedMessageStore.getState(); + const a = store.enqueue("thread-a", makeMessage("a")); + const b = store.enqueue("thread-b", makeMessage("b")); + expect(store.take("thread-a", a.id, null, true)).toEqual(a); + expect(store.take("thread-b", b.id, null, true)).toEqual(b); + expect(store.drain("thread-a")).toEqual([]); + expect(useQueuedMessageStore.getState().backgroundSendsByThreadKey).toEqual({ + "thread-a": { cancelled: true }, + "thread-b": { cancelled: false }, + }); + store.finishBackgroundSend("thread-a"); + store.finishBackgroundSend("thread-b"); + expect(useQueuedMessageStore.getState().backgroundSendsByThreadKey).toEqual({}); + }); it("remove keeps the other messages' anchors", () => { const { enqueue, remove } = useQueuedMessageStore.getState(); const first = enqueue("thread-a", { ...makeMessage("first"), queuedAfterToolActivityId: "t1" }); diff --git a/apps/web/src/queuedMessageStore.ts b/apps/web/src/queuedMessageStore.ts index b342a1380f7a..318726150500 100644 --- a/apps/web/src/queuedMessageStore.ts +++ b/apps/web/src/queuedMessageStore.ts @@ -1,4 +1,10 @@ -import type { PreviewAnnotationPayload } from "@t3tools/contracts"; +import type { + ModelSelection, + PreviewAnnotationPayload, + ProviderInteractionMode, + RuntimeMode, +} from "@t3tools/contracts"; +import type { resolvePromptInjectedEffort } from "@t3tools/shared/model"; import { create } from "zustand"; import type { ComposerSubmissionIntent } from "./composer-logic"; @@ -12,6 +18,13 @@ import type { ReviewCommentContext } from "./reviewCommentContext"; * carries the full draft snapshot so the send path can dispatch it later with * the same text, attachments, and contexts the user pressed Enter on. */ +export interface QueuedMessageSendOptions { + modelSelection: ModelSelection; + runtimeMode: RuntimeMode; + interactionMode: ProviderInteractionMode; + promptEffort: ReturnType; +} + export interface QueuedComposerMessage { id: string; prompt: string; @@ -21,6 +34,8 @@ export interface QueuedComposerMessage { previewAnnotations: PreviewAnnotationPayload[]; reviewComments: ReviewCommentContext[]; submissionIntent: ComposerSubmissionIntent; + /** Snapshot at enqueue time; background sends must never read another thread's composer. */ + sendOptions?: QueuedMessageSendOptions; /** * The newest completed tool activity at queue time. A different id later * means a tool call finished after the user queued, which is the boundary @@ -37,6 +52,8 @@ export interface QueuedComposerMessage { interface QueuedMessageStoreState { queuesByThreadKey: Record; + backgroundSendsByThreadKey: Record; + finishBackgroundSend: (threadKey: string) => void; /** * Bumped by `drain`. A send that took a message before a drain and finishes * its upload after it compares this to the value it captured and gives up, @@ -53,6 +70,7 @@ interface QueuedMessageStoreState { threadKey: string, id: string, toolActivityId: string | null, + background?: boolean, ) => QueuedComposerMessage | null; /** Removes one message without touching the others' anchors. Null when already gone. */ remove: (threadKey: string, id: string) => QueuedComposerMessage | null; @@ -70,6 +88,13 @@ const EMPTY_QUEUE: QueuedComposerMessage[] = []; /** In-memory only: a queued message is a live intent, not a draft worth persisting. */ export const useQueuedMessageStore = create()((set, get) => ({ queuesByThreadKey: {}, + backgroundSendsByThreadKey: {}, + finishBackgroundSend: (threadKey) => + set((state) => { + const backgroundSendsByThreadKey = { ...state.backgroundSendsByThreadKey }; + delete backgroundSendsByThreadKey[threadKey]; + return { backgroundSendsByThreadKey }; + }), drainGeneration: 0, enqueue: (threadKey, message) => { const entry: QueuedComposerMessage = { ...message, id: randomUUID() }; @@ -81,10 +106,10 @@ export const useQueuedMessageStore = create()((set, get })); return entry; }, - take: (threadKey, id, toolActivityId) => { + take: (threadKey, id, toolActivityId, background = false) => { const queue = get().queuesByThreadKey[threadKey]; const entry = queue?.find((message) => message.id === id); - if (!queue || !entry) { + if (!queue || !entry || get().backgroundSendsByThreadKey[threadKey]) { return null; } set((state) => { @@ -101,7 +126,17 @@ export const useQueuedMessageStore = create()((set, get } else { queuesByThreadKey[threadKey] = remaining; } - return { queuesByThreadKey }; + return { + queuesByThreadKey, + ...(background + ? { + backgroundSendsByThreadKey: { + ...state.backgroundSendsByThreadKey, + [threadKey]: { cancelled: false }, + }, + } + : {}), + }; }); return entry; }, @@ -139,6 +174,14 @@ export const useQueuedMessageStore = create()((set, get }); }, drain: (threadKey) => { + if (get().backgroundSendsByThreadKey[threadKey]) { + set((state) => ({ + backgroundSendsByThreadKey: { + ...state.backgroundSendsByThreadKey, + [threadKey]: { cancelled: true }, + }, + })); + } const queue = get().queuesByThreadKey[threadKey]; if (!queue || queue.length === 0) { return EMPTY_QUEUE; diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index eee703ac3f47..60e248ae4ded 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -74,6 +74,7 @@ import { } from "../components/KeybindingsUpdateToast.logic"; import { getDesktopSnapShotBridge } from "../lib/desktopSnapShot"; +import { BackgroundQueueCoordinator } from "../components/BackgroundQueuedMessages"; import { installDesktopPasteAsText } from "../lib/desktopPasteAsText"; import { shouldResumeSnapShotSetupOnStartup } from "../lib/snapShotSetupResume"; @@ -224,6 +225,7 @@ function RootRouteView() { +