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

apps/web/src/components/QueuedMessageSender.test.tsx

@@ -0,0 +1,231 @@
+import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment";
+import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
+import { act, createElement } from "react";
+import { create, type ReactTestRenderer } from "react-test-renderer";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";
+
+import { useQueuedMessageStore, type QueuedComposerMessage } from "../queuedMessageStore";
+import { sendQueuedMessage } from "./chat/sendQueuedMessage";
+import { QueuedMessageSender } from "./QueuedMessageSender";
+
+const io = vi.hoisted(() => ({
+  run: vi.fn(),
+  upload: vi.fn(),
+  toast: vi.fn(),
+  thread: null as unknown,
+  shell: { runtimeMode: "full-access", interactionMode: "default" } as Record<string, unknown>,
+}));
+const config = {
+  environment: { capabilities: { attachmentUploads: true, inlineMessageContext: true } },
+};
+vi.mock("@t3tools/client-runtime/state/runtime", async (load) => ({
+  ...(await load<typeof import("@t3tools/client-runtime/state/runtime")>()),
+  runAtomCommand: (...args: unknown[]) => io.run(...args),
+}));
+vi.mock("../rpc/atomRegistry", () => ({
+  appAtomRegistry: { get: () => new Map([["env-a", config]]) },
+}));
+vi.mock("../state/server", () => ({ environmentServerConfigsAtom: {} }));
+vi.mock("../state/threads", () => ({
+  threadEnvironment: {
+    updateMetadata: "metadata",
+    setRuntimeMode: "runtime",
+    setInteractionMode: "interaction",
+    startTurn: "start",
+  },
+}));
+vi.mock("../state/environments", () => ({
+  useEnvironment: () => ({ connection: { phase: "connected" } }),
+}));
+vi.mock("../state/entities", () => ({
+  useThread: () => io.thread,
+  useThreadStatus: () => "live",
+  useServerConfigs: () => new Map([["env-a", config]]),
+  readThreadShell: () => io.shell,
+  readThread: () => io.thread,
+}));
+vi.mock("./ui/toast", () => ({ toastManager: { add: (...args: unknown[]) => io.toast(...args) } }));
+vi.mock("../lib/attachmentUploadQueue", () => ({
+  startAttachmentUpload: vi.fn(),
+  awaitAttachmentUploads: (...args: unknown[]) => io.upload(...args),
+  getUploadedAttachments: () => [
+    { type: "image", id: "uploaded", name: "a.png", mimeType: "image/png", sizeBytes: 4 },
+  ],
+  releaseDraftAttachments: vi.fn(),
+}));
+
+const threadRef = scopeThreadRef(EnvironmentId.make("env-a"), ThreadId.make("thread-a"));
+const threadKey = scopedThreadKey(threadRef);
+const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" };
+
+function enqueue(overrides: Partial<QueuedComposerMessage> = {}) {
+  return useQueuedMessageStore.getState().enqueue(threadKey, {
+    prompt: "follow up",
+    images: [],
+    files: [],
+    terminalContexts: [],
+    previewAnnotations: [],
+    reviewComments: [],
+    sendSettings: {
+      modelSelection,
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      promptEffort: null,
+    },
+    queuedAfterToolActivityId: null,
+    createdAt: "2026-09-25T00:00:00Z",
+    ...overrides,
+  });
+}
+
+const commandsRun = () => io.run.mock.calls.map((call) => call[1]);
+const queue = () => useQueuedMessageStore.getState().queuesByThreadKey[threadKey];
+
+beforeEach(() => {
+  vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true);
+  useQueuedMessageStore.setState({ queuesByThreadKey: {}, lastDispatchByThreadKey: {} });
+  io.thread = null;
+  io.run.mockReset().mockResolvedValue({ _tag: "Success", value: undefined });
+  io.upload.mockReset().mockResolvedValue(undefined);
+  io.toast.mockReset();
+  io.shell = {
+    modelSelection,
+    branch: null,
+    runtimeMode: "full-access",
+    interactionMode: "default",
+  };
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+});
+
+describe("QueuedMessageSender", () => {
+  const thread = (
+    status: string,
+    { toolActivityIds = [] as string[], userMessageIds = [] as string[] } = {},
+  ) => ({
+    session: { status, activeTurnId: null, updatedAt: status },
+    activities: toolActivityIds.map((id, index) => ({
+      id,
+      kind: "tool.completed",
+      sequence: index,
+      createdAt: "2026-09-25T00:00:01Z",
+    })),
+    messages: userMessageIds.map((id) => ({ id, role: "user" })),
+    latestTurn: null,
+  });
+  let root: ReactTestRenderer | null = null;
+  const render = () =>
+    act(() => {
+      if (root) root.update(createElement(QueuedMessageSender));
+      else root = create(createElement(QueuedMessageSender));
+    });
+  afterEach(async () => {
+    await act(() => root?.unmount());
+    root = null;
+  });
+
+  it("sends a queued message when the turn ends, with no chat view open", async () => {
+    enqueue();
+    io.thread = thread("running");
+    await render();
+    expect(commandsRun()).toEqual([]);
+
+    io.thread = thread("ready");
+    await render();
+
+    expect(commandsRun()).toEqual(["start"]);
+    expect(io.run.mock.calls[0]?.[2]).toMatchObject({
+      environmentId: "env-a",
+      input: { threadId: "thread-a", message: { text: "follow up" }, modelSelection },
+    });
+    expect(queue()).toBeUndefined();
+  });
+
+  it("holds the next message until the server picks up the one before it", async () => {
+    enqueue({ prompt: "first" });
+    enqueue({ prompt: "second" });
+    io.thread = thread("ready");
+    await render();
+    await render();
+    expect(commandsRun()).toEqual(["start"]);
+
+    // The first message started a turn; the second waits for its next tool call.
+    io.thread = thread("running", { userMessageIds: ["first"] });
+    await render();
+    expect(commandsRun()).toEqual(["start"]);
+    io.thread = thread("running", { userMessageIds: ["first"], toolActivityIds: ["tool-1"] });
+    await render();
+    expect(commandsRun()).toEqual(["start", "start"]);
+  });
+
+  it("moves on to the next message after a failed one is cancelled", async () => {
+    io.run.mockResolvedValueOnce({ _tag: "Failure", cause: Cause.fail(new Error("offline")) });
+    const first = enqueue({ prompt: "first" });
+    enqueue({ prompt: "second" });
+    io.thread = thread("ready");
+    await render();
+    expect(queue()?.[0]).toMatchObject({ prompt: "first", holdUntilUserAction: true });
+
+    await act(() => {
+      useQueuedMessageStore.getState().remove(threadKey, first.id);
+    });
+    await render();
+
+    expect(commandsRun()).toEqual(["start", "start"]);
+    expect(io.run.mock.calls[1]?.[2]).toMatchObject({ input: { message: { text: "second" } } });
+  });
+});
+
+describe("sendQueuedMessage", () => {
+  it("saves a mode changed before queueing, then starts the turn", async () => {
+    io.shell = { ...io.shell, runtimeMode: "approval-required" };
+    const message = enqueue();
+
+    await sendQueuedMessage(threadRef, message.id);
+
+    expect(commandsRun()).toEqual(["runtime", "start"]);
+    expect(io.run.mock.calls[1]?.[2]).toMatchObject({ input: { runtimeMode: "full-access" } });
+    expect(queue()).toBeUndefined();
+  });
+
+  it("gives a message back to Stop while its upload runs, without starting a turn", async () => {
+    let finishUpload!: () => void;
+    io.upload.mockReturnValue(new Promise<void>((resolve) => (finishUpload = resolve)));
+    const image = {
+      type: "image" as const,
+      id: "image-1",
+      name: "a.png",
+      mimeType: "image/png",
+      sizeBytes: 4,
+      previewUrl: "data:image/png;base64,AAAA",
+      file: new File(["AAAA"], "a.png", { type: "image/png" }),
+    };
+    const message = enqueue({ images: [image] });
+
+    const sending = sendQueuedMessage(threadRef, message.id);
+    expect(useQueuedMessageStore.getState().drain(threadKey)).toHaveLength(1);
+    finishUpload();
+    await sending;
+
+    expect(commandsRun()).toEqual([]);
+    expect(io.toast).not.toHaveBeenCalled();
+    expect(queue()).toBeUndefined();
+  });
+
+  it("holds a message at the head when the turn start fails", async () => {
+    io.run.mockResolvedValue({ _tag: "Failure", cause: Cause.fail(new Error("offline")) });
+    enqueue({ prompt: "first" });
+    const second = enqueue({ prompt: "second" });
+
+    await sendQueuedMessage(threadRef, second.id);
+
+    expect(queue()?.map((entry) => [entry.prompt, entry.holdUntilUserAction])).toEqual([
+      ["second", true],
+      ["first", undefined],
+    ]);
+    expect(io.toast).toHaveBeenCalledWith(expect.objectContaining({ description: "offline" }));
+  });
+});

