From 543cf337891e2fd9e2baeb1e9bd90a65380975b6 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sun, 13 Sep 2026 02:08:55 +0300 Subject: [PATCH] Notes front door: format-agnostic import into the graph pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The same extractor→auditor→merger pipeline now runs over raw personal notes in arbitrary formats (vision §1, DECIDED 2026-09-13): - NotesSource records (data/graph/sources.json, deterministic id from path hash → idempotent re-imports) play the chatId role in mappings, checkpoints, and observation events; usage lands on the source record (appendSourceUsage, idempotent per startedAt, kept across upserts). - mergeFindings is chat-store-free: callers pass MergeSource {messages, foldedAt} — sessions pass the chat record, notes pass messages:null (no quiz-event rebuild; a deterministic rebuild needs structured quiz tool parts). - Second extractor prompt (NOTES_EXTRACTOR_INSTRUCTIONS) for general- purpose extraction; understanding judges the AUTHOR's grasp. Auditor wording input-agnostic. extract/audit share one generic runSubmitAgent. - graphSnapshot.fileSources (srcId→filePath); notes sources render as static path entries in the detail card (FileTextIcon, never a link). - One shared pipeline instance (exported from chat-handler): same single-flight registry, live status chips (label = file path), usage accounting, and true abort for both front doors. - Transport: POST /api/graph/import (multipart, 3-wide worker pool) + Import notes… button in the graph view; scripts/import-notes.ts for batch runs (concurrency 4). - real-data-extraction/ gitignored (raw import sources never committed). 355/355 tests; typecheck, lint, build green. --- .gitignore | 2 + app/api/graph/import/route.ts | 52 ++++ app/api/graph/status/route.ts | 4 +- components/graph-view.tsx | 82 +++++- config.ts | 7 + docs/vision.md | 27 +- scripts/import-notes.ts | 62 +++++ src/agent/subagents/extraction.ts | 404 ++++++++++++++++++++++++------ src/core/contracts/graph.ts | 30 +++ src/core/graph.ts | 52 ++-- src/core/ports.ts | 10 + src/server/chat-handler.ts | 5 +- src/store/graph.ts | 61 ++++- tests/extraction.test.ts | 136 ++++++++++ tests/graph.test.ts | 101 +++++++- tests/usage.test.ts | 40 ++- 16 files changed, 961 insertions(+), 114 deletions(-) create mode 100644 app/api/graph/import/route.ts create mode 100644 scripts/import-notes.ts diff --git a/.gitignore b/.gitignore index 1143e7e..d1030a5 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,5 @@ data/ .env.local .memory/ SCOPE.md +# raw real-data import sources (user directive 2026-09-13: never commit) +real-data-extraction/ diff --git a/app/api/graph/import/route.ts b/app/api/graph/import/route.ts new file mode 100644 index 0000000..7c44afd --- /dev/null +++ b/app/api/graph/import/route.ts @@ -0,0 +1,52 @@ +/** + * POST /api/graph/import — the format-agnostic front door's transport + * (vision §1, 2026-09-13): raw personal-notes files in, one extraction run + * per file through the SAME pipeline the sessions use. Multipart form data + * (field "files", any text format — the extractor is format-agnostic); the + * pipeline mints a deterministic source id per file path, persists the + * source record (the file's provenance + the run's usage), and runs + * extractor → auditor → merger over the raw text. + * + * Small concurrency cap: the model calls are the slow part, and a burst of + * 25 simultaneous runs would trip provider rate limits — a 3-wide worker + * pool keeps the run registry (the dock's status chips) and the ledgers + * healthy. The dock's chips + graph refetch track progress live through + * /api/graph/status, as with session runs. + */ +import { extraction } from "@/src/server"; +import type { EmissionStats } from "@/src/core"; + +export const dynamic = "force-dynamic"; + +interface ImportResult { + filePath: string; + ok: boolean; + stats?: EmissionStats; + error?: string; +} + +export async function POST(request: Request) { + const form = await request.formData(); + const files = form.getAll("files").filter((f): f is File => f instanceof File); + if (files.length === 0) { + return Response.json({ results: [] as ImportResult[], error: "no files" }, { status: 400 }); + } + + const results: ImportResult[] = []; + const queue = [...files]; + const workers = Array.from({ length: Math.min(3, files.length) }, async () => { + for (;;) { + const file = queue.shift(); + if (!file) return; + try { + const text = await file.text(); + const stats = await extraction.importNotes({ filePath: file.name, text }); + results.push({ filePath: file.name, ok: true, stats }); + } catch (err) { + results.push({ filePath: file.name, ok: false, error: err instanceof Error ? err.message : String(err) }); + } + } + }); + await Promise.all(workers); + return Response.json({ results }); +} \ No newline at end of file diff --git a/app/api/graph/status/route.ts b/app/api/graph/status/route.ts index 0ba43a6..e4a3866 100644 --- a/app/api/graph/status/route.ts +++ b/app/api/graph/status/route.ts @@ -12,7 +12,9 @@ export const dynamic = "force-dynamic"; export async function GET() { const runs = extractionStatuses().map((run) => ({ chatId: run.chatId, - title: chatStore.getChat(run.chatId)?.title ?? run.chatId, + // The run's own label (the file path for notes imports) wins; chat + // runs fall back to the chat store's title. + title: run.label ?? chatStore.getChat(run.chatId)?.title ?? run.chatId, phase: run.phase, startedAt: run.startedAt, // Live per-stage usage (the money channel): model + per-step CallUsage, diff --git a/components/graph-view.tsx b/components/graph-view.tsx index 017664e..cf8fbb9 100644 --- a/components/graph-view.tsx +++ b/components/graph-view.tsx @@ -56,7 +56,7 @@ import { useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; import Link from "next/link"; -import { Loader2Icon, SearchIcon, WaypointsIcon } from "lucide-react"; +import { FileTextIcon, Loader2Icon, SearchIcon, UploadIcon, WaypointsIcon } from "lucide-react"; import Graph from "graphology"; import forceAtlas2 from "graphology-layout-forceatlas2"; import type { NodeHoverDrawingFunction, NodeLabelDrawingFunction } from "sigma/rendering"; @@ -322,7 +322,9 @@ function tokenHex(cssVar: string): string { interface Selection { node: GraphNode; - chatIds: string[]; + /** The sources that touched this node: chat ids (deep-linkable sessions) + * and notes-source ids ("src_…" — imported files, static entries). */ + sourceIds: string[]; /** Node position in container pixels (from graphToViewport at click time). */ anchor: { x: number; y: number }; } @@ -368,6 +370,26 @@ export function GraphView() { }; }, [refreshTick]); + // Notes import (the format-agnostic front door): file picker → POST + // /api/graph/import; the runs surface in the status chips below, and the + // graph refetches when they finish (the runs-empty transition). + const fileInputRef = useRef(null); + const [importing, setImporting] = useState(false); + const onImportFiles = (files: FileList | null) => { + if (!files || files.length === 0) return; + setImporting(true); + const form = new FormData(); + for (const file of Array.from(files)) form.append("files", file); + fetch("/api/graph/import", { method: "POST", body: form }) + .then(async (res) => { + const body = (await res.json()) as { results: { filePath: string; ok: boolean }[] }; + const failed = body.results.filter((r) => !r.ok); + if (failed.length > 0) console.error("notes import failed for:", failed); + }) + .catch((err) => console.error("import failed:", err)) + .finally(() => setImporting(false)); + }; + // Background pipeline status: poll /api/graph/status; when a run // finishes (some -> none), refetch the graph. const [pipelineRuns, setPipelineRuns] = useState<{ chatId: string; title: string; phase: string; startedAt: number }[]>([]); @@ -490,7 +512,7 @@ export function GraphView() { const anchor = renderer?.graphToViewport({ x: pos.x ?? 0, y: pos.y ?? 0 }); setSelected({ node: graphNode, - chatIds: snapshot.nodeSessions[node] ?? [], + sourceIds: snapshot.nodeSessions[node] ?? [], anchor: anchor ?? { x: 0, y: 0 }, }); }); @@ -640,6 +662,28 @@ export function GraphView() { No matching nodes. )} +
+ { + onImportFiles(e.target.files); + e.target.value = ""; // allow re-picking the same file + }} + /> + +
)} @@ -670,20 +714,38 @@ export function GraphView() { >
{selected.node.name}
{selected.node.statement}
- {selected.chatIds.length > 0 ? ( + {selected.sourceIds.length > 0 ? (
Sessions
- {selected.chatIds.map((chatId) => { - // That session's understanding observation — the + {selected.sourceIds.map((sourceId) => { + // That source's understanding observation — the // extractor's named-word note (confidence history). - const note = snapshot?.nodeObservations?.[selected.node.id]?.[chatId]; + const note = snapshot?.nodeObservations?.[selected.node.id]?.[sourceId]; + // An imported-notes file: a STATIC entry — the path, + // plain text, never a link (no session to open). + if (snapshot?.fileSources?.[sourceId] !== undefined) { + return ( +
+ + + {snapshot.fileSources[sourceId]} + + {note && ( + + {note} + + )} +
+ ); + } + // A real session: deep-linkable provenance. return ( -
+
- {snapshot?.sessions[chatId] ?? chatId} + {snapshot?.sessions[sourceId] ?? sourceId} {/* The note is a fixed-scale word (solid/shaky/...) — the session name is the main thing; the chip is bounded so diff --git a/config.ts b/config.ts index 3dbfba7..b385125 100644 --- a/config.ts +++ b/config.ts @@ -72,3 +72,10 @@ export const LEDGER_FILES = { mappings: path.join(PATHS.graph, "mappings.jsonl"), sessions: path.join(PATHS.graph, "sessions.jsonl"), } as const; + +/** + * Notes-import bookkeeping (NOT a ledger): one JSON map, src-id → NotesSource + * record (contracts/graph.ts) — the file path + import time + the run's usage + * log. Read-modify-write like a chat file, not append-only like the ledgers. + */ +export const SOURCES_FILE = path.join(PATHS.graph, "sources.json"); diff --git a/docs/vision.md b/docs/vision.md index ac83348..206d5d5 100644 --- a/docs/vision.md +++ b/docs/vision.md @@ -43,12 +43,37 @@ Every node records every **(chat, message) point** where it came up; every edge Three stages, one pipeline — the sole graph writer: -- **Extractor (LLM) — session-only scope.** Sees ONLY the session transcript (tree-blind, with message ids). Reports freely, with verbatim evidence: propositions from anywhere in the session (one reasoning step each; prose counts as much as anything the teacher formalized), edges as the transcript articulates them, message-granular provenance, retroactive quiz attribution, and per proposition an **understanding level** — how strongly the learner accepted/understood it in this session, a CLOSED named-word scale (`solid` / `shaky` / `accepted` / `resisted` / `not yet engaged`; 2026-09-08: schema-enforced — exactly one word, no numbers, no justifications, which are dropped rather than stored). The levels are data for the future state model; their treatment is TBD. +- **Extractor (LLM) — single-source scope.** Sees ONLY the source text — a session transcript (tree-blind, with message ids) or raw notes (the front door's second prompt, above). Reports freely, with verbatim evidence: propositions from anywhere in the session (one reasoning step each; prose counts as much as anything the teacher formalized), edges as the transcript articulates them, message-granular provenance, retroactive quiz attribution, and per proposition an **understanding level** — how strongly the learner accepted/understood it in this session, a CLOSED named-word scale (`solid` / `shaky` / `accepted` / `resisted` / `not yet engaged`; 2026-09-08: schema-enforced — exactly one word, no numbers, no justifications, which are dropped rather than stored). The levels are data for the future state model; their treatment is TBD. - **Auditor (LLM) — the global-graph scope.** Sees the extractor's output AND the existing graph. Two jobs: the faithfulness net (every candidate verified against verbatim evidence — unchanged), and the **identity judgment**: link each proposition to an existing node (recording the session's phrasing as an alias) or create it new. Conservative: duplicates over wrong merges. - **Merger (deterministic).** Append-only ledger writes; quiz event kinds/refs rebuilt mechanically from quiz tool parts. Every judgment is an LLM's; everything mechanical stays code. **Event ledger (amended):** quiz outcomes are rebuilt deterministically from quiz tool parts (correct / miss / don't-know, plus the learner's note). **There is no `taught` event** — no marking-things-done. The extractor's per-proposition understanding levels join the ledger as session-scoped observations (one closed-scale word per node per session). Identity is no longer exact-string matching — it is the auditor's judgment, with aliases as the phrasing record. +### The format-agnostic front door — notes imports (DECIDED 2026-09-13) + +The same three-stage pipeline (extractor → auditor → merger) runs over **raw +personal notes in arbitrary formats** — an Obsidian vault, plain markdown, +anything textual. The extractor is format-blind by construction (it always +saw rendered text, never ChatRecord JSON), so the notes path needs only a +second extractor prompt (general-purpose extraction; the understanding word +judges the **author's** grasp from their own commentary) and a source +abstraction: each imported file becomes a `NotesSource` record (deterministic +id from the path hash — re-imports are idempotent), playing the `chatId` role +in mappings, checkpoints, and observation events. + +Decided edges: + +- **No session shaping.** Notes are never bent into teacher/learner + transcripts; quiz outcomes inside notes surface only via the extractor's + understanding observations (the deterministic quiz-event rebuild is a + session-pipeline feature — it needs structured quiz tool parts). +- **Static provenance.** Imported nodes show the file path in the sessions + list as a static entry — never a link. Understanding-assessment levels + apply to personal notes as to sessions. +- **Same single-flight registry and observability** as session runs: live + status, usage accounting per stage, true abort. One pipeline instance, + two front doors (quiz-triggered sessions; POST /api/graph/import). + ### Understanding is derived (amended 2026-09-02) Known vs. understood, event ledger as source of truth, derived state recomputed on read — carried over from the design doc in spirit. The state model itself is now an explicit design discussion: **node states should be session-independent and computed from the depth and strength of dependencies** (the `follows`/`supports` topology) plus quiz events and understanding notes. That redesign is pending ideation; the transitional five-class derivation (taught / known / missed / don't-know / candidate) was **removed** (2026-09-08) — the graph view renders state-neutral nodes until the state model is designed. The anchored calibration rubric stays parked until real data. diff --git a/scripts/import-notes.ts b/scripts/import-notes.ts new file mode 100644 index 0000000..c2a9735 --- /dev/null +++ b/scripts/import-notes.ts @@ -0,0 +1,62 @@ +/** + * Import raw notes files through the format-agnostic front door (vision §1, + * 2026-09-13) — every *.md / *.txt file in a directory becomes one + * extraction run (extractor → auditor → merger) over its raw text. Run + * from this worktree only — data never crosses trees: + * + * npx tsx --env-file=.env.local scripts/import-notes.ts [concurrency] + * + * Default dir: real-data-extraction/learn-sessions (the prototype + * learning-session exports — deliberately NOT imported as chats: they run + * through the notes path, raw). Runs in this process: the dev server's + * status chips do NOT see them (registry is per-process) — the dev server + * sees runs triggered through /api/graph/import. + */ +import * as fs from "node:fs"; +import * as path from "node:path"; +import { createExtractionPipeline } from "../src/agent/subagents/extraction.ts"; +import { getGraph } from "../src/core/graph.ts"; +import { stores } from "../src/server/stores.ts"; + +async function main() { + const dir = process.argv[2] ?? "real-data-extraction/learn-sessions"; + const limit = Number(process.argv[3] ?? 4); + const files = fs + .readdirSync(dir) + .filter((f) => f.endsWith(".md") || f.endsWith(".markdown") || f.endsWith(".txt")) + .sort(); + console.log(`[import-notes] ${files.length} note files from ${dir} (concurrency ${limit})`); + + const pipeline = createExtractionPipeline(stores); + const queue = [...files]; + const totals = { nodes: 0, edges: 0, events: 0, failed: 0 }; + const workers = Array.from({ length: Math.min(limit, files.length) }, async () => { + for (;;) { + const file = queue.shift(); + if (!file) return; + const filePath = path.join(dir, file); + const t0 = Date.now(); + try { + const stats = await pipeline.importNotes({ filePath, text: fs.readFileSync(filePath, "utf-8") }); + totals.nodes += stats.nodes; + totals.edges += stats.edges; + totals.events += stats.events; + console.log( + `[import-notes] ${file}: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events in ${((Date.now() - t0) / 1000).toFixed(1)}s`, + ); + } catch (err) { + totals.failed++; + console.error(`[import-notes] ${file} FAILED:`, err); + } + } + }); + await Promise.all(workers); + + const graph = getGraph(stores.graphStore); + console.log( + `[import-notes] done: +${totals.nodes} nodes, +${totals.edges} edges, +${totals.events} events, ${totals.failed} failed — ` + + `graph now ${graph.nodes.length} nodes, ${graph.edges.length} edges, ${graph.events.length} events`, + ); +} + +main(); \ No newline at end of file diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 42bd98e..2f2b608 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -1,7 +1,9 @@ +import * as crypto from "node:crypto"; import type { UIMessage } from "ai"; import type { Usage } from "@earendil-works/pi-ai"; import { z } from "zod"; import { lhTool, toPiTools } from "../lh-tool.ts"; +import type { ModelRole } from "../../../config.ts"; import type { PiEngine } from "../models.ts"; import { runSubagent } from "./pi-run.ts"; import type { @@ -42,6 +44,17 @@ import { getGraph, mergeFindings, type EmissionStats, type GraphDeps } from "../ * whatever the in-flight run missed). On failure nothing is checkpointed — * the next trigger re-runs the whole thing. * + * The FORMAT-AGNOSTIC FRONT DOOR (2026-09-13): importNotes() runs the SAME + * three stages over raw personal notes (arbitrary format — no chat, no + * session, no message ids). A notes-mode extractor prompt replaces the + * session one; the auditor and merger are shared (identity + deterministic + * appends don't care where text came from). Provenance is the file itself: + * source ids ("src_" + path hash) play the chatId role in mappings, events + * and checkpoints, and the source record (graphStore sources seam) carries + * the file path + the run's usage. No quiz events are ever built from + * notes — quiz outcomes inside the text surface only through the + * extractor's understanding observations (user-decided 2026-09-13). + * * Observability (§5 baseline): every stage logs one line per step, its * start, and its duration; the pipeline logs a failure line before * rethrowing, so the server log reads start → stage timings → merged @@ -166,6 +179,44 @@ const findingsSchema = z.object({ .describe("Quiz attribution — your only event judgment."), }); +const NOTES_EXTRACTOR_INSTRUCTIONS = `You are the extraction specialist in a learning application. You read ONE source — +the user's own NOTES, in an ARBITRARY format (personal notes, an exported conversation, a document — anything) — and +report what those notes show the author engaging with, as structured findings for their global knowledge graph. + +There is no session and no chat structure here: the source is raw text, the author's own material. Read it as such. + +Report, with verbatim evidence quotes for everything: + +NODE THEORY — the graph is a set of TRUE propositions about the world, full stop. A node is never anything about the +author or the source. The author's relationship to a proposition is expressed ENTIRELY by your understanding note; it +never leaks into the node itself. So no "the author understands X", no "mentioned in these notes": the node is just X. +A misconception is NOT a node of its own wrong claim — state the TRUE correction as the proposition. + +1. PROPOSITIONS — every distinct reasoning-step-sized proposition the notes actually engage with: + - Each has a "name" — a VERY short label, a few words, a compact handle like a glossary headword ("Will to power", + "Divine command theory"). Not a sentence; it becomes the node's label in the graph, so keep it glanceable. + - And a "statement" — the self-contained TRUE proposition at full precision. + - Notes may explain things at length or quote material the author is studying: extract the propositions themselves, + one reasoning step each — never a topic blob, never a sub-step. + - "messages": ALWAYS an empty list — provenance is the file itself. + - "aliases": verbatim phrasings the notes used for the same proposition. + - "understanding": ONE word from the fixed scale — "solid", "shaky", "accepted", "resisted", or "not yet engaged" — + judging the AUTHOR's grasp of the proposition as the notes show it: their own commentary, self-assessments, + recorded quiz outcomes, the way they write about it. Strictly one word, no numbers, no justification (a + wrong-shaped note is rejected by the schema). When the notes give no signal, omit the field. + +2. EDGES between propositions: + - "follows": to is DERIVABLE from from (understanding from makes understanding to achievable). + - "supports": from GROUNDS to without entailing it (helpful context, not a prerequisite). + - Only relations the notes actually articulate — with evidence. + +There is no quiz attribution for notes — the source has no quiz events, and the schema has no attribution field. + +Evidence discipline: every quote must appear VERBATIM in the source. The auditor rejects anything unbacked. When in +doubt, report; the auditor decides. + +Your entire deliverable is ONE submit_findings call.`; + 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. @@ -210,23 +261,30 @@ transcript). The auditor rejects anything unbacked. When in doubt, report; the a Your entire deliverable is ONE submit_findings call.`; -/** One extractor run: the whole deliverable is ONE zod-validated submit - * call (a pi agent purely so the model can retry a schema-invalid call). - * Returns the submitted findings — captured from the execute seam, which +/** One single-submit agent run — the shared shape of EVERY LLM stage + * here (session + notes): the whole deliverable is ONE zod-validated + * submit call (a pi agent purely so the model can retry a schema-invalid + * call). The submitted value is captured from the execute seam, which * only fires on a schema-valid call. */ -async function runExtractor( - engine: PiEngine | undefined, - prompt: string, - signal: AbortSignal | undefined, - onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void, -): Promise { - let submitted: SessionFindings | undefined; +async function runSubmitAgent(opts: { + role: ModelRole; + toolName: string; + toolDescription: string; + schema: z.ZodType; + systemPrompt: string; + prompt: string; + engine: PiEngine | undefined; + signal: AbortSignal | undefined; + onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void; + neverSubmitted: string; +}): Promise { + let submitted: T | undefined; const result = await runSubagent( toPiTools({ - submit_findings: lhTool({ - name: "submit_findings", - description: "Submit the session findings.", - schema: findingsSchema, + [opts.toolName]: lhTool({ + name: opts.toolName, + description: opts.toolDescription, + schema: opts.schema, execute: async (findings) => { submitted = findings; return findings; @@ -234,28 +292,71 @@ async function runExtractor( }), }), { - role: "extractor", - engine, - systemPrompt: EXTRACTOR_INSTRUCTIONS, - prompt, + role: opts.role, + engine: opts.engine, + systemPrompt: opts.systemPrompt, + prompt: opts.prompt, maxTurns: 4, - abortSignal: signal, + abortSignal: opts.signal, onAssistantMessage: (message, step) => - onStep( + opts.onStep( message.usage, message.model, step, message.stopReason, message.content.filter((c) => c.type === "toolCall").map((c) => (c as { name: string }).name).join(", ") || "(none)", ), - shouldStop: (_message, toolResults) => toolResults.some((r) => r.toolName === "submit_findings" && !r.isError), + shouldStop: (_message, toolResults) => toolResults.some((r) => r.toolName === opts.toolName && !r.isError), }, ); void result; - if (!submitted) throw new Error("the extractor never submitted findings"); + if (!submitted) throw new Error(opts.neverSubmitted); return submitted; } +/** The extractor over ONE session transcript (the chat front door). */ +function runExtractor( + engine: PiEngine | undefined, + prompt: string, + signal: AbortSignal | undefined, + onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void, +): Promise { + return runSubmitAgent({ + role: "extractor", + toolName: "submit_findings", + toolDescription: "Submit the session findings.", + schema: findingsSchema, + systemPrompt: EXTRACTOR_INSTRUCTIONS, + prompt, + engine, + signal, + onStep, + neverSubmitted: "the extractor never submitted findings", + }); +} + +/** The extractor over ONE source of raw personal notes (the format-agnostic + * front door): same single-submit shape, notes-mode prompt + schema. */ +function runNotesExtractor( + engine: PiEngine | undefined, + prompt: string, + signal: AbortSignal | undefined, + onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void, +): Promise { + return runSubmitAgent({ + role: "extractor", + toolName: "submit_findings", + toolDescription: "Submit the notes findings.", + schema: notesFindingsSchema, + systemPrompt: NOTES_EXTRACTOR_INSTRUCTIONS, + prompt, + engine, + signal, + onStep, + neverSubmitted: "the extractor never submitted findings", + }); +} + // ─── the auditor ─────────────────────────────────────────────────────────── const auditSchema = findingsSchema.extend({ @@ -279,11 +380,21 @@ const auditSchema = findingsSchema.extend({ .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, a set of proposed findings (propositions, edges, quiz attribution), and the CURRENT -knowledge graph. Your jobs: the faithfulness net, and identity. +// The notes front door's shapes: everything session findings have, minus +// quiz attribution — raw notes have no quiz tool parts, so there is +// nothing to attribute (and no event to build: outcomes inside the text +// surface only through understanding observations, user-decided). +const notesFindingsSchema = findingsSchema.omit({ attribution: true }); +const notesAuditSchema = auditSchema.omit({ attribution: true }); +type NotesFindings = Omit; +type NotesAuditResult = Omit; -FAITHFULNESS — verify every candidate against the transcript: +const AUDITOR_INSTRUCTIONS = `You are the audit stage of a learning application's extraction pipeline. You receive +ONE source — a session transcript, or the author's personal notes in an ARBITRARY format (the notes front door) — a +set of proposed findings (propositions, edges, and quiz attribution where the source has quiz lines), and the +CURRENT knowledge graph. Your jobs: the faithfulness net, and identity. + +FAITHFULNESS — verify every candidate against the source text: - PROPOSITIONS: every evidence quote must appear (near-)verbatim in the transcript view, and the statement must be a faithful reading of it. The node theory applies: a statement must be a TRUE, self-contained, world-facing fact — reword learner- or session-relative statements ("the learner understands X" → "X"), state @@ -293,8 +404,10 @@ FAITHFULNESS — verify every candidate against the transcript: - EDGES: the evidence must actually articulate the relation, in the direction claimed. Drop direction-swapped or speculative edges. - 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. -- UNDERSTANDING levels: verify against the quiz outcomes and the session's evidence; each note must be + attributions to null, or re-point them to the proposition the question actually tests. (Session + sources only — notes findings carry no attribution; there is nothing to verify.) +- UNDERSTANDING levels: verify against the quiz outcomes and the session's evidence — or, for notes, + the author's own words; each note must be exactly one word from the fixed scale ("solid", "shaky", "accepted", "resisted", "not yet engaged") — reword or drop anything else. No justifications. @@ -324,47 +437,47 @@ function graphView(nodes: GraphNode[]): string { return lines.join("\n"); } -/** One auditor run — same single-submit shape as the extractor. */ -async function runAuditor( +/** The auditor over session findings (the chat front door). */ +function runAuditor( engine: PiEngine | undefined, prompt: string, signal: AbortSignal | undefined, onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void, ): Promise { - let submitted: AuditResult | undefined; - const result = await runSubagent( - toPiTools({ - submit_review: lhTool({ - name: "submit_review", - description: "Submit the verified findings subset (with identity verdicts) + the dropped list.", - schema: auditSchema, - execute: async (review) => { - submitted = review; - return review; - }, - }), - }), - { - role: "auditor", - engine, - systemPrompt: AUDITOR_INSTRUCTIONS, - prompt, - maxTurns: 4, - abortSignal: signal, - onAssistantMessage: (message, step) => - onStep( - message.usage, - message.model, - step, - message.stopReason, - message.content.filter((c) => c.type === "toolCall").map((c) => (c as { name: string }).name).join(", ") || "(none)", - ), - shouldStop: (_message, toolResults) => toolResults.some((r) => r.toolName === "submit_review" && !r.isError), - }, - ); - void result; - if (!submitted) throw new Error("the auditor never submitted a review"); - return submitted; + return runSubmitAgent({ + role: "auditor", + toolName: "submit_review", + toolDescription: "Submit the verified findings subset (with identity verdicts) + the dropped list.", + schema: auditSchema, + systemPrompt: AUDITOR_INSTRUCTIONS, + prompt, + engine, + signal, + onStep, + neverSubmitted: "the auditor never submitted a review", + }); +} + +/** The auditor over notes findings (the format-agnostic front door): same + * faithfulness + identity job, notes-mode schema (no attribution). */ +function runNotesAuditor( + engine: PiEngine | undefined, + prompt: string, + signal: AbortSignal | undefined, + onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void, +): Promise { + return runSubmitAgent({ + role: "auditor", + toolName: "submit_review", + toolDescription: "Submit the verified findings subset (with identity verdicts) + the dropped list.", + schema: notesAuditSchema, + systemPrompt: AUDITOR_INSTRUCTIONS, + prompt, + engine, + signal, + onStep, + neverSubmitted: "the auditor never submitted a review", + }); } // ─── the pipeline status registry ───────────────────────────────────────────── @@ -376,6 +489,13 @@ export type ExtractionPhase = "extracting" | "auditing" | "merging"; * Exported for the status payload's consumer typing (the dock). */ export interface RunningExtraction { chatId: string; + /** Which front door the run came through — label + usage-persistence + * depend on it (chats bill their chat record; notes bill the source + * record). */ + kind: "chat" | "notes"; + /** Display label (the file path for notes imports); the status route + * falls back to the chat store for chat runs. */ + label?: string; phase: ExtractionPhase; startedAt: number; /** Live per-stage usage, accumulated as steps bill (the dock's cost / @@ -388,6 +508,10 @@ export interface RunningExtraction { const runningExtractions = new Map(); +/** Last minted startedAt — kept monotonic so two runs starting in the same + * millisecond never collide on the usage idempotency key. */ +let lastStartedAt = 0; + /** In-flight pipeline runs (oldest first) — the /api/graph/status payload. */ export function extractionStatuses(): RunningExtraction[] { return [...runningExtractions.values()].sort((a, b) => a.startedAt - b.startedAt); @@ -415,22 +539,38 @@ export interface ExtractionModels { export interface ExtractionPipelineDeps extends GraphDeps, ExtractionModels {} +export interface NotesImportInput { + /** The source's file path — its identity (deterministic src id) AND its + * provenance label in the detail card. */ + filePath: string; + /** The file's raw content, VERBATIM — any format. */ + text: string; +} + 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; + /** The format-agnostic front door (2026-09-13): run the same pipeline + * over ONE raw notes file. The source id is deterministic per file + * path, so re-importing the same file re-runs extraction against the + * same source identity (mappings + events dedupe by it). */ + importNotes(input: NotesImportInput): Promise; } export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline { const { chatStore, graphStore } = deps; const inFlight = new Set(); - function setPhase(chatId: string, phase: ExtractionPhase): void { + function setPhase(chatId: string, phase: ExtractionPhase, kind: "chat" | "notes", label?: string): void { const prev = runningExtractions.get(chatId); + if (!prev) lastStartedAt = Math.max(Date.now(), lastStartedAt + 1); runningExtractions.set(chatId, { chatId, + kind, + label: prev?.label ?? label, // phase changes never lose the label phase, - startedAt: prev?.startedAt ?? Date.now(), + startedAt: prev?.startedAt ?? lastStartedAt, usage: prev?.usage ?? {}, // phase changes never lose usage controller: prev?.controller ?? new AbortController(), }); @@ -525,7 +665,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti function persistRunUsage(chatId: string, durationMs: number, stats: EmissionStats | undefined): void { const run = runningExtractions.get(chatId); if (!run) return; - chatStore.appendExtractionUsage(chatId, { + const entry: ExtractionRunUsage = { startedAt: run.startedAt, at: new Date().toISOString(), durationMs, @@ -534,7 +674,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti extractor: run.usage.extractor ?? { model: "unknown", steps: [] }, auditor: run.usage.auditor ?? { model: "unknown", steps: [] }, }, - }); + }; + if (run.kind === "notes") graphStore.appendSourceUsage(chatId, entry); + else chatStore.appendExtractionUsage(chatId, entry); } async function runExtraction(chatId: string): Promise { @@ -543,14 +685,19 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti const t0 = performance.now(); try { console.info(`[extraction ${chatId}] pipeline start`); - setPhase(chatId, "extracting"); + setPhase(chatId, "extracting", "chat"); const findings = await extract(chatId); - setPhase(chatId, "auditing"); + setPhase(chatId, "auditing", "chat"); const auditResult = await audit(chatId, findings); - setPhase(chatId, "merging"); - // The merger re-reads the chat record itself — events are rebuilt - // from the transcript, never from stale copies. - const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }); + setPhase(chatId, "merging", "chat"); + // The merger merges from the CURRENT record — events are rebuilt + // from the live transcript, never from stale copies. + const record = chatStore.getChat(chatId); + if (!record) throw new Error(`Chat not found: ${chatId}`); + const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }, { + messages: record.messages as UIMessage[], + foldedAt: record.updatedAt, + }); persistRunUsage(chatId, Math.round(performance.now() - t0), stats); console.info( `[extraction ${chatId}] merged: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events ` + @@ -577,6 +724,113 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti } } - return { runExtraction }; + // ── the notes front door (format-agnostic: raw text in, same graph out) ── + + /** The raw source view the notes stages see: the file path + the text + * VERBATIM. No message ids — provenance is the file itself. */ + function notesView(input: { filePath: string; text: string }): string { + return `SOURCE (the author's personal notes): ${input.filePath}\n\n${input.text}`; + } + + async function notesExtract(input: { id: string; filePath: string; text: string }): Promise { + const signal = runningExtractions.get(input.id)?.controller.signal; + signal?.throwIfAborted(); // an abort during a sync gap ends the run here + const view = notesView(input); + console.info(`[extraction ${input.id}] notes extractor starting (${view.split("\n").length} source lines)`); + const t0 = performance.now(); + const findings = await runNotesExtractor( + deps.extractor, + view, + signal, + (usage, model, step, finishReason, tools) => { + recordStep(input.id, "extractor", callUsage(usage), model); + console.info(`[extraction] extractor step ${step}: ${finishReason}, tools: ${tools}`); + }, + ); + console.info(`[extraction ${input.id}] notes extractor done in ${Math.round(performance.now() - t0)}ms`); + return findings; + } + + async function notesAudit(input: { id: string; filePath: string; text: string }, findings: NotesFindings): Promise { + const signal = runningExtractions.get(input.id)?.controller.signal; + signal?.throwIfAborted(); // an abort during the extract→audit gap ends here + const graph = getGraph(graphStore); + console.info( + `[extraction ${input.id}] notes auditor starting (${findings.propositions.length} propositions, ` + + `${findings.edges.length} edges, ${graph.nodes.length} graph nodes)`, + ); + const t0 = performance.now(); + const auditResult = await runNotesAuditor( + deps.auditor, + `Source:\n\n${notesView(input)}\n\nProposed findings:\n\n${JSON.stringify(findings, null, 2)}\n\n` + + `Current knowledge graph (${graph.nodes.length} nodes):\n\n${graphView(graph.nodes)}`, + signal, + (usage, model, step, finishReason, tools) => { + recordStep(input.id, "auditor", callUsage(usage), model); + console.info(`[extraction] auditor step ${step}: ${finishReason}, tools: ${tools}`); + }, + ); + if (auditResult.dropped.length > 0) { + console.info( + `[extraction ${input.id}] audit dropped ${auditResult.dropped.length}: ` + + auditResult.dropped.map((d) => `${d.ref} (${d.reason})`).join("; "), + ); + } + console.info( + `[extraction ${input.id}] audit kept ${auditResult.propositions.length}/${findings.propositions.length} propositions, ` + + `${auditResult.propositions.filter((p) => p.linkTo !== undefined).length} linked to existing nodes`, + ); + console.info(`[extraction ${input.id}] notes auditor done in ${Math.round(performance.now() - t0)}ms`); + return auditResult; + } + + async function runNotesExtraction(input: { id: string; filePath: string; text: string; importedAt: string }): Promise { + const { id } = input; + if (inFlight.has(id)) return { nodes: 0, edges: 0, events: 0 }; + inFlight.add(id); + const t0 = performance.now(); + try { + console.info(`[extraction ${id}] notes pipeline start (${input.filePath})`); + setPhase(id, "extracting", "notes", input.filePath); + const findings = await notesExtract(input); + setPhase(id, "auditing", "notes", input.filePath); + const notesResult = await notesAudit(input, findings); + setPhase(id, "merging", "notes", input.filePath); + // Normalize to the merger's findings shape: notes carry no quiz + // attribution, and the merger scans no quiz parts for notes + // (messages: null) — nothing to attribute, nothing to build. + const auditResult: AuditResult = { ...notesResult, attribution: { quiz: {} } }; + const stats = mergeFindings(id, auditResult, { chatStore, graphStore }, { messages: null, foldedAt: input.importedAt }); + persistRunUsage(id, Math.round(performance.now() - t0), stats); + console.info( + `[extraction ${id}] merged: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events ` + + `(${Math.round(performance.now() - t0)}ms total)`, + ); + return stats; + } catch (err) { + persistRunUsage(id, Math.round(performance.now() - t0), undefined); + if (err instanceof Error && err.name === "AbortError") { + console.info(`[extraction ${id}] aborted (${Math.round(performance.now() - t0)}ms) — nothing merged, nothing checkpointed`); + return { nodes: 0, edges: 0, events: 0 }; + } + console.error(`[extraction ${id}] notes pipeline failed (${Math.round(performance.now() - t0)}ms):`, err); + throw err; + } finally { + inFlight.delete(id); + runningExtractions.delete(id); // done or failed — no longer running + } + } + + /** Import ONE raw notes file: register the source (deterministic id per + * file path — re-import keeps the same identity and its usage history) + * and run the pipeline over it. */ + async function importNotes(input: NotesImportInput): Promise { + const id = `src_${crypto.createHash("sha256").update(input.filePath).digest("hex").slice(0, 12)}`; + const importedAt = new Date().toISOString(); + graphStore.upsertSource({ id, filePath: input.filePath, importedAt }); + return runNotesExtraction({ id, ...input, importedAt }); + } + + return { runExtraction, importNotes }; } diff --git a/src/core/contracts/graph.ts b/src/core/contracts/graph.ts index 24ea899..2941541 100644 --- a/src/core/contracts/graph.ts +++ b/src/core/contracts/graph.ts @@ -20,6 +20,8 @@ /** Two relations only (DECIDED 2026-09-02): follows = b is derivable from a; * supports = a grounds b without entailing it. No scalars, no encoded * strength — strength is derived at read time from topology. */ +import type { ExtractionRunUsage } from "./chat.ts"; + export type EdgeType = "follows" | "supports"; /** What one extraction run merged into the graph ledgers — the merger's @@ -37,10 +39,32 @@ export interface EmissionStats { * edges record every (chat, message) where they came up, deep-linkable to * the exact transcript point. */ export interface ProvenancePoint { + /** The chat this point anchors to — or a notes source id ("src_…"): + * imports carry no message anchors (provenance is the file itself, + * recorded through the mapping line + the source record). */ chatId: string; messageId: string; } +/** One imported notes file — the format-agnostic front door's bookkeeping + * (DECIDED 2026-09-13): personal notes in ARBITRARY formats run through the + * same extractor → auditor → merger pipeline, with the file itself as the + * provenance anchor. There is no chat: the detail card shows the file path + * as a STATIC entry (not a link), and understanding observations still + * apply (judged from the author's own voice in the notes). The run's usage + * log lives here, mirroring ChatRecord.extractions. Ids are "src_" + hash + * of the file path — deterministic, so re-importing the same file re-runs + * extraction against the same source id (mappings/events dedupe by it). */ +export interface NotesSource { + id: string; + filePath: string; + /** ISO — when this file was imported (the checkpoint's foldedAt). */ + importedAt: string; + /** The run's usage log — same append-only discipline as + * ChatRecord.extractions, idempotent per startedAt. */ + extractions?: ExtractionRunUsage[]; +} + export interface GraphNode { /** Opaque stable id, store-assigned: "n_" + nanoid-like. */ id: string; @@ -107,6 +131,8 @@ export interface LedgerEvent { /** null = not yet attributed. */ nodeId: string | null; kind: LedgerEventKind; + /** The chat that produced this evidence — or a notes source id + * ("src_…") for imported-notes observations. */ chatId: string; at: string; /** Idempotency key: the quiz part's toolCallId, or `obs:` for @@ -249,6 +275,10 @@ export interface GraphSnapshot { edges: GraphEdge[]; /** chatId → title, for every chat that touched the graph. */ sessions: Record; + /** srcId → file path, for every IMPORTED NOTES file that touched the + * graph (the format-agnostic front door): a static provenance entry — + * the detail card shows the path as plain text, not a link. */ + fileSources: Record; /** nodeId → chatIds that touched the node (click-through provenance). */ nodeSessions: Record; /** nodeId → chatId → that session's understanding observation (the diff --git a/src/core/graph.ts b/src/core/graph.ts index c24bd3a..dee329b 100644 --- a/src/core/graph.ts +++ b/src/core/graph.ts @@ -64,6 +64,17 @@ function newGraphNodeId(): string { return `n_${randomBytes(6).toString("hex")}`; } +/** What one merge run merges FROM (the caller supplies it — the merger never + * reads the chat store): a session's transcript + its `updatedAt` + * checkpoint, or — for the format-agnostic notes front door (2026-09-13) — + * `messages: null`: raw notes carry no tool parts, so no quiz events are + * rebuilt (quiz outcomes inside notes surface only through the extractor's + * understanding observations), and the checkpoint stamps the import. */ +export interface MergeSource { + messages: UIMessage[] | null; + foldedAt: string; +} + // ─── transcript scanning (shared by the merger + trigger) ───────────────────── /** A flattened, order-preserving view of every tool part in a transcript. */ @@ -126,11 +137,15 @@ const samePoint = (a: ProvenancePoint, b: ProvenancePoint): boolean => * parts, nodeId from the extractor's attribution through this run's * mapping) plus one observation event per proposition carrying an * understanding note (ref `obs:`). No taught events, ever. + * + * `chatId` is the run's source key: a real chat, or a notes source id + * ("src_…") for the format-agnostic import path — same ledgers, same + * idempotency, one graph. The `source` parameter carries the transcript + * being merged (null messages = raw notes: no quiz events) and the + * checkpoint stamp; the merger itself is chat-store-free. */ -export function mergeFindings(chatId: string, audit: AuditResult, deps: GraphDeps): EmissionStats { - const { graphStore: ledgers, chatStore } = deps; - const record = chatStore.getChat(chatId); - if (!record) return { nodes: 0, edges: 0, events: 0 }; +export function mergeFindings(chatId: string, audit: AuditResult, deps: GraphDeps, source: MergeSource): EmissionStats { + const { graphStore: ledgers } = deps; const stats: EmissionStats = { nodes: 0, edges: 0, events: 0 }; // 1. Resolution: this run's propositions → graph node ids. Existing @@ -214,7 +229,7 @@ export function mergeFindings(chatId: string, audit: AuditResult, deps: GraphDep for (const event of ledgers.readEvents()) { if (event.chatId === chatId && event.ref) seenRefs.add(event.ref); } - for (const part of toolParts(record.messages)) { + for (const part of source.messages ? toolParts(source.messages) : []) { if (part.name !== "quiz" || part.state !== "output-available") continue; const event = quizEvent(part, chatId, audit, mapping, now); if (event?.ref !== undefined && !seenRefs.has(event.ref)) { @@ -241,8 +256,9 @@ export function mergeFindings(chatId: string, audit: AuditResult, deps: GraphDep stats.events++; } - // 4. Checkpoint: this chat is extracted through its current updatedAt. - ledgers.appendCheckpoint({ chatId, foldedAt: record.updatedAt }); + // 4. Checkpoint: this source is extracted through its foldedAt (the + // chat's updatedAt, or the import time for notes). + ledgers.appendCheckpoint({ chatId, foldedAt: source.foldedAt }); return stats; } @@ -423,20 +439,28 @@ export function graphSnapshot(deps: GraphDeps): GraphSnapshot { if (event.nodeId !== null) touch(event.nodeId, event.chatId); } - // Titles for every chat the graph can point back at. - const chatIds = new Set(); + // Labels for every source the graph can point back at: real chats + // (title, deep-linkable) and imported notes files (path, static — the + // format-agnostic front door's provenance). + const sourceIds = new Set(); for (const chats of nodeChats.values()) { - for (const id of chats) chatIds.add(id); + for (const id of chats) sourceIds.add(id); } const sessions: Record = {}; - for (const id of chatIds) { + const fileSources: Record = {}; + for (const id of sourceIds) { const chat = deps.chatStore.getChat(id); - if (chat) sessions[id] = chat.title; + if (chat) { + sessions[id] = chat.title; + continue; + } + const source = deps.graphStore.readSources().find((s) => s.id === id); + if (source) fileSources[id] = source.filePath; } const nodeSessions: Record = {}; for (const [nodeId, chats] of nodeChats) { - const ids = [...chats].filter((id) => sessions[id] !== undefined); + const ids = [...chats].filter((id) => sessions[id] !== undefined || fileSources[id] !== undefined); if (ids.length > 0) nodeSessions[nodeId] = ids; } @@ -461,7 +485,7 @@ export function graphSnapshot(deps: GraphDeps): GraphSnapshot { (nodeObservations[nodeId] ??= {})[chatId] = note; } - return { nodes, edges, sessions, nodeSessions, nodeObservations }; + return { nodes, edges, sessions, fileSources, nodeSessions, nodeObservations }; } // Re-export: the contract owns the shape; legacy importers still reach it. diff --git a/src/core/ports.ts b/src/core/ports.ts index 6343a2e..e1e6219 100644 --- a/src/core/ports.ts +++ b/src/core/ports.ts @@ -24,6 +24,7 @@ import type { PlanTree, SessionCheckpoint, TurnUsage, + NotesSource, } from "./contracts.ts"; // ─── ChatStore — the session record's home (data/chats) ───────────────────── @@ -80,6 +81,15 @@ export interface GraphStore { /** Append one emission checkpoint. */ appendCheckpoint(checkpoint: SessionCheckpoint): void; readCheckpoints(): SessionCheckpoint[]; + + /** Upsert one imported-notes source record (config.SOURCES_FILE — a + * JSON map, not a ledger; an upsert keeps the existing usage log). */ + upsertSource(source: NotesSource): void; + /** Every notes source record, by id order-insensitive. */ + readSources(): NotesSource[]; + /** Append one run's usage to a source record — idempotent per + * startedAt, mirroring ChatRecord.extractions. */ + appendSourceUsage(id: string, entry: ExtractionRunUsage): void; } // ─── JobStore — background sub-agent jobs (data/jobs) ───────────────────────── diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index a4a3436..589516f 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -37,7 +37,10 @@ const jobRunner = createJobRunner({ jobStore: stores.jobStore, deliver: deliverJobResult, }); -const extraction = createExtractionPipeline(stores); +/** The app's ONE extraction pipeline instance — both front doors share it: + * quiz-triggered session runs (turn-service) and notes imports + * (/api/graph/import) hit the same single-flight registry. */ +export const extraction = createExtractionPipeline(stores); // Wake-pipeline wiring (contracts.ts "THE ASYNC PROTOCOL" §4): the pipeline // fires its server-side teacher turns through THIS composition root, and diff --git a/src/store/graph.ts b/src/store/graph.ts index ac86c45..b67c884 100644 --- a/src/store/graph.ts +++ b/src/store/graph.ts @@ -15,6 +15,10 @@ * (chatId, sourceId) → graphNodeId * sessions.jsonl SessionCheckpoint— emission checkpoints (foldedAt) * + * Plus config.SOURCES_FILE (sources.json): the imported-notes bookkeeping + * (NOT a ledger) — one JSON map, src-id → NotesSource, read-modify-write + * like a chat file. The format-agnostic front door (2026-09-13). + * * Implements core/ports.ts GraphStore. data/graph/* is exclusively this * adapter's. Reads are strict against the contracts.ts shapes (the 2026-09-02 * graph redesign): corrupt lines are skipped, never guessed at — the ledgers @@ -25,12 +29,14 @@ import * as fs from "node:fs"; import * as path from "node:path"; -import { LEDGER_FILES } from "../../config.ts"; +import { LEDGER_FILES, SOURCES_FILE } from "../../config.ts"; import type { GraphEdge, GraphNode, + ExtractionRunUsage, LedgerEvent, LedgerEventKind, + NotesSource, NodeMapping, ProvenancePoint, SessionCheckpoint, @@ -157,4 +163,57 @@ export const graphStore: GraphStore = { readCheckpoints(): SessionCheckpoint[] { return readValid(LEDGER_FILES.sessions, isCheckpoint); }, + + // ── imported-notes bookkeeping (config.SOURCES_FILE — a JSON map) ── + + /** Upsert: keeps the existing record's usage log when re-importing. */ + upsertSource(source: NotesSource): void { + const all = readSourceMap(); + const prev = all[source.id]; + all[source.id] = prev ? { ...source, extractions: prev.extractions } : source; + writeSourceMap(all); + }, + readSources(): NotesSource[] { + return Object.values(readSourceMap()); + }, + /** Idempotent per startedAt — same discipline as chats.appendExtractionUsage. */ + appendSourceUsage(id: string, entry: ExtractionRunUsage): void { + const all = readSourceMap(); + const source = all[id]; + if (!source) return; // no such source — nothing to bill against + source.extractions = source.extractions?.some((r) => r.startedAt === entry.startedAt) + ? source.extractions + : [...(source.extractions ?? []), entry]; + writeSourceMap(all); + }, }; + +/** sources.json — { id: NotesSource }. Read-modify-write, NOT append-only: + * the usage log lives inside the record (like a chat's JSON file). */ +function readSourceMap(): Record { + let raw: string; + try { + raw = fs.readFileSync(SOURCES_FILE, "utf-8"); + } catch { + return {}; + } + try { + const parsed: unknown = JSON.parse(raw); + if (!isRecord(parsed)) return {}; + const out: Record = {}; + for (const [id, value] of Object.entries(parsed)) { + if (isNotesSource(value)) out[id] = value; + } + return out; + } catch { + return {}; + } +} + +function writeSourceMap(map: Record): void { + fs.mkdirSync(path.dirname(SOURCES_FILE), { recursive: true }); + fs.writeFileSync(SOURCES_FILE, JSON.stringify(map, null, "\t"), "utf-8"); +} + +const isNotesSource = (v: unknown): v is NotesSource => + isRecord(v) && isString(v.id) && isString(v.filePath) && isString(v.importedAt); diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index 3b58964..18a0488 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -27,10 +27,12 @@ import type { UIMessage } from "ai"; import type { PiEngine } from "../src/agent/models.ts"; import type { AuditResult, + ExtractionRunUsage, GraphEdge, GraphNode, LedgerEvent, NodeMapping, + NotesSource, SessionCheckpoint, SessionFindings, } from "../src/core/contracts.ts"; @@ -60,6 +62,7 @@ function memoryLedgers() { const events: LedgerEvent[] = []; const mappings: NodeMapping[] = []; const checkpoints: SessionCheckpoint[] = []; + const sources = new Map(); return { appendNode: (n: GraphNode) => void nodes.push(n), readNodes: () => nodes.map((n) => ({ ...n })), @@ -71,6 +74,19 @@ function memoryLedgers() { readMappings: () => mappings.map((m) => ({ ...m })), appendCheckpoint: (c: SessionCheckpoint) => void checkpoints.push(c), readCheckpoints: () => checkpoints.map((c) => ({ ...c })), + upsertSource: (s: NotesSource) => { + // Mirrors the real store: an upsert KEEPS the existing usage log. + const prev = sources.get(s.id); + sources.set(s.id, { ...s, extractions: [...(prev?.extractions ?? [])] }); + }, + readSources: () => [...sources.values()].map((s) => ({ ...s, extractions: [...(s.extractions ?? [])] })), + appendSourceUsage: (id: string, entry: ExtractionRunUsage) => { + const source = sources.get(id); + if (!source) return; + if (!source.extractions?.some((r) => r.startedAt === entry.startedAt)) { + source.extractions = [...(source.extractions ?? []), entry]; + } + }, } satisfies GraphStore; } @@ -513,3 +529,123 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { await assert.rejects(pipeline.runExtraction("chat_does_not_exist"), /Chat not found/); }); }); + +// ── the notes front door (format-agnostic import) ──────────────────────────── + +/** Raw personal notes — arbitrary format, no session, no message ids. */ +const NOTES_TEXT = `> [!quote] YOU +> teach me vim + +> [!abstract] PI +> Vim always starts in Normal mode — keys are commands, not text.`; + +/** What the notes-mode extractor submits: same shapes minus attribution, + * and messages always empty (the file is the provenance). */ +const NOTES_FINDINGS = { + propositions: [ + { + id: "p1", + name: "Vim normal mode", + statement: "Vim starts in Normal mode, where keys are commands.", + aliases: [], + messages: [], + evidence: ["Vim always starts in Normal mode — keys are commands, not text."], + understanding: "solid", + }, + ], + edges: [], +} as const; + +const NOTES_VERDICT = { + propositions: [NOTES_FINDINGS.propositions[0]], + edges: [], + dropped: [], +} as const; + +describe("the notes front door (importNotes — format-agnostic)", () => { + it("imports raw notes end to end: source record, file-scoped provenance, no quiz events", async () => { + const ledgers = memoryLedgers(); + const seen: { kind?: string; label?: string }[] = []; + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", NOTES_FINDINGS), + // During the auditor's call the registry must show the notes run. + auditor: submitModel("submit_review", NOTES_VERDICT, () => { + const run = extractionStatuses().find((r) => r.chatId.startsWith("src_")); + seen.push({ kind: run?.kind, label: run?.label }); + }), + }); + const stats = await pipeline.importNotes({ filePath: "/vault/learn vim.md", text: NOTES_TEXT }); + + assert.deepEqual(stats, { nodes: 1, edges: 0, events: 1 }); + // The run surfaced as a NOTES run labeled with the file path. + assert.deepEqual(seen, [{ kind: "notes", label: "/vault/learn vim.md" }]); + assert.equal(extractionStatuses().length, 0, "the registry clears when the run ends"); + + // The source record: deterministic src id, the file path, the usage log. + const source = ledgers.readSources()[0]!; + assert.match(source.id, /^src_/); + assert.equal(source.filePath, "/vault/learn vim.md"); + assert.ok(Date.parse(source.importedAt)); + assert.equal(source.extractions?.length, 1, "the run's usage persisted into the source record"); + assert.equal(source.extractions![0]!.stats?.nodes, 1); + assert.ok(source.extractions![0]!.stages.extractor.steps.length > 0, "faux usage captured"); + + // File-scoped provenance: the mapping line carries the source id; + // the node carries NO message anchors. + const node = ledgers.readNodes()[0]!; + assert.equal(node.statement, "Vim starts in Normal mode, where keys are commands."); + assert.deepEqual(node.provenance, []); + assert.deepEqual(ledgers.readMappings(), [{ chatId: source.id, sourceId: "p1", graphNodeId: node.id }]); + + // One observation event, zero quiz events — outcomes inside notes + // surface only through understanding observations (user-decided). + const events = ledgers.readEvents(); + assert.deepEqual( + events.map((e) => [e.kind, e.chatId, e.ref, e.note]), + [["observation", source.id, "obs:p1", "solid"]], + ); + // The checkpoint stamps the import. + assert.equal(ledgers.readCheckpoints()[0]!.chatId, source.id); + }); + + it("re-importing the same file is idempotent: same source identity, refs dedupe, usage history grows", async () => { + const ledgers = memoryLedgers(); + const mk = () => + createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", NOTES_FINDINGS), + auditor: submitModel("submit_review", NOTES_VERDICT), + }); + const first = await mk().importNotes({ filePath: "/vault/learn vim.md", text: NOTES_TEXT }); + assert.deepEqual(first, { nodes: 1, edges: 0, events: 1 }); + + const second = await mk().importNotes({ filePath: "/vault/learn vim.md", text: NOTES_TEXT }); + assert.deepEqual(second, { nodes: 0, edges: 0, events: 0 }, "the same source id dedupes everything"); + + assert.equal(ledgers.readSources().length, 1, "one source identity — the path hashes to the same id"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2, "both runs billed their usage"); + assert.equal(ledgers.readNodes().length, 1, "no duplicate node"); + }); + + it("aborts like a chat run: true cancellation, nothing merged, partial usage in the source record", async () => { + const ledgers = memoryLedgers(); + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: hangingModel(), + auditor: submitModel("submit_review", NOTES_VERDICT), + }); + const runPromise = pipeline.importNotes({ filePath: "/vault/learn vim.md", text: NOTES_TEXT }); + + const srcId = extractionStatuses().find((r) => r.chatId.startsWith("src_"))!.chatId; + assert.equal(abortExtraction(srcId), true); + assert.deepEqual(await runPromise, { nodes: 0, edges: 0, events: 0 }); + assert.equal(ledgers.readNodes().length, 0, "nothing merged"); + assert.equal(ledgers.readCheckpoints().length, 0, "nothing checkpointed"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 1, "partial usage still persisted"); + assert.equal(ledgers.readSources()[0]!.extractions![0]!.stats, undefined, "never merged → no stats"); + }); +}); diff --git a/tests/graph.test.ts b/tests/graph.test.ts index 2a104d1..9d27d11 100644 --- a/tests/graph.test.ts +++ b/tests/graph.test.ts @@ -23,14 +23,22 @@ import type { QuestionAnswerData } from "../src/core/question-core.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"; +import { LEDGER_FILES, PATHS, SOURCES_FILE } from "../config.ts"; /** The wired fs adapters, exactly what src/server/stores.ts passes in production. */ const graphDeps = { chatStore, graphStore } as const satisfies GraphDeps; +/** The session merge source for a chat — what the pipeline passes: the live + * transcript + its updatedAt checkpoint stamp. (Notes imports pass + * messages: null — see the notes-merger test below.) */ +function sessionSource(chatId: string): { messages: UIMessage[]; foldedAt: string } { + const record = chatStore.getChat(chatId)!; + return { messages: record.messages as UIMessage[], foldedAt: record.updatedAt }; +} + // ─── ledger isolation (backup-and-clear of data/graph/*.jsonl) ────────────── -const LEDGER_PATHS = Object.values(LEDGER_FILES); +const LEDGER_PATHS = [...Object.values(LEDGER_FILES), SOURCES_FILE]; const backups: Array<{ file: string; bak: string }> = []; const createdChats: string[] = []; @@ -233,7 +241,7 @@ function scopedGraph(chatIds: string[]) { describe("the merger (mergeFindings)", () => { it("creates nodes from the verdicts, rebuilds quiz events deterministically, records observations", () => { - const stats = mergeFindings(chat1.id, fixtureAudit(), graphDeps); + const stats = mergeFindings(chat1.id, fixtureAudit(), graphDeps, sessionSource(chat1.id)); assert.deepEqual(stats, { nodes: 3, edges: 1, events: 6 }); const graph = scopedGraph([chat1.id]); @@ -302,7 +310,7 @@ describe("the merger (mergeFindings)", () => { }); it("is idempotent — re-merging the same verdicts appends nothing", () => { - const stats = mergeFindings(chat1.id, fixtureAudit(), graphDeps); + const stats = mergeFindings(chat1.id, fixtureAudit(), graphDeps, sessionSource(chat1.id)); assert.deepEqual(stats, { nodes: 0, edges: 0, events: 0 }); const graph = scopedGraph([chat1.id]); assert.equal(graph.nodes.length, 3); @@ -346,6 +354,7 @@ describe("the merger (mergeFindings)", () => { dropped: [], }, graphDeps, + sessionSource(chat2.id), ); assert.deepEqual(stats, { nodes: 0, edges: 0, events: 1 }); @@ -363,6 +372,41 @@ describe("the merger (mergeFindings)", () => { assert.deepEqual(m, { chatId: chat2.id, sourceId: "q1", graphNodeId: alphaId }); }); + it("resolves imported-notes ids to their file paths in the snapshot (static provenance)", () => { + // The notes front door in miniature: a source record + a merge under + // its src id — the snapshot must label it with the file path (a + // static entry), never as a chat session. + graphStore.upsertSource({ id: "src_snapfixture", filePath: "/vault/learn vim.md", importedAt: "2026-09-13T00:00:00.000Z" }); + mergeFindings( + "src_snapfixture", + { + propositions: [ + { + id: "p1", + name: "Vim registers", + statement: "Vim registers map single keys to editing commands.", + aliases: [], + messages: [], + evidence: ["fixture evidence"], + understanding: "solid", + }, + ], + edges: [], + attribution: { quiz: {} }, + dropped: [], + }, + graphDeps, + { messages: null, foldedAt: "2026-09-13T00:00:00.000Z" }, + ); + const snap = graphSnapshot(graphDeps); + const node = snap.nodes.find((n) => n.statement === "Vim registers map single keys to editing commands.")!; + assert.equal(snap.fileSources["src_snapfixture"], "/vault/learn vim.md"); + assert.deepEqual(snap.nodeSessions[node.id], ["src_snapfixture"], "the file entry joins the node's sessions list"); + assert.equal(snap.sessions["src_snapfixture"], undefined, "a notes source is never a chat session"); + // The understanding observation joins under the source id like any session's. + assert.equal(snap.nodeObservations[node.id]!["src_snapfixture"], "solid"); + }); + it("derives sessions and per-session understanding observations in graphSnapshot", () => { const snap = graphSnapshot(graphDeps); const idByStatement = new Map(snap.nodes.map((n) => [n.statement, n.id])); @@ -408,17 +452,54 @@ describe("the merger (mergeFindings)", () => { dropped: [], }, graphDeps, + sessionSource(chat3.id), ); assert.equal(stats.nodes, 1, "created — the merger does not match strings"); assert.equal(getGraph(graphStore).nodes.filter((n) => n.statement === "Alpha intuition").length, 2); }); - it("returns zeros for unknown chats", () => { - assert.deepEqual(mergeFindings("chat_does_not_exist", fixtureAudit(), graphDeps), { - nodes: 0, - edges: 0, - events: 0, + it("merges the notes front door: source id as chatId, no quiz events, checkpoint at import time", () => { + // Raw personal notes carry no transcript — the merger gets no + // message parts, so NO quiz events are rebuilt; the understanding + // observation still lands, keyed to the source id. + const stats = mergeFindings( + "src_notesfixture", + { + propositions: [ + { + id: "p1", + name: "Vim normal mode", + statement: "Vim starts in Normal mode, where keys are commands.", + aliases: [], + messages: [], // notes carry no message anchors — the file is the provenance + evidence: ["fixture evidence"], + understanding: "solid", + }, + ], + edges: [], + attribution: { quiz: {} }, // normalized by the notes pipeline — nothing to attribute + dropped: [], + }, + graphDeps, + { messages: null, foldedAt: "2026-09-13T00:00:00.000Z" }, + ); + assert.deepEqual(stats, { nodes: 1, edges: 0, events: 1 }); + + const node = getGraph(graphStore).nodes.find((n) => n.statement.startsWith("Vim starts"))!; + assert.deepEqual(node.provenance, [], "no message anchors — the mapping line carries the source"); + assert.deepEqual(graphStore.readMappings().find((m) => m.chatId === "src_notesfixture"), { + chatId: "src_notesfixture", + sourceId: "p1", + graphNodeId: node.id, }); + // The observation event is file-scoped; no quiz events exist. + const events = graphStore.readEvents().filter((e) => e.chatId === "src_notesfixture"); + assert.deepEqual( + events.map((e) => [e.kind, e.ref, e.note]), + [["observation", "obs:p1", "solid"]], + ); + // The checkpoint stamps the import time. + assert.equal(graphStore.readCheckpoints().find((c) => c.chatId === "src_notesfixture")?.foldedAt, "2026-09-13T00:00:00.000Z"); }); }); @@ -452,7 +533,7 @@ describe("the trigger helper (pendingQuizAnswers)", () => { assert.equal(pendingQuizAnswers(chat.id, graphDeps), 1, "one answered quiz pending"); assert.equal(pendingQuizAnswers("chat_does_not_exist", graphDeps), 0); - mergeFindings(chat.id, fixtureAudit(), graphDeps); + mergeFindings(chat.id, fixtureAudit(), graphDeps, sessionSource(chat.id)); // call-s is unattributed here (fixtureAudit has no call-s attribution) // — but it still becomes an event, so nothing stays pending. assert.equal(pendingQuizAnswers(chat.id, graphDeps), 0, "extracted → nothing pending"); diff --git a/tests/usage.test.ts b/tests/usage.test.ts index 1a2651d..11a5391 100644 --- a/tests/usage.test.ts +++ b/tests/usage.test.ts @@ -10,7 +10,8 @@ import test from "node:test"; import * as fs from "node:fs"; import { chatStore } from "../src/store/chats.ts"; -import { PATHS } from "../config.ts"; +import { graphStore } from "../src/store/graph.ts"; +import { PATHS, SOURCES_FILE } from "../config.ts"; import { callCost, contextWindow, jobCost, sessionCost, sumSteps, turnCost } from "../src/core/prices.ts"; import type { CallUsage, ExtractionRunUsage, JobRecord, TurnUsage } from "../src/core/contracts.ts"; @@ -170,3 +171,40 @@ test("appendExtractionUsage: appends, idempotent per startedAt, not in list payl fs.unlinkSync(`${PATHS.chats}/${record.id}.json`); }); + +test("appendSourceUsage: appends to the source record, idempotent per startedAt, upsert keeps history", () => { + // The real fs graphStore — sources.json backed up and restored around + // this test (same discipline as the chat fixture files). + fs.mkdirSync(PATHS.graph, { recursive: true }); + let bak: string | undefined; + if (fs.existsSync(SOURCES_FILE)) bak = `${SOURCES_FILE}.w5test.bak`; + if (bak) fs.renameSync(SOURCES_FILE, bak); + try { + const runEntry = (startedAt: number): ExtractionRunUsage => ({ + startedAt, + at: "2026-09-13T00:00:00Z", + durationMs: 42_000, + stats: { nodes: 1, edges: 2, events: 3 }, + stages: { + extractor: { model: "z-ai/glm-5.3-flash", steps: [call({ inputTokens: 100, outputTokens: 10 })] }, + auditor: { model: "z-ai/glm-5.3-flash", steps: [call({ inputTokens: 50, outputTokens: 5 })] }, + }, + }); + const id = "src_usagetest01"; + graphStore.upsertSource({ id, filePath: "/vault/usagetest.md", importedAt: "2026-09-13T00:00:00Z" }); + graphStore.appendSourceUsage(id, runEntry(1)); + graphStore.appendSourceUsage(id, runEntry(1)); // same run appended twice → no-op + graphStore.appendSourceUsage(id, runEntry(2)); // a retried run is a NEW spend + assert.equal(graphStore.readSources().find((s) => s.id === id)?.extractions?.length, 2); + + // An upsert (re-import) KEEPS the usage history. + graphStore.upsertSource({ id, filePath: "/vault/usagetest.md", importedAt: "2026-09-14T00:00:00Z" }); + assert.equal(graphStore.readSources().find((s) => s.id === id)?.extractions?.length, 2); + + // Unknown ids are a silent no-op (nothing to bill against). + graphStore.appendSourceUsage("src_nope", runEntry(1)); + } finally { + fs.rmSync(SOURCES_FILE, { force: true }); + if (bak) fs.renameSync(bak, SOURCES_FILE); + } +});