diff --git a/coworker/server/app.py b/coworker/server/app.py index c2e60e159..75abb3007 100644 --- a/coworker/server/app.py +++ b/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,11 +2548,12 @@ 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): @@ -2558,12 +2561,17 @@ async def claim_turn(*, retry: bool = False, content=None, display=None) -> None "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 diff --git a/surfaces/gui/e2e/queue-lifecycle.spec.ts b/surfaces/gui/e2e/queue-lifecycle.spec.ts new file mode 100644 index 000000000..b120ca570 --- /dev/null +++ b/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(); +}); diff --git a/surfaces/gui/e2e/queue-rejection.spec.ts b/surfaces/gui/e2e/queue-rejection.spec.ts new file mode 100644 index 000000000..939efff90 --- /dev/null +++ b/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'); +}); diff --git a/surfaces/gui/src/App.tsx b/surfaces/gui/src/App.tsx index 67cff675c..6b63399df 100644 --- a/surfaces/gui/src/App.tsx +++ b/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([]); const [projects, setProjects] = useState([]); const [sessionId, setSessionId] = useState(newId()); + // Follow-up messages queued while a task is running (#608) + const [queuedMessages, setQueuedMessages] = useState([]); + const pendingQueuedRef = useRef(new Map()); + const [queueReadySession, setQueueReadySession] = useState(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} diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts index a69e0ceca..9c3cf90b3 100644 --- a/surfaces/gui/src/api.ts +++ b/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 } : {}), }); } diff --git a/surfaces/gui/src/components/Composer.tsx b/surfaces/gui/src/components/Composer.tsx index 5e4b0e1bf..dc55d6058 100644 --- a/surfaces/gui/src/components/Composer.tsx +++ b/surfaces/gui/src/components/Composer.tsx @@ -1,6 +1,6 @@ import { useEffect, useLayoutEffect, useRef, useState, type ReactNode } from "react"; import { getI18n, useTranslation } from "react-i18next"; -import type { Attachment, SessionUsage } from "../types"; +import type { Attachment, QueuedMessage, SessionUsage } from "../types"; import { isPdfFile, readFile } from "../attach"; import { ProjectBindMenu } from "./ProjectBindMenu"; import { getSettings, inspectPdf, sessionSkills, type SessionSkillRow } from "../api"; @@ -133,6 +133,10 @@ interface Props { // §8.4 breaker tripped this turn: the mode chip says so quietly until the turn ends // or an ask_user answer resets the streak. reviewerPaused?: boolean; + // Follow-up messages queued while a task is running (#608) + queuedItems?: QueuedMessage[]; + onQueue?: (text: string, attachments?: Attachment[], skill?: string) => void; + onRemoveQueued?: (id: string) => void; } export function Composer(props: Props) { @@ -387,11 +391,20 @@ export function Composer(props: Props) { const body = (skill ? text.slice(skill.length + 1) : text).trim(); if ( (!body && attachments.length === 0 && !skill) || - (props.running && !props.gateOpen) || dictation?.recording || dictationBusy ) return; + // Follow-up queued while a task is running (#608) + if (props.running && !props.gateOpen) { + if (props.onQueue) { + props.onQueue(body, attachments, skill); + setText(""); + setAttachments([]); + setPendingSkill(null); + } + return; + } // No model connected: keep the draft (don't drop it) and send the user to setup instead. if (needsModel) { props.onConnectModel?.(); @@ -532,6 +545,55 @@ export function Composer(props: Props) { )} + {/* Queued follow-up messages strip (#608) */} + {props.queuedItems && props.queuedItems.length > 0 && ( +
+
+ + + + {t("composer.queued_count", { + count: props.queuedItems.length, + defaultValue: `${props.queuedItems.length} queued`, + })} + + +
+
+ {props.queuedItems.map((item, idx) => ( +
+ #{idx + 1} + {item.skill && ( + /{item.skill} + )} + + {item.text || (item.attachments?.length ? `[${item.attachments.length} files]` : "")} + + {item.error && {item.error}} + {props.onRemoveQueued && ( + + )} +
+ ))} +
+
+ )} +
- {t("composer.stop")} - +
+ {hasContent && props.onQueue && ( + + )} + +
) : (