push architecture slice 1: core event bus + /api/events SSE channel (notify->fetch infra, no consumers yet)

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();
+}

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<string, never>;
+	/** The session list changed (create, delete, title). */
+	"chats-changed": Record<string, never>;
+}
+
+export type ServerEventType = keyof ServerEvents;
+
+/** One notification: the type discriminates the payload. */
+export type ServerEvent<K extends ServerEventType = ServerEventType> = {
+	type: K;
+} & ServerEvents[K];
+
+export type ServerEventListener = (event: ServerEvent) => void;
+
+interface LhEventsGlobal {
+	__lhEventBus?: Set<ServerEventListener>;
+}
+
+/** 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<ServerEventListener> {
+	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<K extends ServerEventType>(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);
+		}
+	}
+}

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<Uint8Array>({
+		start(controller) {
+			let closed = false;
+			let heartbeat: ReturnType<typeof setInterval> | 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",
+		},
+	});
+}

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";

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();
+	});
+});