diff --git a/app/layout.tsx b/app/layout.tsx index d488756..6782118 100644 --- a/app/layout.tsx +++ b/app/layout.tsx @@ -4,6 +4,7 @@ import "katex/dist/katex.min.css"; import "./globals.css"; import { AppShell, SidebarProvider } from "@/components/sidebar"; import { ActivityProvider } from "@/components/activity-dock"; +import { ServerEventsProvider } from "@/components/server-events"; import { AppearanceProvider } from "@/components/settings/appearance"; // Chrome = JetBrains Mono, teaching prose = Source Sans 3, titles = CMU @@ -20,13 +21,17 @@ export default function RootLayout({ children }: { children: React.ReactNode }) return ( - - - - {children} - - - + {/* The notification channel rides OUTSIDE everything: sidebar, dock, + and pages all subscribe as the polls retire (push S3, 2026-10-02). */} + + + + + {children} + + + + ); diff --git a/components/chat.tsx b/components/chat.tsx index 10619d9..b7e17d0 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -26,6 +26,7 @@ import { DictationBar, joinDictation, useDictation } from "@/components/dictatio import { JobsContext } from "@/components/jobs-context"; import { StatsBar } from "@/components/stats-bar"; import { useActivity } from "@/components/activity-dock"; +import { useServerEvent } from "@/components/server-events"; import { isProviderKeyRefusal, needsContinuation, type ChatInput, type ChatSettings, type LhMessage, type ModelSlotName } from "@/src/core"; import { PlanBlock, PlanPopup, latestTree } from "@/components/plan-block"; import { PlanMarginalia, planExists } from "@/components/plan-window"; @@ -623,75 +624,40 @@ export function Chat({ void syncTurns(); }, [syncTurns]); // After the learner's own turn ends the log has a fresh entry — and the - // Post-turn adoption + the HANDOFF WATCH. Two things can trail our own - // stream's end: (1) the server's persist/endTurn — milliseconds, and the - // record is only complete once not-in-flight (plan blocks attach at - // persist); (2) a WAKE TURN — the probe→teacher handoff fires - // ~0.75–1.2s later (the pump's grace), and NO jobs poll is running to - // notice it: probe-phase sessions have zero job traffic, so the jobs - // poll is switched off and the open client never learns the teacher's - // turn exists (the freeze the user hit 2026-09-11 — the sidebar saw - // busy→idle while the session sat frozen). So: adopt at the first idle - // read as before, then keep watching briefly — if turnInFlight flips - // true, hand off to the wake-adoption poll below (busy shimmer, adopt - // at finish). + // Post-turn settle (the SLIMMED watch — push-architecture S3, + // 2026-10-02): our own stream just ended. Two instant LOCAL actions + // only — the turns-only refetch lands the log entry while the server's + // persist/endTurn lags by milliseconds, and the coverage ring re-reads + // (this turn's messages just joined the uncovered mass). The ADOPTION + // LOOP and the HANDOFF WATCH that lived here are RETIRED: a + // server-initiated turn (the probe→teach handoff, a job wake, another + // tab) is now ANNOUNCED on the events channel — the turn-started + // subscription (below the jobs poll) attaches to it live. No polling + // window can expire; the freeze class is gone. const prevStatusRef = useRef(status); useEffect(() => { - if (prevStatusRef.current !== "ready" && status === "ready") { - // immediate: turns-only refetch — the stats bar updates even while - // the server's persist/endTurn lags the client by milliseconds - void syncTurns(); - // The coverage ring re-reads at turn settle too (extraction-UX - // redesign, 2026-09-24): this turn's messages just joined the - // uncovered mass; the ring's own running-poll handles runs. - setCoverageTick((t) => t + 1); - // Turn-end nudge (accuracy wave, 2026-09-19): the extraction - // pipeline fires at this turn's onFinish — the dock's poll would - // discover a started run up to 10s late. Refetch the activity feed - // now (stream settle) and again at +600/+1500ms (bounded window — - // covers onFinish→triggerExtraction→setPhase lag); the feed's own - // fast interval takes over once it sees the run. - refreshActivity(); - const nudgeTimers = [600, 1500].map((ms) => setTimeout(() => refreshActivity(), ms)); - let cancelled = false; - let timer: ReturnType | undefined; - let adopted = false; - const WATCH_ATTEMPTS = 8; // 8 × 500ms: persist lag + the ~1.2s handoff - const adopt = async (attempt: number) => { - if (cancelled) return; - try { - const res = await fetch(`/api/chats/${chatId}`); - if (res.ok) { - const record = (await res.json()) as { turnInFlight?: boolean }; - if (!adopted) { - // phase 1: wait out our own turn's persist lag, then adopt once - if (record.turnInFlight !== true) { - adopted = true; - void syncFromServer(); - } - } else if (record.turnInFlight === true) { - // phase 2 (the watch): a wake turn started — the handoff. - // Attach to it live (stream-reattach wave): the same - // resume path as mount reattach; the record poll is - // the fallback while it connects. - attachToRunningTurn(); - return; - } - } - } catch { - /* transient — keep polling */ - } - if (attempt < WATCH_ATTEMPTS && !cancelled) timer = setTimeout(() => void adopt(attempt + 1), 500); - }; - void adopt(0); - return () => { - cancelled = true; - if (timer) clearTimeout(timer); - nudgeTimers.forEach(clearTimeout); - }; - } + const transitioned = prevStatusRef.current !== "ready" && status === "ready"; + // Every path (the old early-return skipped this on the transition + // branch — the re-fire-on-dep-churn wart, fixed S3). prevStatusRef.current = status; - }, [status, chatId, syncFromServer, syncTurns, refreshActivity, attachToRunningTurn]); + if (!transitioned) return; + // immediate: turns-only refetch — the stats bar updates even while + // the server's persist/endTurn lags the client by milliseconds + void syncTurns(); + // The coverage ring re-reads at turn settle too (extraction-UX + // redesign, 2026-09-24); the ring's own running-poll handles runs. + setCoverageTick((t) => t + 1); + // Turn-end nudge (accuracy wave, 2026-09-19): the extraction + // pipeline fires at this turn's onFinish — the dock's poll would + // discover a started run up to 10s late. Refetch the activity feed + // now (stream settle) and again at +600/+1500ms (bounded window — + // covers onFinish→triggerExtraction→setPhase lag); the feed's own + // fast interval takes over once it sees the run. (Dies at S5 when + // the dock goes event-driven.) + refreshActivity(); + const nudgeTimers = [600, 1500].map((ms) => setTimeout(() => refreshActivity(), ms)); + return () => nudgeTimers.forEach(clearTimeout); + }, [status, syncTurns, refreshActivity]); // A stream that died MID-FLIGHT (network): the runner decoupled the turn // from its view, so the server turn runs ON behind the error banner — @@ -858,52 +824,44 @@ export function Chat({ }; }, [pollMs, jobPartCount, chatId, syncFromServer, adoptAppendOnly, refreshActivity, attachToRunningTurn]); - // The FALLBACK liveness poll (stream-reattach wave): runs only while the - // reattach stream couldn't take over — the pre-attach window, a failed - // resume, or a resumed stream that died mid-flight. Polls the record - // until the orphaned turn finishes (turnInFlight flips false server-side - // in onFinish, AFTER persist), then adopts the completed transcript — - // the syncFromServer above already runs the job-continuation check. - // Until then the session reads as busy: pre-first-chunk shimmer, - // composer gated. Steer content shows NOW via append-only adoption. - useEffect(() => { - if (!turnInFlight) return; - let cancelled = false; - let timer: ReturnType | undefined; - const poll = async () => { - if (cancelled) return; - try { - const res = await fetch(`/api/chats/${chatId}`); - if (res.ok) { - const record = (await res.json()) as { - turnInFlight?: boolean; - messages?: LhMessage[]; - }; - // Ground rule (2026-09-15): steer content persisted by the - // wake shows NOW, at the wake turn's start — not at its - // finish. Append-only is safe here: no local stream runs - // under a server-side wake turn. - if (Array.isArray(record.messages)) adoptRecord(record.messages); - if (!record.turnInFlight) { - setTurnInFlight(false); - // Heal a stream-death banner (the failed-attach fallback): the - // record is the truth and the truth just landed. - if (status === "error") clearError(); - void syncFromServer(); - return; - } - } - } catch { - /* transient network hiccup — keep polling */ + // ── Server-initiated turn discovery (push-architecture S3, 2026-10-02) ── + // Every turn THIS chat runs is announced on the events channel — the + // probe→teach handoff, a job wake, another tab's send, the + // missed-on-mount case. This replaces the retired watch loop and the + // 1500ms fallback record poll: discovery is no longer a polling window + // that can expire. Handlers read REFS, not closure state — events fire + // between renders, and a ref holds the live value at event time. + const statusRef = useRef(status); + statusRef.current = status; + useServerEvent("turn-started", (p) => { + // Known non-issue: beginTurn emits before registerTurn in the same + // synchronous request flow; a GET /stream inside that sub-ms window + // gets a settled response — unwinnable by a network round-trip, + // self-healing via the eventual turn-settled. Do not engineer around. + if (p.chatId !== chatId) return; + if (localStreamRef.current) return; // our OWN turn — the 409-race guard + if (turnInFlightRef.current) return; // already attached + attachToRunningTurn(); + }); + useServerEvent("turn-settled", (p) => { + if (p.chatId !== chatId) return; + if (localStreamRef.current) return; // our own [DONE] path owns the settle + if (turnInFlightRef.current) { + // The attached stream's terminal frame owns the normal settle — + // syncing HERE would swap a stale snapshot mid-stream (the + // continuation-coalescing law, see the jobs poll above). Only + // the error case heals: the attached stream died, and the settle + // says persistTurn has made the record durable truth. + if (statusRef.current === "error") { + setTurnInFlight(false); + clearError(); + void syncFromServer(); } - if (!cancelled) timer = setTimeout(poll, 1500); - }; - timer = setTimeout(poll, 1000); - return () => { - cancelled = true; - if (timer) clearTimeout(timer); - }; - }, [turnInFlight, chatId, syncFromServer, status, clearError]); + return; + } + // Fully idle — a turn we never saw (every window missed): adopt now. + void syncFromServer(); + }); const onSend = () => { if (busy || pending) return; diff --git a/components/server-events.tsx b/components/server-events.tsx new file mode 100644 index 0000000..bb0331b --- /dev/null +++ b/components/server-events.tsx @@ -0,0 +1,155 @@ +"use client"; + +/** + * THE NOTIFICATION CHANNEL'S CLIENT HALF (push-architecture slice 3, + * 2026-10-02): ONE EventSource("/api/events") per open window, held for + * the window's lifetime — the channel that retires the app's poll + * lattice. Every server-initiated change arrives as a named SSE frame; + * consumers respond by fetching truth (notify → fetch, the architecture's + * whole trick — see src/core/events.ts and server/events-stream.ts). + * + * This file is the ONE place in the app that knows EventSource exists. + * Auto-reconnect is EventSource's own — never fought, always exploited: + * every (re)open fires the resync callbacks, so a reconnect heals with + * one ordinary fetch per consumer (a dev-server restart self-heals here). + * + * Shape: the provider holds a hub (window-local fan-out — the client twin + * of the server bus; per-window, NOT globalThis: each window's connection + * is its own) and wires ONE source into it. The hooks are thin: latest-ref + * callbacks (events fire between renders — never trust a stale closure) + * plus hub membership for the component's lifetime. + * + * TEST SEAM: `connect` injects the source factory — the hub + wiring are + * plain functions the tests drive directly (no React, no server); the + * React glue is an adapter seam like lh:turn-state (untested by + * precedent, components/session.ts). + */ +import { createContext, useContext, useEffect, useMemo, useRef, type ReactNode } from "react"; +import { SERVER_EVENT_TYPES, type ServerEventType, type ServerEvents } from "@/src/core"; + +/** The minimal source surface the provider consumes — EventSource behind + * this in production, fakes in tests. */ +export interface NotifySource { + close(): void; + onOpen(cb: () => void): void; + onEvent(name: string, cb: (data: string) => void): void; +} + +export function openEventsChannel(): NotifySource { + const es = new EventSource("/api/events"); + return { + close: () => es.close(), + onOpen: (cb) => { + es.onopen = () => cb(); + }, + onEvent: (name, cb) => { + es.addEventListener(name, (ev) => cb((ev as MessageEvent).data)); + }, + }; +} + +/** The per-window fan-out: typed subscriptions over one source. Listener + * errors are isolated per the bus's law (one bad consumer never breaks + * the window), and dispatch iterates a snapshot (an unsubscribe during + * dispatch is safe). */ +export function createNotifyHub() { + const byType = new Map void>>(); + const resyncs = new Set<() => void>(); + return { + subscribe(type: K, cb: (payload: ServerEvents[K]) => void) { + const set = byType.get(type) ?? new Set<(data: never) => void>(); + byType.set(type, set); + const slot = cb as unknown as (data: never) => void; + set.add(slot); + return () => set.delete(slot); + }, + /** Resync callbacks: fired on the source's every (re)open — "you may + * have missed events while disconnected; refetch". A consumer that + * mounts while ALREADY open gets no fire (it missed nothing by + * birth); its own mount fetch is its initial truth. */ + onResync(cb: () => void) { + resyncs.add(cb); + return () => resyncs.delete(cb); + }, + /** One named frame arrived: parse, fan out. Unparseable frames are + * dropped loudly — the channel's payloads are tiny and schemaless, + * a bad frame must never half-fire. */ + dispatch(name: string, raw: string) { + const set = byType.get(name); + if (!set) return; + let payload: unknown; + try { + payload = JSON.parse(raw); + } catch (err) { + console.error(`[events] unparseable frame for ${name}:`, err); + return; + } + for (const cb of [...set]) { + try { + (cb as unknown as (data: unknown) => void)(payload); + } catch (err) { + console.error(`[events] ${name} handler failed:`, err); + } + } + }, + fireOpen() { + for (const cb of [...resyncs]) { + try { + cb(); + } catch (err) { + console.error("[events] resync handler failed:", err); + } + } + }, + }; +} + +/** Wire one source into a hub — the provider's ENTIRE effect body, factored + * out so the wiring is testable without React. Returns the disconnect. */ +export function wireSource(source: NotifySource, hub: ReturnType): () => void { + for (const name of SERVER_EVENT_TYPES) source.onEvent(name, (raw) => hub.dispatch(name, raw)); + source.onOpen(() => hub.fireOpen()); + return () => source.close(); +} + +// ── the React glue ────────────────────────────────────────────────────────── + +const EventsContext = createContext | null>(null); + +export function ServerEventsProvider({ children, connect }: { children: ReactNode; connect?: () => NotifySource }) { + const hub = useMemo(() => createNotifyHub(), []); + useEffect(() => wireSource((connect ?? openEventsChannel)(), hub), [hub, connect]); + return {children}; +} + +/** Subscribe to one event type for this component's lifetime. The callback + * is a latest-ref: pass inline closures freely — the freshest render's + * closure runs, so state captured at the last completed render is safe to + * read (the provider re-stores it every render). */ +export function useServerEvent(type: K, cb: (payload: ServerEvents[K]) => void): void { + const hub = useContext(EventsContext); + if (!hub) throw new Error("useServerEvent requires "); + const cbRef = useRef(cb); + cbRef.current = cb; + useEffect(() => { + const off = hub.subscribe(type, (payload) => cbRef.current(payload)); + return () => { + off(); + }; + }, [hub, type]); +} + +/** Refetch truth on every (re)connect of the window's channel — the + * self-heal seam for EventSource reconnects and dev-server restarts. */ +export function useServerResync(cb: () => void): void { + const hub = useContext(EventsContext); + if (!hub) throw new Error("useServerResync requires "); + const cbRef = useRef(cb); + cbRef.current = cb; + useEffect(() => { + const off = hub.onResync(() => cbRef.current()); + return () => { + off(); + }; + }, [hub]); +} diff --git a/src/core/events.ts b/src/core/events.ts index 3a924fd..bb02528 100644 --- a/src/core/events.ts +++ b/src/core/events.ts @@ -42,6 +42,17 @@ export interface ServerEvents { export type ServerEventType = keyof ServerEvents; +/** The vocabulary's runtime names — the client wires one listener per + * name (types are erased; this list is the single source of truth for + * what the SSE channel can carry). */ +export const SERVER_EVENT_TYPES = [ + "turn-started", + "turn-settled", + "jobs-changed", + "activity-changed", + "chats-changed", +] as const satisfies readonly ServerEventType[]; + /** One notification: the type discriminates the payload. */ export type ServerEvent = { type: K; diff --git a/src/core/index.ts b/src/core/index.ts index 415ba12..761e476 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -20,6 +20,10 @@ export * from "./ports.ts"; export * from "./render-settings.ts"; export * from "./question-core.ts"; +// The event bus (push architecture, 2026-10-02): vocabulary + emit/subscribe +// for every layer; the browser shell consumes the types + the runtime names +// through this barrel (shell law: pure core via @/src/core only). +export * from "./events.ts"; // clusters.ts (Louvain) is deliberately NOT re-exported: nothing in the // shell runs community detection — the snapshot carries the partition // from the server. Keeping it out keeps the heavy clustering deps out of the diff --git a/tests/server-events.test.tsx b/tests/server-events.test.tsx new file mode 100644 index 0000000..40e8cae --- /dev/null +++ b/tests/server-events.test.tsx @@ -0,0 +1,108 @@ +/** + * THE NOTIFICATION CHANNEL'S CLIENT HALF (push-architecture S3, + * 2026-10-02): one EventSource per window fanned out through the hub. + * The provider's ENTIRE effect body is wireSource(source, hub) — a plain + * function — so these pins drive the real wiring with a fake source, no + * React and no server: typed dispatch, resync-on-open, error isolation, + * and close-stops-dispatch. The React glue around it (hooks/context) is + * an adapter seam like lh:turn-state (untested by precedent). + */ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { createNotifyHub, wireSource, type NotifySource } from "../components/server-events.tsx"; + +/** A source the test drives by hand — recordable, closable, honest. */ +function fakeSource() { + const events = new Map void>>(); + const opens = new Set<() => void>(); + const src: NotifySource & { + push(name: string, payload: unknown): void; + open(): void; + closed: boolean; + } = { + closed: false, + close() { + this.closed = true; + }, + onOpen(cb) { + opens.add(cb); + }, + onEvent(name, cb) { + const set = events.get(name) ?? new Set(); + events.set(name, set); + set.add(cb); + }, + push(name, payload) { + for (const cb of events.get(name) ?? []) cb(JSON.stringify(payload)); + }, + open() { + for (const cb of opens) cb(); + }, + }; + return src; +} + +describe("notify hub + wireSource (the client channel wiring)", () => { + it("a named frame dispatches to the right subscriber with the parsed payload", () => { + const hub = createNotifyHub(); + const src = fakeSource(); + wireSource(src, hub); + const got: string[] = []; + hub.subscribe("turn-started", (p) => got.push(p.chatId)); + hub.subscribe("turn-settled", (p) => got.push(`settled:${p.chatId}`)); + src.push("turn-started", { chatId: "chat_wire" }); + src.push("turn-settled", { chatId: "chat_wire" }); + src.push("jobs-changed", { chatId: "chat_wire" }); // no subscriber — no-op + assert.deepEqual(got, ["chat_wire", "settled:chat_wire"]); + }); + + it("open fires resync (initial AND reconnect — EventSource's own retry exploited)", () => { + const hub = createNotifyHub(); + const src = fakeSource(); + wireSource(src, hub); + let resyncs = 0; + hub.onResync(() => resyncs++); + src.open(); + src.open(); // the reconnect case: same callback, fired again + assert.equal(resyncs, 2); + }); + + it("a throwing handler never breaks its neighbors (the bus's isolation law)", () => { + const hub = createNotifyHub(); + const src = fakeSource(); + wireSource(src, hub); + hub.subscribe("chats-changed", () => { + throw new Error("handler bug"); + }); + const got: number[] = []; + hub.subscribe("chats-changed", () => got.push(1)); + src.push("chats-changed", {}); + assert.deepEqual(got, [1]); + }); + + it("an unparseable frame is dropped loudly, never half-fired", () => { + const hub = createNotifyHub(); + const src = fakeSource(); + wireSource(src, hub); + let calls = 0; + hub.subscribe("activity-changed", () => calls++); + // dispatch is the seam's own parser — feed it garbage directly + hub.dispatch("activity-changed", "{not json"); + assert.equal(calls, 0); + }); + + it("unsubscribe stops delivery; close() (the disconnect) is wired through", () => { + const hub = createNotifyHub(); + const src = fakeSource(); + const disconnect = wireSource(src, hub); + let calls = 0; + const off = hub.subscribe("chats-changed", () => calls++); + src.push("chats-changed", {}); + off(); + src.push("chats-changed", {}); + assert.equal(calls, 1); + assert.equal(src.closed, false, "connected until the provider unmounts"); + disconnect(); + assert.equal(src.closed, true, "the effect's cleanup closes the source"); + }); +});