From 65e47486be9f98dbbe26985b84d033c2074da75f Mon Sep 17 00:00:00 2001 From: thekannman Date: Thu, 18 Jun 2026 10:01:16 -0500 Subject: [PATCH] Desktop: broker owns the /events SSE live mirror (Phase 3 complete) The broker now serves the /events live-mirror stream itself, instead of forwarding it -- the last read/mirror piece. Mirrors dashboard_app.event_stream: a snapshot on connect (plate, inflight, pending, queue, runs -- each a BARE array, the SSE shape, using the Node store + liveness), a fresh snapshot whenever SQLite's data_version moves OR a file/ liveness signal changes, and a 10s heartbeat. Holds one read-only connection so data_version is comparable across polls; tears down on client disconnect. Verified against the live FastAPI: the initial snapshot frames (all five events) match byte-for-byte, and adding a task re-emits an updated plate frame (change detection works). - desktop/sse.js (+ tests): snapshot(), signals() (run/session liveness + pending/queue mtimes -- the changes a DB write wouldn't move), createEventStream(). - desktop/broker.js: GET /events is token-gated then streamed (before the JSON SERVED map). With this the broker owns the WHOLE live mirror (every read + the stream). Still forwarded: settings (env-detected terminal) and the action/write endpoints (Phase 4, next). Co-Authored-By: Claude Opus 4.8 (1M context) --- desktop/broker.js | 15 ++++++ desktop/package.json | 2 +- desktop/sse.js | 122 +++++++++++++++++++++++++++++++++++++++++++ desktop/sse.test.js | 56 ++++++++++++++++++++ 4 files changed, 194 insertions(+), 1 deletion(-) create mode 100644 desktop/sse.js create mode 100644 desktop/sse.test.js diff --git a/desktop/broker.js b/desktop/broker.js index 3e99c0f..389f2d7 100644 --- a/desktop/broker.js +++ b/desktop/broker.js @@ -13,6 +13,7 @@ const http = require('http') const crypto = require('crypto') const store = require('./store') const liveness = require('./liveness') +const sse = require('./sse') const { token } = require('./resolve') // GET endpoints the broker answers itself, in FastAPI's response shape (parity-tested against the @@ -74,6 +75,20 @@ function createBroker({ targetHost = '127.0.0.1', targetPort, sopDir }) { // Serve a static read directly from the store (Phase 3). The broker owns the token gate for // these, since FastAPI's check never runs on a broker-served response. const pathname = req.url.split('?')[0] + // The /events live-mirror stream is served by the broker (Phase 3 complete): token-gated, then + // a long-lived SSE response instead of a JSON body. + if (req.method === 'GET' && pathname === '/events' && sopDir) { + const t = new URL(req.url, 'http://x').searchParams.get('t') + if (!tokenOk(t, token({ SOP_DIR: sopDir }))) { + res.writeHead(401, { 'content-type': 'application/json' }) + res.end(JSON.stringify({ detail: 'bad or missing token' })) + return + } + try { sse.createEventStream(req, res, sopDir) } catch (_) { + try { if (!res.headersSent) res.writeHead(500); res.end() } catch (_) { /* already closed */ } + } + return + } const serve = req.method === 'GET' && sopDir ? SERVED[pathname] : undefined if (serve) { const t = new URL(req.url, 'http://x').searchParams.get('t') diff --git a/desktop/package.json b/desktop/package.json index 3b1037b..9d8cda1 100644 --- a/desktop/package.json +++ b/desktop/package.json @@ -6,7 +6,7 @@ "main": "main.js", "scripts": { "start": "electron .", - "check": "node --check main.js && node --check preload.js && node --check resolve.js && node --check broker.js && node --check store.js && node --check liveness.js", + "check": "node --check main.js && node --check preload.js && node --check resolve.js && node --check broker.js && node --check store.js && node --check liveness.js && node --check sse.js", "test": "node --test" }, "devDependencies": { diff --git a/desktop/sse.js b/desktop/sse.js new file mode 100644 index 0000000..a8e9826 --- /dev/null +++ b/desktop/sse.js @@ -0,0 +1,122 @@ +// The /events live-mirror stream, owned by the broker (completes Phase 3: the live mirror moves off +// FastAPI). Mirrors dashboard_app.event_stream: a snapshot on connect, a fresh snapshot whenever the +// DB changes (SQLite PRAGMA data_version on a held connection) OR a file/liveness signal changes, plus +// a heartbeat. Stops on client disconnect. + +const { DatabaseSync } = require('node:sqlite') +const fs = require('fs') +const path = require('path') +const store = require('./store') +const liveness = require('./liveness') + +const POLL_MS = (() => { const v = Number(process.env.SMBOS_SSE_POLL); return Number.isFinite(v) && v > 0 ? v * 1000 : 1000 })() +const HEARTBEAT_MS = (() => { const v = Number(process.env.SMBOS_SSE_HEARTBEAT); return Number.isFinite(v) && v > 0 ? v * 1000 : 10000 })() + +function sse(event, payload) { + return `event: ${event}\ndata: ${payload}\n\n` +} + +// The five live-mirror frames, each the BARE array (dashboard_app._snapshot uses json.dumps(list), +// not a {key: list} wrapper -- the SSE shape differs from the GET endpoints' wrapped shape). +function snapshot(sopDir) { + return [ + sse('plate', JSON.stringify(store.plate(sopDir))), + sse('inflight', JSON.stringify(liveness.inflightWithLiveness(store, sopDir))), + sse('pending', JSON.stringify(store.pending(sopDir))), + sse('queue', JSON.stringify(store.queue(sopDir))), + sse('runs', JSON.stringify(liveness.runsWithLiveness(store, sopDir))), + ] +} + +// [name, mtime-ns] over dir/*.md, sorted, tolerant of a file vanishing between readdir and stat. +// Mirrors dashboard_app._dir_mtime_sig: pending/ and queue/ change with no DB write, so data_version +// misses them. Nanosecond mtime (like Python st_mtime_ns) catches two sub-millisecond rewrites. +function dirSig(dir) { + const sig = [] + let names + try { names = fs.readdirSync(dir) } catch (_) { return [] } + for (const f of names) { + if (!f.endsWith('.md')) continue + try { sig.push([f, String(fs.statSync(path.join(dir, f), { bigint: true }).mtimeNs)]) } catch (_) { /* vanished */ } + } + return sig.sort((a, b) => (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0)) +} + +// Change signals that a DB write wouldn't move: run liveness, session liveness, and the file-based +// pending/ + queue/ dirs. verify=false keeps the once-a-second poll cheap (no ps fork) -- it catches +// the common pid-gone flip; the rendered frame does the full check. Mirrors the _signals() tuple. +// JSON-encoded structured values (not delimiter-joined) so an id/filename can't collide on a separator. +function signals(sopDir) { + return JSON.stringify([ + liveness.activeRuns(sopDir, false).map((r) => [r.sop, r.state]).sort(), + liveness.inflightWithLiveness(store, sopDir, false).map((t) => [t.id, t.state]), + dirSig(path.join(sopDir, 'pending')), + dirSig(path.join(sopDir, 'queue')), + ]) +} + +// Run the SSE loop on an already-Host-guarded, token-checked GET /events request. +function createEventStream(req, res, sopDir) { + res.writeHead(200, { + 'content-type': 'text/event-stream', + 'cache-control': 'no-cache', + connection: 'keep-alive', + 'x-accel-buffering': 'no', // don't let any proxy buffer the stream + }) + // One held read-only connection so data_version is comparable across polls (it's per-connection). + let db = null + let timer = null + const cleanup = () => { + if (timer) { clearInterval(timer); timer = null } + if (db) { try { db.close() } catch (_) { /* already gone */ } db = null } + } + // Attach cleanup BEFORE the first store/liveness read: if a read throws (corrupt/mid-recreate + // state.db), we must still release the connection + timer and not leak or crash the broker. + res.on('close', cleanup) + res.on('error', cleanup) + + // Open lazily and re-open on error, so a state.db that doesn't exist yet at connect (fresh SOP dir) + // or is recreated mid-stream gets picked up -- otherwise data_version stays 0 forever and DB-only + // changes (a new waiting task) never re-emit until the client reconnects. + const ensureDb = () => { + if (db) return + try { db = new DatabaseSync(path.join(sopDir, 'state.db'), { readOnly: true }); db.exec('PRAGMA busy_timeout = 2000') } catch (_) { db = null } + } + const dataVersion = () => { + ensureDb() + try { return db ? db.prepare('PRAGMA data_version').get().data_version : 0 } catch (_) { db = null; return 0 } + } + + try { + + let lastDv = dataVersion() + let lastSig = signals(sopDir) + for (const frame of snapshot(sopDir)) res.write(frame) // initial snapshot + + let sinceBeat = 0 + timer = setInterval(() => { + // Guard the WHOLE tick: dataVersion, signals, AND the snapshot re-emit. A synchronous store/ + // liveness read error (a mid-stream corrupt/recreating state.db) must not escape the interval + // callback and crash the broker -- skip the tick and try again next poll. + try { + sinceBeat += POLL_MS + const dv = dataVersion() + const sig = signals(sopDir) + if (dv !== lastDv || sig !== lastSig) { + lastDv = dv; lastSig = sig + for (const frame of snapshot(sopDir)) res.write(frame) // all frames on any change + } + if (sinceBeat >= HEARTBEAT_MS) { + sinceBeat = 0 + res.write(sse('heartbeat', JSON.stringify({ ts: new Date().toISOString() }))) + } + } catch (_) { /* transient read error: skip this tick */ } + }, POLL_MS) + } catch (_) { + // the initial snapshot/signals read failed: tear down cleanly instead of leaking + throwing + cleanup() + try { res.end() } catch (_) { /* already closed */ } + } +} + +module.exports = { createEventStream, snapshot, signals } diff --git a/desktop/sse.test.js b/desktop/sse.test.js new file mode 100644 index 0000000..43aab07 --- /dev/null +++ b/desktop/sse.test.js @@ -0,0 +1,56 @@ +const test = require('node:test') +const assert = require('node:assert') +const fs = require('fs') +const os = require('os') +const path = require('path') +const { DatabaseSync } = require('node:sqlite') +const sse = require('./sse') + +function seeded() { + const d = fs.mkdtempSync(path.join(os.tmpdir(), 'smbos-sse-')) + const db = new DatabaseSync(path.join(d, 'state.db')) + db.exec(`CREATE TABLE task (id INTEGER PRIMARY KEY, domain TEXT, kind TEXT, subject TEXT, status TEXT, + priority INTEGER DEFAULT 0, source_ref TEXT, created_at TEXT, updated_at TEXT)`) + db.exec(`CREATE TABLE run (id INTEGER PRIMARY KEY, sop_id TEXT, result TEXT, started_at TEXT)`) + db.exec("INSERT INTO task(id,subject,status,created_at,updated_at) VALUES(1,'waiting one','waiting','t','t')") + db.close() + return d +} + +test('snapshot: five frames, each a BARE array (not a {key: ...} wrapper)', () => { + const d = seeded() + const frames = sse.snapshot(d) + assert.equal(frames.length, 5) + for (const ev of ['plate', 'inflight', 'pending', 'queue', 'runs']) { + const frame = frames.find((f) => f.startsWith(`event: ${ev}\n`)) + assert.ok(frame, `has a ${ev} frame`) + assert.ok(frame.endsWith('\n\n'), 'frame is terminated') + const data = JSON.parse(frame.match(/data: (.*)/)[1]) + assert.ok(Array.isArray(data), `${ev} data is a bare array`) + } + // the seeded waiting task shows up in the plate frame + const plate = JSON.parse(frames.find((f) => f.startsWith('event: plate\n')).match(/data: (.*)/)[1]) + assert.equal(plate[0].subject, 'waiting one') +}) + +test('snapshot is resilient to a missing state.db: five empty-array frames, no throw', () => { + const d = fs.mkdtempSync(path.join(os.tmpdir(), 'smbos-sse-')) // a fresh SOP dir, no state.db yet + const frames = sse.snapshot(d) + assert.equal(frames.length, 5) + for (const ev of ['plate', 'inflight', 'pending', 'queue', 'runs']) { + const data = JSON.parse(frames.find((f) => f.startsWith(`event: ${ev}\n`)).match(/data: (.*)/)[1]) + assert.deepEqual(data, []) // the stream still opens; the held connection reopens when the db appears + } +}) + +test('signals: change when a pending/queue file appears (data_version is blind to files)', () => { + const d = seeded() + const s0 = sse.signals(d) + fs.mkdirSync(path.join(d, 'pending')) + fs.writeFileSync(path.join(d, 'pending', 'x.md'), '---\nstatus: pending\n---\n# X\n') + const s1 = sse.signals(d) + assert.notEqual(s0, s1) // a parked result appearing must move the signal + fs.mkdirSync(path.join(d, 'queue')) + fs.writeFileSync(path.join(d, 'queue', 'q.md'), '---\nstatus: queued\n---\n') + assert.notEqual(s1, sse.signals(d)) // a queued run appearing too +})