diff --git a/tests/web/app-render.test.ts b/tests/web/app-render.test.ts index 64ca81ce..f7aeb4a4 100644 --- a/tests/web/app-render.test.ts +++ b/tests/web/app-render.test.ts @@ -1037,6 +1037,52 @@ test("app.js keeps an active agent running when its prompt receipt settles late" assert.equal(app.state.livePhase, "running"); }); +test("app.js reports an admitted native follow-up queue snapshot", async () => { + const app = await renderApp(); + app.context.fetch = async (url: unknown) => { + if (String(url) === "/api/prompt") { + return response({ + id: "received", + accepted: true, + pendingFollowUps: 2, + }); + } + if (String(url).startsWith("/api/snapshot")) return response(SNAPSHOT); + throw new Error(`unexpected request: ${String(url)}`); + }; + const input = app.elements.get("prompt-input"); + assert.ok(input); + input.value = "queue me"; + + await app.sendPrompt(); + + assert.match( + app.elements.get("composer-hint")?.textContent || "", + /2 follow-up messages were waiting when it was received/, + ); +}); + +test("app.js reports acceptance without a queue count when none are pending", async () => { + const app = await renderApp(); + app.context.fetch = async (url: unknown) => { + if (String(url) === "/api/prompt") { + return response({ id: "received", accepted: true, pendingFollowUps: 0 }); + } + if (String(url).startsWith("/api/snapshot")) return response(SNAPSHOT); + throw new Error(`unexpected request: ${String(url)}`); + }; + const input = app.elements.get("prompt-input"); + assert.ok(input); + input.value = "receive me"; + + await app.sendPrompt(); + + assert.equal( + app.elements.get("composer-hint")?.textContent, + "Message accepted by OpenPI Web.", + ); +}); + test("app.js scopes model selection to its session epoch", async () => { const app = await renderApp(); const model = deferred>(); diff --git a/tests/web/pi-adapter.test.ts b/tests/web/pi-adapter.test.ts index bdb4fcff..513bea1f 100644 --- a/tests/web/pi-adapter.test.ts +++ b/tests/web/pi-adapter.test.ts @@ -31,7 +31,7 @@ function runtimeFor( isIdle: () => true, getActiveTurn: () => undefined, cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), - sendPrompt: async () => {}, + sendPrompt: async () => ({ pendingFollowUps: 0 }), newSession: async () => ({ cancelled: false }), switchSession: async () => ({ cancelled: false }), listModels: () => [], diff --git a/tests/web/pi-runtime.test.ts b/tests/web/pi-runtime.test.ts index def9743b..e3a7bee3 100644 --- a/tests/web/pi-runtime.test.ts +++ b/tests/web/pi-runtime.test.ts @@ -55,13 +55,35 @@ function promptSession(sessionId: string) { options: PromptOptions; run: ReturnType; }> = []; + const listeners = new Set< + (event: { + type: "queue_update"; + steering: string[]; + followUp: string[]; + }) => void + >(); + let followUpMessages: string[] = []; return { isStreaming: false, pendingMessageCount: 0, sessionManager: { getSessionId: () => sessionId }, abort: async (): Promise => undefined, - subscribe() { - return () => undefined; + subscribe( + listener: (event: { + type: "queue_update"; + steering: string[]; + followUp: string[]; + }) => void, + ) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + getFollowUpMessages: () => followUpMessages, + emitFollowUpQueue(messages: string[]) { + followUpMessages = messages; + for (const listener of listeners) { + listener({ type: "queue_update", steering: [], followUp: messages }); + } }, prompt(content: string, options: PromptOptions) { const run = deferred(); @@ -283,13 +305,92 @@ test("prompt admission waits for Pi preflight acceptance", async () => { assert.equal(settled, false); session.calls[0].options.preflightResult?.(true); - await admission; + assert.deepEqual(await admission, { pendingFollowUps: 0 }); assert.equal(settled, true); session.calls[0].run.resolve(); await Promise.resolve(); }); +test("prompt admission snapshots Pi follow-up messages", async () => { + const session = promptSession("session-a"); + session.isStreaming = true; + const runtime = promptHarness(session); + const admission = runtime.sendPrompt("queued", { + commandId: "command-queued", + expectedSessionId: "session-a", + }); + await Promise.resolve(); + session.emitFollowUpQueue(["queued"]); + session.calls[0].options.preflightResult?.(true); + assert.deepEqual(await admission, { pendingFollowUps: 1 }); + session.calls[0].run.resolve(); + await Promise.resolve(); +}); + +test("prompt admission snapshots a follow-up queue that shrinks before it grows", async () => { + const session = promptSession("session-a"); + session.isStreaming = true; + session.emitFollowUpQueue(["already pending"]); + const runtime = promptHarness(session); + const admission = runtime.sendPrompt("queued", { + commandId: "command-queued", + expectedSessionId: "session-a", + }); + await Promise.resolve(); + + session.emitFollowUpQueue([]); + session.emitFollowUpQueue(["queued"]); + session.calls[0].options.preflightResult?.(true); + assert.deepEqual(await admission, { pendingFollowUps: 1 }); + session.calls[0].run.resolve(); + await Promise.resolve(); +}); + +test("prompt admission observes streaming after an earlier admission gate", async () => { + const session = promptSession("session-a"); + const runtime = promptHarness(session); + const first = runtime.sendPrompt("first", { + commandId: "command-first", + expectedSessionId: "session-a", + }); + await Promise.resolve(); + const second = runtime.sendPrompt("second", { + commandId: "command-second", + expectedSessionId: "session-a", + }); + await Promise.resolve(); + + session.calls[0].options.preflightResult?.(true); + session.isStreaming = true; + assert.deepEqual(await first, { pendingFollowUps: 0 }); + session.calls[0].run.resolve(); + await Promise.resolve(); + session.emitFollowUpQueue(["second"]); + session.calls[1].options.preflightResult?.(true); + assert.deepEqual(await second, { pendingFollowUps: 1 }); + + session.calls[1].run.resolve(); + await Promise.resolve(); +}); + +test("handled input snapshots an externally pending follow-up without claiming ownership", async () => { + const session = promptSession("session-a"); + session.isStreaming = true; + const runtime = promptHarness(session); + const admission = runtime.sendPrompt("/handled", { + commandId: "command-handled", + expectedSessionId: "session-a", + }); + await Promise.resolve(); + + session.emitFollowUpQueue(["external delivery"]); + session.calls[0].options.preflightResult?.(true); + assert.deepEqual(await admission, { pendingFollowUps: 1 }); + session.calls[0].run.resolve(); + await Promise.resolve(); +}); + test("prompt preflight rejection is a typed non-admission", async () => { const session = promptSession("session-a"); const runtime = promptHarness(session); diff --git a/tests/web/web-host.test.ts b/tests/web/web-host.test.ts index 094c02c4..f0f9f862 100644 --- a/tests/web/web-host.test.ts +++ b/tests/web/web-host.test.ts @@ -54,6 +54,7 @@ test("serves workspaces through a runtime isolated from terminal sessions", asyn cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async (content) => { prompts.push(content); + return { pendingFollowUps: 0 }; }, newSession: async (workspacePath, options) => { newSessions++; @@ -624,6 +625,7 @@ test("an unbound Host exposes no bootstrap Session and rejects prompt bypasses", cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async () => { prompts++; + return { pendingFollowUps: 0 }; }, newSession: async () => ({ cancelled: false }), switchSession: async () => ({ cancelled: false }), @@ -723,6 +725,7 @@ test("returns accepted only after Pi admits the prompt", async () => { sendPrompt: async () => { promptStarted = true; await promptAdmitted; + return { pendingFollowUps: 0 }; }, newSession: async () => ({ cancelled: false }), switchSession: async () => ({ cancelled: false }), @@ -763,7 +766,12 @@ test("returns accepted only after Pi admits the prompt", async () => { resolvePrompt(); const response = await responsePromise; assert.equal(response.status, 202); - assert.equal((await response.json()).accepted, true); + const responseBody = (await response.json()) as { + accepted: boolean; + pendingFollowUps: number; + }; + assert.equal(responseBody.accepted, true); + assert.equal(responseBody.pendingFollowUps, 0); } finally { resolvePrompt(); await host.stop(); @@ -835,7 +843,9 @@ test("returns an exact receipt for a turn-bound cancellation", async () => { function testRuntime( cwd: string, - sendPrompt: WebRuntimeController["sendPrompt"] = async () => {}, + sendPrompt: WebRuntimeController["sendPrompt"] = async () => ({ + pendingFollowUps: 0, + }), ) { const sessionManager = SessionManager.inMemory(cwd); const runtime: WebRuntimeController = { @@ -1026,7 +1036,7 @@ async function readEventRecords(response: Response, count: number) { let buffer = ""; const records: Array<{ id: number; - event: { sequence: number; type: string }; + event: { sequence: number; type: string; detail?: Record }; }> = []; while (records.length < count) { const chunk = await reader.read(); @@ -1046,7 +1056,11 @@ async function readEventRecords(response: Response, count: number) { if (!id || !data) continue; records.push({ id: Number(id), - event: JSON.parse(data) as { sequence: number; type: string }, + event: JSON.parse(data) as { + sequence: number; + type: string; + detail?: Record; + }, }); } } @@ -1085,6 +1099,66 @@ test("rejects prompt admission with the runtime's typed receipt", async () => { } }); +test("returns and publishes the observed follow-up queue receipt", async () => { + const cwd = await mkdtemp(join(tmpdir(), "openpi-web-prompt-queue-")); + const runtime = testRuntime(cwd, async () => ({ + pendingFollowUps: 2, + })); + const { host, launched, headers } = await startTestHost(runtime); + try { + const snapshot = (await ( + await fetch(`${launched.origin}/api/snapshot`, { headers }) + ).json()) as { cursor: number }; + const response = await fetch(`${launched.origin}/api/prompt`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify({ + sessionId: runtime.sessionManager.getSessionId(), + content: "queue me", + }), + }); + assert.equal(response.status, 202); + const receipt = (await response.json()) as { + id: string; + accepted: boolean; + state: string; + pendingFollowUps: number; + cursor: number; + }; + assert.match(receipt.id, /^[0-9a-f-]{36}$/u); + assert.deepEqual( + { ...receipt, id: undefined }, + { + id: undefined, + accepted: true, + state: "accepted", + pendingFollowUps: 2, + cursor: snapshot.cursor + 1, + }, + ); + const events = await readEventRecords( + await fetch(`${launched.origin}/events?cursor=${snapshot.cursor}`, { + headers, + }), + 1, + ); + assert.equal(events[0].event.sequence, snapshot.cursor + 1); + assert.equal(events[0].event.type, "prompt_accepted"); + assert.match(String(events[0].event.detail?.commandId), /^[0-9a-f-]{36}$/u); + assert.deepEqual( + { ...events[0].event.detail, commandId: undefined }, + { + commandId: undefined, + sessionId: runtime.sessionManager.getSessionId(), + pendingFollowUps: 2, + }, + ); + } finally { + await host.stop(); + await rm(cwd, { recursive: true, force: true }); + } +}); + test("replays only events after an exact SSE cursor with event ids", async () => { const cwd = await mkdtemp(join(tmpdir(), "openpi-web-sse-")); const { host, launched, headers } = await startTestHost(testRuntime(cwd)); @@ -1420,6 +1494,7 @@ test("stop rejects a late keepalive mutation before it enters the drain", async const runtime = testRuntime(cwd, async () => { promptStarted(); await promptBarrier; + return { pendingFollowUps: 0 }; }); runtime.dispose = async () => { releasePrompt(); @@ -1604,6 +1679,7 @@ test("stop disposes the runtime before waiting for an in-flight prompt request", const runtime = testRuntime(cwd, async () => { promptStarted(); await pendingPrompt; + return { pendingFollowUps: 0 }; }); runtime.dispose = async () => { disposeCalls++; diff --git a/web/host/web-host.ts b/web/host/web-host.ts index be0cd38f..c06ec666 100644 --- a/web/host/web-host.ts +++ b/web/host/web-host.ts @@ -544,8 +544,9 @@ export class WebHost { sessionId: body.sessionId, chars: content.length, }); + let admission: { pendingFollowUps: number }; try { - await this.runtime.sendPrompt(content, { + admission = await this.runtime.sendPrompt(content, { commandId, expectedSessionId: body.sessionId, }); @@ -565,7 +566,11 @@ export class WebHost { error: failure.error, }); } - this.publish("prompt_accepted", { commandId, sessionId: body.sessionId }); + this.publish("prompt_accepted", { + commandId, + sessionId: body.sessionId, + pendingFollowUps: admission.pendingFollowUps, + }); traceWeb("prompt_response_sent", { commandId, sessionId: body.sessionId, @@ -575,6 +580,7 @@ export class WebHost { id: commandId, accepted: true, state: "accepted", + pendingFollowUps: admission.pendingFollowUps, cursor: this.sequence, }); } diff --git a/web/runtime/pi-runtime.ts b/web/runtime/pi-runtime.ts index b1cb8293..55a8ebc4 100644 --- a/web/runtime/pi-runtime.ts +++ b/web/runtime/pi-runtime.ts @@ -18,6 +18,7 @@ import { type WebActiveTurn, type WebModelSelectionOptions, type WebPromptOptions, + type WebPromptAdmissionReceipt, type WebRuntimeController, type WebRuntimeEvent, type WebSessionCreationOptions, @@ -410,24 +411,25 @@ export class PiWebRuntime implements WebRuntimeController { releaseAdmission = resolveAdmission; }); const startedAt = performance.now(); - const queued = session.isStreaming; const promptTrace: PromptTrace | undefined = options?.commandId ? { commandId: options.commandId, sessionId, startedAt, started: false, - queued, + queued: false, userMessageObserved: false, } : undefined; this.retainRuntimeReference(agentRuntime); - let resolveRequest: () => void = () => undefined; + let resolveRequest: (receipt: WebPromptAdmissionReceipt) => void = () => undefined; let rejectRequest: (error: unknown) => void = () => undefined; - const requestAdmission = new Promise((resolveRequestAdmission, reject) => { - resolveRequest = resolveRequestAdmission; - rejectRequest = reject; - }); + const requestAdmission = new Promise( + (resolveRequestAdmission, reject) => { + resolveRequest = resolveRequestAdmission; + rejectRequest = reject; + }, + ); const operation = (async () => { let preflightObserved = false; let admitted = false; @@ -455,14 +457,15 @@ export class PiWebRuntime implements WebRuntimeController { elapsedMs: elapsed(startedAt), }); } - const pendingMessagesBefore = session.pendingMessageCount; + let followUpMessages = session.getFollowUpMessages().length; unsubscribePromptLifecycle = session.subscribe((event) => { if (event.type === "agent_start") agentLifecycleStarted = true; - if ( - event.type === "queue_update" && - event.steering.length + event.followUp.length > pendingMessagesBefore - ) { - queuedForAgent = true; + if (event.type === "queue_update") { + if (event.followUp.length > followUpMessages) { + queuedForAgent = true; + if (promptTrace) promptTrace.queued = true; + } + followUpMessages = event.followUp.length; } }); await session.prompt(content, { @@ -487,7 +490,9 @@ export class PiWebRuntime implements WebRuntimeController { ); } if (accepted) { - resolveRequest(); + resolveRequest({ + pendingFollowUps: session.getFollowUpMessages().length, + }); } else { rejectRequest( new WebRuntimeRequestError( @@ -576,7 +581,7 @@ export class PiWebRuntime implements WebRuntimeController { () => this.promptOperations.delete(operation), () => this.promptOperations.delete(operation), ); - await requestAdmission; + return await requestAdmission; } newSession(workspacePath: string, options?: WebSessionCreationOptions) { diff --git a/web/runtime/types.ts b/web/runtime/types.ts index 6bc8b9ee..33272786 100644 --- a/web/runtime/types.ts +++ b/web/runtime/types.ts @@ -33,6 +33,10 @@ export interface WebPromptOptions { expectedSessionId?: string; } +export interface WebPromptAdmissionReceipt { + pendingFollowUps: number; +} + export interface WebActiveTurn { sessionId: string; commandId: string; @@ -75,7 +79,10 @@ export interface WebRuntimeController { readonly sessionManager: SessionManager; isIdle(): boolean; getActiveTurn(): WebActiveTurn | undefined; - sendPrompt(content: string, options?: WebPromptOptions): Promise; + sendPrompt( + content: string, + options?: WebPromptOptions, + ): Promise; cancelTurn( options: WebTurnCancellationOptions, ): Promise; diff --git a/web/ui/app.js b/web/ui/app.js index cf8dd64c..442ea25b 100644 --- a/web/ui/app.js +++ b/web/ui/app.js @@ -33,6 +33,7 @@ const state = { snapshotGeneration: 0, livePhase: "idle", liveRetry: null, + pendingFollowUpsReceipt: null, themePreference: "system", query: "", selectedWorkspace: null, @@ -71,6 +72,7 @@ const translations = { enterHint: "Enter to send, Shift+Enter for a new line.", activeOnlyHint: "Only the active Web session accepts messages.", acceptedHint: "Message accepted by OpenPI Web.", + pendingFollowUpsHint: "Message received; {count} follow-up messages were waiting when it was received.", stopTurn: "Stop turn", stoppingTurn: "Stopping current turn...", stoppedTurn: "Current turn stopped.", @@ -110,6 +112,7 @@ const translations = { enterHint: "按 Enter 发送，Shift+Enter 换行。", activeOnlyHint: "只有当前 Web 会话可以接收消息。", acceptedHint: "OpenPI Web 已接收消息。", + pendingFollowUpsHint: "消息已接收；接收时有 {count} 条后续消息等待处理。", stopTurn: "停止当前回合", stoppingTurn: "正在停止当前回合...", stoppedTurn: "当前回合已停止。", @@ -572,12 +575,21 @@ function updateComposer() { : active ? t("promptMessage") : t("promptReadonly"); - $("composer-hint").textContent = + const composerHint = $("composer-hint"); + composerHint.classList.toggle("receipt", state.pendingFollowUpsReceipt > 0); + composerHint.textContent = state.turnCancellationPending ? t("stoppingTurn") : state.turnTerminalStatus === "cancelled" ? t("stoppedTurn") - : canCompose + : state.pendingFollowUpsReceipt !== null + ? state.pendingFollowUpsReceipt > 0 + ? t("pendingFollowUpsHint").replace( + "{count}", + String(state.pendingFollowUpsReceipt), + ) + : t("acceptedHint") + : canCompose ? state.snapshot.runtime.status === "running" || state.liveRunning ? t("queuedHint") : t("enterHint") @@ -791,6 +803,7 @@ async function sendPrompt() { ].slice(-8); state.promptAdmissionPending = true; state.promptAdmissionToken = admissionToken; + state.pendingFollowUpsReceipt = null; state.turnTerminalStatus = null; renderConversation(); $("composer-hint").classList.remove("error"); @@ -802,6 +815,7 @@ async function sendPrompt() { if (epoch !== state.sessionEpoch || state.promptAdmissionToken !== admissionToken) return; const alreadySettled = state.terminalPromptIds.has(receipt.id); applyPromptAcceptedState(alreadySettled); + state.pendingFollowUpsReceipt = receipt.pendingFollowUps; $("prompt-input").value = ""; resizePrompt(); $("composer-hint").textContent = t("acceptedHint"); @@ -1150,6 +1164,9 @@ function applyRuntimeEvent(event) { } else if (event.type === "prompt_accepted") { const alreadySettled = state.terminalPromptIds.has(event.detail?.commandId); applyPromptAcceptedState(alreadySettled); + state.pendingFollowUpsReceipt = Number.isInteger(event.detail?.pendingFollowUps) + ? event.detail.pendingFollowUps + : state.pendingFollowUpsReceipt; state.liveRetry = null; renderConversation(); } else if (event.type === "turn_started") { @@ -1184,6 +1201,7 @@ function applyRuntimeEvent(event) { } renderConversation(); } else if (event.type === "agent_settled") { + state.pendingFollowUpsReceipt = null; if (!state.activeTurn) { state.liveRunning = false; state.livePhase = "idle"; @@ -1287,6 +1305,7 @@ function resetLiveState() { state.turnTerminalStatus = null; state.livePhase = "idle"; state.liveRetry = null; + state.pendingFollowUpsReceipt = null; } async function connectEvents() { diff --git a/web/ui/styles.css b/web/ui/styles.css index c3634c9d..f4c8f4ec 100644 --- a/web/ui/styles.css +++ b/web/ui/styles.css @@ -582,6 +582,7 @@ body.sidebar-collapsed .icon-button { width: 36px; height: 36px; } .stop-button[hidden] { display: none; } .stop-button svg { width: 16px; height: 16px; fill: currentColor; stroke: none; } .composer-hint { display: none; } +.composer-hint.receipt { display: block; color: var(--subtle); } .composer-hint.error { color: var(--error); } .sidebar-scrim { display: none; }