Run one review at a time, and let a browser watch it

Work item W6: the scheduler, the event bus, the startup hook and the stream
that carries a review's progress to a page.

Reviews are slow and expensive and share one account's usage, so two at once
would race for the same rate limit and make both slower for no gain. One runs
and the rest queue. The queue is in memory only: on a restart it is empty and
the reviews that were waiting are still drafts, which is honest for a local
tool, because nothing was promised to them and nothing was spent on them.
Persisting it would mean a crash could start expensive work nobody was
watching.

The manager announces only what the database already says, and reads the row
before it emits. An event arriving before the row it describes would let the
screen show a stage as started while the database still said draft, and
reloading the page would appear to undo it. A test asserts that from inside the
listener: it reads the row on every event and records any disagreement.

Cancelling a queued review takes it out of the line without touching its
status, because nothing was started and there is nothing to unwind. Cancelling
a running one aborts the signal the service already knows how to map.

The bus swallows a listener's error. A browser that disconnected mid-write must
not stop the other watchers being told, and must not fail the review it was
watching. Listeners are copied before iteration, because a done-event listener
unsubscribes itself and mutating the set mid-iteration would skip whoever came
after it.

Two build problems surfaced only in the production build, and both were real
rather than configuration noise. Turbopack reads the migrations folder's URL as
a module it must resolve, so migrations moved to instrumentation and the path is
assembled with join. And Next builds instrumentation for its edge runtime too,
where none of this can load, so every import there is now inside register() and
behind the runtime check.

Six mutations checked and all caught: removing the concurrency cap, a cancel
that aborts nothing, skipping orphan recovery at startup, announcing before
reading the row, a stream that never closes, and a throwing listener taking the
others down with it.

Author: Poolchaos

docs/DECISIONS.md

@@ -539,3 +539,31 @@ verified evidence, in writing, here.
   proved the value was recorded and not that it was the value used. Making the
   wrong thing unreachable is the stronger answer, and the batching behaviour
   itself is already covered by the pipeline's own context-window tests.
+- 2026-07-31 DECIDED (W6): the job queue lives in memory only. On a restart it
+  is empty and the reviews that were waiting are still drafts. That is the
+  honest behaviour for a local tool: nothing was promised to them and nothing
+  was spent on them, and persisting a queue would mean a crash could start
+  expensive work nobody was watching.
+- 2026-07-31 DECIDED (W6): the manager announces only what the database already
+  says, and reads the row before emitting. An event that arrived before the row
+  it describes would let the screen show a stage as started while the database
+  still said draft, and reloading the page would appear to undo it.
+- 2026-07-31 DECIDED (W6): the bus swallows a listener's error rather than
+  letting it escape. A browser that disconnected mid-write must not stop the
+  other watchers being told and must not fail the review it was watching.
+  Listeners are copied before iteration, because a done-event listener
+  unsubscribes itself and mutating the set mid-iteration would skip whoever
+  came after it.
+- 2026-07-31 DECIDED (W6): migrations run in `instrumentation.ts`, not in the
+  manager. Startup migrates, the manager schedules. It also keeps `migrate.ts`
+  out of the SSE route's import graph, which matters because Turbopack reads
+  `new URL("../../../drizzle", import.meta.url)` as a module it must resolve
+  and fails the build over a directory that is only ever read at run time. The
+  path is now assembled with join for the same reason.
+- 2026-07-31 DECIDED (W6): every import in `instrumentation.ts` is inside
+  `register()` and behind the runtime check. Next builds that file for its edge
+  runtime too, where none of it can load, and a top-level import of anything
+  reaching node:path fails the production build.
+- 2026-07-31 DECIDED (W6): the SSE stream takes a narrow watcher, not the
+  manager. It reads a picture and listens for changes; nothing reachable from a
+  route handler should be able to start or cancel a review.

docs/plans/M2-FINISH-PLAN.md

@@ -400,7 +400,7 @@ maintainer at the top of the demo output directory:
 | W3 | checkpointing runner and its tests                          | W1, W2     | DONE   |
 | W4 | review service, artifact lifecycle, its tests               | W1, W2, W3 | DONE   |
 | W5 | pause/resume/cancel determinism tests                       | W4         | DONE   |
-| W6 | bus, manager, instrumentation, SSE route, tests             | W4         |        |
+| W6 | bus, manager, instrumentation, SSE route, tests             | W4         | DONE   |
 | W7 | demo script, tsx alias proof, npm script                    | W2, W6     |        |
 | W8 | docs, gate evidence, FG-2 checklist                         | W7         |        |
 

src/app/api/reviews/[id]/events/route.ts

@@ -0,0 +1,20 @@
+/**
+ * Live progress for one review.
+ *
+ * The Node runtime is required, not preferred: the manager holds an open
+ * SQLite handle and spawns processes, neither of which the edge runtime can do.
+ */
+
+import { jobManager } from "@/server/jobs/manager";
+import { reviewEventStream } from "@/server/jobs/stream";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+export async function GET(
+  request: Request,
+  context: { params: Promise<{ id: string }> },
+): Promise<Response> {
+  const { id } = await context.params;
+  return reviewEventStream(jobManager(), id, request);
+}

src/instrumentation.ts

