api/src/nodes-bridge.test.ts

@@ -2,21 +2,30 @@ import { describe, it, expect } from 'vitest'
 import { createNodeBridge } from './nodes'
 import type { WsEvent } from './types'
 
-/** A bridge whose node subscription is driven by the returned `emit`. */
+/** A bridge whose node subscription is driven by the returned `emit`/`status`. */
-function setup() {
+function setup(pingUp = true) {
   let emit: (e: WsEvent) => void = () => {}
+  let status: (online: boolean) => void = () => {}
   const collective: WsEvent[] = []
   const bridge = createNodeBridge({
     broadcast: (e) => collective.push(e),
-    ping: async () => true,
+    ping: async () => pingUp,
-    subscribe: (_n, onEvent) => {
+    subscribe: (_n, onEvent, onStatus) => {
       emit = onEvent
+      status = onStatus
       return () => {}
     },
   })
-  return { bridge, collective, emit: (e: WsEvent) => emit(e) }
+  return {
+    bridge,
+    collective,
+    emit: (e: WsEvent) => emit(e),
+    status: (online: boolean) => status(online),
+  }
 }
 
+const statusEvents = (feed: WsEvent[]) => feed.filter((e) => e.type === 'node:status')
+
 const flash: WsEvent = { type: 'node:activity', teamId: 't1', kind: 'flash', label: 'Flashed to 0x80F0000', ts: 'T' }
 
 describe('createNodeBridge — per-team activity', () => {
@@ -50,3 +59,53 @@ describe('createNodeBridge — per-team activity', () => {
     expect(collective).toContainEqual(flash) // admin feed still gets it
   })
 })
+
+describe('createNodeBridge — self-healing status', () => {
+  it('re-emits node:status offline on a drop and online on reconnect', async () => {
+    const { bridge, collective, status } = setup()
+    const seen: WsEvent[] = []
+    bridge.onTeamActivity('t1', (e) => seen.push(e))
+    await bridge.register({ teamId: 't1', url: 'u', token: 'k' }) // online (ping)
+
+    status(false) // subscription dropped
+    status(true) // reconnected
+
+    expect(statusEvents(collective)).toEqual([
+      { type: 'node:status', teamId: 't1', online: true },
+      { type: 'node:status', teamId: 't1', online: false },
+      { type: 'node:status', teamId: 't1', online: true },
+    ])
+    // participants get the same recovery signal, not just the collective
+    expect(statusEvents(seen)).toContainEqual({ type: 'node:status', teamId: 't1', online: true })
+    expect(statusEvents(seen)).toContainEqual({ type: 'node:status', teamId: 't1', online: false })
+    // list() reflects the latest state
+    expect(bridge.list()).toEqual([{ teamId: 't1', url: 'u', online: true }])
+  })
+
+  it('dedups repeated status of the same value (quiet while a board stays down)', async () => {
+    const { bridge, collective, status } = setup()
+    await bridge.register({ teamId: 't1', url: 'u', token: 'k' }) // online
+
+    status(false)
+    status(false) // repeated failed reconnect attempts — no new event
+    status(false)
+
+    expect(statusEvents(collective)).toEqual([
+      { type: 'node:status', teamId: 't1', online: true },
+      { type: 'node:status', teamId: 't1', online: false },
+    ])
+  })
+
+  it('promotes a node that was offline at register once its stream connects', async () => {
+    const { bridge, collective, status } = setup(false) // ping says offline
+    await bridge.register({ teamId: 't1', url: 'u', token: 'k' })
+    expect(bridge.list()).toEqual([{ teamId: 't1', url: 'u', online: false }])
+
+    status(true) // the reconnect loop got through
+    expect(bridge.list()).toEqual([{ teamId: 't1', url: 'u', online: true }])
+    expect(statusEvents(collective)).toEqual([
+      { type: 'node:status', teamId: 't1', online: false },
+      { type: 'node:status', teamId: 't1', online: true },
+    ])
+  })
+})

api/src/nodes.ts

@@ -116,49 +116,82 @@ export async function sendPrompt(node: NodeRef, message: string, agent = 'defaul
   })
 }
 
