learnharness/src/agent/subagents/extraction.ts

import * as crypto from "node:crypto";
import type { Usage } from "@earendil-works/pi-ai";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import { Type, type Static, type TSchema } from "typebox";
import { toToolResult } from "../tool-execute.ts";
import type { ModelRole } from "../../../config.ts";
import {
	FACTORY_SETTINGS,
	ModelsNotConfiguredError,
	ProviderKeyMissingError,
	ROLE_SLOT,
	type ModelSlotName,
	type ProviderKeyMissing,
} from "../../core/settings.ts";
import type { PiEngine } from "../models.ts";
import { runSubagent } from "./pi-run.ts";
import { withBgLlmSlot } from "./bg-llm-semaphore.ts";
import type {
	AuditVerdicts,
	CallUsage,
	ChatRecord,
	ExtractionRunUsage,
	ExtractionStageUsage,
	SessionFindings,
} from "../../core/contracts.ts";
import { getGraph, mergeFindings, type EmissionStats, type GraphDeps } from "../../core/graph.ts";
import type { GraphStore } from "../../core/ports.ts";
import { settleGraphLayout } from "../../core/layout-settle.ts";
import { batchRuns, coverageOf, messageViewLines } from "../../core/coverage.ts";
import { teacherProseBlocks } from "../../core/coverage.ts";
import type { LhMessage } from "../../core/native-messages.ts";
import { auditorGraphTools } from "../tools/graph.ts";
import { runSubmitAgent } from "./submit-agent.ts";

/**
 * The extraction pipeline (vision §1 construction, 2026-09-02 redesign;
 * relinked 2026-09-27) — the sole graph writer. Three stages, one
 * orchestrator:
 *
 *   extractor (LLM)  the ONE stage that sees the raw source (a session
 *                    rendered as the teacher's prose, or a notes file
 *                    verbatim) → submits propositions + REASONING CHAINS
 *                    (owner redesign, 2026-09-30), expanded at the seam to
 *                    SessionFindings' follows-edges. TREE-BLIND,
 *                    graph-blind, FULLY TRUSTED — nothing downstream
 *                    re-checks it.
 *   auditor (LLM)    never sees the source. Its job is IDENTITY: for each
 *                    proposition, linkTo an existing graph node or create —
 *                    decided by EXPLORING the graph through tools (find /
 *                    list_sessions / list_propositions), not by an injected
 *                    graph dump (context bloat, owner 2026-09-27). Links are
 *                    its ONLY deliverable — it never mints edges (owner
 *                    2026-09-28: the graph connects across sources solely
 *                    through identified equivalences).
 *   merger (core)    mergeFindings() → deterministic ledger appends
 *
 * The events ledger is GONE (2026-09-27): quiz outcomes and understanding
 * observations are no longer formalized in the graph — the graph is purely
 * extracted reasoning chains (nodes + edges). Understanding-level
 * assessment is practice-module territory.
 *
 * The extractor is a single-tool-call worker (one TypeBox-validated submit
 * call — a pi agent purely so the model can retry a schema-invalid call).
 * The auditor is a small AGENTIC loop: exploration tools plus one submit.
 *
 * Created via createExtractionPipeline(deps) — the composition root
 * (chat-handler) injects the storage ports + models; this module never
 * imports an adapter. runExtraction is single-flight per chat: a second
 * trigger while one run is in flight is a no-op (the next trigger picks up
 * 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 and
 * checkpoints, and the source record (graphStore sources seam) carries
 * the file path + the run's usage.
 *
 * 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
 * stats (or failed) for every run.
 */

// ─── the source view (what the extractor sees) ──────────────────────────────

/**
 * The extraction view: the session as TEACHER PROSE ONLY (owner ruling,
 * 2026-09-28 — supersedes the 2026-09-27 whole-session markdown view):
 * plain prose paragraphs under the session title. The shared markdown
 * engine (core/session-markdown.ts) keeps the full session format for
 * the future user-facing export; the extractor no longer uses it.
 */

/** The whole-record view (used by callers that render a full session;
 *  the pipeline itself batches). TEACHER PROSE ONLY (owner, 2026-09-28).
 */
export function transcriptView(record: ChatRecord): string {
	return proseView(record.messages, record.title);
}

/** The extraction view under the teacher-prose ruling (owner, 2026-09-28):
 *  ONLY the teacher agent's text blocks — no learner messages, no
 *  thinking, no tool calls or results, no quiz blocks. Plain paragraphs
 *  under the session title; the shared markdown engine stays reserved for
 *  the future user-facing export. */
function proseView(messages: ReadonlyArray<LhMessage>, title?: string): string {
	const prose = teacherProseBlocks(messages);
	const head = title && title.length > 0 ? `# ${title}\n\n` : "";
	return head + prose.join("\n\n");
}

// ─── the extractor ──────────────────────────────────────────────────────────

const propositionProps = {
	id: Type.String({ description: 'Local id: "p1", "p2", … — referenced by chains.' }),
	name: Type.String({
		description:
			"A very short label — established terminology, the term the discipline itself uses, a glossary " +
			"headword a textbook would print ('Will to power', 'Divine command theory'). Not a sentence, " +
			"needn't be unique or self-contained: this becomes the node's global label in the graph.",
	}),
	statement: Type.String({
		description:
			"One true, self-contained, world-facing reasoning step at full precision. Never about the " +
			'learner, the teacher, or the reading: not "the learner understands X" but "X". A misconception ' +
			"enters as its TRUE correction.",
	}),
	aliases: Type.Optional(
		Type.Array(Type.String(), {
			description: "Verbatim phrasings the source used for the same proposition, if any.",
		}),
	),
};

