refactor(agent): route all SSE streams through a shared seam with a required heartbeat

CLAUDE.md

@@ -159,6 +159,8 @@ Unified table using TanStack Table v8 (headless) + CSS Grid rows. Key files: `Da
 
 Separate Bun package that runs as a sidecar container alongside Docker hosts. Provides a REST/SSE API for Docker management operations (deploy, logs, stats streaming). Uses raw `Bun.serve()` with manual route matching and timing-safe auth middleware (zero framework dependencies beyond Dockerode). The agent replaces direct Docker API calls from the worker; the main app communicates with agents rather than Docker hosts directly.
 
+Every agent SSE route (`stats`, `logs`, `containers-events`, `zfs`) builds its `Response` through `agent/src/lib/sse-stream.ts`, which owns the headers, initial flush, 5s comment heartbeat, frame grammar, and abort/enqueue-failure teardown. It is a deliberate copy of the web app's `src/lib/sse/create-sse-stream.ts`: the agent is not a workspace member and cannot import web code, so the seam exists once on each side of the split. Routes that own a subprocess or subscription return a cleanup from `onStart` (run exactly once at teardown) and must register it before any long-running loop, otherwise a teardown mid-loop never reaches it.
+
 The agent is intentionally NOT a workspace member of the homelab-manager `package.json`: its `agent/bun.lock` is the only lockfile the docker build (`context: ./agent`) sees, and workspace membership would mask drift by routing local `bun install` to the homelab-manager lockfile. Web/worker import only types from the agent via the TS path alias `@homelab-manager/agent/*` (resolved at compile time, no runtime dependency). Run `bun run setup` for a full install.
 
 ### Authentication (`src/lib/auth/`)

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

@@ -51,6 +51,33 @@ async function readUntil(
   return text;
 }
 
+/** Parse the `data:` frames out of SSE text, ignoring the seam's flush and heartbeat comments. */
+function parseDataFrames(text: string): any[] {
+  return text
+    .split('\n\n')
+    .filter((frame) => frame.startsWith('data: '))
+    .map((frame) => JSON.parse(frame.slice('data: '.length)));
+}
+
+function countDataFrames(text: string): number {
+  return text.split('\n\n').filter((frame) => frame.startsWith('data: ')).length;
+}
+
+/**
+ * A docker events stream whose teardown is awaitable. The broadcaster destroys
+ * it once its last subscriber unsubscribes, which is how a test observes that a
+ * disconnected client released its subscription.
+ */
+function makeDestroyableEventsStream(): { stream: EventEmitter; destroyed: Promise<void> } {
+  const stream = new EventEmitter();
+  let markDestroyed = () => {};
+  const destroyed = new Promise<void>((resolve) => {
+    markDestroyed = resolve;
+  });
+  Object.assign(stream, { destroy: mock(() => markDestroyed()) });
+  return { stream, destroyed };
+}
+
 function makeContainer(id: string, name: string, state = 'running', image = 'nginx:latest') {
   return {
     Id: id,
@@ -118,7 +145,7 @@ describe('handleContainerEvents: init snapshot', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     expect(event.op).toBe('init');
     expect(event.containers).toHaveLength(2);
     expect(event.containers.map((c: { id: string }) => c.id).sort()).toEqual(['c1', 'c2'].sort((a, b) => a.localeCompare(b)));
@@ -138,7 +165,7 @@ describe('handleContainerEvents: init snapshot', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     expect(event.containers).toHaveLength(2);
   });
 
@@ -153,7 +180,7 @@ describe('handleContainerEvents: init snapshot', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     const c = event.containers[0];
     expect(c.id).toBe('abc123');
     expect(c.name).toBe('my-app');
@@ -176,7 +203,7 @@ describe('handleContainerEvents: init snapshot', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     expect(event.containers[0].name).toBe('test-app');
   });
 
@@ -191,7 +218,7 @@ describe('handleContainerEvents: init snapshot', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     expect(event.containers[0].state).toBe('unknown');
   });
 
@@ -205,7 +232,7 @@ describe('handleContainerEvents: init snapshot', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     expect(event.containers).toHaveLength(0);
   });
 });
@@ -231,12 +258,11 @@ describe('handleContainerEvents: start event produces upsert', () => {
     }) + '\n'));
 
     const text = await readUntil(response, (s) => {
-      const events = s.split('\n\n').filter(Boolean);
-      return events.length >= 2;
+      return countDataFrames(s) >= 2;
     });
     ac.abort();
 
-    const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, '')));
+    const events = parseDataFrames(text);
     const upsert = events.find((e) => e.op === 'upsert');
     expect(upsert).toBeDefined();
     expect(upsert.container.id).toBe('c1');
