feat(chat): refine queue panel — hide running, enable input mid-run, SSE refresh

Address three UI remarks on the thread-queue panel (#4155):

1. The queue panel now lists only *waiting* (ENQUEUED) messages. The
   running/PENDING head is excluded — it's already rendered in the chat
   body, so showing it in the queue too was duplicative.

2. The composer is no longer disabled while a run is in progress. A draft
   sends even mid-run (the message enqueues behind the running gate;
   concurrency=1 serializes the thread), so users can stack follow-ups.
   New pure helper `resolveComposerAction({hasDraft,isStreaming,
   isRunInProgress})` → send | stop | disabled, unit-tested; the primary
   button flips Stop→Send the moment there's a draft.

3. The queue is now refreshed from SSE, not a 3s poll. Dropped
   `refetchInterval`; the query re-fetches via `KEYS.threadQueue`
   invalidation on send, on cancel, and on every `conn.status` run
   start/end edge (the SSE stream the chat already consumes).

Verified live against a seeded stranded gate (1 PENDING head + 2 ENQUEUED):
panel shows only the 2 queued, input editable mid-run, per-item cancel +
refresh, and no recurring /queue/ polling in the network log.

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

Author: tlgimenes

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

@@ -999,8 +999,28 @@ export function ActiveTaskProvider({
       onToolCall: (event) => cbRef.current.onToolCall(event as never),
     };
     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.
+    const isActiveStatus = (k: ConnStatus["kind"]) =>
+      k === "submitted" || k === "streaming";
+    let prevActive = isActiveStatus(conn.status.get().kind);
+    const unsubStatus = conn.status.subscribe(() => {
+      const active = isActiveStatus(conn.status.get().kind);
+      if (active === prevActive) return;
+      prevActive = active;
+      const cb = cbRef.current;
+      cb.queryClient.invalidateQueries({
+        queryKey: KEYS.threadQueue(cb.taskId),
+      });
+    });
+
     return () => {
       if (conn.observer === observer) conn.observer = null;
+      unsubStatus();
     };
   }, [conn]);
 

apps/mesh/src/web/components/chat/composer-action.test.ts

@@ -0,0 +1,64 @@
+import { describe, expect, it } from "bun:test";
+import { resolveComposerAction } from "./composer-action";
+
+describe("resolveComposerAction", () => {
+  it("sends when there is a draft, even while a run streams (enqueue)", () => {
+    expect(
+      resolveComposerAction({
+        hasDraft: true,
+        isStreaming: true,
+        isRunInProgress: false,
+      }),
+    ).toBe("send");
+  });
+
+  it("sends when there is a draft and a hosted run is in progress", () => {
+    expect(
+      resolveComposerAction({
+        hasDraft: true,
+        isStreaming: false,
+        isRunInProgress: true,
+      }),
+    ).toBe("send");
+  });
+
+  it("sends when there is a draft and nothing is running", () => {
+    expect(
+      resolveComposerAction({
+        hasDraft: true,
+        isStreaming: false,
+        isRunInProgress: false,
+      }),
+    ).toBe("send");
+  });
+
+  it("offers stop when streaming with no draft", () => {
+    expect(
+      resolveComposerAction({
+        hasDraft: false,
+        isStreaming: true,
+        isRunInProgress: false,
+      }),
+    ).toBe("stop");
+  });
+
+  it("offers stop when a hosted run is in progress with no draft", () => {
+    expect(
+      resolveComposerAction({
+        hasDraft: false,
+        isStreaming: false,
+        isRunInProgress: true,
+      }),
+    ).toBe("stop");
+  });
+
+  it("is disabled when idle with no draft", () => {
+    expect(
+      resolveComposerAction({
+        hasDraft: false,
+        isStreaming: false,
+        isRunInProgress: false,
+      }),
+    ).toBe("disabled");
+  });
+});

apps/mesh/src/web/components/chat/composer-action.ts

@@ -0,0 +1,20 @@
+/** The composer's primary-button action. */
+export type ComposerAction = "send" | "stop" | "disabled";
+
+/**
+ * Decide the composer's primary-button action.
+ *
+ * A draft always sends — even while a run streams or a hosted run is in
+ * progress — because a second message enqueues behind the running one on the
+ * thread gate (concurrency=1 serializes them). With no draft, an active run
+ * offers stop; otherwise the button is disabled. Pure + total.
+ */
+export function resolveComposerAction(state: {
+  hasDraft: boolean;
+  isStreaming: boolean;
+  isRunInProgress: boolean;
+}): ComposerAction {
+  if (state.hasDraft) return "send";
+  if (state.isStreaming || state.isRunInProgress) return "stop";
+  return "disabled";
+}

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

