diff --git a/README.md b/README.md index 0fec74c..2697aa0 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,8 @@ npx tsx scripts/bench-graph.ts # force-layout benchmark (synthetic graphs) |---|---| | `OPENROUTER_API_KEY` | required — all LLM calls go through OpenRouter | | `LH_MODEL` | teacher model (default `anthropic/claude-sonnet-4.5`) | -| `LH_RESEARCH_MODEL`, `LH_DEMO_MODEL` | per-sub-agent overrides (fall back to `LH_MODEL`) | +| `LH_RESEARCH_MODEL`, `LH_DEMO_MODEL`, `LH_EXTRACT_MODEL` | per-sub-agent overrides (fall back to `LH_MODEL`) | +| `LH_EXTRACT_N` | answered quizzes per chat before the extraction pipeline fires (default `3`) | | `LH_VISUAL_MODEL` | visual-maker model — must be **vision-capable** (it looks at rendered SVGs) | | `GOOGLE_SEARCH_API_KEY` + `GOOGLE_CSE_ID` | researcher's `web_search` (Google Custom Search) | | `CHROME_PATH` | optional explicit Chrome path for headless verification (puppeteer-core uses system Chrome; nothing is downloaded) | @@ -58,7 +59,7 @@ graph TB end subgraph TEACHER["Teacher agent (src/agent)"] - CH["chat-handler.ts — turn composition root
(services: context · persist · spawn · fold)"] + CH["chat-handler.ts — turn composition root
(services: context · persist · spawn · extract)"] TA["ToolLoopAgent
system-prompt.md + prompts/*.md fragments
+ graph-context block"] QT["quiz · ask_user_question
(client-side tools, no execute)"] PT["plan_* tools (server-side,
mutate the chat's tree)"] @@ -69,6 +70,7 @@ graph TB RES["researcher
web_search + web_fetch"] DEMO["demo-builder
write → headless-run → fix"] VIS["visual-maker
write → render → look"] + EXT["extraction pipeline
extractor → auditor → merger
(the sole graph writer)"] end subgraph STORE["File stores (src/store adapters → data/, never in git)"] @@ -77,10 +79,10 @@ graph TB LEDG[("data/graph/*.jsonl
nodes/edges/events/mappings/sessions")] end - subgraph PIPE["Graph pipeline (src/core)"] - EMIT["graph.ts emitChat
naive-union, idempotent,
exact/alias match"] + subgraph PIPE["Graph core (src/core) — merger + derivation"] + MERGE["graph.ts mergeFindings
deterministic apply: identity,
event rebuild, dedupe, checkpoint"] GCTX["graph-context.ts
hotspots + 1-hop when large"] - SNAP["graphSnapshot · foldStaleChats"] + SNAP["graphSnapshot (pure read)"] end TR & COMP & DOCK --> RCHAT @@ -95,17 +97,18 @@ graph TB PT --> CHATS CH -->|"onFinish: saveMessages"| CHATS CH -->|"onFinish: spawnPendingJobs"| RES & DEMO & VIS - CH -->|"onFinish: emitChat"| EMIT + CH -->|"onFinish: every N quiz answers
(turn-service seam)"| EXT RES & DEMO & VIS --> JOBS RES & DEMO & VIS -->|"rewrite tool part
→ output-available"| CHATS - EMIT --> LEDG + EXT -->|"audit-passed findings"| MERGE + MERGE --> LEDG LEDG --> GCTX -->|"appended to instructions"| TA LEDG --> SNAP --> RGRAPH ``` -**Core invariant, carried over from the original design:** the graph is never asserted by an LLM post-hoc guesswork — v1 emission is deterministic code over structured tool data. The teacher stays a teacher; everything below the ledgers is derived. +**Core invariant:** the extraction pipeline (extractor → auditor → merger) is the sole graph writer. The LLM stages propose and verify (propositions, edges, probe attribution — each with verbatim evidence); the deterministic merger owns every ledger append and rebuilds anything the model could silently mis-copy (event kinds, timestamps, idempotency refs) from the transcript. The teacher stays a teacher; everything below the ledgers is derived. ## Layout @@ -124,16 +127,18 @@ src/ tools/web.ts web_search (Google CSE) + web_fetch (Readability) tools/jobs.ts research / demo / visual tool surfaces subagents/ researcher · demo-builder · visual-maker, + extraction pipeline (extractor/auditor + prompts), runner (async protocol), headless Chrome helper server/chat-handler.ts one chat turn: compose agent + services, stream server/turn-service.ts turn services: teacher context · persist · - spawn jobs · fold-into-graph (error-contained) + spawn jobs · extraction trigger (the v1 + quiz-answer threshold — one swappable seam) server/stores.ts THE composition root — the one place adapters wire in core/contracts.ts THE contracts file — types + protocol docs core/ports.ts storage ports — ChatStore · GraphStore · JobStore (core code depends on these types only) core/ question-core, plan-tree, phases, pending, - graph (emission/derivation), graph-context, math delims + graph (merger/derivation), graph-context, math delims — all pure: storage arrives as parameters store/ fs adapters implementing the ports: chats · graph ledgers · jobs (plain fs, no framework) @@ -149,15 +154,15 @@ data/ chats/ · graph/ · jobs/ — runtime state, git **The async sub-agent protocol.** `research` / `demo` / `visual` are also client-side tools: the server turn ends at the call, `onFinish` spawns a background runner (fire-and-forget), the job store tracks a live activity line, the client polls `/api/jobs` while parts are pending, and on completion the runner rewrites the tool part in the chat record to `output-available` — the client re-syncs, the continuation fires, the teacher receives the result as a normal tool output. Sub-agents **verify, never one-shot**: the demo-builder headlessly executes its HTML in system Chrome before delivery; the visual-maker renders each SVG and looks at it (needs a vision-capable `LH_VISUAL_MODEL`). -**The live graph.** Every completed turn folds the chat into append-only JSONL ledgers (`emitChat`: tree nodes/edges, quiz outcome events, taught events — idempotent, checkpointed, so partial sessions still deposit value). The home view derives node states from the ledgers and lays them out with forceAtlas2 (run-then-freeze — no idle simulation). At session start the derived graph is rendered into a compact markdown block appended to the teacher's instructions (`buildGraphContext`) — whole graph while small, hotspots + 1-hop neighborhood past 150 nodes. +**The knowledge graph (the two channels).** The write channel: the extraction pipeline — extractor (LLM: propositions, edges, probe attribution, each with verbatim evidence) → auditor (LLM: drops anything unbacked) → merger (`mergeFindings`, deterministic: exact/alias identity resolution, event rebuild from the transcript's tool parts, dedupe, checkpoint). Triggered once a chat accumulates `EXTRACTION.everyNQuizAnswers` (default 3, `LH_EXTRACT_N`) new answered quizzes — a v1 placeholder trigger isolated in one seam (`turn-service.ts`), swappable without touching the pipeline. Failure checkpoints nothing; the next trigger re-runs. The read channel: the home view derives node states from the ledgers (`GET /api/graph` is a pure read — forceAtlas2 run-then-freeze, no idle simulation), and at session start the derived graph is rendered into a compact markdown block appended to the teacher's instructions (`buildGraphContext`) — whole graph while small, hotspots + 1-hop neighborhood past 150 nodes. ## Testing All deterministic, no LLM (models are injected as `MockLanguageModelV4` where an agent loop is under test): -1. **Pure core units** — plan-tree ops, question grading, phases, graph emission, graph-context selection, math delimiters. +1. **Pure core units** — plan-tree ops, question grading, phases, graph merger + trigger helper, graph-context selection, math delimiters. 2. **Server flows** — `chat-handler` driven directly with a scripted mock model: persistence, job spawning, part rewriting, continuation. -3. **Sub-agent loops** — researcher/demo-builder/visual-maker with mock models + fake headless API. +3. **Sub-agent loops** — researcher/demo-builder/visual-maker with mock models + fake headless API; the extraction pipeline end-to-end with mock models (extractor → auditor → in-memory ledgers). 4. **Render tests** — transcript, cards, dock, composer, tree panel via React server-side rendering. ## Lineage diff --git a/config.ts b/config.ts index 4e3d580..b0ceb62 100644 --- a/config.ts +++ b/config.ts @@ -15,7 +15,17 @@ export const PATHS = { } as const; /** - * The LIVE graph store (src/store/graph.ts) — owned by the app's emission + * The extraction pipeline (vision §1 construction): the sole graph writer. + * Triggered per chat once this many NEW quiz answers have landed since the + * chat's last extraction checkpoint (LH_EXTRACT_N). The trigger mechanism is + * a v1 placeholder — deliberately isolated behind one seam + * (server/turn-service.ts) so it can be swapped without touching the + * pipeline. + */ +export const EXTRACTION = { everyNQuizAnswers: Number(process.env.LH_EXTRACT_N ?? 3) } as const; + +/** + * The LIVE graph store (src/store/graph.ts) — owned by the app's extraction * pipeline. Line shapes: src/core/contracts.ts. */ export const LEDGER_FILES = { diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts new file mode 100644 index 0000000..fbe1695 --- /dev/null +++ b/src/agent/subagents/extraction.ts @@ -0,0 +1,309 @@ +import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type UIMessage } from "ai"; +import { openrouter } from "@openrouter/ai-sdk-provider"; +import { z } from "zod"; +import type { AuditResult, ChatRecord, SessionFindings } from "../../core/contracts.ts"; +import { mergeFindings, type EmissionStats, type GraphDeps } from "../../core/graph.ts"; + +/** + * The extraction pipeline (vision §1 construction) — the sole graph writer. + * Three stages, one orchestrator: + * + * extractor (LLM) transcript → SessionFindings (propositions, edges, + * event attribution — each with verbatim evidence) + * auditor (LLM) findings → AuditResult (the verified subset + drops) + * merger (core) mergeFindings() → deterministic ledger appends + * + * The LLM stages supply only the judgment calls; the deterministic rebuild + * in the merger owns everything that must never be mis-copied (event kinds, + * timestamps, refs). Both agents are single-tool-call workers: their whole + * deliverable is one zod-validated submit call — a ToolLoopAgent purely so + * the model can retry a schema-invalid call. + * + * Created via createExtractionPipeline(deps) — the composition root + * (chat-handler) injects the storage ports + models; this module never + * imports an adapter. runExtraction is single-flight per chat: a second + * trigger while one run is in flight is a no-op (the next trigger picks up + * whatever the in-flight run missed). On failure nothing is checkpointed — + * the next trigger re-runs the whole thing. + */ + +const DEFAULT_EXTRACTION_MODEL = + process.env.LH_EXTRACT_MODEL ?? process.env.LH_MODEL ?? "anthropic/claude-sonnet-4.5"; + +// ─── the transcript view (what the LLM sees) ──────────────────────────────── + +/** + * A quiz tool part's outcome, one line: the attribution evidence. + * (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 the extractor + auditor both see: user/teacher + * text verbatim (evidence quotes must match it), the final plan tree, and + * the structured tool events (quiz outcomes, mark_done). Job tool parts + * (research/demo/visual) are noise for extraction — their outcomes are + * summarized as one line each. + */ +export function transcriptView(record: ChatRecord): string { + const lines: string[] = []; + lines.push(`SESSION: ${record.title}`); + + if (record.tree && record.tree.nodes.length > 0) { + const nodes = record.tree.nodes.map((n) => `${n.id} = ${n.title}`).join("; "); + lines.push(`FINAL PLAN TREE (node ids): ${nodes}`); + if (record.tree.edges.length > 0) { + const edges = record.tree.edges.map((e) => `${e.from} requires ${e.to}`).join("; "); + lines.push(`TREE EDGES: ${edges}`); + } + } + + for (const message of record.messages as UIMessage[]) { + const who = message.role === "user" ? "LEARNER" : "TEACHER"; + for (const part of message.parts) { + if (part.type === "text" && part.text.trim()) { + lines.push(`${who}: ${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 (name === "plan_mark_done" && p.state === "output-available") { + const input = p.input as { id?: unknown } | undefined; + lines.push(`TEACHER marked tree node done [${p.toolCallId}]: ${typeof input?.id === "string" ? input.id : "?"}`); + } else if (name.startsWith("plan_")) { + // plan ops are folded into the FINAL PLAN TREE line above + } else if (p.state === "output-available") { + lines.push(`(${name} tool result delivered)`); + } + } + } + } + return lines.join("\n"); +} + +// ─── the extractor ────────────────────────────────────────────────────────── + +const evidence = z + .array(z.string()) + .min(1) + .describe("Verbatim quotes from the transcript that back this. The auditor will check every quote."); + +const findingsSchema = z.object({ + propositions: z + .array( + z.object({ + id: z.string().describe("Local id: \"p1\", \"p2\", … — referenced by edges and attribution."), + statement: z.string().describe("The proposition as one reasoning step — a node statement."), + aliases: z.array(z.string()).describe("Verbatim alternate phrasings from this session, if any."), + kind: z.enum(["truth", "misconception"]).describe("misconception = a wrong idea the learner holds or was warned about."), + treeNodeId: z.string().optional().describe("The plan-tree node id this came from, when it did."), + evidence, + }), + ) + .describe("Every distinct proposition the session touched — from the plan tree AND from prose."), + edges: z + .array( + z.object({ + from: z.string().describe("Proposition id that is the prerequisite / support."), + to: z.string().describe("Proposition id that depends on / is supported by `from`."), + type: z.enum(["requires", "supports"]), + evidence, + }), + ) + .describe("Relations between propositions. `requires` = prerequisite; `supports` = helpful but not prerequisite. Include relations the plan tree does NOT show when the transcript articulates them."), + attribution: z + .object({ + quiz: z + .record(z.string(), z.string().nullable()) + .describe("toolCallId of every UNTAGGED quiz line → the proposition id it was really testing. Probe questions asked BEFORE the tree existed are attributed RETROACTIVELY — the whole transcript is visible; map them to the proposition they were probing (often articulated only later). null when genuinely unattributable. Tagged quizzes (with a tree nodeId in the transcript view) resolve automatically — do not include them."), + }) + .describe("Probe attribution — your only event judgment."), +}); + +const EXTRACTOR_INSTRUCTIONS = `You are the extraction specialist in a learning application. You read ONE session +transcript and report what was learned — as structured findings for the learner's global knowledge graph. + +The transcript view shows: the session's final plan tree (dependency map, with node ids), the conversation +verbatim (LEARNER / TEACHER), and structured events (QUIZ lines with the learner's outcome, mark_done lines). + +Report, with verbatim evidence quotes for everything: + +1. PROPOSITIONS — every distinct reasoning-step-sized proposition the session touched: + - Plan-tree nodes become propositions (pass their treeNodeId) — statement from the node title, enriched to a + complete proposition if the prose elaborates it. + - Propositions that appeared only in prose (teacher explanations, learner questions) get fresh ids. + - Include MISCONCEPTIONS — wrong ideas the learner showed or the teacher warned against — with kind + "misconception". + - One reasoning step per proposition: never a topic blob, never a sub-step. + - aliases: verbatim alternate phrasings this session used for the same thing. + +2. EDGES between propositions: + - The tree's dependency edges are the backbone: from = prerequisite, to = dependent, type "requires". + - Also include relations articulated only in prose. Use "supports" when the transcript presents something as + helpful but NOT a prerequisite. + +3. ATTRIBUTION — UNTAGGED quizzes only: + - QUIZ lines whose question carries a tree nodeId resolve automatically; do not attribute those. + - The others are PROBE-phase questions (asked before the tree was initialized) that test prerequisite + understanding. Attribute them RETROACTIVELY to the proposition they were really probing — even if that + proposition is only articulated later in the session. The QUIZ question text is your guide. If you truly + cannot tell, use null. + +Evidence discipline: every quote must appear VERBATIM in the transcript view (a QUIZ line counts as the +transcript). The auditor rejects anything unbacked. When in doubt, report; the auditor decides. + +Your entire deliverable is ONE submit_findings call.`; + +function createExtractor(model: LanguageModel) { + return new ToolLoopAgent({ + model, + instructions: EXTRACTOR_INSTRUCTIONS, + tools: { + submit_findings: tool({ + description: "Submit the session findings.", + inputSchema: findingsSchema, + execute: async (findings) => findings, + }), + }, + stopWhen: stepCountIs(4), + }); +} + +// ─── the auditor ─────────────────────────────────────────────────────────── + +const auditSchema = findingsSchema.extend({ + dropped: z + .array( + z.object({ + ref: z.string().describe("Proposition id, \"from→to\" for an edge, or attribution toolCallId."), + reason: z.string(), + }), + ) + .describe("Everything you dropped, with the reason — the audit trail."), +}); + +const AUDITOR_INSTRUCTIONS = `You are the audit stage of a learning application's extraction pipeline. You receive +ONE session transcript and a set of proposed findings (propositions, edges, event attribution). Your job is the +faithfulness net: verify every candidate against the transcript, keep only what is backed, and return the +verified subset. + +Verify: +- PROPOSITIONS: every evidence quote must appear (near-)verbatim in the transcript view, and the statement must + be a faithful reading of it. Drop invented or bloated statements. Conservative: when two proposed propositions + are really one, keep the better-stated one and drop the other. +- EDGES: the evidence must actually articulate the relation, in the direction claimed (from = prerequisite). + Drop direction-swapped or speculative edges. +- ATTRIBUTION: only UNTAGGED quizzes carry attribution — the QUIZ line's question must genuinely test the + attributed proposition. Drop wrong attributions to null, or re-point them to the proposition the question + actually tests. + +Also drop propositions not referenced by anything else ONLY if they have no other backing — an unreferenced but +well-evidenced proposition is still knowledge. + +Never invent new candidates. Your entire deliverable is ONE submit_review call containing the verified findings +(the same shapes you received) plus the dropped list.`; + +function createAuditor(model: LanguageModel) { + return new ToolLoopAgent({ + model, + instructions: AUDITOR_INSTRUCTIONS, + tools: { + submit_review: tool({ + description: "Submit the verified findings subset + the dropped list.", + inputSchema: auditSchema, + execute: async (review) => review, + }), + }, + stopWhen: stepCountIs(4), + }); +} + +// ─── the pipeline ─────────────────────────────────────────────────────────── + +export interface ExtractionModels { + /** The extractor's model (tests inject a MockLanguageModelV4). */ + extractor?: LanguageModel; + /** The auditor's model. */ + auditor?: LanguageModel; +} + +export interface ExtractionPipelineDeps extends GraphDeps, ExtractionModels {} + +export interface ExtractionPipeline { + /** Extract + audit + merge ONE chat. Single-flight per chat; resolves to + * the merger's append stats (or throws — no checkpoint on failure). */ + runExtraction(chatId: string): Promise; +} + +export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline { + const { chatStore, graphStore } = deps; + const extractorAgent = createExtractor(deps.extractor ?? openrouter(DEFAULT_EXTRACTION_MODEL)); + const auditorAgent = createAuditor(deps.auditor ?? openrouter(DEFAULT_EXTRACTION_MODEL)); + const inFlight = new Set(); + + async function extract(chatId: string): Promise { + const record = chatStore.getChat(chatId); + if (!record) throw new Error(`Chat not found: ${chatId}`); + const view = transcriptView(record); + + const extraction = await extractorAgent.generate({ prompt: `Transcript:\n\n${view}` }); + for (const part of extraction.content) { + if (part.type === "tool-result" && part.toolName === "submit_findings") { + return part.output as SessionFindings; + } + } + throw new Error("the extractor never submitted findings"); + } + + async function audit(chatId: string, findings: SessionFindings): Promise { + const record = chatStore.getChat(chatId); + if (!record) throw new Error(`Chat not found: ${chatId}`); + const view = transcriptView(record); + + const review = await auditorAgent.generate({ + prompt: `Transcript:\n\n${view}\n\nProposed findings:\n\n${JSON.stringify(findings, null, 2)}`, + }); + let auditResult: AuditResult | null = null; + for (const part of review.content) { + if (part.type === "tool-result" && part.toolName === "submit_review") { + auditResult = part.output as AuditResult; + } + } + if (!auditResult) throw new Error("the auditor never submitted a review"); + if (auditResult.dropped.length > 0) { + console.info( + `[extraction ${chatId}] audit dropped ${auditResult.dropped.length}: ` + + auditResult.dropped.map((d) => `${d.ref} (${d.reason})`).join("; "), + ); + } + const { dropped: _dropped, ...verified } = auditResult; + return verified; + } + + async function runExtraction(chatId: string): Promise { + if (inFlight.has(chatId)) return { nodes: 0, edges: 0, events: 0 }; + inFlight.add(chatId); + try { + const findings = await extract(chatId); + const verified = await audit(chatId, findings); + // The merger re-reads the chat record itself — events are rebuilt + // from the transcript, never from stale copies. + const stats = mergeFindings(chatId, verified, { chatStore, graphStore }); + console.info(`[extraction ${chatId}] merged: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events`); + return stats; + } finally { + inFlight.delete(chatId); + } + } + + return { runExtraction }; +} diff --git a/src/core/contracts.ts b/src/core/contracts.ts index e60cece..0f31e72 100644 --- a/src/core/contracts.ts +++ b/src/core/contracts.ts @@ -83,12 +83,14 @@ export interface GraphEdge { provenance: string[]; } -/** mappings.jsonl line — the naive-union identity map: a chat's tree node → - * the graph node it folded into (the merge audit trail; v1 never merges two - * sessions' tree nodes — W8 exact-title match aside). */ +/** mappings.jsonl line — the session→graph identity map: the extraction + * pipeline's proposition (identified by its source id) → the graph node it + * merged into. The merge audit trail: one line per (chat, graph node). */ export interface NodeMapping { chatId: string; - treeNodeId: string; + /** The writer's local id for the proposition — a plan-tree node id when + * tree-derived, else the extractor's proposition id ("p1", …). */ + sourceId: string; graphNodeId: string; } @@ -125,18 +127,85 @@ export interface LedgerEvent { } /** - * Store format (W4 implements src/store/graph.ts): JSONL ledgers at - * config.LEDGER_FILES (nodes/edges/events/mappings/sessions). The derived - * graph is rebuilt by reading the ledgers — store events, derive state. - * sessions.jsonl tracks which chats have been folded into the graph - * (emission checkpoint — never fold the same turn twice). + * Store format (src/store/graph.ts): JSONL ledgers at config.LEDGER_FILES + * (nodes/edges/events/mappings/sessions). The derived graph is rebuilt by + * reading the ledgers — store events, derive state. sessions.jsonl is the + * extraction checkpoint: which chats the extraction pipeline has processed, + * and through which updatedAt (never extract the same turns twice). * - * v1 emission (naive union, deterministic, no LLM): a chat's plan tree folds - * in — node title → statement (aliases empty), tree edges → requires edges, - * mark_done → "taught" events; quiz parts with input.nodeId → probe / check - * events (+ note); quiz parts without nodeId → events with nodeId null. + * The EXTRACTION PIPELINE is the sole graph writer (extractor → auditor → + * merger): the extractor proposes propositions/edges/attribution with + * verbatim evidence, the auditor drops anything unbacked, and the merger + * (core/graph.ts, deterministic) resolves identity against the existing + * ledgers and appends. See SessionFindings below. */ +// ─── Extraction findings (extractor output → audit → merger input) ─────────── + +/** One proposed proposition — a node candidate. */ +export interface ExtractedProposition { + /** Extractor-local id ("p1", …) referenced by edges + attribution. */ + id: string; + /** Canonical one-reasoning-step statement. */ + statement: string; + /** Verbatim alternate phrasings from the session (merge aliases). */ + aliases: string[]; + kind: "truth" | "misconception"; + /** The plan-tree node this came from, when it did (mapping provenance). */ + treeNodeId?: string; + /** Verbatim transcript quotes backing the proposition. */ + evidence: string[]; +} + +/** One proposed relation between two propositions. */ +export interface ExtractedEdge { + /** Proposition id that is the prerequisite / support. */ + from: string; + /** Proposition id that depends on / is supported by `from`. */ + to: string; + type: EdgeType; + /** Verbatim transcript quotes backing the relation. */ + evidence: string[]; +} + +/** The extractor's event attribution — its ONLY say about events. The kinds, + * timestamps, idempotency refs, notes and question texts are all rebuilt + * deterministically by the merger from the transcript's quiz / mark_done + * tool parts; the LLM supplies exactly the judgment call: WHICH proposition + * each one is about (vision open question: probes attribute retroactively, + * against the whole transcript — a pre-tree probe can map to a proposition + * the tree only articulates later). */ +/** The extractor's event attribution — its ONLY say about events. Event + * kinds, timestamps, idempotency refs, notes and question texts are all + * rebuilt deterministically by the merger from the transcript's tool + * parts, and TAGGED events (quiz nodeId, mark_done tree node) resolve + * deterministically through the proposition mapping. The LLM supplies + * exactly the one judgment call that needs it — the vision's open + * probe-attribution question: WHICH proposition did an UNTAGGED probe + * test? Answered retroactively against the whole transcript — a pre-tree + * probe can map to a proposition the tree only articulates later. */ +export interface ExtractionAttribution { + /** UNTAGGED quiz toolCallId → proposition id; missing / null = + * unattributed (the event lands with nodeId null + the question text, + * exactly like unattributable probes always did). */ + quiz: Record; +} + +/** The extractor's — and, filtered, the auditor's — output shape. */ +export interface SessionFindings { + propositions: ExtractedProposition[]; + edges: ExtractedEdge[]; + attribution: ExtractionAttribution; +} + +/** The auditor's verdict: the verified subset, plus the drops (for the log — + * the audit trail of what was rejected and why). */ +export interface AuditResult extends SessionFindings { + /** Dropped candidates: proposition id, edge "from→to", or attribution + * toolCallId — with the reason. */ + dropped: { ref: string; reason: string }[]; +} + /** Per-node state class for the home view (vision §1 "Home page") — derived, * never stored: taught if ANY taught event; else the LATEST quiz event decides * (known / missed / dont-know); a node with no events is a candidate. */ diff --git a/src/core/graph.ts b/src/core/graph.ts index 709d8a5..c4c6616 100644 --- a/src/core/graph.ts +++ b/src/core/graph.ts @@ -1,24 +1,29 @@ /** - * Knowledge-graph emission + derivation (W4; vision §1 "One substrate", - * contracts.ts "v1 emission"). + * The knowledge graph: merger (the deterministic final stage of the + * extraction pipeline — vision §1 construction) + derivation (reads). * - * emitChat(chatId) folds ONE chat into the global graph — naive union, - * deterministic, no LLM: - * plan-tree nodes → GraphNodes (statement = title, aliases empty; - * W8: a title exactly matching an existing statement - * or alias — case-insensitive, trimmed — maps onto the - * existing node instead of appending a duplicate) - * plan-tree edges → `requires` GraphEdges (from = prerequisite) - * answered quizzes → ledger events (probe-* before the first plan_init - * output, check / dont-know after; ref = toolCallId) - * plan_mark_done → "taught" events (ref = taught:) + * The extraction pipeline is the SOLE graph writer: * - * Idempotent + incremental: mappings dedupe nodes, (from,to,type) dedupes - * edge assertions, chat-scoped refs dedupe events, and sessions.jsonl - * checkpoints which chats have been folded. Re-emitting a chat appends - * nothing. Every function takes the storage ports (core/ports.ts) as - * explicit parameters — pure core, no adapter imports; the composition - * root (src/server/stores.ts) wires the fs adapters in. + * extractor (LLM) → SessionFindings: propositions, edges, and event + * attribution, each with verbatim evidence + * auditor (LLM) → the verified subset (unbacked candidates dropped) + * merger (HERE) → mergeFindings(): deterministic ledger appends + * + * The merger is deliberately the ONLY stage that touches the ledgers, and it + * is pure code: the LLM supplies exactly the judgment calls (what was + * taught, what depends on what, which proposition a quiz tested), while + * event kinds/timestamps/refs — anything the model could silently + * mis-copy — are rebuilt deterministically from the transcript's tool + * parts. Idempotent + incremental: statement/alias matching dedupes nodes, + * (from,to,type) dedupes edge assertions, chat-scoped refs dedupe events, + * one mapping line per (chat, graph node), and sessions.jsonl checkpoints + * which chats have been extracted (never the same turns twice). + * + * Derivation below (getGraph / deriveNodeStates / graphSnapshot) is reads + * only — everything is recomputed from the ledgers. Every function takes + * the storage ports (core/ports.ts) as explicit parameters — pure core, no + * adapter imports; the composition root (src/server/stores.ts) wires the fs + * adapters in. */ import { randomBytes } from "node:crypto"; @@ -30,8 +35,8 @@ import type { LedgerEvent, LedgerEventKind, NodeStateClass, - PlanTree, PlanToolResult, + SessionFindings, } from "./contracts.ts"; import type { QuestionAnswerData } from "./question-core.ts"; import type { ChatStore, GraphStore } from "./ports.ts"; @@ -47,7 +52,7 @@ function newGraphNodeId(): string { return `n_${randomBytes(6).toString("hex")}`; } -// ─── transcript scanning ───────────────────────────────────────────────────── +// ─── transcript scanning (shared by the merger + trigger) ───────────────────── /** A flattened, order-preserving view of every tool part in a transcript. */ interface ToolPartView { @@ -84,23 +89,7 @@ function planResult(part: ToolPartView): PlanToolResult | null { return out as PlanToolResult; } -/** - * The FINAL live tree, replayed from plan tool outputs in message order. - * Fallback for records whose stored `tree` is missing (it is the same state - * saveTree persists, so the two normally agree); also proves the probe/check - * boundary scan below sees the same transcript the teacher wrote. - */ -function replayFinalTree(messages: UIMessage[]): PlanTree { - let tree: PlanTree = { nodes: [], edges: [], currentNodeId: null, doneNodeIds: [], version: 0 }; - for (const part of toolParts(messages)) { - if (!part.name.startsWith("plan_")) continue; - const result = planResult(part); - if (result && result.ok && result.tree) tree = result.tree; - } - return tree; -} - -// ─── emission ──────────────────────────────────────────────────────────────── +// ─── the merger (extraction pipeline stage 3 — deterministic) ──────────────── export interface EmissionStats { /** New graph nodes appended (and mappings with them). */ @@ -112,79 +101,86 @@ export interface EmissionStats { } /** - * Fold one chat into the graph ledgers. Idempotent — safe to call after every - * turn (chat-handler onFinish) and again on catch-up (GET /api/graph). Sync - * and fast: a handful of small-file reads plus one append per new record. + * Merge one chat's audited findings into the graph ledgers — the pipeline's + * apply stage. Deterministic + idempotent: re-merging the same findings (or + * a later re-extraction of the same chat) appends nothing but the new. * - * The storage ports are explicit (tests pass in-memory ledgers; production - * passes the fs adapters from src/server/stores.ts). + * Identity resolution (v1: conservative): a proposition whose statement or + * alias EXACTLY matches (case-insensitive, trimmed) an existing node's + * statement or alias merges onto it; everything else is a new node. One + * mapping line per (chat, graph node) — the merge audit trail. Fuzzy / + * embedding adjudication stays deferred (vision §1 consolidation). + * + * Events are rebuilt deterministically from the transcript's quiz and + * plan_mark_done parts (kinds, timestamps, notes, refs). TAGGED events + * (quiz nodeId, mark_done tree node) resolve through the proposition + * mapping — deterministic; the findings' attribution is consulted ONLY for + * untagged probe quizzes: the retroactive probe-attribution resolution (a + * pre-tree probe maps to a proposition the tree only articulates later). + * Unattributed quizzes land with nodeId null + the question text, exactly + * like the unattributable ones always have. */ -export function emitChat(chatId: string, deps: GraphDeps): EmissionStats { - const { graphStore: ledgers } = deps; - const record = deps.chatStore.getChat(chatId); +export function mergeFindings(chatId: string, findings: SessionFindings, deps: GraphDeps): EmissionStats { + const { graphStore: ledgers, chatStore } = deps; + const record = chatStore.getChat(chatId); if (!record) return { nodes: 0, edges: 0, events: 0 }; const stats: EmissionStats = { nodes: 0, edges: 0, events: 0 }; - // 1. This chat's existing mappings — the identity map is per-chat (v1 is - // naive union: two sessions' tree nodes never merge into one graph node). + // 1. Identity resolution. mapping: proposition id (and tree node id) + // → graph node id — from this chat's existing mapping lines, then + // extended as findings resolve. chatNodeIds: graph nodes this chat has + // already mapped onto — the (chat, graph node) mapping dedupe. const mapping = new Map(); + const chatNodeIds = new Set(); for (const m of ledgers.readMappings()) { - if (m.chatId === chatId) mapping.set(m.treeNodeId, m.graphNodeId); + if (m.chatId !== chatId || typeof m.sourceId !== "string") continue; + mapping.set(m.sourceId, m.graphNodeId); + chatNodeIds.add(m.graphNodeId); } - // 2. The tree state to fold. record.tree is the truth (it is the exact - // state every plan tool persisted); the transcript replay covers records - // that predate tree persistence. Nodes REMOVED from the tree are simply - // absent here — if they were emitted by an earlier fold, their mappings - // and ledger lines stay (the ledgers are append-only evidence). - const tree = - record.tree && record.tree.nodes.length > 0 ? record.tree : replayFinalTree(record.messages); - - // 3. Tree nodes without a mapping → GraphNode + mapping. W8 merge - // improvement: a title that EXACTLY matches (case-insensitive, trimmed) an - // existing node's statement or alias maps onto that node instead of - // appending a duplicate — the mapping line is still appended so provenance - // stays honest. Fuzzy/embedding matching stays deferred (vision §1 - // consolidation pipeline). + // 2. Propositions → graph nodes. Exact statement/alias match against the + // whole ledger first (cross-session reuse), then against propositions + // created earlier in THIS merge (intra-findings dedupe). const norm = (s: string): string => s.trim().toLowerCase(); const existingByTitle = new Map(); for (const node of ledgers.readNodes()) { - const statementKey = norm(node.statement); - if (statementKey && !existingByTitle.has(statementKey)) existingByTitle.set(statementKey, node.id); - for (const alias of node.aliases) { - const aliasKey = norm(alias); - if (aliasKey && !existingByTitle.has(aliasKey)) existingByTitle.set(aliasKey, node.id); + for (const title of [node.statement, ...node.aliases]) { + const key = norm(title); + if (key && !existingByTitle.has(key)) existingByTitle.set(key, node.id); } } const now = new Date().toISOString(); - for (const node of tree.nodes) { - if (mapping.has(node.id)) continue; - const titleKey = norm(node.title); - const existingId = titleKey ? existingByTitle.get(titleKey) : undefined; - if (existingId !== undefined) { - // Exact match: reuse the node, record the mapping, append nothing. - ledgers.appendMapping({ chatId, treeNodeId: node.id, graphNodeId: existingId }); - mapping.set(node.id, existingId); - continue; + for (const p of findings.propositions) { + if (mapping.has(p.id)) continue; + const keys = [p.statement, ...p.aliases].map(norm).filter((k) => k.length > 0); + const matchId = keys.map((k) => existingByTitle.get(k)).find((id) => id !== undefined); + let graphNodeId: string; + if (matchId !== undefined) { + graphNodeId = matchId; // reuse — conservative merge, duplicates over wrong merges + } else { + const graphNode: GraphNode = { + id: newGraphNodeId(), + statement: p.statement, + aliases: [...p.aliases], + kind: p.kind, + createdAt: now, + }; + ledgers.appendNode(graphNode); + stats.nodes++; + graphNodeId = graphNode.id; } - const graphNode: GraphNode = { - id: newGraphNodeId(), - // statement is the title; the summary stays out (contracts). - statement: node.title, - aliases: [], - kind: "truth", - createdAt: now, - }; - ledgers.appendNode(graphNode); - ledgers.appendMapping({ chatId, treeNodeId: node.id, graphNodeId: graphNode.id }); - existingByTitle.set(titleKey, graphNode.id); - mapping.set(node.id, graphNode.id); - stats.nodes++; + for (const key of keys) if (!existingByTitle.has(key)) existingByTitle.set(key, graphNodeId); + if (!chatNodeIds.has(graphNodeId)) { + ledgers.appendMapping({ chatId, sourceId: p.treeNodeId ?? p.id, graphNodeId }); + chatNodeIds.add(graphNodeId); + } + mapping.set(p.id, graphNodeId); + if (p.treeNodeId !== undefined) mapping.set(p.treeNodeId, graphNodeId); } - // 4. Tree edges → `requires` edges. Dedupe by (from,to,type) across the - // whole ledger: an existing assertion just gains this chatId as - // provenance (appended as a new line; reads merge). + // 3. Edges. Dedupe by (from,to,type) across the whole ledger: an existing + // assertion just gains this chatId as provenance (appended as a new + // line; reads merge). const asserted = new Map>(); for (const line of ledgers.readEdges()) { const key = `${line.from}\u0000${line.to}\u0000${line.type}`; @@ -195,23 +191,25 @@ export function emitChat(chatId: string, deps: GraphDeps): EmissionStats { } for (const c of line.provenance) provenance.add(c); } - for (const edge of tree.edges) { + for (const edge of findings.edges) { const from = mapping.get(edge.from); const to = mapping.get(edge.to); if (!from || !to || from === to) continue; // unmapped endpoint — skip - const key = `${from}\u0000${to}\u0000requires`; + const key = `${from}\u0000${to}\u0000${edge.type}`; const provenance = asserted.get(key); if (provenance?.has(chatId)) continue; // already asserted by this chat - ledgers.appendEdge({ from, to, type: "requires", provenance: [chatId] }); + ledgers.appendEdge({ from, to, type: edge.type, provenance: [chatId] }); if (provenance) provenance.add(chatId); else asserted.set(key, new Set([chatId])); stats.edges++; } - // 5. Events, replayed in transcript order so the probe/check boundary is - // exact: quizzes BEFORE the first plan_init output are probes. Refs are - // chat-scoped (a quiz toolCallId or taught:) — a ref already - // in the ledger means this turn was folded before. + // 4. Events, rebuilt deterministically in transcript order so the + // probe/check boundary is exact: quizzes BEFORE the first plan_init + // output are probes. Refs are chat-scoped (a quiz toolCallId or + // taught:) — a ref already in the ledger means this turn + // was extracted before. The findings' attribution supplies the + // nodeId, nothing else. const seenRefs = new Set(); for (const event of ledgers.readEvents()) { if (event.chatId === chatId && event.ref) seenRefs.add(event.ref); @@ -223,7 +221,7 @@ export function emitChat(chatId: string, deps: GraphDeps): EmissionStats { if (part.name === "plan_init" && part.state === "output-available") planSeen = true; if (part.name === "quiz" && part.state === "output-available") { - const event = quizEvent(part, chatId, mapping, planSeen, now); + const event = quizEvent(part, chatId, mapping, findings, planSeen, now); if (event?.ref !== undefined && !seenRefs.has(event.ref)) { ledgers.appendEvent(event); seenRefs.add(event.ref); @@ -237,6 +235,7 @@ export function emitChat(chatId: string, deps: GraphDeps): EmissionStats { if (result?.ok && typeof input?.id === "string") { const ref = `taught:${input.id}`; if (!seenRefs.has(ref)) { + // mark_done is tagged with the tree node id — deterministic. ledgers.appendEvent({ nodeId: mapping.get(input.id) ?? null, kind: "taught", @@ -251,16 +250,18 @@ export function emitChat(chatId: string, deps: GraphDeps): EmissionStats { } } - // 6. Checkpoint: this chat is folded through its current updatedAt. + // 5. Checkpoint: this chat is extracted through its current updatedAt. ledgers.appendCheckpoint({ chatId, foldedAt: record.updatedAt }); return stats; } -/** Build the ledger event for one answered quiz tool part (or null to skip). */ +/** Build the ledger event for one answered quiz tool part (or null to skip). + * Deterministic in everything except the attribution-supplied nodeId. */ function quizEvent( part: ToolPartView, chatId: string, mapping: Map, + findings: SessionFindings, planSeen: boolean, now: string, ): LedgerEvent | null { @@ -275,9 +276,13 @@ function quizEvent( else if (answer.correct === true) kind = planSeen ? "check-correct" : "probe-correct"; else kind = planSeen ? "check-incorrect" : "probe-miss"; + // Tagged (input.nodeId) resolves deterministically through the mapping; + // attribution is the LLM judgment call for UNTAGGED probes only — the + // vision's open probe-attribution question, resolved retroactively. const input = part.input as { nodeId?: unknown } | undefined; - const treeNodeId = typeof input?.nodeId === "string" ? input.nodeId : null; - const nodeId = treeNodeId !== null ? (mapping.get(treeNodeId) ?? null) : null; + const tagged = typeof input?.nodeId === "string" ? mapping.get(input.nodeId) : undefined; + const attributedId = findings.attribution.quiz[part.toolCallId]; + const nodeId = tagged ?? (attributedId ? (mapping.get(attributedId) ?? null) : null); const event: LedgerEvent = { nodeId, @@ -287,11 +292,39 @@ function quizEvent( ref: part.toolCallId, }; if (typeof answer.note === "string" && answer.note.length > 0) event.note = answer.note; - // The question text is the attribution input for untagged/unmapped probes. + // The question text is the attribution input for unattributed probes. if (nodeId === null && typeof answer.questionText === "string") event.question = answer.questionText; 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) ───────────── export interface DerivedGraph { @@ -417,33 +450,3 @@ export function graphSnapshot(deps: GraphDeps): GraphSnapshot { return { nodes, edges, nodeStates, sessions, nodeSessions }; } - -// ─── catch-up ──────────────────────────────────────────────────────────────── - -/** - * Fold every chat whose updatedAt is newer than its checkpoint (partial - * sessions deposit value — vision §1 "frequent updates"). Called at the top - * of GET /api/graph; the chat-handler onFinish hook covers the live path. - */ -export function foldStaleChats(deps: GraphDeps): number { - const checkpoints = new Map(); - for (const checkpoint of deps.graphStore.readCheckpoints()) { - const previous = checkpoints.get(checkpoint.chatId); - if (previous === undefined || checkpoint.foldedAt > previous) { - checkpoints.set(checkpoint.chatId, checkpoint.foldedAt); - } - } - - let folded = 0; - for (const chat of deps.chatStore.listChats()) { - const foldedAt = checkpoints.get(chat.id); - if (foldedAt !== undefined && chat.updatedAt <= foldedAt) continue; - try { - emitChat(chat.id, deps); - folded++; - } catch (err) { - console.error(`[graph] fold failed for chat ${chat.id}:`, err); - } - } - return folded; -} diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index ea0bdb7..ff3e0ec 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -1,9 +1,10 @@ import { createAgentUIStreamResponse, type UIMessage } from "ai"; import { createTeacherAgent, type TeacherAgent } from "../agent/teacher-agent.ts"; import { createJobRunner } from "../agent/subagents/runner.ts"; +import { createExtractionPipeline } from "../agent/subagents/extraction.ts"; import { createTeacherTools } from "../agent/tools.ts"; import { stores } from "./stores.ts"; -import { buildTeacherContext, foldTurnIntoGraph, persistTurn, spawnTurnJobs } from "./turn-service.ts"; +import { buildTeacherContext, persistTurn, spawnTurnJobs, triggerExtraction } from "./turn-service.ts"; /** * One chat turn: run the teacher agent over the client's message list and @@ -18,6 +19,7 @@ import { buildTeacherContext, foldTurnIntoGraph, persistTurn, spawnTurnJobs } fr */ const jobRunner = createJobRunner(stores); +const extraction = createExtractionPipeline(stores); export async function handleChatRequest(req: Request, agent?: TeacherAgent): Promise { let body: { chatId?: unknown; messages?: unknown }; @@ -46,7 +48,7 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro onFinish: ({ messages, isAborted }) => { persistTurn(stores.chatStore, chatId, messages as UIMessage[], isAborted); spawnTurnJobs(jobRunner, chatId, messages as UIMessage[]); - foldTurnIntoGraph(stores, chatId); + triggerExtraction(stores, extraction, chatId); }, }); } diff --git a/src/server/graph.ts b/src/server/graph.ts index b22056e..d175488 100644 --- a/src/server/graph.ts +++ b/src/server/graph.ts @@ -1,15 +1,13 @@ import type { GraphSnapshot } from "../core/contracts.ts"; -import { foldStaleChats, graphSnapshot } from "../core/graph.ts"; +import { graphSnapshot } from "../core/graph.ts"; import { stores } from "./stores.ts"; /** - * Server-side graph entry for the home view (GET /api/graph): fold any chat - * whose updatedAt outstrips its emission checkpoint first (partial sessions - * deposit value), then return the snapshot. Today the fold is the naive-union - * emission; the extraction-pipeline stage replaces the writer behind this - * same read surface. + * Server-side graph entry for the home view (GET /api/graph). A PURE READ: + * the extraction pipeline (the sole graph writer) is triggered by quiz + * answers in sessions (turn-service), never from here — the home view + * reflects whatever the pipeline has merged so far. */ export function graphView(): GraphSnapshot { - foldStaleChats(stores); return graphSnapshot(stores); } diff --git a/src/server/turn-service.ts b/src/server/turn-service.ts index 009ea86..323af20 100644 --- a/src/server/turn-service.ts +++ b/src/server/turn-service.ts @@ -7,10 +7,11 @@ * buildTeacherContext — graph → instructions block (the W8 read channel) * persistTurn — the canonical state write (client + agent messages) * spawnTurnJobs — the async sub-agent protocol kick-off - * foldTurnIntoGraph — session → graph consolidation (the write channel; - * the extraction-pipeline stage swaps this body for - * the extractor/auditor/merger trigger behind the - * same seam) + * triggerExtraction — the graph write channel: fires the extraction + * pipeline (the sole graph writer) once a chat + * crosses the v1 quiz-answer threshold. The trigger + * mechanism is a v1 placeholder — this seam is the + * ONLY thing to swap when a better one lands. * * Logs are the minimal-observability baseline: one line per failure, tagged * [chat ] — structured logging lands behind these seams later. @@ -18,9 +19,11 @@ import type { UIMessage } from "ai"; import { buildGraphContext } from "../core/graph-context.ts"; -import { emitChat, getGraph, type GraphDeps } from "../core/graph.ts"; +import { getGraph, pendingQuizAnswers } from "../core/graph.ts"; import type { ChatStore, GraphStore } from "../core/ports.ts"; +import { EXTRACTION } from "../../config.ts"; import type { JobRunner } from "../agent/subagents/runner.ts"; +import type { ExtractionPipeline } from "../agent/subagents/extraction.ts"; /** * W8 (vision §1 "the graph as session context"): the learner's knowledge @@ -66,16 +69,30 @@ export function spawnTurnJobs(jobRunner: JobRunner, chatId: string, messages: UI } /** - * W4: fold this chat into the knowledge-graph ledgers (tree nodes / edges, - * quiz outcomes, taught events) — synchronous but tiny (append-only JSONL), - * and errors never touch the response. Skipped under tests: scripted chats - * are fixtures, not knowledge. + * 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. + * + * 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). */ -export function foldTurnIntoGraph(deps: GraphDeps, chatId: string): void { +export function triggerExtraction( + deps: { chatStore: ChatStore; graphStore: GraphStore }, + extraction: ExtractionPipeline, + chatId: string, +): void { if (process.env.NODE_ENV === "test") return; try { - emitChat(chatId, deps); + if (pendingQuizAnswers(chatId, deps) < EXTRACTION.everyNQuizAnswers) return; + void extraction.runExtraction(chatId).catch((err) => { + console.error(`[chat ${chatId}] extraction pipeline failed:`, err); + }); } catch (err) { - console.error(`[chat ${chatId}] graph emission failed:`, err); + console.error(`[chat ${chatId}] extraction trigger failed:`, err); } } diff --git a/src/store/graph.ts b/src/store/graph.ts index 6666f83..69b3ca5 100644 --- a/src/store/graph.ts +++ b/src/store/graph.ts @@ -6,8 +6,8 @@ * session asserting the same (from,to,type) * appends another line; reads merge) * events.jsonl LedgerEvent — quiz outcomes + taught events - * mappings.jsonl NodeMapping — the naive-union identity map: - * (chatId, treeNodeId) → graphNodeId + * mappings.jsonl NodeMapping — the session→graph identity map: + * (chatId, sourceId) → graphNodeId * sessions.jsonl SessionCheckpoint— emission checkpoints (foldedAt) * * Implements core/ports.ts GraphStore. data/graph/* is exclusively this @@ -105,7 +105,7 @@ const isLedgerEvent = (v: unknown): v is LedgerEvent => isString(v.at); const isNodeMapping = (v: unknown): v is NodeMapping => - isRecord(v) && isString(v.chatId) && isString(v.treeNodeId) && isString(v.graphNodeId); + isRecord(v) && isString(v.chatId) && isString(v.sourceId) && isString(v.graphNodeId); const isCheckpoint = (v: unknown): v is SessionCheckpoint => isRecord(v) && isString(v.chatId) && isString(v.foldedAt); diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts new file mode 100644 index 0000000..fbdd3f7 --- /dev/null +++ b/tests/extraction.test.ts @@ -0,0 +1,373 @@ +/** + * The extraction pipeline tests (vision §1 construction): extractor → + * auditor → merger, the sole graph writer. + * + * transcriptView — the compact view both LLM stages see (pure) + * pipeline — MockLanguageModelV4 drives both agents: findings out, + * audit verdict in, deterministic merge applied; failure + * paths checkpoint nothing; single-flight per chat + * + * The merger itself (identity resolution, event rebuild, idempotency) is + * tested in tests/graph.test.ts directly against mergeFindings — here it is + * the integration: what one full pipeline run does to the ledgers. + * + * Ledger: in-memory (the GraphStore port) — this suite never touches + * data/graph. Chats: real ChatRecords in the real chat store (created and + * deleted by this suite). + */ + +import assert from "node:assert/strict"; +import * as fs from "node:fs"; +import { after, describe, it } from "node:test"; +import { MockLanguageModelV4 } from "ai/test"; +import type { UIMessage } from "ai"; +import type { GraphEdge, GraphNode, LedgerEvent, NodeMapping, SessionCheckpoint, SessionFindings } from "../src/core/contracts.ts"; +import type { GraphStore } from "../src/core/ports.ts"; +import { createTree, markDone } from "../src/core/plan-tree.ts"; +import { createExtractionPipeline, transcriptView } from "../src/agent/subagents/extraction.ts"; +import { chatStore } from "../src/store/chats.ts"; +import { PATHS } from "../config.ts"; + +// ─── fixtures ──────────────────────────────────────────────────────────────── + +const createdChats: string[] = []; + +function usage() { + return { + inputTokens: { total: 10, noCache: 10, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 5, text: 5, reasoning: 0 }, + }; +} + +/** An in-memory GraphStore (core/ports.ts) — this suite never touches data/graph. */ +function memoryLedgers() { + const nodes: GraphNode[] = []; + const edges: GraphEdge[] = []; + const events: LedgerEvent[] = []; + const mappings: NodeMapping[] = []; + const checkpoints: SessionCheckpoint[] = []; + return { + appendNode: (n: GraphNode) => void nodes.push(n), + readNodes: () => nodes.map((n) => ({ ...n })), + appendEdge: (e: GraphEdge) => void edges.push(e), + readEdges: () => edges.map((e) => ({ ...e })), + appendEvent: (e: LedgerEvent) => void events.push(e), + readEvents: () => events.map((e) => ({ ...e })), + appendMapping: (m: NodeMapping) => void mappings.push(m), + readMappings: () => mappings.map((m) => ({ ...m })), + appendCheckpoint: (c: SessionCheckpoint) => void checkpoints.push(c), + readCheckpoints: () => checkpoints.map((c) => ({ ...c })), + } satisfies GraphStore; +} + +function assistant(id: string, parts: unknown[]): UIMessage { + return { id, role: "assistant", parts } as unknown as UIMessage; +} + +function user(id: string, text: string): UIMessage { + return { id, role: "user", parts: [{ type: "text", text }] } as unknown as UIMessage; +} + +function quizPart(toolCallId: string, input: Record, output: Record): unknown { + return { type: "tool-quiz", toolCallId, state: "output-available", input, output }; +} + +function planPart(toolCallId: string, name: string, input: Record, tree: ReturnType): unknown { + return { + type: `tool-${name}`, + toolCallId, + state: "output-available", + input, + output: { ok: true, tree }, + }; +} + +/** A session-shaped fixture chat: probe → plan → check → mark_done. */ +function fixtureChat() { + const tree = markDone( + createTree( + [ + { id: "cr", title: "Chain rule intuition" }, + { id: "apply", title: "Applying the chain rule" }, + ], + [{ from: "cr", to: "apply" }], + ), + "cr", + ); + const chat = chatStore.createChat("Extraction fixture"); + createdChats.push(chat.id); + chatStore.saveTree(chat.id, tree); + chatStore.saveMessages(chat.id, [ + user("u1", "teach me the chain rule"), + assistant("a1", [ + quizPart("probe-1", { question: "What is a derivative, roughly?" }, { + questionId: "probe-1", kind: "quiz", questionText: "What is a derivative, roughly?", + correct: false, dontKnow: false, answeredAt: "2026-08-22T01:00:00.000Z", + }), + ]), + assistant("a2", [ + { type: "text", text: "The derivative measures how fast a function changes." }, + planPart("p1", "plan_init", { nodes: [], edges: [] }, tree), + ]), + assistant("a3", [ + quizPart("check-1", { question: "Check", nodeId: "apply" }, { + questionId: "check-1", kind: "quiz", questionText: "Check", correct: true, dontKnow: false, + answeredAt: "2026-08-22T02:00:00.000Z", + }), + ]), + assistant("a4", [planPart("p2", "plan_mark_done", { id: "cr" }, tree)]), + ]); + return { chatId: chat.id, record: chatStore.getChat(chat.id)! }; +} + +after(() => { + for (const id of createdChats) { + try { + fs.unlinkSync(`${PATHS.chats}/${id}.json`); + } catch { + /* already gone */ + } + } +}); + +/** What the extractor would produce for the fixture session. */ +const FINDINGS: SessionFindings = { + propositions: [ + { + id: "p1", + statement: "Chain rule intuition", + aliases: [], + kind: "truth", + treeNodeId: "cr", + evidence: ["TEACHER: The derivative measures how fast a function changes."], + }, + { + id: "p2", + statement: "Applying the chain rule", + aliases: [], + kind: "truth", + treeNodeId: "apply", + evidence: ["FINAL PLAN TREE (node ids): cr = Chain rule intuition; apply = Applying the chain rule"], + }, + ], + edges: [{ from: "p1", to: "p2", type: "requires", evidence: ["TREE EDGES: cr requires apply"] }], + attribution: { quiz: { "probe-1": "p1" } }, +}; + +/** A model whose ONE tool call is `name` with the given input; then text. */ +function submitModel(name: string, input: unknown): MockLanguageModelV4 { + let turn = 0; + return new MockLanguageModelV4({ + doGenerate: async () => { + turn++; + if (turn === 1) { + return { + content: [ + { + type: "tool-call" as const, + toolCallId: `${name}-call`, + toolName: name, + input: JSON.stringify(input), + }, + ], + finishReason: { unified: "tool-calls" as const, raw: "tool-calls" }, + usage: usage(), + warnings: [], + }; + } + return { + content: [{ type: "text" as const, text: "submitted" }], + finishReason: { unified: "stop" as const, raw: "stop" }, + usage: usage(), + warnings: [], + }; + }, + }); +} + +/** A model that never calls a tool (the failure path). */ +function silentModel(): MockLanguageModelV4 { + return new MockLanguageModelV4({ + doGenerate: async () => ({ + content: [{ type: "text" as const, text: "I refuse to use tools" }], + finishReason: { unified: "stop" as const, raw: "stop" }, + usage: usage(), + warnings: [], + }), + }); +} + +// ─── transcriptView (pure) ─────────────────────────────────────────────────── + +describe("transcriptView (pure)", () => { + it("shows the final tree, verbatim conversation, structured events — plan ops folded", () => { + const { record } = fixtureChat(); + const view = transcriptView(record); + + // The final tree with node ids (the extractor's backbone + the + // pruned-out stays pruned — only the FINAL tree is shown). + assert.match(view, /SESSION: Extraction fixture/); + assert.match(view, /FINAL PLAN TREE \(node ids\): cr = Chain rule intuition; apply = Applying the chain rule/); + assert.match(view, /TREE EDGES: cr requires apply/); + + // Conversation verbatim — evidence quotes can come from it. + assert.match(view, /LEARNER: teach me the chain rule/); + assert.match(view, /TEACHER: The derivative measures how fast a function changes\./); + + // Structured events, attribution-ready. + assert.match(view, /QUIZ \[probe-1\] What is a derivative, roughly\? — the learner answered incorrectly/); + assert.match(view, /QUIZ \[check-1\] Check — the learner answered correctly/); + assert.match(view, /TEACHER marked tree node done \[p2\]: cr/); + + // plan_init / plan_mark_done tool parts themselves are folded into + // the tree line — no raw JSON noise. + assert.doesNotMatch(view, /plan_init/); + assert.doesNotMatch(view, /output-available/); + }); + + it("summarizes job tool results as one line and drops empty text", () => { + const chat = chatStore.createChat("View noise fixture"); + createdChats.push(chat.id); + chatStore.saveMessages(chat.id, [ + user("u1", "go"), + assistant("a1", [ + { + type: "tool-research", + toolCallId: "r1", + state: "output-available", + input: { question: "q" }, + output: { brief: "long brief", sources: [] }, + }, + { type: "text", text: " " }, + ]), + ]); + const view = transcriptView(chatStore.getChat(chat.id)!); + assert.match(view, /\(research tool result delivered\)/); + assert.doesNotMatch(view, /long brief/); + assert.doesNotMatch(view, /TEACHER: {2}/); + }); +}); + +// ─── the pipeline ─────────────────────────────────────────────────────────── + +describe("the extraction pipeline (extractor → auditor → merger)", () => { + it("runs end to end: findings → audit verdict → deterministic merge + checkpoint", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + + // The auditor passes p1 through, drops p2 (unbacked), re-points the + // probe attribution to p1, and records the drop. + const verified: SessionFindings = { + propositions: [FINDINGS.propositions[0]!], + edges: [], + attribution: { quiz: { "probe-1": "p1" } }, + }; + + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", FINDINGS), + auditor: submitModel("submit_review", { ...verified, dropped: [{ ref: "p2", reason: "evidence not found" }] }), + }); + const stats = await pipeline.runExtraction(chatId); + + assert.deepEqual(stats, { nodes: 1, edges: 0, events: 3 }); + + // Only the audited proposition became a node; mapping provenance kept. + assert.equal(ledgers.readNodes().length, 1); + assert.equal(ledgers.readNodes()[0]!.statement, "Chain rule intuition"); + assert.deepEqual(ledgers.readMappings(), [{ chatId, sourceId: "cr", graphNodeId: ledgers.readNodes()[0]!.id }]); + + // Events rebuilt deterministically from the transcript — the probe + // retroactively attributed, the check tagged, mark_done mapped. + const events = ledgers.readEvents(); + assert.deepEqual( + events.map((e) => [e.kind, e.ref]), + [ + ["probe-miss", "probe-1"], + ["check-correct", "check-1"], + ["taught", "taught:cr"], + ], + ); + const nodeId = ledgers.readNodes()[0]!.id; + assert.equal(events[0]!.nodeId, nodeId, "probe attributed via audit"); + assert.equal(events[1]!.nodeId, null, "tagged to a dropped proposition → null"); + assert.equal(events[2]!.nodeId, nodeId, "taught via tree-node mapping"); + + // Checkpointed through the record's updatedAt — never the same turns twice. + assert.equal(ledgers.readCheckpoints().length, 1); + }); + + it("checkpoints nothing when a stage fails", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", FINDINGS), + auditor: silentModel(), // never submits the review + }); + await assert.rejects(pipeline.runExtraction(chatId), /the auditor never submitted a review/); + assert.equal(ledgers.readNodes().length, 0); + assert.equal(ledgers.readCheckpoints().length, 0, "no checkpoint — the next trigger re-runs"); + + const failingExtractor = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: silentModel(), + auditor: submitModel("submit_review", FINDINGS), + }); + await assert.rejects(failingExtractor.runExtraction(chatId), /the extractor never submitted findings/); + assert.equal(ledgers.readCheckpoints().length, 0); + }); + + it("is single-flight per chat — a concurrent second run is a no-op", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + + // The extractor's first call hangs until we release it. + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let calls = 0; + const gatedExtractor = new MockLanguageModelV4({ + doGenerate: async () => { + calls++; + if (calls === 1) await gate; + return { + content: [{ type: "text" as const, text: "…" }], + finishReason: { unified: "stop" as const, raw: "stop" }, + usage: usage(), + warnings: [], + }; + }, + }); + + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: gatedExtractor, + auditor: silentModel(), + }); + + const first = pipeline.runExtraction(chatId); + const second = await pipeline.runExtraction(chatId); // in-flight → zeros + assert.deepEqual(second, { nodes: 0, edges: 0, events: 0 }); + + release(); + await assert.rejects(first, /the extractor never submitted findings/); + }); + + it("throws for unknown chats", async () => { + const ledgers = memoryLedgers(); + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: silentModel(), + auditor: silentModel(), + }); + await assert.rejects(pipeline.runExtraction("chat_does_not_exist"), /Chat not found/); + }); +}); diff --git a/tests/graph-context.test.ts b/tests/graph-context.test.ts index b15eb85..ab845ec 100644 --- a/tests/graph-context.test.ts +++ b/tests/graph-context.test.ts @@ -5,10 +5,11 @@ * buildGraphContext — per-request entry point (derived-graph → block) * injection plumbing — chat-flow style: the teacher's instructions carry * the block (MockLanguageModelV4 captures the prompt) - * exact-match merge — emitChat dedupes case/whitespace-identical titles - * onto existing nodes (no fuzzy matching) + * exact-match merge — the merger (mergeFindings) dedupes + * case/whitespace-identical statements onto existing + * nodes (no fuzzy matching) * - * Ledger isolation: NONE needed for the merge tests — emitChat's ledger is + * Ledger isolation: NONE needed for the merge tests — the merger's ledger is * injectable (in-memory), so the live data/graph files are only ever READ * here (tests/graph.test.ts owns the backup-and-clear dance; both suites run * concurrently as separate processes). Chat fixtures are real ChatRecords, @@ -21,9 +22,9 @@ import * as path from "node:path"; import { after, describe, it } from "node:test"; import { MockLanguageModelV4, simulateReadableStream } from "ai/test"; import type { LanguageModelV4StreamPart } from "@ai-sdk/provider"; -import type { GraphEdge, GraphNode, LedgerEvent } from "../src/core/contracts.ts"; +import type { GraphEdge, GraphNode, LedgerEvent, SessionFindings } from "../src/core/contracts.ts"; import type { NodeStateClass } from "../src/core/contracts.ts"; -import { emitChat, getGraph } from "../src/core/graph.ts"; +import { getGraph, mergeFindings } from "../src/core/graph.ts"; import type { GraphStore } from "../src/core/ports.ts"; import { buildGraphContext, @@ -295,7 +296,7 @@ describe("graph-context injection (chat-flow style)", () => { }); }); -// ─── exact-match merge (emitChat dedup) ───────────────────────────────────── +// ─── exact-match merge (merger identity resolution) ────────────────────────── /** An in-memory stand-in for the five ledgers (the GraphStore port, core/ports.ts). */ function memoryLedgers() { @@ -320,7 +321,31 @@ function memoryLedgers() { } satisfies GraphStore & { counts: { nodes: number; mappings: number } }; } -/** A real chat record with a stored tree (emitChat reads the real chat store). */ +/** Findings as the extractor would file them for a plan tree: one + * proposition per node (statement = title, treeNodeId set), one `requires` + * edge per tree edge — the merge tests drive the merger with these. */ +function treeFindings(tree: ReturnType): SessionFindings { + const idByNodeId = new Map(tree.nodes.map((n, i) => [n.id, `p${i + 1}`])); + return { + propositions: tree.nodes.map((n, i) => ({ + id: `p${i + 1}`, + statement: n.title, + aliases: [], + kind: "truth" as const, + treeNodeId: n.id, + evidence: ["fixture evidence"], + })), + edges: tree.edges.map((e) => ({ + from: idByNodeId.get(e.from)!, + to: idByNodeId.get(e.to)!, + type: "requires" as const, + evidence: ["fixture evidence"], + })), + attribution: { quiz: {} }, + }; +} + +/** A real chat record with a stored tree (the merger reads the real chat store). */ function fixtureChat(title: string, tree: ReturnType) { const chat = chatStore.createChat(title); createdChats.push(chat.id); @@ -329,24 +354,26 @@ function fixtureChat(title: string, tree: ReturnType) { return chatStore.getChat(chat.id)!; } -describe("exact-match merge (emitChat)", () => { +describe("exact-match merge (merger identity resolution)", () => { it("maps a case/whitespace-identical title onto the existing node instead of duplicating", () => { const ledgers = memoryLedgers(); - const chat1 = fixtureChat("Merge A", createTree([{ id: "cr", title: "Chain rule intuition" }], [])); - let stats = emitChat(chat1.id, { chatStore, graphStore: ledgers }); + const tree1 = createTree([{ id: "cr", title: "Chain rule intuition" }], []); + const chat1 = fixtureChat("Merge A", tree1); + let stats = mergeFindings(chat1.id, treeFindings(tree1), { chatStore, graphStore: ledgers }); assert.deepEqual(stats, { nodes: 1, edges: 0, events: 0 }); const firstId = ledgers.readNodes()[0]!.id; // Second session, same proposition: different casing + padding. - const chat2 = fixtureChat("Merge B", createTree([{ id: "cr2", title: " chain Rule INTUITION " }], [])); - stats = emitChat(chat2.id, { chatStore, graphStore: ledgers }); + const tree2 = createTree([{ id: "cr2", title: " chain Rule INTUITION " }], []); + const chat2 = fixtureChat("Merge B", tree2); + stats = mergeFindings(chat2.id, treeFindings(tree2), { chatStore, graphStore: ledgers }); assert.deepEqual(stats, { nodes: 0, edges: 0, events: 0 }, "no duplicate node appended"); assert.equal(ledgers.counts.nodes, 1, "still exactly one graph node"); assert.equal(ledgers.readNodes()[0]!.id, firstId); - // Provenance stays honest: one mapping line per (chat, treeNodeId). + // Provenance stays honest: one mapping line per (chat, graph node). assert.equal(ledgers.counts.mappings, 2); - assert.deepEqual(ledgers.readMappings()[1], { chatId: chat2.id, treeNodeId: "cr2", graphNodeId: firstId }); + assert.deepEqual(ledgers.readMappings()[1], { chatId: chat2.id, sourceId: "cr2", graphNodeId: firstId }); }); it("matches aliases too (exact, case-insensitive)", () => { @@ -358,52 +385,49 @@ describe("exact-match merge (emitChat)", () => { kind: "truth", createdAt: "2026-08-22T00:00:00.000Z", }); - const chat = fixtureChat("Merge alias", createTree([{ id: "d", title: "The Derivative" }], [])); - const stats = emitChat(chat.id, { chatStore, graphStore: ledgers }); + const tree = createTree([{ id: "d", title: "The Derivative" }], []); + const chat = fixtureChat("Merge alias", tree); + const stats = mergeFindings(chat.id, treeFindings(tree), { chatStore, graphStore: ledgers }); assert.deepEqual(stats, { nodes: 0, edges: 0, events: 0 }, "alias match → no new node"); - assert.deepEqual(ledgers.readMappings(), [{ chatId: chat.id, treeNodeId: "d", graphNodeId: "n_alias" }]); + assert.deepEqual(ledgers.readMappings(), [{ chatId: chat.id, sourceId: "d", graphNodeId: "n_alias" }]); }); it("does NOT fuzzy-match near-identical titles (that stays deferred)", () => { const ledgers = memoryLedgers(); - const chat1 = fixtureChat("Fuzzy A", createTree([{ id: "cr", title: "Chain rule intuition" }], [])); - emitChat(chat1.id, { chatStore, graphStore: ledgers }); - const chat2 = fixtureChat( - "Fuzzy B", - createTree([{ id: "cr2", title: "Chain rules intuition" }], []), - ); - const stats = emitChat(chat2.id, { chatStore, graphStore: ledgers }); - assert.equal(stats.nodes, 1, "near-miss titles stay separate nodes"); + const tree1 = createTree([{ id: "cr", title: "Chain rule intuition" }], []); + const chat1 = fixtureChat("Fuzzy A", tree1); + mergeFindings(chat1.id, treeFindings(tree1), { chatStore, graphStore: ledgers }); + const tree2 = createTree([{ id: "cr2", title: "Chain rules intuition" }], []); + const chat2 = fixtureChat("Fuzzy B", tree2); + const stats = mergeFindings(chat2.id, treeFindings(tree2), { chatStore, graphStore: ledgers }); + assert.equal(stats.nodes, 1, "near-miss statements stay separate nodes"); assert.equal(ledgers.counts.nodes, 2); }); it("merged nodes unify the second chat's edges onto the existing graph node", () => { const ledgers = memoryLedgers(); - const chat1 = fixtureChat( - "Edge A", - createTree( - [ - { id: "base", title: "Derivative as a limit" }, - { id: "chain", title: "Chain rule intuition" }, - ], - [{ from: "base", to: "chain" }], - ), + const tree1 = createTree( + [ + { id: "base", title: "Derivative as a limit" }, + { id: "chain", title: "Chain rule intuition" }, + ], + [{ from: "base", to: "chain" }], ); - emitChat(chat1.id, { chatStore, graphStore: ledgers }); + const chat1 = fixtureChat("Edge A", tree1); + mergeFindings(chat1.id, treeFindings(tree1), { chatStore, graphStore: ledgers }); - // Second session re-teaches the same two propositions (exact titles) - // and asserts the same edge — it must dedupe onto the existing nodes. - const chat2 = fixtureChat( - "Edge B", - createTree( - [ - { id: "base", title: "derivative as a limit" }, - { id: "chain", title: "chain rule intuition" }, - ], - [{ from: "base", to: "chain" }], - ), + // Second session re-teaches the same two propositions (exact statements + // modulo case) and asserts the same edge — it must dedupe onto the + // existing nodes. + const tree2 = createTree( + [ + { id: "base", title: "derivative as a limit" }, + { id: "chain", title: "chain rule intuition" }, + ], + [{ from: "base", to: "chain" }], ); - const stats = emitChat(chat2.id, { chatStore, graphStore: ledgers }); + const chat2 = fixtureChat("Edge B", tree2); + const stats = mergeFindings(chat2.id, treeFindings(tree2), { chatStore, graphStore: ledgers }); // A new ASSERTION line is appended (provenance is honest) — reads merge // by (from,to,type), so the derived graph still has one edge with both // sessions in its provenance. diff --git a/tests/graph.test.ts b/tests/graph.test.ts index 1c55bde..7e74a1a 100644 --- a/tests/graph.test.ts +++ b/tests/graph.test.ts @@ -1,21 +1,27 @@ /** - * Graph store + emission tests (W4; vision §1, contracts.ts "v1 emission"). + * Graph merger + trigger + derivation tests (vision §1; contracts.ts + * "extraction pipeline"). The merger (core/graph.ts mergeFindings) is the + * pipeline's deterministic apply stage: findings in, ledger appends out — + * identity resolution, retroactive probe attribution, event rebuild, and + * the idempotency guarantees. * * Ledger isolation: PATHS is fixed at import time, so the five data/graph * ledgers are backed up and cleared before the suite and restored after * (try/finally-shaped: the after() hook always runs, even on failure). * The chat fixtures are real ChatRecords in the chat store (created and - * deleted by this suite) — emission reads exactly what production reads. + * deleted by this suite) — the merger reads exactly what production reads. + * The full LLM pipeline (extractor → auditor) is tested in + * tests/extraction.test.ts with mock models. */ import assert from "node:assert/strict"; import * as fs from "node:fs"; import { after, describe, it } from "node:test"; import type { UIMessage } from "ai"; -import type { PlanTree, PlanToolResult } from "../src/core/contracts.ts"; +import type { PlanTree, PlanToolResult, SessionFindings } from "../src/core/contracts.ts"; import type { QuestionAnswerData } from "../src/core/question-core.ts"; -import { createTree, markDone, removeNode } from "../src/core/plan-tree.ts"; -import { emitChat, foldStaleChats, getGraph, graphSnapshot, type GraphDeps } from "../src/core/graph.ts"; +import { createTree, markDone } from "../src/core/plan-tree.ts"; +import { getGraph, graphSnapshot, mergeFindings, pendingQuizAnswers, 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 } from "../config.ts"; @@ -32,7 +38,7 @@ const createdChats: string[] = []; fs.mkdirSync(PATHS.graph, { recursive: true }); for (const file of LEDGER_PATHS) { if (fs.existsSync(file)) { - const bak = `${file}.w4test.bak`; + const bak = `${file}.w5test.bak`; fs.renameSync(file, bak); backups.push({ file, bak }); } @@ -120,6 +126,32 @@ function fixtureTree(): PlanTree { ); } +/** + * Findings shaped exactly like what the extractor would produce for a + * session's plan tree: one proposition per tree node (statement = title, + * treeNodeId set), one `requires` edge per tree edge. + */ +function treeFindings(tree: PlanTree): SessionFindings { + const idByNodeId = new Map(tree.nodes.map((n, i) => [n.id, `p${i + 1}`])); + return { + propositions: tree.nodes.map((n, i) => ({ + id: `p${i + 1}`, + statement: n.title, + aliases: [], + kind: "truth" as const, + treeNodeId: n.id, + evidence: ["fixture evidence"], + })), + edges: tree.edges.map((e) => ({ + from: idByNodeId.get(e.from)!, + to: idByNodeId.get(e.to)!, + type: "requires" as const, + evidence: ["fixture evidence"], + })), + attribution: { quiz: {} }, + }; +} + /** The full fixture transcript: probes → plan → checks → mark_done → noise. */ function fixtureMessages(tree: PlanTree): UIMessage[] { return [ @@ -154,7 +186,8 @@ function fixtureMessages(tree: PlanTree): UIMessage[] { ), ]), assistant("a6", [ - // nodeId present but NOT in the tree → unmapped → null + question text + // nodeId present but NOT in the tree → tagged-but-unmapped: the + // attribution can rescue it, or it lands null + question text. quizPart( "call-5", { question: "Q5", nodeId: "ghost", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, @@ -163,7 +196,7 @@ function fixtureMessages(tree: PlanTree): UIMessage[] { ]), assistant("a7", [planPart("p2", "plan_mark_done", { id: "alpha" }, tree)]), assistant("a8", [ - // dismissed quiz + an answered ask — neither emits an event + // dismissed quiz + an answered ask — neither becomes an event quizPart( "call-6", { question: "Q6", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, @@ -180,13 +213,27 @@ function fixtureMessages(tree: PlanTree): UIMessage[] { ]; } -/** The chat every emission test folds (shared, so ordering is intentional). */ +/** + * The extractor's findings for the fixture chat: the tree as propositions + + * the probe attribution. The audit stage is a no-op pass-through in these + * merger tests — its behavior is tested with the pipeline (extraction.test). + */ +function fixtureFindings(): SessionFindings { + const findings = treeFindings(fixtureTree()); + // Untagged probe quizzes: call-1 attributed retroactively to Alpha (the + // vision's open question — resolved by the extractor); call-2 genuinely + // unattributable; call-5's ghost nodeId is rescued onto Gamma. + findings.attribution.quiz = { "call-1": "p1", "call-2": null, "call-5": "p3" }; + return findings; +} + +/** The chat every merger test folds (shared, so ordering is intentional). */ const chat1 = fixtureChat("Graph fixture", fixtureTree(), fixtureMessages(fixtureTree())); -/** Set by the provenance test; the snapshot test (which runs after it) reads it. */ +/** Set by the cross-session test; the snapshot test (which runs after it) reads it. */ let chat2Id = ""; -// ─── emission ──────────────────────────────────────────────────────────────── +// ─── the merger ────────────────────────────────────────────────────────────── /** Everything the graph holds for a set of chats (other processes' test * emissions run concurrently against the same files — scope, never assume). */ @@ -205,9 +252,9 @@ function scopedGraph(chatIds: string[]) { }; } -describe("graph emission (emitChat)", () => { - it("folds tree nodes/edges and quiz/mark_done events with the probe/check boundary", () => { - const stats = emitChat(chat1.id, graphDeps); +describe("the merger (mergeFindings)", () => { + it("merges propositions/edges by identity and rebuilds events with retroactive probe attribution", () => { + const stats = mergeFindings(chat1.id, fixtureFindings(), graphDeps); assert.deepEqual(stats, { nodes: 3, edges: 1, events: 6 }); const graph = scopedGraph([chat1.id]); @@ -219,7 +266,7 @@ describe("graph emission (emitChat)", () => { assert.match(node.id, /^n_/); } - // Tree edge alpha→beta becomes a `requires` edge on the mapped nodes. + // The tree edge alpha→beta becomes a `requires` edge on the mapped nodes. assert.equal(graph.edges.length, 1); const edge = graph.edges[0]!; assert.equal(edge.from, idByStatement.get("Alpha intuition")); @@ -227,77 +274,69 @@ describe("graph emission (emitChat)", () => { assert.equal(edge.type, "requires"); assert.deepEqual(edge.provenance, [chat1.id]); - // Events — kinds, refs, node attribution, notes, question fallback. + // Events — kinds, refs, notes, and the attribution paths. assert.equal(graph.events.length, 6); const byRef = new Map(graph.events.map((e) => [e.ref, e])); const probeMiss = byRef.get("call-1")!; assert.equal(probeMiss.kind, "probe-miss"); // quiz BEFORE plan_init - assert.equal(probeMiss.nodeId, null); - assert.equal(probeMiss.question, "Q1 text"); - assert.equal(probeMiss.chatId, chat1.id); + assert.equal(probeMiss.nodeId, idByStatement.get("Alpha intuition"), "retroactive probe attribution"); + assert.equal(probeMiss.question, undefined, "attributed → no question text needed"); const probeDontKnow = byRef.get("call-2")!; assert.equal(probeDontKnow.kind, "dont-know"); // same kind, both phases - assert.equal(probeDontKnow.nodeId, null); + assert.equal(probeDontKnow.nodeId, null, "unattributable probe → null"); + assert.equal(probeDontKnow.question, "Q2 text"); const check = byRef.get("call-3")!; assert.equal(check.kind, "check-correct"); // quiz AFTER plan_init - assert.equal(check.nodeId, idByStatement.get("Alpha intuition")); + assert.equal(check.nodeId, idByStatement.get("Alpha intuition"), "tagged quiz → deterministic"); assert.equal(check.note, "recalled the diagram"); - assert.equal(check.question, undefined); // resolved nodeId → no question text const checkMiss = byRef.get("call-4")!; assert.equal(checkMiss.kind, "check-incorrect"); assert.equal(checkMiss.nodeId, idByStatement.get("Beta rule")); - const unmapped = byRef.get("call-5")!; - assert.equal(unmapped.kind, "dont-know"); - assert.equal(unmapped.nodeId, null); // nodeId present but unmapped - assert.equal(unmapped.question, "Q5 text"); + const ghost = byRef.get("call-5")!; + assert.equal(ghost.kind, "dont-know"); + assert.equal(ghost.nodeId, idByStatement.get("Gamma application"), "tagged-but-unmapped rescued by attribution"); + assert.equal(ghost.question, undefined); const taught = byRef.get("taught:alpha")!; assert.equal(taught.kind, "taught"); - assert.equal(taught.nodeId, idByStatement.get("Alpha intuition")); + assert.equal(taught.nodeId, idByStatement.get("Alpha intuition"), "mark_done → tree-node mapping"); - // The dismissed quiz and the ask question emitted nothing. + // The dismissed quiz and the ask question became nothing. assert.equal([...byRef.keys()].filter((r) => r === "call-6" || r === "call-7").length, 0); }); - it("is idempotent — re-emission appends nothing", () => { - const stats = emitChat(chat1.id, graphDeps); + it("is idempotent — re-merging the same findings appends nothing", () => { + const stats = mergeFindings(chat1.id, fixtureFindings(), graphDeps); assert.deepEqual(stats, { nodes: 0, edges: 0, events: 0 }); const graph = scopedGraph([chat1.id]); assert.equal(graph.nodes.length, 3); assert.equal(graph.edges.length, 1); assert.equal(graph.events.length, 6); - // exactly one checkpoint line per fold, though — checkpoints are a log + // exactly one checkpoint line per merge, though — checkpoints are a log assert.equal(graphStore.readCheckpoints().filter((c) => c.chatId === chat1.id).length, 2); }); - it("accumulates edge provenance when another chat maps onto the same nodes", () => { - // Simulates what a future consolidation merge produces: chat2's tree - // slugs map onto chat1's graph nodes, so its edge assertion dedupes - // by (from,to,type) and just adds chat2 to the provenance. + it("merges a second session onto the same nodes and accumulates edge provenance", () => { + // chat2 re-teaches the same two propositions (identical statements) — + // the conservative identity resolution recognizes them. const tree2 = createTree( [ - { id: "alpha", title: "Alpha (second session)" }, - { id: "beta", title: "Beta (second session)" }, + { id: "a", title: "Alpha intuition" }, + { id: "b", title: "Beta rule" }, ], - [{ from: "alpha", to: "beta" }], + [{ from: "a", to: "b" }], ); const chat2 = fixtureChat("Graph fixture 2", tree2, [ user("u1", "again"), assistant("a1", [planPart("p1", "plan_init", { nodes: [], edges: [] }, tree2)]), ]); - const before = getGraph(graphStore); - const gAlpha = before.nodes.find((n) => n.statement === "Alpha intuition")!.id; - const gBeta = before.nodes.find((n) => n.statement === "Beta rule")!.id; - graphStore.appendMapping({ chatId: chat2.id, treeNodeId: "alpha", graphNodeId: gAlpha }); - graphStore.appendMapping({ chatId: chat2.id, treeNodeId: "beta", graphNodeId: gBeta }); - - const stats = emitChat(chat2.id, graphDeps); + const stats = mergeFindings(chat2.id, treeFindings(tree2), graphDeps); assert.deepEqual(stats, { nodes: 0, edges: 1, events: 0 }); chat2Id = chat2.id; @@ -312,7 +351,9 @@ describe("graph emission (emitChat)", () => { const idByStatement = new Map(snap.nodes.map((n) => [n.statement, n.id])); assert.equal(snap.nodeStates[idByStatement.get("Alpha intuition")!], "taught"); assert.equal(snap.nodeStates[idByStatement.get("Beta rule")!], "missed"); - assert.equal(snap.nodeStates[idByStatement.get("Gamma application")!], "candidate"); + // Gamma: no taught/checks — but the rescued dont-know probe (call-5) + // makes its latest event a dont-know. + assert.equal(snap.nodeStates[idByStatement.get("Gamma application")!], "dont-know"); assert.equal(snap.sessions[chat1.id], "Graph fixture"); assert.equal(snap.sessions[chat2Id], "Graph fixture 2"); @@ -320,44 +361,25 @@ describe("graph emission (emitChat)", () => { (snap.nodeSessions[idByStatement.get("Alpha intuition")!] ?? []).sort(), [chat1.id, chat2Id].sort(), ); - // Gamma was only ever created (mapping) by chat1. + // Gamma was only ever touched by chat1. assert.deepEqual(snap.nodeSessions[idByStatement.get("Gamma application")!], [chat1.id]); }); - it("replays plan tool outputs when record.tree is missing (removed nodes stay out)", () => { - const initTree = createTree( - [ - { id: "x", title: "X live" }, - { id: "y", title: "Y pruned" }, - ], - [{ from: "x", to: "y" }], - ); - const prunedTree = removeNode(initTree, "y"); - const chat = fixtureChat("Replay fixture", null, [ - user("u1", "go"), - assistant("a1", [planPart("p1", "plan_init", { nodes: [], edges: [] }, initTree)]), - assistant("a2", [planPart("p2", "plan_remove_node", { id: "y" }, prunedTree)]), - ]); - - const stats = emitChat(chat.id, graphDeps); - assert.deepEqual(stats, { nodes: 1, edges: 0, events: 0 }); // y pruned, edge cascaded away - - const graph = scopedGraph([chat.id]); - assert.ok(graph.nodes.some((n) => n.statement === "X live")); - assert.ok(!graph.nodes.some((n) => n.statement === "Y pruned")); - }); - it("returns zeros for unknown chats", () => { - assert.deepEqual(emitChat("chat_does_not_exist", graphDeps), { nodes: 0, edges: 0, events: 0 }); + assert.deepEqual(mergeFindings("chat_does_not_exist", treeFindings(fixtureTree()), graphDeps), { + nodes: 0, + edges: 0, + events: 0, + }); }); }); -// chat2 from the provenance test, reachable by the snapshot test below it. +// ─── the trigger helper ────────────────────────────────────────────────────── -describe("catch-up (foldStaleChats)", () => { - it("folds chats newer than their checkpoint, exactly once, deriving known state", () => { +describe("the trigger helper (pendingQuizAnswers)", () => { + it("counts answered quizzes not yet in the ledger — dismissed/ask never count", () => { const tree = createTree([{ id: "solo", title: "Solo node" }], []); - const chat = fixtureChat("Catch-up fixture", tree, [ + const chat = fixtureChat("Trigger fixture", tree, [ user("u1", "go"), assistant("a1", [planPart("p1", "plan_init", { nodes: [], edges: [] }, tree)]), assistant("a2", [ @@ -366,23 +388,39 @@ describe("catch-up (foldStaleChats)", () => { { question: "QS", nodeId: "solo", 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" }), + }, ]), ]); - const folded = foldStaleChats(graphDeps); - assert.ok(folded >= 1, "the catch-up fixture was folded"); - assert.equal(graphStore.readCheckpoints().filter((c) => c.chatId === chat.id).length, 1); + assert.equal(pendingQuizAnswers(chat.id, graphDeps), 1, "one answered quiz pending"); + assert.equal(pendingQuizAnswers("chat_does_not_exist", graphDeps), 0); - const snap = graphSnapshot(graphDeps); - const solo = snap.nodes.find((n) => n.statement === "Solo node"); - assert.ok(solo, "solo node was emitted by the catch-up fold"); - assert.equal(snap.nodeStates[solo.id], "known"); // correct check, no taught event - assert.equal(snap.sessions[chat.id], "Catch-up fixture"); + mergeFindings(chat.id, treeFindings(tree), graphDeps); + assert.equal(pendingQuizAnswers(chat.id, graphDeps), 0, "extracted → nothing pending"); - // Already checkpointed → not folded again (other test files' chats may - // legitimately fold here, so assert on THIS chat only). - foldStaleChats(graphDeps); - assert.equal(graphStore.readCheckpoints().filter((c) => c.chatId === chat.id).length, 1); + // 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", nodeId: "solo", 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); }); });