fix(web): recover from quiet or stalled connections

tests/web/app-render.test.ts

@@ -348,6 +348,7 @@ async function renderApp(
     URL,
     TextDecoder,
     TextEncoder,
+    AbortController,
     Element: class Element {},
   };
   context.window = {
@@ -401,6 +402,17 @@ async function renderApp(
       "sendPrompt",
       context as vm.Context,
     ) as () => Promise<void>,
+    api: vm.runInContext("api", context as vm.Context) as (
+      path: string,
+      options?: Record<string, unknown>,
+    ) => Promise<unknown>,
+    readEventChunk: vm.runInContext(
+      "readEventChunk",
+      context as vm.Context,
+    ) as (
+      reader: { read(): Promise<unknown> },
+      timeoutMs?: number,
+    ) => Promise<unknown>,
     updateComposer: vm.runInContext(
       "updateComposer",
       context as vm.Context,
@@ -463,6 +475,49 @@ test("app.js resnapshots on an SSE cursor gap", async () => {
   assert.ok(app.readerCancellations() >= 1);
 });
 
+test("app.js uses quiet-stream heartbeats for bounded snapshot recovery", async () => {
+  const heartbeat = ": heartbeat\n\n";
+  const app = await renderApp({
+    eventRecords: [heartbeat.repeat(4)],
+  });
+  assert.equal(app.eventFetches(), 1);
+  assert.ok(app.snapshotFetches() >= 2);
+  assert.equal(app.state.cursor, SNAPSHOT.cursor);
+});
+
+test("app.js bounds API waits and explains duplicate prompt admission", async () => {
+  const app = await renderApp();
+  app.context.fetch = async (
+    _url: unknown,
+    options?: { signal?: AbortSignal },
+  ) =>
+    new Promise((_resolve, reject) => {
+      options?.signal?.addEventListener("abort", () => {
+        const error = new Error("aborted");
+        error.name = "AbortError";
+        reject(error);
+      });
+    });
+  await assert.rejects(
+    app.api("/api/stuck", { timeoutMs: 5, timeoutMessage: "bounded timeout" }),
+    /bounded timeout/u,
+  );
+  await assert.rejects(
+    app.readEventChunk({ read: () => new Promise(() => {}) }, 5),
+    /event stream stalled/u,
+  );
+
+  app.state.promptAdmissionPending = true;
+  const input = app.elements.get("prompt-input");
+  assert.ok(input);
+  input.value = "another message";
+  await app.sendPrompt();
+  assert.equal(
+    app.elements.get("composer-hint")?.textContent,
+    "OpenPI is still accepting the previous message.",
+  );
+});
+
 test("app.js invalidates snapshots for cross-tab session metadata events", async () => {
   const app = await renderApp({
     eventRecords: [

tests/web/web-host.test.ts

@@ -697,6 +697,46 @@ async function startTestHost(runtime: WebRuntimeController) {
   return { host, launched, headers };
 }
 
+test("quiet SSE clients receive heartbeats without advancing the event cursor", async () => {
+  const cwd = await mkdtemp(join(tmpdir(), "openpi-web-heartbeat-"));
+  const host = new WebHost({
+    runtime: testRuntime(cwd),
+    sseHeartbeatMs: 10,
+  });
+  try {
+    await host.start();
+    const launched = new URL(host.url);
+    const token = new URLSearchParams(launched.hash.slice(1)).get("token");
+    assert.ok(token);
+    const headers = { Authorization: `Bearer ${token}` };
+    const before = (await (
+      await fetch(`${launched.origin}/api/snapshot`, { headers })
+    ).json()) as { cursor: number };
+    const response = await fetch(
+      `${launched.origin}/events?cursor=${before.cursor}`,
+      { headers },
+    );
+    assert.equal(response.status, 200);
+    assert.ok(response.body);
+    const reader = response.body.getReader();
+    const decoder = new TextDecoder();
+    let received = "";
+    while (!received.includes(": heartbeat\n\n")) {
+      const chunk = await reader.read();
+      assert.equal(chunk.done, false);
+      received += decoder.decode(chunk.value, { stream: true });
+    }
+    const after = (await (
+      await fetch(`${launched.origin}/api/snapshot`, { headers })
+    ).json()) as { cursor: number };
+    assert.equal(after.cursor, before.cursor);
+    await reader.cancel();
+  } finally {
+    await host.stop();
+    await rm(cwd, { recursive: true, force: true });
+  }
+});
+
 test("adapter initialization fails before the Host starts listening", async () => {
   const cwd = await mkdtemp(join(tmpdir(), "openpi-web-startup-failure-"));
   const runtime = testRuntime(cwd);

web/host/web-host.ts

@@ -33,6 +33,7 @@ const MAX_COMMAND_BYTES = 16 * 1024;
 const MAX_SSE_CLIENTS = 8;
 const MAX_SSE_BUFFER_BYTES = 256 * 1024;
 const MAX_SSE_REPLAY_BYTES = MAX_SSE_BUFFER_BYTES;
+const DEFAULT_SSE_HEARTBEAT_MS = 15_000;
 const SERVER_CLOSE_DRAIN_MS = 500;
 const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000;
 const execFileAsync = promisify(execFile);
@@ -45,13 +46,18 @@ export interface WebHostOptions {
   allowedOrigins?: readonly string[];
   directoryChooser?: (signal: AbortSignal) => Promise<string | undefined>;
   shutdownTimeoutMs?: number;
+  sseHeartbeatMs?: number;
 }
 
 export class WebHost {
   private readonly server: Server;
   private readonly token: Buffer;
   private readonly adapter: PiWebAdapter;
   private readonly clients = new Set<ServerResponse>();
+  private readonly clientHeartbeats = new Map<
+    ServerResponse,
+    ReturnType<typeof setInterval>
+  >();
   private readonly events: WebEvent[] = [];
   private sequence = 0;
   private port = 0;
@@ -63,6 +69,7 @@ export class WebHost {
     WebHostOptions["directoryChooser"]
   >;
   private readonly shutdownTimeoutMs: number;
+  private readonly sseHeartbeatMs: number;
   private readonly unsubscribeCapabilities: () => void;
   private readonly unsubscribeRuntime: () => void;
   private readonly chooserAbort = new AbortController();
@@ -90,6 +97,14 @@ export class WebHost {
     ) {
       throw new Error("Web host shutdown timeout must be a positive integer");
     }
+    this.sseHeartbeatMs =
+      options.sseHeartbeatMs ?? DEFAULT_SSE_HEARTBEAT_MS;
+    if (
+      !Number.isSafeInteger(this.sseHeartbeatMs) ||
+      this.sseHeartbeatMs <= 0
+    ) {
+      throw new Error("SSE heartbeat interval must be a positive integer");
+    }
     this.adapter = new PiWebAdapter(options.runtime);
     this.onEvent = options.onEvent;
     this.unsubscribeCapabilities = subscribeWebCapabilities((scope) => {
@@ -185,8 +200,7 @@ export class WebHost {
         client.writableLength > MAX_SSE_BUFFER_BYTES ||
         !client.write(record)
       ) {
-        this.clients.delete(client);
-        client.destroy();
+        this.removeSseClient(client, "destroy");
       }
     }
     this.onEvent?.(event.type, event.detail);
@@ -204,8 +218,7 @@ export class WebHost {
       this.unsubscribeCapabilities();
       this.unsubscribeRuntime();
       this.chooserAbort.abort();
-      for (const client of this.clients) client.end();
-      this.clients.clear();
+      for (const client of [...this.clients]) this.removeSseClient(client, "end");
       const closeServer = this.server.listening
         ? new Promise<void>((resolve) => {
             const forceClose = setTimeout(
@@ -736,7 +749,31 @@ export class WebHost {
     // ordering without treating normal backpressure as a broken client.
     for (const record of replay) response.write(record);
     this.clients.add(response);
-    response.on("close", () => this.clients.delete(response));
+    const heartbeat = setInterval(() => {
+      if (
+        response.destroyed ||
+        response.writableEnded ||
+        response.writableLength > MAX_SSE_BUFFER_BYTES ||
+        !response.write(": heartbeat\n\n")
+      ) {
+        this.removeSseClient(response, "destroy");
+      }
+    }, this.sseHeartbeatMs);
+    heartbeat.unref();
+    this.clientHeartbeats.set(response, heartbeat);
+    response.on("close", () => this.removeSseClient(response));
+  }
+
+  private removeSseClient(
+    response: ServerResponse,
+    close?: "destroy" | "end",
+  ) {
+    this.clients.delete(response);
+    const heartbeat = this.clientHeartbeats.get(response);
+    if (heartbeat) clearInterval(heartbeat);
+    this.clientHeartbeats.delete(response);
+    if (close === "destroy" && !response.destroyed) response.destroy();
+    else if (close === "end" && !response.writableEnded) response.end();
   }
 
   private parseCursor(value: string | undefined | null) {