@@ -68,6 +68,7 @@ import { useVoiceInput } from "@/web/hooks/use-voice-input.ts";
 import { VoiceWaveform } from "./voice-input";
 import { shouldRenderInlineModeRow } from "./input-mode-row";
 import { ThreadQueuePanel } from "./queue-panel";
+import { resolveComposerAction } from "./composer-action";
 
 // ============================================================================
 // useWindowFileDrop - Reusable hook for window-level file drag & drop
@@ -406,23 +407,25 @@ export function ChatInput({
   // model's context is much fuller than it actually is.
   const lastTotalTokens = lastUsage?.contextTokens ?? 0;
 
-  const canSubmit =
-    !isStreaming && !isModelsLoading && !isTiptapDocEmpty(tiptapDoc);
-
-  const showStopOrCancel = isStreaming || isRunInProgress;
+  // A draft sends even mid-run — the new message enqueues behind the running
+  // gate (concurrency=1 serializes the thread). Stop is offered only when
+  // there's nothing to send. `canSubmit`/`showStopOrCancel` are kept as the
+  // names the button/render logic below already references.
+  const hasDraft = !isModelsLoading && !isTiptapDocEmpty(tiptapDoc);
+  const composerAction = resolveComposerAction({
+    hasDraft,
+    isStreaming,
+    isRunInProgress,
+  });
+  const canSubmit = composerAction === "send";
+  const showStopOrCancel = composerAction === "stop";
   const showInlineModeRow = shouldRenderInlineModeRow({
     messageCount: messages.length,
     showConnectionsBanner,
   });
   const handleSubmit = (e?: FormEvent) => {
     e?.preventDefault();
-    if (isStreaming) {
-      track("chat_message_stopped", { thread_id: taskId });
-      stop();
-    } else if (isRunInProgress) {
-      track("chat_message_stopped", { thread_id: taskId });
-      stop();
-    } else if (canSubmit && tiptapDoc) {
+    if (composerAction === "send" && tiptapDoc) {
       track("chat_message_sent", {
         thread_id: taskId || null,
         mode: chatMode,
@@ -438,6 +441,9 @@ export function ChatInput({
       }
       clearChatDraft(sessionStorage, locator, draftKey);
       setTiptapDoc(undefined);
+    } else if (composerAction === "stop") {
+      track("chat_message_stopped", { thread_id: taskId });
+      stop();
     }
   };
 
@@ -462,17 +468,14 @@ export function ChatInput({
           {stream && taskCtx && <ChatHighlight />}
 
           {stream && taskCtx && taskId ? (
-            <ThreadQueuePanel
-              taskId={taskId}
-              active={isStreaming || isRunInProgress}
-            />
+            <ThreadQueuePanel taskId={taskId} />
           ) : null}
 
           <TiptapProvider
             key={taskId}
             tiptapDoc={tiptapDoc}
             setTiptapDoc={setTiptapDoc}
-            disabled={isStreaming}
+            disabled={false}
             enterToSubmit={true}
             onSubmit={handleSubmit}
           >
@@ -490,7 +493,7 @@ export function ChatInput({
               <div className="group/input relative flex flex-col gap-2 flex-1">
                 <TiptapInput
                   ref={tiptapRef}
-                  disabled={isStreaming || voice.status === "recording"}
+                  disabled={voice.status === "recording"}
                   virtualMcpId={selectedVirtualMcp?.id ?? decopilotId}
                   showFileUploader={true}
                   selectedModel={selectedModel}
@@ -703,8 +706,7 @@ export function ChatInput({
                           if (showStopOrCancel) {
                             e.preventDefault();
                             e.stopPropagation();
-                            if (isStreaming) stop();
-                            else stop();
+                            stop();
                           }
                         }}
                         variant={
@@ -719,11 +721,11 @@ export function ChatInput({
                             "bg-muted text-muted-foreground hover:bg-muted hover:text-muted-foreground cursor-not-allowed",
                         )}
                         title={
-                          isStreaming
-                            ? "Stop generating"
-                            : isRunInProgress
-                              ? "Cancel run"
-                              : "Send message (Enter)"
+                          composerAction === "stop"
+                            ? isStreaming
+                              ? "Stop generating"
+                              : "Cancel run"
+                            : "Send message (Enter)"
                         }
                       >
                         {showStopOrCancel ? (

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

@@ -1,46 +1,31 @@
 import { Button } from "@deco/ui/components/button.tsx";
-import { cn } from "@deco/ui/lib/utils.ts";
 import { X } from "@untitledui/icons";
 import { useCancelQueuedMessage, useThreadQueue } from "./use-thread-queue";
 
 /**
- * Pending-message queue above the composer. Lists the thread's running head +
- * queued messages; each row cancels its own gate workflow. Hidden when empty.
+ * Pending-message queue above the composer. Lists only the messages *waiting*
+ * behind the active run — the running head is excluded here because it's
+ * already rendered in the chat body. Each row cancels its own gate workflow.
+ * Hidden when nothing is queued.
  */
-export function ThreadQueuePanel({
-  taskId,
-  active,
-}: {
-  taskId: string;
-  active: boolean;
-}) {
-  const { items } = useThreadQueue(taskId, { active });
+export function ThreadQueuePanel({ taskId }: { taskId: string }) {
+  const { items } = useThreadQueue(taskId);
   const cancel = useCancelQueuedMessage(taskId);
-  if (items.length === 0) return null;
-
-  const queuedCount = items.filter((i) => i.status === "queued").length;
+  const queued = items.filter((i) => i.status === "queued");
+  if (queued.length === 0) return null;
 
   return (
     <div className="mb-1.5 rounded-xl border border-border bg-muted/50 p-2 text-sm">
       <div className="px-1 pb-1 text-xs text-muted-foreground">
-        {queuedCount > 0
-          ? `${queuedCount} queued message${queuedCount === 1 ? "" : "s"}`
-          : "Running"}
+        {`${queued.length} queued message${queued.length === 1 ? "" : "s"}`}
       </div>
       <ul className="flex flex-col gap-1">
-        {items.map((item) => (
+        {queued.map((item) => (
           <li
             key={item.workflowId}
             className="flex items-center gap-2 rounded-lg bg-background px-2 py-1.5"
           >
-            <span
-              className={cn(
-                "size-1.5 shrink-0 rounded-full",
-                item.status === "running"
-                  ? "bg-primary animate-pulse"
-                  : "bg-muted-foreground/40",
-              )}
-            />
+            <span className="size-1.5 shrink-0 rounded-full bg-muted-foreground/40" />
             <span className="min-w-0 flex-1 truncate text-foreground">
               {item.text || (
                 <span className="text-muted-foreground">(no text)</span>
@@ -51,9 +36,7 @@ export function ThreadQueuePanel({
               variant="ghost"
               size="icon"
               className="size-6 shrink-0 text-muted-foreground hover:text-foreground"
-              title={
-                item.status === "running" ? "Cancel run" : "Remove from queue"
-              }
+              title="Remove from queue"
               onClick={() => cancel(item.workflowId)}
             >
               <X size={14} />

apps/mesh/src/web/components/chat/use-thread-queue.ts

@@ -12,25 +12,17 @@ export interface QueueItemDTO {
 }
 
 /**
- * Poll this thread's pending gate queue. Polls every 3s while the run is active
- * or items exist (cheap, bounded), otherwise idles (re-fetched on send via the
- * KEYS.threadQueue invalidation in chat-context). No useEffect (lint ban).
+ * Read this thread's pending gate queue. Event-driven, not polled: the query
+ * is re-fetched via `KEYS.threadQueue` invalidation on send (chat-context),
+ * on cancel (below), and on every SSE run start/end edge (the `conn.status`
+ * subscription in chat-context). No timer, no useEffect (lint ban).
  */
-export function useThreadQueue(
-  taskId: string,
-  opts: { active: boolean },
-): { items: QueueItemDTO[] } {
+export function useThreadQueue(taskId: string): { items: QueueItemDTO[] } {
   const { org } = useProjectContext();
   const { data } = useQuery({
     queryKey: KEYS.threadQueue(taskId),
     enabled: Boolean(taskId),
     staleTime: 0,
-    refetchInterval: (query) => {
-      const n =
-        (query.state.data as { items?: QueueItemDTO[] } | undefined)?.items
-          ?.length ?? 0;
-      return opts.active || n > 0 ? 3000 : false;
-    },
     queryFn: async (): Promise<{ items: QueueItemDTO[] }> => {
       const res = await fetch(`/api/${org.slug}/decopilot/queue/${taskId}`, {
         credentials: "include",