fix(sse): add periodic heartbeats to web factories and agent streams

agent/src/__tests__/containers-events.test.ts

@@ -102,6 +102,20 @@ describe('handleContainerEvents: response headers', () => {
   });
 });
 
+describe('handleContainerEvents: heartbeat', () => {
+  test('emits heartbeat comment pings while no container events occur', async () => {
+    const docker = makeDocker([]);
+    const ac = new AbortController();
+    const request = new Request('http://localhost/containers/events', { signal: ac.signal });
+    const response = await handleContainerEvents(docker as any, request, 10);
+
+    const text = await readUntil(response, (s) => s.includes(': ping'));
+    ac.abort();
+
+    expect(text).toContain(': ping\n\n');
+  });
+});
+
 describe('handleContainerEvents: init snapshot', () => {
   test('emits init event with all containers on connect', async () => {
     const containers = [

agent/src/__tests__/stats.test.ts

@@ -211,6 +211,18 @@ describe('handleStatsStream', () => {
     expect(response.headers.get('Connection')).toBe('keep-alive');
   });
 
+  test('emits heartbeat comment pings while the stream is idle', async () => {
+    const mockDocker = { listContainers: mock(() => Promise.resolve([])) };
+    const ac = new AbortController();
+    const request = new Request('http://localhost/stats/stream', { signal: ac.signal });
+    const response = handleStatsStream(mockDocker as any, request, { heartbeatIntervalMs: 10 });
+
+    const text = await readUntil(response, (s) => s.includes(': ping'));
+    ac.abort();
+
+    expect(text).toContain(': ping\n\n');
+  });
+
   test('streams flat computed stats as SSE events', async () => {
     const statsEmitter = new EventEmitter();
 

agent/src/__tests__/zfs.test.ts

@@ -141,6 +141,31 @@ describe('handleZfsStatsStream', () => {
     expect(tankEvent.line).toContain('tank');
   });
 
+  test('emits heartbeat comment pings while iostat is silent', async () => {
+    const killMock = mock(() => {});
+    const readableStream = new ReadableStream<Uint8Array>({
+      start() {
+        // Keep stream open with no output; simulates a silent iostat
+      },
+    });
+
+    Bun.spawn = mock(() => ({
+      stdout: readableStream,
+      stderr: new ReadableStream(),
+      kill: killMock,
+      exited: new Promise(() => {}), // never resolves
+    })) as any;
+
+    const ac = new AbortController();
+    const request = new Request('http://localhost/zfs/stats/stream', { signal: ac.signal });
+    const response = handleZfsStatsStream(request, zfsAvailable, 10);
+
+    const text = await readUntil(response, (s) => s.includes(': ping'));
+    ac.abort();
+
+    expect(text).toContain(': ping\n\n');
+  });
+
   test('kills subprocess on client disconnect', async () => {
     const killMock = mock(() => {});
     const readableStream = new ReadableStream<Uint8Array>({

agent/src/lib/__tests__/sse-utils.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, test, mock, beforeAll, beforeEach, afterAll } from 'bun:test';
-import { sendSSE } from '../sse-utils';
+import { sendSSE, startSseHeartbeat } from '../sse-utils';
 
 const originalConsoleError = console.error;
 
@@ -101,3 +101,52 @@ describe('sendSSE', () => {
     }).toThrow(err);
   });
 });
+
+describe('startSseHeartbeat', () => {
+  test('enqueues comment pings on each interval tick', async () => {
+    const { controller, enqueued } = makeController();
+    const stop = startSseHeartbeat(controller, makeEncoder(), () => false, 5);
+
+    await new Promise((r) => setTimeout(r, 20));
+    stop();
+
+    expect(enqueued.length).toBeGreaterThanOrEqual(1);
+    expect(enqueued[0]).toBe(': ping\n\n');
+  });
+
+  test('stops pinging once isClosed reports true', async () => {
+    const { controller, enqueued } = makeController();
+    let closed = false;
+    const stop = startSseHeartbeat(controller, makeEncoder(), () => closed, 5);
+
+    await new Promise((r) => setTimeout(r, 15));
+    closed = true;
+    await new Promise((r) => setTimeout(r, 15));
+    const countAtClose = enqueued.length;
+    await new Promise((r) => setTimeout(r, 15));
+    stop();
+
+    expect(enqueued.length).toBe(countAtClose);
+  });
+
+  test('clears itself when enqueue throws (controller already closed)', async () => {
+    const { controller, enqueued, makeEnqueueThrow } = makeController();
+    makeEnqueueThrow(new TypeError('Controller is closed'));
+    const stop = startSseHeartbeat(controller, makeEncoder(), () => false, 5);
+
+    await new Promise((r) => setTimeout(r, 30));
+    stop();
+
+    expect(enqueued).toHaveLength(0);
+  });
+
+  test('stop function prevents any further pings', async () => {
+    const { controller, enqueued } = makeController();
+    const stop = startSseHeartbeat(controller, makeEncoder(), () => false, 5);
+    stop();
+
+    await new Promise((r) => setTimeout(r, 20));
+
+    expect(enqueued).toHaveLength(0);
+  });
+});

agent/src/lib/sse-utils.ts

@@ -1,3 +1,44 @@
+/**
+ * Default heartbeat period. Must stay below typical idle timeouts that kill
+ * quiet streams (Bun's idleTimeout, nginx's 60 s proxy_read_timeout).
+ */
+export const SSE_HEARTBEAT_INTERVAL_MS = 25_000;
+
+/**
+ * Start a periodic SSE comment ping (`: ping\n\n`) that keeps proxies and
+ * Bun's idleTimeout from killing streams with no traffic.
+ *
+ * The interval clears itself when `isClosed()` reports true or when enqueue
+ * throws (controller already closed). Callers should still invoke the
+ * returned stop function from their teardown path so the timer dies with the
+ * stream instead of one period later.
+ *
+ * @param controller - Stream controller the ping frames are enqueued on
+ * @param encoder - Shared TextEncoder for the session
+ * @param isClosed - Reports whether the session has been torn down
+ * @param intervalMs - Ping period in milliseconds (default: 25s)
+ * @returns Stop function that clears the interval
+ */
+export function startSseHeartbeat(
+  controller: ReadableStreamDefaultController<Uint8Array>,
+  encoder: TextEncoder,
+  isClosed: () => boolean,
+  intervalMs: number = SSE_HEARTBEAT_INTERVAL_MS,
+): () => void {
+  const timer = setInterval(() => {
+    if (isClosed()) {
+      clearInterval(timer);
+      return;
+    }
+    try {
+      controller.enqueue(encoder.encode(': ping\n\n'));
+    } catch {
+      clearInterval(timer);
+    }
+  }, intervalMs);
+  return () => clearInterval(timer);
+}
+
 /** Enqueue an SSE data event, silently swallowing enqueue-after-close TypeError. */
 export function sendSSE(
   controller: ReadableStreamDefaultController<Uint8Array>,

agent/src/routes/containers-events.ts

@@ -1,7 +1,7 @@
 import type Dockerode from 'dockerode';
 import { subscribe as broadcasterSubscribe } from '../lib/docker-events-broadcaster';
 import type { MinimalContainerInfo, BroadcasterEvent } from '../lib/docker-events-broadcaster';
-import { sendSSE } from '../lib/sse-utils';
+import { sendSSE, startSseHeartbeat } from '../lib/sse-utils';
 import type {
   AgentContainerEvent,
   ContainerState,
@@ -63,16 +63,24 @@ function toUpdateContainer(c: MinimalContainerInfo): InventoryUpdateContainer {
  *
  * @param docker - Dockerode client used to interact with the Docker daemon
  * @param request - The HTTP request; its abort signal triggers cleanup
+ * @param heartbeatIntervalMs - Comment ping period in milliseconds (default: 25s)
  */
-export async function handleContainerEvents(docker: Dockerode, request: Request): Promise<Response> {
+export async function handleContainerEvents(
+  docker: Dockerode,
+  request: Request,
+  heartbeatIntervalMs?: number,
+): Promise<Response> {
   const encoder = new TextEncoder();
   const closed = { value: false };
   let unsubscribe: (() => void) | null = null;
 
   const stream = new ReadableStream<Uint8Array>({
     async start(controller) {
+      let stopHeartbeat: () => void = () => {};
+
       request.signal.addEventListener('abort', () => {
         closed.value = true;
+        stopHeartbeat();
         unsubscribe?.();
         try {
           controller.close();
@@ -87,6 +95,16 @@ export async function handleContainerEvents(docker: Dockerode, request: Request)
         return;
       }
 
+      // Inventory events only fire on container state changes, which can be
+      // hours apart; a comment ping keeps proxies and Bun's idleTimeout from
+      // killing the quiet stream.
+      stopHeartbeat = startSseHeartbeat(
+        controller,
+        encoder,
+        () => closed.value,
+        heartbeatIntervalMs,
+      );
+
       const sub = await broadcasterSubscribe(docker, (event: BroadcasterEvent) => {
         let message: AgentContainerEvent;
 

agent/src/routes/stats.ts

@@ -1,5 +1,6 @@
 import type { Readable } from 'node:stream';
 import type Dockerode from 'dockerode';
+import { startSseHeartbeat } from '../lib/sse-utils';
 
 const DEFAULT_REFRESH_INTERVAL_MS = 60_000;
 const DEFAULT_POLL_INTERVAL_MS = 5_000;
@@ -10,6 +11,8 @@ export interface StatsStreamOptions {
   pollIntervalMs?: number;
   /** Close the stream after this many consecutive refresh failures (default: 10). */
   maxConsecutiveFailures?: number;
+  /** Heartbeat comment period in milliseconds (default: 25s). */
+  heartbeatIntervalMs?: number;
 }
 
 /** Flat computed metrics matching the worker's AgentStatsEvent interface. */
@@ -420,8 +423,18 @@ export function handleStatsStream(
         controller,
       };
 
+      // Stats frames stop flowing when no containers run; a comment ping
+      // keeps proxies and Bun's idleTimeout from killing the quiet stream.
+      const stopHeartbeat = startSseHeartbeat(
+        controller,
+        ctx.encoder,
+        () => ctx.closed,
+        options.heartbeatIntervalMs,
+      );
+
       request.signal.addEventListener('abort', () => {
         ctx.closed = true;
+        stopHeartbeat();
         destroyAllStreams(ctx.containerStreams);
         tryCloseController(controller);
       });
@@ -433,6 +446,8 @@ export function handleStatsStream(
         const msg = error instanceof Error ? error.message : String(error);
         sendSSE(ctx, JSON.stringify({ error: msg }), 'error');
         tryCloseController(controller);
+      } finally {
+        stopHeartbeat();
       }
     },
   });

agent/src/routes/zfs.ts

@@ -1,14 +1,18 @@
 import type { ZfsCapabilities } from '../lib/zfs-capabilities';
+import { startSseHeartbeat } from '../lib/sse-utils';
 
 /**
  * GET /zfs/stats/stream: SSE endpoint that streams `zpool iostat -v 1` output.
  *
  * Each non-empty line from the subprocess is emitted as an SSE event with
  * `{ line, timestamp }`. The subprocess is killed when the client disconnects.
+ *
+ * @param heartbeatIntervalMs - Comment ping period in milliseconds (default: 25s)
  */
 export function handleZfsStatsStream(
   request: Request,
   capabilities: ZfsCapabilities,
+  heartbeatIntervalMs?: number,
 ): Response {
   if (!capabilities.available) {
     return Response.json(
@@ -29,8 +33,19 @@ export function handleZfsStatsStream(
 
         let closed = false;
 
+        // `zpool iostat 1` normally emits every second, but a wedged pool or
+        // stalled subprocess can go silent; a comment ping keeps proxies and
+        // Bun's idleTimeout from killing the connection in the meantime.
+        const stopHeartbeat = startSseHeartbeat(
+          controller,
+          encoder,
+          () => closed,
+          heartbeatIntervalMs,
+        );
+
         request.signal.addEventListener('abort', () => {
           closed = true;
+          stopHeartbeat();
           proc.kill();
           try {
             controller.close();
@@ -71,6 +86,7 @@ export function handleZfsStatsStream(
             console.error('ZFS stats stream error:', err);
           }
         } finally {
+          stopHeartbeat();
           proc.kill();
           if (!closed) {
             closed = true;

src/lib/sse/__tests__/create-broadcast-sse-handler.test.ts

@@ -7,7 +7,10 @@ mock.module('@/lib/auth/sse-auth', () => ({
 
 type Event = { n: number };
 
-function setup(serialize: (e: Event) => string = (e) => `data: ${JSON.stringify(e)}\n\n`) {
+function setup(
+  serialize: (e: Event) => string = (e) => `data: ${JSON.stringify(e)}\n\n`,
+  heartbeatIntervalMs?: number,
+) {
   let captured: ((event: Event) => void) | null = null;
   const unsubscribe = mock(() => {});
   const subscribe = mock((cb: (event: Event) => void) => {
@@ -19,6 +22,7 @@ function setup(serialize: (e: Event) => string = (e) => `data: ${JSON.stringify(
     loadSubscribe: async () => subscribe,
     serialize,
     errorEvent: 'test_error',
+    heartbeatIntervalMs,
   });
 
   return {
@@ -214,6 +218,57 @@ describe('createBroadcastSseHandler', () => {
     ac.abort();
   });
 
+  it('emits periodic comment pings on the heartbeat interval', async () => {
+    const { handler } = setup(undefined, 5);
+    const ac = new AbortController();
+
+    const res = await handler({ request: makeRequest(ac) });
+    const reader = readerOf(res);
+    const decoder = new TextDecoder();
+
+    const first = await reader.read();
+    expect(decoder.decode(first.value)).toBe(': ok\n\n');
+
+    const second = await reader.read();
+    expect(decoder.decode(second.value)).toBe(': ping\n\n');
+
+    ac.abort();
+    reader.cancel();
+  });
+
+  it('tears down when the heartbeat ping hits a dead consumer', async () => {
+    const { handler, unsubscribe } = setup(undefined, 5);
+    const ac = new AbortController();
+
+    const res = await handler({ request: makeRequest(ac) });
+    const reader = readerOf(res);
+    await reader.read(); // initial flush comment
+
+    // Cancelling the reader closes the controller without firing abort, so
+    // the next ping's enqueue throws and must trigger teardown.
+    await reader.cancel();
+    await new Promise((r) => setTimeout(r, 30));
+
+    expect(unsubscribe).toHaveBeenCalledTimes(1);
+
+    ac.abort();
+  });
+
+  it('stops the heartbeat after abort', async () => {
+    const clearSpy = spyOn(globalThis, 'clearInterval');
+    try {
+      const { handler } = setup(undefined, 5);
+      const ac = new AbortController();
+
+      await handler({ request: makeRequest(ac) });
+      ac.abort();
+
+      expect(clearSpy).toHaveBeenCalled();
+    } finally {
+      clearSpy.mockRestore();
+    }
+  });
+
   it('returns 401 when authenticateSSE returns null', async () => {
     const { authenticateSSE } = require('@/lib/auth/sse-auth') as { authenticateSSE: ReturnType<typeof mock> };
     authenticateSSE.mockImplementationOnce(async () => null);