// The extractor redesign (owner, 2026-09-30): the extractor submits
// REASONING CHAINS — premises + conclusion — stored AS-IS (chains-primary,
// same evening: chains.jsonl is the truth; edges are a derived view, and
// supports edges are retired entirely).
const chainSchema = Type.Object({
	premises: Type.Array(Type.String(), {
		minItems: 1,
		description:
			"Proposition ids used TOGETHER: they jointly entail the conclusion; independent reasons are separate chains.",
	}),
	conclusion: Type.String({ description: "The proposition id the premises jointly imply." }),
});

// Weak-model tolerance (owner, 2026-09-25): the old zod `.default([])` /
// `.default({ quiz: {} })` on required-but-empty fields becomes Type.Optional
// here + a ??-fill at the submit seam (fillFindings/fillAudit below) — same
// contract on the wire (omission tolerated, empty never means "invalid"),
// no zod left in the agent band.
const findingsProps = {
	propositions: Type.Array(Type.Object(propositionProps), {
		description: "The propositions themselves — every distinct proposition the source's reasoning uses.",
	}),
	chains: Type.Array(chainSchema, {
		description:
			"Every argument the source makes; conclusions reusable by id as premises of later chains; " +
			"bare propositions (no argument) ride the propositions list alone; pass [] when none — " +
			"never omit the field.",
	}),
};

const findingsSchema = Type.Object(findingsProps);

/** What the extractor SUBMITS (the redesign's wire shape) — propositions
 *  + chains. fillFindings normalizes it; validateFindingsChains checks
 *  the chain laws; asSessionFindings lifts it straight into
 *  SessionFindings — chains are stored AS-IS (chains-primary, owner
 *  2026-09-30: no edge translation at any seam). */
type FindingsSubmit = Static<typeof findingsSchema>;

// VERDICTS-ONLY (owner, 2026-09-28): the auditor never alters the
// extractor's findings, so it does not restate them — links are its
// ENTIRE deliverable (owner, 2026-09-28 evening: the edges argument is
// GONE; the auditor never mints edges. The graph connects across sources
// solely through identified equivalences — a link, never an inference).
const auditSchema = Type.Object({
	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. Pass [] when none — never omit the field." },
	),
});

// The notes front door's shapes are IDENTICAL to the session shapes now
// (the only difference was quiz attribution, which no longer exists):
// one schema, one findings type, both front doors.
type NotesFindings = SessionFindings;
type NotesAuditResult = AuditVerdicts;

// ── the weak-model string-heal (owner, 2026-09-28) ──
//
// OpenRouter log evidence: weak models under a big submit payload often
// emit the whole `propositions` array as ONE JSON-STRING instead of a
// native array ("propositions.0: must be object"). pi-agent-core runs
// `prepareArguments` BEFORE TypeBox validation — the sanctioned shim
// point (same doctrine as restoreMathEscapes at the display seam: heal
// weak-model wire behavior, don't punish it with a retry). Best-effort:
// heal only a string that parses to an array; anything else passes
// through untouched and fails validation exactly as before.
function healStringifiedArrays(raw: unknown): unknown {
	if (typeof raw !== "object" || raw === null) return raw;
	const out: Record<string, unknown> = { ...(raw as Record<string, unknown>) };
	for (const key of ["propositions", "chains"]) {
		const v = out[key];
		if (typeof v !== "string" || v.trim().length === 0) continue;
		try {
			const parsed: unknown = JSON.parse(v);
			if (Array.isArray(parsed)) out[key] = parsed;
		} catch {
			// not a clean stringify (e.g. truncated) — leave for validation
		}
	}
	return out;
}

// ── the submit seam (owner, 2026-09-28; redesigned 2026-09-30) ──
//
// Every top-level field on each submit tool is REQUIRED — a payload
// missing `chains` (findings) or `links` (review) fails validation
// instead of silently merging structure-less (the degraded-retry hole:
// the model's first truncated attempt failed validation, then the retry
// passed with propositions only — optional fields ??-filled [] — merging
// a pile of nodes with no structure). Only the per-item aliases stay
// optional with a ??-fill.

function fillFindings(raw: FindingsSubmit): FindingsSubmit {
	return {
		propositions: raw.propositions.map((p) => ({ ...p, aliases: p.aliases ?? [] })),
		chains: raw.chains,
	};
}

/** The chain laws (extractor redesign, 2026-09-30) — the LOUD channel:
 *  runSubmitAgent turns a problem string into an error tool result and
 *  the model re-submits within its turns (the practice-proven hook).
 *  Identical (premise-set, conclusion) chains are NOT errors — they
 *  dedupe silently at the merge (chainKey): restating an argument is a
 *  no-op. */
function validateFindingsChains(value: FindingsSubmit): string | null {
	const ids = new Set(value.propositions.map((p) => p.id));
	for (const chain of value.chains) {
		if (chain.premises.length === 0) {
			return "chains: premises must name at least one proposition — a proposition with no argument stays in the propositions list alone.";
		}
		const unknown = [...chain.premises, chain.conclusion].find((id) => !ids.has(id));
		if (unknown !== undefined) {
			return `chains: "${unknown}" is not one of the submitted proposition ids — every premise and conclusion must appear in this call's propositions.`;
		}
		if (chain.premises.includes(chain.conclusion)) {
			return "chains: a conclusion cannot be its own premise.";
		}
	}
	return null;
}

/** (chains-primary, owner 2026-09-30) the wire IS the pipeline shape
 *  now: propositions + chains, no translation — the stores keep the
 *  chains and the derived edge view is recomputed on read. The fill
 *  already ran inside runSubmitAgent; this only re-asserts the
 *  aliases-filled type. */
