diff --git a/app/api/graph/coverage/extract/route.ts b/app/api/graph/coverage/extract/route.ts new file mode 100644 index 0000000..2687f67 --- /dev/null +++ b/app/api/graph/coverage/extract/route.ts @@ -0,0 +1,36 @@ +/** + * POST /api/graph/coverage/extract — the manual extraction affordance + * (vision-extraction-coverage.md): run the pipeline over ONE session's + * uncovered content, whatever its size — the user's answer to an aborted + * run, a pipeline error, or a session that went dormant under the chunk + * budget. The pipeline itself batches under EXTRACTION.chunkTokens and + * checkpoints coverage per merged batch, so this is the same run the + * trigger fires, just user-initiated. 409 while a run is already in + * flight (the dock is the source of truth for that); failures are logged + * by the pipeline and leave coverage untouched — the button stays honest. + */ +import { extraction, extractionStatuses } from "@/src/server"; + +export const dynamic = "force-dynamic"; + +export async function POST(request: Request) { + let body: unknown; + try { + body = await request.json(); + } catch { + return Response.json({ error: "Invalid JSON body" }, { status: 400 }); + } + const chatId = (body as { chatId?: unknown }).chatId; + if (typeof chatId !== "string" || chatId === "") { + return Response.json({ error: 'Body must be {chatId: string}' }, { status: 400 }); + } + if (extractionStatuses().some((run) => run.kind === "chat" && run.chatId === chatId)) { + return Response.json({ error: "An extraction run is already in flight for this session" }, { status: 409 }); + } + // Fire-and-forget, like the trigger: the dock + the coverage panel's + // running flags track progress; the promise never blocks the response. + void extraction.runExtraction(chatId).catch((err: unknown) => { + console.error(`[chat ${chatId}] manual extraction failed:`, err); + }); + return Response.json({ started: true }, { status: 202 }); +} \ No newline at end of file diff --git a/app/api/graph/coverage/route.ts b/app/api/graph/coverage/route.ts new file mode 100644 index 0000000..8ebecfe --- /dev/null +++ b/app/api/graph/coverage/route.ts @@ -0,0 +1,40 @@ +/** + * GET /api/graph/coverage — the exact extraction coverage of every session + * (vision-extraction-coverage.md, settled 2026-09-15): which sessions have + * unextracted content, how much (tokens + message counts), and which runs + * are in flight. "Uncovered" is DERIVED here per request — fingerprint the + * current transcript, diff against the coverage ledger — never tracked as + * a second ledger. This is the data behind the graph view's coverage panel + * and its per-session "extract what's missing" buttons. + */ +import type { UIMessage } from "ai"; +import { chatStore, extractionStatuses, stores } from "@/src/server"; +import { coverageOf } from "@/src/core"; + +export const dynamic = "force-dynamic"; + +export async function GET() { + const running = new Set( + extractionStatuses().filter((run) => run.kind === "chat").map((run) => run.chatId), + ); + const entries = stores.graphStore.readCoverage(); + const chats = chatStore.listChats().map((summary) => { + const record = chatStore.getChat(summary.id); + const messages = (record?.messages ?? []) as UIMessage[]; + const cov = coverageOf(messages, entries, summary.id); + return { + chatId: summary.id, + title: summary.title, + updatedAt: summary.updatedAt, + totalMessages: cov.slots.length, + coveredMessages: cov.coveredCount, + uncoveredMessages: cov.uncoveredCount, + coveredTokens: cov.coveredTokens, + uncoveredTokens: cov.uncoveredTokens, + running: running.has(summary.id), + }; + }); + // Most uncovered first — the actionable sessions lead the panel. + chats.sort((a, b) => b.uncoveredTokens - a.uncoveredTokens || (a.updatedAt < b.updatedAt ? 1 : -1)); + return Response.json({ chats }); +} \ No newline at end of file diff --git a/components/graph-coverage.tsx b/components/graph-coverage.tsx new file mode 100644 index 0000000..aec46c1 --- /dev/null +++ b/components/graph-coverage.tsx @@ -0,0 +1,190 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import { ListChecksIcon, Loader2Icon } from "lucide-react"; +import { formatTokens } from "@/components/usage-format"; + +/** + * The extraction-coverage control (vision-extraction-coverage.md, settled + * 2026-09-15): the graph view's window into WHAT THE PIPELINE HAS COVERED. + * One button in the top-right chrome opens a panel with the exact coverage + * of every session — covered/uncovered tokens and message counts — and a + * per-session "Extract" button: the manual answer to an aborted run, a + * pipeline error, or a session that went dormant under the chunk budget. + * + * Data: GET /api/graph/coverage (uncovered is DERIVED server-side — + * fingerprint the transcript, diff the ledger — never dual-tracked). + * Manual runs: POST /api/graph/coverage/extract; the run then surfaces in + * the status chips + this panel's running row, and the graph refetches when + * it finishes (the parent's runs-empty transition). The panel polls only + * while open — no new standing poll. + */ + +interface CoverageChat { + chatId: string; + title: string; + updatedAt: string; + totalMessages: number; + coveredMessages: number; + uncoveredMessages: number; + coveredTokens: number; + uncoveredTokens: number; + running: boolean; +} + +export function GraphCoverageControl({ refreshKey = 0 }: { refreshKey?: number }) { + const [open, setOpen] = useState(false); + const [chats, setChats] = useState([]); + const [loaded, setLoaded] = useState(false); + const [extracting, setExtracting] = useState([]); // click-pending ids (optimistic until the poll says running) + + const load = useCallback(async () => { + try { + const res = await fetch("/api/graph/coverage"); + if (!res.ok) return; + const data = (await res.json()) as { chats: CoverageChat[] }; + setChats(data.chats); + setLoaded(true); + // Drop click-pending ids the server now reports as running (or + // that have finished — coverage moved, so the row settled). + setExtracting((ids) => ids.filter((id) => data.chats.find((c) => c.chatId === id)?.running === true)); + } catch { + /* transient — the next tick retries */ + } + }, []); + + useEffect(() => { + if (!open) return; + void load(); + const timer = setInterval(load, 2000); + return () => clearInterval(timer); + }, [open, load, refreshKey]); + + const onExtract = async (chatId: string) => { + setExtracting((ids) => [...ids, chatId]); + try { + const res = await fetch("/api/graph/coverage/extract", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ chatId }), + }); + if (!res.ok && res.status !== 409) { + console.error("manual extraction failed:", res.status); + setExtracting((ids) => ids.filter((id) => id !== chatId)); + return; + } + // 409 = already running (e.g. the trigger raced us): the row's + // running flag takes over on the next poll. + void load(); + } catch { + setExtracting((ids) => ids.filter((id) => id !== chatId)); + } + }; + + const pending = (chat: CoverageChat) => chat.running || extracting.includes(chat.chatId); + return ( +
+ + {open && ( +
+ pending(c)} + onExtract={(c) => void onExtract(c.chatId)} + /> +
+ )} +
+ ); +} + +/** + * The panel body — exported for static-render tests. Rows: exact coverage + * numbers per session, "Extract" for the uncovered ones, "Running" while a + * run is in flight. + */ +export function CoveragePanel({ + chats, + loaded, + isBusy, + onExtract, +}: { + chats: CoverageChat[]; + loaded: boolean; + isBusy: (chat: CoverageChat) => boolean; + onExtract: (chat: CoverageChat) => void; +}) { + const anyUncovered = chats.some((c) => c.uncoveredTokens > 0 || c.uncoveredMessages > 0); + return ( +
+
+ + Extraction coverage + + uncovered first +
+ {!anyUncovered && ( +

+ {loaded ? "Every session is fully extracted — nothing pending." : "…"} +

+ )} +
+ {chats.map((c) => ( + onExtract(c)} /> + ))} +
+
+ ); +} + +function CoverageRow({ chat, busy, onExtract }: { chat: CoverageChat; busy: boolean; onExtract: () => void }) { + return ( +
+
+
{chat.title}
+
+ {busy ? ( + + extracting… + + ) : ( + <> + {formatTokens(chat.coveredTokens)} covered + {chat.uncoveredTokens > 0 && ( + + {formatTokens(chat.uncoveredTokens)} uncovered + + )} + {chat.uncoveredTokens === 0 && · complete} + {chat.uncoveredMessages > 0 && ( + ({chat.uncoveredMessages} msgs) + )} + + )} +
+
+ +
+ ); +} \ No newline at end of file diff --git a/components/graph-view.tsx b/components/graph-view.tsx index 21dc619..1725e0b 100644 --- a/components/graph-view.tsx +++ b/components/graph-view.tsx @@ -66,6 +66,7 @@ import Link from "next/link"; import { FileTextIcon, Loader2Icon, SearchIcon, SettingsIcon, UploadIcon, WaypointsIcon } from "lucide-react"; import { incrementalLayout, type GraphNode, type GraphSnapshot } from "@/src/core"; import { GraphCanvasRenderer, type GraphRenderSettings } from "./graph-canvas.ts"; +import { GraphCoverageControl } from "./graph-coverage.tsx"; import { ARROW_LENGTH_RANGE, CHARGE_RANGE, @@ -843,11 +844,15 @@ export function GraphView() { )} {/* Legend + filters: search box and per-state toggle rows. */} - {/* Top-right controls — three separate floating containers (owner, + {/* Top-right controls — separate floating containers (owner, graph-view round 2): the search box, the settings gear (its own - collapsible component), and the import button. */} - {!isEmpty && ( -
+ collapsible component), the import button, and the coverage + control. Coverage renders even when the graph is EMPTY — an + empty graph with unextracted sessions is exactly when it + matters (extract → the graph gains nodes). */} +
+ {!isEmpty && ( + <> {/* Search — its own container (icon + input + no-match note). */}
@@ -893,8 +898,12 @@ export function GraphView() { gear sits at the far right (owner, round 3) so its panel drops at the very right edge, over less of the graph. */} -
- )} + + )} + {/* Extraction coverage — own container, always present; the + panel re-reads when background runs settle (refreshTick). */} + +
{/* Background pipeline status: what is currently running. */} {pipelineRuns.length > 0 && ( diff --git a/config.ts b/config.ts index 8162991..f01349c 100644 --- a/config.ts +++ b/config.ts @@ -62,7 +62,16 @@ export type ModelRole = keyof typeof MODELS; * (server/turn-service.ts) so it can be swapped without touching the * pipeline. */ -export const EXTRACTION = { everyNQuizAnswers: 3 } as const; +/** + * The extraction trigger (vision-extraction-coverage.md, 2026-09-15): a run + * fires when a chat's UNCOVERED transcript tokens reach the chunk budget. + * Runs are token-budgeted batches of the oldest uncovered messages — this + * knob is the transcript share of the extractor's context (it also carries + * the whole graph + instructions), so it bounds both cost and chunk size. + * Replaces the old every-3-quiz-answers trigger, which never fired for + * quiz-less sessions. + */ +export const EXTRACTION = { chunkTokens: 4096 } as const; /** * The LIVE graph store (src/store/graph.ts) — owned by the app's extraction @@ -74,6 +83,10 @@ export const LEDGER_FILES = { events: path.join(PATHS.graph, "events.jsonl"), mappings: path.join(PATHS.graph, "mappings.jsonl"), sessions: path.join(PATHS.graph, "sessions.jsonl"), + /** Per-message extraction coverage (contracts/graph.ts CoverageEntry; + * the checkpoint's replacement for chat extractions — notes keep + * sessions.jsonl, whole-file as they are). */ + coverage: path.join(PATHS.graph, "coverage.jsonl"), /** LLM-named community records (contracts/graph.ts ClusterLabeling). */ clusterLabels: path.join(PATHS.graph, "cluster-labels.jsonl"), } as const; diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 67157f6..119652b 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -18,6 +18,8 @@ import type { import { getGraph, mergeFindings, type EmissionStats, type GraphDeps } from "../../core/graph.ts"; import type { GraphStore } from "../../core/ports.ts"; import { settleGraphLayout } from "../../core/layout-settle.ts"; +import { batchRuns, batchTag, coverageOf, messageViewLines } from "../../core/coverage.ts"; +import { EXTRACTION } from "../../../config.ts"; import { refreshClusterLabels } from "./cluster-namer.ts"; /** @@ -71,15 +73,6 @@ import { refreshClusterLabels } from "./cluster-namer.ts"; * (question text + what the learner did — kind/correct are rebuilt by the * merger; the extractor only needs enough to attribute.) */ -function quizLine(part: { toolCallId: string; input: unknown; output: unknown }): string { - const input = part.input as { question?: unknown } | undefined; - const out = part.output as Partial<{ questionText: string; correct: boolean; dontKnow: boolean; note: string }> | undefined; - const question = typeof input?.question === "string" ? input.question : out?.questionText ?? "(no question)"; - const outcome = out?.dontKnow === true ? "did not know" : out?.correct === true ? "answered correctly" : "answered incorrectly"; - const note = typeof out?.note === "string" && out.note ? ` (note: ${out.note})` : ""; - return `QUIZ [${part.toolCallId}] ${question} — the learner ${outcome}${note}`; -} - /** * The compact transcript both LLM stages see: user/teacher text VERBATIM, * each tagged with its message id (provenance anchors — the extractor @@ -90,25 +83,12 @@ function quizLine(part: { toolCallId: string; input: unknown; output: unknown }) * are summarized as one line each. */ export function transcriptView(record: ChatRecord): string { - const lines: string[] = []; - lines.push(`SESSION: ${record.title}`); - + // The shared seam (core/coverage.ts messageViewLines): the LLM's prompt + // view and the coverage fingerprint use ONE line builder, so "covered" + // means — exactly — what the extractor saw. + const lines = [`SESSION: ${record.title}`]; record.messages.forEach((message, index) => { - const who = message.role === "user" ? "LEARNER" : "TEACHER"; - const mid = (message as UIMessage).id || `m${index}`; - for (const part of (message as UIMessage).parts) { - if (part.type === "text" && part.text.trim()) { - lines.push(`${who} [${mid}] ${part.text.trim()}`); - } else if (part.type.startsWith("tool-")) { - const name = part.type.slice("tool-".length); - const p = part as unknown as { toolCallId: string; state: string; input: unknown; output: unknown }; - if (name === "quiz" && p.state === "output-available") { - lines.push(quizLine(p)); - } else if (p.state === "output-available") { - lines.push(`(${name} tool result delivered)`); - } - } - } + lines.push(...messageViewLines(message as UIMessage, (message as UIMessage).id || `m${index}`)); }); return lines.join("\n"); } @@ -623,12 +603,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti }; } - async function extract(chatId: string): Promise { - const record = chatStore.getChat(chatId); - if (!record) throw new Error(`Chat not found: ${chatId}`); + async function extract(chatId: string, view: string): Promise { const signal = runningExtractions.get(chatId)?.controller.signal; signal?.throwIfAborted(); // an abort during a sync gap ends the run here - const view = transcriptView(record); console.info(`[extraction ${chatId}] extractor starting (${view.split("\n").length} transcript lines)`); const t0 = performance.now(); @@ -645,12 +622,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti return findings; } - async function audit(chatId: string, findings: SessionFindings): Promise { - const record = chatStore.getChat(chatId); - if (!record) throw new Error(`Chat not found: ${chatId}`); + async function audit(chatId: string, findings: SessionFindings, view: string): Promise { const signal = runningExtractions.get(chatId)?.controller.signal; signal?.throwIfAborted(); // an abort during the extract→audit gap ends here - const view = transcriptView(record); const graph = getGraph(graphStore); console.info( `[extraction ${chatId}] auditor starting (${findings.propositions.length} propositions, ${findings.edges.length} edges, ` + @@ -708,21 +682,66 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti if (inFlight.has(chatId)) return { nodes: 0, edges: 0, events: 0 }; inFlight.add(chatId); const t0 = performance.now(); + // COVERAGE (vision-extraction-coverage.md, 2026-09-15): extract ONLY + // what the per-chat coverage ledger doesn't already hold — never the + // whole session. Batches are token-budgeted runs of the oldest + // uncovered messages; each merged batch checkpoints its messages' + // coverage ATOMICALLY (inside the merger's step 4), so an abort or + // error below loses at most the in-flight batch — every earlier + // batch stays covered, the rest stays visibly uncovered. + let batchesDone = 0; + const total: EmissionStats = { nodes: 0, edges: 0, events: 0 }; try { - console.info(`[extraction ${chatId}] pipeline start`); - setPhase(chatId, "extracting", "chat"); - const findings = await extract(chatId); - setPhase(chatId, "auditing", "chat"); - const auditResult = await audit(chatId, findings); - setPhase(chatId, "merging", "chat"); - // The merger merges from the CURRENT record — events are rebuilt - // from the live transcript, never from stale copies. const record = chatStore.getChat(chatId); if (!record) throw new Error(`Chat not found: ${chatId}`); - const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }, { - messages: record.messages as UIMessage[], - foldedAt: record.updatedAt, - }); + const covered = coverageOf(record.messages as UIMessage[], graphStore.readCoverage(), chatId); + if (covered.uncovered.length === 0) { + console.info(`[extraction ${chatId}] nothing uncovered — no-op`); + return { nodes: 0, edges: 0, events: 0 }; + } + const batches = batchRuns(covered.uncovered, EXTRACTION.chunkTokens); + console.info( + `[extraction ${chatId}] pipeline start — ${covered.uncoveredTokens} uncovered tok, ` + + `${covered.uncoveredCount} messages, ${batches.length} batch(es) of ≤${EXTRACTION.chunkTokens} tok`, + ); + for (const batch of batches) { + // The batch's prompt view: SESSION header + the batch's lines, + // same provenance [id] tags the fingerprints used (coverage + // and prompt share ONE line builder — core/coverage.ts). + const mids = record.messages.map((m, i) => ((m as UIMessage).id || `m${i}`) as string); + const view = [ + `SESSION: ${record.title}`, + ...batch.flatMap((slot) => { + const idx = record.messages.indexOf(slot.message); + return messageViewLines(slot.message, mids[idx] ?? slot.message.id); + }), + ].join("\n"); + setPhase(chatId, "extracting", "chat"); + const findings = await extract(chatId, view); + setPhase(chatId, "auditing", "chat"); + const auditResult = await audit(chatId, findings, view); + setPhase(chatId, "merging", "chat"); + // The batch's messages, snapshotted when the run computed + // coverage — if a message mutates mid-run (a quiz answer + // merges in), the stored fingerprint won't match its new + // one and it re-opens: honest by construction. + const foldedAt = new Date().toISOString(); + const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }, { + messages: batch.map((slot) => slot.message), + foldedAt, + coverage: batch.map((slot) => ({ chatId, fp: slot.fp, tokens: slot.tokens, foldedAt })), + batchTag: batchTag(batch), + }); + total.nodes += stats.nodes; + total.edges += stats.edges; + total.events += stats.events; + batchesDone++; + console.info( + `[extraction ${chatId}] batch ${batchesDone}/${batches.length} merged: ` + + `${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events`, + ); + } + const stats = total; // Semantic embeddings (DECIDED 2026-09-13): topics come from // MEANING, so embed the new propositions BEFORE the settle and // the namer look at clusters. Failure is non-fatal — clustering @@ -747,18 +766,23 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti }).catch(() => {}); return stats; } catch (err) { - persistRunUsage(chatId, Math.round(performance.now() - t0), undefined); + persistRunUsage(chatId, Math.round(performance.now() - t0), batchesDone > 0 ? total : undefined); // Aborted (the dock's abort button): a deliberate stop, not a - // failure — same ledger discipline as a failure (nothing merged, - // nothing checkpointed, partial usage persisted), but the run - // reports zero emissions instead of throwing. + // failure. Per-batch coverage (above) makes the stop CHEAP: every + // merged batch stays covered; only the in-flight one doesn't — it + // stays uncovered, and the next trigger or manual run picks it + // up. Partial usage is persisted either way. if (err instanceof Error && err.name === "AbortError") { - console.info(`[extraction ${chatId}] aborted (${Math.round(performance.now() - t0)}ms) — nothing merged, nothing checkpointed`); - return { nodes: 0, edges: 0, events: 0 }; + console.info( + `[extraction ${chatId}] aborted (${Math.round(performance.now() - t0)}ms) — ` + + `${batchesDone} merged batch(es) stay covered, the in-flight one does not`, + ); + return total; } - // The throw travels to the caller; this line is what makes the - // run's log legible end-to-end (no hanging "pipeline start"). - console.error(`[extraction ${chatId}] pipeline failed (${Math.round(performance.now() - t0)}ms):`, err); + // A failure keeps the same discipline: merged batches stay + // covered, nothing else is checkpointed, and the throw travels + // to the caller (this line keeps the run's log legible). + console.error(`[extraction ${chatId}] pipeline failed (${Math.round(performance.now() - t0)}ms) after ${batchesDone} batch(es):`, err); throw err; } finally { inFlight.delete(chatId); diff --git a/src/core/contracts/graph.ts b/src/core/contracts/graph.ts index 3a69258..1ca8a75 100644 --- a/src/core/contracts/graph.ts +++ b/src/core/contracts/graph.ts @@ -116,6 +116,24 @@ export interface SessionCheckpoint { foldedAt: string; } +/** coverage.jsonl line (vision-extraction-coverage.md, settled 2026-09-15): + * ONE MESSAGE of one chat has been extracted — by content fingerprint over + * exactly what the extraction view sees. The MESSAGE is the unit of + * coverage; "uncovered" is derived (fingerprint the current transcript, + * diff) — never dual-tracked. A quiz answer merging into the asking + * message changes its fingerprint: the message re-opens itself, only + * itself. Chat extractions write these instead of whole-chat checkpoints; + * notes imports keep the checkpoint (a file is one shot). */ +export interface CoverageEntry { + chatId: string; + /** sha256 of the message's view lines — "covered" means exactly this. */ + fp: string; + /** chars/4 estimate of the message's view — display + budgeting. */ + tokens: number; + /** ISO — when this message's last extraction run merged. */ + foldedAt: string; +} + /** * Event kinds (DECIDED 2026-09-02): quiz outcomes only — correct / miss / * dont-know — plus `observation`, the extractor's per-proposition diff --git a/src/core/coverage.ts b/src/core/coverage.ts new file mode 100644 index 0000000..2587ac0 --- /dev/null +++ b/src/core/coverage.ts @@ -0,0 +1,174 @@ +/** + * Extraction coverage (vision-extraction-coverage.md, settled 2026-09-15) — + * the pure core. THE MESSAGE IS THE UNIT OF COVERAGE: a per-chat append-only + * ledger records which messages have been extracted, by CONTENT FINGERPRINT + * over exactly what the extraction view sees; "uncovered" is DERIVED at + * read time (fingerprint the current transcript, diff against the ledger) — + * never dual-tracked. Runs are token-budgeted batches of the oldest + * uncovered messages, so a tool call and its result (which share a message) + * can never be separated. + * + * The fingerprint and the LLM prompt share ONE line builder — + * messageViewLines — so "covered" and "what the model saw" cannot drift + * apart. The line logic is tree-blind, identical to the extractor's view: + * user/teacher text verbatim, quiz outcomes as one-line summaries, other + * tool results as one generic line each, data parts invisible (plan-block + * attach does not re-open a covered message). + */ + +import { createHash } from "node:crypto"; +import type { UIMessage } from "ai"; +import { isToolUIPart } from "ai"; + +import type { CoverageEntry } from "./contracts/graph.ts"; + +export type { CoverageEntry }; + +/** The quiz outcome line — both the fingerprint and the LLM view see the + * quiz answer, so an answered quiz re-opens the message that asked it. */ +function quizLine(part: { toolCallId: string; input: unknown; output: unknown }): string { + const input = part.input as { question?: unknown } | undefined; + const out = part.output as Partial<{ questionText: string; correct: boolean; dontKnow: boolean; note: string }> | undefined; + const question = typeof input?.question === "string" ? input.question : out?.questionText ?? "(no question)"; + const outcome = out?.dontKnow === true ? "did not know" : out?.correct === true ? "answered correctly" : "answered incorrectly"; + const note = typeof out?.note === "string" && out.note ? ` (note: ${out.note})` : ""; + return `QUIZ [${part.toolCallId}] ${question} — the learner ${outcome}${note}`; +} + +/** + * The extraction-relevant lines of ONE message — the shared seam between + * transcriptView (the LLM's prompt) and messageFingerprint (coverage). Same + * logic, same order, so a covered message is by construction exactly what + * the extractor saw. `mid` is the provenance tag (the message id; the + * transcript view passes an index fallback for legacy id-less messages, + * coverage passes the id — persisted messages always have ids). + */ +export function messageViewLines(message: UIMessage, mid: string): string[] { + const lines: string[] = []; + const who = message.role === "user" ? "LEARNER" : "TEACHER"; + for (const part of message.parts) { + if (part.type === "text" && part.text.trim()) { + lines.push(`${who} [${mid}] ${part.text.trim()}`); + } else if (isToolUIPart(part)) { + const name = part.type.slice("tool-".length); + if (name === "quiz" && part.state === "output-available") { + lines.push(quizLine(part as unknown as { toolCallId: string; input: unknown; output: unknown })); + } else if (part.state === "output-available") { + lines.push(`(${name} tool result delivered)`); + } + } + } + return lines; +} + +/** Tokens of a message's view — chars/4, ceil, the app-wide estimate. */ +export function viewTokens(lines: string[]): number { + return Math.ceil(lines.join("\n").length / 4); +} + +/** + * A message's coverage fingerprint: sha256 of role + its view lines (with + * the message's real id as the provenance tag). Changes exactly when the + * extraction view of the message changes: an answered quiz merges into the + * asking message → new line → the message re-opens itself, and only itself. + * Plan-block part attach changes data parts (invisible to the view) → the + * fingerprint is stable → still covered. + */ +export function messageFingerprint(message: UIMessage): string { + const h = createHash("sha256"); + h.update(JSON.stringify([message.role, messageViewLines(message, message.id)])); + return h.digest("hex"); +} + +/** One message's coverage-relevant projection. */ +export interface CoverageSlot { + message: UIMessage; + fp: string; + tokens: number; +} + +/** Fingerprint a whole transcript, in order (used by the trigger, the run, + * and the coverage view — one derivation everywhere). */ +export function coverageSlots(messages: ReadonlyArray): CoverageSlot[] { + return messages.map((message) => { + const lines = messageViewLines(message, message.id); + const h = createHash("sha256"); + h.update(JSON.stringify([message.role, lines])); + return { message, fp: h.digest("hex"), tokens: viewTokens(lines) }; + }); +} + +/** The covered-fingerprint set for one chat, from the ledger. */ +export function coveredSet(entries: ReadonlyArray, chatId: string): Set { + const out = new Set(); + for (const e of entries) if (e.chatId === chatId) out.add(e.fp); + return out; +} + +/** The derived state: which slots are uncovered, plus the token totals. + * "Uncovered" needs no ledger of its own — this diff IS the coverage view. */ +export function coverageOf( + messages: ReadonlyArray, + entries: ReadonlyArray, + chatId: string, +): { + slots: CoverageSlot[]; + uncovered: CoverageSlot[]; + coveredTokens: number; + uncoveredTokens: number; + coveredCount: number; + uncoveredCount: number; +} { + const covered = coveredSet(entries, chatId); + const slots = coverageSlots(messages); + const uncovered = slots.filter((s) => !covered.has(s.fp)); + const coveredCount = slots.length - uncovered.length; + let coveredTokens = 0; + let uncoveredTokens = 0; + for (const s of slots) { + if (covered.has(s.fp)) coveredTokens += s.tokens; + else uncoveredTokens += s.tokens; + } + return { slots, uncovered, coveredTokens, uncoveredTokens, coveredCount, uncoveredCount: uncovered.length }; +} + +/** + * Batch the uncovered messages into token-budgeted RUNS: consecutive slots, + * oldest first, greedy fill — a message is never split (it is the unit), so + * a tool call and its result always travel together. A single message larger + * than the budget becomes its own (over-budget) run rather than being + * dropped. Runs are ephemeral — the durable record is per-message coverage, + * checkpointed after each run merges. + */ +export function batchRuns(uncovered: ReadonlyArray, budget: number): CoverageSlot[][] { + if (budget <= 0 || uncovered.length === 0) return []; + const runs: CoverageSlot[][] = []; + let current: CoverageSlot[] = []; + let currentTokens = 0; + for (const slot of uncovered) { + if (current.length > 0 && currentTokens + slot.tokens > budget) { + runs.push(current); + current = []; + currentTokens = 0; + } + current.push(slot); + currentTokens += slot.tokens; + } + if (current.length > 0) runs.push(current); + return runs; +} + +/** + * A batch's deterministic tag: hash of its member fingerprints. This is the + * observation-ref prefix for the run (obs:: in the merger): + * re-running the SAME batch (crash recovery) produces the same refs and + * dedupes; a re-opened (mutated) message changes the tag, so its fresh + * understanding observations land as new events — the most-recent fold + * sees the update. Run-local source ids ("p1"…) alone could never do this: + * their positions shift between runs. + */ +export function batchTag(batch: ReadonlyArray): string { + const h = createHash("sha256"); + for (const slot of batch) h.update(slot.fp); + return h.digest("hex").slice(0, 12); +} \ No newline at end of file diff --git a/src/core/graph.ts b/src/core/graph.ts index 1c4863c..986c81b 100644 --- a/src/core/graph.ts +++ b/src/core/graph.ts @@ -40,6 +40,7 @@ import { createHash, randomBytes } from "node:crypto"; import { isToolUIPart, type UIMessage } from "ai"; +import type { CoverageEntry } from "./contracts/graph.ts"; import type { AuditResult, Cluster, @@ -67,14 +68,29 @@ function newGraphNodeId(): string { } /** What one merge run merges FROM (the caller supplies it — the merger never - * reads the chat store): a session's transcript + its `updatedAt` - * checkpoint, or — for the format-agnostic notes front door (2026-09-13) — - * `messages: null`: raw notes carry no tool parts, so no quiz events are - * rebuilt (quiz outcomes inside notes surface only through the extractor's - * understanding observations), and the checkpoint stamps the import. */ + * reads the chat store): a session's transcript + its coverage checkpoint, + * or — for the format-agnostic notes front door (2026-09-13) — `messages: + * null`: raw notes carry no tool parts, so no quiz events are rebuilt (quiz + * outcomes inside notes surface only through the extractor's understanding + * observations), and the whole-file checkpoint stamps the import. + * + * Coverage (vision-extraction-coverage.md, settled 2026-09-15): a chat merge + * passes the BATCH it extracted — the caller computed the fingerprints + * before the run, from the same snapshot the merger sees — and step 4 + * appends one CoverageEntry per message on success ("covered" means + * exactly this merge happened). `batchTag` prefixes the observation refs so + * re-runs of the SAME batch dedupe while a re-opened (mutated) message's + * fresh understanding observations land as new events; the notes path + * passes neither and keeps the whole-file checkpoint. */ export interface MergeSource { messages: UIMessage[] | null; foldedAt: string; + /** The batch being merged (chat path): one entry per message in + * `messages`, same order — undefined (notes path / coverage-less + * callers) → the legacy whole-chat checkpoint is written instead. */ + coverage?: CoverageEntry[]; + /** The batch's deterministic tag (core/coverage.ts batchTag) — obs refs. */ + batchTag?: string; } // ─── transcript scanning (shared by the merger + trigger) ───────────────────── @@ -241,10 +257,13 @@ export function mergeFindings(chatId: string, audit: AuditResult, deps: GraphDep } } // Understanding observations: one per proposition carrying a note, - // resolved to the node the verdict mapped it to. + // resolved to the node the verdict mapped it to. Refs carry the batch + // tag when there is one: run-local source ids ("p1"…) shift position + // between runs, so a bare obs:p1 could never both dedupe a re-run of + // the same batch AND let a re-opened message's fresh level through. for (const p of audit.propositions) { if (p.understanding === undefined || p.understanding.length === 0) continue; - const ref = `obs:${p.id}`; + const ref = source.batchTag !== undefined ? `obs:${source.batchTag}:${p.id}` : `obs:${p.id}`; if (seenRefs.has(ref)) continue; ledgers.appendEvent({ nodeId: mapping.get(p.id) ?? null, @@ -258,9 +277,17 @@ export function mergeFindings(chatId: string, audit: AuditResult, deps: GraphDep stats.events++; } - // 4. Checkpoint: this source is extracted through its foldedAt (the - // chat's updatedAt, or the import time for notes). - ledgers.appendCheckpoint({ chatId, foldedAt: source.foldedAt }); + // 4. Checkpoint — THE COVERAGE DISCIPLINE (vision-extraction-coverage.md): + // only now, after every append succeeded, is the batch covered. The + // chat path appends one CoverageEntry per message (per-batch atomic — + // an abort or failure mid-run leaves earlier batches covered, this + // batch's messages uncovered); the notes path keeps the whole-file + // checkpoint (a file is one shot). + if (source.coverage !== undefined) { + for (const entry of source.coverage) ledgers.appendCoverage(entry); + } else { + ledgers.appendCheckpoint({ chatId, foldedAt: source.foldedAt }); + } return stats; } @@ -302,34 +329,6 @@ function quizEvent( return event; } -// ─── the trigger helper (v1: every N quiz answers — swappable seam) ────────── - -/** - * How many ANSWERED quiz parts in this chat are not yet in the ledger — i.e. - * new evidence since the last extraction. The v1 trigger (turn-service) - * fires the pipeline when this crosses EXTRACTION.everyNQuizAnswers. Every - * answered quiz always becomes an event (attribution only decides nodeId), - * so ref-not-in-ledger is a correct "not yet processed" marker. - */ -export function pendingQuizAnswers(chatId: string, deps: GraphDeps): number { - const record = deps.chatStore.getChat(chatId); - if (!record) return 0; - const seenRefs = new Set(); - for (const event of deps.graphStore.readEvents()) { - if (event.chatId === chatId && event.ref) seenRefs.add(event.ref); - } - let count = 0; - for (const part of toolParts(record.messages)) { - if (part.name !== "quiz" || part.state !== "output-available") continue; - const out = part.output; - if (typeof out !== "object" || out === null) continue; - const answer = out as Partial; - if (answer.kind !== "quiz" || answer.dismissed === true) continue; // dismissed = not answered - if (!seenRefs.has(part.toolCallId)) count++; - } - return count; -} - // ─── derivation (reads only — everything below is recomputable) ───────────── /** The content identity of a community: "clu_" + sha256 of the sorted diff --git a/src/core/index.ts b/src/core/index.ts index 753794b..2405463 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -34,4 +34,5 @@ export * from "./ui-message.ts"; export * from "./tool-parts.ts"; export * from "./prices.ts"; export * from "./fuzzy.ts"; +export * from "./coverage.ts"; export * from "./graph-context.ts"; diff --git a/src/core/ports.ts b/src/core/ports.ts index 9f9d893..aa9e3d7 100644 --- a/src/core/ports.ts +++ b/src/core/ports.ts @@ -23,6 +23,7 @@ import type { NodeMapping, PlanTree, Position, + CoverageEntry, SessionCheckpoint, TurnUsage, NotesSource, @@ -84,6 +85,11 @@ export interface GraphStore { /** Append one emission checkpoint. */ appendCheckpoint(checkpoint: SessionCheckpoint): void; readCheckpoints(): SessionCheckpoint[]; + /** Message-granular extraction coverage (vision-extraction-coverage.md): + * append-only; one line per covered message. The chat-extraction + * checkpoint — "uncovered" is derived from these, never tracked. */ + appendCoverage(entry: CoverageEntry): void; + readCoverage(): CoverageEntry[]; /** Append one cluster labeling — idempotent by id (the id hashes the * sorted core, so identical member sets dedupe). */ diff --git a/src/server/turn-service.ts b/src/server/turn-service.ts index ff120cb..62f265e 100644 --- a/src/server/turn-service.ts +++ b/src/server/turn-service.ts @@ -17,7 +17,7 @@ import type { UIMessage } from "ai"; import { withPlanBlocks } from "../core/plan-blocks.ts"; -import { pendingQuizAnswers } from "../core/graph.ts"; +import { coverageOf } from "../core/coverage.ts"; import type { ChatStore, GraphStore } from "../core/ports.ts"; import { EXTRACTION } from "../../config.ts"; import type { ExtractionPipeline } from "../agent/subagents/extraction.ts"; @@ -74,15 +74,17 @@ export function persistTurn(chatStore: ChatStore, chatId: string, messages: UIMe /** * The graph write channel (vision §1 construction): the extraction pipeline * (extractor → auditor → merger — the sole graph writer) runs ASYNC once a - * chat accumulates EXTRACTION.everyNQuizAnswers new answered quizzes since - * its last checkpoint. Fire-and-forget — never blocks the turn; failures - * are logged by the pipeline and leave the checkpoint untouched, so the - * next trigger re-runs the whole thing. + * chat's UNCOVERED transcript tokens reach EXTRACTION.chunkTokens + * (vision-extraction-coverage.md, 2026-09-15 — replacing the old + * every-3-quiz-answers count: content size is the honest signal, and + * quiz-less sessions extract too). The pipeline itself extracts only the + * uncovered batches. Fire-and-forget — never blocks the turn; failures are + * logged by the pipeline and leave coverage untouched, so the next trigger + * (or the manual extract button) re-runs what's left. * - * v1 placeholder trigger, deliberately isolated in this one seam: swap - * here when a better mechanism lands — the pipeline is trigger-agnostic. - * Skipped under tests: scripted chats are fixtures, not knowledge (the - * pipeline itself is tested directly with mock models). + * Deliberately isolated in this one seam — the pipeline is + * trigger-agnostic. Skipped under tests: scripted chats are fixtures, not + * knowledge (the pipeline itself is tested directly with mock models). */ export function triggerExtraction( deps: { chatStore: ChatStore; graphStore: GraphStore }, @@ -91,7 +93,10 @@ export function triggerExtraction( ): void { if (process.env.NODE_ENV === "test") return; try { - if (pendingQuizAnswers(chatId, deps) < EXTRACTION.everyNQuizAnswers) return; + const record = deps.chatStore.getChat(chatId); + if (!record) return; + const covered = coverageOf(record.messages as UIMessage[], deps.graphStore.readCoverage(), chatId); + if (covered.uncoveredTokens < EXTRACTION.chunkTokens) return; void extraction.runExtraction(chatId).catch((err) => { console.error(`[chat ${chatId}] extraction pipeline failed:`, err); }); diff --git a/src/store/graph.ts b/src/store/graph.ts index 8c72007..b18e6de 100644 --- a/src/store/graph.ts +++ b/src/store/graph.ts @@ -42,6 +42,7 @@ import type { NodeMapping, Position, ProvenancePoint, + CoverageEntry, SessionCheckpoint, } from "../core/contracts.ts"; import type { GraphStore } from "../core/ports.ts"; @@ -133,6 +134,9 @@ const isNodeMapping = (v: unknown): v is NodeMapping => const isCheckpoint = (v: unknown): v is SessionCheckpoint => isRecord(v) && isString(v.chatId) && isString(v.foldedAt); +const isCoverageEntry = (v: unknown): v is CoverageEntry => + isRecord(v) && isString(v.chatId) && isString(v.fp) && typeof v.tokens === "number" && isString(v.foldedAt); + // ─── the port implementation ──────────────────────────────────────────────── export const graphStore: GraphStore = { @@ -175,6 +179,13 @@ export const graphStore: GraphStore = { return readValid(LEDGER_FILES.sessions, isCheckpoint); }, + appendCoverage(entry: CoverageEntry): void { + appendLine(LEDGER_FILES.coverage, entry); + }, + readCoverage(): CoverageEntry[] { + return readValid(LEDGER_FILES.coverage, isCoverageEntry); + }, + // ── cluster labelings (config.LEDGER_FILES.clusterLabels) ── /** Idempotent by id: the id is a hash of the sorted core, so re-appending diff --git a/tests/coverage.test.ts b/tests/coverage.test.ts new file mode 100644 index 0000000..3b7fdb5 --- /dev/null +++ b/tests/coverage.test.ts @@ -0,0 +1,159 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import type { UIMessage } from "ai"; + +import { + batchRuns, + batchTag, + coverageOf, + coverageSlots, + messageFingerprint, + messageViewLines, +} from "../src/core/coverage.ts"; +import type { CoverageEntry } from "../src/core/contracts.ts"; + +/** + * Extraction coverage — the pure core (vision-extraction-coverage.md, + * settled 2026-09-15). THE MESSAGE IS THE UNIT OF COVERAGE; the fingerprint + * and the LLM prompt share ONE line builder (messageViewLines), so "covered" + * is by construction exactly "what the extractor saw". + */ + +function user(id: string, text: string): UIMessage { + return { id, role: "user", parts: [{ type: "text", text }] } as unknown as UIMessage; +} + +function slot(messages: UIMessage[], id: string) { + const found = coverageSlots(messages).find((s) => s.message.id === id); + assert.ok(found, `slot ${id}`); + return found; +} + +describe("messageFingerprint — what changes it, what never does", () => { + it("a text edit changes the fingerprint; role participates", () => { + const a = messageFingerprint(user("m1", "hello world")); + const b = messageFingerprint(user("m1", "hello worlds")); + const same = messageFingerprint(user("m1", "hello world")); + assert.notEqual(a, b); + assert.equal(a, same); + // Role is hashed too — a user and an assistant line never collide. + const teacher = { id: "m1", role: "assistant", parts: [{ type: "text", text: "hello world" }] } as unknown as UIMessage; + assert.notEqual(a, messageFingerprint(teacher)); + }); + + it("the message id is the provenance tag — part of exactly what the extractor saw", () => { + // The view lines carry the [id] tag, so identical content under a + // different id is a DIFFERENT fingerprint: coverage matches the + // extraction view byte-for-byte. (Ids are stable in practice — the + // write-ahead persist owns them.) + assert.notEqual(messageFingerprint(user("m1", "same")), messageFingerprint(user("m2", "same"))); + // The id is the ONLY tag: the same id + content re-fingerprints equal. + assert.equal(messageFingerprint(user("m1", "same")), messageFingerprint(user("m1", "same"))); + }); + + it("data parts are invisible — plan-block attach never re-opens a covered message", () => { + const plain = user("m1", "teach me the chain rule"); + const withData = { + ...plain, + parts: [...plain.parts, { type: "data-plan-blocks", planBlocks: [] }], + } as unknown as UIMessage; + assert.equal(messageFingerprint(plain), messageFingerprint(withData)); + }); + + it("a quiz answer merging in changes the fingerprint — and only that message's", () => { + const asking = { + id: "a1", + role: "assistant", + parts: [{ type: "tool-quiz", toolCallId: "q1", state: "input-available", input: { question: "Q?" } }], + } as unknown as UIMessage; + const answered = { + id: "a1", + role: "assistant", + parts: [ + { type: "tool-quiz", toolCallId: "q1", state: "output-available", input: { question: "Q?" }, output: { questionText: "Q?", correct: true } }, + ], + } as unknown as UIMessage; + const fpBefore = messageFingerprint(asking); + assert.notEqual(fpBefore, messageFingerprint(answered)); + // An output-bearing line's mass counts toward coverage tokens. + assert.ok( + messageViewLines(answered, answered.id).join("").length > + messageViewLines(asking, asking.id).join("").length, + ); + }); +}); + +describe("coverageOf — uncovered is DERIVED, never dual-tracked", () => { + it("diffs fingerprints: covered stays, uncovered counts, tokens split", () => { + const messages = [user("m1", "aaaa"), user("m2", "bbbb"), user("m3", "cccc")]; + const fp1 = slot(messages, "m1").fp; + const fp2 = slot(messages, "m2").fp; + const entries: CoverageEntry[] = [ + { chatId: "c1", fp: fp1, tokens: 1, foldedAt: "2026-09-15T00:00:00.000Z" }, + { chatId: "c1", fp: fp2, tokens: 1, foldedAt: "2026-09-15T00:00:00.000Z" }, + // Another chat's entries are irrelevant to c1's coverage. + { chatId: "c2", fp: slot(messages, "m3").fp, tokens: 1, foldedAt: "2026-09-15T00:00:00.000Z" }, + ]; + const cov = coverageOf(messages, entries, "c1"); + assert.deepEqual(cov.uncovered.map((s) => s.message.id), ["m3"]); + assert.equal(cov.coveredCount, 2); + assert.equal(cov.uncoveredCount, 1); + assert.equal(cov.uncoveredTokens, slot(messages, "m3").tokens); + assert.equal(cov.coveredTokens, slot(messages, "m1").tokens + slot(messages, "m2").tokens); + }); + + it("a mutated covered message re-opens itself, and only itself", () => { + const messages = [user("m1", "aaaa"), user("m2", "bbbb")]; + const entries: CoverageEntry[] = coverageSlots(messages).map((s) => ({ + chatId: "c1", + fp: s.fp, + tokens: s.tokens, + foldedAt: "2026-09-15T00:00:00.000Z", + })); + assert.equal(coverageOf(messages, entries, "c1").uncoveredCount, 0); + const mutated = [user("m1", "aaaa!"), messages[1]!]; + const cov = coverageOf(mutated, entries, "c1"); + assert.deepEqual(cov.uncovered.map((s) => s.message.id), ["m1"]); + }); +}); + +describe("batchRuns — token-budgeted, consecutive, never splits a message", () => { + it("greedy oldest-first fill; the overflowing message starts the next run", () => { + // Each message's view line is ~14 tok (chars/4) → budget 45 fits + // three, the fourth overflows: [m1,m2,m3] + [m4]. + const slots = coverageSlots([ + user("m1", "a".repeat(40)), + user("m2", "b".repeat(40)), + user("m3", "c".repeat(40)), + user("m4", "d".repeat(40)), + ]); + const runs = batchRuns(slots, 45); + assert.deepEqual(runs.map((r) => r.map((s) => s.message.id)), [["m1", "m2", "m3"], ["m4"]]); + }); + + it("a single message larger than the budget becomes its own over-budget run", () => { + const slots = coverageSlots([user("m1", "x".repeat(400)), user("m2", "y".repeat(20))]); + const runs = batchRuns(slots, 10); + assert.deepEqual(runs.map((r) => r.map((s) => s.message.id)), [["m1"], ["m2"]]); + }); + + it("an empty or non-positive budget never drops messages into a void", () => { + assert.deepEqual(batchRuns(coverageSlots([user("m1", "hi")]), 0), []); + assert.deepEqual(batchRuns([], 4096), []); + }); +}); + +describe("batchTag — deterministic, ref-stable, mutation-sensitive", () => { + it("the same members in the same order → the same 12-hex tag; a different order is a different batch", () => { + const slots = coverageSlots([user("m1", "aaaa"), user("m2", "bbbb")]); + assert.equal(batchTag(slots), batchTag(coverageSlots([user("m1", "aaaa"), user("m2", "bbbb")]))); + assert.notEqual(batchTag(slots), batchTag([...slots].reverse()), "batches are ordered runs"); + assert.match(batchTag(slots), /^[0-9a-f]{12}$/); + }); + + it("a re-opened member changes the tag — its observations land as new events", () => { + const slots = coverageSlots([user("m1", "aaaa"), user("m2", "bbbb")]); + const mutated = coverageSlots([user("m1", "aaaa!"), user("m2", "bbbb")]); + assert.notEqual(batchTag(slots), batchTag(mutated)); + }); +}); \ No newline at end of file diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index 24e90ca..f713fe9 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -28,6 +28,7 @@ import type { PiEngine } from "../src/agent/models.ts"; import type { AuditResult, ClusterLabeling, + CoverageEntry, ExtractionRunUsage, GraphEdge, GraphNode, @@ -39,6 +40,7 @@ import type { } from "../src/core/contracts.ts"; import type { GraphStore } from "../src/core/ports.ts"; import { abortExtraction, createExtractionPipeline, extractionStatuses, transcriptView } from "../src/agent/subagents/extraction.ts"; +import { batchTag, coverageOf, coverageSlots } from "../src/core/coverage.ts"; import { clusterNamingSettled, refreshClusterLabels } from "../src/agent/subagents/cluster-namer.ts"; import { semanticClusters } from "../src/core/semantic-clusters.ts"; import { chatStore } from "../src/store/chats.ts"; @@ -65,6 +67,7 @@ function memoryLedgers() { const events: LedgerEvent[] = []; const mappings: NodeMapping[] = []; const checkpoints: SessionCheckpoint[] = []; + const coverage: CoverageEntry[] = []; const sources = new Map(); const clusterLabels: ClusterLabeling[] = []; const positions = new Map(); @@ -86,6 +89,10 @@ function memoryLedgers() { readMappings: () => mappings.map((m) => ({ ...m })), appendCheckpoint: (c: SessionCheckpoint) => void checkpoints.push(c), readCheckpoints: () => checkpoints.map((c) => ({ ...c })), + // Extraction coverage (vision-extraction-coverage.md) — append-only; + // the chat pipeline checkpoints one entry per covered message. + appendCoverage: (e) => void coverage.push({ ...e }), + readCoverage: () => coverage.map((e) => ({ ...e })), upsertSource: (s: NotesSource) => { // Mirrors the real store: an upsert KEEPS the existing usage log. const prev = sources.get(s.id); @@ -356,20 +363,35 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { // Events rebuilt deterministically: quizzes in transcript order, then // the observation. check-1's attribution pointed at the DROPPED p2 → // unresolvable → null (the question text is kept for later attribution). + // The observation ref is BATCH-TAGGED (obs::p1) so a re-run of + // the same batch dedupes while a re-opened message's update lands as + // a new event (vision-extraction-coverage.md). const events = ledgers.readEvents(); + const record = chatStore.getChat(chatId)!; + const expectedTag = batchTag(coverageSlots(record.messages as UIMessage[])); assert.deepEqual( events.map((e) => [e.kind, e.ref, e.nodeId]), [ ["miss", "probe-1", existingId], ["correct", "check-1", null], - ["observation", "obs:p1", existingId], + ["observation", `obs:${expectedTag}:p1`, existingId], ], ); assert.equal(events[0]!.nodeId, existingId, "probe attributed via the verdict's mapping"); assert.equal(events[2]!.note, "solid", "the understanding note joins the ledger"); - // Checkpointed through the record's updatedAt — never the same turns twice. - assert.equal(ledgers.readCheckpoints().length, 1); + // COVERAGE checkpointed per message (not a whole-chat checkpoint): + // one entry per transcript message, fingerprinted content each — + // the same turns are never extracted twice. + const covered = ledgers.readCoverage(); + assert.equal(covered.length, record.messages.length); + assert.ok(covered.every((c) => c.chatId === chatId && c.fp.length === 64 && c.tokens > 0)); + assert.equal(ledgers.readCheckpoints().length, 0, "the chat path writes coverage, not the legacy checkpoint"); + assert.equal( + coverageOf(record.messages as UIMessage[], covered, chatId).uncoveredCount, + 0, + "every message is now covered", + ); }); it("checkpoints nothing when a stage fails", async () => { @@ -791,3 +813,225 @@ describe("the notes front door (importNotes — format-agnostic)", () => { assert.equal(ledgers.readSources()[0]!.extractions![0]!.stats, undefined, "never merged → no stats"); }); }); + +// ── the chunked pipeline (extraction coverage — vision-extraction-coverage.md) ── + +/** Pull the pipeline's prompt view out of a model call: the pipeline sends + * `Transcript:\n\n` as the user turn, so the LAST user message's + * text IS what the extractor saw. */ +function capturedView(context: unknown): string { + const messages = (context as { messages?: { role: string; content: string }[] }).messages ?? []; + const last = [...messages].reverse().find((m) => m.role === "user"); + return typeof last?.content === "string" ? last.content : ""; +} + +/** An extractor step that submits FINDINGS, optionally capturing its view. */ +function findingsStep(capture?: (view: string) => void): FauxResponseStep { + return (context) => { + capture?.(capturedView(context)); + return fauxAssistantMessage( + [fauxToolCall("submit_findings", FINDINGS as unknown as Record, { id: "findings-call" })], + { stopReason: "toolUse" }, + ); + }; +} + +/** An auditor step that passes p1 through (no linkTo — a fresh node), drops p2. */ +function passThroughAuditStep(): FauxResponseStep { + return () => + fauxAssistantMessage( + [ + fauxToolCall( + "submit_review", + { + propositions: [{ ...FINDINGS.propositions[0]! }], + edges: [], + attribution: { quiz: {} }, + dropped: [{ ref: "p2", reason: "evidence not found" }], + } as unknown as Record, + { id: "review-call" }, + ), + ], + { stopReason: "toolUse" }, + ); +} + +/** A session with UNCOVERED-multiple-batch mass: five long distinct messages + * (~1200 tok each — 3600+1200 > EXTRACTION.chunkTokens 4096 → batches of + * [m1,m2,m3] and [m4,m5]). */ +function chunkyChat() { + const chat = chatStore.createChat("Chunky coverage fixture"); + createdChats.push(chat.id); + // ~1230 tok per message (chars/4): three fill the 4096 budget, the + // fourth would overflow — deterministic batches [m1,m2,m3] + [m4,m5]. + chatStore.saveMessages( + chat.id, + [1, 2, 3, 4, 5].map((n) => user(`u${n}`, `topic ${n}: ${"detail ".repeat(700)}${n}`)), + ); + return { chatId: chat.id, record: chatStore.getChat(chat.id)! }; +} + +describe("the chunked pipeline (extraction coverage)", () => { + it("extracts uncovered messages in token-budgeted batches, checkpoints coverage per batch, and no-ops once covered", async () => { + const { chatId, record } = chunkyChat(); + const ledgers = memoryLedgers(); + const views: string[] = []; + + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + extractor: fauxEngine([findingsStep((v) => views.push(v)), findingsStep((v) => views.push(v))]), + auditor: fauxEngine([passThroughAuditStep(), passThroughAuditStep()]), + labeler: submitModel("submit_labels", { labels: [] }), + }); + const stats = await pipeline.runExtraction(chatId); + + // Two batches: the first held messages 1-3, the second 4-5 — a + // message is never split, batches are consecutive oldest-first. + assert.equal(views.length, 2, "one extractor call per batch"); + assert.ok(views[0]!.includes("LEARNER [u1]") && views[0]!.includes("LEARNER [u3]")); + assert.ok(!views[0]!.includes("LEARNER [u4]"), "the batch stops at the token budget"); + assert.ok(views[1]!.includes("LEARNER [u4]") && views[1]!.includes("LEARNER [u5]")); + assert.ok(!views[1]!.includes("LEARNER [u3]")); + + // p1 was a fresh node in batch 1; batch 2's p1 mapped to the same + // node (chat-scoped mapping) — but its understanding note is a NEW + // batch-tagged observation. + assert.deepEqual(stats, { nodes: 1, edges: 0, events: 2 }); + + // Per-batch coverage: every message checkpointed, nothing legacy. + const coverage = ledgers.readCoverage(); + assert.equal(coverage.length, record.messages.length); + assert.equal(ledgers.readCheckpoints().length, 0); + assert.equal(coverageOf(record.messages as UIMessage[], coverage, chatId).uncoveredCount, 0); + + // A second run is a NO-OP: nothing uncovered → no model calls, no + // new usage entry (no money spent on covered ground). + const before = views.length; + const again = await pipeline.runExtraction(chatId); + assert.deepEqual(again, { nodes: 0, edges: 0, events: 0 }); + assert.equal(views.length, before); + assert.equal(chatStore.getChat(chatId)!.extractions?.length, 1, "the no-op run persists no usage"); + }); + + it("aborting after a merged batch keeps it covered; the re-run extracts only the remainder", async () => { + const { chatId, record } = chunkyChat(); + const ledgers = memoryLedgers(); + const views: string[] = []; + + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + // Batch 1 submits and merges; batch 2's extractor call HANGS + // until the abort fires — the dock's abort button, mid-run. + extractor: fauxEngine([ + findingsStep((v) => views.push(v)), + (_context, options) => + new Promise((resolve) => { + const sig = options?.signal; + if (sig?.aborted) { + resolve(fauxAssistantMessage("late")); + return; + } + sig?.addEventListener("abort", () => resolve(fauxAssistantMessage("late")), { once: true }); + }), + ]), + auditor: fauxEngine([passThroughAuditStep()]), + labeler: submitModel("submit_labels", { labels: [] }), + }); + const runPromise = pipeline.runExtraction(chatId); + // The abort must land while batch 2's extractor hangs. The faux + // stream only resolves on abort, so wait for the batch-1 merge to + // show in coverage, then abort. + await new Promise((resolve) => { + const check = setInterval(() => { + if (ledgers.readCoverage().length > 0) { + clearInterval(check); + resolve(); + } + }, 5); + }); + assert.equal(abortExtraction(chatId), true); + const stats = await runPromise; + + // Batch 1's emissions are kept and its messages stay covered. + assert.deepEqual(stats, { nodes: 1, edges: 0, events: 1 }); + assert.equal(ledgers.readCoverage().length, 3, "batch 1 checkpointed its three messages"); + const cov = coverageOf(record.messages as UIMessage[], ledgers.readCoverage(), chatId); + assert.equal(cov.uncoveredCount, 2, "the in-flight batch's messages stay uncovered"); + + // The re-run — the manual extract button — picks up ONLY the + // remainder: one batch, messages 4-5 alone in the prompt. + const reRunViews: string[] = []; + const finisher = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + extractor: fauxEngine([findingsStep((v) => reRunViews.push(v))]), + auditor: fauxEngine([passThroughAuditStep()]), + labeler: submitModel("submit_labels", { labels: [] }), + }); + await finisher.runExtraction(chatId); + assert.equal(reRunViews.length, 1); + assert.ok(reRunViews[0]!.includes("LEARNER [u4]") && reRunViews[0]!.includes("LEARNER [u5]")); + assert.ok(!reRunViews[0]!.includes("LEARNER [u1]"), "the covered remainder never re-enters the prompt"); + assert.equal(ledgers.readCoverage().length, 5, "the re-run closed the gap"); + }); + + it("a re-opened message re-extracts alone; a data-part attach never re-opens it", async () => { + // A small session: one batch covers everything, so re-openings are + // the ONLY thing later runs can see. + const chat = chatStore.createChat("Re-open fixture"); + createdChats.push(chat.id); + chatStore.saveMessages(chat.id, [user("s1", "short topic one"), user("s2", "short topic two")]); + const chatId = chat.id; + const ledgers = memoryLedgers(); + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + extractor: fauxEngine([findingsStep()]), + auditor: fauxEngine([passThroughAuditStep()]), + labeler: submitModel("submit_labels", { labels: [] }), + }); + await pipeline.runExtraction(chatId); + const messages = () => chatStore.getChat(chatId)!.messages as UIMessage[]; + assert.equal(coverageOf(messages(), ledgers.readCoverage(), chatId).uncoveredCount, 0); + + // Data parts are invisible to the extraction view — plan-block + // attach does NOT re-open a covered message. + const withDataPart = messages().map((m) => + m.id === "u2" + ? { ...m, parts: [...m.parts, { type: "data-plan-blocks", planBlocks: [] } as unknown as UIMessage["parts"][number]] } + : m, + ); + chatStore.saveMessages(chatId, withDataPart); + const afterAttach = coverageOf(messages(), ledgers.readCoverage(), chatId); + assert.equal(afterAttach.uncoveredCount, 0, "data parts don't change the fingerprint"); + + // A quiz answer merging into an ASKING message changes its view → + // exactly that message re-opens, and the next run extracts it alone. + const mutated = messages().map((m) => + m.id === "s1" ? { ...m, parts: [...m.parts, { type: "text", text: "the answer merged in late" }] } : m, + ) as unknown as UIMessage[]; + chatStore.saveMessages(chatId, mutated); + const afterMerge = coverageOf(messages(), ledgers.readCoverage(), chatId); + assert.equal(afterMerge.uncoveredCount, 1, "only the mutated message re-opens"); + + const reRunViews: string[] = []; + const reOpener = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + extractor: fauxEngine([findingsStep((v) => reRunViews.push(v))]), + auditor: fauxEngine([passThroughAuditStep()]), + labeler: submitModel("submit_labels", { labels: [] }), + }); + await reOpener.runExtraction(chatId); + assert.equal(reRunViews.length, 1, "one batch: the single re-opened message"); + assert.ok(reRunViews[0]!.includes("LEARNER [s1]") && !reRunViews[0]!.includes("LEARNER [s2]")); + assert.equal(coverageOf(messages(), ledgers.readCoverage(), chatId).uncoveredCount, 0, "re-covered"); + }); +}); diff --git a/tests/graph.test.ts b/tests/graph.test.ts index 2bfc79c..7f7f908 100644 --- a/tests/graph.test.ts +++ b/tests/graph.test.ts @@ -20,7 +20,7 @@ import { after, describe, it } from "node:test"; import type { UIMessage } from "ai"; import type { AuditResult } from "../src/core/contracts.ts"; import type { QuestionAnswerData } from "../src/core/question-core.ts"; -import { clusterContentId, getGraph, graphSnapshot, mergeFindings, pendingQuizAnswers, type GraphDeps } from "../src/core/graph.ts"; +import { clusterContentId, getGraph, graphSnapshot, mergeFindings, type GraphDeps } from "../src/core/graph.ts"; import { graphStore } from "../src/store/graph.ts"; import { chatStore } from "../src/store/chats.ts"; import { LEDGER_FILES, PATHS, SOURCES_FILE, EMBEDDINGS_FILE } from "../config.ts"; @@ -564,55 +564,6 @@ describe("the merger (mergeFindings)", () => { }); }); -// ─── the trigger helper ────────────────────────────────────────────────────── - -describe("the trigger helper (pendingQuizAnswers)", () => { - it("counts answered quizzes not yet in the ledger — dismissed/ask never count", () => { - const chat = fixtureChat("Trigger fixture", [ - user("u1", "go"), - assistant("a1", [ - quizPart( - "call-s", - { question: "QS", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, - answer({ questionId: "call-s", questionText: "QS text", correct: true, answeredAt: "2026-08-22T03:00:00.000Z" }), - ), - quizPart( - "call-s2", - { question: "QS2", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, - answer({ questionId: "call-s2", questionText: "QS2 text", dismissed: true }), - ), - { - type: "tool-ask_user_question", - toolCallId: "call-s3", - state: "output-available", - input: { question: "free?" }, - output: answer({ questionId: "call-s3", kind: "ask", questionText: "free?", text: "y" }), - }, - ]), - ]); - - assert.equal(pendingQuizAnswers(chat.id, graphDeps), 1, "one answered quiz pending"); - assert.equal(pendingQuizAnswers("chat_does_not_exist", graphDeps), 0); - - mergeFindings(chat.id, fixtureAudit(), graphDeps, sessionSource(chat.id)); - // call-s is unattributed here (fixtureAudit has no call-s attribution) - // — but it still becomes an event, so nothing stays pending. - assert.equal(pendingQuizAnswers(chat.id, graphDeps), 0, "extracted → nothing pending"); - - // A NEW answered quiz raises the count again (the re-trigger path). - chatStore.saveMessages(chat.id, [ - ...chatStore.getChat(chat.id)!.messages, - assistant("a3", [ - quizPart( - "call-s4", - { question: "QS4", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, - answer({ questionId: "call-s4", questionText: "QS4 text", correct: false, answeredAt: "2026-08-22T03:01:00.000Z" }), - ), - ]), - ]); - assert.equal(pendingQuizAnswers(chat.id, graphDeps), 1); - }); -}); // ─── store primitives ──────────────────────────────────────────────────────── diff --git a/tests/render-graph-coverage.test.tsx b/tests/render-graph-coverage.test.tsx new file mode 100644 index 0000000..a2b38d5 --- /dev/null +++ b/tests/render-graph-coverage.test.tsx @@ -0,0 +1,85 @@ +import { renderToStaticMarkup } from "react-dom/server"; +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { CoveragePanel, GraphCoverageControl } from "../components/graph-coverage.tsx"; + +// The extraction-coverage panel (vision-extraction-coverage.md, settled +// 2026-09-15): the graph view's window into WHAT the pipeline has covered, +// with a per-session "Extract" button for the uncovered remainder. Static +// markup can't click — the closed control and the panel body are pinned +// separately. (Interaction — the fetch + POST flow — is covered by the live +// smoke.) + +describe("GraphCoverageControl", () => { + it("renders the coverage button in the chrome, closed by default", () => { + const html = renderToStaticMarkup(); + assert.ok(html.includes(" { + const base = { + chatId: "chat_a", + title: "Modal logic", + updatedAt: "2026-09-15T00:00:00.000Z", + totalMessages: 12, + coveredMessages: 10, + uncoveredMessages: 2, + coveredTokens: 1536, + uncoveredTokens: 384, + running: false, + }; + + it("shows each session's covered/uncovered tokens and the Extract button", () => { + const html = renderToStaticMarkup( + false} onExtract={() => {}} />, + ); + assert.ok(html.includes("Modal logic")); + assert.ok(html.includes("1.5k covered")); + assert.ok(html.includes("384 uncovered")); + assert.ok(html.includes("(2 msgs)")); + assert.ok(html.includes(">Extract<"), "the uncovered session offers a manual extract"); + }); + + it("a fully covered session shows · complete with a disabled Extract button", () => { + const html = renderToStaticMarkup( + false} + onExtract={() => {}} + />, + ); + assert.ok(html.includes("· complete")); + assert.ok(html.includes("disabled"), "nothing to extract — the button is honest"); + assert.ok(!html.includes(">Extract<") || html.includes("disabled")); + }); + + it("a running session shows extracting… and no clickable duplicate start", () => { + const html = renderToStaticMarkup( + c.running} + onExtract={() => {}} + />, + ); + assert.ok(html.includes("extracting…")); + assert.ok(html.includes(">Running<")); + assert.ok(html.includes("disabled"), "single-flight: the in-flight run can't be double-started"); + }); + + it("all-covered panel: the nothing-pending line, not a row list", () => { + const html = renderToStaticMarkup( + false} + onExtract={() => {}} + />, + ); + assert.ok(html.includes("Every session is fully extracted — nothing pending.")); + }); +}); \ No newline at end of file