apps/web/src/components/QueuedMessageSender.tsx

@@ -0,0 +1,100 @@
+import { parseScopedThreadKey } from "@t3tools/client-runtime/environment";
+import { derivePendingRequests } from "@t3tools/client-runtime/pending-requests";
+import { useEffect, useMemo } from "react";
+import { useShallow } from "zustand/react/shallow";
+
+import { useComposerDraftStore } from "../composerDraftStore";
+import {
+  isQueuedMessageDue,
+  latestCompletedToolActivityId,
+  useQueuedMessageStore,
+  useQueuedMessages,
+} from "../queuedMessageStore";
+import { derivePhase } from "../session-logic";
+import { useServerConfigs, useThread, useThreadStatus } from "../state/entities";
+import { useEnvironment } from "../state/environments";
+import { hasServerAcknowledgedLocalDispatch, latestTurnStartFailureId } from "./ChatView.logic";
+import { sendQueuedMessage } from "./chat/sendQueuedMessage";
+
+/**
+ * Sends queued messages when they are due, for every thread with a queue,
+ * whether or not the thread is on screen. Mounted once at the root.
+ */
+export function QueuedMessageSender() {
+  const threadKeys = useQueuedMessageStore(
+    useShallow((state) => Object.keys(state.queuesByThreadKey)),
+  );
+  return threadKeys.map((threadKey) => <ThreadQueueSender key={threadKey} threadKey={threadKey} />);
+}
+
+/**
+ * Watches one thread while it has queued messages. Reading the thread keeps
+ * its detail subscribed, so tool boundaries and the end of the turn are
+ * visible while the user is elsewhere.
+ */
+function ThreadQueueSender({ threadKey }: { threadKey: string }) {
+  const threadRef = useMemo(() => parseScopedThreadKey(threadKey), [threadKey]);
+  const thread = useThread(threadRef);
+  const threadStatus = useThreadStatus(threadRef);
+  const environmentId = threadRef?.environmentId ?? null;
+  const environment = useEnvironment(environmentId);
+  const serverConfigs = useServerConfigs();
+  const serverConfigLoaded = environmentId !== null && serverConfigs.has(environmentId);
+  const rewinding = useComposerDraftStore((store) => store.rewindingThreadKeys.has(threadKey));
+  const queue = useQueuedMessages(threadKey);
+  const next = queue[0];
+  const sending = queue.some((message) => message.sending);
+  const activities = thread?.activities;
+  const latestToolActivityId = useMemo(
+    () => latestCompletedToolActivityId(activities ?? []),
+    [activities],
+  );
+  const pendingRequests = useMemo(() => derivePendingRequests(activities ?? []), [activities]);
+  const phase = derivePhase(thread?.session ?? null);
+
+  // A send that starts a new turn leaves the thread idle until the server
+  // picks it up. Hold the next message until then, as the composer does for
+  // its own sends.
+  const lastDispatch = useQueuedMessageStore(
+    (state) => state.lastDispatchByThreadKey[threadKey]?.thread ?? null,
+  );
+  const latestUserMessageId = thread?.messages.findLast((m) => m.role === "user")?.id ?? null;
+  const waitingForServer =
+    lastDispatch !== null &&
+    !hasServerAcknowledgedLocalDispatch({
+      localDispatch: lastDispatch,
+      phase,
+      latestTurn: thread?.latestTurn ?? null,
+      latestUserMessageId,
+      session: thread?.session ?? null,
+      hasPendingApproval: pendingRequests.approvals.length > 0,
+      hasPendingUserInput: pendingRequests.userInputs.length > 0,
+      latestTurnStartFailureId: latestTurnStartFailureId(thread ?? undefined, latestUserMessageId),
+      threadError: null,
+    });
+
+  // Approvals and questions block the agent; a steer landing on top of them
+  // would answer nothing and confuse the turn, so the queue holds until the
+  // user resolves them.
+  const blocked =
+    threadRef === null ||
+    thread === null ||
+    threadStatus !== "live" ||
+    (environment !== null && environment.connection.phase !== "connected") ||
+    !serverConfigLoaded ||
+    rewinding ||
+    sending ||
+    waitingForServer ||
+    pendingRequests.approvals.length > 0 ||
+    pendingRequests.userInputs.length > 0;
+  const due =
+    next !== undefined &&
+    !blocked &&
+    isQueuedMessageDue({ message: next, phase, latestToolActivityId });
+  const nextId = next?.id;
+  useEffect(() => {
+    if (!due || !threadRef || nextId === undefined) return;
+    void sendQueuedMessage(threadRef, nextId);
+  }, [due, nextId, threadRef]);
+  return null;
+}

