feat(composer): queue follow-up messages while task is running (#608)

coworker/server/app.py

@@ -2496,7 +2496,7 @@ def _resolve_pending(resolution: str) -> None:
             "iteration_end",
         }
 
-        async def run_turn(content, *, retry: bool = False, display=None) -> None:
+        async def run_turn(content, *, retry: bool = False, display=None, request_id=None) -> None:
             # The receive loop atomically claims this session before scheduling the task.
             # Keeping the claim outside prevents two back-to-back frames from both starting.
             try:
@@ -2506,6 +2506,8 @@ async def run_turn(content, *, retry: bool = False, display=None) -> None:
                     else engine.run(content, display=display)
                 )
                 async for event in events:
+                    if event.type.value == "turn_start" and request_id:
+                        event.data = {**event.data, "request_id": request_id}
                     # Broadcast to every socket viewing this session (this socket included — it's a
                     # registered client), so a second view of the same session stays in sync too.
                     await manager.broadcast_session(
@@ -2546,24 +2548,30 @@ async def run_turn(content, *, retry: bool = False, display=None) -> None:
                 }
             )
         inbound_times: deque[float] = deque()
+        request_id = None
 
         async def reject_input(reason: str) -> None:
             # Input validation failures are not provider failures and must not offer "Retry"
             # or flush an in-progress assistant stream in the GUI.
-            await ws.send_json({"type": "input_rejected", "data": {"error": reason}})
+            await ws.send_json({"type": "input_rejected", "data": {"error": reason, **({"request_id": request_id} if request_id else {})}})
 
         async def claim_turn(*, retry: bool = False, content=None, display=None) -> None:
             if not manager.try_mark_running(session_id):
                 await reject_input(
                     "This session is already running a turn. Wait for it to finish or stop it."
                 )
                 return
-            asyncio.create_task(run_turn(content, retry=retry, display=display))
+            asyncio.create_task(run_turn(content, retry=retry, display=display, request_id=request_id))
 
         try:
             while True:
+                request_id = None
                 try:
                     message = await ws.receive_json()
+                    if isinstance(message, dict) and message.get("type") == "user_message":
+                        candidate = message.get("request_id")
+                        if isinstance(candidate, str) and 0 < len(candidate) <= 128:
+                            request_id = candidate
                 except (json.JSONDecodeError, UnicodeDecodeError):
                     await reject_input("Invalid WebSocket message: expected JSON.")
                     continue

surfaces/gui/e2e/queue-lifecycle.spec.ts

@@ -0,0 +1,68 @@
+import { expect } from '@playwright/test';
+import { test } from './fixtures';
+test('queued messages stay in their original session', async ({page}) => {
+ const sent: {sid:string,text:string}[]=[];
+ await page.routeWebSocket(/\/ws\/session\//,ws=>{
+  const sid=ws.url().split('/ws/session/')[1].split('?')[0];
+  ws.send(JSON.stringify({type:'ready',data:{running:sid==='resume-live-1'}}));
+  ws.onMessage(raw=>{const m=JSON.parse(String(raw));if(m.type==='user_message')sent.push({sid,text:m.text});});
+ });
+ await page.goto('/');await page.getByRole('button',{name:/Show more/}).first().click();
+ await page.getByTitle('Long audit').click();await expect(page.getByRole('button',{name:/Stop/})).toBeVisible();
+ const box=page.getByPlaceholder(/Ask the coworker/);await box.fill('private audit followup');await box.press('Enter');
+ await page.getByTitle('Draft the launch note').first().click();await expect(page.getByRole('button',{name:'Send',exact:true})).toBeVisible();
+ await expect.poll(()=>sent).toEqual([]);
+});
+
+test('queue sends one message per acknowledged turn', async ({page}) => {
+ const sent:string[]=[];let complete:()=>void=()=>{};
+ await page.routeWebSocket(/\/ws\/session\//,ws=>{
+  const sid=ws.url().split('/ws/session/')[1].split('?')[0];
+  const send=(type:string,data={})=>ws.send(JSON.stringify({type,data}));
+  send('ready',{running:sid==='resume-live-1'});
+  if(sid==='resume-live-1')complete=()=>send('turn_done');
+  ws.onMessage(raw=>{const m=JSON.parse(String(raw));if(m.type==='user_message'){sent.push(m.text);send('turn_start',{input:m.text,request_id:m.request_id});}});
+ });
+ await page.goto('/');await page.getByRole('button',{name:/Show more/}).first().click();await page.getByTitle('Long audit').click();
+ await expect(page.getByRole('button',{name:/Stop/})).toBeVisible();const box=page.getByPlaceholder(/Ask the coworker/);
+ for(const text of ['first','second']){await box.fill(text);await box.press('Enter');}
+ complete();await expect.poll(()=>sent).toEqual(['first']);
+ await expect(page.getByTestId('composer-queue')).toContainText('second');
+ complete();await expect.poll(()=>sent).toEqual(['first','second']);
+ await expect(page.getByTestId('composer-queue')).toHaveCount(0);
+});
+
+
+test('an attached queued message is acknowledged by ID', async ({ page }) => {
+  let complete: () => void = () => {};
+  await page.routeWebSocket(/\/ws\/session\//, (ws) => {
+    const sid = ws.url().split('/ws/session/')[1].split('?')[0];
+    ws.send(JSON.stringify({ type: 'ready', data: { running: sid === 'resume-live-1' } }));
+    if (sid === 'resume-live-1') complete = () => ws.send(JSON.stringify({ type: 'turn_done', data: {} }));
+    ws.onMessage((raw) => {
+      const m = JSON.parse(String(raw));
+      if (m.type === 'user_message') {
+        expect(m.attachments).toHaveLength(1);
+        ws.send(JSON.stringify({ type: 'turn_start', data: {
+          input: [{ type: 'text', text: m.text }, { type: 'text', text: 'attachment framing' }],
+          request_id: m.request_id,
+        } }));
+      }
+    });
+  });
+  await page.goto('/');
+  await page.getByRole('button', { name: /Show more/ }).first().click();
+  await page.getByTitle('Long audit').click();
+  await expect(page.getByRole('button', { name: /Stop/ })).toBeVisible();
+  await page.locator('input[type="file"]').setInputFiles({
+    name: 'notes.txt', mimeType: 'text/plain', buffer: Buffer.from('attached notes'),
+  });
+  await expect(page.getByText('notes.txt').first()).toBeVisible();
+  const box = page.getByPlaceholder(/Ask the coworker/);
+  await box.fill('inspect attached notes');
+  await box.press('Enter');
+  await expect(page.getByTestId('composer-queue')).toBeVisible();
+  complete();
+  await expect(page.getByTestId('composer-queue')).toHaveCount(0);
+  await expect(page.getByText('inspect attached notes', { exact: false })).toBeVisible();
+});

