diff --git a/src/feed.mjs b/src/feed.mjs index 421a387..bd158a1 100644 --- a/src/feed.mjs +++ b/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() { diff --git a/src/live-feed.mjs b/src/live-feed.mjs index 52a440f..add55c6 100644 --- a/src/live-feed.mjs +++ b/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, + }; +} diff --git a/test/feed-stream.test.mjs b/test/feed-stream.test.mjs new file mode 100644 index 0000000..5dcc61d --- /dev/null +++ b/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(); +}); diff --git a/test/live-feed.test.mjs b/test/live-feed.test.mjs index d00b881..714f37c 100644 --- a/test/live-feed.test.mjs +++ b/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"); +});