feat(chat): frontend message queue — send-while-streaming queues, no disabled input

Reworks the queue UX per design feedback:

- **Send renders in the body.** A message sent to an idle thread submits
  normally and renders in the thread's user-message place + streams.

- **No disabled composer state, only a streaming state.** Tools, the
  microphone, and the mode pills are always enabled now; only the primary
  button reflects streaming (Send when there's a draft, Stop otherwise).

- **Send while a run is active → the queue, not the body.** New
  `conn.enqueue()` POSTs the message to the gate quietly (no optimistic body
  row, no run-status change); the message is mirrored into a per-thread
  frontend queue and surfaces in the panel. It renders in the body only when
  its turn dequeues and streams.

- **Frontend message queue.** New module-scoped per-thread store
  (`message-queue-store.ts`) with `useMessageQueue(threadId)` +
  `useMessageQueueActions()` hooks (enqueue / cancel / refresh). Seeded from
  the gate on mount, written optimistically on send, and re-synced on every
  SSE run start/end edge and on terminal `onFinish`. Replaces the
  React-Query `useThreadQueue` (deleted) and the `KEYS.threadQueue` key.

`selectWaitingQueueItems` still excludes the active head so the running
message never double-renders.

Verified live (Codex-desktop thread): message renders in the body + streams;
Tools/mic stay enabled mid-run; a message sent behind a running gate lands in
the panel (not the body) and cancels cleanly.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tlgimenes

apps/mesh/src/web/components/chat/chat-context.tsx

@@ -94,6 +94,18 @@ function statusToString(s: ConnStatus): ChatStreamContextValue["status"] {
   return s.kind;
 }
 
+/** Concatenate the text parts of a message for the queue-row display. */
+function queueItemTextFromParts(parts: ChatMessage["parts"]): string {
+  return parts
+    .map((p) =>
+      p && typeof p === "object" && (p as { type?: string }).type === "text"
+        ? ((p as { text?: string }).text ?? "")
+        : "",
+    )
+    .join("")
+    .trim();
+}
+
 import { useChatNavigation } from "./hooks/use-chat-navigation";
 import { useThreadActions, useThreadManager } from "./store/hooks";
 import { derivePartsFromTiptapDoc } from "./derive-parts";
@@ -105,6 +117,7 @@ import type { RunStatusStage } from "./run-status";
 import { useLocalStorage } from "../../hooks/use-local-storage";
 import { LOCALSTORAGE_KEYS } from "../../lib/localstorage-keys";
 import { KEYS } from "../../lib/query-keys";
+import { enqueueMessage, refreshMessageQueue } from "./message-queue-store";
 import { formatDeckTabId } from "@/web/layouts/main-panel-tabs/tab-id";
 import { useSimpleMode } from "../../hooks/use-organization-settings";
 
@@ -937,6 +950,9 @@ export function ActiveTaskProvider({
       },
       onFinish: (message, _messages, finishReason) => {
         const cb = cbRef.current;
+        // Terminal event: the gate advanced — re-sync the queue so the
+        // dequeued message drops and the next one surfaces.
+        if (cb.taskId) void refreshMessageQueue(cb.orgSlug, cb.taskId);
         if (cb.taskId) {
           cb.manager.patchThread({
             id: cb.taskId,
@@ -1000,11 +1016,15 @@ export function ActiveTaskProvider({
     };
     conn.observer = observer;
 
-    // Refresh the pending-message queue on every SSE run start/end edge. A
-    // gate dequeue flips conn.status idle→active (the next queued message
-    // began streaming); a finish flips active→idle (the slot freed). This is
-    // the SSE-driven replacement for the old 3s poll — the queue panel updates
-    // exactly when the thread gate advances, off the same stream the chat uses.
+    // Seed the frontend message queue from the gate on (re)mount so a reload
+    // or thread switch still shows what's queued.
+    void refreshMessageQueue(cbRef.current.orgSlug, cbRef.current.taskId);
+
+    // Re-sync the queue on every SSE run start/end edge. A gate dequeue flips
+    // conn.status idle→active (the next queued message began streaming); a
+    // finish flips active→idle (the slot freed). This is the SSE-driven
+    // refresh — the queue panel updates exactly when the thread gate advances,
+    // off the same stream the chat uses.
     const isActiveStatus = (k: ConnStatus["kind"]) =>
       k === "submitted" || k === "streaming";
     let prevActive = isActiveStatus(conn.status.get().kind);
@@ -1013,9 +1033,7 @@ export function ActiveTaskProvider({
       if (active === prevActive) return;
       prevActive = active;
       const cb = cbRef.current;
-      cb.queryClient.invalidateQueries({
-        queryKey: KEYS.threadQueue(cb.taskId),
-      });
+      void refreshMessageQueue(cb.orgSlug, cb.taskId);
     });
 
     return () => {
@@ -1100,35 +1118,60 @@ export function ActiveTaskProvider({
       metadata: messageMetadata,
     };
 
-    await conn.submit(
-      { kind: "message", message: userMessage },
-      {
-        tier: activeTier,
-        mode: modeToSend,
-        toolApprovalLevel:
-          preferences.toolApprovalLevel ?? readToolApprovalLevel(),
-        system: system || undefined,
-        agent: { id: capturedVirtualMcpId },
-        thread_id: capturedTaskId,
-        ...resolveSubmitSettings({
-          thread: activeTask
-            ? {
-                harness_id: activeTask.harness_id ?? null,
-                sandbox_provider_kind: activeTask.sandbox_provider_kind ?? null,
-                branch: activeTask.branch ?? null,
-              }
-            : null,
-          globals: {
-            harnessId: pendingHarnessId ?? undefined,
-            sandboxProviderKind: pendingSandboxProviderKind ?? undefined,
-            branch: currentBranch,
-          },
-        }),
-      },
-    );
-    queryClient.invalidateQueries({
-      queryKey: KEYS.threadQueue(capturedTaskId),
-    });
+    const requestOptions = {
+      tier: activeTier,
+      mode: modeToSend,
+      toolApprovalLevel:
+        preferences.toolApprovalLevel ?? readToolApprovalLevel(),
+      system: system || undefined,
+      agent: { id: capturedVirtualMcpId },
+      thread_id: capturedTaskId,
+      ...resolveSubmitSettings({
+        thread: activeTask
+          ? {
+              harness_id: activeTask.harness_id ?? null,
+              sandbox_provider_kind: activeTask.sandbox_provider_kind ?? null,
+              branch: activeTask.branch ?? null,
+            }
+          : null,
+        globals: {
+          harnessId: pendingHarnessId ?? undefined,
+          sandboxProviderKind: pendingSandboxProviderKind ?? undefined,
+          branch: currentBranch,
+        },
+      }),
+    };
+
+    // A run is already streaming (this thread) or in progress (hosted): the
+    // message can't be the current turn, so queue it behind the running one.
+    // POST it to the gate WITHOUT an optimistic body row, then mirror it into
+    // the frontend message queue (it renders in the body when its turn
+    // dequeues and streams). Otherwise this is the current turn — submit
+    // normally so it renders in the body and streams immediately.
+    if (isStreaming || isRunInProgress) {
+      try {
+        await conn.enqueue(userMessage, requestOptions);
+        enqueueMessage(capturedTaskId, {
+          workflowId: `thread-run:${capturedTaskId}:${userMessage.id}`,
+          messageId: userMessage.id,
+          text: queueItemTextFromParts(parts),
+          status: "queued",
+          enqueuedAt: Date.now(),
+        });
+      } catch (err) {
+        setChatError(err instanceof Error ? err : new Error(String(err)));
+        toast.error(
+          err instanceof Error ? err.message : "Failed to queue message",
+        );
+      }
+      // Reconcile the optimistic row against the gate's authoritative list.
+      void refreshMessageQueue(org.slug, capturedTaskId);
+    } else {
+      await conn.submit(
+        { kind: "message", message: userMessage },
+        requestOptions,
+      );
+    }
   }
 
   // Cancel run

apps/mesh/src/web/components/chat/input.tsx

@@ -534,7 +534,7 @@ export function ChatInput({
                     {/* Left Actions (+, Tools, active tool pills, stats) */}
                     <div className="flex items-center gap-1.5 min-w-0">
                       <ToolsPopover
-                        disabled={isStreaming}
+                        disabled={false}
                         onOpenConnections={() => {
                           track("connections_dialog_opened", {
                             source: "tools_popover",
@@ -576,7 +576,6 @@ export function ChatInput({
                       {chatMode === "gen-image" && imageModel && (
                         <button
                           type="button"
-                          disabled={isStreaming}
                           onClick={() => {
                             playSwitchSound();
                             track("chat_mode_changed", {
@@ -603,7 +602,6 @@ export function ChatInput({
                       {chatMode === "web-search" && webSearchModel && (
                         <button
                           type="button"
-                          disabled={isStreaming}
                           onClick={() => {
                             playSwitchSound();
                             track("chat_mode_changed", {
@@ -630,7 +628,6 @@ export function ChatInput({
                       {chatMode === "deep-research" && deepResearchModel && (
                         <button
                           type="button"
-                          disabled={isStreaming}
                           onClick={() => {
                             playSwitchSound();
                             track("chat_mode_changed", {
@@ -674,14 +671,14 @@ export function ChatInput({
                       )}
                       <TierTrigger />
 
-                      {/* Microphone button — kept mounted (and disabled)
-                          during streaming/run to avoid layout shift when
-                          the send button morphs into stop/cancel. */}
+                      {/* Microphone button — always enabled; the composer has
+                          no disabled state, only a streaming state reflected by
+                          the send/stop button. */}
                       {voice.isSupported && (
                         <Button
                           type="button"
                           onClick={handleVoiceStart}
-                          disabled={isStreaming || isRunInProgress}
+                          disabled={false}
                           variant="ghost"
                           size="icon"
                           className={cn(

apps/mesh/src/web/components/chat/message-queue-store.ts

@@ -0,0 +1,61 @@
+import {
+  dropQueueItem,
+  upsertQueueItem,
+  type QueueItemDTO,
+} from "./queue-items";
+import { Store } from "./store/store-primitive";
+
+/**
+ * Per-thread frontend message queue. A module-scoped registry of Stores keyed
+ * by thread id, so a thread's queue outlives any single chat mount (switching
+ * threads and back keeps the optimistic queue). Seeded from the server
+ * `/queue` endpoint and written optimistically when the user sends a message
+ * behind a running turn. The panel reads this — not the network — so a queued
+ * message appears instantly.
+ */
+const stores = new Map<string, Store<QueueItemDTO[]>>();
+
+export function messageQueueStore(threadId: string): Store<QueueItemDTO[]> {
+  let store = stores.get(threadId);
+  if (!store) {
+    store = new Store<QueueItemDTO[]>([]);
+    stores.set(threadId, store);
+  }
+  return store;
+}
+
+/** Optimistically append a just-sent message (idempotent by messageId). */
+export function enqueueMessage(threadId: string, item: QueueItemDTO): void {
+  messageQueueStore(threadId).update((cur) => upsertQueueItem(cur, item));
+}
+
+/** Drop a message — cancelled, or dequeued into the running slot. */
+export function removeMessage(threadId: string, messageId: string): void {
+  messageQueueStore(threadId).update((cur) => dropQueueItem(cur, messageId));
+}
+
+/** Replace a thread's queue with the server's authoritative list. */
+export function setQueue(threadId: string, items: QueueItemDTO[]): void {
+  messageQueueStore(threadId).set(items);
+}
+
+/**
+ * Fetch the server's gate queue for a thread and replace the local store.
+ * Best-effort: on any error the optimistic store stands until the next call.
+ */
+export async function refreshMessageQueue(
+  orgSlug: string,
+  threadId: string,
+): Promise<void> {
+  try {
+    const res = await fetch(
+      `/api/${encodeURIComponent(orgSlug)}/decopilot/queue/${encodeURIComponent(threadId)}`,
+      { credentials: "include" },
+    );
+    if (!res.ok) return;
+    const body = (await res.json()) as { items?: QueueItemDTO[] };
+    setQueue(threadId, body.items ?? []);
+  } catch {
+    // best-effort — leave the current store contents in place
+  }
+}

apps/mesh/src/web/components/chat/queue-items.test.ts

@@ -1,6 +1,10 @@
 import { describe, expect, it } from "bun:test";
-import { selectWaitingQueueItems } from "./queue-items";
-import type { QueueItemDTO } from "./use-thread-queue";
+import {
+  dropQueueItem,
+  selectWaitingQueueItems,
+  upsertQueueItem,
+  type QueueItemDTO,
+} from "./queue-items";
 
 const item = (
   id: string,
@@ -55,3 +59,36 @@ describe("selectWaitingQueueItems", () => {
     expect(out.map((i) => i.messageId)).toEqual(["q1", "q2"]);
   });
 });
+
+describe("upsertQueueItem", () => {
+  it("appends a new item", () => {
+    const out = upsertQueueItem(
+      [item("a", "queued", 1)],
+      item("b", "queued", 2),
+    );
+    expect(out.map((i) => i.messageId)).toEqual(["a", "b"]);
+  });
+
+  it("is idempotent by messageId (no duplicate when re-enqueued)", () => {
+    const out = upsertQueueItem(
+      [item("a", "queued", 1)],
+      item("a", "queued", 9),
+    );
+    expect(out.map((i) => i.messageId)).toEqual(["a"]);
+  });
+});
+
+describe("dropQueueItem", () => {
+  it("removes the matching messageId", () => {
+    const out = dropQueueItem(
+      [item("a", "queued", 1), item("b", "queued", 2)],
+      "a",
+    );
+    expect(out.map((i) => i.messageId)).toEqual(["b"]);
+  });
+
+  it("is a no-op when the id is absent", () => {
+    const out = dropQueueItem([item("a", "queued", 1)], "zzz");
+    expect(out.map((i) => i.messageId)).toEqual(["a"]);
+  });
+});

apps/mesh/src/web/components/chat/queue-items.ts

@@ -1,4 +1,16 @@
-import type { QueueItemDTO } from "./use-thread-queue";
+/** A pending message in a thread's gate queue, as surfaced to the UI. */
+export interface QueueItemDTO {
+  /** Full DBOS workflow id: `thread-run:{threadId}:{messageId}`. */
+  workflowId: string;
+  /** Trailing segment of the workflow id (the user message id). */
+  messageId: string;
+  /** Display text of the queued user message. */
+  text: string;
+  /** PENDING ("running") → being processed; ENQUEUED ("queued") → waiting. */
+  status: "running" | "queued";
+  /** Epoch ms the gate was created/enqueued. */
+  enqueuedAt: number;
+}
 
 /**
  * The messages *waiting behind* the active run, for the queue panel.
@@ -14,3 +26,21 @@ export function selectWaitingQueueItems(items: QueueItemDTO[]): QueueItemDTO[] {
   const sorted = [...items].sort((a, b) => a.enqueuedAt - b.enqueuedAt);
   return sorted.slice(1);
 }
+
+/** Append an item unless its messageId is already present (idempotent). */
+export function upsertQueueItem(
+  items: QueueItemDTO[],
+  item: QueueItemDTO,
+): QueueItemDTO[] {
+  return items.some((i) => i.messageId === item.messageId)
+    ? items
+    : [...items, item];
+}
+
+/** Remove the item with the given messageId. */
+export function dropQueueItem(
+  items: QueueItemDTO[],
+  messageId: string,
+): QueueItemDTO[] {
+  return items.filter((i) => i.messageId !== messageId);
+}

apps/mesh/src/web/components/chat/queue-panel.tsx

@@ -1,7 +1,7 @@
 import { Button } from "@deco/ui/components/button.tsx";
 import { X } from "@untitledui/icons";
 import { selectWaitingQueueItems } from "./queue-items";
-import { useCancelQueuedMessage, useThreadQueue } from "./use-thread-queue";
+import { useMessageQueue, useMessageQueueActions } from "./use-message-queue";
 
 /**
  * Pending-message queue above the composer. Lists only the messages *waiting*
@@ -10,8 +10,8 @@ import { useCancelQueuedMessage, useThreadQueue } from "./use-thread-queue";
  * chat body. Each row cancels its own gate workflow. Hidden when nothing waits.
  */
 export function ThreadQueuePanel({ taskId }: { taskId: string }) {
-  const { items } = useThreadQueue(taskId);
-  const cancel = useCancelQueuedMessage(taskId);
+  const items = useMessageQueue(taskId);
+  const { cancel } = useMessageQueueActions();
   const queued = selectWaitingQueueItems(items);
   if (queued.length === 0) return null;
 
@@ -38,7 +38,7 @@ export function ThreadQueuePanel({ taskId }: { taskId: string }) {
               size="icon"
               className="size-6 shrink-0 text-muted-foreground hover:text-foreground"
               title="Remove from queue"
-              onClick={() => cancel(item.workflowId)}
+              onClick={() => cancel(taskId, item.messageId)}
             >
               <X size={14} />
             </Button>