@@ -275,12 +301,11 @@ describe('handleContainerEvents: die event produces upsert with exited state', (
     }) + '\n'));
 
     const text = await readUntil(response, (s) => {
-      const events = s.split('\n\n').filter(Boolean);
-      return events.length >= 2;
+      return countDataFrames(s) >= 2;
     });
     ac.abort();
 
-    const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, '')));
+    const events = parseDataFrames(text);
     const upsert = events.find((e) => e.op === 'upsert');
     expect(upsert).toBeDefined();
     expect(upsert.container.id).toBe('c1');
@@ -307,12 +332,11 @@ describe('handleContainerEvents: destroy event', () => {
     }) + '\n'));
 
     const text = await readUntil(response, (s) => {
-      const events = s.split('\n\n').filter(Boolean);
-      return events.length >= 2;
+      return countDataFrames(s) >= 2;
     });
     ac.abort();
 
-    const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, '')));
+    const events = parseDataFrames(text);
     const destroy = events.find((e) => e.op === 'destroy');
     expect(destroy).toBeDefined();
     expect(destroy.containerId).toBe('c1');
@@ -420,27 +444,32 @@ describe('handleContainerEvents: request abort cleanup', () => {
     expect(done).toBe(true);
   });
 
-  test('already-aborted request does not register a broadcaster subscriber', async () => {
-    const docker = makeDocker([]);
+  test('already-aborted request unsubscribes the broadcaster once subscribe resolves', async () => {
+    const { stream, destroyed } = makeDestroyableEventsStream();
+    const docker = makeDocker([], stream);
     const ac = new AbortController();
     ac.abort();
     const request = new Request('http://localhost/containers/events', { signal: ac.signal });
     const response = await handleContainerEvents(docker as any, request);
 
     const reader = response.body!.getReader();
-    const result = await reader.read();
-    expect(result.done).toBe(true);
-    expect(docker.getEvents).not.toHaveBeenCalled();
+    let done = false;
+    while (!done) {
+      done = (await reader.read()).done;
+    }
+
+    await destroyed;
   });
 
   test('abort during broadcasterSubscribe tears down the late-arriving subscriber', async () => {
     // Slow listContainers so subscribe()'s await is in-flight when we abort.
     const listHolder: { resolve: ((v: unknown[]) => void) | null } = { resolve: null };
+    const { stream, destroyed } = makeDestroyableEventsStream();
     const docker = {
       listContainers: mock(() => new Promise<unknown[]>((resolve) => {
         listHolder.resolve = resolve;
       })),
-      getEvents: mock(() => Promise.resolve(new EventEmitter())),
+      getEvents: mock(() => Promise.resolve(stream)),
       getContainer: mock(() => ({
         inspect: mock(() => Promise.reject(Object.assign(new Error('n/a'), { statusCode: 404 }))),
       })),
@@ -455,8 +484,11 @@ describe('handleContainerEvents: request abort cleanup', () => {
     ac.abort();
     listHolder.resolve?.([]);
 
-    const result = await reader.read();
-    expect(result.done).toBe(true);
+    let done = false;
+    while (!done) {
+      done = (await reader.read()).done;
+    }
+    await destroyed;
   });
 });
 
@@ -512,7 +544,7 @@ describe('handleContainerEvents: ports and mounts pass-through', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     const c = event.containers[0];
     expect(c.ports).toEqual([{ containerPort: 80, protocol: 'tcp', hostIp: '0.0.0.0', hostPort: 8080 }]);
     expect(c.mounts).toEqual([{ type: 'bind', source: '/host', destination: '/data', rw: true }]);
@@ -529,7 +561,7 @@ describe('handleContainerEvents: ports and mounts pass-through', () => {
     const text = await readUntil(response, (s) => s.includes('"op":"init"'));
     ac.abort();
 
-    const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, ''));
+    const event = parseDataFrames(text)[0];
     const c = event.containers[0];
     expect(c.ports).toEqual([]);
     expect(c.mounts).toEqual([]);
@@ -569,12 +601,11 @@ describe('handleContainerEvents: ports and mounts pass-through', () => {
     }) + '\n'));
 
     const text = await readUntil(response, (s) => {
-      const events = s.split('\n\n').filter(Boolean);
-      return events.length >= 2;
+      return countDataFrames(s) >= 2;
     });
     ac.abort();
 
-    const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, '')));
+    const events = parseDataFrames(text);
     const upsert = events.find((e) => e.op === 'upsert');
     expect(upsert.container.ports).toEqual([{ containerPort: 53, protocol: 'udp', hostIp: '::', hostPort: 5353 }]);
     expect(upsert.container.mounts).toEqual([{ type: 'volume', source: 'vol1', destination: '/var/data', rw: true }]);
