From 46e128d46c5848b3348af5f9a287f25e5f340c7b Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Mon, 28 Sep 2026 18:00:56 +0300 Subject: [PATCH] serialized audit: the audit+merge critical section is process-wide, one run at a time (owner 2026-09-28) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extraction may run in parallel across sources (extractors are source-local and graph-blind), but the auditor explores the LIVE graph through its tools and stamps identity verdicts against what it finds — two concurrent audits would each link into a graph the other is about to change, the structural mint of duplicate nodes. withAuditLock: a FIFO promise-chain mutex on globalThis (the same module-instance idiom as the run registries — a module-level lock would be per-route-instance and useless in dev). The check runs before any LLM is invoked or the graph read; the lock spans audit through merge inclusive (the merge is the graph mutation) and is reacquired per batch, so interleaved multi-batch runs each audit against the latest graph instead of starving each other. Applied at both front doors — chat batches and notes imports. New 'waiting' phase (extracting → waiting → auditing → merging): visible in the status feed so a queued run never looks stalled in extracting. Abort-aware on both sides of the queue: a run aborted while waiting leaves without auditing (its chain link still releases — successors never hang); a run aborted mid-audit releases in finally. Tests pin: a second run queues at the gate and its auditor is invoked only after the first run's merge is in the ledgers (the post-merge-graph invariant itself); a run aborted while waiting never audits, checkpoints nothing, and leaves a healthy chain for a third run. Gates: typecheck, 1061/1061, lint, build. --- docs/architecture.md | 2 + src/agent/subagents/extraction.ts | 126 +++++++++++++++++++++------- tests/extraction.test.ts | 135 ++++++++++++++++++++++++++++++ 3 files changed, 231 insertions(+), 32 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 0ecfb89..5b631a1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -340,6 +340,8 @@ flowchart TB **Raw reasoning chains only** (2026-09-27 cleanup): the extractor reads ONLY the session — the whole thing, rendered as readable markdown by the shared engine (`core/session-markdown.ts`; the same renderer will serve future user-facing session export) — and proposes free-form propositions (true statements; a misconception is stated as its true correction) and `follows`/`supports` edges with verbatim evidence. Every node is an objective, world-facing fact. No quiz outcomes, no understanding assessments — the events ledger and `UnderstandingLevel` are deleted; the learner's relationship to a proposition becomes practice-module territory (assessment records, later). +**Serialized audit** (owner ruling 2026-09-28): extraction may run in parallel across sources, but the audit→merge section is a process-wide critical section — a FIFO mutex on `globalThis` (the same module-instance idiom as the run registries). Before any auditor LLM call or graph read, a run checks the gate and waits (`waiting` phase, visible in the status feed) until every earlier run's merge has landed, so each auditor explores the graph as it actually stands — two concurrent audits would each link into a graph the other is about to change, the structural mint of duplicate nodes. The lock is reacquired per batch (interleaved runs each see the latest graph instead of starving) and is abort-aware on both sides of the queue. + **Division of trust** (the pipeline's core rule): the extractor is fully trusted and is the only stage that ever sees the source. The auditor never sees the transcript — it receives the findings JSON alone and explores the existing graph through read-only tools (`find`, `list_sessions`, `list_propositions` — statement text included, no timestamps) to reach identity verdicts: `linkTo` an existing node or let the merger create one. Equivalence bias (owner ruling 2026-09-27): paraphrases, rewordings, and compressions of the same claim MERGE; only genuinely distinct propositions stay apart. The auditor may also draw bridging edges — typed `follows`/`supports` connections between existing graph nodes and the batch's propositions (the auditor is the asserter: chat-scoped provenance, no message anchor). The deterministic merger validates and appends: linkTo resolution, dedupe by `(from, to, type)`, coverage checkpoints. There is no drop channel — nothing rejects the extractor's output. **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. diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 9751f34..5df0a01 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -480,8 +480,12 @@ function runAuditor( * The post-batch embedding tail was its own phase (accuracy wave, * 2026-09-19) until wave 7 killed the embedder seam wholesale (this * interlude) — birth-session clusters need no post-merge pass. The - * phases tell the truth about what the loop is actually waiting on. */ -export type ExtractionPhase = "extracting" | "auditing" | "merging"; + * phases tell the truth about what the loop is actually waiting on. + * "waiting" (owner 2026-09-28): the serialized-audit gate — the run's + * audit+merge critical section is queued behind another run's; the + * extractor is done but the auditor may not start until the graph is + * stable. */ +export type ExtractionPhase = "extracting" | "waiting" | "auditing" | "merging"; /** One in-flight pipeline run. Module-level: one process, one registry. * Exported for the status payload's consumer typing (the dock). */ @@ -517,6 +521,11 @@ interface LhExtractionGlobal { * same millisecond never collide on the usage idempotency key. */ lastStartedAt: number; }; + /** The serialized-audit mutex (owner 2026-09-28): a FIFO promise + * chain — the audit+merge critical section runs one run at a time, + * process-wide. globalThis for the same module-instance reason as + * the run registries. */ + __lhAuditLock?: { tail: Promise }; } /** THE RUN REGISTRIES live on globalThis, not module level — the @@ -681,6 +690,46 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti run.usage[stage]!.steps.push(usage); } + /** SERIALIZED AUDIT (owner 2026-09-28): the auditor explores the LIVE + * graph through its tools and stamps identity verdicts against what + * it finds — so the audit+merge section is a process-wide critical + * section. Two concurrent audits would each link into a graph the + * other is about to change: the structural mint of duplicate nodes. + * The extractor stays parallel (source-local, graph-blind). The + * check runs BEFORE any LLM is invoked or the graph read; the lock + * is FIFO, held through the merge, reacquired per batch, and + * abort-aware on both sides of the queue (a waiter that aborts + * leaves without auditing — its chain link still resolves, so + * successors never hang). */ + async function withAuditLock(chatId: string, kind: "chat" | "notes", label: string | undefined, critical: () => Promise): Promise { + setPhase(chatId, "waiting", kind, label); + const g = globalThis as typeof globalThis & LhExtractionGlobal; + const lock = (g.__lhAuditLock ??= { tail: Promise.resolve() }); + const prev = lock.tail; + let release!: () => void; + lock.tail = new Promise((resolve) => (release = resolve)); + const signal = runningExtractions.get(chatId)?.controller.signal; + let onAbort: (() => void) | undefined; + const aborted = new Promise((_, reject) => { + if (!signal) return; + onAbort = () => { + const err = new Error("extraction aborted while waiting for the audit gate"); + err.name = "AbortError"; + reject(err); + }; + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + }); + try { + await Promise.race([prev, aborted]); + signal?.throwIfAborted(); // aborted exactly at handoff + return await critical(); + } finally { + if (onAbort && signal && !signal.aborted) signal.removeEventListener("abort", onAbort); + release(); + } + } + async function extract(chatId: string, view: string): Promise { const signal = runningExtractions.get(chatId)?.controller.signal; signal?.throwIfAborted(); // an abort during a sync gap ends the run here @@ -808,36 +857,44 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti ); setPhase(chatId, "extracting", "chat"); const findings = await extract(chatId, view); - setPhase(chatId, "auditing", "chat"); - const auditResult = await audit(chatId, findings); - setPhase(chatId, "merging", "chat"); - // SESSION-DELETION GUARD (wave 6): a chat deleted mid-run must never - // merge. The deletion route aborts this run first and runs its purge - // as one sync block; this closes the last window — a run that slipped - // past the wind-down finds the chat gone HERE and rides the existing - // abort path (info, not error): nothing merged, nothing checkpointed. - if (!chatStore.getChat(chatId)) { - const gone = new Error(`chat deleted mid-run: ${chatId}`); - gone.name = "AbortError"; - throw gone; - } - // The batch's messages, snapshotted when the run computed - // coverage — if a message mutates mid-run (a quiz answer - // merges in), the stored fingerprint won't match its new - // one and it re-opens: honest by construction. - const foldedAt = new Date().toISOString(); - const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }, { - foldedAt, - coverage: batch - .filter((slot) => messageViewLines(slot.message, slot.message.id).length > 0) - .map((slot) => ({ chatId, fp: slot.fp, tokens: slot.tokens, foldedAt })), + // SERIALIZED AUDIT (owner 2026-09-28): audit + merge are the + // graph-mutating critical section — one run at a time, + // process-wide, so the auditor explores the graph as it + // stands AFTER every earlier run's merge. Reacquired per + // batch: interleaved runs each audit against the latest + // graph instead of starving each other. + await withAuditLock(chatId, "chat", undefined, async () => { + setPhase(chatId, "auditing", "chat"); + const auditResult = await audit(chatId, findings); + setPhase(chatId, "merging", "chat"); + // SESSION-DELETION GUARD (wave 6): a chat deleted mid-run must never + // merge. The deletion route aborts this run first and runs its purge + // as one sync block; this closes the last window — a run that slipped + // past the wind-down finds the chat gone HERE and rides the existing + // abort path (info, not error): nothing merged, nothing checkpointed. + if (!chatStore.getChat(chatId)) { + const gone = new Error(`chat deleted mid-run: ${chatId}`); + gone.name = "AbortError"; + throw gone; + } + // The batch's messages, snapshotted when the run computed + // coverage — if a message mutates mid-run (a quiz answer + // merges in), the stored fingerprint won't match its new + // one and it re-opens: honest by construction. + const foldedAt = new Date().toISOString(); + const batchStats = mergeFindings(chatId, auditResult, { chatStore, graphStore }, { + foldedAt, + coverage: batch + .filter((slot) => messageViewLines(slot.message, slot.message.id).length > 0) + .map((slot) => ({ chatId, fp: slot.fp, tokens: slot.tokens, foldedAt })), + }); + total.nodes += batchStats.nodes; + total.edges += batchStats.edges; }); - total.nodes += stats.nodes; - total.edges += stats.edges; batchesDone++; console.info( `[extraction ${chatId}] batch ${batchesDone}/${batches.length} merged: ` + - `${stats.nodes} nodes, ${stats.edges} edges`, + `${total.nodes} nodes, ${total.edges} edges (cumulative)`, ); } const stats = total; @@ -918,10 +975,15 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti 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 auditResult = await audit(id, findings); - setPhase(id, "merging", "notes", input.filePath); - const stats = mergeFindings(id, auditResult, { chatStore, graphStore }, { foldedAt: input.importedAt }); + // 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, auditResult, { chatStore, graphStore }, { foldedAt: input.importedAt }); + }); // 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 9fde09b..8321c5d 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -270,6 +270,31 @@ function submitModel(name: string, input: unknown, onCall?: () => void): PiEngin ]); } +/** A submit model whose single response WAITS on a gate before answering — + * the timing control for the serialized-audit tests (owner 2026-09-28): + * the model call has STARTED (onCall fired, the run holds the audit + * gate) but will not resolve until the test opens the gate. */ +function gatedSubmitModel(name: string, input: unknown, gate: Promise, onCall?: () => void): PiEngine { + return fauxEngine([ + async () => { + onCall?.(); + await gate; + return fauxAssistantMessage( + [fauxToolCall(name, input as ToolCall["arguments"], { id: `${name}-call` })], + { stopReason: "toolUse" }, + ); + }, + ]); +} + +/** Poll until true — the queued-phase observations need real ticks. */ +async function until(cond: () => boolean, ms = 2000): Promise { + for (const t0 = Date.now(); !cond(); ) { + if (Date.now() - t0 > ms) throw new Error("condition never became true in time"); + await new Promise((r) => setTimeout(r, 2)); + } +} + /** An engine whose model call never resolves until its abort signal fires * — mimicking a cancelled provider request (the abort path). Like a real * provider, an ALREADY-aborted signal resolves immediately (the abort can @@ -600,6 +625,116 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { assert.equal(extractionStatuses().length, 0, "the registry clears when the run ends"); }); + it("serialized audit (owner 2026-09-28): a second run QUEUES at the gate; its auditor explores the graph only AFTER the first run's merge", async () => { + const a = fixtureChat(); + const b = fixtureChat(); + const ledgers = memoryLedgers(); + + const single = (id: string, name: string, statement: string): SessionFindings => ({ + propositions: [{ id, name, statement, aliases: [], messages: ["a2"], evidence: ["e"] }], + edges: [], + }); + const findingsA = single("pa", "Alpha claim", "Alpha statement, full precision"); + const findingsB = single("pb", "Beta claim", "Beta statement, full precision"); + + let gateOpen!: () => void; + const gate = new Promise((resolve) => (gateOpen = resolve)); + let aAuditing = false; + const pipelineA = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", findingsA), + auditor: gatedSubmitModel("submit_review", findingsA, gate, () => { + aAuditing = true; + }), + }); + const runA = pipelineA.runExtraction(a.chatId); + await until(() => extractionStatuses().some((r) => r.chatId === a.chatId && r.phase === "auditing")); + assert.ok(aAuditing, "run A holds the gate — its audit model call is in flight"); + + let bAuditorCalled = false; + let bSawAlphasMerge = false; + const pipelineB = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", findingsB), + // 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, () => { + bAuditorCalled = true; + bSawAlphasMerge = ledgers.readNodes().some((n) => n.name === "Alpha claim"); + }), + }); + const runB = pipelineB.runExtraction(b.chatId); + + await until(() => extractionStatuses().some((r) => r.chatId === b.chatId && r.phase === "waiting")); + assert.ok(!bAuditorCalled, "B's auditor has not been invoked — it queues behind A"); + assert.equal(ledgers.readNodes().filter((n) => n.name === "Beta claim").length, 0, "nothing of B merged yet"); + assert.equal(extractionStatuses().find((r) => r.chatId === a.chatId)?.phase, "auditing", "A still audits while B waits"); + + gateOpen(); + assert.deepEqual(await runA, { nodes: 1, edges: 0 }); + assert.deepEqual(await runB, { nodes: 1, edges: 0 }); + assert.ok(bAuditorCalled); + assert.ok(bSawAlphasMerge, "B's auditor saw A's merged node — audit serialized after merge"); + assert.equal(extractionStatuses().length, 0, "both runs left the registry"); + }); + + it("serialized audit: a run ABORTED while waiting never audits — and the queue chain stays healthy for successors", async () => { + const a = fixtureChat(); + const b = fixtureChat(); + const c = fixtureChat(); + const ledgers = memoryLedgers(); + + const single = (id: string, name: string, statement: string): SessionFindings => ({ + propositions: [{ id, name, statement, aliases: [], messages: ["a2"], evidence: ["e"] }], + edges: [], + }); + + let gateOpen!: () => void; + const gate = new Promise((resolve) => (gateOpen = resolve)); + const pipelineA = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", single("pa", "Alpha claim", "Alpha statement")), + auditor: gatedSubmitModel("submit_review", single("pa", "Alpha claim", "Alpha statement"), gate), + }); + const runA = pipelineA.runExtraction(a.chatId); + await until(() => extractionStatuses().some((r) => r.chatId === a.chatId && r.phase === "auditing")); + + let bAuditorCalled = false; + const pipelineB = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", single("pb", "Beta claim", "Beta statement")), + auditor: submitModel("submit_review", single("pb", "Beta claim", "Beta statement"), () => { + bAuditorCalled = true; + }), + }); + const runB = pipelineB.runExtraction(b.chatId); + await until(() => extractionStatuses().some((r) => r.chatId === b.chatId && r.phase === "waiting")); + + // The dock's abort button, pressed while B queues. + assert.equal(abortExtraction(b.chatId), true); + assert.deepEqual(await runB, { nodes: 0, edges: 0 }, "aborted before auditing — zero emissions"); + assert.ok(!bAuditorCalled, "B's auditor was never invoked"); + assert.ok(!extractionStatuses().some((r) => r.chatId === b.chatId), "B left the registry"); + assert.equal(ledgers.readCoverage().some((row) => row.chatId === b.chatId), false, "B checkpointed nothing"); + + // The chain link B abandoned still resolved: A finishes, and a + // THIRD run audits normally — no deadlock residue. + gateOpen(); + assert.deepEqual(await runA, { nodes: 1, edges: 0 }); + const pipelineC = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: submitModel("submit_findings", single("pc", "Gamma claim", "Gamma statement")), + auditor: submitModel("submit_review", single("pc", "Gamma claim", "Gamma statement")), + }); + assert.deepEqual(await pipelineC.runExtraction(c.chatId), { nodes: 1, edges: 0 }); + }); + it("captures per-stage usage live, persists the run's usage into the chat record", async () => { const { chatId } = fixtureChat(); const ledgers = memoryLedgers();