diff --git a/app/api/graph/import/route.ts b/app/api/graph/import/route.ts index 03f5bf3..e719e1d 100644 --- a/app/api/graph/import/route.ts +++ b/app/api/graph/import/route.ts @@ -13,7 +13,7 @@ * healthy. The dock's chips + graph refetch track progress live through * /api/graph/status, as with session runs. */ -import { extraction } from "@/src/server"; +import { extraction, NotesImportError } from "@/src/server"; import { ModelsNotConfiguredError, PROVIDER_KEY_MISSING_CODE, @@ -24,6 +24,8 @@ import { export const dynamic = "force-dynamic"; interface ImportResult { + requestIndex: number; + outcome: "completed" | "failed" | "partial" | "aborted" | "busy"; filePath: string; ok: boolean; stats?: EmissionStats; @@ -46,20 +48,22 @@ export async function POST(request: Request) { } const results: ImportResult[] = []; - const queue = [...files]; + const queue = files.map((file, requestIndex) => ({ file, requestIndex })); const workers = Array.from({ length: Math.min(3, files.length) }, async () => { for (;;) { - const file = queue.shift(); - if (!file) return; + const next = queue.shift(); + if (!next) return; + const { file, requestIndex } = next; try { const text = await file.text(); const stats = await extraction.importNotes({ filePath: file.name, text }); - results.push({ filePath: file.name, ok: true, stats }); + results.push({ filePath: file.name, requestIndex, outcome: "completed", ok: true, stats }); } catch (err) { if (err instanceof ModelsNotConfiguredError) { results.push({ filePath: file.name, ok: false, + requestIndex, outcome: "failed", error: err.envelope.error, code: "models_not_configured", missing: err.envelope.missing, @@ -70,6 +74,7 @@ export async function POST(request: Request) { results.push({ filePath: file.name, ok: false, + requestIndex, outcome: "failed", error: err.envelope.error, code: PROVIDER_KEY_MISSING_CODE, provider: err.envelope.provider, @@ -77,7 +82,12 @@ export async function POST(request: Request) { }); continue; } - results.push({ filePath: file.name, ok: false, error: err instanceof Error ? err.message : String(err) }); + results.push({ + filePath: file.name, requestIndex, ok: false, + outcome: err instanceof NotesImportError ? err.outcome : "failed", + ...(err instanceof NotesImportError ? { stats: err.partialStats } : {}), + error: err instanceof Error ? err.message : String(err), + }); } } }); diff --git a/components/graph-view.tsx b/components/graph-view.tsx index 3a94b9a..379294e 100644 --- a/components/graph-view.tsx +++ b/components/graph-view.tsx @@ -64,6 +64,7 @@ */ import { useCallback, useDeferredValue, useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; +import { NotesImportResults, useNotesImport } from "@/components/notes-import.tsx"; import Link from "next/link"; import { FileTextIcon, ListTreeIcon, Loader2Icon, SearchIcon, SettingsIcon, UploadIcon, WaypointsIcon } from "lucide-react"; import { incrementalLayout, propositionSessionHref, type GraphCostSummary, type GraphNode, type GraphSnapshot } from "@/src/core"; @@ -1088,29 +1089,19 @@ export function GraphView({ // /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); // The dock's start-site nudge (accuracy wave, 2026-09-19): the import // POST awaits the WHOLE pipeline server-side, so the run registers long // before the response — nudge at click and again at +600ms to catch the // start; from its first setPhase the run announces itself on the events // channel (S5), and the dock rides those announcements thereafter. const refreshActivity = useActivity().refresh; - const onImportFiles = (files: FileList | null) => { - if (!files || files.length === 0) return; - setImporting(true); - refreshActivity(); - setTimeout(() => refreshActivity(), 600); - 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)); - }; + const notesImport = useNotesImport({ + onStart: () => { + refreshActivity(); + setTimeout(() => refreshActivity(), 600); + }, + }); + const { importing, onImportFiles } = notesImport; // Background pipeline status: read /api/graph/status; whenever a run // LEAVES the feed (finished / aborted / failed), refetch the graph. @@ -1614,6 +1605,7 @@ export function GraphView({ /> + {/* Top-right tail (owner, 2026-09-20): the pipeline cost number — just a number, hover-highlighted, dropdown breakdown — then the running-pipeline pills under it. Lives BELOW the actions diff --git a/components/minimal-home.tsx b/components/minimal-home.tsx index 47b14b8..569030b 100644 --- a/components/minimal-home.tsx +++ b/components/minimal-home.tsx @@ -22,6 +22,7 @@ import { useCallback, useEffect, useRef, useState } from "react"; import { ListTreeIcon, UploadIcon } from "lucide-react"; import { ChainTableOverlay } from "@/components/chain-table.tsx"; import { GraphCostControl } from "@/components/graph-view.tsx"; +import { NotesImportResults, useNotesImport } from "@/components/notes-import.tsx"; import { Hint } from "@/components/ui/hint"; import type { ChatSummary, GraphCostSummary, GraphSnapshot } from "@/src/core"; @@ -96,7 +97,6 @@ export function MinimalHome({ const [costs, setCosts] = useState(null); const [chats, setChats] = useState([]); const [chainsOpen, setChainsOpen] = useState(false); - const [importing, setImporting] = useState(false); const fileInputRef = useRef(null); const readState = useCallback(() => { @@ -123,25 +123,9 @@ export function MinimalHome({ if (chainSelectOpen) setChainsOpen(true); }, [chainSelectOpen]); - // Notes import — the same front door as the graph home: POST awaits - // the whole pipeline server-side, then one refetch lands the numbers. - 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); - readState(); - }); - }; + // Refetch even after incomplete imports: earlier writes and spend remain. + const notesImport = useNotesImport({ onFinish: readState }); + const { importing, onImportFiles } = notesImport; const teachSessions = chats.filter((c) => c.kind === "teach").length; const practiceSessions = chats.filter((c) => c.kind === "practice").length; @@ -210,6 +194,7 @@ export function MinimalHome({ + {costs && costs.runs.length > 0 && } diff --git a/components/notes-import.tsx b/components/notes-import.tsx new file mode 100644 index 0000000..41c6d59 --- /dev/null +++ b/components/notes-import.tsx @@ -0,0 +1,124 @@ +"use client"; + +import { useRef, useState } from "react"; + +// Transport-only types: never import the server's extraction/agent module here. +export interface NotesImportResult { + requestIndex: number; + filePath: string; + ok: boolean; + outcome: "completed" | "failed" | "partial" | "aborted" | "busy"; + stats?: { nodes: number; chains?: number; droppedChains?: number }; + error?: string; + code?: string; + missing?: string[]; + provider?: string; + slots?: string[]; +} +export interface NotesImportRow { + file: File; + pending: boolean; + result?: NotesImportResult; +} + +/** Validate the entire response before applying any row: a malformed index + * must never attribute another file's success to this file (same names included). */ +export function parseNotesImportResults(body: unknown, files: readonly File[]): NotesImportResult[] { + const invalid = () => new Error("Invalid import response. Please retry the file."); + if (!body || typeof body !== "object" || !("results" in body) || !Array.isArray(body.results) || body.results.length !== files.length) throw invalid(); + const seen = new Set(); + for (const raw of body.results) { + if (!raw || typeof raw !== "object") throw invalid(); + const r = raw as NotesImportResult; + if (!Number.isInteger(r.requestIndex) || r.requestIndex < 0 || r.requestIndex >= files.length || seen.has(r.requestIndex) || r.filePath !== files[r.requestIndex]!.name) throw invalid(); + seen.add(r.requestIndex); + if (!["completed", "failed", "partial", "aborted", "busy"].includes(r.outcome) || typeof r.ok !== "boolean" || r.ok !== (r.outcome === "completed")) throw invalid(); + if (r.stats !== undefined && (!r.stats || typeof r.stats !== "object" || !Number.isFinite(r.stats.nodes) || r.stats.nodes < 0 || (r.stats.chains !== undefined && (!Number.isFinite(r.stats.chains) || r.stats.chains < 0)))) throw invalid(); + if (r.ok && !r.stats) throw invalid(); + if (!r.ok && typeof r.error !== "string") throw invalid(); + for (const field of ["error", "code", "provider"] as const) if (r[field] !== undefined && typeof r[field] !== "string") throw invalid(); + for (const field of ["missing", "slots"] as const) if (r[field] !== undefined && (!Array.isArray(r[field]) || !r[field]!.every((v) => typeof v === "string"))) throw invalid(); + } + return body.results as NotesImportResult[]; +} + +export async function postNotesImport(files: readonly File[]): Promise { + const form = new FormData(); + for (const file of files) form.append("files", file); + const response = await fetch("/api/graph/import", { method: "POST", body: form }); + let body: unknown; + try { body = await response.json(); } + catch { throw new Error("Invalid import response. Please retry the file."); } + if (!response.ok) { + const detail = body && typeof body === "object" && "error" in body && typeof body.error === "string" ? `: ${body.error}` : ""; + throw new Error(`Import request failed (HTTP ${response.status})${detail}. Please retry the file.`); + } + return parseNotesImportResults(body, files); +} + +/** Files live only with the latest selection. Retry submits exactly one captured + * File, not its basename; no automatic retries or persistent import history. */ +export function useNotesImport(callbacks: { onStart?: () => void; onFinish?: () => void } = {}) { + const [rows, setRows] = useState([]); + const [importing, setImporting] = useState(false); + const pending = useRef(false); + const submit = async (files: File[], retryIndex?: number) => { + if (pending.current || files.length === 0) return; + pending.current = true; + setImporting(true); + if (retryIndex === undefined) setRows(files.map((file) => ({ file, pending: true }))); + else setRows((prev) => prev.map((row, i) => i === retryIndex ? { file: row.file, pending: true } : row)); + try { + callbacks.onStart?.(); + const results = await postNotesImport(files); + setRows((prev) => { + const next = [...prev]; + for (const result of results) { + const index = retryIndex ?? result.requestIndex; + next[index] = { file: files[result.requestIndex]!, pending: false, result }; + } + return next; + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + setRows((prev) => prev.map((row, i) => retryIndex === undefined || i === retryIndex ? { + file: row.file, pending: false, + result: { filePath: row.file.name, requestIndex: retryIndex === undefined ? i : 0, ok: false, outcome: "failed", error: message }, + } : row)); + } finally { + pending.current = false; + setImporting(false); + callbacks.onFinish?.(); + } + }; + return { + rows, importing, + onImportFiles: (files: FileList | null) => { if (files) void submit(Array.from(files)); }, + retry: (index: number) => { if (rows[index]) void submit([rows[index].file], index); }, + dismiss: () => { if (!pending.current) setRows([]); }, + }; +} + +export function NotesImportResults({ rows, importing, retry, dismiss }: Pick, "rows" | "importing" | "retry" | "dismiss">) { + if (rows.length === 0) return null; + return ( +
+
    + {rows.map((row, i) => { + const r = row.result; + const message = row.pending ? "Importing…" : r?.outcome === "completed" ? `Import complete — ${r.stats!.nodes} new propositions, ${r.stats!.chains ?? 0} new chains.` : r?.outcome === "partial" ? "Some parts completed; others failed. Saved changes were kept." : r?.outcome === "aborted" ? "Import stopped. Any changes already saved were kept." : r?.outcome === "busy" ? "This file is already being imported. Try again when it finishes." : "Import did not complete. Any changes already saved were kept."; + return
  • +

    {row.file.name}

    +

    {message}

    + {r && !r.ok && <> +

    {r.error}

    + {r.code &&

    {r.code}{r.provider ? ` — ${r.provider}` : ""}{r.missing?.length ? ` — ${r.missing.join(", ")}` : ""}{r.slots?.length ? ` — ${r.slots.join(", ")}` : ""}

    } + + } +
    • ; + })} +