surfaces/gui/e2e/queue-rejection.spec.ts

@@ -0,0 +1,16 @@
+import { expect } from '@playwright/test';
+import { test } from './fixtures';
+test('rejected queued input must not leave the session running',async({page})=>{
+ let complete:()=>void=()=>{};
+ await page.routeWebSocket(/\/ws\/session\//,ws=>{
+  const sid=ws.url().split('/ws/session/')[1].split('?')[0];
+  ws.send(JSON.stringify({type:'ready',data:{running:sid==='resume-live-1'}}));
+  if(sid==='resume-live-1')complete=()=>ws.send(JSON.stringify({type:'turn_done',data:{}}));
+  ws.onMessage(raw=>{const m=JSON.parse(String(raw));if(m.type==='user_message')ws.send(JSON.stringify({type:'input_rejected',data:{error:'Message too long',request_id:m.request_id}}));});
+ });
+ await page.goto('/');await page.getByRole('button',{name:/Show more/}).first().click();await page.getByTitle('Long audit').click();
+ await expect(page.getByRole('button',{name:/Stop/})).toBeVisible();const box=page.getByPlaceholder(/Ask the coworker/);await box.fill('rejected followup');await box.press('Enter');
+ complete();await expect(page.getByText('Message too long').first()).toBeVisible();
+ await expect(page.getByRole('button',{name:/Stop/})).toHaveCount(0);
+ await expect(page.getByTestId('composer-queue')).toContainText('rejected followup');
+});

surfaces/gui/src/App.tsx