+export interface SubscribeOptions {
+  /** Aborts the whole reconnect loop when fired. */
+  signal?: AbortSignal
+  /** Called with `true` on each successful connect, `false` when it drops/fails. */
+  onStatus?: (online: boolean) => void
+  minBackoffMs?: number
+  maxBackoffMs?: number
+}
+
+/** Resolve after `ms`, or immediately if `signal` aborts first. */
+function sleep(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve) => {
+    const t = setTimeout(resolve, ms)
+    signal.addEventListener('abort', () => { clearTimeout(t); resolve() }, { once: true })
+  })
+}
+
 /**
  * Subscribe to a node's `/api/events` SSE stream, mapping each event to a
- * WsEvent via `mapNodeEvent` and handing it to `onEvent`. Returns a stop fn.
+ * WsEvent via `mapNodeEvent` and handing it to `onEvent`. Self-healing: the
+ * stream ends whenever the board drops or its daemon restarts, so we reconnect
+ * with exponential backoff until aborted — otherwise the participant feed goes
+ * permanently blind after the first disconnect. `onStatus` reports each
+ * connect/drop so the bridge can track `online` and re-emit node:status.
+ * Returns a stop fn.
  */
 export function subscribeNodeEvents(
   node: NodeRef,
   onEvent: (e: WsEvent) => void,
-  signal?: AbortSignal,
+  opts: SubscribeOptions = {},
 ): () => void {
   const ctl = new AbortController()
-  if (signal) signal.addEventListener('abort', () => ctl.abort())
+  if (opts.signal) opts.signal.addEventListener('abort', () => ctl.abort())
+  const minBackoff = opts.minBackoffMs ?? 1000
+  const maxBackoff = opts.maxBackoffMs ?? 15000
 
   ;(async () => {
-    try {
+    let backoff = minBackoff
-      const res = await fetch(`${node.url}/api/events`, {
+    while (!ctl.signal.aborted) {
-        headers: { authorization: `Bearer ${node.token}` },
+      try {
-        signal: ctl.signal,
+        const res = await fetch(`${node.url}/api/events`, {
-      })
+          headers: { authorization: `Bearer ${node.token}` },
-      if (!res.body) return
+          signal: ctl.signal,
-      const reader = res.body.getReader()
+        })
-      const decoder = new TextDecoder()
+        if (!res.ok || !res.body) throw new Error(`events ${res.status}`)
-      let buf = ''
+        opts.onStatus?.(true)
-      for (;;) {
+        backoff = minBackoff // healthy connection — reset backoff
-        const { value, done } = await reader.read()
+        const reader = res.body.getReader()
-        if (done) break
+        const decoder = new TextDecoder()
-        buf += decoder.decode(value, { stream: true })
+        let buf = ''
-        let nl: number
+        for (;;) {
-        while ((nl = buf.indexOf('\n')) >= 0) {
+          const { value, done } = await reader.read()
-          const line = buf.slice(0, nl).trim()
+          if (done) break
-          buf = buf.slice(nl + 1)
+          buf += decoder.decode(value, { stream: true })
-          if (!line.startsWith('data:')) continue
+          let nl: number
-          const payload = line.slice(5).trim()
+          while ((nl = buf.indexOf('\n')) >= 0) {
-          if (!payload) continue
+            const line = buf.slice(0, nl).trim()
-          try {
+            buf = buf.slice(nl + 1)
-            const mapped = mapNodeEvent(node.teamId, JSON.parse(payload))
+            if (!line.startsWith('data:')) continue
-            if (mapped) onEvent(mapped)
+            const payload = line.slice(5).trim()
-          } catch {
+            if (!payload) continue
-            /* skip malformed line */
+            try {
+              const mapped = mapNodeEvent(node.teamId, JSON.parse(payload))
+              if (mapped) onEvent(mapped)
+            } catch {
+              /* skip malformed line */
+            }
           }
         }
+      } catch {
+        /* connect failed or the stream dropped — fall through to reconnect */
       }
-    } catch {
+      if (ctl.signal.aborted) break
-      /* stream ended or aborted */
+      opts.onStatus?.(false)
+      await sleep(backoff, ctl.signal)
+      backoff = Math.min(backoff * 2, maxBackoff)
     }
   })()
 
@@ -190,7 +223,7 @@ export interface NodeBridgeDeps {
   /** Injectable for tests. */
   ping?: (n: NodeRef) => Promise<boolean>
   send?: (n: NodeRef, m: string, agent?: string) => Promise<void>
-  subscribe?: (n: NodeRef, onEvent: (e: WsEvent) => void) => () => void
+  subscribe?: (n: NodeRef, onEvent: (e: WsEvent) => void, onStatus: (online: boolean) => void) => () => void
 }
 
 /**
@@ -203,7 +236,7 @@ export function createNodeBridge(deps: NodeBridgeDeps): NodeBridge {
   const registry = deps.registry ?? createNodeRegistry()
   const ping = deps.ping ?? pingNode
   const send = deps.send ?? sendPrompt
-  const subscribe = deps.subscribe ?? ((n, on) => subscribeNodeEvents(n, on))
+  const subscribe = deps.subscribe ?? ((n, on, onStatus) => subscribeNodeEvents(n, on, { onStatus }))
   const online = new Map<string, boolean>()
   const stops = new Map<string, () => void>()
   const teamListeners = new Map<string, Set<(e: WsEvent) => void>>()
@@ -213,14 +246,23 @@ export function createNodeBridge(deps: NodeBridgeDeps): NodeBridge {
     teamListeners.get(teamId)?.forEach((l) => l(e)) // that team's participant(s)
   }
 
+  // Emit node:status only on a real change — the reconnect loop reports every
+  // connect/drop, so dedup keeps the feed quiet while a board stays down.
+  const setStatus = (teamId: string, up: boolean) => {
+    if (online.get(teamId) === up) return
+    online.set(teamId, up)
+    fanOut(teamId, { type: 'node:status', teamId, online: up })
+  }
+
   return {
     async register(ref) {
       stops.get(ref.teamId)?.()
       registry.register(ref)
-      const up = await ping(ref)
+      setStatus(ref.teamId, await ping(ref))
-      online.set(ref.teamId, up)
+      stops.set(
-      fanOut(ref.teamId, { type: 'node:status', teamId: ref.teamId, online: up })
+        ref.teamId,
-      stops.set(ref.teamId, subscribe(ref, (e) => fanOut(ref.teamId, e)))
+        subscribe(ref, (e) => fanOut(ref.teamId, e), (up) => setStatus(ref.teamId, up)),
+      )
     },
     remove(teamId) {
       stops.get(teamId)?.()