fix(web): report native follow-up queue snapshots

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<ReturnType<typeof response>>();

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: () => [],

tests/web/pi-runtime.test.ts

@@ -55,13 +55,35 @@ function promptSession(sessionId: string) {
     options: PromptOptions;
     run: ReturnType<typeof deferred>;
   }> = [];
+  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<void> => 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);

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<string, unknown> };
   }> = [];
   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<string, unknown>;
+        },
       });
     }
   }
@@ -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++;

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,
       });
     }