@@ -0,0 +1,29 @@
+/**
+ * What runs once, when the server starts.
+ *
+ * Next calls `register()` a single time per server process, which is the only
+ * hook that fits work that must not run per request: opening the database,
+ * migrating it, and recovering reviews that a previous process left marked as
+ * running. A review in that state cannot be running, because nothing survived
+ * the restart that could be running it, and until it is recovered it can
+ * neither be started nor cancelled.
+ *
+ * Every import is inside the function and behind the runtime check. Next builds
+ * this file for its edge runtime as well, where none of it can load, and a
+ * top-level import of anything touching node:path fails that build.
+ */
+
+export async function register(): Promise<void> {
+  if (process.env.NEXT_RUNTIME !== "nodejs") return;
+
+  const { homedir } = await import("node:os");
+  const { dbPath, resolveDataDir } = await import("@/lib/paths");
+  const { createDb } = await import("@/server/db/client");
+  const { runMigrations } = await import("@/server/db/migrate");
+  const { jobManager } = await import("@/server/jobs/manager");
+
+  const dataDir = resolveDataDir(process.env, homedir());
+  const db = createDb(dbPath(dataDir));
+  runMigrations(db);
+  jobManager().init({ db, dataDir });
+}

src/server/db/migrate.ts

@@ -6,11 +6,21 @@
  * anyone running a generator first.
  */
 
+import { dirname, join } from "node:path";
 import { fileURLToPath } from "node:url";
 import { migrate } from "drizzle-orm/better-sqlite3/migrator";
 import type { Db } from "./client";
 
-const MIGRATIONS_FOLDER = fileURLToPath(new URL("../../../drizzle", import.meta.url));
+// Assembled from parts rather than written as one URL literal, because a
+// bundler reads `new URL("../../../drizzle", import.meta.url)` as a module it
+// should resolve and fails the build. This is a directory read at run time.
+const MIGRATIONS_FOLDER = join(
+  dirname(fileURLToPath(import.meta.url)),
+  "..",
+  "..",
+  "..",
+  "drizzle",
+);
 
 export function runMigrations(db: Db, migrationsFolder: string = MIGRATIONS_FOLDER): void {
   migrate(db, { migrationsFolder });

src/server/jobs/bus.ts

@@ -0,0 +1,91 @@
+/**
+ * What a running review tells anyone watching it.
+ *
+ * The bus carries no state a restart would need. Every durable fact, the
+ * status, the stage rows, the run notes, the usage, is written to the database
+ * by the service and the checkpointing runner before it is announced here, so
+ * a listener that joins late or reconnects after a restart can read the whole
+ * truth from the snapshot and lose nothing. That ordering is the point: an
+ * event that arrived before the row it describes would let the UI show a stage
+ * as finished while the database still says it is running, and reloading the
+ * page would appear to undo it.
+ *
+ * Listeners are untrusted with respect to each other. One that throws must not
+ * stop the others from being told, and must not fail the review it is watching.
+ */
+
+import type { ReviewStage, StageErrorClass } from "@/lib/domain/enums";
+import type { ReviewStatus } from "@/lib/domain/state-machines";
+import type { RunNote } from "../db/repositories/reviews";
+
+export interface StageUsage {
+  inputTokens: number;
+  outputTokens: number;
+  costEquivalentUsd: number;
+}
+
+export type ReviewEvent =
+  | { kind: "status"; status: ReviewStatus; pausedReason?: string | undefined }
+  | {
+      kind: "stage";
+      stage: ReviewStage;
+      /** `replayed` costs nothing and spawns nothing; it is not a live call. */
+      phase: "started" | "replayed" | "finished" | "failed";
+      attempt?: number | undefined;
+      usage?: StageUsage | undefined;
+      errorClass?: StageErrorClass | undefined;
+    }
+  | { kind: "engine"; stage: ReviewStage; event: "tool-use" | "text"; detail: string }
+  | { kind: "rate-limit"; status: string; resetsAt?: number | undefined }
+  | { kind: "note"; note: RunNote }
+  | {
+      kind: "done";
+      outcome: "completed" | "paused" | "cancelled" | "failed";
+      reason?: string | undefined;
+    };
+
+export type ReviewListener = (event: ReviewEvent) => void;
+
+export interface ReviewBus {
+  emit: (reviewId: string, event: ReviewEvent) => void;
+  subscribe: (reviewId: string, listener: ReviewListener) => () => void;
+  /** How many listeners a review has, so a test can prove unsubscribe worked. */
+  listenerCount: (reviewId: string) => number;
+}
+
+export function createReviewBus(onListenerError: (error: unknown) => void = () => {}): ReviewBus {
+  const listeners = new Map<string, Set<ReviewListener>>();
+
+  return {
+    emit(reviewId, event) {
+      // Copied before iterating: a listener may unsubscribe itself in response
+      // to a done event, and mutating the set mid-iteration would skip whoever
+      // came after it.
+      for (const listener of [...(listeners.get(reviewId) ?? [])]) {
+        try {
+          listener(event);
+        } catch (error) {
+          // A browser that disconnected mid-write must not fail the review.
+          onListenerError(error);
+        }
+      }
+    },
+
+    subscribe(reviewId, listener) {
+      const forReview = listeners.get(reviewId) ?? new Set<ReviewListener>();
+      forReview.add(listener);
+      listeners.set(reviewId, forReview);
+
+      return () => {
+        const current = listeners.get(reviewId);
+        if (!current) return;
+        current.delete(listener);
+        if (current.size === 0) listeners.delete(reviewId);
+      };
+    },
+
+    listenerCount(reviewId) {
+      return listeners.get(reviewId)?.size ?? 0;
+    },
+  };
+}