apps/web/src/components/chat/MessagesTimeline.logic.test.ts

@@ -1100,7 +1100,12 @@ describe("deriveMessagesTimelineRows", () => {
     terminalContexts: [],
     previewAnnotations: [],
     reviewComments: [],
-    submissionIntent: "foreground" as const,
+    sendSettings: {
+      modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
+      runtimeMode: "full-access" as const,
+      interactionMode: "default" as const,
+      promptEffort: null,
+    },
     queuedAfterToolActivityId: null,
     createdAt: "2026-01-01T00:00:01Z",
   });

apps/web/src/components/chat/MessagesTimeline.tsx

@@ -1788,11 +1788,14 @@ function QueuedMessageTimelineRow({
     queuedMessage.previewAnnotations.length +
     queuedMessage.reviewComments.length;
   const text = queuedMessage.prompt.trim();
-  const statusLabel = queuedMessage.holdUntilUserAction
-    ? "Waits for Send now"
-    : row.isNext
-      ? "Sends after the next tool call or when the turn ends"
-      : "Sends after the messages above it";
+  const sending = queuedMessage.sending !== undefined;
+  const statusLabel = sending
+    ? "Sending to the agent"
+    : queuedMessage.holdUntilUserAction
+      ? "Waits for Send now"
+      : row.isNext
+        ? "Sends after the next tool call or when the turn ends"
+        : "Sends after the messages above it";
   return (
     <div className="flex flex-col items-end" data-queued-message-id={queuedMessage.id}>
       <div className="max-w-[80%] rounded-2xl border border-dashed border-border p-3 text-message-foreground/80">
@@ -1820,14 +1823,14 @@ function QueuedMessageTimelineRow({
           <Tooltip>
             <TooltipTrigger
               render={<span className="inline-flex h-6 items-center gap-1" />}
-              aria-label={`Queued. ${statusLabel}.`}
+              aria-label={`${sending ? "Sending" : "Queued"}. ${statusLabel}.`}
             >
               <ClockIcon className="size-3.5" aria-hidden />
-              Queued
+              {sending ? "Sending" : "Queued"}
             </TooltipTrigger>
             <TooltipPopup side="bottom">{statusLabel}</TooltipPopup>
           </Tooltip>
-          <div className="ml-auto flex items-center gap-0.5">
+          <div className={cn("ml-auto flex items-center gap-0.5", sending && "invisible")}>
             <Tooltip>
               <TooltipTrigger
                 render={