diff --git a/plugins/web-ui/server/index.ts b/plugins/web-ui/server/index.ts index 08d0b2a46..faa0c9431 100644 --- a/plugins/web-ui/server/index.ts +++ b/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 => { 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; diff --git a/plugins/web-ui/src/chat.ts b/plugins/web-ui/src/chat.ts index 35db6f6aa..012afca82 100644 --- a/plugins/web-ui/src/chat.ts +++ b/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 { + let active: Awaited>; + 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, diff --git a/plugins/web-ui/src/composer.ts b/plugins/web-ui/src/composer.ts index 1e0e1a63c..1194e04fa 100644 --- a/plugins/web-ui/src/composer.ts +++ b/plugins/web-ui/src/composer.ts @@ -9,6 +9,7 @@ import { Brain, Check, ChevronDown, + CornerDownRight, FileText, Paperclip, ScrollText, @@ -20,10 +21,14 @@ import { } from "lucide"; import { api, + ApiError, fetchRuntimeConfig, + queueTurn, updateRuntimeConfig, + withdrawRun, type ApprovalDecision, type PendingApproval, + type QueuedRun, type RuntimeConfig, } from "./core-bridge"; import { errMessage, swallow } from "../../chassis/src/errors"; @@ -39,6 +44,7 @@ import { getModelOptionsForHarness, harnessSupportsEffort, harnessSupportsFastMode, + harnessSupportsSteer, type EffortLevel, type ModelOption, type ModelOptionValue, @@ -227,6 +233,24 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { const pastedTextIds = new Set(); + const queuedRuns = new Map(); + + function queuedRunsFor(threadRef: string | null): QueuedRun[] { + return (threadRef ? queuedRuns.get(threadRef) : undefined) ?? []; + } + + function setQueuedRuns(threadRef: string, runs: QueuedRun[]): void { + if (runs.length) queuedRuns.set(threadRef, runs); + else queuedRuns.delete(threadRef); + } + + function forgetQueuedRun(threadRef: string, runId: string): void { + setQueuedRuns( + threadRef, + queuedRunsFor(threadRef).filter((r) => r.runId !== runId), + ); + } + let dragDepth = 0; let skillsLoading = false; let slashActiveIndex = 0; @@ -351,7 +375,7 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { const attachingDisabled = inputBlocked; let placeholder = "Ask anything"; if (inputBlocked) placeholder = runtimePending ? "Loading runtime…" : "Approve or deny to continue"; - else if (agent.state.isStreaming) placeholder = "Steer the running task…"; + else if (agent.state.isStreaming) placeholder = "Queue a message for after this turn…"; let composerNotice: TemplateResult | typeof nothing = nothing; if (composerState.processingFiles) { composerNotice = html`
Preparing files...
`; @@ -429,6 +453,7 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { ` : nothing } + ${queuedStrip(agent)} ${ approvalPauses.length ? composerApprovalPanel(approvalPauses) @@ -663,20 +688,64 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { ${icon(ArrowUp, 17)} `; } - const canSteer = Boolean(composerState.draft.trim()); - const steerTitle = composerState.attachments.length - ? "Steer the running task (attachments stay for your next message)" - : "Steer the running task"; + const canQueue = Boolean(composerState.draft.trim()); return html` - `; } + function queuedStrip(agent: Agent): TemplateResult | typeof nothing { + const queued = queuedRunsFor(ctx.chat.state.threadRef); + if (!queued.length) return nothing; + const steerable = + agent.state.isStreaming && ctx.chat.hasLiveRun() && harnessSupportsSteer(currentModelOption().harnessId); + return html` +
+ ${queued.map( + (q) => html` +
+ Queued + ${q.text} + + +
+ `, + )} +
+ `; + } + function composerApprovalPanel(approvals: PendingApproval[]): TemplateResult { const busy = ctx.chat.state.resolvingApprovals.size > 0; const decide = (decision: ApprovalDecision): void => { @@ -1115,66 +1184,87 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { agent.abort(); } - async function sendSteer(agent: Agent): Promise { + async function queueDraft(agent: Agent): Promise { + const threadRef = ctx.chat.state.threadRef; const text = composerState.draft.trim(); - if (!text) return; - if (ctx.chat.state.threadRef) bumpSessionActivity(ctx.chat.state.threadRef); + if (!text || !threadRef) return; clearActiveDraft(); composerState.draft = ""; composerState.error = ""; - agent.state.messages.push({ - role: "user", - content: text, - timestamp: Date.now(), - steered: true, - } as unknown as AgentMessage); ctx.chat.drawActiveChat(agent); clearComposerDom(agent); - if (!ctx.chat.hasLiveRun()) { - // The turn is between run states: submitted but /api/turn hasn't returned the - // run id yet, or the stream is tearing down. Dropping the message here is a - // silent no-op the user reads as a dead composer — hold it and deliver when - // the run slot settles (steer the live run, or resend as an ordinary prompt). - steerWhenLive(agent, text, 0); - return; - } - await deliverSteer(agent, text); + if (!(await enqueueTurn(agent, threadRef, text))) composerState.draft = text; + ctx.chat.drawActiveChat(agent); } - async function deliverSteer(agent: Agent, text: string): Promise { + async function enqueueTurn(agent: Agent, threadRef: string, text: string): Promise { try { - const outcome = await ctx.chat.signalLiveRun("steer", text); - if (!outcome.ok) recoverEndedRunSteer(agent, text, outcome); + const queued = await queueTurn(threadRef, text, agent, ctx.chat.currentTurnOptions); + setQueuedRuns(threadRef, [...queuedRunsFor(threadRef), queued]); + bumpSessionActivity(threadRef); + return true; } catch (err) { - composerState.error = errMessage(err, "Could not steer the running task."); - ctx.chat.drawActiveChat(agent); + composerState.error = errMessage(err, "Could not queue the message."); + return false; } } - function steerWhenLive(agent: Agent, text: string, attempt: number): void { - if (agent !== ctx.chat.state.agent) return; - if (ctx.chat.hasLiveRun()) { - void deliverSteer(agent, text); - return; + async function removeQueued(agent: Agent, queued: QueuedRun): Promise { + const threadRef = ctx.chat.state.threadRef; + if (!threadRef) return; + composerState.error = ""; + try { + await withdrawRun(queued.runId); + } catch (err) { + if (!(err instanceof ApiError && (err.status === 409 || err.status === 404))) { + composerState.error = errMessage(err, "Could not remove the queued message."); + return ctx.chat.drawActiveChat(agent); + } } - if (!agent.state.isStreaming) { - // The run ended without the slot ever going live — recover exactly like a - // steer that raced the run's end: resend the text as an ordinary prompt. - recoverEndedRunSteer(agent, text, {}); - return; + forgetQueuedRun(threadRef, queued.runId); + ctx.chat.drawActiveChat(agent); + } + + async function steerQueued(agent: Agent, queued: QueuedRun): Promise { + const threadRef = ctx.chat.state.threadRef; + if (!threadRef) return; + if (!ctx.chat.hasLiveRun()) { + composerState.error = "That turn already finished — this message will run as its own turn."; + return ctx.chat.drawActiveChat(agent); } - if (attempt < 40) { - window.setTimeout(() => steerWhenLive(agent, text, attempt + 1), 250); - return; + composerState.error = ""; + try { + if (!(await withdrawRun(queued.runId))) return ctx.chat.drawActiveChat(agent); + } catch (err) { + const started = err instanceof ApiError && err.status === 409; + const gone = err instanceof ApiError && err.status === 404; + if (started) composerState.error = "That message already started — it's the running turn now."; + else if (gone) composerState.error = "That message was already removed in another tab."; + else composerState.error = errMessage(err, "Could not steer with that message."); + if (started || gone) forgetQueuedRun(threadRef, queued.runId); + return ctx.chat.drawActiveChat(agent); } - const last = agent.state.messages[agent.state.messages.length - 1] as - { role?: string; content?: unknown } | undefined; - if (last?.role === "user" && last.content === text) agent.state.messages.pop(); - // Don't clobber anything typed while the message was held: put the held text - // back in front of the newer draft instead of overwriting it. - composerState.draft = composerState.draft.trim() ? `${text}\n\n${composerState.draft}` : text; - composerState.error = "Could not deliver the message — the running task never settled. It is back in the composer."; + forgetQueuedRun(threadRef, queued.runId); + bumpSessionActivity(threadRef); + agent.state.messages.push({ + role: "user", + content: queued.text, + timestamp: Date.now(), + steered: true, + } as unknown as AgentMessage); ctx.chat.drawActiveChat(agent); + + try { + const outcome = await ctx.chat.signalLiveRun("steer", queued.text); + if (!outcome.ok) recoverEndedRunSteer(agent, queued.text, outcome); + } catch (err) { + composerState.error = errMessage(err, "Could not steer the running task."); + const last = agent.state.messages[agent.state.messages.length - 1] as + { role?: string; content?: unknown } | undefined; + if (last?.role === "user" && last.content === queued.text) agent.state.messages.pop(); + if (!(await enqueueTurn(agent, threadRef, queued.text))) composerState.draft = queued.text; + ctx.chat.drawActiveChat(agent); + } } // The run ended before the steer landed (the client believed it was still live). @@ -1229,7 +1319,7 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { if (composerState.pasteView) closePasteView(agent); if (ctx.chat.state.resolvingApprovals.size > 0) return; if (ctx.chat.hasUnresolvedApproval()) return; - if (agent.state.isStreaming) return sendSteer(agent); + if (agent.state.isStreaming) return queueDraft(agent); const text = composerState.draft.trim(); if (!text && composerState.attachments.length === 0) return; if (ctx.chat.state.threadRef) { @@ -1512,6 +1602,8 @@ export function createComposerSurface(ctx: ConvCtx): ComposerSurface { return { state: composerState, composerForm, + queuedRunsFor, + setQueuedRuns, resetComposer, focusComposerEnd, resizeComposer, diff --git a/plugins/web-ui/src/conv-types.ts b/plugins/web-ui/src/conv-types.ts index 6ae240709..652c368be 100644 --- a/plugins/web-ui/src/conv-types.ts +++ b/plugins/web-ui/src/conv-types.ts @@ -2,7 +2,14 @@ import type { Agent } from "@earendil-works/pi-agent-core"; import type { TemplateResult } from "lit"; import type { DensityTier } from "./density"; import type { Attachment } from "@earendil-works/pi-web-ui"; -import type { ApprovalDecision, CoreSession, PendingApproval, entriesToMessages } from "./core-bridge"; +import type { + ApprovalDecision, + CoreSession, + PendingApproval, + QueuedRun, + TurnOptions, + entriesToMessages, +} from "./core-bridge"; import type { EffortLevel, ModelOption } from "./model-options"; import type { ComposerMenu } from "./composer"; @@ -56,6 +63,7 @@ export interface ChatSurface { state: ChatState; hasLiveRun(): boolean; signalLiveRun(kind: "abort" | "steer", text?: string): Promise; + currentTurnOptions(): TurnOptions; newChat(context?: { scopeId: string; name: string | null }): string; teardown(): void; resetChatState(): void; @@ -107,6 +115,8 @@ interface ComposerState { export interface ComposerSurface { state: ComposerState; composerForm(agent: Agent): TemplateResult; + queuedRunsFor(threadRef: string | null): QueuedRun[]; + setQueuedRuns(threadRef: string, runs: QueuedRun[]): void; resetComposer(): void; focusComposerEnd(): void; resizeComposer(): void; diff --git a/plugins/web-ui/src/core-bridge.ts b/plugins/web-ui/src/core-bridge.ts index a7cef34db..706d7094e 100644 --- a/plugins/web-ui/src/core-bridge.ts +++ b/plugins/web-ui/src/core-bridge.ts @@ -344,8 +344,14 @@ export interface TurnOptions { } export interface ActiveRun { + runId: string | null; + run: RunPoll | null; + queued: QueuedRun[]; +} + +export interface QueuedRun { runId: string; - run: RunPoll; + text: string; } export function isContinuable(s: Pick, user: string): boolean { @@ -575,10 +581,32 @@ export function makeCoreStreamFn( return fn as unknown as StreamFn; } -export async function activeRunForThread(threadRef: string): Promise { +export async function activeRunForThread(threadRef: string): Promise { const q = new URLSearchParams({ threadRef }); - const r = await api<{ runId?: string | null; run?: RunPoll | null }>(`/api/runs/active?${q.toString()}`); - return r.runId && r.run ? { runId: r.runId, run: r.run } : null; + const r = await api<{ runId?: string | null; run?: RunPoll | null; queued?: QueuedRun[] }>( + `/api/runs/active?${q.toString()}`, + ); + const live = r.runId && r.run ? { runId: r.runId, run: r.run } : { runId: null, run: null }; + return { ...live, queued: r.queued ?? [] }; +} + +export async function queueTurn( + threadRef: string, + text: string, + agent: Agent, + getTurnOptions?: () => TurnOptions, +): Promise { + const submit = await api<{ runId?: string }>("/api/turn", { + method: "POST", + body: JSON.stringify(turnRequestBody(threadRef, text, agent.state.model, agent, getTurnOptions)), + }); + if (!submit.runId) throw new Error("Could not queue the message."); + return { runId: submit.runId, text }; +} + +export async function withdrawRun(runId: string): Promise { + const r = await api<{ withdrawn?: boolean }>(runPath(runId, "/withdraw"), { method: "POST" }); + return r.withdrawn === true; } export function makeRunResumeStreamFn( @@ -634,6 +662,34 @@ export function makeOpenerStreamFn( return fn as unknown as StreamFn; } +function turnRequestBody( + threadRef: string, + text: string, + model: Model, + agent: Agent, + getTurnOptions?: () => TurnOptions, + attachments: CoreAttachment[] = [], +): Record { + const turnOptions = getTurnOptions?.() ?? {}; + const thinkingLevel = + !turnOptions.harness || harnessSupportsEffort(turnOptions.harness) + ? (turnOptions.effortLevel ?? agent.state.thinkingLevel ?? defaultEffortForModel(model)) + : undefined; + const timezone = browserTimezone(); + return { + text, + threadRef, + ...(turnOptions.harness ? { harness: turnOptions.harness } : {}), + model: model.id, + ...(thinkingLevel ? { thinkingLevel } : {}), + ...(typeof turnOptions.fastMode === "boolean" ? { fastMode: turnOptions.fastMode } : {}), + ...(timezone ? { timezone } : {}), + ...(turnOptions.scopeId ? { scopeId: turnOptions.scopeId } : {}), + ...(turnOptions.channelName ? { channelName: turnOptions.channelName } : {}), + ...(attachments.length ? { attachments } : {}), + }; +} + async function drive( stream: AssistantMessageEventStream, model: Model, @@ -650,12 +706,6 @@ async function drive( const work: WorkBlock = { status: "thinking", activity: [] }; (partial as AssistantWork).work = work; const notify = (): void => onWork?.(work); - const turnOptions = getTurnOptions?.() ?? {}; - const thinkingLevel = - !turnOptions.harness || harnessSupportsEffort(turnOptions.harness) - ? (turnOptions.effortLevel ?? agent.state.thinkingLevel ?? defaultEffortForModel(model)) - : undefined; - const timezone = browserTimezone(); try { notify(); stream.push({ type: "start", partial }); @@ -668,16 +718,7 @@ async function drive( const submit = await api<{ status?: string; runId?: string; reply?: string }>("/api/turn", { method: "POST", body: JSON.stringify({ - text, - threadRef, - ...(turnOptions.harness ? { harness: turnOptions.harness } : {}), - model: model.id, - ...(thinkingLevel ? { thinkingLevel } : {}), - ...(typeof turnOptions.fastMode === "boolean" ? { fastMode: turnOptions.fastMode } : {}), - ...(timezone ? { timezone } : {}), - ...(turnOptions.scopeId ? { scopeId: turnOptions.scopeId } : {}), - ...(turnOptions.channelName ? { channelName: turnOptions.channelName } : {}), - ...(attachments.length ? { attachments } : {}), + ...turnRequestBody(threadRef, text, model, agent, getTurnOptions, attachments), ...(approval ? { approval } : {}), ...(opener ? { proactiveOpener: true } : {}), }), diff --git a/plugins/web-ui/src/model-options.ts b/plugins/web-ui/src/model-options.ts index 5ee2c799f..658cebe4c 100644 --- a/plugins/web-ui/src/model-options.ts +++ b/plugins/web-ui/src/model-options.ts @@ -212,6 +212,10 @@ export function harnessSupportsFastMode(harnessId: string): boolean { return harnessId === "pi" || harnessId === "claude"; } +export function harnessSupportsSteer(harnessId: string): boolean { + return harnessId === "pi" || harnessId === "claude" || harnessId === "codex" || harnessId === "opencode"; +} + export function defaultEffortForModel(model: Model): EffortLevel { const provider = String(model.provider ?? model.api ?? "").toLowerCase(); return provider.includes("anthropic") ? "low" : "auto"; diff --git a/plugins/web-ui/src/shell.css b/plugins/web-ui/src/shell.css index 46e636b9f..f2497daba 100644 --- a/plugins/web-ui/src/shell.css +++ b/plugins/web-ui/src/shell.css @@ -1454,6 +1454,65 @@ a.chat-row-open { gap: 6px; margin-top: 9px; } +.queued-strip { + display: flex; + flex-direction: column; + gap: 4px; + margin-top: 9px; +} +.queued-chip { + display: flex; + align-items: center; + gap: 8px; + min-height: 28px; + padding: 4px 6px 4px 8px; + border: 1px dashed var(--border); + border-radius: 8px; + background: var(--background); + color: var(--muted-foreground); + font-size: 12px; + line-height: 1.2; +} +.queued-tag { + flex: none; + text-transform: uppercase; + letter-spacing: 0.04em; + font-size: 10px; + opacity: 0.75; +} +.queued-text { + flex: 1; + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + color: var(--foreground); +} +.queued-steer { + flex: none; + display: inline-flex; + align-items: center; + gap: 4px; + padding: 2px 8px; + border: 1px solid var(--border); + border-radius: 999px; + background: var(--background); + color: var(--foreground); + font-size: 11px; + cursor: pointer; + transition: background 0.12s ease; +} +.queued-steer:hover:not(:disabled) { + background: var(--secondary); +} +.queued-steer:disabled { + opacity: 0.45; + cursor: default; +} +.queued-steer:focus-visible { + outline: 2px solid color-mix(in srgb, var(--foreground) 35%, transparent); + outline-offset: 2px; +} .file-chip { display: inline-flex; align-items: center; @@ -6907,6 +6966,7 @@ h3.ambient-field-label { } [data-density] .runtime-upgrade, [data-density] .attachment-strip, +[data-density] .queued-strip, [data-density] .composer-approval-panel, [data-density] .composer-note, [data-density] .composer-error { diff --git a/plugins/web-ui/test/active-run-discovery-route.test.ts b/plugins/web-ui/test/active-run-discovery-route.test.ts new file mode 100644 index 000000000..25c0a49fe --- /dev/null +++ b/plugins/web-ui/test/active-run-discovery-route.test.ts @@ -0,0 +1,108 @@ +import { mintPortalIdentity, PORTAL_IDENTITY_HEADER } from "../../chassis/src/portal-identity.ts"; +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { createServer, type IncomingMessage } from "node:http"; +import type { AddressInfo } from "node:net"; + +// The scenario Bugbot caught on this feature: a surface instance that saw only the QUEUE +// submission holds the pending run in its local index and nothing else. Core's durable answer +// names the actual head. Discovery must report core's head as the live run — a pending turn +// queued behind a running one is never "active", however this instance learned of it. +const THREAD = "web:alice:t-disc"; +const runStatus = new Map([ + ["run-live", "running"], + ["run-queued", "pending"], +]); +let durableHead: string | null = "run-live"; +let queued: Array<{ runId: string; text: string }> = [{ runId: "run-queued", text: "after this" }]; + +const core = createServer((req: IncomingMessage, res) => { + const u = new URL(req.url ?? "", "http://core"); + const reply = (status: number, body: unknown): void => { + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify(body)); + }; + if (req.method === "POST" && u.pathname === "/v1/turns") { + let body = ""; + req.on("data", (c) => (body += c)); + return void req.on("end", () => reply(200, { status: "queued", runId: "run-queued" })); + } + if (req.method === "GET" && u.pathname === "/v1/runs") + return reply(200, { runId: durableHead, ...(queued.length ? { queued } : {}) }); + const m = /^\/v1\/runs\/([^/]+)$/.exec(u.pathname); + if (req.method === "GET" && m) { + const status = runStatus.get(decodeURIComponent(m[1]!)); + return status ? reply(200, { status }) : reply(404, { error: "not_found" }); + } + reply(404, { error: "not_found" }); +}); +await new Promise((r) => core.listen(0, r)); + +const SECRET = "active-run-discovery-test"; +process.env.CORE_API_URL = `http://localhost:${(core.address() as AddressInfo).port}`; +process.env.CORE_SIGNING_SECRET = SECRET; +process.env.WEB_UI_PRINCIPALS = "alice"; + +const { handler } = await import("../server/index.ts"); +const surface = createServer((req, res) => void handler(req, res)); +await new Promise((r) => surface.listen(0, r)); +const base = `http://localhost:${(surface.address() as AddressInfo).port}`; +const IDENTITY = { + cookie: "webuiuser=alice", + [PORTAL_IDENTITY_HEADER]: mintPortalIdentity({ p: "alice", exp: Date.now() + 60_000 }, SECRET), + "content-type": "application/json", +}; + +test.after(() => { + surface.close(); + core.close(); +}); + +test("a pending run this instance remembered never masks core's running head", async () => { + // Seed the instance-local index with ONLY the queued (pending) run — exactly what an instance + // that handled the queue submission but not the original send looks like. + const seed = await fetch(`${base}/api/turn`, { + method: "POST", + headers: IDENTITY, + body: JSON.stringify({ text: "after this", threadRef: THREAD }), + }); + assert.equal(seed.status, 200); + assert.equal(((await seed.json()) as { runId?: string }).runId, "run-queued"); + + const r = await fetch(`${base}/api/runs/active?threadRef=${encodeURIComponent(THREAD)}`, { headers: IDENTITY }); + assert.equal(r.status, 200); + const body = (await r.json()) as { runId: string | null; queued?: Array<{ runId: string }> }; + assert.equal(body.runId, "run-live", "core's durable head is the live run, not the remembered pending one"); + assert.deepEqual( + body.queued?.map((q) => q.runId), + ["run-queued"], + "the pending turn stays reported as queued", + ); +}); + +test("when core's head IS the remembered run, it reports live once and never doubles as queued", async () => { + durableHead = "run-queued"; + runStatus.set("run-queued", "running"); + queued = [{ runId: "run-queued", text: "after this" }]; + const r = await fetch(`${base}/api/runs/active?threadRef=${encodeURIComponent(THREAD)}`, { headers: IDENTITY }); + assert.equal(r.status, 200); + const body = (await r.json()) as { runId: string | null; queued?: Array<{ runId: string }> }; + assert.equal(body.runId, "run-queued"); + assert.equal(body.queued, undefined, "the followed run is filtered out of the queue"); +}); + +test("nothing in flight answers null with whatever core still holds queued", async () => { + durableHead = null; + // Both remembered runs are terminal now: the walk prunes them instead of reporting one. + runStatus.set("run-queued", "done"); + runStatus.set("run-live", "done"); + queued = [{ runId: "run-later", text: "still waiting" }]; + const r = await fetch(`${base}/api/runs/active?threadRef=${encodeURIComponent(THREAD)}`, { headers: IDENTITY }); + assert.equal(r.status, 200); + const body = (await r.json()) as { runId: string | null; queued?: Array<{ runId: string }> }; + assert.equal(body.runId, null); + assert.deepEqual( + body.queued?.map((q) => q.runId), + ["run-later"], + ); +}); diff --git a/plugins/web-ui/test/composer-source.test.ts b/plugins/web-ui/test/composer-source.test.ts index 5a0a31455..5f503874c 100644 --- a/plugins/web-ui/test/composer-source.test.ts +++ b/plugins/web-ui/test/composer-source.test.ts @@ -54,14 +54,17 @@ test("attaching files is allowed while a turn is streaming", () => { assert.ok(guards.length > 0, "streaming still gates steer/send routing"); }); -test("steering with pending attachments explains they ride the next message", () => { - assert.match(composer, /attachments stay for your next message/); +test("a mid-turn submit queues — attachments cannot ride a queued message and stay for the next", () => { + // Mid-turn Enter queues through core (queueDraft), so the steer-button attachment note is gone; + // the queue button gates only on draft text, never on the run slot. + assert.match(composer, /title="Queue for after this turn"/); + assert.doesNotMatch(composer, /attachments stay for your next message/); }); test("a steer whose run already ended is recovered, never silently dropped", () => { - // sendSteer must inspect the signal outcome and route a failed steer through recovery. - assert.match(composer, /const outcome = await ctx\.chat\.signalLiveRun\("steer", text\);/); - assert.match(composer, /if \(!outcome\.ok\) recoverEndedRunSteer\(agent, text, outcome\);/); + // steerQueued must inspect the signal outcome and route a failed steer through recovery. + assert.match(composer, /const outcome = await ctx\.chat\.signalLiveRun\("steer", queued\.text\);/); + assert.match(composer, /if \(!outcome\.ok\) recoverEndedRunSteer\(agent, queued\.text, outcome\);/); // Replayed by core → detach from the stale stream and attach to the fresh run. assert.match(composer, /function recoverEndedRunSteer\(/); assert.match(composer, /attachWhenIdle\(agent, 0\);/); diff --git a/plugins/web-ui/test/queue-by-default.test.ts b/plugins/web-ui/test/queue-by-default.test.ts new file mode 100644 index 000000000..cc76c77e4 --- /dev/null +++ b/plugins/web-ui/test/queue-by-default.test.ts @@ -0,0 +1,228 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { existsSync, readFileSync, readdirSync } from "node:fs"; +import { harnessSupportsSteer } from "../src/model-options.ts"; + +const composer = readFileSync(new URL("../src/composer.ts", import.meta.url), "utf8"); +const chat = readFileSync(new URL("../src/chat.ts", import.meta.url), "utf8"); +const bridge = readFileSync(new URL("../src/core-bridge.ts", import.meta.url), "utf8"); +const server = readFileSync(new URL("../server/index.ts", import.meta.url), "utf8"); + +test("a mid-turn Enter queues the message — it no longer steers the running turn", () => { + assert.match(composer, /if \(agent\.state\.isStreaming\) return queueDraft\(agent\);/); + assert.doesNotMatch(composer, /isStreaming\) return sendSteer\(/); + assert.match(composer, /placeholder = "Queue a message for after this turn…"/); + assert.match(composer, /title="Queue for after this turn"/); +}); + +// The whole point of the rewrite: the queue is core's, not the browser's. A queued message is a +// real run core will execute whether or not this tab survives, so there is nothing to flush, no +// per-tab store to keep in sync, and no way for two tabs to send the same message twice. +test("the queue lives in core, not in the browser", () => { + assert.ok(!existsSync(new URL("../src/message-queue.ts", import.meta.url)), "the localStorage queue module is gone"); + for (const [name, src] of [ + ["composer", composer], + ["chat", chat], + ] as const) { + assert.doesNotMatch(src, /"web-ui:queued"/, `${name} persists no queue of its own`); + assert.doesNotMatch(src, /flushQueuedMessages/, `${name} has no flush path left`); + } + assert.match( + composer, + /const queuedRuns = new Map\(\);/, + "what the composer holds is a view of core's queue, rebuilt from core — not a store", + ); + assert.match(composer, /const queued = await queueTurn\(threadRef, text, agent, ctx\.chat\.currentTurnOptions\);/); + assert.match( + bridge, + /export async function queueTurn\([\s\S]{0,600}?api<\{ runId\?: string \}>\("\/api\/turn"/, + "queuing is an ordinary turn submission; core enqueues it behind the live run", + ); +}); + +test("a queued turn carries the same model, effort and scope a typed one would", () => { + assert.match(bridge, /function turnRequestBody\(/); + const body = bridge.slice(bridge.indexOf("function turnRequestBody")); + assert.match(body, /thinkingLevel/); + assert.match(body, /scopeId: turnOptions\.scopeId/); + assert.match(bridge, /\.\.\.turnRequestBody\(threadRef, text, model, agent, getTurnOptions, attachments\),/); + assert.match(bridge, /turnRequestBody\(threadRef, text, agent\.state\.model, agent, getTurnOptions\)/); +}); + +// Found in live QA: when the Steer control was swapped in and out of an existing strip, a tab that +// opened the conversation BEFORE its run attached (the queue draws first, the live run a moment +// later) ended up with a Steer button that received clicks but ran nothing. Whether the harness can +// steer is fixed for the conversation, so the button's presence must be too — only its enabled +// state may change. +test("the Steer control's presence is fixed for the conversation; only enablement changes", () => { + const strip = composer.slice( + composer.indexOf("function queuedStrip"), + composer.indexOf("function composerApprovalPanel"), + ); + assert.match(strip, /\?disabled=\$\{!steerable\}/, "an unusable Steer is disabled, never removed"); + assert.doesNotMatch( + strip, + /steerable\s*\?\s*html`