diff --git a/docs/architecture.md b/docs/architecture.md index 6bdc9f5..04e360e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -346,7 +346,7 @@ flowchart TB **Coverage, not quiz-counting** (supersedes the v1 every-N-quiz-answers trigger): each message's *view* (the compact line format `core/coverage.ts` has always hashed) is hashed into a per-chat fingerprint ledger (`coverage.jsonl`); "covered" means exactly "a run committed on this content." The prompt view is the richer markdown document — a deliberate 2026-09-27 split: re-formatting presentation never re-opens covered history. Uncovered = fingerprint diff, never dual-tracked; runs are greedy token-budgeted batches (`graph.extractionChunkTokens` — a settings knob, default 4096, floor 500, read fresh each turn) of the oldest uncovered messages — quiz-less sessions extract too. A coarse `sessions.jsonl` checkpoint still marks each merged run. The trigger seam (`turn-service.ts`) is trigger-agnostic from the pipeline's side: when a better mechanism lands (session end, idle, explicit flush), only the seam changes. Single-flight per source; the run registry powers the coverage UI and true abort. -**Two front doors, one pipeline**: chat transcripts (quiz-triggered) and raw personal notes (`POST /api/graph/import` — each file becomes a `NotesSource` playing the `chatId` role, deterministic id from the path hash). +**Two front doors, one pipeline**: chat transcripts (quiz-triggered) and raw personal notes (`POST /api/graph/import` — each file becomes a `NotesSource` playing the `chatId` role, deterministic id from the path hash). Notes chunk like chats (owner, 2026-09-28): large files split at line boundaries under the same token budget, each chunk a standalone extract → audit → merge under the serialized gate — no single model call carries a whole file. Both submit tools require their top-level fields (`edges`/`links` present even when empty): a payload missing them fails validation rather than silently merging structure-less nodes. **The read channel is pure derivation.** `GET /api/graph` reads the ledgers (`graphSnapshot`: merged nodes/edges, birth-session clusters, positions) — the home view never triggers a write. Node detail cards show a node's edges — outgoing under "What follows from this" / "What this supports", incoming under "Follows from" / "Supported by" (an edge A→B means A implies or supports B, owner semantics 2026-09-28) — and its birth sessions. diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 164567b..3208e96 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -142,12 +142,10 @@ const findingsProps = { propositions: Type.Array(Type.Object(propositionProps), { description: "Every distinct proposition the session touched — true statements, misconception-free.", }), - edges: Type.Optional( - Type.Array(edgeSchema, { - description: - "Relations between propositions: follows (derivable) / supports (grounds without entailing). Pass [] when there are none — never omit the field.", - }), - ), + edges: Type.Array(edgeSchema, { + description: + "Relations between propositions: follows (derivable) / supports (grounds without entailing). Pass [] when there are none — never omit the field.", + }), }; const findingsSchema = Type.Object(findingsProps); @@ -157,24 +155,20 @@ const findingsSchema = Type.Object(findingsProps); // edges are its entire deliverable. An order-of-magnitude smaller submit // than the old full-findings echo. const auditSchema = Type.Object({ - links: Type.Optional( - Type.Array( - Type.Object({ - id: Type.String({ description: 'The proposition id from the findings ("p1", …).' }), - linkTo: Type.String({ - description: - "The id of the EXISTING graph node this proposition IS — your identity verdict. EQUIVALENCE BIAS (2026-09-27): paraphrases, rewordings, and compressions of the same claim MERGE; only genuinely distinct propositions stay apart.", - }), + links: Type.Array( + Type.Object({ + id: Type.String({ description: 'The proposition id from the findings ("p1", …).' }), + linkTo: Type.String({ + description: + "The id of the EXISTING graph node this proposition IS — your identity verdict. EQUIVALENCE BIAS (2026-09-27): paraphrases, rewordings, and compressions of the same claim MERGE; only genuinely distinct propositions stay apart.", }), - { description: "Identity verdicts — which findings propositions ARE existing graph nodes. A proposition absent from here = create a new node." }, - ), - ), - edges: Type.Optional( - Type.Array(edgeSchema, { - description: - "BRIDGING edges ONLY — new edges welding this batch into the existing graph, with an EXISTING graph node id as one endpoint (or both) and/or a findings proposition id. NEVER restate the extractor's own edges.", }), + { description: "Identity verdicts — which findings propositions ARE existing graph nodes. A proposition absent from here = create a new node. Pass [] when none — never omit the field." }, ), + edges: Type.Array(edgeSchema, { + description: + "BRIDGING edges ONLY — new edges welding this batch into the existing graph, with an EXISTING graph node id as one endpoint (or both) and/or a findings proposition id. NEVER restate the extractor's own edges. Pass [] when none — never omit the field.", + }), }); // The notes front door's shapes are IDENTICAL to the session shapes now @@ -209,19 +203,26 @@ function healStringifiedArrays(raw: unknown): unknown { return out; } -// ── the weak-model ??-fills (the old zod defaults, applied at submit) ── +// ── the submit seam (owner, 2026-09-28) ── +// +// Both top-level fields on both submit tools are REQUIRED — a payload +// missing `edges`/`links` fails validation instead of silently merging +// edge-less (the degraded-retry hole: the model's first truncated +// attempt failed validation, then the retry passed with propositions +// only — optional edges ??-filled [] — merging a pile of nodes with no +// structure). Only the per-item aliases stay optional with a ??-fill. function fillFindings(raw: Static): SessionFindings { return { propositions: raw.propositions.map((p) => ({ ...p, aliases: p.aliases ?? [] })), - edges: raw.edges ?? [], + edges: raw.edges, }; } function fillAudit(raw: Static): AuditVerdicts { return { - links: raw.links ?? [], - edges: raw.edges ?? [], + links: raw.links, + edges: raw.edges, }; } @@ -231,6 +232,8 @@ report what those notes show the author engaging with, as structured findings fo There is no session and no chat structure here: the source is raw text, the author's own material. Read it as such. You are the ONLY stage that sees this source — you are fully trusted; report carefully. +You may be seeing ONE PART of a longer file (large files are chunked): extract exactly what THIS part holds — +the auditor links equivalent propositions across parts, so overlap worries are not yours. The graph is a set of TRUE propositions about the world, full stop. A node is never anything about the author or the source — no "the author understands X", no "mentioned in these notes": the node is just X. A misconception is NOT a @@ -954,11 +957,34 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti /** 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}`; + function notesView(input: { filePath: string; text: string; part?: { i: number; n: number } }): string { + const part = input.part && input.part.n > 1 ? ` (part ${input.part.i}/${input.part.n})` : ""; + return `SOURCE (the author's personal notes): ${input.filePath}${part}\n\n${input.text}`; } - async function notesExtract(input: { id: string; filePath: string; text: string }): Promise { + /** Chunk the notes text under the same budget the chat path uses + * (owner, 2026-09-28: the whole-file-in-one-call design collapsed a + * weak model — 26 propositions, structure bleed mid-emission). + * Greedy by line, chars/4 like every other estimate in the app; a + * single over-budget line rides alone (never split a line). */ + function chunkNotes(text: string, budget: number): string[] { + if (text.trim().length === 0) return []; + const out: string[] = []; + let current = ""; + for (const line of text.split("\n")) { + const candidate = current.length === 0 ? line : current + "\n" + line; + if (current.length > 0 && Math.ceil(candidate.length / 4) > budget) { + out.push(current); + current = line; + } else { + current = candidate; + } + } + if (current.trim().length > 0) out.push(current); + return out; + } + + async function notesExtract(input: { id: string; filePath: string; text: string; part?: { i: number; n: number } }): 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); @@ -990,18 +1016,32 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti // (2026-09-25): same refusal as the chat front door — no // extractor/auditor engines, or a stage key gone, no run. refuseIfStagesUnready(); - console.info(`[extraction ${id}] notes pipeline start (${input.filePath})`); - setPhase(id, "extracting", "notes", input.filePath); - const findings = await notesExtract(input); - // SERIALIZED AUDIT (owner 2026-09-28): the same gate as the chat - // path — a notes import and a chat run may extract in parallel, - // but audit + merge one at a time, process-wide. - const stats = await withAuditLock(id, "notes", input.filePath, async () => { - setPhase(id, "auditing", "notes", input.filePath); - const auditResult = await audit(id, findings); - setPhase(id, "merging", "notes", input.filePath); - return mergeFindings(id, findings, auditResult, { chatStore, graphStore }, { foldedAt: input.importedAt }); - }); + // CHUNKED like the chat path (owner, 2026-09-28): each chunk is a + // standalone extract → audit → merge, so no single call carries a + // whole large file. Abort mid-run keeps earlier chunks merged + // (same doctrine as chat batches). + const chunkTokens = deps.getChunkTokens?.() ?? FACTORY_SETTINGS.graph.extractionChunkTokens; + const chunks = chunkNotes(input.text, chunkTokens); + console.info( + `[extraction ${id}] notes pipeline start (${input.filePath}) — ${chunks.length} chunk(s) of ≤${chunkTokens} tok`, + ); + const stats: EmissionStats = { nodes: 0, edges: 0 }; + for (const [i, chunk] of chunks.entries()) { + setPhase(id, "extracting", "notes", input.filePath); + const findings = await notesExtract({ ...input, text: chunk, part: { i: i + 1, n: chunks.length } }); + // SERIALIZED AUDIT (owner, 2026-09-28): the same gate as the chat + // path — a notes import and a chat run may extract in parallel, + // but audit + merge one at a time, process-wide. Reacquired per + // chunk so multi-chunk runs interleave fairly. + const merged = await withAuditLock(id, "notes", input.filePath, async () => { + setPhase(id, "auditing", "notes", input.filePath); + const auditResult = await audit(id, findings); + setPhase(id, "merging", "notes", input.filePath); + return mergeFindings(id, findings, auditResult, { chatStore, graphStore }, { foldedAt: input.importedAt }); + }); + stats.nodes += merged.nodes; + stats.edges += merged.edges; + } // Same post-merge settle as the chat path. settleGraphLayout(graphStore); persistRunUsage(id, Math.round(performance.now() - t0), stats); diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index 39aaf43..09448bc 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -598,7 +598,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { chatStore, graphStore: ledgers, extractor: silentModel(), - auditor: submitModel("submit_review", FINDINGS), + auditor: submitModel("submit_review", { links: [], edges: [] }), }); await assert.rejects(failingExtractor.runExtraction(chatId), /the extractor never submitted findings/); assert.equal(ledgers.readCheckpoints().length, 0); @@ -654,7 +654,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { chatStore, graphStore: ledgers, extractor: submitModel("submit_findings", findingsA), - auditor: gatedSubmitModel("submit_review", findingsA, gate, () => { + auditor: gatedSubmitModel("submit_review", { links: [], edges: [] }, gate, () => { aAuditing = true; }), }); @@ -671,7 +671,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { // THE pin: when B's auditor is finally invoked, A's merge must // already be in the ledgers — B explores the post-merge graph, // which is the entire point of the gate. - auditor: submitModel("submit_review", findingsB, () => { + auditor: submitModel("submit_review", { links: [], edges: [] }, () => { bAuditorCalled = true; bSawAlphasMerge = ledgers.readNodes().some((n) => n.name === "Alpha claim"); }), @@ -708,7 +708,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { chatStore, graphStore: ledgers, extractor: submitModel("submit_findings", single("pa", "Alpha claim", "Alpha statement")), - auditor: gatedSubmitModel("submit_review", single("pa", "Alpha claim", "Alpha statement"), gate), + auditor: gatedSubmitModel("submit_review", { links: [], edges: [] }, gate), }); const runA = pipelineA.runExtraction(a.chatId); await until(() => extractionStatuses().some((r) => r.chatId === a.chatId && r.phase === "auditing")); @@ -718,7 +718,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { chatStore, graphStore: ledgers, extractor: submitModel("submit_findings", single("pb", "Beta claim", "Beta statement")), - auditor: submitModel("submit_review", single("pb", "Beta claim", "Beta statement"), () => { + auditor: submitModel("submit_review", { links: [], edges: [] }, () => { bAuditorCalled = true; }), }); @@ -740,7 +740,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { chatStore, graphStore: ledgers, extractor: submitModel("submit_findings", single("pc", "Gamma claim", "Gamma statement")), - auditor: submitModel("submit_review", single("pc", "Gamma claim", "Gamma statement")), + auditor: submitModel("submit_review", { links: [], edges: [] }), }); assert.deepEqual(await pipelineC.runExtraction(c.chatId), { nodes: 1, edges: 0 }); }); @@ -806,7 +806,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { chatStore, graphStore: ledgers, extractor: hangingModel(), // hangs until aborted — no step finishes - auditor: submitModel("submit_review", { propositions: [], edges: [] }), + auditor: submitModel("submit_review", { links: [], edges: [] }), }); const runPromise = pipeline.runExtraction(chatId); @@ -954,7 +954,7 @@ const NOTES_FINDINGS = { } as const; const NOTES_VERDICT = { - propositions: [NOTES_FINDINGS.propositions[0]], + links: [], edges: [], } as const; @@ -999,6 +999,75 @@ describe("the notes front door (importNotes — format-agnostic)", () => { assert.equal(ledgers.readCheckpoints()[0]!.chatId, source.id); }); + it("chunks a large file: one extract→audit→merge per chunk, part headers in each view (owner, 2026-09-28)", async () => { + const ledgers = memoryLedgers(); + const views: string[] = []; + let calls = 0; + // Fresh engine PER CHUNK via the engineFor factory — same seam + // production uses (one scripted response per model call). + const extractorStep = () => + fauxEngine([ + (context) => { + calls++; + views.push(capturedView(context)); + return fauxAssistantMessage( + [ + fauxToolCall("submit_findings", NOTES_FINDINGS as unknown as ToolCall["arguments"], { + id: `findings-${calls}`, + }), + ], + { stopReason: "toolUse" }, + ); + }, + ]); + const text = Array.from({ length: 8 }, (_, i) => `note line ${i}: ${"x".repeat(30)}`).join("\n"); + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + engineFor: (role) => + role === "extractor" + ? extractorStep() + : submitModel("submit_review", { links: [], edges: [] }), + getChunkTokens: () => 20, // 80 chars/line-groups → several chunks + }); + const stats = await pipeline.importNotes({ filePath: "/vault/big.md", text }); + + assert.ok(calls >= 2, `expected ≥2 extractor calls, got ${calls}`); + assert.match(views[0]!, new RegExp(`part 1/${calls}`)); + assert.match(views.at(-1)!, new RegExp(`part ${calls}/${calls}`)); + // Same findings in every chunk → ONE node total (mapping dedupe + // across chunks), stats never double-count. + assert.deepEqual(stats, { nodes: 1, edges: 0 }); + assert.equal(ledgers.readNodes().length, 1); + }); + + it("required top-level fields (owner, 2026-09-28): a findings submit MISSING edges fails validation — no silent edge-less merge", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + // The degraded-retry shape from the OpenRouter log: propositions + // only, edges omitted. Required schema → validation fails on every + // turn → the run rejects instead of merging a pile of structureless + // nodes. Four queued attempts exhaust the extractor's turn budget. + // Four queued attempts — one per turn of the extractor's budget. + const noEdges = () => + fauxEngine( + Array.from({ length: 4 }, () => (context) => { + void context; + return fauxAssistantMessage( + [fauxToolCall("submit_findings", { propositions: FINDINGS.propositions } as unknown as ToolCall["arguments"])], + { stopReason: "toolUse" }, + ); + }), + ); + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + engineFor: (role) => (role === "extractor" ? noEdges() : submitModel("submit_review", { links: [], edges: [] })), + }); + await assert.rejects(pipeline.runExtraction(chatId), /never submitted findings/); + assert.equal(ledgers.readNodes().length, 0); + }); + it("re-importing the same file is idempotent: same source identity, refs dedupe, usage history grows", async () => { const ledgers = memoryLedgers(); const mk = () => @@ -1270,7 +1339,7 @@ describe("the session-deletion guard (chat deleted mid-run)", () => { chatStore, graphStore: ledgers, extractor: submitModel("submit_findings", FINDINGS), - auditor: submitModel("submit_review", { propositions: FINDINGS.propositions, edges: [] }, () => chatStore.deleteChat(chatId)), + auditor: submitModel("submit_review", { links: [], edges: [] }, () => chatStore.deleteChat(chatId)), }); const stats = await pipeline.runExtraction(chatId);