diff --git a/app/api/graph/route.ts b/app/api/graph/route.ts index 3cbfc0a..96f5508 100644 --- a/app/api/graph/route.ts +++ b/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 - * returns nodes, edges, derived node states, and session provenance. + * transport: the server entry (src/server/graph.ts) returns nodes, edges, + * session provenance, and per-session understanding observations. */ export const dynamic = "force-dynamic"; diff --git a/app/api/graph/status/route.ts b/app/api/graph/status/route.ts new file mode 100644 index 0000000..e2a22f4 --- /dev/null +++ b/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 }); +} diff --git a/components/graph-view.tsx b/components/graph-view.tsx index bb4e349..f8c2e47 100644 --- a/components/graph-view.tsx +++ b/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() { )} + {/* Background pipeline status: what is currently running. */} + {pipelineRuns.length > 0 && ( +
+ {pipelineRuns.map((run) => ( +
+ + {run.phase} + {run.title} + {Math.max(0, Math.round((Date.now() - run.startedAt) / 1000))}s +
+ ))} +
+ )} + {selected && !isEmpty && (
(); + +/** 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(); + 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 { 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 } } diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index b533895..5a5fbec 100644 --- a/tests/extraction.test.ts +++ b/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. */ -function submitModel(name: string, input: unknown): MockLanguageModelV4 { +/** A model whose ONE tool call is `name` with the given input; then text. + * `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();