From 78c739390971cf347f4796876ebc110519a22f02 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 2 Oct 2026 17:04:22 +0300 Subject: [PATCH] push architecture slice 1: core event bus + /api/events SSE channel (notify->fetch infra, no consumers yet) --- app/api/events/route.ts | 15 +++++ src/core/events.ts | 86 +++++++++++++++++++++++ src/server/events-stream.ts | 81 ++++++++++++++++++++++ src/server/index.ts | 1 + tests/events.test.ts | 131 ++++++++++++++++++++++++++++++++++++ 5 files changed, 314 insertions(+) create mode 100644 app/api/events/route.ts create mode 100644 src/core/events.ts create mode 100644 src/server/events-stream.ts create mode 100644 tests/events.test.ts diff --git a/app/api/events/route.ts b/app/api/events/route.ts new file mode 100644 index 0000000..68d2579 --- /dev/null +++ b/app/api/events/route.ts @@ -0,0 +1,15 @@ +/** + * THE NOTIFICATION CHANNEL ROUTE (push-architecture slice 1, 2026-10-02): + * one SSE connection per open window over the core event bus — the + * endpoint every future consumer (chat view, sidebar, activity dock) + * subscribes to as the polls retire. The response plumbing lives in the + * server seam (events-stream.ts); the framework route stays a thin + * wrapper (the import law; tests drive the seam directly). + */ +import { eventsResponse } from "@/src/server"; + +export const dynamic = "force-dynamic"; + +export function GET(): Response { + return eventsResponse(); +} diff --git a/src/core/events.ts b/src/core/events.ts new file mode 100644 index 0000000..3a924fd --- /dev/null +++ b/src/core/events.ts @@ -0,0 +1,86 @@ +/** + * THE EVENT BUS (push-architecture slice 1, owner ruling 2026-10-02: go + * directly to the endgame — notify → fetch, no standing polls). + * + * One bus, every layer: server emits turn lifecycle here, the store + * adapters emit job/chat mutations, the agent layer's background + * registries (extraction, practice) emit activity changes — and + * GET /api/events turns every emission into an SSE frame for any open + * window. The events carry the FACT, never the data: consumers fetch + * truth on receipt (push the poke, pull the data), so payloads stay + * tiny, schemaless, and idempotent — a reconnect heals by refetching. + * + * WHY globalThis (the turn-liveness discipline, same rationale): Next + * dev compiles each route as its own module entry, so a module-level + * singleton is a DIFFERENT instance in the page render, the events + * route, and every emitter's entry — they would each hold a private + * listener set and nothing would propagate. globalThis spans module + * instances within the process; it is the standard Next singleton + * channel (see src/server/turn-liveness.ts for the original incident). + * + * Pure core by law: no React, no IO, no imports — the one layer that + * server, store, and agent may all depend on. + */ + +/** The full event vocabulary — fixed from day one (slice plan, 2026-10-02). */ +export interface ServerEvents { + /** A teach turn began server-side (beginTurn) — client-initiated sends, + * wake turns, other-tab turns: every origin rides the same event. */ + "turn-started": { chatId: string }; + /** A turn settled (endTurn — AFTER persistTurn, so any fetch triggered + * by this event reads a durable record; ordering is guaranteed in + * chat-handler's onFinish). */ + "turn-settled": { chatId: string }; + /** A job record changed (dispatch, progress, settle, abort, recovery). + * chatId when known — consumers that own one chat filter on it. */ + "jobs-changed": { chatId?: string }; + /** An extraction or practice run changed phase (activity dock). */ + "activity-changed": Record; + /** The session list changed (create, delete, title). */ + "chats-changed": Record; +} + +export type ServerEventType = keyof ServerEvents; + +/** One notification: the type discriminates the payload. */ +export type ServerEvent = { + type: K; +} & ServerEvents[K]; + +export type ServerEventListener = (event: ServerEvent) => void; + +interface LhEventsGlobal { + __lhEventBus?: Set; +} + +/** The shared listener set — globalThis-backed so every Next module entry + * (and every layer) sees ONE bus. Ephemeral by design: a process restart + * drops the listeners with their connections; emitters never queue. */ +function listeners(): Set { + const g = globalThis as typeof globalThis & LhEventsGlobal; + return (g.__lhEventBus ??= new Set()); +} + +/** Listen. Returns the unsubscribe function — the SSE route's cancel and + * any test cleanup call it; a dropped listener leaks nothing. */ +export function subscribe(listener: ServerEventListener): () => void { + const set = listeners(); + set.add(listener); + return () => set.delete(listener); +} + +/** Notify every listener. Zero subscribers is a no-op (no window open — + * the emission was for the stores' own bookkeeping anyway). One + * listener throwing must never break the others or the emitter: each + * call is isolated, the failure logged, the loop continues. */ +export function emit(type: K, payload: ServerEvents[K]): void { + // Snapshot: a listener that unsubscribes (or subscribes) mid-emit must + // not mutate the set under iteration. + for (const listener of [...listeners()]) { + try { + listener({ type, ...payload } as ServerEvent); + } catch (err) { + console.error(`[events] listener failed on ${type}:`, err); + } + } +} diff --git a/src/server/events-stream.ts b/src/server/events-stream.ts new file mode 100644 index 0000000..6b583c4 --- /dev/null +++ b/src/server/events-stream.ts @@ -0,0 +1,81 @@ +/** + * THE NOTIFICATION CHANNEL (push-architecture slice 1): GET /api/events is + * an SSE response over the core event bus — one connection per open + * window, held open until the view goes away, fed by every emitter + * (turn lifecycle, job/chat stores, extraction/practice registries). + * + * NOTIFY → FETCH (the architecture's whole trick): frames carry only the + * event type + a tiny payload; consumers respond by fetching truth + * (/api/chats/:id, /api/jobs, /api/activity). No payload versioning, no + * replay protocol — a reconnect resyncs with one ordinary fetch per + * consumer. This is a DESKTOP app talking to localhost: holding N≈1-3 + * connections costs one socket pair each and zero traffic while idle + * (owner ruling 2026-10-02 — polling's cloud economics don't exist at + * loopback; the poll lattice it left behind is what this replaces). + * + * Wire dialect mirrors the turn stream (pi/ui-stream.ts): text/event-stream, + * no-cache/no-transform, X-Accel-Buffering off — plus a `: ping` comment + * every 25s to keep any intermediary from idling the socket out (locally + * unnecessary, harmless, and it makes dead-connection detection cheap). + * + * The DIFFERENCE from the turn stream is lifetime: a turn's stream ends at + * settle ([DONE]); this channel ends when the CLIENT goes away — cancel() + * is the only end, and it must leave the bus untouched for other windows. + * The closure shape (teardown assigned in start, called in cancel) is the + * ui-stream subscribe() pattern. + */ +import { subscribe, type ServerEvent } from "../core/events.ts"; + +/** One window's subscription as an SSE Response. */ +export function eventsResponse(): Response { + const encoder = new TextEncoder(); + let teardown: (() => void) | undefined; + const stream = new ReadableStream({ + start(controller) { + let closed = false; + let heartbeat: ReturnType | undefined; + let unsubscribe: (() => void) | undefined; + + const stop = () => { + if (closed) return; + closed = true; + unsubscribe?.(); + if (heartbeat) clearInterval(heartbeat); + }; + // Enqueue is the ONLY place the socket can reject us (the reader + // died) — any failure routes through stop() so the listener and + // the heartbeat never outlive the connection. + const frame = (chunk: string) => { + if (closed) return; + try { + controller.enqueue(encoder.encode(chunk)); + } catch { + stop(); + } + }; + + // Attach FIRST, synchronously: no event can fall between + // construction and subscription (single-threaded — the same + // attach-then-replay law as ui-stream's subscribe()). + unsubscribe = subscribe((event: ServerEvent) => { + const { type, ...payload } = event; + frame(`event: ${type}\ndata: ${JSON.stringify(payload)}\n\n`); + }); + heartbeat = setInterval(() => frame(": ping\n\n"), 25_000); + + teardown = stop; + }, + cancel() { + // The view went away: drop the listener + timer. The bus keeps + // serving every other window untouched. + teardown?.(); + }, + }); + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + }, + }); +} diff --git a/src/server/index.ts b/src/server/index.ts index 2191837..ac83d3d 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -5,6 +5,7 @@ */ export * from "./chat-handler.ts"; +export * from "./events-stream.ts"; export * from "./chats.ts"; export * from "./practice.ts"; export * from "./graph.ts"; diff --git a/tests/events.test.ts b/tests/events.test.ts new file mode 100644 index 0000000..a4c6246 --- /dev/null +++ b/tests/events.test.ts @@ -0,0 +1,131 @@ +/** + * THE EVENT BUS + NOTIFICATION CHANNEL (push-architecture slice 1, + * 2026-10-02): the bus is pure core law (typed, globalThis-shared, error- + * isolated); the SSE seam is one connection per open window — frames in, + * cancel out, bus untouched. These pins are the channel's contract: every + * future consumer (chat view, sidebar, dock) builds on exactly this. + */ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { emit, subscribe, type ServerEvent } from "../src/core/events.ts"; +import { eventsResponse } from "../src/server/events-stream.ts"; + +// ─── the bus ──────────────────────────────────────────────────────────────── + +describe("event bus", () => { + it("delivers typed events with their payload to every subscriber", () => { + const got: ServerEvent[] = []; + const off = subscribe((e) => got.push(e)); + emit("turn-started", { chatId: "chat_a" }); + emit("jobs-changed", { chatId: "chat_a" }); + emit("activity-changed", {}); + off(); + assert.deepEqual(got, [ + { type: "turn-started", chatId: "chat_a" }, + { type: "jobs-changed", chatId: "chat_a" }, + { type: "activity-changed" }, + ]); + }); + + it("unsubscribe stops delivery, exactly once", () => { + let calls = 0; + const off = subscribe(() => calls++); + emit("chats-changed", {}); + off(); + emit("chats-changed", {}); + assert.equal(calls, 1); + }); + + it("one listener throwing NEVER breaks the others (error isolation)", () => { + const got: string[] = []; + subscribe(() => { + throw new Error("listener bug"); + }); + const off = subscribe((e) => got.push(e.type)); + emit("turn-settled", { chatId: "chat_b" }); + off(); + assert.deepEqual(got, ["turn-settled"]); + }); + + it("emit with zero subscribers is a silent no-op", () => { + assert.doesNotThrow(() => emit("chats-changed", {})); + }); + + it("the registry is globalThis-backed — the Next multi-entry law (turn-liveness discipline)", () => { + // Every module instance must share ONE bus or nothing propagates + // (the original incident: page render vs route each held a private + // singleton). The key IS the contract. + const g = globalThis as { __lhEventBus?: unknown }; + assert.ok(g.__lhEventBus instanceof Set, "the bus lives on globalThis"); + }); + + it("an unsubscribe DURING emit is safe (snapshot iteration)", () => { + const got: string[] = []; + let off2: () => void = () => {}; + const off1 = subscribe(() => off2()); // listener 2 leaves mid-emit + off2 = subscribe((e) => got.push(e.type)); + emit("chats-changed", {}); + off1(); + off2(); + assert.deepEqual(got, ["chats-changed"], "listener 2 still got THIS event (snapshot), no crash"); + }); +}); + +// ─── the SSE seam ─────────────────────────────────────────────────────────── + +function decode(chunk: Uint8Array): string { + return new TextDecoder().decode(chunk); +} + +describe("eventsResponse (the SSE seam)", () => { + it("holds an SSE response: event frames as they are emitted", async () => { + const res = eventsResponse(); + assert.equal(res.status, 200); + assert.equal(res.headers.get("Content-Type"), "text/event-stream"); + assert.equal(res.headers.get("Cache-Control"), "no-cache, no-transform"); + assert.equal(res.headers.get("X-Accel-Buffering"), "no"); + const reader = res.body!.getReader(); + // The listener attached synchronously in start() — emit NOW lands + // as a queued chunk (no race, single-threaded). + emit("turn-started", { chatId: "chat_x" }); + const { value, done } = await reader.read(); + assert.equal(done, false); + assert.equal( + decode(value!), + 'event: turn-started\ndata: {"chatId":"chat_x"}\n\n', + "one parseable SSE frame: type on the event line, payload on the data line", + ); + await reader.cancel(); + }); + + it("cancel() unsubscribes — later emits reach OTHER windows only", async () => { + const first = eventsResponse(); + const second = eventsResponse(); + const r1 = first.body!.getReader(); + const r2 = second.body!.getReader(); + await r1.cancel(); // window one goes away + emit("turn-settled", { chatId: "chat_y" }); + // window two still receives… + const { value } = await r2.read(); + assert.match(decode(value!), /event: turn-settled/); + // …and window one's stream is finished — no frame, just done. + const after = await r1.read(); + assert.equal(after.done, true, "the cancelled connection carries nothing further"); + await r2.cancel(); + }); + + it("the connection ends WITHOUT a [DONE] sentinel — the client owns the lifetime", async () => { + // A turn's stream terminates itself; the notification channel never + // self-closes. Pin the difference: settled events do NOT close the + // stream — it stays live for the next event. + const res = eventsResponse(); + const reader = res.body!.getReader(); + emit("turn-settled", { chatId: "chat_z" }); + await reader.read(); + emit("chats-changed", {}); + const { value, done } = await reader.read(); + assert.equal(done, false, "the channel survives every event"); + assert.match(decode(value!), /event: chats-changed/); + await reader.cancel(); + }); +});