@@ -44,6 +44,7 @@ import type {
   ApprovalDecision,
   Attachment,
   Item,
+  QueuedMessage,
   SessionInfo,
   SessionUsage,
   TodoItem,
@@ -249,6 +250,16 @@ export function App() {
   const [sessions, setSessions] = useState<SessionInfo[]>([]);
   const [projects, setProjects] = useState<RecentWorkspace[]>([]);
   const [sessionId, setSessionId] = useState<string>(newId());
+  // Follow-up messages queued while a task is running (#608)
+  const [queuedMessages, setQueuedMessages] = useState<QueuedMessage[]>([]);
+  const pendingQueuedRef = useRef(new Map<string, QueuedMessage>());
+  const [queueReadySession, setQueueReadySession] = useState<string | null>(null);
+  const rejectQueued = (sid: string, error: string) => {
+    const pending = pendingQueuedRef.current.get(sid);
+    if (!pending) return;
+    pendingQueuedRef.current.delete(sid);
+    setQueuedMessages((items) => items.map((m) => m.id === pending.id ? { ...m, error } : m));
+  };
   // Automation-run context (§ owner ask 2026-07-04): which task an open __run__ session belongs
   // to, driving the banner + "Back to runs". Best-effort — a run session without context still
   // shows a generic banner (detected by its __run__ id).
@@ -695,6 +706,7 @@ export function App() {
       if (ev.type !== "compacting") setCompacting(false);
       switch (ev.type) {
         case "ready":
+          setQueueReadySession(sessionId);
           setConnected(true);
           if (d.model) setModel(d.model);
           if (d.mode) setMode(d.mode);
@@ -707,7 +719,15 @@ export function App() {
           // without this the Stop button and waiting row vanish (owner catch 2026-08-24).
           if (typeof d.running === "boolean") setRunning(d.running);
           break;
-        case "turn_start":
+        case "turn_start": {
+          const pending = pendingQueuedRef.current.get(sessionId);
+          if (pending && d.request_id === pending.id) {
+            const shown = pending.skill ? `/${pending.skill}${pending.text ? ` ${pending.text}` : ""}` : pending.text;
+            setItems((items) => [...items, { kind: "user", text: shown, attachments: pending.attachments, ts: Date.now() / 1000 }]);
+            pendingQueuedRef.current.delete(sessionId);
+            setQueuedMessages((items) => items.filter((m) => m.id !== pending.id));
+          }
+        }
           setRunning(true);
           setReviewerPaused(false); // a fresh user message resets the denial streak
           setStreaming("");
@@ -944,6 +964,9 @@ export function App() {
           ]);
           break;
         case "input_rejected":
+          if (d.request_id === pendingQueuedRef.current.get(sessionId)?.id) {
+            rejectQueued(sessionId, d.error || t("app.notice.input_rejected"));
+          }
           setItems((p) => [
             ...p,
             { kind: "notice", tone: "warn", text: d.error || t("app.notice.input_rejected") },
@@ -988,10 +1011,18 @@ export function App() {
           sessionRef.current?.userMessage(p.text, p.attachments, p.model, p.skill);
         }
       },
-      onClose: () => setConnected(false),
+      onClose: () => {
+        setConnected(false);
+        setQueueReadySession(null);
+        rejectQueued(sessionId, t("composer.queue_disconnected"));
+      },
     });
     sessionRef.current = session;
-    return () => session.close();
+    return () => {
+      setQueueReadySession(null);
+      rejectQueued(sessionId, t("composer.queue_disconnected"));
+      session.close();
+    };
     // NOTE: `workspace` is intentionally NOT a dependency. Every real workspace change
     // (pick folder, select/switch session, new session) is paired with a `sessionId`
     // change, so the socket still reconnects when it should. The one workspace-only change
@@ -1133,6 +1164,34 @@ export function App() {
     sessionRef.current?.userMessage(text, attachments, model, skill);
     followLatest(); // sending always re-engages stream-following, wherever the user had scrolled
   };
+
+  const handleQueue = (text: string, attachments?: Attachment[], skill?: string) => {
+    if (!text.trim() && (!attachments || attachments.length === 0) && !skill) return;
+    const newQueued: QueuedMessage = {
+      id: `${Date.now()}-${Math.random().toString(36).slice(2, 7)}`,
+      sessionId,
+      text,
+      attachments,
+      skill,
+      createdAt: Date.now(),
+    };
+    setQueuedMessages((prev) => [...prev, newQueued]);
+  };
+
+  const handleRemoveQueued = (id: string) => {
+    setQueuedMessages((prev) => prev.filter((m) => m.id !== id));
+  };
+
+  useEffect(() => {
+    if (running || !connected || queueReadySession !== sessionId || pendingQueuedRef.current.has(sessionId)) return;
+    const next = queuedMessages.find((m) => m.sessionId === sessionId);
+    if (!next || next.error) return;
+    // Keep the item until turn_start acknowledges it. A rejection leaves an
+    // removable record and must not claim that an agent turn is running.
+    pendingQueuedRef.current.set(sessionId, next);
+    sessionRef.current?.userMessage(next.text, next.attachments, model, next.skill, next.id);
+  }, [running, connected, queueReadySession, sessionId, queuedMessages]);
+
   // Resolving a LIVE prompt also resolves its parked Inbox mirror server-side, but the polled
   // `sessionInbox` copy stays "pending" for up to a poll cycle — long enough for the docked
   // answer-in-context card to flash the SAME request again right after the user answered it
@@ -2090,6 +2149,9 @@ export function App() {
               onUnattendedChange={agent !== "chat" ? toggleUnattended : undefined}
               prefill={composerPrefill}
               resetKey={sessionId}
+              queuedItems={queuedMessages.filter((m) => m.sessionId === sessionId)}
+              onQueue={handleQueue}
+              onRemoveQueued={handleRemoveQueued}
               usage={usage}
               contextWindow={modelContextWindows[model]}
               contextBar={contextBar}

surfaces/gui/src/api.ts

@@ -2408,7 +2408,7 @@ export class Session {
    * exactly what the user sees — immune to set_model races across reconnects (a new cowork
    * session always reconnects once to adopt its scratch dir, which could drop a queued
    * set_model and leave the engine on a stale/resumed model; found 2026-07-04). */
-  userMessage(text: string, attachments?: unknown[], model?: string, skill?: string) {
+  userMessage(text: string, attachments?: unknown[], model?: string, skill?: string, requestId?: string) {
     this.send({
       type: "user_message",
       text,
@@ -2417,6 +2417,7 @@ export class Session {
       // Force-run (SKILLS-SPEC §4.1): the composer's /skill pick rides as its own field;
       // the server validates it against the session's effective menu and frames the turn.
       ...(skill ? { skill } : {}),
+      ...(requestId ? { request_id: requestId } : {}),
     });
   }