web-ui: queue a mid-turn message by default; steer only on request

plugins/web-ui/server/index.ts

@@ -1554,6 +1554,18 @@ const routeRequest = async (req: IncomingMessage, res: ServerResponse) => {
     if (method === "GET" && path === "/api/runs/active") {
       const threadRef = url.searchParams.get("threadRef") ?? "";
       if (!threadRef.startsWith("web:")) return json(res, 404, { error: "not_found" });
+      let queued: Array<{ runId: string; text: string }> = [];
+      let durableRunId: string | null = null;
+      const durable = await coreFetch("GET", `/v1/runs?threadRef=${encodeURIComponent(threadRef)}`);
+      if (durable.status >= 200 && durable.status < 300) {
+        try {
+          const parsed = JSON.parse(durable.text) as { runId?: string | null; queued?: typeof queued };
+          durableRunId = parsed.runId ?? null;
+          queued = parsed.queued ?? [];
+        } catch {
+          /* leave the queue empty; the run lookups below still answer */
+        }
+      }
       const tryRun = async (runId: string, ownedByUser = true): Promise<boolean> => {
         const r = await coreFetch("GET", `/v1/runs/${encodeURIComponent(runId)}`);
         if (r.status < 200 || r.status >= 300) {
@@ -1572,23 +1584,15 @@ const routeRequest = async (req: IncomingMessage, res: ServerResponse) => {
           return false;
         }
         rememberRun(runId, user, threadRef);
-        json(res, 200, { runId, run });
+        const waiting = queued.filter((q) => q.runId !== runId);
+        json(res, 200, { runId, run, ...(waiting.length ? { queued: waiting } : {}) });
         return true;
       };
+      if (durableRunId && (await tryRun(durableRunId, false))) return;
       for (const runId of Array.from(activeRunsByThread.get(threadKey(user, threadRef)) ?? [])) {
         if (await tryRun(runId)) return;
       }
-      const d = await coreFetch("GET", `/v1/runs?threadRef=${encodeURIComponent(threadRef)}`);
-      if (d.status >= 200 && d.status < 300) {
-        let runId: string | null = null;
-        try {
-          runId = (JSON.parse(d.text) as { runId?: string | null }).runId ?? null;
-        } catch {
-          void 0;
-        }
-        if (runId && (await tryRun(runId, false))) return;
-      }
-      json(res, 200, { runId: null, run: null });
+      json(res, 200, { runId: null, run: null, ...(queued.length ? { queued } : {}) });
       return;
     }
 
@@ -1612,6 +1616,13 @@ const routeRequest = async (req: IncomingMessage, res: ServerResponse) => {
       return relay(res, r);
     }
 
+    if (method === "POST" && path.startsWith("/api/runs/") && path.endsWith("/withdraw")) {
+      const id = decodeURIComponent(path.slice("/api/runs/".length, -"/withdraw".length));
+      const r = await coreFetch("POST", `/v1/runs/${encodeURIComponent(id)}/withdraw`);
+      if (r.status >= 200 && r.status < 300) forgetRun(id);
+      return relay(res, r);
+    }
+
     if (method === "GET" && path.startsWith("/api/runs/") && path.endsWith("/events")) {
       const id = decodeURIComponent(path.slice("/api/runs/".length, -"/events".length));
       let closed = false;

plugins/web-ui/src/chat.ts

@@ -380,6 +380,7 @@ export function createChatSurface(
         if (agent !== chatState.agent) return;
         adoptActiveSessionFromList(agent);
         await refreshTranscriptFromEntries(agent);
+        void followNextQueuedRun(agent, threadRef, normalStreamFn, onWork);
         if (wasUnsaved && chatState.sessionId) void settleNewSessionTitle(agent, threadRef);
       });
     });
@@ -628,6 +629,37 @@ export function createChatSurface(
     };
   }
 
+  async function followNextQueuedRun(
+    agent: Agent,
+    threadRef: string,
+    normalStreamFn: Agent["streamFn"],
+    onWork: (work: WorkBlock) => void,
+  ): Promise<void> {
+    let active: Awaited<ReturnType<typeof activeRunForThread>>;
+    try {
+      active = await activeRunForThread(threadRef);
+    } catch {
+      return;
+    }
+    if (agent !== chatState.agent || threadRef !== chatState.threadRef || agent.state.isStreaming) return;
+    const next = ctx.composer.queuedRunsFor(threadRef).find((r) => r.runId === active.runId);
+    ctx.composer.setQueuedRuns(threadRef, active.queued);
+    if (!active.runId || !active.run) return drawActiveChat(agent);
+    const recorded = (agent.state.messages.at(-1) as { role?: string } | undefined)?.role === "user";
+    if (!recorded && !next) return drawActiveChat(agent);
+    agent.streamFn = makeRunResumeStreamFn(active.runId, active.run, onWork, runSlot);
+    try {
+      await (recorded ? agent.continue() : agent.prompt(next!.text));
+    } catch (err) {
+      if (agent === chatState.agent) ctx.composer.state.error = errMessage(err, "Could not follow the queued message.");
+    } finally {
+      if (agent === chatState.agent) {
+        agent.streamFn = normalStreamFn;
+        await refreshTranscriptFromEntries(agent);
+      }
+    }
+  }
+
   async function resumeTrackedRun(
     agent: Agent,
     threadRef: string,
@@ -641,7 +673,15 @@ export function createChatSurface(
     } catch {
       return false;
     }
-    if (!activeRun || agent !== chatState.agent || appState.currentView !== "chats" || agent.state.isStreaming)
+    if (agent === chatState.agent && threadRef === chatState.threadRef)
+      ctx.composer.setQueuedRuns(threadRef, activeRun.queued);
+    if (
+      !activeRun.runId ||
+      !activeRun.run ||
+      agent !== chatState.agent ||
+      appState.currentView !== "chats" ||
+      agent.state.isStreaming
+    )
       return false;
     // Pull the transcript before attaching so the turn's triggering user message
     // (written by core, not by this tab) is on screen while the run streams.
@@ -2178,6 +2218,7 @@ export function createChatSurface(
     state: chatState,
     hasLiveRun: () => hasLiveRun(runSlot),
     signalLiveRun: (kind, text) => signalLiveRun(runSlot, kind, text),
+    currentTurnOptions,
     newChat,
     teardown: teardownActiveChat,
     resetChatState,