const asSessionFindings = (raw: FindingsSubmit): SessionFindings => ({
	propositions: raw.propositions.map((p) => ({ ...p, aliases: p.aliases ?? [] })),
	chains: raw.chains,
});

function fillAudit(raw: Static<typeof auditSchema>): AuditVerdicts {
	return {
		links: raw.links,
	};
}

// ── the extractor's system prompt (owner redesign, 2026-09-30) ──
//
// GOAL-FIRST, NO RULE ARMOR: the old doctrine's defensive lists (DROP
// lists, merge-your-own-output, name walls, when-in-doubt-omit, the
// faithfulness paragraph) are ALL CUT — the prompt states the goal and
// the submission mechanics; the tool's field descriptions carry the
// details. One shared core + two contextualizations, assembled here —
// session and notes share the core by construction. (The old opening
// describing Quiz/Recall-check/Question blocks was stale since the
// 2026-09-28 teacher-prose ruling and is simply omitted.)

export const SESSION_PREAMBLE =
	"You are the extraction stage of a learning application. You read one source — a teaching session, " +
	"rendered as the teacher's prose in reading order — and report what it teaches as findings for the " +
	"learner's global knowledge graph.";

export const NOTES_PREAMBLE =
	"You are the extraction stage of a learning application. You read one source — one chunk of the " +
	"author's personal notes, rendered in reading order — and report what it teaches as findings for the " +
	"learner's global knowledge graph.";

export const EXTRACTION_CORE = `THE GOAL: extract the source's REASONING CHAINS. The source teaches by making arguments — propositions
worked out from other propositions. Your deliverable is twofold: the propositions themselves, and every
argument the source makes. Each argument is a chain: premises that JOINTLY entail a conclusion.

The graph holds TRUE propositions about the world, nothing else — never anything about the learner, the
teacher, or this reading: not "the learner understands X" but X. A misconception enters as its TRUE
correction. Each statement is one self-contained reasoning step at full precision.

SUBMIT with one submit_findings call: every proposition with a short name and a precise statement, and
every chain as premises + conclusion. Premises within one chain are used TOGETHER — independent arguments
for the same conclusion are separate chains. A conclusion may reappear, by id, as a premise of later
chains. Include every field even when empty — "chains": [] when the source makes no arguments. The
tool's field descriptions carry the details.`;

export const EXTRACTOR_INSTRUCTIONS = `${SESSION_PREAMBLE}\n\n${EXTRACTION_CORE}`;
export const NOTES_EXTRACTOR_INSTRUCTIONS = `${NOTES_PREAMBLE}\n\n${EXTRACTION_CORE}`;

/** 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> {
	// Chains in (the wire), follows-edges out (the pipeline) — the
	// translation seam of the 2026-09-30 redesign. Validation runs at the
	// submit seam (error-back); expansion runs once, after a clean submit.
	return runSubmitAgent<FindingsSubmit>({
		role: "extractor",
		toolName: "submit_findings",
		toolDescription: "Submit the session findings.",
		parameters: findingsSchema,
		fill: fillFindings,
		validate: validateFindingsChains,
		systemPrompt: EXTRACTOR_INSTRUCTIONS,
		prompt,
		engine,
		signal,
		onStep,
		neverSubmitted: "the extractor never submitted findings",
		prepareArguments: healStringifiedArrays,
	}).then(asSessionFindings);
}

/** 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<FindingsSubmit>({
		role: "extractor",
		toolName: "submit_findings",
		toolDescription: "Submit the notes findings.",
		parameters: findingsSchema,
		fill: fillFindings,
		validate: validateFindingsChains,
		systemPrompt: NOTES_EXTRACTOR_INSTRUCTIONS,
		prompt,
		engine,
		signal,
		onStep,
		neverSubmitted: "the extractor never submitted findings",
		prepareArguments: healStringifiedArrays,
	}).then(asSessionFindings);
}

// ─── the auditor ───────────────────────────────────────────────────────────


const AUDITOR_INSTRUCTIONS = `You are the linking stage of a learning application's extraction pipeline. An extractor
has read a source (a teaching session, or the author's personal notes) and reported findings: propositions and
their reasoning chains. You NEVER see the source itself — the extractor is fully trusted, and you do not re-check
its work. Your job is IDENTITY: connecting those findings to the CURRENT knowledge graph, preparing the merge.

You explore the graph through your tools: list_sessions, list_propositions, and find (glob over names +
statements + aliases; each hit also says which session it was born in). The graph is SMALL and FINITE — a handful of
sessions — and your job is done as soon as you have SEEN the plausible candidates. FOLLOW THIS PROCEDURE:

1. GROUND yourself first: call list_sessions ONCE. This one call shows the entire space — every session that ever
   contributed propositions, and how many. Now you know the extent of what you are searching; nothing is hidden.
2. READ the plausibly-related sessions: for each session whose topic could overlap these findings, call
   list_propositions — every proposition born there, with full statement text. A few calls and you have read
   every realistic identity candidate in the graph.
3. find is your targeted sweep: use it for phrasings, synonyms, and neighboring ideas the session reads made you
   wonder about — its hits point back at sessions you can read whole.
4. SUBMIT. Once every plausibly-related session has been read (and stray finds come back empty), you are DONE.
   Do not re-search ground you have already seen; more patterns over the same nodes are wasted work.

IDENTITY — links, per proposition:
- For each proposition: does an EXISTING node — one you have now SEEN — already state this same proposition?
- If yes: add { "id": <its proposition id>, "linkTo": <that node's id> } to "links". The session's phrasings join
  its aliases and its provenance; the node keeps its existing name.
- If no: add nothing — a new node is created with the extractor's name as its label.
- EQUIVALENCE BIAS (owner ruling): paraphrases, rewordings, and compressions of the same claim MERGE. When two
  statements are the same claim at different precision or from different angles, LINK them. Only genuinely distinct
  propositions — different claims about the world — stay apart. You are NOT guarding against wrong merges; you are
  hunting for right ones.

Your deliverable is ONLY the identity verdicts: "links". NEVER restate the propositions —
the merger already has them. You NEVER create relations between propositions — they come from the
extractor alone, and the batch connects to the existing graph ONLY through the equivalences you identify. Your
entire deliverable is ONE submit_review call.

Include the "links" field even when empty — "links": [] when nothing links. NEVER omit it because it is empty.`;

/** The auditor — one shape for both front doors now (the notes schema was
 *  identical once attribution died). An agentic loop: graph-exploration
 *  tools plus the one submit call. */