+ +
+ ); +} diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 2dd101b..e011869 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -742,6 +742,20 @@ export interface NotesImportInput { text: string; } +/** A notes import did not finish. Earlier graph writes and billed usage are + * retained; partialStats describe known merges, not an all-or-nothing result. */ +export class NotesImportError extends Error { + constructor( + public readonly outcome: "failed" | "partial" | "aborted" | "busy", + public readonly partialStats: EmissionStats, + message: string, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "NotesImportError"; + } +} + 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). */ @@ -1285,9 +1299,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti async function runNotesExtraction(input: { id: string; filePath: string; text: string; importedAt: string }): Promise { const { id } = input; - if (inFlight.has(id)) return { nodes: 0, chains: 0 }; - inFlight.add(id); + // importNotes owns the source exclusion, including metadata and cleanup. const t0 = performance.now(); + const mergedStats: EmissionStats = { nodes: 0, chains: 0 }; try { // BYOK per-need gate (owner, 2026-09-22) + key presence // (2026-09-25): same refusal as the chat front door — no @@ -1313,11 +1327,15 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti const settled: PromiseSettledResult[] = []; for (const [chunkIndex, chunk] of chunks.entries()) { const part = chunks.length > 1 ? { i: chunkIndex + 1, n: chunks.length } : undefined; + // Keep the actual signal after runJob tears down its registry row: + // usage persistence can throw and mask the body's AbortError. + let chunkSignal: AbortSignal | undefined; try { settled.push({ status: "fulfilled", value: await runJob({ id, kind: "notes", label: input.filePath, part }, async (jobKey) => { setPhase(jobKey, "queued", "notes", input.filePath, part); + chunkSignal = runningExtractions.get(jobKey)?.controller.signal; const findings = await withExtractSlot(jobKey, async () => { setPhase(jobKey, "extracting", "notes", input.filePath, part); return notesExtract(jobKey, { ...input, text: chunk, part }); @@ -1337,6 +1355,11 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti { chatStore, graphStore }, { foldedAt: input.importedAt }, ); + // Capture the committed merge BEFORE layout/usage can fail. + mergedStats.nodes += merged.nodes; + mergedStats.chains! += merged.chains ?? 0; + if (merged.droppedChains) + mergedStats.droppedChains = (mergedStats.droppedChains ?? 0) + merged.droppedChains; // The layout ledger evolves with each merge — the next // snapshot fetch reflects every finished chunk. settleGraphLayout(graphStore); @@ -1346,46 +1369,63 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti }), }); } catch (err) { - settled.push({ status: "rejected", reason: err }); + let failure = err; + if (chunkSignal?.aborted && !(err instanceof Error && err.name === "AbortError")) { + // Notes-only: never start a fresh chunk after user cancellation, + // even when usage cleanup replaced the original abort rejection. + // Keep the secondary failure visible in the final import error. + const aborted = new Error(`Notes import aborted; job cleanup failed: ${err instanceof Error ? err.message : String(err)}`, { cause: err }); + aborted.name = "AbortError"; + failure = aborted; + } + settled.push({ status: "rejected", reason: failure }); // Same source-wide-abort law as the chat door: stop the // run; isolated failures keep going. - if (err instanceof Error && err.name === "AbortError") break; + if (failure instanceof Error && failure.name === "AbortError") break; } } const stats = aggregateJobs(id, settled); + // Completion is a chunk-status contract, never a node-count test. + // Zero findings (and an empty file after the readiness gate) are valid. + const rejected = settled.filter((r) => r.status === "rejected"); + if (rejected.length > 0 || settled.length !== chunks.length) { + const aborted = rejected.some((r) => r.reason instanceof Error && r.reason.name === "AbortError"); + const outcome = aborted ? "aborted" : settled.some((r) => r.status === "fulfilled") ? "partial" : "failed"; + const details = rejected.map((r) => r.reason instanceof Error ? r.reason.message : String(r.reason)).join("; "); + throw new NotesImportError(outcome, mergedStats, details || "Notes import did not complete"); + } console.info( `[extraction ${id}] merged: ${stats.nodes} nodes, ${stats.chains ?? 0} chains ` + `(${Math.round(performance.now() - t0)}ms total, ${chunks.length} chunk job(s))`, ); return stats; } catch (err) { - // Pre-fan-out failures only (stage gate, chunking): job-level - // failures and aborts are absorbed at the fan-out site, per job, - // with their usage records already persisted. console.error(`[extraction ${id}] notes pipeline failed (${Math.round(performance.now() - t0)}ms):`, err); - throw err; + if (err instanceof NotesImportError || err instanceof ModelsNotConfiguredError || err instanceof ProviderKeyMissingError) throw err; + throw new NotesImportError(err instanceof Error && err.name === "AbortError" ? "aborted" : "failed", mergedStats, + err instanceof Error ? err.message : String(err), { cause: err }); } finally { - inFlight.delete(id); finishExtraction(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. Re-import guard (owner, 2026-09-29): - * the file's content fingerprint decides — unchanged text is a TRUE - * no-op (the mappings ledger used to dedupe this at merge time by - * accident); changed text re-runs the whole file. */ + /** Every explicit import processes the whole file. Identity, source digest, + * auditor and usage history remain source-scoped; this is not exactly-once. */ async function importNotes(input: NotesImportInput): Promise { const id = `src_${crypto.createHash("sha256").update(input.filePath).digest("hex").slice(0, 12)}`; - const contentFp = crypto.createHash("sha256").update(input.text).digest("hex").slice(0, 16); - if (graphStore.readSources().find((s) => s.id === id)?.contentFp === contentFp) { - console.info(`[extraction ${id}] re-import with unchanged content (fp match) — skipping`); - return { nodes: 0, chains: 0 }; + // Acquire BEFORE source upsert: overlapping contents must not alter metadata. + if (inFlight.has(id)) throw new NotesImportError("busy", { nodes: 0, chains: 0 }, "This file is already being imported. Try again when it finishes."); + inFlight.add(id); + try { + const importedAt = new Date().toISOString(); + graphStore.upsertSource({ id, filePath: input.filePath, importedAt }); + return await runNotesExtraction({ id, ...input, importedAt }); + } catch (err) { + if (err instanceof NotesImportError || err instanceof ModelsNotConfiguredError || err instanceof ProviderKeyMissingError) throw err; + throw new NotesImportError("failed", { nodes: 0, chains: 0 }, err instanceof Error ? err.message : String(err), { cause: err }); + } finally { + inFlight.delete(id); } - const importedAt = new Date().toISOString(); - graphStore.upsertSource({ id, filePath: input.filePath, importedAt, contentFp }); - return runNotesExtraction({ id, ...input, importedAt }); } return { runExtraction, importNotes }; diff --git a/src/core/contracts/graph.ts b/src/core/contracts/graph.ts index d578276..3b21bc3 100644 --- a/src/core/contracts/graph.ts +++ b/src/core/contracts/graph.ts @@ -95,11 +95,9 @@ export interface NotesSource { filePath: string; /** ISO — when this file was imported (the checkpoint's foldedAt). */ importedAt: string; - /** sha256 of the imported text at the last EXTRACTION run — the - * same-path re-import guard (owner, 2026-09-29): a matching - * fingerprint is a TRUE no-op (no model calls, no usage record); - * a mismatch or a legacy missing stamp re-runs the whole file. - * The mappings ledger used to give this idempotency by accident. */ + /** Legacy content fingerprint. Accepted when reading old records, but + * ignored: every explicit notes import processes the whole file. New + * imports do not compute or write this field. No migration is needed. */ contentFp?: string; /** The run's usage log — same append-only discipline as * ChatRecord.extractions, idempotent per startedAt. */ diff --git a/src/server/index.ts b/src/server/index.ts index c614f91..bf0212b 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -8,6 +8,7 @@ export * from "./chat-handler.ts"; export * from "./events-stream.ts"; export * from "./chats.ts"; export * from "./graph.ts"; +export { NotesImportError } from "../agent/subagents/extraction.ts"; export * from "./commit-positions.ts"; export * from "./jobs.ts"; export * from "./stores.ts"; diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index ba22333..37078ad 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -22,6 +22,7 @@ */ import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; import * as fs from "node:fs"; import { after, describe, it } from "node:test"; import { createFauxCore, fauxAssistantMessage, fauxToolCall, type FauxResponseStep, type ToolCall } from "@earendil-works/pi-ai"; @@ -44,6 +45,7 @@ import type { GraphStore } from "../src/core/ports.ts"; import { abortExtraction, createExtractionPipeline, + NotesImportError, extractionStatuses, transcriptView, sourceDigest, @@ -1584,7 +1586,7 @@ describe("the notes front door (importNotes — format-agnostic)", () => { assert.equal(ledgers.readCoverage().length, 0, "nothing checkpointed — the messages stay re-runnable"); }); - it("re-importing is fingerprint-guarded: unchanged text is a TRUE no-op; changed text re-runs (owner, 2026-09-29)", async () => { + it("every explicit import runs the whole file, even identical text and legacy fingerprints", async () => { const ledgers = memoryLedgers(); const mk = () => createExtractionPipeline({ @@ -1596,23 +1598,21 @@ describe("the notes front door (importNotes — format-agnostic)", () => { const first = await mk().importNotes({ filePath: "/vault/learn vim.md", text: NOTES_TEXT }); assert.deepEqual(first, { nodes: 1, chains: 0 }); - // Unchanged text: the content fingerprint matches — no extraction - // at all (no model calls, no usage record; the mappings ledger - // used to give this no-op at merge time by accident). + // A legacy stamp must never suppress an explicit import. + ledgers.upsertSource({ ...ledgers.readSources()[0]!, contentFp: createHash("sha256").update(NOTES_TEXT).digest("hex").slice(0, 16) }); const second = await mk().importNotes({ filePath: "/vault/learn vim.md", text: NOTES_TEXT }); - assert.deepEqual(second, { nodes: 0, chains: 0 }, "unchanged content is a true no-op"); + assert.deepEqual(second, { nodes: 1, chains: 0 }, "identical content re-extracts"); + assert.equal(ledgers.readSources()[0]!.contentFp, undefined, "new imports write no fingerprint"); assert.equal(ledgers.readSources().length, 1, "one source identity — the path hashes to the same id"); - assert.equal(ledgers.readSources()[0]!.extractions?.length, 1, "only the real run billed its usage"); - assert.equal(ledgers.readNodes().length, 1, "no duplicate node"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2, "both runs bill usage"); + assert.equal(ledgers.readNodes().length, 2, "the auditor, not a content skip, owns identity"); - // Changed text under the same path: the fingerprint misses — the - // whole file re-runs (fresh merge; the auditor's links would dedupe - // identity on a real model). + // Changed text also re-runs the whole file. const third = await mk().importNotes({ filePath: "/vault/learn vim.md", text: `${NOTES_TEXT}\nA new closing line.` }); assert.deepEqual(third, { nodes: 1, chains: 0 }, "changed content re-extracts"); - assert.equal(ledgers.readSources()[0]!.extractions?.length, 2, "the re-run bills its usage"); - assert.equal(ledgers.readNodes().length, 2); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 3, "the re-run bills its usage"); + assert.equal(ledgers.readNodes().length, 3); }); it("aborts like a chat run: true cancellation, nothing merged, partial usage in the source record", async () => { @@ -1627,7 +1627,7 @@ describe("the notes front door (importNotes — format-agnostic)", () => { const srcId = extractionStatuses().find((r) => r.chatId.startsWith("src_"))!.chatId; assert.equal(abortExtraction(srcId), true); - assert.deepEqual(await runPromise, { nodes: 0, chains: 0 }); + await assert.rejects(runPromise, (error) => error instanceof NotesImportError && error.outcome === "aborted" && error.partialStats.nodes === 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"); @@ -1935,7 +1935,7 @@ describe("per-chunk parallel jobs (owner, 2026-09-28)", () => { // SOURCE-WIDE abort: the parked job dies and the run ENDS — no // later chunk ever mints (fresh rows would hang un-aborted). assert.equal(abortExtraction(srcId), true); - assert.deepEqual(await runPromise, { nodes: 0, chains: 0 }); + await assert.rejects(runPromise, (error) => error instanceof NotesImportError && error.outcome === "aborted"); assert.equal(extractionStatuses().filter((r) => r.chatId === srcId).length, 0, "all rows leave the registry"); assert.equal(ledgers.readNodes().length, 0, "nothing merged"); // Per-job usage: the ONE minted job persisted its (failed) spend. @@ -2318,3 +2318,190 @@ describe("chain simplification (the merger's maintenance pass, owner 2026-09-30 assert.equal(stored.length, 2, "the ledger is back to the two lean chains"); }); }); + + +describe("notes import completion and retry reliability", () => { + const input = { filePath: "reliability.md", text: NOTES_TEXT }; + const make = (graphStore: GraphStore, extractor = submitModel("submit_findings", NOTES_FINDINGS)) => + createExtractionPipeline({ chatStore, graphStore, extractor, auditor: submitModel("submit_review", NOTES_VERDICT) }); + const incomplete = (outcome: NotesImportError["outcome"], nodes: number) => (error: unknown) => { + assert.ok(error instanceof NotesImportError); + assert.equal(error.outcome, outcome); + assert.equal(error.partialStats.nodes, nodes); + return true; + }; + + it("failure followed by an unchanged retry really extracts and preserves both usage records", async () => { + const ledgers = memoryLedgers(); + await assert.rejects(make(ledgers, silentModel()).importNotes(input), incomplete("failed", 0)); + assert.deepEqual(await make(ledgers).importNotes(input), { nodes: 1, chains: 0 }); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2); + }); + + it("zero new nodes and an empty file are successful, but the readiness gate still applies", async () => { + const ledgers = memoryLedgers(); + assert.deepEqual(await make(ledgers, submitModel("submit_findings", { propositions: [], chains: [] })).importNotes(input), { nodes: 0, chains: 0 }); + assert.deepEqual(await make(ledgers).importNotes({ ...input, text: " \n" }), { nodes: 0, chains: 0 }); + await assert.rejects(createExtractionPipeline({ chatStore, graphStore: ledgers }).importNotes({ ...input, text: "" }), ModelsNotConfiguredError); + await assert.rejects(createExtractionPipeline({ chatStore, graphStore: ledgers, extractor: submitModel("submit_findings", NOTES_FINDINGS), auditor: submitModel("submit_review", NOTES_VERDICT), providerKeyMissing: () => providerKeyMissing("openai", ["cheap"]) }).importNotes(input), ProviderKeyMissingError); + }); + + it("isolates an ordinary failed chunk, reports partial, and keeps earlier/later merges and usage", async () => { + const ledgers = memoryLedgers(); + let calls = 0; + const pipeline = createExtractionPipeline({ + chatStore, graphStore: ledgers, getChunkTokens: () => 1, + engineFor: (role) => role === "auditor" ? submitModel("submit_review", NOTES_VERDICT) : fauxEngine([() => { + calls++; + if (calls === 2) throw new Error("synthetic failed chunk"); + return fauxAssistantMessage([fauxToolCall("submit_findings", NOTES_FINDINGS as unknown as ToolCall["arguments"])], { stopReason: "toolUse" }); + }]), + }); + await assert.rejects(pipeline.importNotes({ ...input, text: "first chunk\nsecond chunk\nthird chunk" }), incomplete("partial", 2)); + assert.equal(calls, 3); + assert.equal(ledgers.readNodes().length, 2); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 3); + }); + + it("an abort after an earlier merge stops later chunks without losing graph or billed usage", async () => { + const ledgers = memoryLedgers(); + let calls = 0; + const pipeline = createExtractionPipeline({ + chatStore, graphStore: ledgers, getChunkTokens: () => 1, + engineFor: (role) => role === "auditor" ? submitModel("submit_review", NOTES_VERDICT) : submitModel("submit_findings", NOTES_FINDINGS, () => { + calls++; + if (calls === 2) abortExtraction(ledgers.readSources()[0]!.id); + }), + }); + await assert.rejects(pipeline.importNotes({ ...input, text: "first chunk\nsecond chunk\nthird chunk" }), incomplete("aborted", 1)); + assert.equal(calls, 2); + assert.equal(ledgers.readNodes().length, 1); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2); + }); + + for (const abortAt of [1, 2]) it(`abort in chunk ${abortAt} survives a usage-persistence error and stops later chunks`, async () => { + const ledgers = memoryLedgers(); + let calls = 0; + let cancelled = false; + let usageFailures = 0; + const graphStore: GraphStore = { ...ledgers, appendSourceUsage: (id, entry) => { + if (cancelled) { usageFailures++; throw new Error("synthetic usage persistence failure during abort"); } + ledgers.appendSourceUsage(id, entry); + } }; + const pipeline = createExtractionPipeline({ + chatStore, graphStore, getChunkTokens: () => 1, + engineFor: (role) => role === "auditor" ? submitModel("submit_review", NOTES_VERDICT) : submitModel("submit_findings", NOTES_FINDINGS, () => { + calls++; + if (calls === abortAt) { + cancelled = true; + assert.equal(abortExtraction(ledgers.readSources()[0]!.id), true); + } + }), + }); + const wholeFile = { ...input, text: "first chunk\nsecond chunk\nthird chunk" }; + await assert.rejects(pipeline.importNotes(wholeFile), (error) => { + incomplete("aborted", abortAt - 1)(error); + assert.match((error as NotesImportError).message, /synthetic usage persistence failure during abort/); + return true; + }); + assert.equal(calls, abortAt, "no later extractor runs with a fresh controller after cancellation"); + assert.equal(usageFailures, 1, "the aborted chunk attempted to persist usage and surfaced its failure"); + assert.equal(ledgers.readNodes().length, abortAt - 1, "earlier graph merges retained"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, abortAt - 1, "earlier usage retained"); + assert.equal(extractionStatuses().length, 0, "cancelled job row removed"); + assert.deepEqual(await make(ledgers).importNotes(wholeFile), { nodes: 1, chains: 0 }, "source exclusion released for an explicit unchanged whole-file retry"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, abortAt); + }); + + it("a usage-persistence error without an aborted signal remains an isolated failure and later chunks run", async () => { + const ledgers = memoryLedgers(); + let calls = 0; + const graphStore: GraphStore = { ...ledgers, appendSourceUsage: (id, entry) => { + if (calls === 1) throw new Error("synthetic isolated usage failure"); + ledgers.appendSourceUsage(id, entry); + } }; + const pipeline = createExtractionPipeline({ + chatStore, graphStore, getChunkTokens: () => 1, + engineFor: (role) => role === "auditor" ? submitModel("submit_review", NOTES_VERDICT) : submitModel("submit_findings", NOTES_FINDINGS, () => { calls++; }), + }); + await assert.rejects(pipeline.importNotes({ ...input, text: "first chunk\nsecond chunk\nthird chunk" }), (error) => { + incomplete("partial", 3)(error); + assert.match((error as NotesImportError).message, /synthetic isolated usage failure/); + return true; + }); + assert.equal(calls, 3, "ordinary cleanup failure does not imply cancellation"); + assert.equal(ledgers.readNodes().length, 3); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2); + }); + + it("overlapping identical or different contents are busy BEFORE source metadata changes, across pipeline instances", async () => { + const ledgers = memoryLedgers(); + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + let upserts = 0; + const tracked = { ...ledgers, upsertSource: (source: NotesSource) => { upserts++; ledgers.upsertSource(source); } }; + const first = make(tracked, gatedSubmitModel("submit_findings", NOTES_FINDINGS, gate)).importNotes(input); + const sourceBefore = structuredClone(ledgers.readSources()); + try { + await assert.rejects(make(tracked).importNotes(input), incomplete("busy", 0)); + await assert.rejects(make(tracked).importNotes({ ...input, text: "changed" }), incomplete("busy", 0)); + assert.equal(upserts, 1, "busy calls cannot even attempt source metadata writes"); + assert.deepEqual(ledgers.readSources(), sourceBefore); + } finally { release(); } + await first; + assert.deepEqual(await make(ledgers).importNotes(input), { nodes: 1, chains: 0 }, "lock released after lifecycle"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2); + }); + + it("a fulfilled zero-node chunk plus a failed chunk is partial, not failed or completed", async () => { + const ledgers = memoryLedgers(); + let calls = 0; + const pipeline = createExtractionPipeline({ + chatStore, graphStore: ledgers, getChunkTokens: () => 1, + engineFor: (role) => role === "auditor" ? submitModel("submit_review", NOTES_VERDICT) : fauxEngine([() => { + if (++calls === 2) throw new Error("synthetic failed chunk"); + return fauxAssistantMessage([fauxToolCall("submit_findings", { propositions: [], chains: [] })], { stopReason: "toolUse" }); + }]), + }); + await assert.rejects(pipeline.importNotes({ ...input, text: "first chunk\nsecond chunk" }), incomplete("partial", 0)); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2); + }); + + it("a merger failure after a node append is incomplete without rolling back that write", async () => { + const ledgers = memoryLedgers(); + const broken = { ...ledgers, appendCheckpoint: () => { throw new Error("synthetic checkpoint failure after node append"); } }; + await assert.rejects(make(broken).importNotes(input), incomplete("failed", 0)); + assert.equal(ledgers.readNodes().length, 1, "already appended graph write retained even though merge never returned stats"); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 1); + }); + + for (const seam of ["layout", "usage"] as const) it(`a post-merge ${seam} failure is not false success and reports committed stats`, async () => { + const ledgers = memoryLedgers(); + const originalUsage = ledgers.appendSourceUsage; + let failed = false; + const broken: GraphStore = { ...ledgers, + ...(seam === "layout" ? { writePositions: () => { throw new Error("synthetic layout failure"); } } : { + appendSourceUsage: (id, entry) => { if (!failed) { failed = true; throw new Error("synthetic usage failure"); } originalUsage(id, entry); }, + }), + }; + await assert.rejects(make(broken).importNotes(input), incomplete("failed", 1)); + assert.equal(ledgers.readNodes().length, 1); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 1); + await make(ledgers).importNotes(input); + }); + + it("source-upsert failure is typed and releases exclusion for retry", async () => { + const ledgers = memoryLedgers(); + await assert.rejects(make({ ...ledgers, upsertSource: () => { throw new Error("synthetic metadata failure"); } }).importNotes(input), incomplete("failed", 0)); + await make(ledgers).importNotes(input); + }); + + for (const filePath of ["1.md", "1.1.md", "1 notes.md", "1.1 notes.md"]) it(`imports and repeats identical text for ${filePath}`, async () => { + const ledgers = memoryLedgers(); + await make(ledgers).importNotes({ ...input, filePath }); + await make(ledgers).importNotes({ ...input, filePath }); + assert.equal(ledgers.readSources().length, 1); + assert.equal(ledgers.readSources()[0]!.extractions?.length, 2); + assert.equal(ledgers.readNodes().length, 2); + }); +}); diff --git a/tests/notes-import-route.test.ts b/tests/notes-import-route.test.ts new file mode 100644 index 0000000..7ab5a44 --- /dev/null +++ b/tests/notes-import-route.test.ts @@ -0,0 +1,103 @@ +import assert from "node:assert/strict"; +import { after, it } from "node:test"; +import { POST } from "../app/api/graph/import/route.ts"; +import { extraction } from "../src/server/index.ts"; +import { ModelsNotConfiguredError, ProviderKeyMissingError } from "../src/core/settings.ts"; +import { createExtractionPipeline, NotesImportError } from "../src/agent/subagents/extraction.ts"; +import { createFauxCore, fauxAssistantMessage, fauxToolCall } from "@earendil-works/pi-ai"; +import { stores } from "../src/server/stores.ts"; +import type { GraphStore } from "../src/core/ports.ts"; +import type { GraphNode, GraphChain, NotesSource, SessionCheckpoint } from "../src/core/contracts.ts"; + +const original = extraction.importNotes; +after(() => { extraction.importNotes = original; }); +function request(names: string[]) { + const form = new FormData(); + for (const name of names) form.append("files", new File(["identical text"], name)); + return new Request("http://localhost/api/graph/import", { method: "POST", body: form }); +} + +it("route keeps three-worker batching, request positions and mixed per-file outcomes/config details", async () => { + let active = 0; + let peak = 0; + extraction.importNotes = async ({ filePath }) => { + active++; peak = Math.max(peak, active); + try { + await new Promise((resolve) => setTimeout(resolve, filePath === "slow.md" ? 25 : 1)); + if (filePath === "models.md") throw new ModelsNotConfiguredError(["cheap"]); + if (filePath === "key.md") throw new ProviderKeyMissingError("openai", ["cheap"]); + if (["failed", "partial", "aborted", "busy"].includes(filePath)) throw new NotesImportError(filePath as NotesImportError["outcome"], { nodes: 2, chains: 1 }, `synthetic ${filePath}`); + if (filePath === "unknown.md") throw new Error("synthetic unknown error"); + return { nodes: 0, chains: 0 }; + } finally { active--; } + }; + const names = ["slow.md", "same.md", "same.md", "failed", "partial", "aborted", "busy", "models.md", "key.md", "unknown.md"]; + const response = await POST(request(names)); + assert.equal(response.status, 200); + const { results } = await response.json(); + assert.equal(peak, 3); + assert.notEqual(results[0].requestIndex, 0, "out-of-order completion remains supported"); + assert.equal(results.length, names.length); + for (const [requestIndex, filePath] of names.entries()) { + const result = results.find((r: { requestIndex: number }) => r.requestIndex === requestIndex); + assert.equal(result.filePath, filePath); + if (["failed", "partial", "aborted", "busy"].includes(filePath)) { + assert.equal(result.outcome, filePath); + assert.equal(result.ok, false); + assert.deepEqual(result.stats, { nodes: 2, chains: 1 }); + } else if (["slow.md", "same.md"].includes(filePath)) { + assert.equal(result.outcome, "completed"); assert.equal(result.ok, true); + } else { assert.equal(result.ok, false); assert.equal(result.outcome, "failed"); } + } + assert.deepEqual(results.find((r: { filePath: string }) => r.filePath === "models.md").missing, ["cheap"]); + const key = results.find((r: { filePath: string }) => r.filePath === "key.md"); + assert.equal(key.code, "provider_key_missing"); assert.equal(key.provider, "openai"); assert.deepEqual(key.slots, ["cheap"]); +}); + +it("real route plus in-memory ledgers and faux stages reprocess identical imports for all four filename cases", async () => { + const nodes: GraphNode[] = []; + const chains: GraphChain[] = []; + const checkpoints: SessionCheckpoint[] = []; + const sources = new Map(); + const graphStore: GraphStore = { + appendNode: (node) => { nodes.push(node); }, readNodes: () => [...nodes], + appendChain: (chain) => { chains.push(chain); }, readChains: () => [...chains], + appendCheckpoint: (checkpoint) => { checkpoints.push(checkpoint); }, readCheckpoints: () => [...checkpoints], + appendCoverage: () => {}, readCoverage: () => [], + upsertSource: (source) => { sources.set(source.id, { ...source, extractions: sources.get(source.id)?.extractions ?? [] }); }, + readSources: () => [...sources.values()], + appendSourceUsage: (id, usage) => { sources.get(id)!.extractions!.push(usage); }, + readPositions: () => ({}), writePositions: () => {}, rewriteLedgers: () => {}, + }; + let calls = 0; + const pipeline = createExtractionPipeline({ chatStore: stores.chatStore, graphStore, + engineFor: (role) => { + const faux = createFauxCore({}); + faux.setResponses([() => { + if (role === "extractor") calls++; + return fauxAssistantMessage([fauxToolCall(role === "extractor" ? "submit_findings" : "submit_review", role === "extractor" ? { + propositions: [{ id: "p1", name: "Synthetic import", statement: "Explicit notes imports process the whole file.", aliases: [] }], chains: [], + } : { links: [] })], { stopReason: "toolUse" }); + }]); + return { model: faux.getModel(), streamFn: faux.streamSimple }; + }, + }); + extraction.importNotes = pipeline.importNotes; + const names = ["1.md", "1.1.md", "1 notes.md", "1.1 notes.md"]; + for (let pass = 0; pass < 2; pass++) { + const response = await POST(request(names)); + const { results } = await response.json(); + assert.equal(results.length, 4); + assert.ok(results.every((r: { ok: boolean; outcome: string }) => r.ok && r.outcome === "completed")); + } + assert.equal(calls, 8, "repeat identical imports execute real faux extractor calls"); + assert.equal(nodes.length, 8); + assert.equal(sources.size, 4); + assert.ok([...sources.values()].every((source) => source.extractions?.length === 2 && source.contentFp === undefined)); +}); + +it("route returns 400 without files", async () => { + const response = await POST(request([])); + assert.equal(response.status, 400); + assert.deepEqual(await response.json(), { results: [], error: "no files" }); +}); diff --git a/tests/notes-import.test.tsx b/tests/notes-import.test.tsx new file mode 100644 index 0000000..b66d093 --- /dev/null +++ b/tests/notes-import.test.tsx @@ -0,0 +1,70 @@ +import assert from "node:assert/strict"; +import { after, it } from "node:test"; +import { renderToStaticMarkup } from "react-dom/server"; +import { NotesImportResults, parseNotesImportResults, postNotesImport, type NotesImportResult } from "../components/notes-import.tsx"; + +const originalFetch = globalThis.fetch; +after(() => { globalThis.fetch = originalFetch; }); +const files = [new File(["one"], "same.md"), new File(["two"], "same.md")]; +const completed = (requestIndex = 0): NotesImportResult => ({ requestIndex, filePath: "same.md", ok: true, outcome: "completed", stats: { nodes: 0, chains: 0 } }); + +it("maps out-of-order responses by index, not duplicate basenames", () => { + const results = [completed(1), { ...completed(0), stats: { nodes: 7 } }]; + assert.deepEqual(parseNotesImportResults({ results }, files), results); +}); +for (const [name, results] of [ + ["missing index", [{ ...completed(), requestIndex: undefined }, completed(1)]], + ["duplicate index", [completed(), completed()]], + ["missing row", [completed()]], + ["out-of-range index", [completed(), completed(2)]], + ["wrong filename", [{ ...completed(), filePath: "different.md" }, completed(1)]], + ["missing stats", [{ ...completed(), stats: undefined }, completed(1)]], + ["unknown outcome", [{ ...completed(), outcome: "unchanged" }, completed(1)]], + ["conflicting ok", [{ ...completed(), ok: false }, completed(1)]], + ["invalid stats", [{ ...completed(), stats: { nodes: "bad" } }, completed(1)]], + ["invalid details", [{ ...completed(), slots: [42] }, completed(1)]], +] as const) it(`malformed response (${name}) is rejected for retry`, () => { + assert.throws(() => parseNotesImportResults({ results }, files), /Please retry/); +}); + +it("POST sends only captured files, and retains config/provider details", async () => { + const refused: NotesImportResult = { requestIndex: 0, filePath: files[1]!.name, ok: false, outcome: "failed", code: "provider_key_missing", provider: "openai", slots: ["cheap"], error: "No key is set" }; + globalThis.fetch = async (url, init) => { + assert.equal(url, "/api/graph/import"); assert.equal(init?.method, "POST"); + const sent = (init?.body as FormData).getAll("files") as File[]; + assert.equal(sent.length, 1); assert.equal(await sent[0]!.text(), "two"); + return Response.json({ results: [refused] }); + }; + assert.deepEqual(await postNotesImport([files[1]!]), [refused]); +}); +for (const kind of ["network", "non-JSON", "non-2xx", "malformed"] as const) it(`${kind} response rejects rather than false success`, async () => { + globalThis.fetch = async () => { + if (kind === "network") throw new Error("synthetic offline"); + if (kind === "non-JSON") return new Response("not JSON"); + if (kind === "non-2xx") return Response.json({ error: "synthetic refused" }, { status: 503 }); + return Response.json({ results: [] }); + }; + await assert.rejects(postNotesImport([files[0]!])); +}); + +for (const outcome of ["completed", "failed", "partial", "aborted", "busy"] as const) it(`renders ${outcome}, escaped filename/details, Retry and one Dismiss action`, () => { + const file = new File(["notes"], "