Background pipeline status: registry + /api/graph/status + graph-view chips

The background extraction pipeline is invisible on the home surface —
essential observability per the user's directive: the graph view must
show what is currently running (extracting / auditing / merging) and
for which chat.

- extraction.ts: module-level registry of in-flight runs (chatId,
  phase, startedAt), updated at each phase transition in runExtraction
  and cleared in finally (done or failed). extractionStatuses() is the
  read; one process, one registry.
- GET /api/graph/status: thin transport over the registry, chat titles
  joined from the chat store.
- graph-view: polls every 2s and shows one chip per running pipeline
  (spinner + phase + title + elapsed seconds, bottom-left); when a run
  finishes (some -> none) the graph refetches, so merges land live.
  Status is cosmetic — poll errors never surface.
- Unit test drives a mocked pipeline run and observes the registry
  mid-flight: 'extracting' during the extractor's generate, 'auditing'
  during the auditor's, empty after.

app/api/graph/route.ts

@@ -2,8 +2,8 @@ import { graphView } from "@/src/server";
 
 /**
  * The home graph view's data endpoint (W4; vision §1 "Home page") — thin
- * transport: the server entry (src/server/graph.ts) folds stale chats and
+ * transport: the server entry (src/server/graph.ts) returns nodes, edges,
- * returns nodes, edges, derived node states, and session provenance.
+ * session provenance, and per-session understanding observations.
  */
 
 export const dynamic = "force-dynamic";

app/api/graph/status/route.ts

@@ -0,0 +1,21 @@
+/**
+ * GET /api/graph/status — the in-flight background extraction runs
+ * (vision §1: the pipeline status must be visible on the home graph).
+ * Thin transport over the pipeline's module-level registry: phase
+ * (extracting / auditing / merging) per chat, oldest first. Titles joined
+ * from the chat store; a missing chat record still reports the run.
+ */
+import { extractionStatuses } from "@/src/agent/subagents/extraction.ts";
+import { chatStore } from "@/src/store/chats.ts";
+
+export const dynamic = "force-dynamic";
+
+export async function GET() {
+	const runs = extractionStatuses().map((run) => ({
+		chatId: run.chatId,
+		title: chatStore.getChat(run.chatId)?.title ?? run.chatId,
+		phase: run.phase,
+		startedAt: run.startedAt,
+	}));
+	return Response.json({ runs });
+}

components/graph-view.tsx

@@ -56,7 +56,7 @@
 
 import { useEffect, useLayoutEffect, useMemo, useRef, useState } from "react";
 import Link from "next/link";
-import { SearchIcon, WaypointsIcon } from "lucide-react";
+import { Loader2Icon, SearchIcon, WaypointsIcon } from "lucide-react";
 import Graph from "graphology";
 import forceAtlas2 from "graphology-layout-forceatlas2";
 import type { NodeHoverDrawingFunction, NodeLabelDrawingFunction } from "sigma/rendering";
@@ -352,7 +352,9 @@ export function GraphView() {
 		return set;
 	}, [snapshot, query]);
 
-	// Fetch the derived graph (the route folds stale chats first).
+	// Fetch the derived graph. Re-runs when a background pipeline run
+	// finishes (refreshTick) — its merges just landed in the ledger.
+	const [refreshTick, setRefreshTick] = useState(0);
 	useEffect(() => {
 		let cancelled = false;
 		fetch("/api/graph")
@@ -364,6 +366,30 @@ export function GraphView() {
 		return () => {
 			cancelled = true;
 		};
+	}, [refreshTick]);
+
+	// Background pipeline status: poll /api/graph/status; when a run
+	// finishes (some -> none), refetch the graph.
+	const [pipelineRuns, setPipelineRuns] = useState<{ chatId: string; title: string; phase: string; startedAt: number }[]>([]);
+	useEffect(() => {
+		let cancelled = false;
+		let hadRuns = false;
+		const poll = () =>
+			fetch("/api/graph/status")
+				.then((res) => res.json() as Promise<{ runs: typeof pipelineRuns }>)
+				.then(({ runs }) => {
+					if (cancelled) return;
+					setPipelineRuns(runs);
+					if (hadRuns && runs.length === 0) setRefreshTick((t) => t + 1);
+					hadRuns = runs.length > 0;
+				})
+				.catch(() => {}); // status is cosmetic — never surface poll errors
+		poll();
+		const timer = setInterval(poll, 2000);
+		return () => {
+			cancelled = true;
+			clearInterval(timer);
+		};
 	}, []);
 
 	// Build the graphology graph, lay it out once, mount sigma.
@@ -612,6 +638,23 @@ export function GraphView() {
 				</div>
 			)}
 
+			{/* Background pipeline status: what is currently running. */}
+			{pipelineRuns.length > 0 && (
+				<div className="absolute bottom-3 left-3 z-10 flex flex-col gap-1.5">
+					{pipelineRuns.map((run) => (
+						<div
+							key={run.chatId}
+							className="flex items-center gap-2 rounded-full border bg-card/85 px-3 py-1.5 text-xs text-muted-foreground shadow-lg backdrop-blur"
+						>
+							<Loader2Icon className="size-3.5 shrink-0 animate-spin" />
+							<span className="capitalize">{run.phase}</span>
+							<span className="max-w-48 truncate text-foreground">{run.title}</span>
+							<span className="tabular-nums">{Math.max(0, Math.round((Date.now() - run.startedAt) / 1000))}s</span>
+						</div>
+					))}
+				</div>
+			)}
+
 			{selected && !isEmpty && (
 				<div
 					ref={cardRef}

src/agent/subagents/extraction.ts

@@ -304,6 +304,25 @@ function createAuditor(model: LanguageModel) {
 	});
 }
 