function runAuditor(
	engine: PiEngine | undefined,
	prompt: string,
	signal: AbortSignal | undefined,
	onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void,
	tools: Record<string, AgentTool<any>>,
): Promise<AuditVerdicts> {
	return runSubmitAgent({
		role: "auditor",
		toolName: "submit_review",
		toolDescription: "Submit the identity verdicts ONLY (links) — never the propositions themselves.",
		parameters: auditSchema,
		fill: fillAudit,
		systemPrompt: AUDITOR_INSTRUCTIONS,
		prompt,
		engine,
		signal,
		onStep,
		neverSubmitted: "the auditor never submitted a review",
		extraTools: tools,
		maxTurns: 16,
		prepareArguments: healStringifiedArrays,
	});
}

// ─── the pipeline status registry ─────────────────────────────────────────────

/** The pipeline phases, in order — what the home status endpoint shows.
 *  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 job is actually waiting on.
 *  "queued" (owner, 2026-09-28): the global extract semaphore — 3
 *  concurrent extractor calls process-wide; the job is waiting for an
 *  extract slot, not for the model.
 *  "waiting" (owner 2026-09-28): the serialized-audit gate — the job's
 *  audit+merge critical section is queued behind another job's; the
 *  extractor is done but the auditor may not start until the graph is
 *  stable. */
export type ExtractionPhase = "queued" | "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). */
export interface RunningExtraction {
	/** The SOURCE id (chat id / notes src id) — the payload's identity and
	 *  the abort key. The registry KEY adds "#i" for chunk jobs (a
	 *  multi-chunk source owns several entries); this field stays the
	 *  source id so the feed's reveal logic and abort stay source-scoped. */
	chatId: string;
	/** PER-CHUNK JOBS (owner, 2026-09-28): which chunk of how many this
	 *  job is — present only when the source fanned out (n > 1). Drives
	 *  the feed row's "· i/N" suffix. */
	part?: { i: number; n: number };
	/** 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 /
	 *  context-window blobs read this through /api/graph/status). */
	usage: {
		extractor?: ExtractionStageUsage;
		auditor?: ExtractionStageUsage;
	};
	/** True-cancellation handle (the dock's abort button): one controller per
	 *  run, shared across phases; abort() cancels the in-flight model call. */
	controller: AbortController;
}

interface LhExtractionGlobal {
	__lhExtractionRuns?: {
		/** In-flight pipeline runs, keyed chatId (chat) / src id (notes). */
		running: Map<string, RunningExtraction>;
		/** The single-flight guard, shared by every pipeline instance. */
		inFlight: Set<string>;
		/** Last minted startedAt — kept monotonic so two runs starting in the
		 *  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 job at a time,
	 *  process-wide. globalThis for the same module-instance reason as
	 *  the run registries. */
	__lhAuditLock?: { tail: Promise<void> };
	/** The EXTRACT SEMAPHORE (owner, 2026-09-28): 3 concurrent extractor
	 *  calls process-wide — the provider-facing rate stance (the import
	 *  route's old 3-wide pool doctrine, now global). Per-chunk jobs fan
	 *  out in parallel but queue here. */
}

/** THE RUN REGISTRIES live on globalThis, not module level — the
 *  turn-liveness lesson: Next dev compiles each route entry as its own
 *  module instance, so a module-level Map is a different object in the
 *  /api/activity route than in the pipeline's chat-handler instance —
 *  the status routes would read an always-empty registry. One process,
 *  one truth — the same idiom as the turn, wake, and abort registries
 *  (moved 2026-09-19, the observability-accuracy wave; also makes the
 *  single-flight guard cross route instances). */
function runs(): NonNullable<LhExtractionGlobal["__lhExtractionRuns"]> {
	const g = globalThis as typeof globalThis & LhExtractionGlobal;
	return (g.__lhExtractionRuns ??= { running: new Map(), inFlight: new Set(), lastStartedAt: 0 });
}

/** In-flight pipeline runs (oldest first) — the /api/graph/status payload. */
export function extractionStatuses(): RunningExtraction[] {
	return [...runs().running.values()].sort((a, b) => a.startedAt - b.startedAt);
}

/** True cancellation of the in-flight run for a chat (the dock's abort
 *  button): the pending model call receives the signal — the provider
 *  cancels the request, spend stops — and the run ends like a failure:
 *  nothing merged, nothing checkpointed, partial usage still persisted.
 *  Returns false when no run is in flight (idempotent no-op). */
export function abortExtraction(chatId: string): boolean {
	// PER-CHUNK JOBS (owner, 2026-09-28): a source may own several registry
	// entries (one per chunk/batch) — aborting the source aborts EVERY
	// job of its fan-out.
	let found = false;
	for (const run of runs().running.values()) {
		if (run.chatId !== chatId) continue;
		run.controller.abort();
		found = true;
	}
	return found;
}
// ─── the pipeline ───────────────────────────────────────────────────────────

export interface ExtractionModels {
	/** Test seam: a scripted engine per stage — checked FIRST, ahead of the
	 *  engineFor chain (production wires no direct values). */
	extractor?: PiEngine;
	auditor?: PiEngine;
	/** Settings-resolved engine per role — the same slot treatment the
	 *  job runner got, read FRESH per stage call (a settings PUT between
	 *  stages lands). BYOK (owner, 2026-09-22): the chain (role override →
	 *  slot → NOTHING) may end EMPTY → undefined; the front doors refuse
	 *  with ModelsNotConfiguredError before any stage runs. */
	engineFor?: (role: ModelRole) => PiEngine | undefined;
	/** Key-presence gate (provider-key interlude, 2026-09-25): given a
	 *  stage engine's provider + the role's slot tier, the injected check
	 *  returns the refusal envelope when the provider's key is missing
	 *  (null = run). Same injection discipline as the job runner's seam
	 *  (the agent layer may not read the key store); absent in tests that
	 *  inject engines directly (test engines never owe a key). A keyless
	 *  stage refuses with ProviderKeyMissingError BEFORE any batch runs. */
	providerKeyMissing?: (provider: string, tier: ModelSlotName) => ProviderKeyMissing | null;
}

export interface ExtractionPipelineDeps extends GraphDeps, ExtractionModels {
	/** The chunk budget — the transcript-token share of the extractor's
	 *  context; batches fill to it (the trigger fires on the same number,
	 *  read fresh per turn by the composition root). OPTIONAL with the
	 *  factory default (4096) so test fixtures stay minimal; the server
	 *  callers wire the live effective settings (extraction-UX redesign,
	 *  2026-09-24). */
	getChunkTokens?: () => number;
}

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 dedupe by it). */
	importNotes(input: NotesImportInput): Promise<EmissionStats>;
}

