authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-26 01:46:14-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-26 01:46:14-07:00
log295d7cba09bb4b4084482a74b5a42a0cac72a0b1
tree05144bc2d2589b46c128ea589da2cb9a2ba0384b
parent81e0491d2bdec153bf1ee12d4e1deab3909b7428
signature Signed by PGP key B5690EEEBB952194

fix(web): queued messages send while their thread is not open (#13764)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

11 files changed, 892 insertions(+), 302 deletions(-)

apps/web/src/components/ChatView.tsx+61-147
......@@ -324,12 +324,13 @@ import {
324324 terminalContextReference,
325325} from "../lib/composerContextRecords";
326326import {
327 isQueuedMessageDue,
328327 latestCompletedToolActivityId,
329328 type QueuedComposerMessage,
329 type QueuedMessageSendSettings,
330330 useQueuedMessages,
331331 useQueuedMessageStore,
332332} from "../queuedMessageStore";
333import { sendQueuedMessage } from "./chat/sendQueuedMessage";
333334import { type ReviewCommentContext } from "../reviewCommentContext";
334335import { environmentCatalog } from "../connection/catalog";
335336import { isDesktopLocalConnectionTarget } from "../connection/desktopLocal";
......@@ -7248,10 +7249,27 @@ export default function ChatView(props: ChatViewProps) {
72487249 };
72497250
72507251 const queuedMessages = useQueuedMessages(activeThreadKey ?? "");
7251 // Puts queued messages back into the composer, e.g. after Stop or a failed
7252 // send. Prompts join with blank lines; attachments and contexts are added.
7252 // The composer's model and modes, as a queued message keeps them for its send.
7253 const readComposerSendSettings = (
7254 sendCtx: ReturnType<ChatComposerHandle["getSendContext"]>,
7255 ): QueuedMessageSendSettings => ({
7256 modelSelection: sendCtx.selectedModelSelection,
7257 runtimeMode,
7258 interactionMode: sendCtx.interactionMode,
7259 promptEffort: resolvePromptInjectedEffort(
7260 getProviderModelCapabilities(
7261 sendCtx.selectedProviderModels,
7262 sendCtx.selectedModel,
7263 sendCtx.selectedProvider,
7264 ),
7265 sendCtx.selectedPromptEffort,
7266 ),
7267 });
7268 // Puts queued messages back into the composer after Stop or Cancel. Prompts
7269 // join with blank lines; attachments and contexts are added.
72537270 const restoreQueuedMessagesToComposer = (messages: ReadonlyArray<QueuedComposerMessage>) => {
7254 if (messages.length === 0) return;
7271 const [firstMessage] = messages;
7272 if (!firstMessage) return;
72557273 const prompts = [promptRef.current, ...messages.map((message) => message.prompt)]
72567274 .map((prompt) => prompt.trim())
72577275 .filter((prompt) => prompt.length > 0);
......@@ -7279,6 +7297,8 @@ export default function ChatView(props: ChatViewProps) {
72797297 if (restoredImages.length > 0) addComposerDraftImages(composerDraftTarget, restoredImages);
72807298 if (restoredFiles.length > 0) addComposerDraftFiles(composerDraftTarget, restoredFiles);
72817299 if (overflow.length > 0 && activeThreadKey) {
7300 // The overflow is the rest of the restored draft, so it follows the composer.
7301 const sendCtx = composerRef.current?.getSendContext();
72827302 useQueuedMessageStore.getState().enqueue(activeThreadKey, {
72837303 prompt: "",
72847304 images: overflow.filter((attachment) => attachment.type === "image"),
......@@ -7286,7 +7306,7 @@ export default function ChatView(props: ChatViewProps) {
72867306 terminalContexts: [],
72877307 previewAnnotations: [],
72887308 reviewComments: [],
7289 submissionIntent: "foreground",
7309 sendSettings: sendCtx ? readComposerSendSettings(sendCtx) : firstMessage.sendSettings,
72907310 queuedAfterToolActivityId: latestCompletedToolActivityId(threadActivities),
72917311 // Restoration is not a send. The user decides when the overflow goes.
72927312 holdUntilUserAction: true,
......@@ -7329,8 +7349,6 @@ export default function ChatView(props: ChatViewProps) {
73297349 annotation: PreviewAnnotationPayload;
73307350 image: ComposerImageAttachment | null;
73317351 },
7332 /** A queued message being sent now instead of the live composer draft. */
7333 queuedMessage?: QueuedComposerMessage,
73347352 ) => {
73357353 e?.preventDefault();
73367354 // Typed out in full rather than picked from the menu. Attachments or contexts
......@@ -7339,7 +7357,6 @@ export default function ChatView(props: ChatViewProps) {
73397357 usageLimitsOffered &&
73407358 usageLimitsKey !== null &&
73417359 !directAnnotation &&
7342 !queuedMessage &&
73437360 !composerHasNonPromptContent &&
73447361 isUsageLimitsCommand(promptRef.current)
73457362 ) {
......@@ -7401,9 +7418,7 @@ export default function ChatView(props: ChatViewProps) {
74017418 return;
74027419 }
74037420 if (activePendingProgress) {
7404 // A queued message waits until the question is answered; it must not
7405 // be submitted as the answer.
7406 if (directAnnotation || queuedMessage) {
7421 if (directAnnotation) {
74077422 notifyDirectAnnotationAttached();
74087423 return;
74097424 }
......@@ -7415,7 +7430,7 @@ export default function ChatView(props: ChatViewProps) {
74157430 notifyDirectAnnotationAttached();
74167431 return;
74177432 }
7418 const multipleModelSelections = queuedMessage ? null : sendCtx.multipleModelSelections;
7433 const multipleModelSelections = sendCtx.multipleModelSelections;
74197434 if (
74207435 multipleModelSelections !== null &&
74217436 serverConfig?.environment.capabilities.requiredWorktreeBootstrap !== true
......@@ -7446,8 +7461,6 @@ export default function ChatView(props: ChatViewProps) {
74467461 terminalContexts: composerTerminalContexts,
74477462 previewAnnotations: sendContextPreviewAnnotations,
74487463 reviewComments: composerReviewComments,
7449 } = queuedMessage ?? sendCtx;
7450 const {
74517464 selectedProvider: ctxSelectedProvider,
74527465 selectedModel: ctxSelectedModel,
74537466 selectedProviderModels: ctxSelectedProviderModels,
......@@ -7490,13 +7503,11 @@ export default function ChatView(props: ChatViewProps) {
74907503 : sendContextPreviewAnnotations;
74917504 // A direct "send annotation" writes the draft and sends in the same tick; the reference
74927505 // must be in the text now, not after the next render.
7493 const promptForSend = queuedMessage
7494 ? queuedMessage.prompt
7495 : directAnnotation
7496 ? ensureInlineContextReferences(promptRef.current, [
7497 previewAnnotationContextReference(directAnnotation.annotation),
7498 ])
7499 : promptRef.current;
7506 const promptForSend = directAnnotation
7507 ? ensureInlineContextReferences(promptRef.current, [
7508 previewAnnotationContextReference(directAnnotation.annotation),
7509 ])
7510 : promptRef.current;
75007511 const {
75017512 trimmedPrompt: trimmed,
75027513 sendableTerminalContexts: sendableComposerTerminalContexts,
......@@ -7517,7 +7528,7 @@ export default function ChatView(props: ChatViewProps) {
75177528 composerReviewComments.length === 0
75187529 ? parseCodexFeedbackCommand(trimmed)
75197530 : null;
7520 if (feedbackCommand && !queuedMessage && multipleModelSelections === null) {
7531 if (feedbackCommand && multipleModelSelections === null) {
75217532 if (!isServerThread || activeThread.session === null) {
75227533 toastManager.add(
75237534 stackedThreadToast({
......@@ -7568,7 +7579,6 @@ export default function ChatView(props: ChatViewProps) {
75687579 }
75697580 if (
75707581 !directAnnotation &&
7571 !queuedMessage &&
75727582 sendInteractionModeEnabled &&
75737583 showPlanFollowUpPrompt &&
75747584 activeProposedPlan &&
......@@ -7640,7 +7650,7 @@ export default function ChatView(props: ChatViewProps) {
76407650 composerReviewComments.length === 0
76417651 ? parseStandaloneComposerSlashCommand(trimmed)
76427652 : null;
7643 if (standaloneSlashCommand && !queuedMessage && multipleModelSelections === null) {
7653 if (standaloneSlashCommand && multipleModelSelections === null) {
76447654 handleInteractionModeChange(standaloneSlashCommand);
76457655 promptRef.current = "";
76467656 clearComposerDraftContent(composerDraftTarget);
......@@ -7661,12 +7671,6 @@ export default function ChatView(props: ChatViewProps) {
76617671 }),
76627672 );
76637673 }
7664 // A queued message whose only content expired would retry on every
7665 // boundary and block the rest of the queue. Nothing sendable is left
7666 // in it, so drop it and let the queue move on.
7667 if (queuedMessage && activeThreadKey) {
7668 useQueuedMessageStore.getState().remove(activeThreadKey, queuedMessage.id);
7669 }
76707674 return;
76717675 }
76727676 if (!activeProject) {
......@@ -7679,14 +7683,26 @@ export default function ChatView(props: ChatViewProps) {
76797683 );
76807684 return;
76817685 }
7686 // A queued message that will still leave on its own goes first, so a new
7687 // send lines up behind it instead of overtaking it.
7688 const queueStillSending =
7689 activeThreadKey !== null &&
7690 (useQueuedMessageStore.getState().queuesByThreadKey[activeThreadKey] ?? []).some(
7691 (message) => message.sending !== undefined || !message.holdUntilUserAction,
7692 );
76827693 if (
7683 !queuedMessage &&
76847694 !directAnnotation &&
7685 phase === "running" &&
76867695 activeThreadKey &&
7687 (settings.followUpBehavior === "queue") !== (submissionIntent === "alternate")
7696 (queueStillSending ||
7697 (phase === "running" &&
7698 (settings.followUpBehavior === "queue") !== (submissionIntent === "alternate")))
76887699 ) {
7689 if (composerRef.current?.validateProviderInput(promptForSend) === false) {
7700 const sendSettings = readComposerSendSettings(sendCtx);
7701 if (
7702 composerRef.current?.validateProviderInput(
7703 applyClaudePromptEffortPrefix(promptForSend, sendSettings.promptEffort),
7704 ) === false
7705 ) {
76907706 return;
76917707 }
76927708 useQueuedMessageStore.getState().enqueue(activeThreadKey, {
......@@ -7696,7 +7712,7 @@ export default function ChatView(props: ChatViewProps) {
76967712 terminalContexts: [...composerTerminalContexts],
76977713 previewAnnotations: [...composerPreviewAnnotations],
76987714 reviewComments: [...composerReviewComments],
7699 submissionIntent,
7715 sendSettings,
77007716 queuedAfterToolActivityId: latestCompletedToolActivityId(threadActivities),
77017717 createdAt: new Date().toISOString(),
77027718 });
......@@ -7766,11 +7782,6 @@ export default function ChatView(props: ChatViewProps) {
77667782 text: messageTextForSend || ATTACHMENT_ONLY_BOOTSTRAP_PROMPT,
77677783 });
77687784 if (composerRef.current?.validateProviderInput(outgoingMessageText) === false) {
7769 // A queued message that no longer fits is held at the head for the
7770 // user to edit via Cancel, instead of failing on every boundary.
7771 if (queuedMessage && activeThreadKey) {
7772 useQueuedMessageStore.getState().holdAtFront(activeThreadKey, queuedMessage);
7773 }
77747785 return;
77757786 }
77767787
......@@ -7837,41 +7848,10 @@ export default function ChatView(props: ChatViewProps) {
78377848
78387849 sendInFlightRef.current = true;
78397850 const sendGeneration = ++composerSendGenerationRef.current;
7840 // Every early return above leaves a queued message in the queue for a
7841 // later retry. From here on a failure hands it back to the composer.
7842 if (queuedMessage) {
7843 const taken = activeThreadKey
7844 ? useQueuedMessageStore
7845 .getState()
7846 .take(
7847 activeThreadKey,
7848 queuedMessage.id,
7849 latestCompletedToolActivityId(threadActivities),
7850 )
7851 : null;
7852 if (!taken) {
7853 sendInFlightRef.current = false;
7854 return;
7855 }
7856 }
7857 // Stop drains the queue. A queued send whose upload was still running at
7858 // that moment must not start a turn afterwards; it checks this before
7859 // dispatch and hands the message back to the composer instead.
7860 const drainGenerationAtTake = useQueuedMessageStore.getState().drainGeneration;
7861 // A queued send that fails goes back to the head of the queue, held. The
7862 // messages behind it keep their order and wait; the composer is not
7863 // touched, which also keeps a failure after navigation off the new
7864 // thread's draft. The user retries with Send now or edits with Cancel.
7865 const abortQueuedReplay = () => {
7866 if (queuedMessage && activeThreadKey) {
7867 useQueuedMessageStore.getState().holdAtFront(activeThreadKey, queuedMessage);
7868 }
7869 };
78707851 const attachmentCapabilitiesBeforeUpload = readLiveAttachmentCapabilities();
78717852 if (attachmentCapabilitiesBeforeUpload.fileBlockReason !== null) {
78727853 sendInFlightRef.current = false;
78737854 setThreadError(threadIdForSend, attachmentCapabilitiesBeforeUpload.fileBlockReason);
7874 abortQueuedReplay();
78757855 return;
78767856 }
78777857 const turnUsesAttachmentUploads =
......@@ -7891,26 +7871,15 @@ export default function ChatView(props: ChatViewProps) {
78917871 if (attachmentCapabilitiesAfterUpload.fileBlockReason !== null) {
78927872 sendInFlightRef.current = false;
78937873 setThreadError(threadIdForSend, attachmentCapabilitiesAfterUpload.fileBlockReason);
7894 abortQueuedReplay();
78957874 return;
78967875 }
78977876 if (getUploadedAttachments({ environmentId, images: composerAttachmentsSnapshot }) === null) {
78987877 sendInFlightRef.current = false;
78997878 setThreadError(threadIdForSend, "Retry or remove failed uploads before sending.");
7900 abortQueuedReplay();
79017879 return;
79027880 }
79037881 }
79047882
7905 if (
7906 queuedMessage &&
7907 useQueuedMessageStore.getState().drainGeneration !== drainGenerationAtTake
7908 ) {
7909 sendInFlightRef.current = false;
7910 restoreQueuedMessagesToComposer([queuedMessage]);
7911 return;
7912 }
7913
79147883 const resolvedSubmissionIntent =
79157884 (multipleModelSelections !== null || submissionIntent === "background") && isLocalDraftThread
79167885 ? "background"
......@@ -7951,7 +7920,6 @@ export default function ChatView(props: ChatViewProps) {
79517920 setDockedDraftHeroThreadKey((currentThreadKey) =>
79527921 currentThreadKey === activeThreadKey ? null : currentThreadKey,
79537922 );
7954 abortQueuedReplay();
79557923 return;
79567924 }
79577925 beginLocalDispatch({
......@@ -8313,11 +8281,9 @@ export default function ChatView(props: ChatViewProps) {
83138281 }),
83148282 );
83158283 }
8316 if (!queuedMessage) {
8317 promptRef.current = "";
8318 clearComposerDraftContent(composerDraftTarget);
8319 composerRef.current?.resetCursorState();
8320 }
8284 promptRef.current = "";
8285 clearComposerDraftContent(composerDraftTarget);
8286 composerRef.current?.resetCursorState();
83218287
83228288 let firstComposerImageName: string | null = null;
83238289 if (composerImagesSnapshot.length > 0) {
......@@ -8552,24 +8518,7 @@ export default function ChatView(props: ChatViewProps) {
85528518 );
85538519 clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend));
85548520 }
8555 if (queuedMessage) {
8556 setOptimisticUserMessages((existing) => {
8557 const removed = existing.filter((message) => message.id === messageIdForSend);
8558 for (const message of removed) {
8559 revokeUserMessagePreviewUrls(message);
8560 }
8561 const next = existing.filter((message) => message.id !== messageIdForSend);
8562 return next.length === existing.length ? existing : next;
8563 });
8564 // The optimistic row's preview URLs were just revoked, so the images
8565 // need fresh ones before the row can show them again.
8566 if (activeThreadKey) {
8567 useQueuedMessageStore.getState().holdAtFront(activeThreadKey, {
8568 ...queuedMessage,
8569 images: queuedMessage.images.map(cloneComposerImageForRetry),
8570 });
8571 }
8572 } else if (
8521 if (
85738522 backgroundDraftOpened
85748523 ? !composerDraftHasUserContent(
85758524 useComposerDraftStore.getState().getComposerDraft(composerDraftTarget),
......@@ -8659,46 +8608,12 @@ export default function ChatView(props: ChatViewProps) {
86598608 }
86608609 };
86618610
8662 // Sends the oldest queued message once it is due: a tool call finished
8663 // after it was queued, or the turn ended. Only one leaves per boundary; the
8664 // take inside onSend re-anchors the rest.
8665 const sendQueuedMessage = useEffectEvent((message: QueuedComposerMessage) => {
8666 void onSend(undefined, message.submissionIntent, undefined, message);
8667 });
8668 const nextQueuedMessage = queuedMessages[0] ?? null;
8669 const latestToolActivityId = useMemo(
8670 () => (nextQueuedMessage ? latestCompletedToolActivityId(threadActivities) : null),
8671 [nextQueuedMessage, threadActivities],
8672 );
8673 // Approvals and questions block the agent; a steer landing on top of them
8674 // would answer nothing and confuse the turn, so the queue holds until the
8675 // user resolves them.
8611 // Queued messages go out from QueuedMessageSender, which also covers
8612 // threads that are not on screen. Send now uses the same path but skips the
8613 // wait for a boundary. Approvals and questions still hold it: a steer on
8614 // top of them would answer nothing and confuse the turn.
86768615 const queueBlockedByPendingRequest =
86778616 activePendingApproval !== null || pendingUserInputs.length > 0;
8678 // onSend bails early on transient gates (environment offline, settings not
8679 // hydrated, checkpoint rewinding, messages loading, machine not chosen) and
8680 // leaves the message queued. Re-run when any of them clear so a due message
8681 // does not wait for an unrelated phase change.
8682 const queueSendGate =
8683 activeEnvironmentUnavailable ||
8684 !clientSettingsHydrated ||
8685 isRevertingCheckpoint ||
8686 threadDetailLoading ||
8687 needsLoadBalancing ||
8688 activeProviderStatus === null;
8689 useEffect(() => {
8690 if (!nextQueuedMessage || isSendBusy || queueBlockedByPendingRequest || queueSendGate) return;
8691 if (sendInFlightRef.current) return;
8692 if (!isQueuedMessageDue({ message: nextQueuedMessage, phase, latestToolActivityId })) return;
8693 sendQueuedMessage(nextQueuedMessage);
8694 }, [
8695 isSendBusy,
8696 latestToolActivityId,
8697 nextQueuedMessage,
8698 phase,
8699 queueBlockedByPendingRequest,
8700 queueSendGate,
8701 ]);
87028617
87038618 // The row handlers are read from refs at call-time so their identity stays
87048619 // stable and does not bust TimelineRowCtx on every ChatView render.
......@@ -8708,9 +8623,8 @@ export default function ChatView(props: ChatViewProps) {
87088623 });
87098624 queuedMessageActionsRef.current = {
87108625 steer: (id) => {
8711 const message = queuedMessages.find((entry) => entry.id === id);
8712 if (!message || sendInFlightRef.current || queueBlockedByPendingRequest) return;
8713 void onSend(undefined, message.submissionIntent, undefined, message);
8626 if (!activeThreadRef || queueBlockedByPendingRequest) return;
8627 void sendQueuedMessage(activeThreadRef, id);
87148628 },
87158629 remove: (id) => {
87168630 if (!activeThreadKey) return;
apps/web/src/components/QueuedMessageSender.test.tsx created+231
......@@ -0,0 +1,231 @@
1import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment";
2import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
3import * as Cause from "effect/Cause";
4import { act, createElement } from "react";
5import { create, type ReactTestRenderer } from "react-test-renderer";
6import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";
7
8import { useQueuedMessageStore, type QueuedComposerMessage } from "../queuedMessageStore";
9import { sendQueuedMessage } from "./chat/sendQueuedMessage";
10import { QueuedMessageSender } from "./QueuedMessageSender";
11
12const io = vi.hoisted(() => ({
13 run: vi.fn(),
14 upload: vi.fn(),
15 toast: vi.fn(),
16 thread: null as unknown,
17 shell: { runtimeMode: "full-access", interactionMode: "default" } as Record<string, unknown>,
18}));
19const config = {
20 environment: { capabilities: { attachmentUploads: true, inlineMessageContext: true } },
21};
22vi.mock("@t3tools/client-runtime/state/runtime", async (load) => ({
23 ...(await load<typeof import("@t3tools/client-runtime/state/runtime")>()),
24 runAtomCommand: (...args: unknown[]) => io.run(...args),
25}));
26vi.mock("../rpc/atomRegistry", () => ({
27 appAtomRegistry: { get: () => new Map([["env-a", config]]) },
28}));
29vi.mock("../state/server", () => ({ environmentServerConfigsAtom: {} }));
30vi.mock("../state/threads", () => ({
31 threadEnvironment: {
32 updateMetadata: "metadata",
33 setRuntimeMode: "runtime",
34 setInteractionMode: "interaction",
35 startTurn: "start",
36 },
37}));
38vi.mock("../state/environments", () => ({
39 useEnvironment: () => ({ connection: { phase: "connected" } }),
40}));
41vi.mock("../state/entities", () => ({
42 useThread: () => io.thread,
43 useThreadStatus: () => "live",
44 useServerConfigs: () => new Map([["env-a", config]]),
45 readThreadShell: () => io.shell,
46 readThread: () => io.thread,
47}));
48vi.mock("./ui/toast", () => ({ toastManager: { add: (...args: unknown[]) => io.toast(...args) } }));
49vi.mock("../lib/attachmentUploadQueue", () => ({
50 startAttachmentUpload: vi.fn(),
51 awaitAttachmentUploads: (...args: unknown[]) => io.upload(...args),
52 getUploadedAttachments: () => [
53 { type: "image", id: "uploaded", name: "a.png", mimeType: "image/png", sizeBytes: 4 },
54 ],
55 releaseDraftAttachments: vi.fn(),
56}));
57
58const threadRef = scopeThreadRef(EnvironmentId.make("env-a"), ThreadId.make("thread-a"));
59const threadKey = scopedThreadKey(threadRef);
60const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" };
61
62function enqueue(overrides: Partial<QueuedComposerMessage> = {}) {
63 return useQueuedMessageStore.getState().enqueue(threadKey, {
64 prompt: "follow up",
65 images: [],
66 files: [],
67 terminalContexts: [],
68 previewAnnotations: [],
69 reviewComments: [],
70 sendSettings: {
71 modelSelection,
72 runtimeMode: "full-access",
73 interactionMode: "default",
74 promptEffort: null,
75 },
76 queuedAfterToolActivityId: null,
77 createdAt: "2026-09-25T00:00:00Z",
78 ...overrides,
79 });
80}
81
82const commandsRun = () => io.run.mock.calls.map((call) => call[1]);
83const queue = () => useQueuedMessageStore.getState().queuesByThreadKey[threadKey];
84
85beforeEach(() => {
86 vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true);
87 useQueuedMessageStore.setState({ queuesByThreadKey: {}, lastDispatchByThreadKey: {} });
88 io.thread = null;
89 io.run.mockReset().mockResolvedValue({ _tag: "Success", value: undefined });
90 io.upload.mockReset().mockResolvedValue(undefined);
91 io.toast.mockReset();
92 io.shell = {
93 modelSelection,
94 branch: null,
95 runtimeMode: "full-access",
96 interactionMode: "default",
97 };
98});
99
100afterEach(() => {
101 vi.unstubAllGlobals();
102});
103
104describe("QueuedMessageSender", () => {
105 const thread = (
106 status: string,
107 { toolActivityIds = [] as string[], userMessageIds = [] as string[] } = {},
108 ) => ({
109 session: { status, activeTurnId: null, updatedAt: status },
110 activities: toolActivityIds.map((id, index) => ({
111 id,
112 kind: "tool.completed",
113 sequence: index,
114 createdAt: "2026-09-25T00:00:01Z",
115 })),
116 messages: userMessageIds.map((id) => ({ id, role: "user" })),
117 latestTurn: null,
118 });
119 let root: ReactTestRenderer | null = null;
120 const render = () =>
121 act(() => {
122 if (root) root.update(createElement(QueuedMessageSender));
123 else root = create(createElement(QueuedMessageSender));
124 });
125 afterEach(async () => {
126 await act(() => root?.unmount());
127 root = null;
128 });
129
130 it("sends a queued message when the turn ends, with no chat view open", async () => {
131 enqueue();
132 io.thread = thread("running");
133 await render();
134 expect(commandsRun()).toEqual([]);
135
136 io.thread = thread("ready");
137 await render();
138
139 expect(commandsRun()).toEqual(["start"]);
140 expect(io.run.mock.calls[0]?.[2]).toMatchObject({
141 environmentId: "env-a",
142 input: { threadId: "thread-a", message: { text: "follow up" }, modelSelection },
143 });
144 expect(queue()).toBeUndefined();
145 });
146
147 it("holds the next message until the server picks up the one before it", async () => {
148 enqueue({ prompt: "first" });
149 enqueue({ prompt: "second" });
150 io.thread = thread("ready");
151 await render();
152 await render();
153 expect(commandsRun()).toEqual(["start"]);
154
155 // The first message started a turn; the second waits for its next tool call.
156 io.thread = thread("running", { userMessageIds: ["first"] });
157 await render();
158 expect(commandsRun()).toEqual(["start"]);
159 io.thread = thread("running", { userMessageIds: ["first"], toolActivityIds: ["tool-1"] });
160 await render();
161 expect(commandsRun()).toEqual(["start", "start"]);
162 });
163
164 it("moves on to the next message after a failed one is cancelled", async () => {
165 io.run.mockResolvedValueOnce({ _tag: "Failure", cause: Cause.fail(new Error("offline")) });
166 const first = enqueue({ prompt: "first" });
167 enqueue({ prompt: "second" });
168 io.thread = thread("ready");
169 await render();
170 expect(queue()?.[0]).toMatchObject({ prompt: "first", holdUntilUserAction: true });
171
172 await act(() => {
173 useQueuedMessageStore.getState().remove(threadKey, first.id);
174 });
175 await render();
176
177 expect(commandsRun()).toEqual(["start", "start"]);
178 expect(io.run.mock.calls[1]?.[2]).toMatchObject({ input: { message: { text: "second" } } });
179 });
180});
181
182describe("sendQueuedMessage", () => {
183 it("saves a mode changed before queueing, then starts the turn", async () => {
184 io.shell = { ...io.shell, runtimeMode: "approval-required" };
185 const message = enqueue();
186
187 await sendQueuedMessage(threadRef, message.id);
188
189 expect(commandsRun()).toEqual(["runtime", "start"]);
190 expect(io.run.mock.calls[1]?.[2]).toMatchObject({ input: { runtimeMode: "full-access" } });
191 expect(queue()).toBeUndefined();
192 });
193
194 it("gives a message back to Stop while its upload runs, without starting a turn", async () => {
195 let finishUpload!: () => void;
196 io.upload.mockReturnValue(new Promise<void>((resolve) => (finishUpload = resolve)));
197 const image = {
198 type: "image" as const,
199 id: "image-1",
200 name: "a.png",
201 mimeType: "image/png",
202 sizeBytes: 4,
203 previewUrl: "data:image/png;base64,AAAA",
204 file: new File(["AAAA"], "a.png", { type: "image/png" }),
205 };
206 const message = enqueue({ images: [image] });
207
208 const sending = sendQueuedMessage(threadRef, message.id);
209 expect(useQueuedMessageStore.getState().drain(threadKey)).toHaveLength(1);
210 finishUpload();
211 await sending;
212
213 expect(commandsRun()).toEqual([]);
214 expect(io.toast).not.toHaveBeenCalled();
215 expect(queue()).toBeUndefined();
216 });
217
218 it("holds a message at the head when the turn start fails", async () => {
219 io.run.mockResolvedValue({ _tag: "Failure", cause: Cause.fail(new Error("offline")) });
220 enqueue({ prompt: "first" });
221 const second = enqueue({ prompt: "second" });
222
223 await sendQueuedMessage(threadRef, second.id);
224
225 expect(queue()?.map((entry) => [entry.prompt, entry.holdUntilUserAction])).toEqual([
226 ["second", true],
227 ["first", undefined],
228 ]);
229 expect(io.toast).toHaveBeenCalledWith(expect.objectContaining({ description: "offline" }));
230 });
231});
apps/web/src/components/QueuedMessageSender.tsx created+100
......@@ -0,0 +1,100 @@
1import { parseScopedThreadKey } from "@t3tools/client-runtime/environment";
2import { derivePendingRequests } from "@t3tools/client-runtime/pending-requests";
3import { useEffect, useMemo } from "react";
4import { useShallow } from "zustand/react/shallow";
5
6import { useComposerDraftStore } from "../composerDraftStore";
7import {
8 isQueuedMessageDue,
9 latestCompletedToolActivityId,
10 useQueuedMessageStore,
11 useQueuedMessages,
12} from "../queuedMessageStore";
13import { derivePhase } from "../session-logic";
14import { useServerConfigs, useThread, useThreadStatus } from "../state/entities";
15import { useEnvironment } from "../state/environments";
16import { hasServerAcknowledgedLocalDispatch, latestTurnStartFailureId } from "./ChatView.logic";
17import { sendQueuedMessage } from "./chat/sendQueuedMessage";
18
19/**
20 * Sends queued messages when they are due, for every thread with a queue,
21 * whether or not the thread is on screen. Mounted once at the root.
22 */
23export function QueuedMessageSender() {
24 const threadKeys = useQueuedMessageStore(
25 useShallow((state) => Object.keys(state.queuesByThreadKey)),
26 );
27 return threadKeys.map((threadKey) => <ThreadQueueSender key={threadKey} threadKey={threadKey} />);
28}
29
30/**
31 * Watches one thread while it has queued messages. Reading the thread keeps
32 * its detail subscribed, so tool boundaries and the end of the turn are
33 * visible while the user is elsewhere.
34 */
35function ThreadQueueSender({ threadKey }: { threadKey: string }) {
36 const threadRef = useMemo(() => parseScopedThreadKey(threadKey), [threadKey]);
37 const thread = useThread(threadRef);
38 const threadStatus = useThreadStatus(threadRef);
39 const environmentId = threadRef?.environmentId ?? null;
40 const environment = useEnvironment(environmentId);
41 const serverConfigs = useServerConfigs();
42 const serverConfigLoaded = environmentId !== null && serverConfigs.has(environmentId);
43 const rewinding = useComposerDraftStore((store) => store.rewindingThreadKeys.has(threadKey));
44 const queue = useQueuedMessages(threadKey);
45 const next = queue[0];
46 const sending = queue.some((message) => message.sending);
47 const activities = thread?.activities;
48 const latestToolActivityId = useMemo(
49 () => latestCompletedToolActivityId(activities ?? []),
50 [activities],
51 );
52 const pendingRequests = useMemo(() => derivePendingRequests(activities ?? []), [activities]);
53 const phase = derivePhase(thread?.session ?? null);
54
55 // A send that starts a new turn leaves the thread idle until the server
56 // picks it up. Hold the next message until then, as the composer does for
57 // its own sends.
58 const lastDispatch = useQueuedMessageStore(
59 (state) => state.lastDispatchByThreadKey[threadKey]?.thread ?? null,
60 );
61 const latestUserMessageId = thread?.messages.findLast((m) => m.role === "user")?.id ?? null;
62 const waitingForServer =
63 lastDispatch !== null &&
64 !hasServerAcknowledgedLocalDispatch({
65 localDispatch: lastDispatch,
66 phase,
67 latestTurn: thread?.latestTurn ?? null,
68 latestUserMessageId,
69 session: thread?.session ?? null,
70 hasPendingApproval: pendingRequests.approvals.length > 0,
71 hasPendingUserInput: pendingRequests.userInputs.length > 0,
72 latestTurnStartFailureId: latestTurnStartFailureId(thread ?? undefined, latestUserMessageId),
73 threadError: null,
74 });
75
76 // Approvals and questions block the agent; a steer landing on top of them
77 // would answer nothing and confuse the turn, so the queue holds until the
78 // user resolves them.
79 const blocked =
80 threadRef === null ||
81 thread === null ||
82 threadStatus !== "live" ||
83 (environment !== null && environment.connection.phase !== "connected") ||
84 !serverConfigLoaded ||
85 rewinding ||
86 sending ||
87 waitingForServer ||
88 pendingRequests.approvals.length > 0 ||
89 pendingRequests.userInputs.length > 0;
90 const due =
91 next !== undefined &&
92 !blocked &&
93 isQueuedMessageDue({ message: next, phase, latestToolActivityId });
94 const nextId = next?.id;
95 useEffect(() => {
96 if (!due || !threadRef || nextId === undefined) return;
97 void sendQueuedMessage(threadRef, nextId);
98 }, [due, nextId, threadRef]);
99 return null;
100}
apps/web/src/components/chat/MessagesTimeline.logic.test.ts+6-1
......@@ -1100,7 +1100,12 @@ describe("deriveMessagesTimelineRows", () => {
11001100 terminalContexts: [],
11011101 previewAnnotations: [],
11021102 reviewComments: [],
1103 submissionIntent: "foreground" as const,
1103 sendSettings: {
1104 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
1105 runtimeMode: "full-access" as const,
1106 interactionMode: "default" as const,
1107 promptEffort: null,
1108 },
11041109 queuedAfterToolActivityId: null,
11051110 createdAt: "2026-01-01T00:00:01Z",
11061111 });
apps/web/src/components/chat/MessagesTimeline.tsx+11-8
......@@ -1788,11 +1788,14 @@ function QueuedMessageTimelineRow({
17881788 queuedMessage.previewAnnotations.length +
17891789 queuedMessage.reviewComments.length;
17901790 const text = queuedMessage.prompt.trim();
1791 const statusLabel = queuedMessage.holdUntilUserAction
1792 ? "Waits for Send now"
1793 : row.isNext
1794 ? "Sends after the next tool call or when the turn ends"
1795 : "Sends after the messages above it";
1791 const sending = queuedMessage.sending !== undefined;
1792 const statusLabel = sending
1793 ? "Sending to the agent"
1794 : queuedMessage.holdUntilUserAction
1795 ? "Waits for Send now"
1796 : row.isNext
1797 ? "Sends after the next tool call or when the turn ends"
1798 : "Sends after the messages above it";
17961799 return (
17971800 <div className="flex flex-col items-end" data-queued-message-id={queuedMessage.id}>
17981801 <div className="max-w-[80%] rounded-2xl border border-dashed border-border p-3 text-message-foreground/80">
......@@ -1820,14 +1823,14 @@ function QueuedMessageTimelineRow({
18201823 <Tooltip>
18211824 <TooltipTrigger
18221825 render={<span className="inline-flex h-6 items-center gap-1" />}
1823 aria-label={`Queued. ${statusLabel}.`}
1826 aria-label={`${sending ? "Sending" : "Queued"}. ${statusLabel}.`}
18241827 >
18251828 <ClockIcon className="size-3.5" aria-hidden />
1826 Queued
1829 {sending ? "Sending" : "Queued"}
18271830 </TooltipTrigger>
18281831 <TooltipPopup side="bottom">{statusLabel}</TooltipPopup>
18291832 </Tooltip>
1830 <div className="ml-auto flex items-center gap-0.5">
1833 <div className={cn("ml-auto flex items-center gap-0.5", sending && "invisible")}>
18311834 <Tooltip>
18321835 <TooltipTrigger
18331836 render={
apps/web/src/components/chat/sendQueuedMessage.ts created+211
......@@ -0,0 +1,211 @@
1import { scopedThreadKey } from "@t3tools/client-runtime/environment";
2import {
3 runAtomCommand,
4 squashAtomCommandFailure,
5 type AtomCommand,
6} from "@t3tools/client-runtime/state/runtime";
7import type { ScopedThreadRef } from "@t3tools/contracts";
8import { serializeLegacyContextMessage } from "@t3tools/shared/composerContextLegacySend";
9import { applyClaudePromptEffortPrefix } from "@t3tools/shared/model";
10
11import { buildMessageContext, terminalContextReference } from "../../lib/composerContextRecords";
12import { removeInlineContextReference } from "../../lib/composerContextReferences";
13import {
14 awaitAttachmentUploads,
15 getUploadedAttachments,
16 releaseDraftAttachments,
17 startAttachmentUpload,
18} from "../../lib/attachmentUploadQueue";
19import { newMessageId } from "../../lib/utils";
20import { latestCompletedToolActivityId, useQueuedMessageStore } from "../../queuedMessageStore";
21import { appAtomRegistry } from "../../rpc/atomRegistry";
22import { readThread, readThreadShell } from "../../state/entities";
23import { environmentServerConfigsAtom } from "../../state/server";
24import { threadEnvironment } from "../../state/threads";
25import {
26 createLocalDispatchSnapshot,
27 deriveComposerSendState,
28 readFileAsDataUrl,
29 resolveThreadMetadataUpdateForNextTurn,
30 revokeBlobPreviewUrl,
31} from "../ChatView.logic";
32import { toastManager } from "../ui/toast";
33import { fileAttachmentCapabilityBlockReason } from "./composerAttachmentFiles";
34import { ATTACHMENT_ONLY_BOOTSTRAP_PROMPT } from "./composerPromptHistory";
35
36async function run<W, A, E>(command: AtomCommand<W, A, E>, input: W): Promise<A> {
37 const result = await runAtomCommand(appAtomRegistry, command, input, { reportFailure: false });
38 if (result._tag === "Failure") throw squashAtomCommandFailure(result);
39 return result.value;
40}
41
42/**
43 * Sends one queued message as a turn on its thread. `QueuedMessageSender`
44 * calls it when the head message is due, and Send now calls it directly. It
45 * reads nothing from the composer, so it works for threads that are not on
46 * screen. A failed send stays at the head of the queue, held for Send now.
47 */
48export async function sendQueuedMessage(
49 threadRef: ScopedThreadRef,
50 messageId: string,
51): Promise<void> {
52 const { environmentId, threadId } = threadRef;
53 const threadKey = scopedThreadKey(threadRef);
54 const queue = useQueuedMessageStore.getState();
55 const message = queue.beginSend(
56 threadKey,
57 messageId,
58 latestCompletedToolActivityId(readThread(threadRef)?.activities ?? []),
59 );
60 if (!message) return;
61 const { sendSettings } = message;
62 const attachments = [...message.images, ...message.files];
63 const readConfig = () => appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId);
64 const assertFilesAllowed = () => {
65 const config = readConfig();
66 const reason = fileAttachmentCapabilityBlockReason({
67 files: message.files,
68 attachmentUploadsCapabilityKnown: config !== undefined,
69 supportsAttachmentUploads: config?.environment.capabilities.attachmentUploads === true,
70 maxFileAttachmentBytes:
71 config?.environment.capabilities.fileAttachments?.maxUploadBytes ?? null,
72 });
73 if (reason !== null) throw new Error(reason);
74 };
75 try {
76 const { sendableTerminalContexts, hasSendableContent } = deriveComposerSendState({
77 prompt: message.prompt,
78 imageCount: attachments.length,
79 terminalContexts: message.terminalContexts,
80 elementContextCount: message.previewAnnotations.length + message.reviewComments.length,
81 });
82 // Only expired terminal context was left. Retrying would block the queue
83 // on every boundary, so drop it and let the queue move on.
84 if (!hasSendableContent) {
85 queue.finishSend(threadKey, message.id);
86 return;
87 }
88 // Expired terminal excerpts are not sent; their chips leave the text with them.
89 const prompt = message.terminalContexts
90 .filter((context) => !sendableTerminalContexts.includes(context))
91 .reduce(
92 (text, context) =>
93 removeInlineContextReference(text, terminalContextReference(context).contextId).prompt,
94 message.prompt,
95 )
96 .trim();
97 const text = applyClaudePromptEffortPrefix(
98 prompt || ATTACHMENT_ONLY_BOOTSTRAP_PROMPT,
99 sendSettings.promptEffort,
100 );
101
102 assertFilesAllowed();
103 const useUploads = readConfig()?.environment.capabilities.attachmentUploads === true;
104 if (useUploads && attachments.length > 0) {
105 for (const attachment of attachments) {
106 startAttachmentUpload({ environmentId, image: attachment, draftTarget: threadRef });
107 }
108 await awaitAttachmentUploads(attachments.map((attachment) => attachment.id));
109 }
110 const wireAttachments = await Promise.all(
111 attachments.map(async (attachment) => {
112 if (useUploads) {
113 const uploaded = getUploadedAttachments({ environmentId, images: [attachment] })?.[0];
114 if (!uploaded) throw new Error(`Attachment '${attachment.name}' did not upload.`);
115 return uploaded;
116 }
117 if (attachment.type !== "image") {
118 throw new Error("This server does not support file attachments.");
119 }
120 return {
121 type: "image" as const,
122 id: attachment.id,
123 name: attachment.name,
124 mimeType: attachment.mimeType,
125 sizeBytes: attachment.sizeBytes,
126 dataUrl: await readFileAsDataUrl(attachment.file),
127 ...(attachment.source ? { source: attachment.source } : {}),
128 };
129 }),
130 );
131 assertFilesAllowed();
132
133 // The server starts the turn with the thread's stored modes, so a change
134 // made in the composer before queueing is saved first.
135 const createdAt = new Date().toISOString();
136 const shell = readThreadShell(threadRef);
137 const metadataUpdate = shell
138 ? resolveThreadMetadataUpdateForNextTurn({
139 currentModelSelection: shell.modelSelection,
140 nextModelSelection: sendSettings.modelSelection,
141 currentBranch: shell.branch,
142 })
143 : null;
144 if (metadataUpdate) {
145 await run(threadEnvironment.updateMetadata, {
146 environmentId,
147 input: { threadId, ...metadataUpdate },
148 });
149 }
150 if (shell && shell.runtimeMode !== sendSettings.runtimeMode) {
151 await run(threadEnvironment.setRuntimeMode, {
152 environmentId,
153 input: { threadId, runtimeMode: sendSettings.runtimeMode, createdAt },
154 });
155 }
156 if (shell && shell.interactionMode !== sendSettings.interactionMode) {
157 await run(threadEnvironment.setInteractionMode, {
158 environmentId,
159 input: { threadId, interactionMode: sendSettings.interactionMode, createdAt },
160 });
161 }
162
163 // Stop hands a preparing message back to the composer. Past this point
164 // the send can no longer be taken back.
165 const thread = readThread(threadRef) ?? undefined;
166 if (!queue.markDispatching(threadKey, message.id, createLocalDispatchSnapshot(thread))) return;
167 const context = buildMessageContext({
168 terminalContexts: sendableTerminalContexts,
169 reviewComments: message.reviewComments,
170 previewAnnotations: message.previewAnnotations,
171 attachments: attachments.map((attachment, index) => ({
172 attachment,
173 attachmentId: wireAttachments[index]?.id ?? attachment.id,
174 })),
175 });
176 // Servers from before inline context drop the records, so their turns
177 // carry the payload in the text instead.
178 const inlineContext = readConfig()?.environment.capabilities.inlineMessageContext === true;
179 await run(threadEnvironment.startTurn, {
180 environmentId,
181 input: {
182 threadId,
183 message: {
184 messageId: newMessageId(),
185 role: "user",
186 text:
187 context !== undefined && !inlineContext
188 ? serializeLegacyContextMessage({ text, records: context.records })
189 : text,
190 attachments: wireAttachments,
191 ...(context !== undefined && inlineContext ? { context } : {}),
192 },
193 modelSelection: sendSettings.modelSelection,
194 runtimeMode: sendSettings.runtimeMode,
195 interactionMode: sendSettings.interactionMode,
196 createdAt,
197 },
198 });
199 queue.finishSend(threadKey, message.id);
200 if (useUploads) releaseDraftAttachments(attachments);
201 for (const image of message.images) revokeBlobPreviewUrl(image.previewUrl);
202 } catch (error) {
203 if (!queue.failSend(threadKey, message.id)) return;
204 const title = readThreadShell(threadRef)?.title;
205 toastManager.add({
206 type: "error",
207 title: title ? `Queued message not sent in "${title}"` : "Queued message not sent",
208 description: error instanceof Error ? error.message : "Use Send now to try again.",
209 });
210 }
211}
apps/web/src/queuedMessageStore.test.ts+96-45
......@@ -1,5 +1,7 @@
1import { ProviderInstanceId } from "@t3tools/contracts";
12import { beforeEach, describe, expect, it } from "vite-plus/test";
23
4import { createLocalDispatchSnapshot } from "./components/ChatView.logic";
35import {
46 isQueuedMessageDue,
57 latestCompletedToolActivityId,
......@@ -15,15 +17,23 @@ function makeMessage(prompt: string): Omit<QueuedComposerMessage, "id"> {
1517 terminalContexts: [],
1618 previewAnnotations: [],
1719 reviewComments: [],
18 submissionIntent: "foreground",
20 sendSettings: {
21 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
22 runtimeMode: "full-access",
23 interactionMode: "default",
24 promptEffort: null,
25 },
1926 queuedAfterToolActivityId: null,
2027 createdAt: "2026-09-11T00:00:00.000Z",
2128 };
2229}
2330
31const queue = (threadKey: string) =>
32 useQueuedMessageStore.getState().queuesByThreadKey[threadKey] ?? [];
33
2434describe("queuedMessageStore", () => {
2535 beforeEach(() => {
26 useQueuedMessageStore.setState({ queuesByThreadKey: {}, drainGeneration: 0 });
36 useQueuedMessageStore.setState({ queuesByThreadKey: {}, lastDispatchByThreadKey: {} });
2737 });
2838
2939 it("keeps messages in submission order per thread", () => {
......@@ -32,71 +42,112 @@ describe("queuedMessageStore", () => {
3242 enqueue("thread-a", makeMessage("second"));
3343 enqueue("thread-b", makeMessage("other"));
3444
35 const queues = useQueuedMessageStore.getState().queuesByThreadKey;
36 expect(queues["thread-a"]?.map((message) => message.prompt)).toEqual(["first", "second"]);
37 expect(queues["thread-b"]?.map((message) => message.prompt)).toEqual(["other"]);
45 expect(queue("thread-a").map((message) => message.prompt)).toEqual(["first", "second"]);
46 expect(queue("thread-b").map((message) => message.prompt)).toEqual(["other"]);
3847 });
3948
40 it("take hands the message to exactly one caller", () => {
41 const { enqueue, take } = useQueuedMessageStore.getState();
42 const entry = enqueue("thread-a", makeMessage("first"));
49 it("allows one send per thread and re-anchors the rest to the current tool boundary", () => {
50 const { enqueue, beginSend } = useQueuedMessageStore.getState();
51 const first = enqueue("thread-a", makeMessage("first"));
52 const second = enqueue("thread-a", makeMessage("second"));
4353
44 expect(take("thread-a", entry.id, null)?.prompt).toBe("first");
45 expect(take("thread-a", entry.id, null)).toBeNull();
46 expect(useQueuedMessageStore.getState().queuesByThreadKey["thread-a"]).toBeUndefined();
54 expect(beginSend("thread-a", first.id, "tool-2")?.prompt).toBe("first");
55 expect(beginSend("thread-a", first.id, "tool-2")).toBeNull();
56 expect(beginSend("thread-a", second.id, "tool-2")).toBeNull();
57
58 const [sending, waiting] = queue("thread-a");
59 expect(sending?.sending).toBe("preparing");
60 expect(waiting?.queuedAfterToolActivityId).toBe("tool-2");
61 expect(
62 isQueuedMessageDue({ message: waiting!, phase: "running", latestToolActivityId: "tool-2" }),
63 ).toBe(false);
4764 });
4865
49 it("take re-anchors the remaining messages to the current tool boundary", () => {
50 const { enqueue, take } = useQueuedMessageStore.getState();
66 it("finishSend drops the sent message", () => {
67 const { enqueue, beginSend, finishSend } = useQueuedMessageStore.getState();
5168 const first = enqueue("thread-a", makeMessage("first"));
52 enqueue("thread-a", makeMessage("second"));
69 beginSend("thread-a", first.id, null);
5370
54 take("thread-a", first.id, "tool-2");
71 finishSend("thread-a", first.id);
5572
56 const [second] = useQueuedMessageStore.getState().queuesByThreadKey["thread-a"] ?? [];
57 expect(second?.queuedAfterToolActivityId).toBe("tool-2");
58 expect(
59 isQueuedMessageDue({ message: second!, phase: "running", latestToolActivityId: "tool-2" }),
60 ).toBe(false);
73 expect(useQueuedMessageStore.getState().queuesByThreadKey["thread-a"]).toBeUndefined();
74 });
75
76 it("failSend returns the message to the head, held", () => {
77 const { enqueue, beginSend, failSend } = useQueuedMessageStore.getState();
78 enqueue("thread-a", makeMessage("first"));
79 const second = enqueue("thread-a", makeMessage("second"));
80 beginSend("thread-a", second.id, null);
81
82 expect(failSend("thread-a", second.id)).toBe(true);
83
84 const [head] = queue("thread-a");
85 expect(queue("thread-a").map((message) => message.prompt)).toEqual(["second", "first"]);
86 expect(head?.sending).toBeUndefined();
87 expect(isQueuedMessageDue({ message: head!, phase: "ready", latestToolActivityId: null })).toBe(
88 false,
89 );
6190 });
6291
63 it("remove keeps the other messages' anchors", () => {
64 const { enqueue, remove } = useQueuedMessageStore.getState();
92 it("remove keeps the other messages' anchors and refuses a message being sent", () => {
93 const { enqueue, remove, beginSend } = useQueuedMessageStore.getState();
6594 const first = enqueue("thread-a", { ...makeMessage("first"), queuedAfterToolActivityId: "t1" });
6695 const second = enqueue("thread-a", makeMessage("second"));
6796
6897 expect(remove("thread-a", second.id)?.prompt).toBe("second");
6998 expect(remove("thread-a", second.id)).toBeNull();
70 expect(useQueuedMessageStore.getState().queuesByThreadKey["thread-a"]).toEqual([first]);
99 expect(queue("thread-a")).toEqual([first]);
100
101 beginSend("thread-a", first.id, "t1");
102 expect(remove("thread-a", first.id)).toBeNull();
71103 });
72104
73 it("holdAtFront returns a failed message to the head, held", () => {
74 const { enqueue, take, holdAtFront } = useQueuedMessageStore.getState();
105 it("a failed send keeps waiting on the dispatch before it", () => {
106 const { enqueue, beginSend, markDispatching, finishSend, failSend } =
107 useQueuedMessageStore.getState();
108 const earlier = createLocalDispatchSnapshot(undefined);
75109 const first = enqueue("thread-a", makeMessage("first"));
76 enqueue("thread-a", makeMessage("second"));
77 const taken = take("thread-a", first.id, "t1")!;
110 const second = enqueue("thread-a", makeMessage("second"));
111 const third = enqueue("thread-a", makeMessage("third"));
112 const lastDispatch = () => useQueuedMessageStore.getState().lastDispatchByThreadKey["thread-a"];
113 beginSend("thread-a", first.id, null);
114 markDispatching("thread-a", first.id, earlier);
115 finishSend("thread-a", first.id);
116
117 // Fails before its turn start went out: the first send is still the one to wait on.
118 beginSend("thread-a", second.id, null);
119 failSend("thread-a", second.id);
120 expect(lastDispatch()?.thread).toBe(earlier);
121
122 // Fails after going out: it never reached the server, so the first still counts.
123 beginSend("thread-a", third.id, null);
124 markDispatching("thread-a", third.id, { ...earlier, startedAt: "later" });
125 failSend("thread-a", third.id);
126 expect(lastDispatch()?.thread).toBe(earlier);
127 });
78128
79 holdAtFront("thread-a", taken);
129 it("Stop takes back a preparing send but not one already dispatching", () => {
130 const { enqueue, beginSend, markDispatching, drain, failSend } =
131 useQueuedMessageStore.getState();
132 const preparing = enqueue("thread-a", makeMessage("preparing"));
133 enqueue("thread-a", makeMessage("waiting"));
134 beginSend("thread-a", preparing.id, null);
80135
81 const queue = useQueuedMessageStore.getState().queuesByThreadKey["thread-a"] ?? [];
82 expect(queue.map((message) => message.prompt)).toEqual(["first", "second"]);
83 expect(queue[0]?.holdUntilUserAction).toBe(true);
84 expect(
85 isQueuedMessageDue({ message: queue[0]!, phase: "ready", latestToolActivityId: null }),
86 ).toBe(false);
87 });
136 expect(drain("thread-a").map((message) => message.prompt)).toEqual(["preparing", "waiting"]);
137 expect(markDispatching("thread-a", preparing.id, createLocalDispatchSnapshot(undefined))).toBe(
138 false,
139 );
140 expect(failSend("thread-a", preparing.id)).toBe(false);
88141
89 it("drain empties one thread's queue in order", () => {
90 const { enqueue, drain } = useQueuedMessageStore.getState();
91 enqueue("thread-a", makeMessage("first"));
92 enqueue("thread-a", makeMessage("second"));
93 enqueue("thread-b", makeMessage("other"));
142 const dispatching = enqueue("thread-b", makeMessage("dispatching"));
143 enqueue("thread-b", makeMessage("waiting"));
144 beginSend("thread-b", dispatching.id, null);
145 expect(
146 markDispatching("thread-b", dispatching.id, createLocalDispatchSnapshot(undefined)),
147 ).toBe(true);
94148
95 expect(drain("thread-a").map((message) => message.prompt)).toEqual(["first", "second"]);
96 expect(useQueuedMessageStore.getState().drainGeneration).toBe(1);
97 expect(drain("thread-a")).toEqual([]);
98 expect(useQueuedMessageStore.getState().drainGeneration).toBe(1);
99 expect(useQueuedMessageStore.getState().queuesByThreadKey["thread-b"]).toHaveLength(1);
149 expect(drain("thread-b").map((message) => message.prompt)).toEqual(["waiting"]);
150 expect(queue("thread-b").map((message) => message.prompt)).toEqual(["dispatching"]);
100151 });
101152});
102153
apps/web/src/queuedMessageStore.ts+165-100
......@@ -1,12 +1,30 @@
1import type { PreviewAnnotationPayload } from "@t3tools/contracts";
1import type {
2 ModelSelection,
3 PreviewAnnotationPayload,
4 ProviderInteractionMode,
5 RuntimeMode,
6} from "@t3tools/contracts";
27import { create } from "zustand";
38
4import type { ComposerSubmissionIntent } from "./composer-logic";
9import type { LocalDispatchSnapshot } from "./components/ChatView.logic";
510import type { ComposerFileAttachment, ComposerImageAttachment } from "./composerDraftStore";
611import type { TerminalContextDraft } from "./lib/terminalContext";
712import { randomUUID } from "./lib/utils";
813import type { ReviewCommentContext } from "./reviewCommentContext";
914
15/**
16 * The composer's model and modes when the message was queued. The send uses
17 * these instead of the live composer, so it can go out while the user is on
18 * another thread.
19 */
20export interface QueuedMessageSendSettings {
21 modelSelection: ModelSelection;
22 runtimeMode: RuntimeMode;
23 interactionMode: ProviderInteractionMode;
24 /** Effort written into the prompt text, for providers that read it there. */
25 promptEffort: string | null;
26}
27
1028/**
1129 * A composer submission held back while the thread's turn is running. It
1230 * carries the full draft snapshot so the send path can dispatch it later with
......@@ -20,7 +38,7 @@ export interface QueuedComposerMessage {
2038 terminalContexts: TerminalContextDraft[];
2139 previewAnnotations: PreviewAnnotationPayload[];
2240 reviewComments: ReviewCommentContext[];
23 submissionIntent: ComposerSubmissionIntent;
41 sendSettings: QueuedMessageSendSettings;
2442 /**
2543 * The newest completed tool activity at queue time. A different id later
2644 * means a tool call finished after the user queued, which is the boundary
......@@ -32,125 +50,172 @@ export interface QueuedComposerMessage {
3250 * user pressing send. It waits for Send now instead of leaving on its own.
3351 */
3452 holdUntilUserAction?: boolean;
53 /**
54 * Set while a send is under way; the row stays until it settles. Stop can
55 * still take a "preparing" message back (uploads, thread settings), but not
56 * a "dispatching" one, whose turn start is already on the wire.
57 */
58 sending?: "preparing" | "dispatching";
3559 createdAt: string;
3660}
3761
62/**
63 * The thread as it was when its last queued turn start went out. The next
64 * message waits until the server has moved past it, so a send that starts a
65 * new turn and the message after it do not leave on one boundary.
66 */
67interface QueuedDispatch {
68 /** The message that went out, or null for a dispatch restored after a failure. */
69 messageId: string | null;
70 thread: LocalDispatchSnapshot;
71 /** The dispatch this one replaced. If this send fails, that one still counts. */
72 previous: LocalDispatchSnapshot | null;
73}
74
3875interface QueuedMessageStoreState {
3976 queuesByThreadKey: Record<string, QueuedComposerMessage[]>;
40 /**
41 * Bumped by `drain`. A send that took a message before a drain and finishes
42 * its upload after it compares this to the value it captured and gives up,
43 * so Stop cannot be followed by a queued message starting a new turn.
44 */
45 drainGeneration: number;
77 lastDispatchByThreadKey: Record<string, QueuedDispatch>;
4678 enqueue: (threadKey: string, message: Omit<QueuedComposerMessage, "id">) => QueuedComposerMessage;
4779 /**
48 * Removes one message and returns it, or null when another caller already
49 * took it. The remaining messages are re-anchored to `toolActivityId` so
50 * only one queued message leaves per tool boundary.
80 * Marks one message as sending and returns it, or null when it is gone or
81 * the thread already has a send under way. The other messages are
82 * re-anchored to `toolActivityId` so only one leaves per tool boundary.
5183 */
52 take: (
84 beginSend: (
5385 threadKey: string,
5486 id: string,
5587 toolActivityId: string | null,
5688 ) => QueuedComposerMessage | null;
57 /** Removes one message without touching the others' anchors. Null when already gone. */
58 remove: (threadKey: string, id: string) => QueuedComposerMessage | null;
89 /** The turn start is going out. False when Stop took the message back first. */
90 markDispatching: (threadKey: string, id: string, thread: LocalDispatchSnapshot) => boolean;
91 /** Drops a message whose send went out, or that had nothing left to send. */
92 finishSend: (threadKey: string, id: string) => void;
5993 /**
60 * Puts a message back at the head, held for user action. Used when its
61 * send failed: the queue keeps its order and nothing behind it overtakes.
94 * Moves a message whose send failed back to the head, held for user action.
95 * The queue keeps its order and nothing behind it overtakes. False when
96 * Stop already took the message back.
6297 */
63 holdAtFront: (threadKey: string, message: QueuedComposerMessage) => void;
64 /** Removes and returns every queued message for the thread, oldest first. */
98 failSend: (threadKey: string, id: string) => boolean;
99 /** Removes one message without touching the others' anchors. Null when gone or sending. */
100 remove: (threadKey: string, id: string) => QueuedComposerMessage | null;
101 /** Removes and returns every message for the thread that is not already on the wire. */
65102 drain: (threadKey: string) => QueuedComposerMessage[];
66103}
67104
68105const EMPTY_QUEUE: QueuedComposerMessage[] = [];
69106
107type QueueState = Pick<QueuedMessageStoreState, "queuesByThreadKey" | "lastDispatchByThreadKey">;
108
109/** Replaces one thread's queue. `lastDispatch` null forgets it; an empty queue always does. */
110function withQueue(
111 state: QueueState,
112 threadKey: string,
113 queue: QueuedComposerMessage[],
114 lastDispatch?: QueuedDispatch | null,
115): QueueState {
116 const queuesByThreadKey = { ...state.queuesByThreadKey, [threadKey]: queue };
117 const lastDispatchByThreadKey = { ...state.lastDispatchByThreadKey };
118 if (lastDispatch) lastDispatchByThreadKey[threadKey] = lastDispatch;
119 if (queue.length === 0) delete queuesByThreadKey[threadKey];
120 if (queue.length === 0 || lastDispatch === null) delete lastDispatchByThreadKey[threadKey];
121 return { queuesByThreadKey, lastDispatchByThreadKey };
122}
123
70124/** In-memory only: a queued message is a live intent, not a draft worth persisting. */
71export const useQueuedMessageStore = create<QueuedMessageStoreState>()((set, get) => ({
72 queuesByThreadKey: {},
73 drainGeneration: 0,
74 enqueue: (threadKey, message) => {
75 const entry: QueuedComposerMessage = { ...message, id: randomUUID() };
76 set((state) => ({
77 queuesByThreadKey: {
78 ...state.queuesByThreadKey,
79 [threadKey]: [...(state.queuesByThreadKey[threadKey] ?? EMPTY_QUEUE), entry],
80 },
81 }));
82 return entry;
83 },
84 take: (threadKey, id, toolActivityId) => {
85 const queue = get().queuesByThreadKey[threadKey];
86 const entry = queue?.find((message) => message.id === id);
87 if (!queue || !entry) {
88 return null;
89 }
90 set((state) => {
91 const remaining = (state.queuesByThreadKey[threadKey] ?? EMPTY_QUEUE)
92 .filter((message) => message.id !== id)
93 .map((message) =>
94 message.queuedAfterToolActivityId === toolActivityId
95 ? message
96 : { ...message, queuedAfterToolActivityId: toolActivityId },
97 );
98 const queuesByThreadKey = { ...state.queuesByThreadKey };
99 if (remaining.length === 0) {
100 delete queuesByThreadKey[threadKey];
101 } else {
102 queuesByThreadKey[threadKey] = remaining;
103 }
104 return { queuesByThreadKey };
105 });
106 return entry;
107 },
108 remove: (threadKey, id) => {
109 const queue = get().queuesByThreadKey[threadKey];
110 const entry = queue?.find((message) => message.id === id);
111 if (!queue || !entry) {
112 return null;
113 }
114 set((state) => {
115 const remaining = (state.queuesByThreadKey[threadKey] ?? EMPTY_QUEUE).filter(
116 (message) => message.id !== id,
117 );
118 const queuesByThreadKey = { ...state.queuesByThreadKey };
119 if (remaining.length === 0) {
120 delete queuesByThreadKey[threadKey];
121 } else {
122 queuesByThreadKey[threadKey] = remaining;
123 }
124 return { queuesByThreadKey };
125 });
126 return entry;
127 },
128 holdAtFront: (threadKey, message) => {
129 set((state) => {
130 const rest = (state.queuesByThreadKey[threadKey] ?? EMPTY_QUEUE).filter(
131 (entry) => entry.id !== message.id,
125export const useQueuedMessageStore = create<QueuedMessageStoreState>()((set, get) => {
126 const queueOf = (threadKey: string) => get().queuesByThreadKey[threadKey] ?? EMPTY_QUEUE;
127 const update = (
128 threadKey: string,
129 queue: QueuedComposerMessage[],
130 lastDispatch?: QueuedDispatch | null,
131 ) => set((state) => withQueue(state, threadKey, queue, lastDispatch));
132 return {
133 queuesByThreadKey: {},
134 lastDispatchByThreadKey: {},
135 enqueue: (threadKey, message) => {
136 const entry: QueuedComposerMessage = { ...message, id: randomUUID() };
137 update(threadKey, [...queueOf(threadKey), entry]);
138 return entry;
139 },
140 beginSend: (threadKey, id, toolActivityId) => {
141 const queue = queueOf(threadKey);
142 const entry = queue.find((message) => message.id === id);
143 if (!entry || queue.some((message) => message.sending)) return null;
144 update(
145 threadKey,
146 queue.map((message) =>
147 message.id === id
148 ? { ...message, sending: "preparing" }
149 : message.queuedAfterToolActivityId === toolActivityId
150 ? message
151 : { ...message, queuedAfterToolActivityId: toolActivityId },
152 ),
132153 );
133 return {
134 queuesByThreadKey: {
135 ...state.queuesByThreadKey,
136 [threadKey]: [{ ...message, holdUntilUserAction: true }, ...rest],
154 return entry;
155 },
156 markDispatching: (threadKey, id, thread) => {
157 const queue = queueOf(threadKey);
158 if (!queue.some((message) => message.id === id && message.sending)) return false;
159 update(
160 threadKey,
161 queue.map((message) =>
162 message.id === id ? { ...message, sending: "dispatching" } : message,
163 ),
164 {
165 messageId: id,
166 thread,
167 previous: get().lastDispatchByThreadKey[threadKey]?.thread ?? null,
137168 },
138 };
139 });
140 },
141 drain: (threadKey) => {
142 const queue = get().queuesByThreadKey[threadKey];
143 if (!queue || queue.length === 0) {
144 return EMPTY_QUEUE;
145 }
146 set((state) => {
147 const queuesByThreadKey = { ...state.queuesByThreadKey };
148 delete queuesByThreadKey[threadKey];
149 return { queuesByThreadKey, drainGeneration: state.drainGeneration + 1 };
150 });
151 return queue;
152 },
153}));
169 );
170 return true;
171 },
172 finishSend: (threadKey, id) => {
173 const queue = queueOf(threadKey);
174 if (!queue.some((message) => message.id === id)) return;
175 update(
176 threadKey,
177 queue.filter((message) => message.id !== id),
178 );
179 },
180 failSend: (threadKey, id) => {
181 const queue = queueOf(threadKey);
182 const entry = queue.find((message) => message.id === id);
183 if (!entry) return false;
184 const { sending: _sending, ...rest } = entry;
185 // This send never reached the server, so only an earlier one is worth
186 // waiting for.
187 const dispatch = get().lastDispatchByThreadKey[threadKey];
188 update(
189 threadKey,
190 [{ ...rest, holdUntilUserAction: true }, ...queue.filter((message) => message.id !== id)],
191 dispatch?.messageId !== id
192 ? undefined
193 : dispatch.previous && { messageId: null, thread: dispatch.previous, previous: null },
194 );
195 return true;
196 },
197 remove: (threadKey, id) => {
198 const queue = queueOf(threadKey);
199 const entry = queue.find((message) => message.id === id);
200 if (!entry || entry.sending) return null;
201 update(
202 threadKey,
203 queue.filter((message) => message.id !== id),
204 );
205 return entry;
206 },
207 drain: (threadKey) => {
208 const queue = queueOf(threadKey);
209 const drained = queue.filter((message) => message.sending !== "dispatching");
210 if (drained.length === 0) return EMPTY_QUEUE;
211 update(
212 threadKey,
213 queue.filter((message) => message.sending === "dispatching"),
214 );
215 return drained;
216 },
217 };
218});
154219
155220/**
156221 * The newest finished tool call. Its id changing is the boundary a queued
apps/web/src/routes/__root.tsx+2
......@@ -29,6 +29,7 @@ import { DesktopAppActivationCoordinator } from "../components/desktop/DesktopAp
2929import { RunningThreadKeepAlive } from "../components/desktop/RunningThreadKeepAlive";
3030import { ProviderUpdateLaunchNotification } from "../components/ProviderUpdateLaunchNotification";
3131import { ThreadNotificationCoordinator } from "../components/ThreadNotificationCoordinator";
32import { QueuedMessageSender } from "../components/QueuedMessageSender";
3233import { ProjectCloneToastCoordinator } from "../components/ProjectCloneToastCoordinator";
3334import { SlowRpcRequestToastCoordinator } from "../components/SlowRpcRequestToastCoordinator";
3435import { ThemeEditorHost } from "../components/settings/ThemeEditorHost";
......@@ -227,6 +228,7 @@ function RootRouteView() {
227228 <SshPasswordPromptDialog />
228229 <SnapShotCoordinator />
229230 <ThreadNotificationCoordinator />
231 <QueuedMessageSender />
230232 <ConfirmDialogHost />
231233 <CustomSnoozeDialogHost />
232234 <SlowRpcRequestToastCoordinator />
apps/web/src/state/entities.ts+8
......@@ -183,6 +183,14 @@ export function readThreadShell(ref: ScopedThreadRef): EnvironmentThreadShell |
183183 return appAtomRegistry.get(environmentThreadShells.threadShellAtom(ref));
184184}
185185
186/** The thread as `useThread` returns it, read outside React. */
187export function readThread(ref: ScopedThreadRef): EnvironmentThread | null {
188 return mergeEnvironmentThread(
189 appAtomRegistry.get(environmentThreadDetails.detailAtom(ref)),
190 readThreadShell(ref),
191 );
192}
193
186194/** Whether the environment's server understands thread.settle/unsettle.
187195 False for pre-settlement servers (capability defaults false on decode),
188196 so clients under version skew fall back instead of erroring. */
docs/user/composer.md+1-1
......@@ -36,7 +36,7 @@ See [images and videos](#images-and-videos-in-messages) for previewing and savin
3636
3737On web and desktop, a message sent during a running turn waits at the end of the conversation as a
3838dashed bubble. It goes out on its own when the agent finishes its next tool
39call, or when the turn ends. Use the arrow under the bubble to send it right
39call, or when the turn ends, even while you have another thread open. Use the arrow under the bubble to send it right
4040away, or the X to move it back into the composer. Stop returns every queued
4141message to the composer.
4242