Notes front door: format-agnostic import into the graph pipeline

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.
This commit is contained in:
amosblomqvist
2026-09-13 02:08:55 +03:00
parent b769d32bc6
commit 543cf33789
16 changed files with 961 additions and 114 deletions
+329 -75
View File
@@ -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<SessionFindings> {
let submitted: SessionFindings | undefined;
async function runSubmitAgent<T>(opts: {
role: ModelRole;
toolName: string;
toolDescription: string;
schema: z.ZodType<T>;
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<T> {
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<SessionFindings> {
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<NotesFindings> {
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<SessionFindings, "attribution">;
type NotesAuditResult = Omit<AuditResult, "attribution">;
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<AuditResult> {
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<NotesAuditResult> {
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<string, RunningExtraction>();
/** 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<EmissionStats>;
/** 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<EmissionStats>;
}
export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline {
const { chatStore, graphStore } = deps;
const inFlight = new Set<string>();
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<EmissionStats> {
@@ -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<NotesFindings> {
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<NotesAuditResult> {
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<EmissionStats> {
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<EmissionStats> {
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 };
}
+30
View File
@@ -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:<sourceId>` for
@@ -249,6 +275,10 @@ export interface GraphSnapshot {
edges: GraphEdge[];
/** chatId → title, for every chat that touched the graph. */
sessions: Record<string, string>;
/** 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<string, string>;
/** nodeId → chatIds that touched the node (click-through provenance). */
nodeSessions: Record<string, string[]>;
/** nodeId → chatId → that session's understanding observation (the
+38 -14
View File
@@ -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:<sourceId>`). 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<string>();
// 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<string>();
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<string, string> = {};
for (const id of chatIds) {
const fileSources: Record<string, string> = {};
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<string, string[]> = {};
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.
+10
View File
@@ -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) ─────────────────────────
+4 -1
View File
@@ -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
+60 -1
View File
@@ -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<string, NotesSource> {
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<string, NotesSource> = {};
for (const [id, value] of Object.entries(parsed)) {
if (isNotesSource(value)) out[id] = value;
}
return out;
} catch {
return {};
}
}
function writeSourceMap(map: Record<string, NotesSource>): 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);