feat(feed): notify browsers of canonical inventory changes

src/feed.mjs

@@ -237,6 +237,53 @@ export function createFeed({
   let stale = false;
   let errors = [];
   let scanning = null;
+  let scanAgain = false;
+  let catchupTimer = null;
+  let started = false;
+  let revision = 0;
+  let published = "";
+  const listeners = new Set();
+
+  function publish() {
+    // Observation timestamps advance during fallback polling. Only a changed
+    // public snapshot needs to wake browsers; their clocks already tick locally.
+    const value = snapshot();
+    const signature = JSON.stringify({ ...value, checkedAt: undefined });
+    if (signature === published) return;
+    published = signature;
+    revision++;
+    for (const listener of listeners) {
+      try { listener(revision); } catch { /* One closed client cannot block the feed. */ }
+    }
+  }
+
+  function scan({ latest = false } = {}) {
+    if (scanning) {
+      if (latest) scanAgain = true;
+      return scanning;
+    }
+    if (catchupTimer !== null) { clearTimeout(catchupTimer); catchupTimer = null; }
+    scanning = (async () => {
+      try {
+        for (let pass = 0; pass < 2; pass++) {
+          scanAgain = false;
+          await doScan();
+          if (!scanAgain) break;
+        }
+        return snapshot();
+      } finally {
+        scanning = null;
+        if (scanAgain && started && catchupTimer === null) {
+          catchupTimer = setTimeout(() => {
+            catchupTimer = null;
+            if (started) void scan({ latest: true }).catch(() => {});
+          }, 0);
+          catchupTimer.unref?.();
+        }
+      }
+    })();
+    return scanning;
+  }
 
   async function doScan() {
     const nextErrors = [];
@@ -381,6 +428,7 @@ export function createFeed({
     errors = nextErrors;
     stale = errors.length > 0;
     if (!stale) lastSuccessAt = now();
+    publish();
     return snapshot();
   }
 
@@ -419,9 +467,23 @@ export function createFeed({
 
   return {
     snapshot,
-    scan() {
-      if (!scanning) scanning = doScan().finally(() => { scanning = null; });
-      return scanning;
+    scan,
+    subscribe(listener) {
+      listeners.add(listener);
+      return () => listeners.delete(listener);
+    },
+    start() {
+      if (started) return;
+      started = true;
+      if (hubInventory.configured) hubInventory.start?.(() => {
+        if (started) void scan({ latest: true }).catch(() => {});
+      });
+    },
+    stop() {
+      started = false;
+      scanAgain = false;
+      if (catchupTimer !== null) { clearTimeout(catchupTimer); catchupTimer = null; }
+      hubInventory.stop?.();
     },
     async flushEnrichment() {
       if (hubInventory.configured || typeof enrich.flush !== "function") return snapshot();
@@ -472,13 +534,14 @@ function isSameOriginRequest(req) {
 }
 
 export function createFeedServer(feed, { directory = HERE, messages = createHubMessenger() } = {}) {
-  return createServer(async (req, res) => {
+  const streams = new Set();
+  const server = createServer(async (req, res) => {
     const headers = { "cache-control": "no-store" };
     const json = (status, body) => { res.writeHead(status, { ...headers, "content-type": "application/json" }); res.end(JSON.stringify(body)); };
     try {
       const url = new URL(req.url, "http://localhost");
       const messageRoute = url.pathname.match(/^\/agents\/([^/]+)\/messages$/);
-      const isAgentRoute = url.pathname === "/agents" || /^\/agents\/[^/]+\/activity$/.test(url.pathname);
+      const isAgentRoute = ["/agents", "/agents/events"].includes(url.pathname) || /^\/agents\/[^/]+\/activity$/.test(url.pathname);
       if (isAgentRoute && !isSameOriginRequest(req)) return json(403, { error: "Cross-origin agent access is not allowed" });
       if (req.method === "OPTIONS") { res.writeHead(204, { ...headers, "access-control-allow-methods": "GET, POST, OPTIONS" }); return res.end(); }
       if (req.method === "POST" && messageRoute) {
@@ -503,6 +566,26 @@ export function createFeedServer(feed, { directory = HERE, messages = createHubM
         return json(sent.status, sent.body);
       }
       if (req.method !== "GET" && req.method !== "HEAD") return json(405, { error: "Read-only endpoint" });
+      if (url.pathname === "/agents/events") {
+        if (typeof feed.subscribe !== "function") return json(404, { error: "Live notifications unavailable" });
+        res.writeHead(200, { ...headers, "content-type": "text/event-stream", "x-accel-buffering": "no" });
+        if (req.method === "HEAD") return res.end();
+        res.flushHeaders();
+        streams.add(res);
+        const notify = revision => {
+          // Notifications contain no Hub payload. Clients read the committed,
+          // normalized /agents snapshot, including its messaging capabilities.
+          if (!res.destroyed && !res.write(`event: change\ndata: ${JSON.stringify({ revision })}\n\n`)) res.destroy();
+        };
+        const unsubscribe = feed.subscribe(notify);
+        const heartbeat = setInterval(() => {
+          if (!res.destroyed && !res.write(": heartbeat\n\n")) res.destroy();
+        }, 15000);
+        heartbeat.unref?.();
+        res.once("close", () => { clearInterval(heartbeat); unsubscribe(); streams.delete(res); });
+        notify(null); // Every connection resynchronizes, including after restart.
+        return;
+      }
       if (url.pathname === "/agents") {
         const snapshot = feed.snapshot();
         return json(200, { ...snapshot, agents: snapshot.agents.map(agent => ({ ...agent, conversation: messages.capability(agent, { stale: !!snapshot.stale, checkedAt: snapshot.checkedAt }) })) });
@@ -564,6 +647,14 @@ export function createFeedServer(feed, { directory = HERE, messages = createHubM
       else res.end();
     }
   });
+  server.on("listening", () => feed.start?.());
+  server.on("close", () => feed.stop?.());
+  const close = server.close;
+  server.close = function (...args) {
+    for (const response of streams) response.end();
+    return close.apply(this, args);
+  };
+  return server;
 }
 
 async function main() {

src/live-feed.mjs

@@ -61,3 +61,40 @@ export function createLatestRefresh(run) {
     return active;
   };
 }
+
+/** Push wakeups are optional and belong only to the bundled same-origin feed. */
+export function createFeedNotifications({
+  origin,
+  EventSourceClass = globalThis.EventSource,
+  onChange = () => {},
+} = {}) {
+  let source = null, current = null, generation = 0;
+  function stop() {
+    generation++;
+    source?.close();
+    source = null;
+    current = null;
+  }
+  return {
+    sync(endpoint, { enabled = true } = {}) {
+      let url = null;
+      try {
+        const candidate = new URL(endpoint, origin);
+        if (enabled && endpoint && candidate.origin === origin && candidate.pathname === "/agents" &&
+          !candidate.search && !candidate.hash && !candidate.username && !candidate.password) url = candidate;
+      } catch { /* Custom feeds retain ordinary JSON polling. */ }
+      if (!url || typeof EventSourceClass !== "function") { stop(); return; }
+      if (source && current === url.href) return;
+      stop();
+      const epoch = generation;
+      current = url.href;
+      try {
+        source = new EventSourceClass(new URL("/agents/events", origin).href);
+        source.addEventListener("change", () => {
+          if (epoch === generation && source) onChange();
+        });
+      } catch { stop(); } // Polling is still active when SSE is unsupported.
+    },
+    stop,
+  };
+}

test/feed-stream.test.mjs

@@ -0,0 +1,105 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import { once } from "node:events";
+import { createFeed, createFeedServer } from "../src/feed.mjs";
+
+const inventory = agents => ({ ok: true, agents, keys: new Set(agents.map(a => a.id)), links: new Map() });
+const task = status => ({ id: "child", taskId: "child", sessionId: "session", source: "hub", town: "HubTown", name: "Child", status, parent: "parent" });
+
+test("stream invalidation during a scan gets another pass and publishes committed state", async () => {
+  let reads = 0, release, changed, stopped = 0;
+  const gate = new Promise(resolve => { release = resolve; });
+  const feed = createFeed({ hubInventory: {
+    configured: true,
+    start(callback) { changed = callback; },
+    stop() { stopped++; },
+    async read() { reads++; if (reads === 1) { await gate; return inventory([task("working")]); } return inventory([task("done")]); },
+  } });
+  const observed = [];
+  const unsubscribe = feed.subscribe(() => observed.push(feed.snapshot().agents[0].status));
+  feed.start();
+  const scanning = feed.scan();
+  await Promise.resolve();
+  changed();
+  changed();
+  release();
+  await scanning;
+  assert.equal(reads, 2, "event bursts coalesce but never disappear behind an active scan");
+  assert.deepEqual(observed, ["working", "done"], "callbacks see the committed snapshot");
+  unsubscribe(); feed.stop();
+  assert.equal(stopped, 1);
+});
+
+test("continuous invalidation yields after two passes and a later scan catches up", async t => {
+  let reads=0,changed,release;const gate=new Promise(resolve=>{release=resolve;});
+  const feed=createFeed({hubInventory:{configured:true,start(callback){changed=callback;},stop(){},async read(){reads++;if(reads<=2)changed();if(reads===3)await gate;return inventory([task(reads>=3?'done':'working')]);}}});
+  feed.start();t.after(()=>{release();feed.stop();});
+  const first=feed.scan(),joined=feed.scan();assert.equal(first,joined,'ordinary callers share the active promise');
+  let resolved=false;const settled=first.then(()=>{resolved=true;});
+  await new Promise(resolve=>setTimeout(resolve,25));
+  const resolvedBeforeCatchup=resolved;
+  release();await settled;
+  assert.equal(resolvedBeforeCatchup,true,'the shared scan resolves after its two-pass budget');
+  await new Promise(resolve=>setTimeout(resolve,10));
+  assert.equal(reads,3,'the coalesced catch-up pass runs after yielding');
+  assert.equal(feed.snapshot().agents[0].status,'done');
+});
+
+test("a fresh scan coalesces pending catch-up and stop cancels it and ignores invalidations", async t => {
+  let reads=0,invalidations=2,changed;const feed=createFeed({hubInventory:{configured:true,start(callback){changed=callback;},stop(){},async read(){reads++;if(invalidations>0){invalidations--;changed();}return inventory([task('working')]);}}});
+  feed.start();t.after(()=>feed.stop());
+  await feed.scan();assert.equal(reads,2);
+  await feed.scan();assert.equal(reads,3,'an explicit fresh scan consumes the dirty state');
+  await new Promise(resolve=>setTimeout(resolve,10));assert.equal(reads,3,'the superseded timer does not duplicate the fresh scan');
+  // Create another dirty scan, then stop before its queued catch-up can run.
+  invalidations=2;changed();await feed.scan();
+  const beforeStop=reads;feed.stop();changed();
+  await new Promise(resolve=>setTimeout(resolve,10));assert.equal(reads,beforeStop,'stopped feeds ignore callbacks and pending work');
+});
+
+test("message POST completes while invalidations continue during its shared scan", async t => {
+  let reads=0,changed,release,thirdReadStarted,sends=0;const gate=new Promise(resolve=>{release=resolve;}),thirdRead=new Promise(resolve=>{thirdReadStarted=resolve;});
+  const feed=createFeed({hubInventory:{configured:true,start(callback){changed=callback;},stop(){},async read(){reads++;if(reads<=2)changed();if(reads===3){thirdReadStarted();await gate;}return inventory([task('working')]);}}});
+  feed.start();const server=createFeedServer(feed,{messages:{async send(){sends++;return {status:200,body:{ok:true,delivery:'accepted'}};}}}).listen(0,'127.0.0.1');
+  await once(server,'listening');t.after(async()=>{release();feed.stop();server.closeAllConnections();await new Promise(resolve=>server.close(resolve));});
+  const base=`http://127.0.0.1:${server.address().port}`;
+  const request=fetch(`${base}/agents/child/messages`,{method:'POST',headers:{origin:base,'content-type':'application/json','x-cottagecode-request':'user-message'},body:JSON.stringify({taskId:'child',message:'stub test',requestId:'stub-request'})});
+  await thirdRead;
+  let timeout;
+  try {
+    const result=await Promise.race([request.then(response=>response.status),new Promise(resolve=>{timeout=setTimeout(()=>resolve('pending'),1500);})]);
+    assert.equal(result,200,'POST returns while the coalesced catch-up read is still gated');
+  } finally { clearTimeout(timeout);release(); }
+  const response=await request;assert.equal(response.status,200);
+  assert.equal(sends,1);
+});
+
+test("local SSE wakes after commit, exposes only revision, and rejects foreign origins", async t => {
+  let status = "working", stamp = 1_800_000_000_000;
+  const feed = createFeed({ now: () => stamp, hubInventory: { configured: true, read: async () => inventory([task(status)]) } });
+  await feed.scan();
+  const server = createFeedServer(feed).listen(0, "127.0.0.1");
+  await once(server, "listening");
+  const base = `http://127.0.0.1:${server.address().port}`;
+  t.after(() => new Promise(resolve => { server.close(resolve); server.closeAllConnections(); }));
+  assert.equal((await fetch(base + "/agents/events", { headers: { origin: "https://foreign.example" } })).status, 403);
+  const controller = new AbortController();
+  t.after(() => controller.abort());
+  const response = await fetch(base + "/agents/events", { signal: controller.signal, headers: { origin: base } });
+  assert.match(response.headers.get("content-type"), /text\/event-stream/);
+  const reader = response.body.getReader(), decode = new TextDecoder();
+  assert.equal(decode.decode((await reader.read()).value), 'event: change\ndata: {"revision":null}\n\n');
+  let notifications = 0;
+  const off = feed.subscribe(() => notifications++);
+  stamp += 2000;
+  await feed.scan();
+  assert.equal(notifications, 0, "unchanged fallback scan does not publish checkedAt churn");
+  status = "done";
+  const next = reader.read();
+  await feed.scan();
+  const frame = decode.decode((await next).value);
+  assert.match(frame, /^event: change\ndata: \{"revision":\d+\}\n\n$/);
+  assert.equal((await (await fetch(base + "/agents")).json()).agents[0].status, "done");
+  assert.doesNotMatch(frame, /taskId|session|authorization|parent/);
+  off(); await reader.cancel();
+});

test/live-feed.test.mjs

@@ -1,6 +1,6 @@
 import test from "node:test";
 import assert from "node:assert/strict";
-import { blankFeedMeta, blankFeedState, createLatestRefresh, readCurrentFeed, snapshotForEndpoint } from "../src/live-feed.mjs";
+import { blankFeedMeta, blankFeedState, createLatestRefresh, createFeedNotifications, readCurrentFeed, snapshotForEndpoint } from "../src/live-feed.mjs";
 
 function response(body, status = 200) {
   return { ok: status >= 200 && status < 300, status, json: async () => body };
@@ -65,3 +65,41 @@ test("a demo fallback clears a prior endpoint snapshot and metadata", () => {
     "a later local-feed failure must not restore real cottages into the demo namespace");
   assert.notEqual(fallback.meta, prior.meta);
 });
+
+test("push wakeups follow only the current bundled feed and stop for pause/demo", () => {
+  const sources = [];
+  class EventSourceStub {
+    constructor(url) { this.url = url; this.closed = false; sources.push(this); }
+    addEventListener(name, callback) { assert.equal(name, "change"); this.notify = callback; }
+    close() { this.closed = true; }
+  }
+  let updates = 0;
+  const origin = "http://127.0.0.1:8787";
+  const notifications = createFeedNotifications({ origin, EventSourceClass: EventSourceStub, onChange: () => updates++ });
+  notifications.sync(origin + "/agents");
+  notifications.sync(origin + "/agents");
+  assert.equal(sources.length, 1, "one subscription per active endpoint");
+  assert.equal(sources[0].url, origin + "/agents/events");
+  sources[0].notify();
+  assert.equal(updates, 1);
+  notifications.sync(origin + "/agents", { enabled: false });
+  sources[0].notify();
+  assert.equal(updates, 1, "queued notifications cannot update a paused feed");
+  assert.equal(sources[0].closed, true);
+  notifications.sync(origin + "/agents");
+  sources[1].notify();
+  assert.equal(updates, 2);
+  for (const endpoint of ["https://custom.example/agents", origin + "/custom.json", null, origin + "/agents?source=other"]) {
+    notifications.sync(endpoint);
+    sources[1].notify();
+  }
+  assert.equal(updates, 2, "old connections cannot cross endpoint or demo boundaries");
+  assert.equal(sources.length, 2, "custom feeds are not assumed to support SSE");
+});
+
+test("an unavailable notification transport leaves JSON polling usable", async () => {
+  const notifications = createFeedNotifications({ origin: "http://localhost", EventSourceClass: class { constructor() { throw new Error("Unavailable"); } } });
+  assert.doesNotThrow(() => notifications.sync("http://localhost/agents"));
+  const result = await readCurrentFeed("http://localhost/agents", { fetchFn: async () => response({ agents: [{ id: "still-polling" }] }) });
+  assert.equal(result.kind, "success");
+});