@@ -612,55 +643,32 @@ describe('zInventorySnapshotContainer: ports/mounts schema round-trip', () => {
 
 describe('handleContainerEvents: idle heartbeat', () => {
   const realSetInterval = globalThis.setInterval;
-  const realClearInterval = globalThis.clearInterval;
 
   afterEach(() => {
     globalThis.setInterval = realSetInterval;
-    globalThis.clearInterval = realClearInterval;
   });
 
-  test('enqueues a comment heartbeat to keep the idle socket alive', async () => {
-    // A quiet host emits no container events, so without this the socket sits
-    // silent past Bun's 10s HTTP idleTimeout and the worker reconnects in a loop.
-    let captured: (() => void) | null = null;
-    globalThis.setInterval = mock((cb: () => void) => {
-      captured = cb;
-      return 1 as unknown as ReturnType<typeof setInterval>;
+  test('a quiet host still gets a comment heartbeat on the 5s cadence', async () => {
+    // Bun's HTTP idleTimeout defaults to 10s, so a host with no container
+    // activity would otherwise go silent long enough to drop the socket.
+    const ticks: Array<{ cb: () => void; ms: number }> = [];
+    globalThis.setInterval = mock((cb: () => void, ms: number) => {
+      ticks.push({ cb, ms });
+      return ticks.length as unknown as ReturnType<typeof setInterval>;
     }) as unknown as typeof setInterval;
 
     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);
 
-    // Let start() finish broadcaster setup and register the interval.
-    await new Promise((r) => setTimeout(r, 50));
-    expect(captured).not.toBeNull();
+    expect(ticks).toHaveLength(1);
+    expect(ticks[0].ms).toBe(5000);
 
-    captured!();
-    const text = await readUntil(response, (s) => s.includes(':\n\n'));
+    ticks[0].cb();
+    const text = await readUntil(response, (s) => s.split('\n\n').includes(':'));
     ac.abort();
 
-    // Init frame plus a bare ':' comment heartbeat.
     expect(text.split('\n\n')).toContain(':');
   });
-
-  test('clears the heartbeat interval when the request aborts', async () => {
-    const fakeId = Symbol('hb') as unknown as ReturnType<typeof setInterval>;
-    globalThis.setInterval = mock(() => fakeId) as unknown as typeof setInterval;
-    const clearSpy = mock(() => {});
-    globalThis.clearInterval = clearSpy as unknown as typeof clearInterval;
-
-    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);
-
-    await new Promise((r) => setTimeout(r, 50)); // let the interval register
-    ac.abort();
-    await new Promise((r) => setTimeout(r, 10));
-
-    expect(clearSpy).toHaveBeenCalledWith(fakeId);
-    void response;
-  });
 });

agent/src/__tests__/logs.test.ts

@@ -381,9 +381,11 @@ describe('handleLogStream', () => {
     const originalSetInterval = globalThis.setInterval;
     const originalClearInterval = globalThis.clearInterval;
     let heartbeatCb: (() => void) | null = null;
+    const heartbeatDelays: number[] = [];
     const fakeTimerId = 999;
-    globalThis.setInterval = ((cb: () => void) => {
+    globalThis.setInterval = ((cb: () => void, ms: number) => {
       heartbeatCb = cb;
+      heartbeatDelays.push(ms);
       return fakeTimerId as unknown as ReturnType<typeof setInterval>;
     }) as typeof setInterval;
     globalThis.clearInterval = mock(() => {}) as typeof clearInterval;
@@ -399,6 +401,7 @@ describe('handleLogStream', () => {
 
       // Fire the heartbeat callback
       expect(heartbeatCb).not.toBeNull();
+      expect(heartbeatDelays).toEqual([5000]);
       heartbeatCb!();
 
       liveEmitter.emit('end');

agent/src/__tests__/stats.test.ts

@@ -623,4 +623,33 @@ describe('handleStatsStream: container refresh', () => {
     expect(text).toContain('"type":"refresh_failed"');
     expect(text).toContain('"error":"Docker daemon down"');
   });
+
+  test('emits a comment heartbeat when the host runs no containers', async () => {
+    // With nothing running there are no stats frames at all, so the stream would
+    // otherwise sit silent past Bun's 10s HTTP idleTimeout.
+    const realSetInterval = globalThis.setInterval;
+    const ticks: Array<{ cb: () => void; ms: number }> = [];
+    globalThis.setInterval = mock((cb: () => void, ms: number) => {
+      ticks.push({ cb, ms });
+      return ticks.length as unknown as ReturnType<typeof setInterval>;
+    }) as unknown as typeof setInterval;
+
+    try {
+      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, fastOptions);
+
+      expect(ticks).toHaveLength(1);
+      expect(ticks[0].ms).toBe(5000);
+
+      ticks[0].cb();
+      const text = await readUntil(response, (s) => s.split('\n\n').includes(':'));
+      ac.abort();
+
+      expect(text.split('\n\n')).toContain(':');
+    } finally {
+      globalThis.setInterval = realSetInterval;
+    }
+  });
 });