/** pi's per-step usage, flattened to the persisted CallUsage fields
 *  (the money channel's capture shape, contracts/chat.ts). Exported for
 *  the cluster namer — the same capture shape (2026-09 graph cost
 *  metrics wave). Module scope: no closure deps, pure projection. */
export function callUsage(usage: Usage): CallUsage {
	return {
		inputTokens: usage.input,
		outputTokens: usage.output,
		cacheReadTokens: usage.cacheRead,
		cacheWriteTokens: usage.cacheWrite,
		// Optional in pi's Usage (providers that never emit reasoning
		// tokens leave it unset) — persisted as 0, never dropped.
		reasoningTokens: usage.reasoning ?? 0,
	};
}

export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline {
	const { chatStore, graphStore } = deps;
	// Stage-engine resolution: a direct test value wins; otherwise the
	// engineFor chain (role override → slot → nothing, resolved FRESH per
	// stage call server-side) — the job runner's exact treatment. undefined
	// = the BYOK chain came up empty; the front-door guards turn that into
	// an honest refusal.
	const stageEngine = (role: ModelRole, override?: PiEngine): PiEngine | undefined =>
		override ?? deps.engineFor?.(role);
	/** The per-need refusal guard: which slot tiers the pipeline's stages
	 *  found unset (test values count as configured — they ARE engines),
	 *  and whether a configured stage's provider key is gone (provider-key
	 *  interlude, 2026-09-25) — both refuse BEFORE any batch runs. */
	function refuseIfStagesUnready(): void {
		const stages: readonly [ModelRole, PiEngine | undefined][] = [
			["extractor", stageEngine("extractor", deps.extractor)],
			["auditor", stageEngine("auditor", deps.auditor)],
		];
		const missing = new Set<ModelSlotName>();
		for (const [role, engine] of stages) if (!engine) missing.add(ROLE_SLOT[role]);
		if (missing.size > 0) {
			console.info(`[extraction] refused — no model configured for: ${[...missing].join(", ")}`);
			throw new ModelsNotConfiguredError([...missing]);
		}
		// Key-presence parity: first offender surfaces (extractor before
		// auditor); fixing that key re-gates and the next run surfaces the
		// next offender, if any.
		for (const [role, engine] of stages) {
			const refusal = deps.providerKeyMissing?.(engine!.model.provider, ROLE_SLOT[role]);
			if (refusal) {
				console.info(`[extraction] refused — ${JSON.stringify(refusal)}`);
				throw new ProviderKeyMissingError(refusal.provider, refusal.slots);
			}
		}
	}
	// THE SHARED REGISTRIES (runs() above — globalThis): every pipeline
	// instance (route entries in dev, tests) sees the same runs, the same
	// single-flight guard. No module-level state left in this factory.
	const runningExtractions = runs().running;
	const inFlight = runs().inFlight;

	function setPhase(
		jobKey: string,
		phase: ExtractionPhase,
		kind: "chat" | "notes",
		label?: string,
		part?: { i: number; n: number },
	): void {
		const R = runs();
		const prev = runningExtractions.get(jobKey);
		if (!prev) R.lastStartedAt = Math.max(Date.now(), R.lastStartedAt + 1);
		runningExtractions.set(jobKey, {
			chatId: prev?.chatId ?? jobKey.split("#")[0]!,
			part: prev?.part ?? part, // phase changes never lose the part
			kind,
			label: prev?.label ?? label, // phase changes never lose the label
			phase,
			startedAt: prev?.startedAt ?? R.lastStartedAt,
			usage: prev?.usage ?? {}, // phase changes never lose usage
			controller: prev?.controller ?? new AbortController(),
		});
	}

	/** Accumulate one billed step into the run's registry entry (live). */
	function recordStep(jobKey: string, stage: "extractor" | "auditor", usage: CallUsage, model: string): void {
		const run = runningExtractions.get(jobKey);
		if (!run) return; // not running (tests driving stages directly)
		run.usage[stage] ??= { model, steps: [] };
		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<T>(jobKey: string, kind: "chat" | "notes", label: string | undefined, critical: () => Promise<T>): Promise<T> {
		setPhase(jobKey, "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<void>((resolve) => (release = resolve));
		const signal = runningExtractions.get(jobKey)?.controller.signal;
		let onAbort: (() => void) | undefined;
		const aborted = new Promise<never>((_, 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(jobKey: string, view: string): Promise<SessionFindings> {
		const signal = runningExtractions.get(jobKey)?.controller.signal;
		signal?.throwIfAborted(); // an abort during a sync gap ends the job here
		console.info(`[extraction ${jobKey}] extractor starting (${view.split("\n").length} markdown lines)`);
		const t0 = performance.now();

		const findings = await runExtractor(
			stageEngine("extractor", deps.extractor),
			view,
			signal,
			(usage, model, step, finishReason, tools) => {
				recordStep(jobKey, "extractor", callUsage(usage), model);
				console.info(`[extraction] extractor step ${step}: ${finishReason}, tools: ${tools}`);
			},
		);
		console.info(`[extraction ${jobKey}] extractor done in ${Math.round(performance.now() - t0)}ms`);
		return findings;
	}

	async function audit(jobKey: string, findings: SessionFindings): Promise<AuditVerdicts> {
		const signal = runningExtractions.get(jobKey)?.controller.signal;
		signal?.throwIfAborted(); // an abort during the extract→audit gap ends here
		const graph = getGraph(graphStore);
		console.info(
			`[extraction ${jobKey}] auditor starting (${findings.propositions.length} propositions, ` +
				`${graph.nodes.length} graph nodes)`,
		);
		const t0 = performance.now();

		// The auditor NEVER sees the source (owner 2026-09-27): findings
		// JSON + identity instructions, and the graph reached through
		// exploration tools instead of an injected dump.
		const auditResult = await runAuditor(
			stageEngine("auditor", deps.auditor),
			`Proposed findings from the extractor (the source itself is not available to you — the extractor is fully trusted; the propositions only — the reasoning chains are the merger's business, owner 2026-09-30):\n\n` +
				`${JSON.stringify({ propositions: findings.propositions }, null, 2)}\n\n` +
				`The current knowledge graph has ${graph.nodes.length} node(s). Explore it with your tools and link what links.`,
			signal,
			(usage, model, step, finishReason, tools) => {
				recordStep(jobKey, "auditor", callUsage(usage), model);
				console.info(`[extraction] auditor step ${step}: ${finishReason}, tools: ${tools}`);
			},
			auditorGraphTools({ chatStore, graphStore }),
		);
		console.info(
			`[extraction ${jobKey}] audit: ${findings.propositions.length} propositions through, ` +
				`${auditResult.links.length} linked to existing nodes`,
		);
		console.info(`[extraction ${jobKey}] auditor done in ${Math.round(performance.now() - t0)}ms`);
		return auditResult;
	}


	/** THE EXTRACT SLOT (unified 2026-09-29): extraction chunk jobs run
	 *  under the SHARED background-LLM semaphore — one process-wide pool
	 *  of 3 permits for extraction chunks AND practice's partitioner/
	 *  generators/graders (owner ruling; see bg-llm-semaphore.ts). The
	 *  job's registry-row signal kills a queued ticket (dead-ticket
	 *  handoff). The audit side needs no slot: audits already serialize
	 *  behind their own global lock. */
	async function withExtractSlot<T>(jobKey: string, critical: () => Promise<T>): Promise<T> {
		const signal = runningExtractions.get(jobKey)?.controller.signal;
		return withBgLlmSlot(signal, critical);
	}

		/** Append the run's usage record to the chat's extractions log — success
	 *  or failure: a failed run still spent money, and its partial usage is
	 *  exactly what the record should show (stats absent = never merged).
	 *  Idempotent per startedAt (the store dedupes). */
	function persistRunUsage(jobKey: string, durationMs: number, stats: EmissionStats | undefined): void {
		const run = runningExtractions.get(jobKey);
		if (!run) return; // not running (tests driving stages directly)
		// PER-CHUNK JOBS (owner, 2026-09-28): the registry KEY may carry a
		// "#i" suffix — the usage LEDGERS key by SOURCE (run.chatId): one
		// job's record per chunk, all appended to the source's log.
		// SESSION-DELETION GUARD (wave 6): a chat deleted mid-run has no
		// usage log to append to — the record's death is the user's choice
		// and its partial spend logs with it (appendSourceUsage below
		// already no-ops for a vanished source; the chat path guards here).
		if (run.kind === "chat" && !chatStore.getChat(run.chatId)) return;
		const entry: ExtractionRunUsage = {
			startedAt: run.startedAt,
			at: new Date().toISOString(),
			durationMs,
			stats,
			stages: {
				extractor: run.usage.extractor ?? { model: "unknown", steps: [] },
				auditor: run.usage.auditor ?? { model: "unknown", steps: [] },
			},
		};
		if (run.kind === "notes") graphStore.appendSourceUsage(run.chatId, entry);
		else chatStore.appendExtractionUsage(run.chatId, entry);
	}

	/** One chunk/batch JOB (per-chunk jobs, owner 2026-09-28): the registry
	 *  entry's lifecycle — minted with its part, usage persisted per job
	 *  (the ledger shows each chunk's cost), removed on any exit. The body
	 *  runs the full queued → extract → audit-locked merge lifecycle;
	 *  failure isolation lives at the fan-out site (one bad chunk never
	 *  kills its siblings). */
	async function runJob(
		source: { id: string; kind: "chat" | "notes"; label?: string; part?: { i: number; n: number } },
		body: (jobKey: string) => Promise<EmissionStats>,
	): Promise<EmissionStats> {
		const jobKey = source.part ? `${source.id}#${source.part.i}` : source.id;
		const t0 = performance.now();
		try {
			const stats = await body(jobKey);
			persistRunUsage(jobKey, Math.round(performance.now() - t0), stats);
			return stats;
		} catch (err) {
			persistRunUsage(jobKey, Math.round(performance.now() - t0), undefined);
			throw err;
		} finally {
			runningExtractions.delete(jobKey);
		}
	}

	/** The fan-out aggregation (per-chunk jobs, owner 2026-09-28): successes
	 *  count (earlier chunks keep their merges — checkpoints make aborts
	 *  cheap), aborts log and pass as partial stats, isolated failures log
	 *  per job and never throw — the source's run returns what actually
	 *  merged. */
	function aggregateJobs(sourceId: string, settled: PromiseSettledResult<EmissionStats>[]): EmissionStats {
		const stats: EmissionStats = { nodes: 0, chains: 0 };
		let aborted = false;
		let failed = 0;
		for (const result of settled) {
			if (result.status === "fulfilled") {
				stats.nodes += result.value.nodes;
				stats.chains! += result.value.chains ?? 0;
				continue;
			}
			const err = result.reason;
			if (err instanceof Error && err.name === "AbortError") aborted = true;
			else {
				failed++;
				console.error(`[extraction ${sourceId}] job failed (isolated):`, err);
			}
		}
		if (aborted)
			console.info(
				`[extraction ${sourceId}] aborted — merged chunks keep their coverage, the rest stay uncovered for the next run`,
			);
		if (failed > 0)
			console.error(`[extraction ${sourceId}] ${failed} job(s) failed; successful jobs still merged`);
		return stats;
	}

	async function runExtraction(chatId: string): Promise<EmissionStats> {
		if (inFlight.has(chatId)) return { nodes: 0, chains: 0 };
		inFlight.add(chatId);
		const t0 = performance.now();
		// COVERAGE (vision-extraction-coverage.md, 2026-09-15): extract ONLY
		// what the per-chat coverage ledger doesn't already hold — never the
		// whole session. Batches are token-budgeted runs of the oldest
		// uncovered messages; each merged batch checkpoints its messages'
		// coverage ATOMICALLY (inside the merger's step 4), so an abort or
		// error below loses at most the in-flight batch — every earlier
		// batch stays covered, the rest stays visibly uncovered.
		try {
			const record = chatStore.getChat(chatId);
			if (!record) throw new Error(`Chat not found: ${chatId}`);
			// BYOK per-need gate (owner, 2026-09-22) + key presence
			// (2026-09-25): the write channel needs extractor + auditor
			// engines AND their providers' keys; a refusal lands BEFORE any
			// stage runs (typed — the trigger skips, routes surface 503).
			refuseIfStagesUnready();
			const covered = coverageOf(record.messages, graphStore.readCoverage(), chatId);
			if (covered.uncovered.length === 0) {
				console.info(`[extraction ${chatId}] nothing uncovered — no-op`);
				return { nodes: 0, chains: 0 };
			}
			const chunkTokens = deps.getChunkTokens?.() ?? FACTORY_SETTINGS.graph.extractionChunkTokens;
			// The batcher's one rule (owner, Wave D): a tool-call message and
			// its result message(s) are INSEPARABLE — one chunk, covered
			// together, read together (core/coverage.ts batchRuns).
			const batches = batchRuns(covered.slots, covered.uncovered, chunkTokens);
			console.info(
				`[extraction ${chatId}] pipeline start — ${covered.uncoveredTokens} uncovered tok, ` +
					`${covered.uncoveredCount} messages, ${batches.length} batch(es) of ≤${chunkTokens} tok`,
			);
			// PER-CHUNK JOBS (owner, 2026-09-28): every batch is its own job —
			// extracts fan out behind the global 3-slot semaphore ("queued"),
			// audits serialize through the process-wide gate ("waiting"). A
			// batch that fails is ISOLATED (its messages stay uncovered —
			// honestly re-runnable); its siblings keep merging.
			const settled = await Promise.allSettled(
				batches.map((batch, i) => {
					const part = batches.length > 1 ? { i: i + 1, n: batches.length } : undefined;
					return runJob({ id: chatId, kind: "chat", part }, async (jobKey) => {
						// The batch's source view: TEACHER PROSE ONLY (owner,
						// 2026-09-28) — plain paragraphs, same selection as the
						// coverage fingerprints (messageViewLines parity).
						const view = proseView(
							batch.map((slot) => slot.message),
							record.title,
						);
						setPhase(jobKey, "queued", "chat", undefined, part);
						const findings = await withExtractSlot(jobKey, async () => {
							setPhase(jobKey, "extracting", "chat", undefined, part);
							return extract(jobKey, view);
						});
						// SERIALIZED AUDIT (owner 2026-09-28): audit + merge are the
						// graph-mutating critical section — one job at a time,
						// process-wide, so the auditor explores the graph as it
						// stands AFTER every earlier job's merge.
						return withAuditLock(jobKey, "chat", undefined, async () => {
							setPhase(jobKey, "auditing", "chat", undefined, part);
							const auditResult = await audit(jobKey, findings);
							setPhase(jobKey, "merging", "chat", undefined, part);
							// 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, findings, 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 })),
							});
							// The layout ledger evolves with the merge (sync — the
							// next snapshot fetch must already reflect the new nodes).
							settleGraphLayout(graphStore);
							console.info(
								`[extraction ${jobKey}] merged: ${batchStats.nodes} nodes, ${batchStats.chains ?? 0} chains`,
							);
							return batchStats;
						});
					});
				}),
			);
			const stats = aggregateJobs(chatId, settled);
			console.info(
				`[extraction ${chatId}] merged: ${stats.nodes} nodes, ${stats.chains ?? 0} chains ` +
					`(${Math.round(performance.now() - t0)}ms total, ${batches.length} batch job(s))`,
			);
			return stats;
		} catch (err) {
			// Pre-fan-out failures only (stage gate, coverage read): aborts
			// and job-level failures are absorbed at the fan-out site, per
			// job — merged batches stay covered, partial usage persists per
			// job, the aggregate returns what actually merged.
			console.error(`[extraction ${chatId}] pipeline failed (${Math.round(performance.now() - t0)}ms):`, err);
			throw err;
		} finally {
			inFlight.delete(chatId);
			runningExtractions.delete(chatId); // done or failed — no longer running
		}
	}

	// ── 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; 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}`;
	}

	/** 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(
		jobKey: string,
		input: { id: string; filePath: string; text: string; part?: { i: number; n: number } },
	): Promise<NotesFindings> {
		// jobKey (per-chunk jobs, owner 2026-09-28): the registry key —
		// "src#i" for chunk jobs — so the abort signal reaches the call.
		// input.id stays the SOURCE id for the view and store ledgers.
		const signal = runningExtractions.get(jobKey)?.controller.signal;
		signal?.throwIfAborted(); // an abort during a sync gap ends the job here
		const view = notesView(input);
		console.info(`[extraction ${jobKey}] notes extractor starting (${view.split("\n").length} source lines)`);
		const t0 = performance.now();
		const findings = await runNotesExtractor(
			stageEngine("extractor", deps.extractor),
			view,
			signal,
			(usage, model, step, finishReason, tools) => {
				recordStep(jobKey, "extractor", callUsage(usage), model);
				console.info(`[extraction] extractor step ${step}: ${finishReason}, tools: ${tools}`);
			},
		);
		console.info(`[extraction ${jobKey}] notes extractor done in ${Math.round(performance.now() - t0)}ms`);
		return findings;
	}

	// The notes auditor is the SAME identity-linker as the chat one (one
	// schema now that attribution is gone) — audit() above, verbatim.

	async function runNotesExtraction(input: { id: string; filePath: string; text: string; importedAt: string }): Promise<EmissionStats> {
		const { id } = input;
		if (inFlight.has(id)) return { nodes: 0, chains: 0 };
		inFlight.add(id);
		const t0 = performance.now();
		try {
			// BYOK per-need gate (owner, 2026-09-22) + key presence
			// (2026-09-25): same refusal as the chat front door — no
			// extractor/auditor engines, or a stage key gone, no run.
			refuseIfStagesUnready();
			// 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`,
			);
			// PER-CHUNK JOBS (owner, 2026-09-28): every chunk is its own job —
			// registry row, usage record, failure domain. Extracts fan out in
			// parallel behind the global 3-slot extract semaphore ("queued");
			// audits serialize through the process-wide gate exactly as before
			// ("waiting"). The part rides the registry entry (feed suffix
			// "· i/N") only when the file actually chunked.
			const settled = await Promise.allSettled(
				chunks.map((chunk, i) => {
					const part = chunks.length > 1 ? { i: i + 1, n: chunks.length } : undefined;
					return runJob({ id, kind: "notes", label: input.filePath, part }, async (jobKey) => {
						setPhase(jobKey, "queued", "notes", input.filePath, part);
						const findings = await withExtractSlot(jobKey, async () => {
							setPhase(jobKey, "extracting", "notes", input.filePath, part);
							return notesExtract(jobKey, { ...input, text: chunk, part });
						});
						// SERIALIZED AUDIT (owner, 2026-09-28): extracts run in
						// parallel, but audit + merge one at a time, process-wide —
						// the auditor explores the graph as it stands AFTER every
						// earlier job's merge. Reacquired per chunk so runs
						// interleave fairly.
						return withAuditLock(jobKey, "notes", input.filePath, async () => {
							setPhase(jobKey, "auditing", "notes", input.filePath, part);
							const auditResult = await audit(jobKey, findings);
							setPhase(jobKey, "merging", "notes", input.filePath, part);
							const merged = mergeFindings(
								id,
								findings,
								auditResult,
								{ chatStore, graphStore },
								{ foldedAt: input.importedAt },
							);
							// The layout ledger evolves with each merge — the next
							// snapshot fetch reflects every finished chunk.
							settleGraphLayout(graphStore);
							console.info(`[extraction ${jobKey}] merged: ${merged.nodes} nodes, ${merged.chains ?? 0} chains`);
							return merged;
						});
					});
				}),
			);
			const stats = aggregateJobs(id, settled);
			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;
		} 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. 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. */
	async function importNotes(input: NotesImportInput): Promise<EmissionStats> {
		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 };
		}
		const importedAt = new Date().toISOString();
		graphStore.upsertSource({ id, filePath: input.filePath, importedAt, contentFp });
		return runNotesExtraction({ id, ...input, importedAt });
	}

	return { runExtraction, importNotes };
}