+// ─── the pipeline status registry ─────────────────────────────────────────────
+
+/** The pipeline phases, in order — what the home status endpoint shows. */
+export type ExtractionPhase = "extracting" | "auditing" | "merging";
+
+/** One in-flight pipeline run. Module-level: one process, one registry. */
+interface RunningExtraction {
+	chatId: string;
+	phase: ExtractionPhase;
+	startedAt: number;
+}
+
+const runningExtractions = new Map<string, RunningExtraction>();
+
+/** In-flight pipeline runs (oldest first) — the /api/graph/status payload. */
+export function extractionStatuses(): RunningExtraction[] {
+	return [...runningExtractions.values()].sort((a, b) => a.startedAt - b.startedAt);
+}
+
 // ─── the pipeline ───────────────────────────────────────────────────────────
 
 export interface ExtractionModels {
@@ -327,6 +346,14 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
 	const auditorAgent = createAuditor(deps.auditor ?? modelFor("auditor"));
 	const inFlight = new Set<string>();
 
+	function setPhase(chatId: string, phase: ExtractionPhase): void {
+		runningExtractions.set(chatId, {
+			chatId,
+			phase,
+			startedAt: runningExtractions.get(chatId)?.startedAt ?? Date.now(),
+		});
+	}
+
 	async function extract(chatId: string): Promise<SessionFindings> {
 		const record = chatStore.getChat(chatId);
 		if (!record) throw new Error(`Chat not found: ${chatId}`);
@@ -388,8 +415,11 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
 		const t0 = performance.now();
 		try {
 			console.info(`[extraction ${chatId}] pipeline start`);
+			setPhase(chatId, "extracting");
 			const findings = await extract(chatId);
+			setPhase(chatId, "auditing");
 			const auditResult = await audit(chatId, findings);
+			setPhase(chatId, "merging");
 			// The merger re-reads the chat record itself — events are rebuilt
 			// from the transcript, never from stale copies.
 			const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore });
@@ -405,6 +435,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
 			throw err;
 		} finally {
 			inFlight.delete(chatId);
+			runningExtractions.delete(chatId); // done or failed — no longer running
 		}
 	}
 

tests/extraction.test.ts

@@ -34,7 +34,7 @@ import type {
 	SessionFindings,
 } from "../src/core/contracts.ts";
 import type { GraphStore } from "../src/core/ports.ts";
-import { createExtractionPipeline, transcriptView } from "../src/agent/subagents/extraction.ts";
+import { createExtractionPipeline, extractionStatuses, transcriptView } from "../src/agent/subagents/extraction.ts";
 import { chatStore } from "../src/store/chats.ts";
 import { PATHS } from "../config.ts";
 
@@ -163,13 +163,16 @@ const FINDINGS: SessionFindings = {
 	attribution: { quiz: { "probe-1": "p1", "check-1": "p2" } },
 };
 
-/** A model whose ONE tool call is `name` with the given input; then text. */
+/** A model whose ONE tool call is `name` with the given input; then text.
-function submitModel(name: string, input: unknown): MockLanguageModelV4 {
+ *  `onCall` (optional) fires at the moment of the first generate — the
+ *  observation point for mid-run assertions. */
+function submitModel(name: string, input: unknown, onCall?: () => void): MockLanguageModelV4 {
 	let turn = 0;
 	return new MockLanguageModelV4({
 		doGenerate: async () => {
 			turn++;
 			if (turn === 1) {
+				onCall?.();
 				return {
 					content: [
 						{
@@ -342,6 +345,39 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
 		assert.equal(ledgers.readCheckpoints().length, 0);
 	});
 
+	it("reports its phase through the status registry while running, and clears it when done", async () => {
+		const { chatId } = fixtureChat();
+		const ledgers = memoryLedgers();
+		const existingId = "n_existing";
+		ledgers.appendNode({
+			id: existingId,
+			name: "Derivative",
+			statement: "Derivatives measure rates of change",
+			aliases: [],
+			createdAt: "2026-08-01T00:00:00.000Z",
+			provenance: [],
+		});
+		const seen: (string | undefined)[] = [];
+		const phaseOf = () => extractionStatuses().find((r) => r.chatId === chatId)?.phase;
+		const verdict: AuditResult = {
+			propositions: [{ ...FINDINGS.propositions[0]!, linkTo: existingId }],
+			edges: [],
+			attribution: { quiz: {} },
+			dropped: [{ ref: "p2", reason: "evidence not found" }],
+		};
+		const pipeline = createExtractionPipeline({
+			chatStore,
+			graphStore: ledgers,
+			// Each stage's first generate is the observation point: the
+			// registry must already carry this chat at the right phase.
+			extractor: submitModel("submit_findings", FINDINGS, () => seen.push(phaseOf())),
+			auditor: submitModel("submit_review", verdict, () => seen.push(phaseOf())),
+		});
+		await pipeline.runExtraction(chatId);
+		assert.deepEqual(seen, ["extracting", "auditing"]);
+		assert.equal(extractionStatuses().length, 0, "the registry clears when the run ends");
+	});
+
 	it("is single-flight per chat — a concurrent second run is a no-op", async () => {
 		const { chatId } = fixtureChat();
 		const ledgers = memoryLedgers();