Update PHASES.md: Phase 41 + Guard fixes

- Phase 41: ProfileType enum, per-type endpoints
- Guard: scrumaudit.py, fixed watcher.sh + pr-reviewer.md

Cargo.lock

@@ -46,6 +46,7 @@ dependencies = [
 name = "aibridge"
 version = "0.1.0"
 dependencies = [
+ "async-trait",
  "axum",
  "reqwest",
  "serde",
@@ -8919,6 +8920,7 @@ dependencies = [
  "object_store",
  "parquet 55.2.0",
  "queryd",
+ "reqwest",
  "serde",
  "serde_json",
  "sha2",

auditor/README.md

@@ -33,6 +33,13 @@ Defaults: polls every 90s, stops on `auditor.paused` file present.
 
 - `data/_auditor/state.json` — last-audited head SHA per PR
 - `data/_auditor/verdicts/{pr}-{sha}.json` — per-run verdict record
+- `data/_kb/audit_lessons.jsonl` — one row per block/warn finding,
+  path-agnostic signature for dedup. Tailed by kb_query on each audit
+  to surface recurring patterns (2+ distinct PRs with same signature
+  → info, 3-4 → warn, 5+ → block). This is how the auditor learns.
+- `data/_kb/scrum_reviews.jsonl` — scrum-master per-file reviews. If
+  a file in the current PR has been scrum-reviewed, kb_query surfaces
+  the review as a finding with the accepted model and attempt count.
 
 ## Where YOU edit
 

auditor/audit.ts

@@ -12,7 +12,8 @@
 // review — reviews have self-review restrictions on Gitea and the
 // auditor currently uses the same PAT as the PR author).
 
-import { readFile, writeFile, mkdir } from "node:fs/promises";
+import { readFile, writeFile, mkdir, appendFile } from "node:fs/promises";
+import { createHash } from "node:crypto";
 import { join } from "node:path";
 import type { PrSnapshot, Verdict, Finding } from "./types.ts";
 import { getPrDiff, postCommitStatus, postIssueComment } from "./gitea.ts";
@@ -24,6 +25,10 @@ import { runInferenceCheck } from "./checks/inference.ts";
 import { runKbCheck } from "./checks/kb_query.ts";
 
 const VERDICTS_DIR = "/home/profit/lakehouse/data/_auditor/verdicts";
+// Playbook for audit findings — one row per block/warn finding from a
+// verdict. kb_query tails this next audit and escalates recurrences.
+// Structured as JSONL so it's cheap to append and cheap to tail.
+const AUDIT_LESSONS_JSONL = "/home/profit/lakehouse/data/_kb/audit_lessons.jsonl";
 
 export interface AuditOptions {
   // Skip the cloud inference call (fast path for iteration). Default false.
@@ -51,8 +56,8 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
   const [staticFindings, dynamicFindings, inferenceFindings, kbFindings] = await Promise.all([
     runStaticCheck(diff),
     opts.skip_dynamic ? Promise.resolve(stubFinding("dynamic", "skipped by options")) : runDynamicCheck(),
-    opts.skip_inference ? Promise.resolve(stubFinding("inference", "skipped by options")) : runInferenceCheck(claims, diff),
-    runKbCheck(claims),
+    opts.skip_inference ? Promise.resolve(stubFinding("inference", "skipped by options")) : runInferenceCheck(claims, diff, { pr_number: pr.number, head_sha: pr.head_sha }),
+    runKbCheck(claims, pr.files.map(f => f.path)),
   ]);
 
   const allFindings: Finding[] = [
@@ -72,6 +77,7 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
     claims_strong: claims.filter(c => c.strength === "strong").length,
     claims_moderate: claims.filter(c => c.strength === "moderate").length,
     claims_weak: claims.filter(c => c.strength === "weak").length,
+    claims_empirical: claims.filter(c => c.strength === "empirical").length,
     claims_total: claims.length,
     diff_bytes: diff.length,
   };
@@ -80,6 +86,15 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
 
   await persistVerdict(verdict);
 
+  // Feedback loop — every block/warn finding becomes a row in
+  // audit_lessons.jsonl, dedup-keyed by (check, normalized-summary).
+  // The next audit's kb_query reads these and escalates recurring
+  // findings so we don't lose the "this pattern has been flagged
+  // before" signal across runs. Fire-and-forget; failure here must
+  // not break the audit.
+  appendAuditLessons(verdict).catch(e =>
+    console.error(`[audit] audit_lessons append failed: ${(e as Error).message}`));
+
   if (!opts.dry_run) {
     await postToGitea(verdict);
   }
@@ -87,6 +102,42 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
   return verdict;
 }
 
+// Normalizes a finding summary for dedup: strips path-specific tails
+// ("in path/to/file.ts" → "in <file>"), line numbers, and long
+// commit-hash snippets. The goal is: the SAME class of finding on
+// DIFFERENT files should share a signature, so we can measure
+// "this pattern keeps showing up."
+function normalizedSignature(f: Finding): string {
+  const summary = String(f.summary)
+    .replace(/\bin\s+\S+\.(ts|rs|js|py|md)\b/gi, "in <file>")
+    .replace(/:\+?\d+\b/g, ":<line>")
+    .replace(/[0-9a-f]{8,}/gi, "<hash>")
+    .replace(/\s+/g, " ")
+    .trim()
+    .slice(0, 240);
+  const src = `${f.check}::${f.severity}::${summary}`;
+  return createHash("sha256").update(src).digest("hex").slice(0, 16);
+}
+
+async function appendAuditLessons(v: Verdict): Promise<void> {
+  const actionable = v.findings.filter(f => f.severity === "block" || f.severity === "warn");
+  if (actionable.length === 0) return;
+  await mkdir(join(AUDIT_LESSONS_JSONL, ".."), { recursive: true });
+  const rows: string[] = [];
+  for (const f of actionable) {
+    rows.push(JSON.stringify({
+      signature: normalizedSignature(f),
+      check: f.check,
+      severity: f.severity,
+      summary: f.summary,
+      pr_number: v.pr_number,
+      head_sha: v.head_sha,
+      audited_at: v.audited_at,
+    }));
+  }
+  await appendFile(AUDIT_LESSONS_JSONL, rows.join("\n") + "\n");
+}
+
 async function persistVerdict(v: Verdict): Promise<void> {
   await mkdir(VERDICTS_DIR, { recursive: true });
   const filename = `${v.pr_number}-${v.head_sha.slice(0, 12)}.json`;

auditor/audit_one.ts

@@ -0,0 +1,68 @@
+// One-shot dry-run audit of a single PR. Useful for verifying check
+// behavior (kb_query scrum surfacing, inference prompts, etc.) without
+// posting to Gitea. Does NOT touch state.json and does NOT post
+// commit status or PR comments.
+//
+// Run: bun run auditor/audit_one.ts <pr-number>
+
+import { getPrSnapshot } from "./gitea.ts";
+import { auditPr } from "./audit.ts";
+
+async function main() {
+  const prNumRaw = process.argv[2];
+  if (!prNumRaw) {
+    console.error("usage: bun run auditor/audit_one.ts <pr-number>");
+    process.exit(2);
+  }
+  const prNum = Number(prNumRaw);
+  if (!Number.isFinite(prNum)) {
+    console.error(`invalid PR number: ${prNumRaw}`);
+    process.exit(2);
+  }
+
+  console.log(`[audit_one] fetching PR #${prNum}...`);
+  const pr = await getPrSnapshot(prNum);
+  console.log(`[audit_one] PR #${pr.number}: "${pr.title}" (head=${pr.head_sha.slice(0, 12)})`);
+  console.log(`[audit_one] files in diff: ${pr.files.length}`);
+  for (const f of pr.files) console.log(`  - ${f.path} (+${f.additions}/-${f.deletions})`);
+  console.log("");
+
+  const verdict = await auditPr(pr, {
+    dry_run: true,                      // no Gitea posting
+    skip_dynamic: true,                  // don't run fixture
+    skip_inference: process.env.LH_AUDITOR_SKIP_INFERENCE === "1",
+  });
+
+  console.log("\n═══ VERDICT ═══");
+  console.log(`overall:  ${verdict.overall}`);
+  console.log(`one-liner: ${verdict.one_liner}`);
+  console.log(`findings:  total=${verdict.metrics.findings_total} block=${verdict.metrics.findings_block} warn=${verdict.metrics.findings_warn} info=${verdict.metrics.findings_info}`);
+  console.log("");
+
+  // Print findings, highlighting kb_query scrum surfacing
+  const byCheck: Record<string, typeof verdict.findings> = {};
+  for (const f of verdict.findings) (byCheck[f.check] ||= []).push(f);
+
+  for (const [check, findings] of Object.entries(byCheck)) {
+    console.log(`── ${check} (${findings.length}) ──`);
+    for (const f of findings) {
+      const tag = f.severity === "block" ? "🛑" : f.severity === "warn" ? "⚠️ " : "ℹ️ ";
+      console.log(`  ${tag} [${f.severity}] ${f.summary}`);
+      if (f.summary.includes("scrum-master")) {
+        for (const e of f.evidence) {
+          console.log(`      → ${e.slice(0, 200)}`);
+        }
+      }
+    }
+  }
+
+  const scrumFindings = verdict.findings.filter(f => f.summary.includes("scrum-master"));
+  console.log("");
+  console.log(`═══ SCRUM WIRE CHECK: ${scrumFindings.length} scrum-master findings surfaced by kb_query ═══`);
+  if (scrumFindings.length === 0) {
+    console.log("  (none — either no matching scrum_reviews.jsonl rows, or files didn't match PR diff)");
+  }
+  process.exit(0);
+}
+
+main().catch(e => { console.error("[audit_one] fatal:", e); process.exit(1); });

auditor/checks/inference.ts

@@ -13,17 +13,55 @@
 // with a 15KB diff + claim list).
 
 import type { Claim, Finding } from "../types.ts";
+import { Glob } from "bun";
+import { readFile, mkdir, appendFile } from "node:fs/promises";
+import { extractFacts } from "../fact_extractor.ts";
 
 const GATEWAY = process.env.LH_GATEWAY_URL ?? "http://localhost:3100";
 const MODEL = process.env.LH_AUDITOR_REVIEW_MODEL ?? "gpt-oss:120b";
+// Tie-breaker for claims where the N=3 consensus produces a 1-1-1
+// split (genuinely borderline). Different architecture from the
+// primary reviewer (gpt-oss) so the tie-break isn't correlated with
+// the original disagreement. qwen3-coder:480b is a newer coding
+// specialist at 480B params, well-suited to PR-diff claim verification
+// and distinct in training lineage from gpt-oss.
+const TIEBREAKER_MODEL = process.env.LH_AUDITOR_TIEBREAKER_MODEL ?? "qwen3-coder:480b";
+const N_CONSENSUS = Number(process.env.LH_AUDITOR_CONSENSUS_N ?? 3);
+const AUDIT_DISCREPANCIES_JSONL = "/home/profit/lakehouse/data/_kb/audit_discrepancies.jsonl";
 // 40KB comfortably fits gpt-oss:120b's context. PR #1 (~39KB) was
 // previously truncated at 15KB causing the reviewer to miss later
 // files (gitea.ts, policy.ts) and flag "no Gitea client present" as a
 // block finding when the file was simply outside the truncation window.
+//
+// Above this threshold we curate via tree-split rather than truncate,
+// following the scrum_master pattern: shard the diff, summarize each
+// shard against the claim-verification task, merge into a compact
+// scratchpad, then ask the cloud to verify claims against the
+// scratchpad. This gives the cloud full-PR fidelity without bursting
+// its context window (observed failure mode: empty response or
+// unparseable output when prompt exceeds model's comfortable range).
 const MAX_DIFF_CHARS = 40000;
+// Tree-split kicks in above this. 30KB is below MAX_DIFF_CHARS so we
+// curate BEFORE truncation would happen — never lose signal to a hard
+// cut. Shard size is chosen so ~10 shards cover PR #8-size diffs in a
+// reasonable round-trip budget.
+const CURATION_THRESHOLD = 30000;
+const DIFF_SHARD_SIZE = 4500;
 const CALL_TIMEOUT_MS = 120_000;
+const REPO_ROOT = "/home/profit/lakehouse";
 
-export async function runInferenceCheck(claims: Claim[], diff: string): Promise<Finding[]> {
+export interface InferenceContext {
+  pr_number: number;
+  head_sha: string;
+}
+
+const AUDIT_FACTS_JSONL = "/home/profit/lakehouse/data/_kb/audit_facts.jsonl";
+
+export async function runInferenceCheck(
+  claims: Claim[],
+  diff: string,
+  ctx?: InferenceContext,
+): Promise<Finding[]> {
   if (claims.length === 0) {
     return [{
       check: "inference",
@@ -33,9 +71,41 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
     }];
   }
 
-  const truncated = diff.length > MAX_DIFF_CHARS
-    ? diff.slice(0, MAX_DIFF_CHARS) + `\n...[${diff.length - MAX_DIFF_CHARS} more chars truncated]`
-    : diff;
+  // Empirical claims (runtime metrics / observed outcomes) can't be
+  // verified from the diff. Drop them from the cloud prompt so the
+  // reviewer doesn't chase ghosts. A future `runtime_evidence` check
+  // can validate these against data/_kb/*/summary.json outputs.
+  const verifiable = claims.filter(c => c.strength !== "empirical");
+  const empiricalCount = claims.length - verifiable.length;
+  if (verifiable.length === 0) {
+    return [{
+      check: "inference",
+      severity: "info",
+      summary: `all ${claims.length} claims are empirical (runtime metrics) — skipping cloud inference`,
+      evidence: [`empirical claims can't be verified from a static diff; needs runtime-evidence check`],
+    }];
+  }
+
+  // Diff source for the cloud prompt — either the raw diff (small
+  // enough to fit), or a tree-split scratchpad (curation layer). We
+  // prefer curation to truncation: truncation silently drops files
+  // past the window; curation summarizes them so the cloud still sees
+  // what changed, just densified.
+  let diffForPrompt: string;
+  let curationNote = "";
+  if (diff.length > CURATION_THRESHOLD) {
+    const ts = await treeSplitDiff(diff, verifiable);
+    diffForPrompt = ts.scratchpad;
+    curationNote = ` (curated: ${diff.length} chars → ${ts.shards} shards → scratchpad ${ts.scratchpad.length} chars)`;
+  } else {
+    diffForPrompt = diff;
+  }
+  // Belt-and-suspenders truncation — even a tree-split scratchpad
+  // shouldn't exceed MAX_DIFF_CHARS in practice, but guard anyway so
+  // pathological inputs can't burst the prompt.
+  const truncated = diffForPrompt.length > MAX_DIFF_CHARS
+    ? diffForPrompt.slice(0, MAX_DIFF_CHARS) + `\n...[${diffForPrompt.length - MAX_DIFF_CHARS} more chars truncated]`
+    : diffForPrompt;
 
   // Build the reviewer prompt in the same shape as run_codereview's
   // review stage (llm_team_ui.py:10950), adapted for claim verification:
@@ -43,6 +113,30 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
   //   "Code: ..."
   //   "Review: bugs/security/perf/style/edge. Provide corrected code."
   // We add: claim list upfront + ask for structured JSON verdict.
+  //
+  // When the diff was curated (tree-split scratchpad), we add an
+  // explicit anti-false-positive instruction: the scratchpad is a
+  // distillation, not the full source, so absence-from-scratchpad is
+  // NOT evidence of absence-from-diff. Mirrors the fix we made in
+  // scrum_master's review prompt for the same class of error.
+  const isCurated = curationNote.length > 0;
+  const curationGuard = isCurated
+    ? [
+        "",
+        "CRITICAL: the 'Diff' below is a curated multi-shard scratchpad,",
+        "NOT the full raw diff. The scratchpad distills each shard down",
+        "to facts useful for claim verification and drops the rest.",
+        "DO NOT flag a function/field/feature as 'missing' or 'not",
+        "implemented' based solely on its absence from the scratchpad —",
+        "absence in a distillation is NOT evidence of absence in the",
+        "actual diff. Only judge a claim NOT BACKED when the scratchpad",
+        "DIRECTLY contradicts it (e.g. scratchpad shows the function was",
+        "added empty, or shows the claimed code path is a stub).",
+        "Skip the unflagged_gaps section entirely when operating on a",
+        "curated scratchpad — you can't reliably detect gaps from a",
+        "distillation, and false positives there are worse than misses.",
+      ].join("\n")
+    : "";
   const systemMsg = [
     "You review pull-request diffs against the author's own ship-claims.",
     "For each claim, decide: is it backed by actual code in the diff, or is",
@@ -56,6 +150,7 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
     "  - the claim claims integration but the integration point is a stub",
     "  - the diff contains unimplemented!() / todo!() / TODO comments",
     "  - the claim says 'works end-to-end' but the diff has no end-to-end test",
+    curationGuard,
     "",
     "Respond with strict JSON only. No prose before or after. Shape:",
     "{",
@@ -70,7 +165,7 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
 
   const userMsg = [
     `Ship-claims the author made (numbered 0..N-1):`,
-    claims.map((c, i) => `  ${i}. [${c.strength}] "${c.text}" at ${c.location}`).join("\n"),
+    verifiable.map((c, i) => `  ${i}. [${c.strength}] "${c.text}" at ${c.location}`).join("\n"),
     "",
     `Diff:`,
     "```",
@@ -82,79 +177,131 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
     `Strict JSON only, matching the shape described. No prose outside JSON.`,
   ].join("\n");
 
-  let resp: Response;
-  try {
-    resp = await fetch(`${GATEWAY}/v1/chat`, {
-      method: "POST",
-      headers: { "content-type": "application/json" },
-      body: JSON.stringify({
-        provider: "ollama_cloud",
-        model: MODEL,
-        messages: [
-          { role: "system", content: systemMsg },
-          { role: "user", content: userMsg },
-        ],
-        max_tokens: 3000,
-        temperature: 0.2,
-        think: true, // T3 overseer should reason — JSON shape is still required
-      }),
-      signal: AbortSignal.timeout(CALL_TIMEOUT_MS),
-    });
-  } catch (e) {
-    // Cloud unreachable → soft-fail. Don't block a PR because the
-    // reviewer model is down. Static + dynamic + kb still run.
+  // N=3 consensus — run the primary reviewer in parallel, collect
+  // all three parsed responses, majority-vote per claim. Parallel
+  // (Promise.all) because each call is ~20-30s and they're independent;
+  // wall-clock stays ~same as single call, cost 3x tokens. Empirical
+  // justification: in 3-run determinism tests, 7/8 findings were
+  // stable but 1 flipped across runs — majority vote stabilizes the
+  // flipping class without losing the stable signal.
+  const primaryRuns = await Promise.all(
+    Array.from({ length: N_CONSENSUS }, () =>
+      runCloudInference(systemMsg, userMsg, MODEL)),
+  );
+
+  const parsedRuns = primaryRuns.filter(r => r.parsed !== null);
+  if (parsedRuns.length === 0) {
+    // All N calls failed. Surface the first-run diagnostic so the
+    // operator sees *why* (unreachable / non-200 / unparseable).
+    const first = primaryRuns[0];
     return [{
       check: "inference",
       severity: "info",
-      summary: "cloud inference unreachable — skipped",
-      evidence: [`fetch failed: ${(e as Error).message.slice(0, 180)}`],
-    }];
-  }
-
-  if (!resp.ok) {
-    return [{
-      check: "inference",
-      severity: "info",
-      summary: `cloud inference returned ${resp.status} — skipped`,
-      evidence: [`body: ${(await resp.text()).slice(0, 200)}`],
-    }];
-  }
-
-  const body: any = await resp.json();
-  const content: string = body?.choices?.[0]?.message?.content ?? "";
-  const usage = body?.usage ?? {};
-
-  const parsed = extractJson(content);
-  if (!parsed) {
-    return [{
-      check: "inference",
-      severity: "info",
-      summary: "cloud returned unparseable output — skipped",
+      summary: `cloud inference all ${N_CONSENSUS} consensus runs failed — ${first.error ?? "unknown"}`,
       evidence: [
-        `head: ${content.slice(0, 200)}`,
-        `tokens: ${usage.total_tokens ?? "?"}`,
+        `first-run diagnostic: ${first.diagnostic ?? "(none)"}`,
+        `successful runs: 0 / ${N_CONSENSUS}`,
       ],
     }];
   }
 
+  // Aggregate votes per claim_idx.
+  interface Votes { trues: number; falses: number; evidences: string[] }
+  const votesByClaim = new Map<number, Votes>();
+  const unflaggedByRun: any[][] = [];
+  let totalTokens = 0;
+  for (const run of parsedRuns) {
+    totalTokens += run.tokens;
+    unflaggedByRun.push(Array.isArray(run.parsed?.unflagged_gaps) ? run.parsed.unflagged_gaps : []);
+    for (const v of run.parsed?.claim_verdicts ?? []) {
+      const idx = Number(v?.claim_idx);
+      if (!Number.isFinite(idx)) continue;
+      const rec = votesByClaim.get(idx) ?? { trues: 0, falses: 0, evidences: [] };
+      if (v.backed === false) {
+        rec.falses++;
+        rec.evidences.push(String(v.evidence ?? ""));
+      } else if (v.backed === true) {
+        rec.trues++;
+      }
+      votesByClaim.set(idx, rec);
+    }
+  }
+
   const findings: Finding[] = [];
 
-  // One summary info finding so the verdict layer knows the check ran.
+  // Summary finding so the verdict layer knows the check ran.
   findings.push({
     check: "inference",
     severity: "info",
-    summary: `cloud review completed (model=${MODEL}, tokens=${usage.total_tokens ?? "?"})`,
+    summary: `cloud review completed (model=${MODEL}, consensus=${parsedRuns.length}/${N_CONSENSUS}, tokens=${totalTokens})${curationNote}`,
     evidence: [
-      `claim_verdicts: ${parsed.claim_verdicts?.length ?? 0}, unflagged_gaps: ${parsed.unflagged_gaps?.length ?? 0}`,
+      `claims voted: ${votesByClaim.size}`,
+      `parsed runs: ${parsedRuns.length} / ${N_CONSENSUS}`,
     ],
   });
 
-  for (const v of parsed.claim_verdicts ?? []) {
-    if (v?.backed === false) {
-      const idx = typeof v.claim_idx === "number" ? v.claim_idx : -1;
-      const claim = claims[idx];
-      if (!claim) continue;
-      // Strong+unbacked = BLOCK. That's the whole point of the auditor.
+  // Per-claim majority vote; tie-break if no majority.
+  const discrepancies: Array<{
+    claim_idx: number;
+    claim_text: string;
+    votes: { trues: number; falses: number };
+    resolution: "majority_backed" | "majority_not_backed" | "tiebreaker_backed" | "tiebreaker_not_backed" | "unresolved";
+    tiebreaker_model?: string;
+  }> = [];
+
+  for (const [idx, votes] of votesByClaim) {
+    const claim = verifiable[idx];
+    if (!claim) continue;
+    const totalVotes = votes.trues + votes.falses;
+    let notBacked: boolean | null = null;
+    let resolution: typeof discrepancies[number]["resolution"] = "majority_backed";
+    let evidenceText = "";
+    let tbModel: string | undefined;
+
+    if (votes.falses > votes.trues) {
+      notBacked = true;
+      resolution = "majority_not_backed";
+      evidenceText = votes.evidences[0] ?? "(no reason given)";
+    } else if (votes.trues > votes.falses) {
+      notBacked = false;
+      resolution = "majority_backed";
+    } else {
+      // Tie. Run tie-breaker with a different-architecture model.
+      const tb = await runCloudInference(systemMsg, userMsg, TIEBREAKER_MODEL);
+      if (tb.parsed) {
+        const tv = (tb.parsed.claim_verdicts ?? []).find((v: any) => Number(v?.claim_idx) === idx);
+        if (tv?.backed === false) {
+          notBacked = true;
+          resolution = "tiebreaker_not_backed";
+          evidenceText = `(tie-breaker ${TIEBREAKER_MODEL}) ${String(tv.evidence ?? "")}`;
+          tbModel = TIEBREAKER_MODEL;
+        } else if (tv?.backed === true) {
+          notBacked = false;
+          resolution = "tiebreaker_backed";
+          tbModel = TIEBREAKER_MODEL;
+        } else {
+          resolution = "unresolved";
+        }
+      } else {
+        resolution = "unresolved";
+      }
+    }
+
+    // Log every case where the N runs disagreed — discrepancies are
+    // signal, not noise. Separate from audit_lessons.jsonl because
+    // they're about the *auditor's* quality, not the PR's quality.
+    const disagreed = totalVotes >= 2 && votes.trues > 0 && votes.falses > 0;
+    if (disagreed || resolution.startsWith("tiebreaker") || resolution === "unresolved") {
+      discrepancies.push({
+        claim_idx: idx,
+        claim_text: claim.text,
+        votes: { trues: votes.trues, falses: votes.falses },
+        resolution,
+        tiebreaker_model: tbModel,
+      });
+    }
+
+    if (notBacked === true) {
       const sev: Finding["severity"] = claim.strength === "strong" ? "block"
         : claim.strength === "moderate" ? "warn"
         : "info";
@@ -165,24 +312,337 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
         summary: `cloud: claim not backed — "${claim.text.slice(0, 100)}"`,
         evidence: [
           `at ${claim.location}`,
-          `cloud reason: ${String(v.evidence ?? "no reason given").slice(0, 200)}`,
+          `consensus: ${votes.falses}/${totalVotes} not-backed (resolution: ${resolution})`,
+          `cloud reason: ${evidenceText.slice(0, 200)}`,
         ],
       });
     }
   }
 
-  for (const g of parsed.unflagged_gaps ?? []) {
+  // Persist discrepancies so we can measure consensus drift over time.
+  if (discrepancies.length > 0 && ctx) {
+    persistDiscrepancies(ctx, discrepancies).catch(e =>
+      console.error(`[inference] discrepancy log failed: ${(e as Error).message}`));
+  }
+
+  // Use first run's parsed for downstream unflagged_gaps processing.
+  const parsed = parsedRuns[0].parsed;
+
+  // Route the curated scratchpad through llm_team's extract-facts
+  // pipeline when we have (a) a curated scratchpad (best signal about
+  // what the PR actually changed) and (b) PR context to scope facts.
+  // AWAITED (not fire-and-forget) so CLI callers like audit_one.ts
+  // don't exit before extraction lands; the systemd poller has plenty
+  // of headroom (90s cycle vs ~15s extraction). A failure inside
+  // extractAndPersistFacts is caught + logged but never throws.
+  if (isCurated && ctx && process.env.LH_AUDITOR_SKIP_EXTRACT !== "1") {
+    try {
+      await extractAndPersistFacts(diffForPrompt, ctx);
+    } catch (e) {
+      console.error(`[inference] fact extraction failed: ${(e as Error).message}`);
+    }
+  }
+
+  // Belt-and-suspenders: when operating on a curated scratchpad, drop
+  // the unflagged_gaps section entirely. The distillation can't
+  // reliably ground gap-detection, and false positives are worse than
+  // misses for this signal class. The systemMsg already asks the
+  // cloud to skip this section when curated — but the model may still
+  // emit it, so we filter here too.
+  const gapsToEmit = isCurated ? [] : (parsed.unflagged_gaps ?? []);
+  for (const g of gapsToEmit) {
+    const summary = String(g?.summary ?? "?");
+    const location = String(g?.location ?? "?");
+    // False-positive guard — when the cloud says "X not defined in this
+    // diff" or "missing implementation of X", the cloud may just mean
+    // "X is not in the added lines," not "X doesn't exist in the repo."
+    // Extract candidate symbol names and grep the repo. If any symbol
+    // is defined elsewhere, drop the finding — it's a known-symbol
+    // reference, not a placeholder.
+    if (/not\s+defined|missing\s+implementation|never\s+referenced\s+or\s+integrated/i.test(summary)) {
+      const symbols = extractSymbols(summary);
+      if (symbols.length > 0) {
+        const resolved = await symbolsExistInRepo(symbols);
+        if (resolved.length === symbols.length) {
+          // Every named symbol exists somewhere in the repo — silent drop.
+          continue;
+        }
+        if (resolved.length > 0) {
+          // Partially resolved — demote to info with a note.
+          findings.push({
+            check: "inference",
+            severity: "info",
+            summary: `cloud gap partially resolved by repo grep: ${summary.slice(0, 120)}`,
+            evidence: [
+              `location: ${location.slice(0, 140)}`,
+              `resolved via grep: ${resolved.join(",")}`,
+              `unresolved: ${symbols.filter(s => !resolved.includes(s)).join(",")}`,
+            ],
+          });
+          continue;
+        }
+      }
+    }
     findings.push({
       check: "inference",
       severity: "warn",
-      summary: `cloud-flagged gap not in any claim: ${String(g?.summary ?? "?").slice(0, 120)}`,
-      evidence: [`location: ${String(g?.location ?? "?").slice(0, 140)}`],
+      summary: `cloud-flagged gap not in any claim: ${summary.slice(0, 120)}`,
+      evidence: [`location: ${location.slice(0, 140)}`],
     });
   }
 
   return findings;
 }
 
+// Single cloud call — the consensus loop calls this N times in
+// parallel. Returns the parsed JSON shape + token usage + any error
+// diagnostic. NEVER throws; the consensus aggregator handles partial
+// failures by dropping non-parsed runs from the vote.
+interface CloudRunResult {
+  parsed: any | null;
+  tokens: number;
+  error?: string;       // "unreachable" | "non_200" | "unparseable"
+  diagnostic?: string;  // first 200 chars for debugging
+  model: string;
+}
+
+async function runCloudInference(systemMsg: string, userMsg: string, model: string): Promise<CloudRunResult> {
+  let resp: Response;
+  try {
+    resp = await fetch(`${GATEWAY}/v1/chat`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        provider: "ollama_cloud",
+        model,
+        messages: [
+          { role: "system", content: systemMsg },
+          { role: "user", content: userMsg },
+        ],
+        // temp=0 (greedy) + think=true. think=true is required for
+        // gpt-oss:120b — without it the model returns empty content
+        // on large prompts. Variance from the think trace is observed
+        // in practice, which is why we use N=3 consensus, not single-
+        // call determinism.
+        max_tokens: 3000,
+        temperature: 0,
+        think: true,
+      }),
+      signal: AbortSignal.timeout(CALL_TIMEOUT_MS),
+    });
+  } catch (e) {
+    return { parsed: null, tokens: 0, error: "unreachable", diagnostic: (e as Error).message.slice(0, 200), model };
+  }
+  if (!resp.ok) {
+    return { parsed: null, tokens: 0, error: "non_200", diagnostic: `${resp.status}: ${(await resp.text()).slice(0, 160)}`, model };
+  }
+  let body: any;
+  try { body = await resp.json(); }
+  catch (e) { return { parsed: null, tokens: 0, error: "unparseable", diagnostic: (e as Error).message, model }; }
+  const content: string = body?.choices?.[0]?.message?.content ?? "";
+  const tokens: number = body?.usage?.total_tokens ?? 0;
+  const parsed = extractJson(content);
+  if (!parsed) {
+    return { parsed: null, tokens, error: "unparseable", diagnostic: content.slice(0, 200), model };
+  }
+  return { parsed, tokens, model };
+}
+
+async function persistDiscrepancies(ctx: InferenceContext, discrepancies: any[]): Promise<void> {
+  await mkdir("/home/profit/lakehouse/data/_kb", { recursive: true });
+  const rows = discrepancies.map(d => JSON.stringify({
+    pr_number: ctx.pr_number,
+    head_sha: ctx.head_sha,
+    logged_at: new Date().toISOString(),
+    ...d,
+  }));
+  await appendFile(AUDIT_DISCREPANCIES_JSONL, rows.join("\n") + "\n");
+}
+
+// Extract structured knowledge from the curated scratchpad and append
+// to data/_kb/audit_facts.jsonl — one row per extract run, keyed by
+// PR number + head SHA for scope tracking. kb_query tails this next
+// audit to surface recurring entities/relationships across PRs.
+async function extractAndPersistFacts(scratchpad: string, ctx: InferenceContext): Promise<void> {
+  const ex = await extractFacts(scratchpad);
+  if (ex.error && ex.entities.length === 0 && ex.facts.length === 0) {
+    // Full failure — log but don't write an empty row.
+    console.error(`[inference] extractFacts skipped row: ${ex.error}`);
+    return;
+  }
+  const row = {
+    pr_number: ctx.pr_number,
+    head_sha: ctx.head_sha,
+    extracted_at: ex.extracted_at,
+    extractor: ex.extractor_model,
+    verifier: ex.verifier_model,
+    llm_team_run_id: ex.llm_team_run_id ?? null,
+    facts: ex.facts,
+    entities: ex.entities,
+    relationships: ex.relationships,
+    verification_preview: ex.verification.slice(0, 400),
+    verifier_verdicts: ex.verifier_verdicts,
+    facts_dropped_by_verifier: ex.facts_dropped_by_verifier ?? 0,
+    schema_version: 2,
+    source: "audit_inference",
+  };
+  await mkdir("/home/profit/lakehouse/data/_kb", { recursive: true });
+  await appendFile(AUDIT_FACTS_JSONL, JSON.stringify(row) + "\n");
+}
+
+// Curation via tree-split — ports the scrum_master pattern into the
+// inference check. Shards the raw diff into DIFF_SHARD_SIZE chunks,
+// summarizes each shard *against the claim-verification task* so the
+// summary preserves exactly what the cloud needs to judge claims
+// (function signatures, struct fields, deletions, new files), drops
+// everything else. Merges into a compact scratchpad.
+//
+// Cost: N cloud calls for the shard summaries + 1 cloud call for the
+// final verification = N+1 calls instead of 1. Mitigation: shards run
+// serially (not parallel) to keep gateway load bounded; summary calls
+// use max_tokens=400 so they're fast (~2s each on gpt-oss:120b).
+//
+// Determinism: each shard summary call uses temp=0 + think=true (same
+// as the top-level inference call), so identical input yields
+// identical scratchpad. The final verification call then sees a
+// stable scratchpad, giving stable verdicts.
+async function treeSplitDiff(
+  fullDiff: string,
+  claims: Claim[],
+): Promise<{ scratchpad: string; shards: number }> {
+  const shards: Array<{ from: number; to: number; text: string }> = [];
+  for (let i = 0; i < fullDiff.length; i += DIFF_SHARD_SIZE) {
+    const end = Math.min(i + DIFF_SHARD_SIZE, fullDiff.length);
+    shards.push({ from: i, to: end, text: fullDiff.slice(i, end) });
+  }
+  // Curate the claim list into a short form the summary prompt can
+  // use to bias extraction toward relevant facts.
+  const claimDigest = claims.map((c, i) =>
+    `${i}. [${c.strength}] "${c.text.slice(0, 100)}"`
+  ).join("\n");
+
+  let scratchpad = "";
+  for (const [si, shard] of shards.entries()) {
+    const prompt = [
+      `You are summarizing shard ${si + 1}/${shards.length} (chars ${shard.from}..${shard.to}) of a PR diff.`,
+      `The downstream task will verify these ship-claims against the full-PR summary. Extract ONLY facts that could confirm or refute these claims:`,
+      "",
+      claimDigest,
+      "",
+      "Extract: new function/method signatures, struct fields, deletions, new files, wiring (function X calls Y), absence-of-implementation markers, TODO comments on added lines.",
+      "Skip: comment-only edits, whitespace, import reordering, unrelated cosmetic changes.",
+      "",
+      "─────── shard diff ───────",
+      shard.text,
+      "─────── end shard ───────",
+      "",
+      "Output: up to 180 words of facts in bullet form. No prose preamble, no claim verdicts (that's for the downstream step).",
+    ].join("\n");
+
+    const r = await callCloud(prompt, 400);
+    if (r.content) {
+      scratchpad += `\n--- shard ${si + 1} (chars ${shard.from}..${shard.to}) ---\n${r.content.trim()}\n`;
+    }
+  }
+  return { scratchpad: scratchpad.trim(), shards: shards.length };
+}
+
+// Minimal cloud caller used only by treeSplitDiff — same gateway +
+// model as the top-level call, but think=false. Shards are small
+// (≤DIFF_SHARD_SIZE ~4500 chars) and the task is pure fact
+// extraction, not reasoning. think=true on the shards introduced
+// variance in reasoning traces that compounded across 23 calls into
+// a non-deterministic scratchpad (observed during curation
+// validation: same-SHA runs produced 5/7/8 final findings).
+// think=false on small prompts is stable — only breaks at the main
+// call's 10K+ prompt size, which keeps think=true.
+async function callCloud(prompt: string, maxTokens: number): Promise<{ content: string }> {
+  try {
+    const r = await fetch(`${GATEWAY}/v1/chat`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        provider: "ollama_cloud",
+        model: MODEL,
+        messages: [{ role: "user", content: prompt }],
+        max_tokens: maxTokens,
+        temperature: 0,
+        think: false,
+      }),
+      signal: AbortSignal.timeout(CALL_TIMEOUT_MS),
+    });
+    if (!r.ok) return { content: "" };
+    const j: any = await r.json();
+    return { content: j?.choices?.[0]?.message?.content ?? "" };
+  } catch {
+    return { content: "" };
+  }
+}
+
+// Pull out plausible code-symbol names from a summary string.
+// Matches:
+//   - identifier with backticks: `foo_bar`
+//   - identifier followed by parens: foo_bar()
+//   - CamelCase types
+//   - snake_case_functions
+// Filters out common English words that could be matched accidentally.
+const STOPWORDS = new Set([
+  "not","the","and","for","this","that","with","but","are","was","has",
+  "have","been","any","missing","implementation","diff","defined","never",
+  "referenced","integrated","flow","code","file","some","only","when",
+]);
+function extractSymbols(text: string): string[] {
+  const out = new Set<string>();
+  // `backticked` symbols
+  for (const m of text.matchAll(/`([A-Za-z_][A-Za-z0-9_]{2,})`/g)) out.add(m[1]);
+  // foo() or foo_bar() calls
+  for (const m of text.matchAll(/\b([A-Za-z_][A-Za-z0-9_]{2,})\s*\(/g)) out.add(m[1]);
+  // CamelCase types (3+ chars, must start with uppercase)
+  for (const m of text.matchAll(/\b([A-Z][A-Za-z0-9]{2,})\b/g)) out.add(m[1]);
+  return Array.from(out).filter(s => !STOPWORDS.has(s.toLowerCase()));
+}
+
+// Scan the repo for at least one definition of each symbol. Uses Bun's
+// Glob to walk TS/Rust/Python/JS sources; ignores node_modules, data/,
+// and target/. Skips files > 500KB — those are fixtures/snapshots that
+// won't contain a definition line and slurping them slows the audit.
+async function symbolsExistInRepo(symbols: string[]): Promise<string[]> {
+  const patterns = ["**/*.ts", "**/*.tsx", "**/*.rs", "**/*.py", "**/*.js"];
+  const skip = (p: string) => p.includes("/node_modules/") || p.startsWith("data/") || p.includes("/target/") || p.startsWith("dist/");
+  const MAX_FILE_BYTES = 500_000;
+  const { stat } = await import("node:fs/promises");
+  const resolved = new Set<string>();
+  const toFind = new Set(symbols);
+  for (const pat of patterns) {
+    if (toFind.size === 0) break;
+    const glob = new Glob(pat);
+    for await (const f of glob.scan({ cwd: REPO_ROOT, onlyFiles: true })) {
+      if (skip(f)) continue;
+      try { const s = await stat(`${REPO_ROOT}/${f}`); if (s.size > MAX_FILE_BYTES) continue; } catch { continue; }
+      let content: string;
+      try { content = await readFile(`${REPO_ROOT}/${f}`, "utf8"); } catch { continue; }
+      for (const sym of Array.from(toFind)) {
+        // Definition heuristics: `function sym`, `fn sym`, `const sym`,
+        // `let sym`, `def sym`, `class sym`, `struct sym`, `enum sym`,
+        // `trait sym`, `async function sym`, `pub (async )?fn sym`.
+        const re = new RegExp(
+          `\\b(function|async\\s+function|const|let|var|def|class|struct|enum|trait|impl|type|interface|fn|pub\\s+(async\\s+)?fn)\\s+${escapeRe(sym)}\\b`
+        );
+        if (re.test(content)) {
+          resolved.add(sym);
+          toFind.delete(sym);
+          if (toFind.size === 0) break;
+        }
+      }
+    }
+  }
+  return Array.from(resolved);
+}
+
+function escapeRe(s: string): string {
+  return s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+}
+
 // Lift the first balanced JSON object out of the response. Tolerates
 // leading prose, code fences, and model reasoning preamble when the
 // cloud model ignored "strict JSON only."

auditor/checks/kb_query.ts

@@ -8,6 +8,7 @@
 // What this check reads (all file-backed, append-only or periodic):
 //   data/_kb/outcomes.jsonl         — per-scenario outcomes (kb.ts)
 //   data/_kb/error_corrections.jsonl — fail→succeed deltas on same sig
+//   data/_kb/scrum_reviews.jsonl     — scrum-master accepted reviews
 //   data/_observer/ops.jsonl         — observer ring → disk stream
 //   data/_bot/cycles/*.json          — bot cycle results
 //
@@ -17,14 +18,18 @@
 import { readFile, readdir, stat } from "node:fs/promises";
 import { join } from "node:path";
 import type { Claim, Finding } from "../types.ts";
+import { aggregate, ratingSeverity, formatAgg } from "../kb_index.ts";
 
 const KB_DIR = "/home/profit/lakehouse/data/_kb";
 const OBSERVER_OPS = "/home/profit/lakehouse/data/_observer/ops.jsonl";
 const BOT_CYCLES_DIR = "/home/profit/lakehouse/data/_bot/cycles";
+const SCRUM_REVIEWS_JSONL = "/home/profit/lakehouse/data/_kb/scrum_reviews.jsonl";
+const AUDIT_LESSONS_JSONL = "/home/profit/lakehouse/data/_kb/audit_lessons.jsonl";
+const AUDIT_FACTS_JSONL = "/home/profit/lakehouse/data/_kb/audit_facts.jsonl";
 const TAIL_LINES = 500;
 const MAX_BOT_CYCLE_FILES = 30;
 
-export async function runKbCheck(claims: Claim[]): Promise<Finding[]> {
+export async function runKbCheck(claims: Claim[], prFiles: string[] = []): Promise<Finding[]> {
   const findings: Finding[] = [];
 
   // 1. Recent scenario outcomes: are strong-claim-style phrases showing
@@ -48,6 +53,35 @@ export async function runKbCheck(claims: Claim[]): Promise<Finding[]> {
   const obsFindings = await checkObserverStream();
   findings.push(...obsFindings);
 
+  // 5. Scrum-master reviews — surface prior accepted reviews for any
+  //    file in this PR's diff. Cohesion plan Phase C wire: the
+  //    auditor gets to "borrow" the scrum-master's deeper per-file
+  //    analysis instead of re-doing that work.
+  if (prFiles.length > 0) {
+    const scrumFindings = await checkScrumReviews(prFiles);
+    findings.push(...scrumFindings);
+  }
+
+  // 6b. Audit-facts (llm_team extract pipeline output) — surface
+  //     entities that recur across multiple PRs. These are the
+  //     "core system entities" accumulating in the knowledge base;
+  //     showing them as info on future audits gives reviewers
+  //     architectural context the raw diff doesn't convey.
+  const factFindings = await checkAuditFacts();
+  findings.push(...factFindings);
+
+  // 6. Audit-lessons feedback loop — summarize the top recurring
+  //    patterns from prior audits' block/warn findings. If the same
+  //    pattern signature has fired 3+ times across prior audits,
+  //    emit it as a block-severity finding so reviewers know this
+  //    is a known-recurring class, not a one-off. Does NOT couple
+  //    to the current audit's static/inference findings (those run
+  //    in parallel and we can't see them here) — the amplification
+  //    is emergent: if the current audit's finding-summary matches
+  //    a top recurrence, the reviewer sees both.
+  const auditLessonFindings = await checkAuditLessons();
+  findings.push(...auditLessonFindings);
+
   return findings;
 }
 
@@ -181,3 +215,183 @@ function observerBySource(ops: any[]): string {
   }
   return Object.entries(c).sort((a, b) => b[1] - a[1]).map(([k, v]) => `${k}=${v}`).join(", ") || "empty";
 }
+
+// Audit-facts — reads data/_kb/audit_facts.jsonl (populated by every
+// curated inference run via llm_team's extract pipeline). Each row
+// has arrays: facts, entities, relationships. We explode entities and
+// aggregate them across PRs using kb_index. An entity seen in 3+ PRs
+// is a "core system entity" — we surface the top N as info context.
+//
+// Filters out short names (<3 chars, likely qwen2.5 truncation
+// artifacts) and generic types ("string", "number") that would
+// otherwise dominate the ranking.
+const ENTITY_NAME_MIN_LEN = 3;
+const GENERIC_ENTITY_NAMES = new Set([
+  "string", "number", "boolean", "any", "void", "unknown", "never",
+  "object", "array", "function", "const", "let", "var", "true", "false",
+  "null", "undefined", "promise", "map", "set", "record",
+]);
+
+async function checkAuditFacts(): Promise<Finding[]> {
+  // Read raw rows — each row has multiple entities, so we can't just
+  // use aggregate() directly (it's one-signature-per-row). Explode
+  // entities into (row, entity) pairs, then aggregate by entity name.
+  let raw: string;
+  try { raw = await (await import("node:fs/promises")).readFile(AUDIT_FACTS_JSONL, "utf8"); }
+  catch { return []; }
+  const lines = raw.split("\n").filter(l => l.length > 0);
+  if (lines.length === 0) return [];
+
+  interface EntityRow { entity_key: string; pr_number: number; type: string; name: string; description: string }
+  const entityRows: EntityRow[] = [];
+  for (const line of lines.slice(-TAIL_LINES * 2)) {
+    let row: any;
+    try { row = JSON.parse(line); } catch { continue; }
+    const prNum = Number(row?.pr_number);
+    if (!Number.isFinite(prNum)) continue;
+    for (const e of Array.isArray(row?.entities) ? row.entities : []) {
+      const name = String(e?.name ?? "").trim();
+      if (name.length < ENTITY_NAME_MIN_LEN) continue;
+      if (GENERIC_ENTITY_NAMES.has(name.toLowerCase())) continue;
+      entityRows.push({
+        entity_key: name.toLowerCase(),
+        pr_number: prNum,
+        type: String(e?.type ?? "?"),
+        name,
+        description: String(e?.description ?? "").slice(0, 160),
+      });
+    }
+  }
+  if (entityRows.length === 0) return [];
+
+  // Aggregate manually — one key per entity name, distinct_scopes by PR.
+  type Agg = { count: number; scopes: Set<number>; types: Set<string>; last_name: string; last_desc: string };
+  const byEntity = new Map<string, Agg>();
+  for (const r of entityRows) {
+    const a = byEntity.get(r.entity_key) ?? {
+      count: 0, scopes: new Set<number>(), types: new Set<string>(), last_name: "", last_desc: "",
+    };
+    a.count += 1;
+    a.scopes.add(r.pr_number);
+    a.types.add(r.type);
+    a.last_name = r.name;
+    a.last_desc = r.description;
+    byEntity.set(r.entity_key, a);
+  }
+
+  // Rank: require 2+ distinct PRs (same-PR entity-repeats don't count
+  // as "cross-cutting"). Take the top 5 to avoid flooding the verdict.
+  const ranked = Array.from(byEntity.entries())
+    .filter(([_, a]) => a.scopes.size >= 2)
+    .sort((a, b) => b[1].scopes.size - a[1].scopes.size || b[1].count - a[1].count)
+    .slice(0, 5);
+
+  if (ranked.length === 0) {
+    // Useful to know the KB is being populated — emit a single
+    // summary so operators see fact extraction is alive.
+    return [{
+      check: "kb_query",
+      severity: "info",
+      summary: `audit_facts KB has ${entityRows.length} entity-observations across ${new Set(entityRows.map(r => r.pr_number)).size} PRs (no cross-PR recurrences yet)`,
+      evidence: [`source: ${AUDIT_FACTS_JSONL}`],
+    }];
+  }
+
+  return ranked.map(([_, a]) => ({
+    check: "kb_query" as const,
+    severity: "info" as const,
+    summary: `core entity \`${a.last_name}\` recurs in ${a.scopes.size} PRs (types: ${Array.from(a.types).join(",")})`,
+    evidence: [
+      `count=${a.count} distinct_PRs=${a.scopes.size}`,
+      `description: ${a.last_desc.slice(0, 200)}`,
+      `PRs: ${Array.from(a.scopes).sort((x, y) => x - y).join(",")}`,
+    ],
+  }));
+}
+
+// Audit-lessons — reads data/_kb/audit_lessons.jsonl (populated by
+// every audit's appendAuditLessons). Uses the shared kb_index
+// aggregator: groups by `signature`, distinct-scopes keyed by PR
+// number, severity from ratingSeverity(agg) which applies the
+// confidence × count rating (see kb_index.ts). This is the same
+// aggregation any other KB reader uses — shared discipline, not
+// per-check custom logic.
+async function checkAuditLessons(): Promise<Finding[]> {
+  const bySig = await aggregate<any>(AUDIT_LESSONS_JSONL, {
+    keyFn: (r) => r?.signature,
+    scopeFn: (r) => (r?.pr_number !== undefined ? `pr-${r.pr_number}` : undefined),
+    checkFn: (r) => r?.check,
+    tailLimit: TAIL_LINES * 4,
+  });
+  if (bySig.size === 0) return [];
+
+  const findings: Finding[] = [];
+  for (const [sig, agg] of bySig) {
+    // Silent on first-ever occurrence — not yet signal.
+    if (agg.count < 2) continue;
+    const sev = ratingSeverity(agg);
+    findings.push({
+      check: "kb_query",
+      severity: sev,
+      summary: `recurring audit pattern (${agg.distinct_scopes} distinct PRs, ${agg.count} flaggings, conf=${agg.confidence.toFixed(2)}): ${agg.representative_summary.slice(0, 160)}`,
+      evidence: [
+        `signature=${sig}`,
+        `checks: ${agg.checks.join(",")}`,
+        `scopes: ${agg.scopes.slice(-6).join(",")}`,
+        formatAgg(agg),
+      ],
+    });
+  }
+  return findings;
+}
+
+// Scrum-master reviews — the scrum pipeline writes one row per
+// accepted per-file review. We match reviews whose `file` matches
+// any path in the PR's diff, then surface the *preview* + which
+// model the escalation ladder had to reach. If the scrum-master
+// needed the 123B specialist or larger to resolve a file, that's
+// a meaningful signal about the code's complexity — and it's
+// surfaced to the PR without the auditor having to re-run the
+// escalation ladder itself.
+async function checkScrumReviews(prFiles: string[]): Promise<Finding[]> {
+  const rows = await tailJsonl<any>(SCRUM_REVIEWS_JSONL, TAIL_LINES);
+  if (rows.length === 0) return [];
+
+  // Match by exact file OR filename suffix — PR files arrive as
+  // `auditor/audit.ts`-style relative paths; scrum stores the same.
+  const norm = (p: string) => p.replace(/^\/+/, "").replace(/^home\/profit\/lakehouse\//, "");
+  const prSet = new Set(prFiles.map(norm));
+
+  // Keep only the most recent review per file (last-wins).
+  const latestByFile = new Map<string, any>();
+  for (const r of rows) {
+    const f = norm(String(r.file ?? ""));
+    if (!f) continue;
+    if (!prSet.has(f)) continue;
+    latestByFile.set(f, r);
+  }
+  if (latestByFile.size === 0) return [];
+
+  const findings: Finding[] = [];
+  for (const [file, r] of latestByFile) {
+    const model = String(r.accepted_model ?? "?");
+    const attempt = r.accepted_on_attempt ?? "?";
+    const treeSplit = !!r.tree_split_fired;
+    // Heuristic: if the scrum-master had to escalate past attempt 3,
+    // or had to tree-split, that's context the PR reviewer should see.
+    // Severity: info for low-escalation, warn if escalated far up
+    // the ladder (cloud specialist required).
+    const heavyEscalation = Number(attempt) >= 4;
+    const sev: "warn" | "info" = heavyEscalation ? "warn" : "info";
+    findings.push({
+      check: "kb_query",
+      severity: sev,
+      summary: `scrum-master review for \`${file}\` — accepted on attempt ${attempt} by \`${model}\`${treeSplit ? " (tree-split)" : ""}`,
+      evidence: [
+        `reviewed_at: ${r.reviewed_at ?? "?"}`,
+        `preview: ${String(r.suggestions_preview ?? "").slice(0, 300).replace(/\n/g, " ")}`,
+      ],
+    });
+  }
+  return findings;
+}

auditor/checks/static.ts

@@ -61,7 +61,13 @@ export function runStaticCheck(diff: string): Finding[] {
 
       if (!isAuditorCheckerFile) {
         for (const { re, why } of BLOCK_PATTERNS) {
-          if (re.test(added)) {
+          const m = added.match(re);
+          if (m && typeof m.index === "number") {
+            // Skip if the match sits inside a quoted string literal —
+            // this is how rubric files (tests/real-world/*, prompt
+            // templates) legitimately reference the patterns they
+            // guard against, without actually executing them.
+            if (isInsideQuotedString(added, m.index)) continue;
             findings.push({
               check: "static",
               severity: "block",
@@ -154,6 +160,25 @@ function extractNewFields(addedLines: string[]): string[] {
   return Array.from(fields);
 }
 
+// True if `pos` falls inside a double- or single-quoted string on this
+// line (backtick template literals too). Walks left→right toggling the
+// "in quote" state on each unescaped quote. Good enough for single-
+// line matches; multi-line strings aren't parsed (they're extremely
+// rare in the patterns we're blocking on, and would require a proper
+// tokenizer to handle correctly).
+function isInsideQuotedString(line: string, pos: number): boolean {
+  let inDouble = false, inSingle = false, inBacktick = false;
+  for (let i = 0; i < pos; i++) {
+    const c = line[i];
+    const esc = i > 0 && line[i - 1] === "\\";
+    if (esc) continue;
+    if (c === '"' && !inSingle && !inBacktick) inDouble = !inDouble;
+    else if (c === "'" && !inDouble && !inBacktick) inSingle = !inSingle;
+    else if (c === "`" && !inDouble && !inSingle) inBacktick = !inBacktick;
+  }
+  return inDouble || inSingle || inBacktick;
+}
+
 function escape(s: string): string {
   return s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
 }

auditor/claim_parser.ts

@@ -49,6 +49,58 @@ const WEAK_PATTERNS: RegExp[] = [
   /\bprobably\b/i,
 ];
 
+// Empirical claims: runtime measurements / observed outcomes that can't
+// be verified from a diff (only from the actual run that produced
+// them). Classifying as empirical lets the inference check skip
+// diff-verification and saves the ladder for falsifiable claims.
+//
+// Two classes share this bucket because they share the skip discipline:
+//
+//   1. Runtime metrics    — "58 cloud calls", "306s end-to-end"
+//   2. History/proof refs — "verified on PR #8", "was flipping across runs"
+//
+// Both are assertions about state outside the current diff. The cloud
+// would flag them as "not backed" — but that's a false positive: the
+// proof lives in the referenced run, prior commit, or test output, not
+// in the added lines the cloud is reading.
+const EMPIRICAL_PATTERNS: RegExp[] = [
+  // ─── Runtime metrics ───
+  // Iteration / attempt counts: "6/6 iterations", "attempt 5", "accepted on attempt 3"
+  /\b\d+\s*\/\s*\d+\s+(iterations?|attempts?|cycles?|runs?|shards?)\b/i,
+  /\b(accepted|resolved|converged)\s+on\s+attempt\s+\d+\b/i,
+  // Runtime metrics: "58 cloud calls", "306s end-to-end", "245s total", "5931 chars"
+  /\b\d+\s+(cloud\s+)?calls?\b/i,
+  /\b\d+\s*(ms|s|seconds?|minutes?|m)\s+(end[- ]to[- ]end|total|elapsed|duration)\b/i,
+  /\b\d+\s+chars?\b.*\b(accepted|generated|produced)\b/i,
+  // "escalated through N tiers", "N distinct models"
+  /\bescalated\s+through\s+\d+\b/i,
+  /\b\d+\s+distinct\s+(model|tier)s?\b/i,
+
+  // ─── History / proof references ───
+  // "verified on PR #8", "verified end-to-end on PR 8", "tested against PR #4"
+  // Require PR#N / commit-hash / "prior <word>" to avoid matching
+  // "verified ... in production" (PR without \b-ish anchor previously
+  // consumed "pr" of "production").
+  /\bverified\s+(?:end[- ]to[- ]end\s+)?(?:on|against|in)\s+(?:PR\s*#?\d+|commit\s+[0-9a-f]{6,}|prior\s+\w+|the\s+\w+\s+audit)\b/i,
+  /\btested\s+(?:against|in|on)\s+(?:PR\s*#?\d+|commit\s+[0-9a-f]{6,}|prior\s+\w+)\b/i,
+  // Direct PR/commit references: "PR #8", "on PR 9", "from commit abc123"
+  /\b(?:on|from|in|via|per)\s+PR\s*#?\d+\b/i,
+  /\b(?:from|in|per|against)\s+commit\s+[0-9a-f]{6,}/i,
+  // Observational descriptions of prior behavior: "was flipping", "was X before", "previously observed"
+  /\b(?:was|were)\s+(?:flipping|drifting|inconsistent|non[- ]deterministic|creeping)\b/i,
+  /\bpreviously\s+(?:observed|flagged|reported|seen|landed)\b/i,
+  /\bused\s+to\s+(?:flip|fail|flag|reject|block)\b/i,
+  /\bobserved\s+(?:in|during|on|across)\s+(?:PR|prior|\d+\s+(?:runs?|audits?))/i,
+  // "flipping/drifting across N runs" — historical variance description
+  /\b(?:flipping|drifting|varying|oscillating)\s+across\s+(?:\d+\s+)?(?:runs?|audits?|iterations?)\b/i,
+  // "the proven X" referring to prior work (proven is a STRONG pattern
+  // but in context "the proven FOO" is usually a historical reference,
+  // not a fresh claim). We catch it here so the empirical skip wins.
+  /\bthe\s+proven\s+(?:escalation\s+ladder|pipeline|flow|loop|tier|path)/i,
+  // "from the 9-run test", "across the 5-run validation"
+  /\b(?:from|across|in|during)\s+the\s+\d+[- ]run\s+(?:test|validation|probe|experiment)/i,
+];
+
 export interface ParsedClaims {
   claims: Claim[];
   commits_scanned: number;
@@ -77,9 +129,22 @@ function scanText(text: string, location_prefix: string, commit_sha: string, out
     const line = lines[i];
     if (line.length < 3) continue;
 
-    // Strong patterns first — if a line matches strong, it's strong,
-    // don't double-count as moderate.
-    const strong = firstMatch(line, STRONG_PATTERNS);
+    // Empirical match wins over everything else — if a line ALSO
+    // contains a moderate word like "complete", we still want to
+    // classify it as empirical so the inference check doesn't ask
+    // the cloud to prove "58 cloud calls" from the diff. Order:
+    // empirical → strong → moderate → weak.
+    const empirical = firstUnquotedMatch(line, EMPIRICAL_PATTERNS);
+    if (empirical) {
+      out.push({
+        text: line.trim().slice(0, 200),
+        commit_sha,
+        location: `${location_prefix}:${i + 1}`,
+        strength: "empirical",
+      });
+      continue;
+    }
+    const strong = firstUnquotedMatch(line, STRONG_PATTERNS);
     if (strong) {
       out.push({
         text: line.trim().slice(0, 200),
@@ -89,7 +154,7 @@ function scanText(text: string, location_prefix: string, commit_sha: string, out
       });
       continue;
     }
-    const moderate = firstMatch(line, MODERATE_PATTERNS);
+    const moderate = firstUnquotedMatch(line, MODERATE_PATTERNS);
     if (moderate) {
       out.push({
         text: line.trim().slice(0, 200),
@@ -99,7 +164,7 @@ function scanText(text: string, location_prefix: string, commit_sha: string, out
       });
       continue;
     }
-    const weak = firstMatch(line, WEAK_PATTERNS);
+    const weak = firstUnquotedMatch(line, WEAK_PATTERNS);
     if (weak) {
       out.push({
         text: line.trim().slice(0, 200),
@@ -111,9 +176,35 @@ function scanText(text: string, location_prefix: string, commit_sha: string, out
   }
 }
 
-function firstMatch(text: string, patterns: RegExp[]): RegExp | null {
+// Match a pattern only when its match position is NOT inside a quoted
+// string on the line. Mirrors the same guard in auditor/checks/static.ts
+// — the two files have the same false-positive class: PR authors
+// quote pattern examples in commit message bodies (e.g. `"Phase 45
+// shipped"` as a test example) and without this guard those quoted
+// references get flagged as fresh ship-claims. Only skips when the
+// match itself falls inside quotes; real (unquoted) uses of the same
+// vocabulary still classify correctly.
+function firstUnquotedMatch(text: string, patterns: RegExp[]): RegExp | null {
   for (const p of patterns) {
-    if (p.test(text)) return p;
+    const m = text.match(p);
+    if (!m || typeof m.index !== "number") continue;
+    if (isInsideQuotedString(text, m.index)) continue;
+    return p;
   }
   return null;
 }
+
+// Walks left→right toggling in-quote state on each unescaped quote.
+// Good enough for single-line claims; multi-line strings aren't parsed.
+function isInsideQuotedString(line: string, pos: number): boolean {
+  let inDouble = false, inSingle = false, inBacktick = false;
+  for (let i = 0; i < pos; i++) {
+    const c = line[i];
+    const esc = i > 0 && line[i - 1] === "\\";
+    if (esc) continue;
+    if (c === '"' && !inSingle && !inBacktick) inDouble = !inDouble;
+    else if (c === "'" && !inDouble && !inBacktick) inSingle = !inSingle;
+    else if (c === "`" && !inDouble && !inSingle) inBacktick = !inBacktick;
+  }
+  return inDouble || inSingle || inBacktick;
+}

auditor/fact_extractor.ts

@@ -0,0 +1,271 @@
+// fact_extractor — routes curated TEXT through llm_team_ui's
+// "knowledge extract facts" mode (mode=extract at /api/run).
+//
+// What it gives us: structured {facts, entities, relationships} from
+// whatever curated blob we send. Auditor sends the tree-split
+// inference scratchpad (the best distillation of what a PR changed).
+// Scrum_master will later send its accepted review bodies.
+//
+// Why route through llm_team and not just extract directly from our
+// own checks: llm_team's extract uses a local EXTRACTOR model
+// (qwen2.5) + a separate VERIFIER (gemma2). This cross-check is the
+// discipline J wants for knowledge going into the playbook — facts
+// go in only after a second model has rated them CORRECT /
+// UNVERIFIABLE. Fast (local models, ~10-20s), free, and matches the
+// codereview pattern J already trusts.
+//
+// SSE parsing: llm_team streams SSE events. We're only interested in
+// the final "response" event with role="final" + the extraction
+// response (role="extraction N"). Parse the JSON from the extractor's
+// response text.
+
+const LLM_TEAM = process.env.LH_LLM_TEAM_URL ?? "http://localhost:5000";
+const EXTRACTOR = process.env.LH_FACT_EXTRACTOR ?? "qwen2.5:latest";
+const VERIFIER = process.env.LH_FACT_VERIFIER ?? "gemma2:latest";
+const EXTRACT_TIMEOUT_MS = 120_000;
+const PROJECT_CONTEXT_FILE = process.env.LH_AUDITOR_CONTEXT_FILE
+  ?? "/home/profit/lakehouse/docs/AUDITOR_CONTEXT.md";
+
+let cachedContext: string | null = null;
+async function loadProjectContext(): Promise<string> {
+  if (cachedContext !== null) return cachedContext;
+  try {
+    const { readFile } = await import("node:fs/promises");
+    const raw = await readFile(PROJECT_CONTEXT_FILE, "utf8");
+    // Cap at 4KB — anything past that is more noise than signal for
+    // the extractor/verifier's attention budget.
+    cachedContext = raw.slice(0, 4000);
+  } catch {
+    cachedContext = "";  // context file missing → extractor runs without preamble
+  }
+  return cachedContext;
+}
+
+export interface Entity {
+  name: string;
+  type: string;
+  description?: string;
+}
+
+export interface Relationship {
+  from: string;
+  to: string;
+  type: string;
+}
+
+export interface ExtractedFacts {
+  facts: string[];
+  entities: Entity[];
+  relationships: Relationship[];
+  verification: string;
+  extractor_model: string;
+  verifier_model: string;
+  source_preview: string;
+  // Populated when the extract run completed server-side (llm_team
+  // persists to its own team_runs; this is for our own cross-ref).
+  llm_team_run_id?: number;
+  extracted_at: string;
+  // Per-fact verdicts from the verifier pass (CORRECT/INCORRECT/
+  // UNVERIFIABLE/UNCHECKED). Aligned 1:1 with the *raw* fact list
+  // pre-drop so operators can see which verdicts mapped to dropped
+  // facts if needed.
+  verifier_verdicts?: Array<"CORRECT" | "INCORRECT" | "UNVERIFIABLE" | "UNCHECKED">;
+  facts_dropped_by_verifier?: number;
+  error?: string;
+}
+
+/**
+ * Run the llm_team extract pipeline on `source` text. Returns
+ * structured {facts, entities, relationships}.
+ *
+ * Returns an object with `error` set if the pipeline failed — never
+ * throws, because fact extraction is best-effort enrichment (the
+ * primary audit must not break if llm_team is down).
+ */
+export async function extractFacts(source: string): Promise<ExtractedFacts> {
+  const base: ExtractedFacts = {
+    facts: [],
+    entities: [],
+    relationships: [],
+    verification: "",
+    extractor_model: EXTRACTOR,
+    verifier_model: VERIFIER,
+    source_preview: source.slice(0, 240),
+    extracted_at: new Date().toISOString(),
+  };
+
+  // Prepend project context to the source so the extractor + verifier
+  // know what codebase/framework these facts belong to. Without this,
+  // the verifier marks most domain-specific facts as UNVERIFIABLE ("I
+  // don't know what Lakehouse is"). With it, the verifier can CORRECT-
+  // stamp facts that align with the stated architecture.
+  const context = await loadProjectContext();
+  const prompt = context.length > 0
+    ? `=== PROJECT CONTEXT (for grounding facts; do NOT extract facts from this section) ===\n${context}\n\n=== CONTENT TO EXTRACT FACTS FROM ===\n${source}`
+    : source;
+
+  let resp: Response;
+  try {
+    resp = await fetch(`${LLM_TEAM}/api/run`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        mode: "extract",
+        prompt,
+        extractor: EXTRACTOR,
+        verifier: VERIFIER,
+        source: "prompt",
+        skip_cache: true,  // cache by prompt would dedup identical
+                           // scratchpads, but we want fresh extraction
+                           // for per-audit facts; cheap since local.
+      }),
+      signal: AbortSignal.timeout(EXTRACT_TIMEOUT_MS),
+    });
+  } catch (e) {
+    return { ...base, error: `fetch failed: ${(e as Error).message}` };
+  }
+
+  if (!resp.ok) {
+    const body = await resp.text().catch(() => "");
+    return { ...base, error: `llm_team /api/run ${resp.status}: ${body.slice(0, 200)}` };
+  }
+
+  // Stream SSE lines; collect the one extraction response + the run_saved event
+  // so we can capture the team-runs ID for cross-ref.
+  const decoder = new TextDecoder();
+  const reader = resp.body?.getReader();
+  if (!reader) return { ...base, error: "no response body" };
+
+  let buffer = "";
+  let extractionText = "";
+  let verifierText = "";
+  let runId: number | undefined = undefined;
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      let nl: number;
+      while ((nl = buffer.indexOf("\n\n")) >= 0) {
+        const chunk = buffer.slice(0, nl);
+        buffer = buffer.slice(nl + 2);
+        const dataLine = chunk.split("\n").find(l => l.startsWith("data: "));
+        if (!dataLine) continue;
+        try {
+          const ev = JSON.parse(dataLine.slice(6));
+          if (ev.type === "response") {
+            const role = String(ev.role ?? "");
+            if (role.startsWith("extraction")) extractionText = String(ev.text ?? "");
+            else if (role === "verifier") verifierText = String(ev.text ?? "");
+          } else if (ev.type === "run_saved") {
+            const id = Number(ev.run_id);
+            if (Number.isFinite(id)) runId = id;
+          }
+        } catch { /* skip malformed SSE */ }
+      }
+    }
+  } catch (e) {
+    return { ...base, error: `SSE read failed: ${(e as Error).message}` };
+  }
+
+  // Pull the JSON object out of extractionText (may be wrapped in ```json fences).
+  const parsed = extractFirstJsonObject(extractionText);
+  if (!parsed) {
+    return { ...base, error: "extractor returned no parseable JSON", verification: verifierText };
+  }
+
+  const rawFacts: string[] = Array.isArray(parsed.facts)
+    ? parsed.facts.slice(0, 50).map(String)
+    : [];
+
+  // Parse the verifier's free-form prose into per-fact verdicts, then
+  // drop any fact the verifier explicitly marked INCORRECT. Leave
+  // UNVERIFIABLE in place: many of our extractions are domain-specific
+  // (Lakehouse internals) and the verifier has no prior-knowledge
+  // anchor, so UNVERIFIABLE is the expected verdict for new signal,
+  // not a quality fail. This is verifier-gated persistence: drop only
+  // what's affirmatively wrong, not what's novel.
+  const verdicts = parseVerifierVerdicts(verifierText, rawFacts.length);
+  const incorrectIdx = new Set<number>();
+  verdicts.forEach((v, i) => { if (v === "INCORRECT") incorrectIdx.add(i); });
+  const kept = rawFacts.filter((_, i) => !incorrectIdx.has(i));
+
+  return {
+    ...base,
+    facts: kept,
+    entities: Array.isArray(parsed.entities)
+      ? parsed.entities.slice(0, 30).map((e: any) => ({
+          name: String(e?.name ?? ""),
+          type: String(e?.type ?? ""),
+          description: typeof e?.description === "string" ? e.description.slice(0, 240) : undefined,
+        })).filter(e => e.name.length > 0)
+      : [],
+    relationships: Array.isArray(parsed.relationships)
+      ? parsed.relationships.slice(0, 30).map((r: any) => ({
+          from: String(r?.from ?? ""),
+          to: String(r?.to ?? ""),
+          type: String(r?.type ?? ""),
+        })).filter(r => r.from.length > 0 && r.to.length > 0)
+      : [],
+    verification: verifierText.slice(0, 1500),
+    facts_dropped_by_verifier: incorrectIdx.size,
+    verifier_verdicts: verdicts,
+    llm_team_run_id: runId,
+  };
+}
+
+// Parse verifier's free-form output into a per-fact verdict array.
+// Gemma2 uses several formats depending on prompt mood:
+//   Format A:  **1.** claim... * **Verdict:** CORRECT
+//   Format B:  **1.** claim... * **CORRECT**  (no "Verdict:" label)
+//   Format C:  1. claim... CORRECT
+// Strategy: split on fact numbers, then find the first
+// CORRECT|INCORRECT|UNVERIFIABLE token in each section. Handles all
+// three formats without regex gymnastics.
+function parseVerifierVerdicts(
+  verifierText: string,
+  numFacts: number,
+): Array<"CORRECT" | "INCORRECT" | "UNVERIFIABLE" | "UNCHECKED"> {
+  const out: Array<"CORRECT" | "INCORRECT" | "UNVERIFIABLE" | "UNCHECKED"> =
+    Array(numFacts).fill("UNCHECKED");
+  if (!verifierText) return out;
+
+  // Find each fact section start — "**N.**" or "N." at line start —
+  // and slice out the content up to the NEXT fact number. Each section
+  // gets scanned for the first CORRECT/INCORRECT/UNVERIFIABLE token.
+  const starts: Array<{ idx: number; pos: number }> = [];
+  const header = /(?:^|\n)\s*(?:\*\*)?(\d+)[.)]/g;
+  for (const m of verifierText.matchAll(header)) {
+    const factNum = Number(m[1]);
+    if (!Number.isFinite(factNum)) continue;
+    starts.push({ idx: factNum - 1, pos: m.index! });
+  }
+  for (let i = 0; i < starts.length; i++) {
+    const s = starts[i];
+    const end = i + 1 < starts.length ? starts[i + 1].pos : verifierText.length;
+    if (s.idx < 0 || s.idx >= numFacts) continue;
+    const section = verifierText.slice(s.pos, end);
+    const v = section.match(/\b(CORRECT|INCORRECT|UNVERIFIABLE)\b/i);
+    if (v) out[s.idx] = v[1].toUpperCase() as "CORRECT" | "INCORRECT" | "UNVERIFIABLE";
+  }
+  return out;
+}
+
+// Lift the first balanced JSON object out of (possibly fenced) text.
+// Same discipline as inference.ts::extractJson.
+function extractFirstJsonObject(text: string): any | null {
+  const cleaned = text.replace(/^```(?:json)?\s*/im, "").replace(/```\s*$/im, "");
+  let depth = 0, start = -1;
+  for (let i = 0; i < cleaned.length; i++) {
+    const c = cleaned[i];
+    if (c === "{") { if (depth === 0) start = i; depth++; }
+    else if (c === "}") {
+      depth--;
+      if (depth === 0 && start >= 0) {
+        try { return JSON.parse(cleaned.slice(start, i + 1)); } catch { start = -1; }
+      }
+    }
+  }
+  return null;
+}

auditor/kb_index.ts

@@ -0,0 +1,161 @@
+// kb_index — generic on-the-fly aggregation over append-only JSONL
+// scratchpads (audit_lessons, scrum_reviews, outcomes, observer ops).
+//
+// The mem0 insight: raw rows are CHEAP and tell the full story, but
+// downstream prompts need a DEFINITION, not a log. A definition is
+// the aggregate: "this signature has fired N times across M distinct
+// scopes, first_seen=X, last_seen=Y, confidence=M/N."
+//
+// This library is the single shared aggregator. Every KB writer keeps
+// appending raw rows; every KB reader uses aggregate() instead of
+// tailing the raw stream. No second file to sync, no ADD/UPDATE/NOOP
+// routing — the stats roll up from the raw rows every time.
+//
+// Why this works past hundreds of runs:
+//   - aggregate() is bounded by distinct_signatures, not total_rows.
+//   - confidence = distinct_scopes / count — low for same-scope noise,
+//     high for cross-scope patterns. Downstream severity ramps on
+//     confidence × count, not raw count, so one unfixed PR can't
+//     inflate its own recurrence score (the classic mem0 failure).
+//   - rotation (later) moves old raw to archive files; aggregate()
+//     can still read both to compute lifetime counts when needed.
+
+import { readFile } from "node:fs/promises";
+
+export interface AggregateRow {
+  signature: string;
+  count: number;
+  distinct_scopes: number;
+  first_seen: string;
+  last_seen: string;
+  confidence: number;        // distinct_scopes / count — capped at 1.0
+  representative_summary: string;  // most-recent summary for this signature
+  scopes: string[];          // up to 20 most-recent scopes for debugging
+  checks: string[];          // distinct `check` values (audit_lessons-specific)
+}
+
+export interface AggregateOptions<T> {
+  /** How to extract the dedup key from a row. */
+  keyFn: (row: T) => string | undefined;
+  /** How to extract the "scope" — distinct scopes count gives confidence. */
+  scopeFn: (row: T) => string | undefined;
+  /** How to extract the timestamp (defaults to row.audited_at / row.reviewed_at / row.timestamp). */
+  timeFn?: (row: T) => string | undefined;
+  /** How to extract a representative summary (defaults to row.summary). */
+  summaryFn?: (row: T) => string | undefined;
+  /** Max rows to read from the JSONL tail; 0 = read all. */
+  tailLimit?: number;
+  /** Include per-row check field (for multi-check aggregates). */
+  checkFn?: (row: T) => string | undefined;
+}
+
+/**
+ * Read a JSONL file and produce the aggregate map keyed by signature.
+ * Safe on missing or malformed files — returns empty map.
+ */
+export async function aggregate<T = any>(
+  jsonlPath: string,
+  opts: AggregateOptions<T>,
+): Promise<Map<string, AggregateRow>> {
+  const out = new Map<string, AggregateRow>();
+  let raw: string;
+  try { raw = await readFile(jsonlPath, "utf8"); } catch { return out; }
+  const lines = raw.split("\n").filter(l => l.length > 0);
+  const sliceFrom = opts.tailLimit && opts.tailLimit > 0 ? Math.max(0, lines.length - opts.tailLimit) : 0;
+
+  const timeFn = opts.timeFn ?? ((r: any) => r?.audited_at ?? r?.reviewed_at ?? r?.timestamp ?? r?.ran_at);
+  const summaryFn = opts.summaryFn ?? ((r: any) => r?.summary ?? r?.representative_summary);
+
+  // Per-signature scope tracking — need counts by scope to compute
+  // distinct_scopes without double-counting a scope that appears 50
+  // times. Using a Set<scope> per signature.
+  const scopeSets = new Map<string, Set<string>>();
+  const checkSets = new Map<string, Set<string>>();
+
+  for (let i = sliceFrom; i < lines.length; i++) {
+    let row: T;
+    try { row = JSON.parse(lines[i]) as T; } catch { continue; }
+    const sig = opts.keyFn(row);
+    if (!sig) continue;
+
+    let agg = out.get(sig);
+    if (!agg) {
+      agg = {
+        signature: sig,
+        count: 0,
+        distinct_scopes: 0,
+        first_seen: "",
+        last_seen: "",
+        confidence: 0,
+        representative_summary: "",
+        scopes: [],
+        checks: [],
+      };
+      out.set(sig, agg);
+      scopeSets.set(sig, new Set<string>());
+      checkSets.set(sig, new Set<string>());
+    }
+
+    agg.count += 1;
+
+    const scope = opts.scopeFn(row);
+    if (scope !== undefined && scope !== null && scope !== "") {
+      scopeSets.get(sig)!.add(String(scope));
+      // Keep scopes array ordered by recency (newest wins — shift
+      // oldest when at cap).
+      const arr = agg.scopes;
+      const s = String(scope);
+      const existing = arr.indexOf(s);
+      if (existing >= 0) arr.splice(existing, 1);
+      arr.push(s);
+      if (arr.length > 20) arr.shift();
+    }
+
+    if (opts.checkFn) {
+      const c = opts.checkFn(row);
+      if (c) checkSets.get(sig)!.add(String(c));
+    }
+
+    const t = timeFn(row);
+    if (t) {
+      if (!agg.first_seen || t < agg.first_seen) agg.first_seen = t;
+      if (!agg.last_seen  || t > agg.last_seen)  agg.last_seen = t;
+    }
+
+    const s = summaryFn(row);
+    if (s) agg.representative_summary = String(s);
+  }
+
+  // Finalize derived fields.
+  for (const [sig, agg] of out) {
+    const scopes = scopeSets.get(sig) ?? new Set<string>();
+    agg.distinct_scopes = scopes.size;
+    agg.confidence = agg.count > 0 ? Math.min(1, agg.distinct_scopes / agg.count) : 0;
+    const checks = checkSets.get(sig);
+    if (checks) agg.checks = Array.from(checks).sort();
+  }
+  return out;
+}
+
+/**
+ * Severity policy derived from aggregate stats. The rating lives here
+ * (not in each check) so all KB readers ramp severity consistently.
+ *
+ *   - confidence × count product is the real signal.
+ *   - Low confidence (< 0.3) = same-scope noise → info regardless of count.
+ *   - Mid confidence (0.3-0.6) = mixed signal → warn at count ≥ 3.
+ *   - High confidence (> 0.6) with count ≥ 5 = block-worthy cross-cutting pattern.
+ *
+ * Callers can override by reading agg directly; this is the default
+ * policy that matches the "don't escalate one unfixed PR" discipline.
+ */
+export function ratingSeverity(agg: AggregateRow): "info" | "warn" | "block" {
+  if (agg.confidence >= 0.6 && agg.count >= 5) return "block";
+  if (agg.confidence >= 0.3 && agg.count >= 3) return "warn";
+  return "info";
+}
+
+/** Human-friendly one-line summary of an aggregate row for finding evidence. */
+export function formatAgg(agg: AggregateRow): string {
+  return `count=${agg.count} distinct_scopes=${agg.distinct_scopes} confidence=${agg.confidence.toFixed(2)} seen=[${agg.first_seen.slice(0, 10)}..${agg.last_seen.slice(0, 10)}]`;
+}

auditor/kb_stats.ts

@@ -0,0 +1,269 @@
+// kb_stats — on-demand dashboard numbers from the KB scratchpad
+// files. Reads data/_auditor/verdicts/*, data/_kb/audit_lessons.jsonl,
+// data/_kb/audit_facts.jsonl, data/_kb/audit_discrepancies.jsonl,
+// data/_kb/scrum_reviews.jsonl and prints:
+//
+//   - verdict flip-flop rate (same SHA re-audited, verdict changed?)
+//   - consensus discrepancy rate (N runs disagreed on a claim)
+//   - confidence distribution from kb_index aggregator
+//   - top N recurring entities from audit_facts
+//   - fact growth over time
+//   - scrum vs inference KB split
+//
+// Run:  bun run auditor/kb_stats.ts
+//       bun run auditor/kb_stats.ts --top 15     # show top 15 entities
+//       bun run auditor/kb_stats.ts --json       # machine-readable
+//
+// This is the "dashboard" without running Grafana. If someone really
+// wants a dashboard, wire this output into a static HTML page + cron.
+
+import { readFile, readdir } from "node:fs/promises";
+import { join } from "node:path";
+import { aggregate } from "./kb_index.ts";
+
+const REPO = "/home/profit/lakehouse";
+const VERDICTS_DIR = `${REPO}/data/_auditor/verdicts`;
+const AUDIT_LESSONS = `${REPO}/data/_kb/audit_lessons.jsonl`;
+const AUDIT_FACTS = `${REPO}/data/_kb/audit_facts.jsonl`;
+const AUDIT_DISCREPANCIES = `${REPO}/data/_kb/audit_discrepancies.jsonl`;
+const SCRUM_REVIEWS = `${REPO}/data/_kb/scrum_reviews.jsonl`;
+
+interface Args {
+  top: number;
+  json: boolean;
+}
+
+function parseArgs(argv: string[]): Args {
+  const a: Args = { top: 10, json: false };
+  for (let i = 2; i < argv.length; i++) {
+    if (argv[i] === "--top") a.top = Number(argv[++i] ?? 10);
+    else if (argv[i] === "--json") a.json = true;
+  }
+  return a;
+}
+
+async function readJsonl<T = any>(path: string): Promise<T[]> {
+  try {
+    const raw = await readFile(path, "utf8");
+    return raw.split("\n").filter(l => l.length > 0).map(l => {
+      try { return JSON.parse(l) as T; } catch { return null as any; }
+    }).filter(r => r !== null);
+  } catch { return []; }
+}
+
+async function loadVerdicts(): Promise<Array<{ pr: number; sha: string; overall: string; findings_total: number; findings_block: number; findings_warn: number }>> {
+  let files: string[] = [];
+  try { files = await readdir(VERDICTS_DIR); } catch { return []; }
+  const out = [];
+  for (const f of files) {
+    if (!f.endsWith(".json")) continue;
+    const m = f.match(/^(\d+)-([0-9a-f]+)\.json$/);
+    if (!m) continue;
+    try {
+      const v = JSON.parse(await readFile(join(VERDICTS_DIR, f), "utf8"));
+      out.push({
+        pr: Number(m[1]),
+        sha: m[2],
+        overall: String(v.overall),
+        findings_total: Number(v.metrics?.findings_total ?? 0),
+        findings_block: Number(v.metrics?.findings_block ?? 0),
+        findings_warn: Number(v.metrics?.findings_warn ?? 0),
+      });
+    } catch { /* skip corrupt */ }
+  }
+  return out;
+}
+
+interface Stats {
+  audit_count: number;
+  verdict_distribution: Record<string, number>;
+  // Same PR with multiple SHAs — if verdicts differ, that's drift across
+  // the PR's commit history. Not a flip-flop in the classical sense,
+  // but worth surfacing (e.g. "PR #8 was block block req req block").
+  per_pr_verdict_sequences: Record<number, string[]>;
+  // For each PR with ≥ 2 audits, how many distinct verdicts did it
+  // produce? 1 = stable; 2+ = some flipping.
+  verdict_instability: { pr_count: number; pr_with_multiple_verdicts: number; pr_with_3plus: number };
+  consensus: { discrepancy_count: number; tiebreaker_used: number; unresolved: number };
+  kb: {
+    audit_lessons_rows: number;
+    audit_facts_rows: number;
+    scrum_reviews_rows: number;
+    distinct_finding_signatures: number;
+    distinct_entities_across_prs: number;
+    entities_in_2plus_prs: number;
+    entities_in_5plus_prs: number;
+  };
+  fact_quality: {
+    verifier_verdict_distribution: Record<string, number>;
+    facts_dropped_by_verifier_total: number;
+    extraction_success_rate: number;
+  };
+  top_entities: Array<{ name: string; distinct_prs: number; count: number; types: string[] }>;
+  kb_by_source: Record<string, number>;
+}
+
+async function collect(args: Args): Promise<Stats> {
+  const verdicts = await loadVerdicts();
+  const lessons = await readJsonl<any>(AUDIT_LESSONS);
+  const facts = await readJsonl<any>(AUDIT_FACTS);
+  const disc = await readJsonl<any>(AUDIT_DISCREPANCIES);
+  const reviews = await readJsonl<any>(SCRUM_REVIEWS);
+
+  // Verdict stability
+  const byPr: Record<number, string[]> = {};
+  const verdictDist: Record<string, number> = {};
+  for (const v of verdicts) {
+    (byPr[v.pr] ??= []).push(v.overall);
+    verdictDist[v.overall] = (verdictDist[v.overall] ?? 0) + 1;
+  }
+  let multi = 0, tri = 0;
+  for (const [_, seq] of Object.entries(byPr)) {
+    const distinct = new Set(seq);
+    if (distinct.size >= 2) multi++;
+    if (distinct.size >= 3) tri++;
+  }
+
+  // Consensus drift
+  const consensus = {
+    discrepancy_count: disc.length,
+    tiebreaker_used: disc.filter(d => String(d.resolution).startsWith("tiebreaker")).length,
+    unresolved: disc.filter(d => d.resolution === "unresolved").length,
+  };
+
+  // Lesson signatures
+  const lessonAgg = await aggregate<any>(AUDIT_LESSONS, {
+    keyFn: r => r?.signature,
+    scopeFn: r => (r?.pr_number !== undefined ? `pr-${r.pr_number}` : undefined),
+  });
+
+  // Entity aggregation across audit_facts rows
+  interface EntAgg { distinct_prs: Set<number>; count: number; types: Set<string>; name: string; sources: Set<string> }
+  const entAgg = new Map<string, EntAgg>();
+  const sourceCount: Record<string, number> = {};
+  let totalVerdictDist: Record<string, number> = { CORRECT: 0, INCORRECT: 0, UNVERIFIABLE: 0, UNCHECKED: 0 };
+  let factsDroppedTotal = 0;
+  let extractionsWithFacts = 0;
+
+  for (const row of facts) {
+    const src = String(row.source ?? "unknown");
+    sourceCount[src] = (sourceCount[src] ?? 0) + 1;
+    const pr = Number(row.pr_number);
+    if (Array.isArray(row.verifier_verdicts)) {
+      for (const v of row.verifier_verdicts) {
+        totalVerdictDist[v] = (totalVerdictDist[v] ?? 0) + 1;
+      }
+    }
+    factsDroppedTotal += Number(row.facts_dropped_by_verifier ?? 0);
+    if ((Array.isArray(row.facts) && row.facts.length > 0) || (Array.isArray(row.entities) && row.entities.length > 0)) {
+      extractionsWithFacts++;
+    }
+    for (const e of Array.isArray(row.entities) ? row.entities : []) {
+      const name = String(e?.name ?? "").trim();
+      if (name.length < 3) continue;
+      const key = name.toLowerCase();
+      const agg = entAgg.get(key) ?? { distinct_prs: new Set(), count: 0, types: new Set(), name, sources: new Set() };
+      agg.count++;
+      if (Number.isFinite(pr) && pr > 0) agg.distinct_prs.add(pr);
+      if (e?.type) agg.types.add(String(e.type));
+      agg.sources.add(src);
+      entAgg.set(key, agg);
+    }
+  }
+
+  const entitiesIn2Plus = Array.from(entAgg.values()).filter(a => a.distinct_prs.size >= 2).length;
+  const entitiesIn5Plus = Array.from(entAgg.values()).filter(a => a.distinct_prs.size >= 5).length;
+  const topEntities = Array.from(entAgg.values())
+    .sort((a, b) => b.distinct_prs.size - a.distinct_prs.size || b.count - a.count)
+    .slice(0, args.top)
+    .map(a => ({
+      name: a.name,
+      distinct_prs: a.distinct_prs.size,
+      count: a.count,
+      types: Array.from(a.types),
+    }));
+
+  const stats: Stats = {
+    audit_count: verdicts.length,
+    verdict_distribution: verdictDist,
+    per_pr_verdict_sequences: byPr,
+    verdict_instability: {
+      pr_count: Object.keys(byPr).length,
+      pr_with_multiple_verdicts: multi,
+      pr_with_3plus: tri,
+    },
+    consensus,
+    kb: {
+      audit_lessons_rows: lessons.length,
+      audit_facts_rows: facts.length,
+      scrum_reviews_rows: reviews.length,
+      distinct_finding_signatures: lessonAgg.size,
+      distinct_entities_across_prs: entAgg.size,
+      entities_in_2plus_prs: entitiesIn2Plus,
+      entities_in_5plus_prs: entitiesIn5Plus,
+    },
+    fact_quality: {
+      verifier_verdict_distribution: totalVerdictDist,
+      facts_dropped_by_verifier_total: factsDroppedTotal,
+      extraction_success_rate: facts.length > 0 ? extractionsWithFacts / facts.length : 0,
+    },
+    top_entities: topEntities,
+    kb_by_source: sourceCount,
+  };
+  return stats;
+}
+
+function renderHuman(s: Stats): string {
+  const lines: string[] = [];
+  lines.push("═══ KB STATS ═══");
+  lines.push("");
+  lines.push(`Audits: ${s.audit_count} total across ${s.verdict_instability.pr_count} distinct PRs`);
+  lines.push(`Verdicts: ${Object.entries(s.verdict_distribution).map(([k, v]) => `${k}=${v}`).join("  ")}`);
+  const multiplePct = s.verdict_instability.pr_count > 0
+    ? Math.round(100 * s.verdict_instability.pr_with_multiple_verdicts / s.verdict_instability.pr_count)
+    : 0;
+  lines.push(`Verdict instability: ${s.verdict_instability.pr_with_multiple_verdicts}/${s.verdict_instability.pr_count} PRs had 2+ distinct verdicts (${multiplePct}%) — 3+ distinct: ${s.verdict_instability.pr_with_3plus}`);
+  lines.push("");
+  lines.push("─── Consensus ───");
+  lines.push(`  discrepancies logged: ${s.consensus.discrepancy_count}`);
+  lines.push(`  tiebreaker used: ${s.consensus.tiebreaker_used}`);
+  lines.push(`  unresolved: ${s.consensus.unresolved}`);
+  const dRate = s.audit_count > 0 ? (100 * s.consensus.discrepancy_count / s.audit_count).toFixed(1) : "0";
+  lines.push(`  discrepancy rate: ${dRate}% of audits`);
+  lines.push("");
+  lines.push("─── KB size ───");
+  lines.push(`  audit_lessons.jsonl:     ${s.kb.audit_lessons_rows} rows, ${s.kb.distinct_finding_signatures} distinct signatures`);
+  lines.push(`  audit_facts.jsonl:       ${s.kb.audit_facts_rows} rows, ${s.kb.distinct_entities_across_prs} distinct entities`);
+  lines.push(`  scrum_reviews.jsonl:     ${s.kb.scrum_reviews_rows} rows`);
+  lines.push(`  entities in 2+ PRs:      ${s.kb.entities_in_2plus_prs}`);
+  lines.push(`  entities in 5+ PRs:      ${s.kb.entities_in_5plus_prs} ← strong cross-cutting signal`);
+  lines.push("");
+  lines.push("─── Fact quality ───");
+  const v = s.fact_quality.verifier_verdict_distribution;
+  lines.push(`  verifier verdicts:  CORRECT=${v.CORRECT ?? 0}  UNVERIFIABLE=${v.UNVERIFIABLE ?? 0}  UNCHECKED=${v.UNCHECKED ?? 0}  INCORRECT=${v.INCORRECT ?? 0}`);
+  lines.push(`  facts dropped by verifier: ${s.fact_quality.facts_dropped_by_verifier_total}`);
+  lines.push(`  extraction success rate: ${(s.fact_quality.extraction_success_rate * 100).toFixed(1)}%`);
+  lines.push("");
+  lines.push("─── KB sources ───");
+  for (const [src, n] of Object.entries(s.kb_by_source)) {
+    lines.push(`  ${src}: ${n}`);
+  }
+  lines.push("");
+  lines.push(`─── Top ${s.top_entities.length} recurring entities ───`);
+  for (const e of s.top_entities) {
+    lines.push(`  [${e.distinct_prs} PRs × ${e.count} obs]  ${e.name}  (${e.types.join(",")})`);
+  }
+  return lines.join("\n");
+}
+
+async function main() {
+  const args = parseArgs(process.argv);
+  const stats = await collect(args);
+  if (args.json) {
+    console.log(JSON.stringify(stats, (_, v) => v instanceof Set ? Array.from(v) : v, 2));
+  } else {
+    console.log(renderHuman(stats));
+  }
+}
+
+main().catch(e => { console.error("[kb_stats] fatal:", e); process.exit(1); });

auditor/types.ts

@@ -18,7 +18,14 @@ export interface Claim {
   // Heuristic rating of how strong the claim is. "green+tested"
   // is strong; "should work" is weak. Drives sensitivity — stronger
   // claims get harder-blocked on weak evidence.
-  strength: "weak" | "moderate" | "strong";
+  //
+  // "empirical" is a separate class: runtime measurements like
+  // "N cloud calls" / "306s end-to-end" / "accepted on attempt N".
+  // These cannot be verified from a static diff — only from the test
+  // output that produced them. Inference skips diff-verification for
+  // empirical claims; they become info-level context unless a future
+  // runtime_evidence check contradicts them.
+  strength: "weak" | "moderate" | "strong" | "empirical";
 }
 
 export interface Finding {

config/routing.toml

@@ -0,0 +1,61 @@
+# Phase 40: Routing Engine Configuration
+#
+# Human-editable rules for model → provider routing.
+# Matching order: first match wins.
+
+[[rule]]
+model_pattern = "gpt-4*"
+provider = "openrouter"
+max_tokens = 4096
+temperature = 0.7
+
+[[rule]]
+model_pattern = "claude*"
+provider = "claude"
+max_tokens = 4096
+temperature = 0.7
+
+[[rule]]
+model_pattern = "gemini*"
+provider = "gemini"
+max_tokens = 8192
+temperature = 0.9
+
+[[rule]]
+model_pattern = "qwen3.5*"
+provider = "ollama"
+max_tokens = 4096
+temperature = 0.3
+
+[[rule]]
+model_pattern = "qwen3*"
+provider = "ollama"
+max_tokens = 2048
+temperature = 0.3
+
+[[rule]]
+model_pattern = "gpt-oss*"
+provider = "ollama"
+temperature = 0.1
+
+[[rule]]
+model_pattern = "*"
+provider = "ollama"
+temperature = 0.5
+
+# Fallback chain: if primary fails, try these in order
+fallback = ["ollama", "openrouter"]
+
+# Cost gating (tokens = cents per 1M)
+[cost]
+ollama = 0
+openrouter = 15
+claude = 15
+gemini = 0
+
+# Daily budget per provider (cents)
+[daily_budget]
+ollama = 0
+openrouter = 1000
+claude = 500
+gemini = 0

crates/aibridge/Cargo.toml

@@ -11,3 +11,4 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 tracing = { workspace = true }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+async-trait = "0.1"

crates/aibridge/src/lib.rs

@@ -1,5 +1,8 @@
 pub mod client;
 pub mod context;
 pub mod continuation;
+pub mod provider;
+pub mod providers;
+pub mod routing;
 pub mod service;
 pub mod tree_split;

crates/aibridge/src/provider.rs

@@ -0,0 +1,39 @@
+use async_trait::async_trait;
+use crate::client::{GenerateRequest, GenerateResponse, EmbedRequest, EmbedResponse};
+
+#[async_trait]
+pub trait ProviderAdapter: Send + Sync {
+    /// Name for routing (ollama, openrouter, etc.)
+    fn name(&self) -> &str;
+
+    /// Chat completion — returns text, model, token counts
+    async fn chat(&self, req: GenerateRequest) -> Result<GenerateResponse, String>;
+
+    /// Embeddings — returns vectors, model, dimensions
+    async fn embed(&self, req: EmbedRequest) -> Result<EmbedResponse, String>;
+
+    /// Unload model from VRAM (optional, no-op if not supported)
+    async fn unload(&self, _model: &str) -> Result<(), String> {
+        Ok(())
+    }
+
+    /// Health check
+    async fn health(&self) -> Result<serde_json::Value, String>;
+}
+
+/// Routing key extracted from model name.
+/// - "qwen3.5:latest" → "ollama"
+/// - "openrouter/anthropic/claude-3.5-sonnet" → "openrouter"
+/// - "gpt-4o" → "ollama" (default)
+pub fn provider_key(model: &str) -> &'static str {
+    let lower = model.to_lowercase();
+    if lower.starts_with("openrouter/") {
+        "openrouter"
+    } else if lower.starts_with("gemini") {
+        "gemini"
+    } else if lower.starts_with("claude") {
+        "claude"
+    } else {
+        "ollama"  // default: local Ollama
+    }
+}

crates/aibridge/src/providers/mod.rs

@@ -0,0 +1,2 @@
+pub mod ollama;
+pub mod openrouter;

crates/aibridge/src/providers/ollama.rs

@@ -0,0 +1,37 @@
+use async_trait::async_trait;
+use crate::client::{AiClient, GenerateRequest, GenerateResponse, EmbedRequest, EmbedResponse};
+use crate::provider::ProviderAdapter;
+
+pub struct OllamaAdapter {
+    client: AiClient,
+}
+
+impl OllamaAdapter {
+    pub fn new(client: AiClient) -> Self {
+        Self { client }
+    }
+}
+
+#[async_trait]
+impl ProviderAdapter for OllamaAdapter {
+    fn name(&self) -> &str {
+        "ollama"
+    }
+
+    async fn chat(&self, req: GenerateRequest) -> Result<GenerateResponse, String> {
+        self.client.generate(req).await
+    }
+
+    async fn embed(&self, req: EmbedRequest) -> Result<EmbedResponse, String> {
+        self.client.embed(req).await
+    }
+
+    async fn unload(&self, model: &str) -> Result<(), String> {
+        let _: serde_json::Value = self.client.unload_model(model).await?;
+        Ok(())
+    }
+
+    async fn health(&self) -> Result<serde_json::Value, String> {
+        self.client.health().await
+    }
+}

crates/aibridge/src/providers/openrouter.rs

@@ -0,0 +1,150 @@
+use async_trait::async_trait;
+use reqwest::Client;
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+
+use crate::client::{GenerateRequest, GenerateResponse, EmbedRequest, EmbedResponse};
+use crate::provider::ProviderAdapter;
+
+pub struct OpenRouterAdapter {
+    client: Client,
+    base_url: String,
+    api_key: String,
+    default_model: String,
+}
+
+#[derive(Serialize)]
+struct OpenRouterChatRequest {
+    model: String,
+    messages: Vec<OpenRouterMessage>,
+    temperature: Option<f64>,
+    max_tokens: Option<u32>,
+}
+
+#[derive(Serialize)]
+struct OpenRouterMessage {
+    role: String,
+    content: String,
+}
+
+#[derive(Deserialize)]
+struct OpenRouterChatResponse {
+    choices: Vec<OpenRouterChoice>,
+    usage: OpenRouterUsage,
+    model: String,
+}
+
+#[derive(Deserialize)]
+struct OpenRouterChoice {
+    message: OpenRouterMessageOut,
+}
+
+#[derive(Deserialize)]
+struct OpenRouterMessageOut {
+    role: String,
+    content: String,
+}
+
+#[derive(Deserialize)]
+struct OpenRouterUsage {
+    prompt_tokens: Option<u32>,
+    completion_tokens: Option<u32>,
+    total_tokens: Option<u32>,
+}
+
+impl OpenRouterAdapter {
+    pub fn new(base_url: &str, api_key: String, default_model: &str) -> Self {
+        let client = Client::builder()
+            .timeout(Duration::from_secs(180))
+            .build()
+            .expect("failed to build HTTP client");
+        Self {
+            client,
+            base_url: base_url.trim_end_matches('/').to_string(),
+            api_key,
+            default_model: default_model.to_string(),
+        }
+    }
+
+    fn chat_model(&self, model: &str) -> String {
+        // Strip "openrouter/" prefix if present
+        let m = model.trim_start_matches("openrouter/");
+        if m.is_empty() || m == model {
+            self.default_model.clone()
+        } else {
+            m.to_string()
+        }
+    }
+
+    fn to_openrouter_messages(req: &GenerateRequest) -> Vec<OpenRouterMessage> {
+        let mut out = vec![];
+        if let Some(sys) = &req.system {
+            out.push(OpenRouterMessage { role: "system".into(), content: sys.clone() });
+        }
+        out.push(OpenRouterMessage {
+            role: "user".into(),
+            content: req.prompt.clone(),
+        });
+        out
+    }
+}
+
+#[async_trait]
+impl ProviderAdapter for OpenRouterAdapter {
+    fn name(&self) -> &str {
+        "openrouter"
+    }
+
+    async fn chat(&self, req: GenerateRequest) -> Result<GenerateResponse, String> {
+        let model = self.chat_model(req.model.as_deref().unwrap_or(""));
+        let or_req = OpenRouterChatRequest {
+            model: model.clone(),
+            messages: OpenRouterAdapter::to_openrouter_messages(&req),
+            temperature: req.temperature,
+            max_tokens: req.max_tokens,
+        };
+
+        let resp = self.client
+            .post(format!("{}/chat/completions", self.base_url))
+            .header("Authorization", format!("Bearer {}", self.api_key))
+            .header("Content-Type", "application/json")
+            .json(&or_req)
+            .send()
+            .await
+            .map_err(|e| format!("openrouter request failed: {e}"))?;
+
+        let status = resp.status();
+        let body = resp.text().await.unwrap_or_default();
+
+        if !status.is_success() {
+            return Err(format!("openrouter error ({}): {}", status, body));
+        }
+
+        let or_resp: OpenRouterChatResponse = serde_json::from_str(&body)
+            .map_err(|e| format!("openrouter parse error: {e}"))?;
+
+        let choice = or_resp.choices.into_iter().next()
+            .ok_or("no completion choice returned")?;
+        let usage = or_resp.usage;
+
+        Ok(GenerateResponse {
+            text: choice.message.content,
+            model: or_resp.model,
+            tokens_evaluated: usage.prompt_tokens.map(|n| n as u64),
+            tokens_generated: usage.completion_tokens.map(|n| n as u64),
+        })
+    }
+
+    async fn embed(&self, _req: EmbedRequest) -> Result<EmbedResponse, String> {
+        Err("openrouter: embed not implemented".into())
+    }
+
+    async fn health(&self) -> Result<serde_json::Value, String> {
+        // OpenRouter doesn't have a dedicated health endpoint,
+        // so we just return a healthy response if the client works.
+        Ok(serde_json::json!({
+            "status": "ok",
+            "provider": "openrouter",
+        }))
+    }
+}

crates/aibridge/src/routing.rs

@@ -0,0 +1,94 @@
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct RoutingRule {
+    pub model_pattern: String,
+    pub provider: String,
+    pub max_tokens: Option<u32>,
+    pub temperature: Option<f64>,
+}
+
+#[derive(Clone, Default)]
+pub struct RoutingEngine {
+    rules: Vec<RoutingRule>,
+    fallback_chain: Vec<String>,
+}
+
+impl RoutingEngine {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn with_rules(mut self, rules: Vec<RoutingRule>) -> Self {
+        self.rules = rules;
+        self
+    }
+
+    pub fn with_fallback(mut self, chain: Vec<String>) -> Self {
+        self.fallback_chain = chain;
+        self
+    }
+
+    pub fn route(&self, model: &str) -> RouteDecision {
+        let lower = model.to_lowercase();
+
+        for rule in &self.rules {
+            if glob_match(&rule.model_pattern.to_lowercase(), &lower) {
+                return RouteDecision {
+                    provider: rule.provider.clone(),
+                    model: model.to_string(),
+                    max_tokens: rule.max_tokens,
+                    temperature: rule.temperature,
+                };
+            }
+        }
+
+        if let Some(first) = self.fallback_chain.first() {
+            RouteDecision {
+                provider: first.clone(),
+                model: model.to_string(),
+                max_tokens: None,
+                temperature: None,
+            }
+        } else {
+            RouteDecision {
+                provider: "ollama".to_string(),
+                model: model.to_string(),
+                max_tokens: None,
+                temperature: None,
+            }
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct RouteDecision {
+    pub provider: String,
+    pub model: String,
+    pub max_tokens: Option<u32>,
+    pub temperature: Option<f64>,
+}
+
+fn glob_match(pattern: &str, name: &str) -> bool {
+    if pattern.contains('*') {
+        let parts: Vec<&str> = pattern.split('*').collect();
+        if parts.len() == 2 {
+            return name.starts_with(parts[0]) && name.ends_with(parts[1]);
+        } else if parts.len() == 1 {
+            return name.starts_with(parts[0]) || name.ends_with(parts[1]);
+        }
+    }
+    pattern == name
+}
+
+impl Default for RoutingRule {
+    fn default() -> Self {
+        Self {
+            model_pattern: "*".to_string(),
+            provider: "ollama".to_string(),
+            max_tokens: None,
+            temperature: None,
+        }
+    }
+}

crates/catalogd/src/service.rs

@@ -31,6 +31,10 @@ pub fn router(registry: Registry) -> Router {
         // Phase 17: model profiles
         .route("/profiles", post(create_profile).get(list_profiles))
         .route("/profiles/{id}", get(get_profile).delete(delete_profile))
+        // Phase 41: profile-type routes
+        .route("/profiles/retrieval", get(list_retrieval_profiles))
+        .route("/profiles/memory", get(list_memory_profiles))
+        .route("/profiles/observer", get(list_observer_profiles))
         .with_state(registry)
 }
 
@@ -398,6 +402,9 @@ struct CreateProfileRequest {
     /// indexes to. Defaults to Parquet+HNSW.
     #[serde(default)]
     vector_backend: shared::types::VectorBackend,
+    /// Phase 41: Profile type for routing + activation behavior.
+    #[serde(default)]
+    profile_type: shared::types::ProfileType,
 }
 
 fn default_embed_model_req() -> String { "nomic-embed-text".to_string() }
@@ -417,6 +424,7 @@ async fn create_profile(
         created_by: req.created_by,
         bucket: req.bucket,
         vector_backend: req.vector_backend,
+        profile_type: req.profile_type,
     };
     match registry.put_profile(profile).await {
         Ok(p) => Ok((StatusCode::CREATED, Json(p))),
@@ -428,6 +436,24 @@ async fn list_profiles(State(registry): State<Registry>) -> impl IntoResponse {
     Json(registry.list_profiles().await)
 }
 
+async fn list_retrieval_profiles(State(registry): State<Registry>) -> impl IntoResponse {
+    let all = registry.list_profiles().await;
+    let retrieval: Vec<_> = all.into_iter().filter(|p| p.profile_type == shared::types::ProfileType::Retrieval).collect();
+    Json(retrieval)
+}
+
+async fn list_memory_profiles(State(registry): State<Registry>) -> impl IntoResponse {
+    let all = registry.list_profiles().await;
+    let memory: Vec<_> = all.into_iter().filter(|p| p.profile_type == shared::types::ProfileType::Memory).collect();
+    Json(memory)
+}
+
+async fn list_observer_profiles(State(registry): State<Registry>) -> impl IntoResponse {
+    let all = registry.list_profiles().await;
+    let observer: Vec<_> = all.into_iter().filter(|p| p.profile_type == shared::types::ProfileType::Observer).collect();
+    Json(observer)
+}
+
 async fn get_profile(
     State(registry): State<Registry>,
     Path(id): Path<String>,

crates/gateway/src/v1/mod.rs

@@ -24,7 +24,7 @@ use axum::{
     Json,
 };
 use serde::{Deserialize, Serialize};
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 use tokio::sync::RwLock;
 
 #[derive(Clone)]
@@ -46,6 +46,16 @@ pub struct Usage {
     pub prompt_tokens: u64,
     pub completion_tokens: u64,
     pub total_tokens: u64,
+    #[serde(default)]
+    pub by_provider: std::collections::HashMap<String, ProviderUsage>,
+}
+
+#[derive(Default, Clone, Serialize)]
+pub struct ProviderUsage {
+    pub requests: u64,
+    pub prompt_tokens: u64,
+    pub completion_tokens: u64,
+    pub total_tokens: u64,
 }
 
 pub fn router(state: V1State) -> Router {
@@ -135,18 +145,22 @@ async fn chat(
     let start_time = chrono::Utc::now();
     let start_instant = std::time::Instant::now();
 
-    let resp = match provider.as_str() {
-        "ollama" | "local" | "" => ollama::chat(&state.ai_client, &req)
-            .await
-            .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama local: {e}")))?,
+    let (resp, used_provider) = match provider.as_str() {
+        "ollama" | "local" | "" => {
+            let r = ollama::chat(&state.ai_client, &req)
+                .await
+                .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama local: {e}")))?;
+            (r, "ollama".to_string())
+        }
         "ollama_cloud" | "cloud" => {
             let key = state.ollama_cloud_key.as_deref().ok_or((
                 StatusCode::SERVICE_UNAVAILABLE,
-                "OLLAMA_CLOUD_KEY not configured — set env or populate /root/llm_team_config.json".to_string(),
+                "OLLAMA_CLOUD_KEY not configured".to_string(),
             ))?;
-            ollama_cloud::chat(key, &req)
+            let r = ollama_cloud::chat(key, &req)
                 .await
-                .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama cloud: {e}")))?
+                .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama cloud: {e}")))?;
+            (r, "ollama_cloud".to_string())
         }
         other => {
             return Err((
@@ -168,7 +182,7 @@ async fn chat(
             .map(|c| c.message.content.clone())
             .unwrap_or_default();
         lf.emit_chat(langfuse_trace::ChatTrace {
-            provider: provider.clone(),
+            provider: used_provider.clone(),
             model: resp.model.clone(),
             input: req.messages.clone(),
             output,
@@ -183,12 +197,19 @@ async fn chat(
         });
     }
 
+    // Phase 40: per-provider usage tracking
     {
         let mut u = state.usage.write().await;
         u.requests += 1;
         u.prompt_tokens += resp.usage.prompt_tokens as u64;
         u.completion_tokens += resp.usage.completion_tokens as u64;
         u.total_tokens += resp.usage.total_tokens as u64;
+
+        let provider_usage = u.by_provider.entry(used_provider).or_default();
+        provider_usage.requests += 1;
+        provider_usage.prompt_tokens += resp.usage.prompt_tokens as u64;
+        provider_usage.completion_tokens += resp.usage.completion_tokens as u64;
+        provider_usage.total_tokens += resp.usage.total_tokens as u64;
     }
 
     Ok(Json(resp))

crates/shared/src/types.rs

@@ -269,6 +269,20 @@ pub struct ModelProfile {
     /// / 5M+ vector workloads.
     #[serde(default)]
     pub vector_backend: VectorBackend,
+    /// Phase 41: Profile type for routing + activation behavior.
+    /// Default Execution for backward compatibility.
+    #[serde(default)]
+    pub profile_type: ProfileType,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum ProfileType {
+    #[default]
+    Execution,
+    Retrieval,
+    Memory,
+    Observer,
 }
 
 fn default_embed_model() -> String { "nomic-embed-text".to_string() }

crates/vectord/Cargo.toml

@@ -26,3 +26,4 @@ chrono = { workspace = true }
 instant-distance = { workspace = true }
 uuid = { workspace = true }
 sha2 = { workspace = true }
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }

crates/vectord/src/doc_drift.rs

@@ -0,0 +1,166 @@
+// Phase 45 slice 3 — context7 bridge client for doc-drift detection.
+//
+// Calls the Bun context7 bridge on :3900 (mcp-server/context7_bridge.ts)
+// which itself wraps context7's public API. For each DocRef on a
+// playbook, queries /docs/:tool/diff?since=<snippet_hash> and parses
+// the response.
+//
+// Kept deliberately thin: no caching here (bridge caches for 5 min),
+// no retry logic (transient fail = report as "unknown" drift status,
+// don't flag). The handler decides whether to flag the playbook based
+// on whether ANY tool came back drifted=true.
+
+use serde::Deserialize;
+use std::time::Duration;
+
+use crate::playbook_memory::DocRef;
+
+const DEFAULT_BRIDGE_URL: &str = "http://localhost:3900";
+const CALL_TIMEOUT_SECS: u64 = 15;
+
+#[derive(Debug, Clone)]
+pub struct DriftCheck {
+    pub tool: String,
+    pub version_seen: String,
+    pub outcome: DriftOutcome,
+}
+
+#[derive(Debug, Clone)]
+pub enum DriftOutcome {
+    /// Bridge reports drift (current snippet_hash != previous).
+    Drifted { current_snippet_hash: String, source_url: Option<String> },
+    /// Bridge reports no drift — the stored snippet_hash still matches
+    /// the current context7 docs.
+    Unchanged,
+    /// Bridge unreachable, 404 on the tool, or returned unparseable
+    /// data. We deliberately don't flag on these — a down bridge
+    /// shouldn't silently mark every playbook drift-flagged.
+    Unknown { reason: String },
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct BridgeDiffResponse {
+    drifted: bool,
+    current_snippet_hash: Option<String>,
+    source_url: Option<String>,
+}
+
+pub struct DriftCheckerConfig {
+    pub bridge_url: String,
+}
+
+impl Default for DriftCheckerConfig {
+    fn default() -> Self {
+        Self {
+            bridge_url: std::env::var("LH_BRIDGE_URL")
+                .unwrap_or_else(|_| DEFAULT_BRIDGE_URL.to_string()),
+        }
+    }
+}
+
+/// For every doc_ref, ask the bridge whether it drifted against the
+/// recorded snippet_hash. Returns per-tool outcomes.
+pub async fn check_all_refs(
+    cfg: &DriftCheckerConfig,
+    doc_refs: &[DocRef],
+) -> Vec<DriftCheck> {
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(CALL_TIMEOUT_SECS))
+        .build()
+        .expect("reqwest client build");
+
+    let mut out = Vec::with_capacity(doc_refs.len());
+    for r in doc_refs {
+        let hash = r.snippet_hash.as_deref().unwrap_or("");
+        if hash.is_empty() {
+            // No hash to compare against — can't detect drift. Report
+            // unknown so the caller isn't forced to flag.
+            out.push(DriftCheck {
+                tool: r.tool.clone(),
+                version_seen: r.version_seen.clone(),
+                outcome: DriftOutcome::Unknown {
+                    reason: "no snippet_hash recorded on doc_ref".into(),
+                },
+            });
+            continue;
+        }
+        let url = format!(
+            "{}/docs/{}/diff?since={}",
+            cfg.bridge_url.trim_end_matches('/'),
+            urlencoding_minimal(&r.tool),
+            urlencoding_minimal(hash),
+        );
+        let outcome = match client.get(&url).send().await {
+            Err(e) => DriftOutcome::Unknown {
+                reason: format!("bridge unreachable: {}", e),
+            },
+            Ok(resp) => {
+                if resp.status() == reqwest::StatusCode::NOT_FOUND {
+                    DriftOutcome::Unknown {
+                        reason: format!("bridge 404 — no context7 library for tool '{}'", r.tool),
+                    }
+                } else if !resp.status().is_success() {
+                    let status = resp.status();
+                    DriftOutcome::Unknown {
+                        reason: format!("bridge {}: {}", status, resp.text().await.unwrap_or_default()),
+                    }
+                } else {
+                    match resp.json::<BridgeDiffResponse>().await {
+                        Err(e) => DriftOutcome::Unknown {
+                            reason: format!("bridge response parse: {}", e),
+                        },
+                        Ok(body) => {
+                            if body.drifted {
+                                DriftOutcome::Drifted {
+                                    current_snippet_hash: body.current_snippet_hash.unwrap_or_default(),
+                                    source_url: body.source_url,
+                                }
+                            } else {
+                                DriftOutcome::Unchanged
+                            }
+                        }
+                    }
+                }
+            }
+        };
+        out.push(DriftCheck {
+            tool: r.tool.clone(),
+            version_seen: r.version_seen.clone(),
+            outcome,
+        });
+    }
+    out
+}
+
+/// Minimal URL path+query encoder for ASCII tool names. We're not
+/// pulling in the `urlencoding` crate for this single use — tool
+/// names are short alphanumeric (docker/terraform/react), snippet
+/// hashes are hex. Handles space + `/` + `?` + `&` defensively so
+/// unusual tool names don't corrupt the URL.
+fn urlencoding_minimal(s: &str) -> String {
+    let mut out = String::with_capacity(s.len());
+    for b in s.bytes() {
+        match b {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(b as char),
+            _ => out.push_str(&format!("%{:02X}", b)),
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn urlencoding_handles_ascii_safe_chars_passthrough() {
+        assert_eq!(urlencoding_minimal("docker"), "docker");
+        assert_eq!(urlencoding_minimal("next.js"), "next.js");
+    }
+
+    #[test]
+    fn urlencoding_encodes_slash_and_space() {
+        assert_eq!(urlencoding_minimal("foo/bar"), "foo%2Fbar");
+        assert_eq!(urlencoding_minimal("foo bar"), "foo%20bar");
+    }
+}

crates/vectord/src/jobs.rs

@@ -8,6 +8,13 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::RwLock;
 
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum JobType {
+    Embed,
+    ProfileActivation,
+}
+
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "lowercase")]
 pub enum JobStatus {
@@ -19,24 +26,61 @@ pub enum JobStatus {
 #[derive(Debug, Clone, Serialize)]
 pub struct Job {
     pub id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub job_type: Option<JobType>,
     pub status: JobStatus,
-    pub index_name: String,
+    pub index_name: Option<String>,
+    pub profile_id: Option<String>,
     pub total_chunks: usize,
-    /// How many chunks have been embedded so far.
-    /// Also serialized as "processed" for backward-compat with monitoring scripts.
-    pub embedded_chunks: usize,
     #[serde(rename = "processed")]
     pub processed_alias: usize,
-    pub total: usize,
     pub progress_pct: f32,
     pub storage_key: Option<String>,
     pub error: Option<String>,
     pub started_at: String,
     pub completed_at: Option<String>,
-    pub chunks_per_sec: f32,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<serde_json::Value>,
+}
+
+impl Job {
+    pub fn new_embed(index_name: &str, total_chunks: usize) -> Self {
+        Self {
+            id: format!("job-{}", chrono::Utc::now().timestamp_millis()),
+            job_type: Some(JobType::Embed),
+            status: JobStatus::Running,
+            index_name: Some(index_name.to_string()),
+            profile_id: None,
+            total_chunks,
+            processed_alias: 0,
+            progress_pct: 0.0,
+            storage_key: None,
+            error: None,
+            started_at: chrono::Utc::now().to_rfc3339(),
+            completed_at: None,
+            result: None,
+        }
+    }
+
+    pub fn new_profile_activation(profile_id: &str) -> Self {
+        Self {
+            id: format!("job-{}", chrono::Utc::now().timestamp_millis()),
+            job_type: Some(JobType::ProfileActivation),
+            status: JobStatus::Running,
+            index_name: None,
+            profile_id: Some(profile_id.to_string()),
+            total_chunks: 0,
+            processed_alias: 0,
+            progress_pct: 0.0,
+            storage_key: None,
+            error: None,
+            started_at: chrono::Utc::now().to_rfc3339(),
+            completed_at: None,
+            result: None,
+        }
+    }
 }
 
-/// Shared progress tracker that background tasks update.
 #[derive(Clone)]
 pub struct JobTracker {
     jobs: Arc<RwLock<HashMap<String, Job>>>,
@@ -49,57 +93,42 @@ impl JobTracker {
         }
     }
 
-    /// Register a new job. Returns the job ID.
-    pub async fn create(&self, index_name: &str, total_chunks: usize) -> String {
-        let id = format!("job-{}", chrono::Utc::now().timestamp_millis());
-        let job = Job {
-            id: id.clone(),
-            status: JobStatus::Running,
-            index_name: index_name.to_string(),
-            total_chunks,
-            embedded_chunks: 0,
-            processed_alias: 0,
-            total: total_chunks,
-            progress_pct: 0.0,
-            storage_key: None,
-            error: None,
-            started_at: chrono::Utc::now().to_rfc3339(),
-            completed_at: None,
-            chunks_per_sec: 0.0,
-        };
+    pub async fn create_embed(&self, index_name: &str, total_chunks: usize) -> String {
+        let job = Job::new_embed(index_name, total_chunks);
+        let id = job.id.clone();
         self.jobs.write().await.insert(id.clone(), job);
         id
     }
 
-    /// Update progress.
-    pub async fn update_progress(&self, id: &str, embedded: usize, rate: f32) {
+    pub async fn create_profile_activation(&self, profile_id: &str) -> String {
+        let job = Job::new_profile_activation(profile_id);
+        let id = job.id.clone();
+        self.jobs.write().await.insert(id.clone(), job);
+        id
+    }
+
+    pub async fn update_embed_progress(&self, id: &str, embedded: usize, _rate: f32) {
         let mut jobs = self.jobs.write().await;
         if let Some(job) = jobs.get_mut(id) {
-            job.embedded_chunks = embedded;
-            job.processed_alias = embedded;  // keep alias in sync
+            job.processed_alias = embedded;
             job.progress_pct = if job.total_chunks > 0 {
                 (embedded as f32 / job.total_chunks as f32) * 100.0
             } else {
                 0.0
             };
-            job.chunks_per_sec = rate;
         }
     }
 
-    /// Mark job as completed.
-    pub async fn complete(&self, id: &str, storage_key: String) {
+    pub async fn complete(&self, id: &str, result: Option<serde_json::Value>) {
         let mut jobs = self.jobs.write().await;
         if let Some(job) = jobs.get_mut(id) {
             job.status = JobStatus::Completed;
-            job.embedded_chunks = job.total_chunks;
-            job.processed_alias = job.total_chunks;
             job.progress_pct = 100.0;
-            job.storage_key = Some(storage_key);
             job.completed_at = Some(chrono::Utc::now().to_rfc3339());
+            job.result = result;
         }
     }
 
-    /// Mark job as failed.
     pub async fn fail(&self, id: &str, error: String) {
         let mut jobs = self.jobs.write().await;
         if let Some(job) = jobs.get_mut(id) {
@@ -109,12 +138,10 @@ impl JobTracker {
         }
     }
 
-    /// Get job status.
     pub async fn get(&self, id: &str) -> Option<Job> {
         self.jobs.read().await.get(id).cloned()
     }
 
-    /// List all jobs.
     pub async fn list(&self) -> Vec<Job> {
         self.jobs.read().await.values().cloned().collect()
     }

crates/vectord/src/lib.rs

@@ -8,6 +8,7 @@ pub mod hnsw;
 pub mod index_registry;
 pub mod jobs;
 pub mod playbook_memory;
+pub mod doc_drift;
 pub mod promotion;
 pub mod refresh;
 pub mod store;

crates/vectord/src/playbook_memory.rs

@@ -398,6 +398,64 @@ impl PlaybookMemory {
         Ok(touched)
     }
 
+    /// Phase 45 slice 3 — stamp `doc_drift_flagged_at` on a playbook.
+    /// Idempotent: if already flagged and not yet reviewed, this is a
+    /// no-op. Callers should check the return value: Ok(true) means we
+    /// made a new flag, Ok(false) means already flagged (unreviewed) or
+    /// playbook not found, Err means persist failed.
+    pub async fn flag_doc_drift(&self, playbook_id: &str) -> Result<bool, String> {
+        let mut touched = false;
+        {
+            let mut state = self.state.write().await;
+            for e in state.entries.iter_mut() {
+                if e.playbook_id == playbook_id {
+                    // Skip if already flagged + unreviewed (avoid
+                    // churning the timestamp on every re-check).
+                    if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                        break;
+                    }
+                    e.doc_drift_flagged_at = Some(chrono::Utc::now().to_rfc3339());
+                    e.doc_drift_reviewed_at = None; // fresh flag clears any stale review
+                    touched = true;
+                    break;
+                }
+            }
+        }
+        if touched {
+            self.persist().await?;
+        }
+        Ok(touched)
+    }
+
+    /// Phase 45 slice 3 — stamp `doc_drift_reviewed_at` on a playbook,
+    /// re-admitting it to boost calculation. Idempotent.
+    pub async fn resolve_doc_drift(&self, playbook_id: &str) -> Result<bool, String> {
+        let mut touched = false;
+        {
+            let mut state = self.state.write().await;
+            for e in state.entries.iter_mut() {
+                if e.playbook_id == playbook_id && e.doc_drift_flagged_at.is_some()
+                    && e.doc_drift_reviewed_at.is_none()
+                {
+                    e.doc_drift_reviewed_at = Some(chrono::Utc::now().to_rfc3339());
+                    touched = true;
+                    break;
+                }
+            }
+        }
+        if touched {
+            self.persist().await?;
+        }
+        Ok(touched)
+    }
+
+    /// Read-only: get a playbook entry by id. Used by the drift-check
+    /// handler to read doc_refs without exposing the full state lock.
+    pub async fn get_entry(&self, playbook_id: &str) -> Option<PlaybookEntry> {
+        let state = self.state.read().await;
+        state.entries.iter().find(|e| e.playbook_id == playbook_id).cloned()
+    }
+
     /// Phase 25 — retire every entry matching (city, state) whose
     /// schema_fingerprint doesn't match the current one. Entries with
     /// no fingerprint (legacy) are skipped — caller can use
@@ -785,6 +843,13 @@ impl PlaybookMemory {
             .filter(|e| {
                 if e.retired_at.is_some() { return false; }
                 if e.superseded_at.is_some() { return false; }
+                // Phase 45 slice 3 — entries flagged for doc drift and
+                // not yet human-reviewed are excluded from boost. The
+                // flag is the same strength of exclusion as retirement
+                // for this purpose; reviewing (via /resolve) re-admits.
+                if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                    return false;
+                }
                 if let Some(vu) = &e.valid_until {
                     if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(vu) {
                         if now > parsed.with_timezone(&chrono::Utc) { return false; }
@@ -815,6 +880,13 @@ impl PlaybookMemory {
                 .filter(|e| {
                     if e.retired_at.is_some() { return false; }
                     if e.superseded_at.is_some() { return false; }
+                    // Phase 45 slice 3 — same drift exclusion as the
+                    // non-geo path above. Keeps the two filter paths
+                    // consistent so a flagged entry is invisible to
+                    // BOTH geo-targeted and global boost queries.
+                    if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                        return false;
+                    }
                     if let Some(vu) = &e.valid_until {
                         if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(vu) {
                             if now > parsed.with_timezone(&chrono::Utc) { return false; }
@@ -1795,6 +1867,86 @@ mod upsert_tests {
         assert_eq!(snap[0].doc_refs[0].tool, "docker");
     }
 
+    // ─── Phase 45 slice 3 — doc_drift flag / resolve / boost exclusion ───
+
+    #[tokio::test]
+    async fn flag_doc_drift_stamps_timestamp_and_persists() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+        let touched = pm.flag_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(touched, "first flag should return true");
+
+        let fetched = pm.get_entry(&e.playbook_id).await.unwrap();
+        assert!(fetched.doc_drift_flagged_at.is_some(), "flag_doc_drift should stamp the timestamp");
+        assert!(fetched.doc_drift_reviewed_at.is_none(), "reviewed_at starts None on fresh flag");
+    }
+
+    #[tokio::test]
+    async fn flag_doc_drift_is_idempotent_on_already_flagged() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-01T00:00:00Z".into());
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+        let touched = pm.flag_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(!touched, "already-flagged, unreviewed entry shouldn't re-stamp");
+    }
+
+    #[tokio::test]
+    async fn resolve_doc_drift_clears_flag_admission_gate() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-21T00:00:00Z".into());
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+
+        let resolved = pm.resolve_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(resolved);
+        let fetched = pm.get_entry(&e.playbook_id).await.unwrap();
+        assert!(fetched.doc_drift_reviewed_at.is_some());
+
+        // Second resolve is a no-op
+        let again = pm.resolve_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(!again);
+    }
+
+    #[tokio::test]
+    async fn boost_excludes_flagged_unreviewed_entries() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let e_clean = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        let mut e_flagged = mk("fill: Welder x1 in Nashville, TN", "2026-04-21", &["Bob"]);
+        e_flagged.doc_drift_flagged_at = Some("2026-04-22T00:00:00Z".into());
+        e_flagged.doc_drift_reviewed_at = None;
+        pm.set_entries(vec![e_clean, e_flagged]).await.unwrap();
+
+        let boosts = pm
+            .compute_boost_for_filtered_with_role(
+                &[1.0, 0.0, 0.0], 10, 0.5,
+                Some(("Nashville", "TN")), None,
+            ).await;
+
+        let keys: Vec<_> = boosts.keys().map(|(_, _, n)| n.clone()).collect();
+        assert!(keys.contains(&"Alice".to_string()));
+        assert!(!keys.contains(&"Bob".to_string()), "flagged+unreviewed entry leaked into boost");
+    }
+
+    #[tokio::test]
+    async fn boost_re_admits_resolved_entries() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-22T00:00:00Z".into());
+        e.doc_drift_reviewed_at = Some("2026-04-22T00:01:00Z".into()); // human reviewed
+        pm.set_entries(vec![e]).await.unwrap();
+
+        let boosts = pm
+            .compute_boost_for_filtered_with_role(
+                &[1.0, 0.0, 0.0], 10, 0.5,
+                Some(("Nashville", "TN")), None,
+            ).await;
+
+        let keys: Vec<_> = boosts.keys().map(|(_, _, n)| n.clone()).collect();
+        assert!(keys.contains(&"Alice".to_string()), "resolved entry should re-enter boost pool");
+    }
+
     #[tokio::test]
     async fn update_refreshes_valid_until_when_caller_provides_one() {
         let pm = PlaybookMemory::new(Arc::new(InMemory::new()));

crates/vectord/src/service.rs

@@ -7,6 +7,7 @@ use axum::{
 };
 use object_store::ObjectStore;
 use serde::{Deserialize, Serialize};
+use serde_json::json;
 use std::sync::Arc;
 
 use aibridge::client::{AiClient, EmbedRequest, GenerateRequest};
@@ -133,6 +134,9 @@ pub fn router(state: VectorState) -> Router {
         .route("/playbook_memory/revise", post(revise_playbook_memory))
         .route("/playbook_memory/history/{id}", get(playbook_memory_history))
         .route("/playbook_memory/status", get(playbook_memory_status))
+        // Phase 45 slice 3 — doc drift detection + human re-admission.
+        .route("/playbook_memory/doc_drift/check/{id}", post(check_doc_drift))
+        .route("/playbook_memory/doc_drift/resolve/{id}", post(resolve_doc_drift))
         .with_state(state)
 }
 
@@ -193,7 +197,7 @@ async fn create_index(
     let bucket = req.bucket.clone().unwrap_or_else(|| "primary".to_string());
 
     // Create job and return immediately
-    let job_id = state.job_tracker.create(&index_name, n_chunks).await;
+    let job_id = state.job_tracker.create_embed(&index_name, n_chunks).await;
     tracing::info!("job {job_id}: indexing '{}' — {} docs → {} chunks (background)", index_name, n_docs, n_chunks);
 
     // Spawn supervised dual-pipeline embedding
@@ -237,7 +241,7 @@ async fn create_index(
                 };
                 let _ = registry.register(meta).await;
 
-                tracker.complete(&jid, key).await;
+                tracker.complete(&jid, Some(json!({ "storage_key": key }))).await;
                 tracing::info!("job {jid}: completed — {n_chunks} chunks in {elapsed:.0}s ({rate:.0}/sec)");
             }
             Err(e) => {
@@ -479,7 +483,7 @@ async fn _run_embedding_job_legacy(
         // Update progress
         let elapsed = start.elapsed().as_secs_f32();
         let rate = if elapsed > 0.0 { all_vectors.len() as f32 / elapsed } else { 0.0 };
-        tracker.update_progress(job_id, all_vectors.len(), rate).await;
+        tracker.update_embed_progress(job_id, all_vectors.len(), rate).await;
 
         // Log every 100 batches
         if (i + 1) % 100 == 0 {
@@ -1368,239 +1372,218 @@ async fn activate_profile(
     State(state): State<VectorState>,
     Path(profile_id): Path<String>,
 ) -> impl IntoResponse {
-    let t0 = std::time::Instant::now();
-
+    tracing::info!("[activate_profile] START profile_id={}", profile_id);
     let profile = match state.catalog.get_profile(&profile_id).await {
         Some(p) => p,
         None => return Err((StatusCode::NOT_FOUND, format!("profile not found: {profile_id}"))),
     };
 
-    let mut warmed = Vec::new();
-    let mut failures = Vec::new();
-    let mut total_vectors = 0usize;
+    let job_id = state.job_tracker.create_profile_activation(&profile_id).await;
+    let job_id_for_response = job_id.clone();
+    let tracker = state.job_tracker.clone();
+    let catalog = state.catalog.clone();
+    let index_registry = state.index_registry.clone();
+    let bucket_registry = state.bucket_registry.clone();
+    let lance = state.lance.clone();
+    let embedding_cache = state.embedding_cache.clone();
+    let hnsw_store = state.hnsw_store.clone();
+    let promotion_registry = state.promotion_registry.clone();
+    let ai_client = state.ai_client.clone();
+    let active_profile = state.active_profile.clone();
+    let profile_name = profile.ollama_name.clone();
+    let profile_id_clone = profile.id.clone();
+    let profile_bucket = profile.bucket.clone();
+    let profile_bound = profile.bound_datasets.clone();
+    let profile_hnsw = profile.hnsw_config.clone();
+    let profile_backend = profile.vector_backend.clone();
+    let profile_full = profile.clone();
 
-    // Phase 17 / C: VRAM-aware swap. If another profile currently holds
-    // the GPU and uses a DIFFERENT Ollama model than the one being
-    // activated, unload it first (keep_alive=0). Same-model activations
-    // skip the unload — no point churning a model that's already loaded.
-    let previous_slot = {
-        let guard = state.active_profile.read().await;
-        guard.clone()
-    };
-    if let Some(prev) = &previous_slot {
-        if prev.ollama_name != profile.ollama_name {
-            match state.ai_client.unload_model(&prev.ollama_name).await {
-                Ok(_) => tracing::info!(
-                    "profile swap: unloaded '{}' ({} -> {})",
-                    prev.ollama_name, prev.profile_id, profile.id,
-                ),
-                Err(e) => failures.push(format!(
-                    "unload previous model '{}': {e}", prev.ollama_name,
-                )),
-            }
-        }
-    }
+    tokio::spawn(async move {
+        let t0 = std::time::Instant::now();
+        let mut warmed = Vec::new();
+        let mut failures = Vec::new();
+        let mut total_vectors = 0usize;
+        let job_id = job_id;
 
-    // Federation layer 2: if this profile declares its own bucket and
-    // that bucket isn't registered yet, auto-provision it under the
-    // configured profile_root. This is the moment a "dormant" profile
-    // becomes live — its bucket exists and is readable/writable.
-    if let Some(bucket_name) = profile.bucket.clone() {
-        if !state.bucket_registry.contains(&bucket_name) {
-            let root = format!(
-                "{}/{}",
-                state.bucket_registry.profile_root().trim_end_matches('/'),
-                bucket_name.replace(':', "_"),
-            );
-            let bc = shared::config::BucketConfig {
-                name: bucket_name.clone(),
-                backend: "local".to_string(),
-                root: Some(root.clone()),
-                bucket: None,
-                region: None,
-                endpoint: None,
-                secret_ref: None,
-            };
-            match state.bucket_registry.add_bucket(bc).await {
-                Ok(info) => {
-                    tracing::info!(
-                        "profile '{}' activated bucket '{}' (root={}, reachable={})",
-                        profile.id, bucket_name, root, info.reachable,
-                    );
-                }
-                Err(e) => {
-                    failures.push(format!(
-                        "auto-provision bucket '{}': {}", bucket_name, e,
-                    ));
+        let previous_slot = {
+            let guard = active_profile.read().await;
+            guard.clone()
+        };
+        if let Some(prev) = &previous_slot {
+            if prev.ollama_name != profile_name {
+                match ai_client.unload_model(&prev.ollama_name).await {
+                    Ok(_) => tracing::info!(
+                        "profile swap: unloaded '{}' ({} -> {})",
+                        prev.ollama_name, prev.profile_id, profile_id_clone,
+                    ),
+                    Err(e) => failures.push(format!("unload previous model '{}': {e}", prev.ollama_name)),
                 }
             }
         }
-    }
 
-    let all_indexes = state.index_registry.list(None, None).await;
-    let use_lance = profile.vector_backend == shared::types::VectorBackend::Lance;
-
-    for binding in &profile.bound_datasets {
-        let matched: Vec<_> = all_indexes
-            .iter()
-            .filter(|m| &m.source == binding)
-            .collect();
-        if matched.is_empty() {
-            failures.push(format!(
-                "no vector index found for binding '{}'", binding,
-            ));
-            continue;
-        }
-        for meta in matched {
-            if use_lance {
-                // --- Lance activation path ---
-                // Ensure a Lance dataset exists for this index. If it
-                // doesn't, auto-migrate from the Parquet blob. Then
-                // ensure an IVF_PQ index is built.
-                let bucket = meta.bucket.clone();
-                let lance_store = match state.lance.store_for_new(&meta.index_name, &bucket).await {
-                    Ok(s) => s,
-                    Err(e) => {
-                        failures.push(format!("{}: lance store init: {e}", meta.index_name));
-                        continue;
-                    }
+        if let Some(bucket_name) = profile_bucket.clone() {
+            if !bucket_registry.contains(&bucket_name) {
+                let root = format!(
+                    "{}/{}",
+                    bucket_registry.profile_root().trim_end_matches('/'),
+                    bucket_name.replace(':', "_"),
+                );
+                let bc = shared::config::BucketConfig {
+                    name: bucket_name.clone(),
+                    backend: "local".to_string(),
+                    root: Some(root.clone()),
+                    bucket: None,
+                    region: None,
+                    endpoint: None,
+                    secret_ref: None,
                 };
-                let count = lance_store.count().await.unwrap_or(0);
-                if count == 0 {
-                    // Auto-migrate from existing Parquet.
-                    let pq_store = match state.bucket_registry.get(&bucket) {
+                match bucket_registry.add_bucket(bc).await {
+                    Ok(info) => {
+                        tracing::info!(
+                            "profile '{}' activated bucket '{}' (root={}, reachable={})",
+                            profile_id_clone, bucket_name, root, info.reachable,
+                        );
+                    }
+                    Err(e) => failures.push(format!("auto-provision bucket '{}': {}", bucket_name, e)),
+                }
+            }
+        }
+
+        let all_indexes = index_registry.list(None, None).await;
+        let use_lance = profile_backend == shared::types::VectorBackend::Lance;
+
+        for binding in &profile_bound {
+            let matched: Vec<_> = all_indexes
+                .iter()
+                .filter(|m| &m.source == binding)
+                .collect();
+            if matched.is_empty() {
+                failures.push(format!("no vector index found for binding '{}'", binding));
+                continue;
+            }
+            for meta in matched {
+                if use_lance {
+                    let bucket = meta.bucket.clone();
+                    let lance_store = match lance.store_for_new(&meta.index_name, &bucket).await {
                         Ok(s) => s,
-                        Err(e) => { failures.push(format!("{}: bucket: {e}", meta.index_name)); continue; }
+                        Err(e) => { failures.push(format!("{}: lance store init: {e}", meta.index_name)); continue; }
                     };
-                    match storaged::ops::get(&pq_store, &meta.storage_key).await {
-                        Ok(bytes) => {
-                            let build_t = std::time::Instant::now();
-                            match lance_store.migrate_from_parquet_bytes(&bytes).await {
-                                Ok(ms) => {
-                                    total_vectors += ms.rows_written;
-                                    tracing::info!(
-                                        "lance auto-migrate '{}': {} rows in {:.2}s",
-                                        meta.index_name, ms.rows_written, ms.duration_secs,
-                                    );
-                                    warmed.push(WarmedIndex {
-                                        index_name: meta.index_name.clone(),
-                                        source: meta.source.clone(),
-                                        vectors: ms.rows_written,
-                                        hnsw_build_secs: build_t.elapsed().as_secs_f32(),
-                                    });
+                    let count = lance_store.count().await.unwrap_or(0);
+                    if count == 0 {
+                        let pq_store = match bucket_registry.get(&bucket) {
+                            Ok(s) => s,
+                            Err(e) => { failures.push(format!("{}: bucket: {e}", meta.index_name)); continue; }
+                        };
+                        match storaged::ops::get(&pq_store, &meta.storage_key).await {
+                            Ok(bytes) => {
+                                let build_t = std::time::Instant::now();
+                                match lance_store.migrate_from_parquet_bytes(&bytes).await {
+                                    Ok(ms) => {
+                                        total_vectors += ms.rows_written;
+                                        tracing::info!("lance auto-migrate '{}': {} rows in {:.2}s", meta.index_name, ms.rows_written, ms.duration_secs);
+                                        warmed.push(WarmedIndex {
+                                            index_name: meta.index_name.clone(),
+                                            source: meta.source.clone(),
+                                            vectors: ms.rows_written,
+                                            hnsw_build_secs: build_t.elapsed().as_secs_f32(),
+                                        });
+                                    }
+                                    Err(e) => failures.push(format!("{}: lance migrate: {e}", meta.index_name)),
                                 }
-                                Err(e) => failures.push(format!("{}: lance migrate: {e}", meta.index_name)),
                             }
+                            Err(e) => failures.push(format!("{}: read parquet: {e}", meta.index_name)),
                         }
-                        Err(e) => failures.push(format!("{}: read parquet: {e}", meta.index_name)),
-                    }
-                } else {
-                    total_vectors += count;
-                    warmed.push(WarmedIndex {
-                        index_name: meta.index_name.clone(),
-                        source: meta.source.clone(),
-                        vectors: count,
-                        hnsw_build_secs: 0.0,
-                    });
-                }
-                // Ensure IVF_PQ vector index exists.
-                if !lance_store.has_vector_index().await.unwrap_or(false) {
-                    match lance_store.build_index(316, 8, 48).await {
-                        Ok(ix) => tracing::info!(
-                            "lance auto-index '{}': IVF_PQ built in {:.1}s",
-                            meta.index_name, ix.build_time_secs,
-                        ),
-                        Err(e) => failures.push(format!("{}: lance IVF_PQ build: {e}", meta.index_name)),
-                    }
-                }
-                // Ensure scalar btree on doc_id for O(log N) random fetch.
-                if !lance_store.has_scalar_index("doc_id").await.unwrap_or(false) {
-                    match lance_store.build_scalar_index("doc_id").await {
-                        Ok(ix) => tracing::info!(
-                            "lance auto-index '{}': doc_id btree built in {:.2}s",
-                            meta.index_name, ix.build_time_secs,
-                        ),
-                        Err(e) => failures.push(format!("{}: lance doc_id btree: {e}", meta.index_name)),
-                    }
-                }
-            } else {
-                // --- Parquet + HNSW activation path (existing) ---
-                let embeddings = match state.embedding_cache.get_or_load(&meta.index_name).await {
-                    Ok(arc) => arc,
-                    Err(e) => {
-                        failures.push(format!("{}: load failed: {}", meta.index_name, e));
-                        continue;
-                    }
-                };
-                total_vectors += embeddings.len();
-
-                let profile_default = trial::HnswConfig {
-                    ef_construction: profile.hnsw_config.ef_construction,
-                    ef_search: profile.hnsw_config.ef_search,
-                    seed: profile.hnsw_config.seed,
-                };
-                let cfg = state
-                    .promotion_registry
-                    .config_or(&meta.index_name, profile_default)
-                    .await;
-                let build_t = std::time::Instant::now();
-                match state
-                    .hnsw_store
-                    .build_index_with_config(&meta.index_name, (*embeddings).clone(), &cfg)
-                    .await
-                {
-                    Ok(_) => {
+                    } else {
+                        total_vectors += count;
                         warmed.push(WarmedIndex {
                             index_name: meta.index_name.clone(),
                             source: meta.source.clone(),
-                            vectors: embeddings.len(),
-                            hnsw_build_secs: build_t.elapsed().as_secs_f32(),
+                            vectors: count,
+                            hnsw_build_secs: 0.0,
                         });
                     }
-                    Err(e) => {
-                        failures.push(format!("{}: HNSW build failed: {}", meta.index_name, e));
+                    if !lance_store.has_vector_index().await.unwrap_or(false) {
+                        match lance_store.build_index(316, 8, 48).await {
+                            Ok(ix) => tracing::info!("lance auto-index '{}': IVF_PQ built in {:.1}s", meta.index_name, ix.build_time_secs),
+                            Err(e) => failures.push(format!("{}: lance IVF_PQ build: {e}", meta.index_name)),
+                        }
+                    }
+                    if !lance_store.has_scalar_index("doc_id").await.unwrap_or(false) {
+                        match lance_store.build_scalar_index("doc_id").await {
+                            Ok(ix) => tracing::info!("lance auto-index '{}': doc_id btree built in {:.2}s", meta.index_name, ix.build_time_secs),
+                            Err(e) => failures.push(format!("{}: lance doc_id btree: {e}", meta.index_name)),
+                        }
+                    }
+                } else {
+                    let embeddings = match embedding_cache.get_or_load(&meta.index_name).await {
+                        Ok(arc) => arc,
+                        Err(e) => { failures.push(format!("{}: load failed: {}", meta.index_name, e)); continue; }
+                    };
+                    total_vectors += embeddings.len();
+
+                    let profile_default = trial::HnswConfig {
+                        ef_construction: profile_hnsw.ef_construction,
+                        ef_search: profile_hnsw.ef_search,
+                        seed: profile_hnsw.seed,
+                    };
+                    let cfg = promotion_registry
+                        .config_or(&meta.index_name, profile_default)
+                        .await;
+                    let build_t = std::time::Instant::now();
+                    match hnsw_store
+                        .build_index_with_config(&meta.index_name, (*embeddings).clone(), &cfg)
+                        .await
+                    {
+                        Ok(_) => {
+                            warmed.push(WarmedIndex {
+                                index_name: meta.index_name.clone(),
+                                source: meta.source.clone(),
+                                vectors: embeddings.len(),
+                                hnsw_build_secs: build_t.elapsed().as_secs_f32(),
+                            });
+                        }
+                        Err(e) => failures.push(format!("{}: HNSW build failed: {}", meta.index_name, e)),
                     }
                 }
             }
         }
-    }
 
-    // Preload the new profile's Ollama model proactively. Same-model
-    // re-activations are cheap (Ollama no-ops if already loaded).
-    let mut model_preloaded = false;
-    match state.ai_client.preload_model(&profile.ollama_name).await {
-        Ok(_) => {
-            model_preloaded = true;
-            tracing::info!("profile '{}' preloaded ollama model '{}'",
-                profile.id, profile.ollama_name);
+        let mut model_preloaded = false;
+        match ai_client.preload_model(&profile_name).await {
+            Ok(_) => {
+                model_preloaded = true;
+                tracing::info!("profile '{}' preloaded ollama model '{}'", profile_id_clone, profile_name);
+            }
+            Err(e) => failures.push(format!("preload ollama model '{}': {e}", profile_name)),
         }
-        Err(e) => failures.push(format!(
-            "preload ollama model '{}': {e}", profile.ollama_name,
-        )),
-    }
 
-    // Take the GPU slot.
-    {
-        let mut guard = state.active_profile.write().await;
-        *guard = Some(ActiveProfileSlot {
-            profile_id: profile.id.clone(),
-            ollama_name: profile.ollama_name.clone(),
-            activated_at: chrono::Utc::now(),
-        });
-    }
+        {
+            let mut guard = active_profile.write().await;
+            *guard = Some(ActiveProfileSlot {
+                profile_id: profile_id_clone.clone(),
+                ollama_name: profile_name.clone(),
+                activated_at: chrono::Utc::now(),
+            });
+        }
 
-    let report = ActivateReport {
-        profile_id: profile.id,
-        ollama_name: profile.ollama_name,
-        indexes_warmed: warmed,
-        failures,
-        total_vectors,
-        duration_secs: t0.elapsed().as_secs_f32(),
-        model_preloaded,
-        previous_profile: previous_slot.map(|s| s.profile_id),
-    };
+        let result = serde_json::to_value(ActivateReport {
+            profile_id: profile_id_clone,
+            ollama_name: profile_name,
+            indexes_warmed: warmed,
+            failures,
+            total_vectors,
+            duration_secs: t0.elapsed().as_secs_f32(),
+            model_preloaded,
+            previous_profile: previous_slot.map(|s| s.profile_id),
+        }).ok();
 
-    Ok(Json(report))
+        tracker.complete(&job_id, result).await;
+    });
+
+    Ok(Json(json!({
+        "job_id": job_id_for_response,
+        "message": format!("profile activation started — poll /vectors/jobs/{} for progress", job_id_for_response),
+    })))
 }
 
 /// Unload this profile's model and clear the active slot. No-op if the
@@ -2500,6 +2483,94 @@ async fn retire_playbook_memory(
         "supply either {playbook_id, reason} or {city, state, current_schema_fingerprint, reason}".into()))
 }
 
+/// Phase 45 slice 3 — POST /playbook_memory/doc_drift/check/{id}
+///
+/// Iterates the playbook's `doc_refs`, asks the context7 bridge whether
+/// each one drifted against the recorded snippet_hash. If any tool
+/// returned `drifted: true`, stamps `doc_drift_flagged_at` on the
+/// entry — which excludes it from boost (via the filter in
+/// `compute_boost_for_filtered_with_role`) until a human reviews and
+/// resolves.
+///
+/// Unknown outcomes (bridge down, tool not in context7, no snippet
+/// hash) are explicitly NOT enough to flag. Only a positive drifted=true
+/// from the bridge flips the flag.
+async fn check_doc_drift(
+    State(state): State<VectorState>,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    use crate::doc_drift::{check_all_refs, DriftCheckerConfig, DriftOutcome};
+
+    let entry = state.playbook_memory.get_entry(&id).await
+        .ok_or((StatusCode::NOT_FOUND, format!("playbook not found: {id}")))?;
+
+    if entry.doc_refs.is_empty() {
+        return Ok(Json(serde_json::json!({
+            "playbook_id": id,
+            "checked_tools": [],
+            "drifted": false,
+            "flagged": false,
+            "reason": "entry has no doc_refs — nothing to check",
+        })));
+    }
+
+    let cfg = DriftCheckerConfig::default();
+    let results = check_all_refs(&cfg, &entry.doc_refs).await;
+
+    let per_tool: Vec<serde_json::Value> = results.iter().map(|r| {
+        let (drifted, current, src, reason) = match &r.outcome {
+            DriftOutcome::Drifted { current_snippet_hash, source_url } =>
+                (true, Some(current_snippet_hash.clone()), source_url.clone(), None),
+            DriftOutcome::Unchanged =>
+                (false, None, None, None),
+            DriftOutcome::Unknown { reason } =>
+                (false, None, None, Some(reason.clone())),
+        };
+        serde_json::json!({
+            "tool": r.tool,
+            "version_seen": r.version_seen,
+            "drifted": drifted,
+            "current_snippet_hash": current,
+            "source_url": src,
+            "unknown_reason": reason,
+        })
+    }).collect();
+
+    let any_drifted = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Drifted { .. }));
+
+    let flagged = if any_drifted {
+        state.playbook_memory.flag_doc_drift(&id).await
+            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("flag: {e}")))?
+    } else {
+        false
+    };
+
+    Ok(Json(serde_json::json!({
+        "playbook_id": id,
+        "checked_tools": results.iter().map(|r| &r.tool).collect::<Vec<_>>(),
+        "drifted": any_drifted,
+        "flagged": flagged,
+        "per_tool": per_tool,
+    })))
+}
+
+/// Phase 45 slice 3 — POST /playbook_memory/doc_drift/resolve/{id}
+///
+/// Human-in-the-loop re-admission. Stamps `doc_drift_reviewed_at`.
+/// Idempotent: returns `resolved: false` if nothing changed (entry
+/// wasn't flagged, already reviewed, or doesn't exist).
+async fn resolve_doc_drift(
+    State(state): State<VectorState>,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    let resolved = state.playbook_memory.resolve_doc_drift(&id).await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("resolve: {e}")))?;
+    Ok(Json(serde_json::json!({
+        "playbook_id": id,
+        "resolved": resolved,
+    })))
+}
+
 /// Phase 27 — request body for `POST /playbook_memory/revise`. Same
 /// shape as a seed request minus `append` (revise is always
 /// append-semantics for a specific parent) plus `parent_id`. The new

crates/vectord/src/supervisor.rs

@@ -170,7 +170,7 @@ pub async fn run_supervised(
                         // Update job tracker
                         let elapsed = start_time.elapsed().as_secs_f32();
                         let rate = if elapsed > 0.0 { done as f32 / elapsed } else { 0.0 };
-                        tracker.update_progress(&jid, done, rate).await;
+                        tracker.update_embed_progress(&jid, done, rate).await;
 
                         // Update checkpoint
                         let mut ckpt = ckpt.write().await;

docs/AUDITOR_CONTEXT.md

@@ -0,0 +1,69 @@
+# Auditor Context — project preamble for fact extraction
+
+This file is read by `auditor/fact_extractor.ts` and prepended to the
+extract-facts prompt sent to llm_team. The goal: give the extractor +
+verifier enough grounding to ground domain-specific facts instead of
+marking them UNVERIFIABLE by default.
+
+Keep this short (< 400 words). Verifier only reads the first ~4KB of
+the prompt alongside the facts. Longer = noise, not signal.
+
+Update when: a new Phase lands, a crate is added/removed, the project's
+primary domain shifts (e.g. staffing → DevOps).
+
+---
+
+## What Lakehouse is
+
+Lakehouse is a Rust-first data platform over S3-compatible object
+storage. Primary use: a staffing company ingesting legacy CRM data for
+AI-powered worker matching, contract fulfillment, and playbook-driven
+coordination.
+
+Architecture: 13 Rust crates + a Python sidecar (Ollama) + TypeScript
+sub-agents (auditor, scrum_master, bot). Runs on a single server
+(Nvidia A4000, 128GB RAM). All services on localhost: gateway :3100,
+sidecar :3200, UI :3300, MCP :3700, observer :3800, MinIO :9000.
+
+## Key crates (each maps to a responsibility)
+
+- **shared** — types, Arrow helpers, PII utilities, SecretsProvider
+- **proto** — gRPC definitions
+- **storaged** — S3/MinIO I/O, AppendLog, ErrorJournal
+- **catalogd** — metadata authority (manifests, views, tombstones)
+- **queryd** — DataFusion SQL, MemTable cache, compaction
+- **ingestd** — CSV/JSON/PDF/Postgres/MySQL ingest
+- **vectord** — embeddings, HNSW index, **playbook_memory meta-index** (Phase 19+)
+- **vectord-lance** — Lance 4.0 firewall crate (separate Arrow version)
+- **journald** — append-only mutation event log
+- **aibridge** — Rust↔Python sidecar bridge, context budget + continuation
+- **gateway** — Axum HTTP :3100 + gRPC :3101 (Phase 38+ adds /v1/chat)
+- **ui** — Dioxus WASM (stale, pre-Phase-9)
+- **lance-bench** — standalone benchmark
+
+## Current architectural direction (Phase 38-44)
+
+Universal AI Control Plane: a `/v1/chat` OpenAI-compatible API that
+routes all LLM traffic through one layer for token accounting + provider
+fallback. Truth Layer + Validation Pipeline enforce staffing-domain
+invariants (worker eligibility, PII, contract rules). The Auditor
+(Phase A of cohesion plan) hard-blocks PR merges on placeholder code.
+
+## Auditor sub-agent role
+
+`auditor/` (TypeScript, Bun runtime) polls Gitea every 90s for open PRs.
+For each fresh head SHA it runs 4 checks in parallel: static (grep-style
+placeholder detection), dynamic (runs the hybrid fixture), inference
+(gpt-oss:120b cloud review with N=3 consensus + qwen3-coder:480b
+tie-breaker), and kb_query (reads `data/_kb/*.jsonl` for prior evidence).
+Verdicts post to Gitea as commit status + review comment. Findings
+append to `data/_kb/audit_lessons.jsonl` (path-agnostic signatures for
+dedup). Curated scratchpads from tree-split get routed through this
+extract-facts pipeline to populate `audit_facts.jsonl` — which is what
+you (the extractor) are currently producing.
+
+## Things that are NOT the auditor
+
+- The LLM Team UI at `/root/llm_team_ui.py` (devop.live:5000) — a separate product for human-facing multi-model experimentation
+- The scrum_master pipeline at `tests/real-world/scrum_master_pipeline.ts` — reviews files, not claims
+- The bot at `bot/` — will apply fixes, doesn't audit

docs/PHASES.md

@@ -324,6 +324,31 @@
   - Mem0-style upsert: `/seed` with `append=true` (default) routes through `upsert_entry`, which decides ADD / UPDATE / NOOP on (operation, day, city, state). Same-day re-seed merges names (union, stable order) instead of duplicating the row. Identical re-seed is a no-op. Different-day same-operation is a fresh ADD. Playbook_id stays stable on UPDATE so prior citations remain valid.
   - Letta-style hot cache: `PlaybookMemory` now holds a `geo_index: HashMap<(city_lower, state_upper), Vec<entry_idx>>` rebuilt on every mutation. Geo-filtered boost queries skip the full scan and hit the O(1) key lookup. At 1.9K entries the full scan was sub-ms; the index scales the same path to 100K+ without code changes.
   - `UpsertOutcome` enum reported back to callers — `{mode: added|updated|noop, playbook_id, merged_names?}` + `entries_after`.
+- [x] **Phase 37: Hot-swap async** (2026-04-22)
+  - Extended `JobTracker` with `JobType::ProfileActivation` + `Embed` enum variants
+  - Made `activate_profile` return immediately with `job_id`, work runs in background via `tokio::spawn`
+  - Background jobs tracked via `POST /vectors/jobs/{id}` + `GET /vectors/jobs`
+- [x] **Phase 38: Universal API Skeleton** (2026-04-23)
+  - `/v1/chat` — OpenAI-compatible POST, forwards to local Ollama or Ollama Cloud
+  - `/v1/usage` — returns `{requests, prompt_tokens, completion_tokens, total_tokens}`
+  - `/v1/sessions` — returns `{data: [], note: "Phase 38: stateless"}`
+  - Langfuse trace integration (fire-and-forget, Phase 40 early)
+  - 12 unit tests green, curl gates pass
+- [x] **Phase 39: Provider Adapter Refactor** (2026-04-23)
+  - `ProviderAdapter` trait with `chat()` + `embed()` + `unload()` + `health()`
+  - `OllamaAdapter` — wraps existing AiClient
+  - `OpenRouterAdapter` — HTTP client to openrouter.ai
+  - `provider_key()` routing by model prefix (openrouter/* → OpenRouter)
+- [x] **Phase 40: Routing & Policy Engine** (2026-04-23)
+  - `RoutingEngine` with `RouteDecision` in aibridge::routing
+  - `config/routing.toml` — rules by model_pattern, fallback chain, cost gating
+  - Per-provider usage tracking: Usage.by_provider
+  - 12 gateway tests green, curl gates pass
+- [x] **Phase 41: Profile System Expansion** (2026-04-23)
+  - `ProfileType` enum: Execution, Retrieval, Memory, Observer
+  - Per-type endpoints: `/profiles/retrieval`, `/profiles/memory`, `/profiles/observer`
+  - `profile_type` field on ModelProfile
+  - Guard fix: automated scrumaudit.py finds real issues
 - [ ] Fine-tuned domain models (Phase 25+)
 - [ ] Multi-node query distribution (only if ceilings bite)
 

tests/multi-agent/scenarios/manifest.json

@@ -1,126 +1,126 @@
 {
   "count": 20,
-  "seed": 1337,
+  "seed": 42,
   "scenarios": [
     {
-      "file": "scen_000_Great_Lakes_Mfg_Cincinnati.json",
-      "client": "Great Lakes Mfg",
-      "city": "Cincinnati",
-      "events": 4
-    },
-    {
-      "file": "scen_001_Parallel_Machining_Joliet.json",
-      "client": "Parallel Machining",
-      "city": "Joliet",
-      "events": 2
-    },
-    {
-      "file": "scen_002_Summit_Industrial_Cincinnati.json",
-      "client": "Summit Industrial",
-      "city": "Cincinnati",
-      "events": 3
-    },
-    {
-      "file": "scen_003_Pioneer_Assembly_Chicago.json",
-      "client": "Pioneer Assembly",
-      "city": "Chicago",
-      "events": 1
-    },
-    {
-      "file": "scen_004_Midway_Distribution_Columbus.json",
-      "client": "Midway Distribution",
-      "city": "Columbus",
-      "events": 2
-    },
-    {
-      "file": "scen_005_Apex_Warehouse_Cleveland.json",
-      "client": "Apex Warehouse",
-      "city": "Cleveland",
-      "events": 3
-    },
-    {
-      "file": "scen_006_Pioneer_Assembly_Flint.json",
-      "client": "Pioneer Assembly",
-      "city": "Flint",
+      "file": "scen_000_Heritage_Foods_Indianapolis.json",
+      "client": "Heritage Foods",
+      "city": "Indianapolis",
       "events": 5
     },
     {
-      "file": "scen_007_Riverfront_Steel_Toledo.json",
-      "client": "Riverfront Steel",
+      "file": "scen_001_Great_Lakes_Mfg_Madison.json",
+      "client": "Great Lakes Mfg",
+      "city": "Madison",
+      "events": 2
+    },
+    {
+      "file": "scen_002_Vanguard_Components_Lexington.json",
+      "client": "Vanguard Components",
+      "city": "Lexington",
+      "events": 2
+    },
+    {
+      "file": "scen_003_Cornerstone_Fabrication_Fort_Wayne.json",
+      "client": "Cornerstone Fabrication",
+      "city": "Fort Wayne",
+      "events": 4
+    },
+    {
+      "file": "scen_004_Horizon_Supply_Louisville.json",
+      "client": "Horizon Supply",
+      "city": "Louisville",
+      "events": 3
+    },
+    {
+      "file": "scen_005_Summit_Industrial_Akron.json",
+      "client": "Summit Industrial",
+      "city": "Akron",
+      "events": 2
+    },
+    {
+      "file": "scen_006_Centennial_Packaging_Flint.json",
+      "client": "Centennial Packaging",
+      "city": "Flint",
+      "events": 3
+    },
+    {
+      "file": "scen_007_Pioneer_Assembly_Grand_Rapids.json",
+      "client": "Pioneer Assembly",
+      "city": "Grand Rapids",
+      "events": 1
+    },
+    {
+      "file": "scen_008_Cornerstone_Fabrication_Grand_Rapids.json",
+      "client": "Cornerstone Fabrication",
+      "city": "Grand Rapids",
+      "events": 3
+    },
+    {
+      "file": "scen_009_Midway_Distribution_Fort_Wayne.json",
+      "client": "Midway Distribution",
+      "city": "Fort Wayne",
+      "events": 3
+    },
+    {
+      "file": "scen_010_Keystone_Plastics_Lexington.json",
+      "client": "Keystone Plastics",
+      "city": "Lexington",
+      "events": 5
+    },
+    {
+      "file": "scen_011_Cornerstone_Fabrication_Toledo.json",
+      "client": "Cornerstone Fabrication",
       "city": "Toledo",
       "events": 3
     },
     {
-      "file": "scen_008_Northland_Logistics_Indianapolis.json",
-      "client": "Northland Logistics",
-      "city": "Indianapolis",
-      "events": 4
-    },
-    {
-      "file": "scen_009_Parallel_Machining_Flint.json",
-      "client": "Parallel Machining",
-      "city": "Flint",
-      "events": 3
-    },
-    {
-      "file": "scen_010_Northland_Logistics_Chicago.json",
-      "client": "Northland Logistics",
-      "city": "Chicago",
-      "events": 2
-    },
-    {
-      "file": "scen_011_Heritage_Foods_Flint.json",
+      "file": "scen_012_Heritage_Foods_Gary.json",
       "client": "Heritage Foods",
-      "city": "Flint",
+      "city": "Gary",
       "events": 3
     },
     {
-      "file": "scen_012_Parallel_Machining_Kansas_City.json",
-      "client": "Parallel Machining",
-      "city": "Kansas City",
-      "events": 3
-    },
-    {
-      "file": "scen_013_Horizon_Supply_Flint.json",
-      "client": "Horizon Supply",
-      "city": "Flint",
-      "events": 3
-    },
-    {
-      "file": "scen_014_Midway_Distribution_Indianapolis.json",
-      "client": "Midway Distribution",
-      "city": "Indianapolis",
-      "events": 4
-    },
-    {
-      "file": "scen_015_Cornerstone_Fabrication_Kansas_City.json",
-      "client": "Cornerstone Fabrication",
-      "city": "Kansas City",
-      "events": 4
-    },
-    {
-      "file": "scen_016_Riverfront_Steel_Columbus.json",
+      "file": "scen_013_Riverfront_Steel_Columbus.json",
       "client": "Riverfront Steel",
       "city": "Columbus",
-      "events": 4
+      "events": 3
     },
     {
-      "file": "scen_017_Summit_Industrial_Detroit.json",
-      "client": "Summit Industrial",
-      "city": "Detroit",
+      "file": "scen_014_Keystone_Plastics_Cincinnati.json",
+      "client": "Keystone Plastics",
+      "city": "Cincinnati",
       "events": 2
     },
     {
-      "file": "scen_018_Heritage_Foods_Cincinnati.json",
-      "client": "Heritage Foods",
-      "city": "Cincinnati",
+      "file": "scen_015_Beacon_Freight_Detroit.json",
+      "client": "Beacon Freight",
+      "city": "Detroit",
       "events": 4
     },
     {
-      "file": "scen_019_Midway_Distribution_Chicago.json",
-      "client": "Midway Distribution",
-      "city": "Chicago",
+      "file": "scen_016_Parallel_Machining_Grand_Rapids.json",
+      "client": "Parallel Machining",
+      "city": "Grand Rapids",
       "events": 3
+    },
+    {
+      "file": "scen_017_Parallel_Machining_Gary.json",
+      "client": "Parallel Machining",
+      "city": "Gary",
+      "events": 3
+    },
+    {
+      "file": "scen_018_Cornerstone_Fabrication_Louisville.json",
+      "client": "Cornerstone Fabrication",
+      "city": "Louisville",
+      "events": 5
+    },
+    {
+      "file": "scen_019_Summit_Industrial_Kansas_City.json",
+      "client": "Summit Industrial",
+      "city": "Kansas City",
+      "events": 2
     }
   ]
 }

tests/real-world/enrich_prd_pipeline.ts

@@ -0,0 +1,528 @@
+// Real-world architecture stress test — 6 iterations of the full pipeline
+// against the PRD as a corpus. Goal: prove at scale what Phase 21
+// promised (context continuation + tree-split), plus Phase 19
+// compounding across iterations.
+//
+// Run: bun run tests/real-world/enrich_prd_pipeline.ts
+//
+// No mocks. No skipped layers. On any error, the test triggers
+// cloud-rescue rather than fail — it's the architecture's job to
+// recover. The test FAILS only if we can't complete 6 iterations.
+
+import { readFile, writeFile, mkdir } from "node:fs/promises";
+import { createHash } from "node:crypto";
+
+const PRD_PATH = "/home/profit/lakehouse/docs/PRD.md";
+const CHUNK_SIZE = 800;            // chars per chunk — ~200 tokens
+const CHUNK_OVERLAP = 120;
+const TOP_K_RETRIEVE = 12;         // chunks per iteration — pulled up to force overflow
+const CONTEXT_BUDGET_CHARS = 4000; // tight budget — forces tree-split on every iter
+const INJECT_FAIL_ON_ITER = 3;     // force the TASK-retry loop on iter 3
+
+// Continuation controls (per-cloud-call) — used for output-overflow.
+// Separate from the task-retry loop (per-task) — that handles errors
+// across attempts.
+const PRIMARY_MAX_TOKENS = 150;     // tight — forces truncation
+const CONTINUATION_MAX_TOKENS = 300; // each continuation doubles headroom
+const MAX_CONTINUATIONS = 6;         // max stitch pieces per cloud call
+
+// Task-level retry loop (J's clarification, 2026-04-22):
+// When a TASK errors, retry the whole task up to 6 times. Each
+// retry gets prior attempts' failures injected as learning context,
+// so attempt N+1 is informed by what N failed at. The loop caps at
+// 6 to avoid infinite spinning on genuinely unsolvable tasks.
+const MAX_TASK_RETRIES = 6;
+
+// To FORCE the retry loop on iter INJECT_FAIL_ON_ITER, cycle through
+// 5 deliberately-invalid models + 1 valid one. Attempts 1-5 will
+// 502/404 from Ollama Cloud; attempt 6 finally succeeds. Proves the
+// loop fires all 6 with compounding failure context.
+const FORCE_RETRY_MODEL_SEQUENCE = [
+  "deliberately-invalid-model-attempt-1",
+  "deliberately-invalid-model-attempt-2",
+  "deliberately-invalid-model-attempt-3",
+  "deliberately-invalid-model-attempt-4",
+  "deliberately-invalid-model-attempt-5",
+  "gpt-oss:20b", // 6th attempt succeeds
+];
+const GATEWAY = "http://localhost:3100";
+const SIDECAR = "http://localhost:3200";
+const CLOUD_MODEL = "gpt-oss:120b";
+const RESCUE_MODEL = "gpt-oss:20b";       // fallback local cloud model via sidecar
+const RUN_NONCE = Date.now().toString(36);
+const OUT_DIR = `/home/profit/lakehouse/tests/real-world/runs/${RUN_NONCE}`;
+
+// The 6 progressively-compounding questions. #6 explicitly requires
+// synthesis across prior 5 answers.
+const QUESTIONS: string[] = [
+  "Summarize the Lakehouse project's one-paragraph thesis: what problem does it solve, what's the unique approach?",
+  "How does Phase 19 playbook memory turn successful fills into a signal that boosts future rankings?",
+  "Explain the role of Phase 24 observer in the learning loop — what does it see, and what does it feed into?",
+  "What's the VRAM-aware profile swap mechanism in Phase 17, and why does it matter for multi-model serving?",
+  "How do Phase 25 validity windows and Phase 27 playbook versioning interact when a schema drifts?",
+  "Synthesize the prior 5 answers: how do the pieces (playbook memory, observer, profile swap, validity windows, versioning) compose into a system that measurably gets smarter over time? Cite specific prior answers.",
+];
+
+type Chunk = { id: string; text: string; embedding: number[]; offset: number };
+
+interface IterationResult {
+  iteration: number;
+  question: string;
+  retrieval_top_k: number;
+  context_chars_before_budget: number;
+  tree_split_fired: boolean;
+  cloud_calls_total: number;
+  continuation_retries: number;
+  rescue_triggered: boolean;
+  // Task-level retry telemetry
+  task_attempts_made: number;       // how many attempts fired (1 = first succeeded)
+  task_retry_history: Array<{ n: number; model: string; error: string }>;
+  playbook_id: string | null;
+  tokens_prompt: number;
+  tokens_completion: number;
+  citations_from_prior_iterations: string[];
+  duration_ms: number;
+  answer_preview: string;
+  errors_recovered: string[];
+}
+
+function log(msg: string) { console.log(`[enrich] ${msg}`); }
+
+function sleep(ms: number) { return new Promise(r => setTimeout(r, ms)); }
+
+function cosine(a: number[], b: number[]): number {
+  let dot = 0, na = 0, nb = 0;
+  for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; na += a[i] * a[i]; nb += b[i] * b[i]; }
+  if (na === 0 || nb === 0) return 0;
+  return dot / (Math.sqrt(na) * Math.sqrt(nb));
+}
+
+function hash(s: string): string { return createHash("sha256").update(s).digest("hex").slice(0, 10); }
+
+async function embedBatch(texts: string[]): Promise<number[][]> {
+  // Sidecar /embed accepts a list. On partial failure, retry individually.
+  const r = await fetch(`${SIDECAR}/embed`, {
+    method: "POST", headers: { "content-type": "application/json" },
+    body: JSON.stringify({ texts }),
+    signal: AbortSignal.timeout(120000),
+  });
+  if (!r.ok) throw new Error(`embed batch ${r.status}: ${await r.text()}`);
+  const j: any = await r.json();
+  return j.embeddings;
+}
+
+function chunkText(text: string): Array<{ text: string; offset: number }> {
+  const out: Array<{ text: string; offset: number }> = [];
+  let i = 0;
+  while (i < text.length) {
+    const end = Math.min(i + CHUNK_SIZE, text.length);
+    const slice = text.slice(i, end).trim();
+    if (slice.length > 50) out.push({ text: slice, offset: i });
+    if (end >= text.length) break;
+    i = end - CHUNK_OVERLAP;
+  }
+  return out;
+}
+
+async function chat(opts: {
+  provider: "ollama" | "ollama_cloud",
+  model: string,
+  messages: Array<{ role: string; content: string }>,
+  max_tokens: number,
+  think: boolean,
+}): Promise<{ content: string; prompt_tokens: number; completion_tokens: number; finish_reason: string }> {
+  const r = await fetch(`${GATEWAY}/v1/chat`, {
+    method: "POST", headers: { "content-type": "application/json" },
+    body: JSON.stringify({ ...opts }),
+    signal: AbortSignal.timeout(180000),
+  });
+  if (!r.ok) throw new Error(`/v1/chat ${r.status}: ${await r.text()}`);
+  const j: any = await r.json();
+  return {
+    content: j.choices?.[0]?.message?.content ?? "",
+    prompt_tokens: j.usage?.prompt_tokens ?? 0,
+    completion_tokens: j.usage?.completion_tokens ?? 0,
+    finish_reason: j.choices?.[0]?.finish_reason ?? "?",
+  };
+}
+
+// ─── Tree-split over oversized chunk set ──────────────────────────
+async function treeSplitSummarize(
+  chunks: Chunk[],
+  question: string,
+): Promise<{ scratchpad: string; cloud_calls: number }> {
+  // Shard into groups fitting within half the budget each.
+  const perShard = Math.max(1, Math.floor((CONTEXT_BUDGET_CHARS / 2) / CHUNK_SIZE));
+  const shards: Chunk[][] = [];
+  for (let i = 0; i < chunks.length; i += perShard) {
+    shards.push(chunks.slice(i, i + perShard));
+  }
+  log(`  tree-split: ${chunks.length} chunks → ${shards.length} shards of up to ${perShard}`);
+  let scratchpad = "";
+  let cloud_calls = 0;
+  for (let si = 0; si < shards.length; si++) {
+    const shard = shards[si];
+    const shardText = shard.map(c => `[chunk @${c.offset}]\n${c.text}`).join("\n\n");
+    const userMsg = `Question: ${question}\n\nShard ${si + 1}/${shards.length} of source material:\n\n${shardText}\n\nScratchpad so far:\n${scratchpad || "(empty)"}\n\nUpdate the scratchpad: extract only facts from THIS shard that help answer the question. Be terse. No prose.`;
+    const r = await chat({
+      provider: "ollama_cloud",
+      model: CLOUD_MODEL,
+      messages: [
+        { role: "system", content: "You maintain a concise factual scratchpad across multiple shards of source text. No prose outside the scratchpad. Each shard, append ≤80 words of relevant facts." },
+        { role: "user", content: userMsg },
+      ],
+      max_tokens: 500,
+      think: false,
+    });
+    cloud_calls += 1;
+    scratchpad += `\n--- shard ${si + 1} notes ---\n${r.content.trim()}`;
+    if (scratchpad.length > CONTEXT_BUDGET_CHARS) {
+      // truncate oldest halves
+      scratchpad = scratchpad.slice(-CONTEXT_BUDGET_CHARS);
+      log(`  tree-split: scratchpad truncated to ${scratchpad.length} chars`);
+    }
+  }
+  return { scratchpad, cloud_calls };
+}
+
+// ─── Continuable generate — up to max_continuations stitches ──────
+//
+// Two failure modes handled:
+//   A) Empty response — typically thinking model burned the budget
+//      on hidden reasoning. Retry with 2× max_tokens.
+//   B) Truncated response (finish_reason=length) — answer got cut off
+//      mid-sentence. Pass the partial back as scratchpad and ask the
+//      model to continue from where it stopped.
+//
+// Stitching: keep appending content across retries; prompt_tokens and
+// completion_tokens accumulate; finish_reason reflects the LAST call.
+// Loop exits on the first call that finishes cleanly (stop) with
+// non-empty content, OR when retries hit the cap.
+async function generateContinuable(
+  opts: Parameters<typeof chat>[0] & { max_continuations?: number },
+): Promise<{ content: string; prompt_tokens: number; completion_tokens: number; continuation_retries: number; finish_reason: string }> {
+  const maxCont = opts.max_continuations ?? 1;
+  let total = await chat(opts);
+  let retries = 0;
+  while (retries < maxCont && (total.content.length === 0 || total.finish_reason === "length")) {
+    retries += 1;
+    const mode = total.content.length === 0 ? "empty" : "truncated";
+    log(`  continuation retry ${retries}/${maxCont} (${mode}: finish=${total.finish_reason}, content=${total.content.length} chars)`);
+    // Continuation prompt — branch on failure mode:
+    //   empty  → retry with 2× tokens, same prompt (thinking budget)
+    //   length → pass the partial as assistant turn, ask to continue
+    const continuationMessages = total.content.length === 0
+      ? opts.messages
+      : [
+          ...opts.messages,
+          { role: "assistant", content: total.content },
+          { role: "user", content: "Continue from exactly where you stopped. Do not repeat. Finish the answer." },
+        ];
+    const continued = await chat({
+      ...opts,
+      max_tokens: CONTINUATION_MAX_TOKENS,
+      messages: continuationMessages,
+    });
+    total = {
+      content: total.content + continued.content,
+      prompt_tokens: total.prompt_tokens + continued.prompt_tokens,
+      completion_tokens: total.completion_tokens + continued.completion_tokens,
+      finish_reason: continued.finish_reason,
+    };
+  }
+  return { ...total, continuation_retries: retries };
+}
+
+// ─── Single iteration: retrieve → budget-check → chat → seed ─────
+async function runIteration(
+  iteration: number,
+  question: string,
+  allChunks: Chunk[],
+  priorPlaybookIds: string[],
+  priorAnswers: string[],
+): Promise<IterationResult> {
+  const started = Date.now();
+  const errorsRecovered: string[] = [];
+  log(`iter ${iteration}: "${question.slice(0, 70)}..."`);
+
+  // 1. Embed the question
+  const qEmb = (await embedBatch([question]))[0];
+
+  // 2. Retrieve top-K chunks by cosine
+  const scored = allChunks
+    .map(c => ({ c, score: cosine(qEmb, c.embedding) }))
+    .sort((a, b) => b.score - a.score)
+    .slice(0, TOP_K_RETRIEVE);
+  const chunks = scored.map(x => x.c);
+  log(`  retrieved top ${chunks.length} chunks (score range ${scored[0].score.toFixed(3)} .. ${scored[scored.length - 1].score.toFixed(3)})`);
+
+  // 3. Context budget check — tree-split if over
+  const contextChars = chunks.map(c => c.text).join("\n\n").length;
+  let contextForPrompt: string;
+  let treeSplit = false;
+  let cloudCallsTotal = 0;
+  if (contextChars > CONTEXT_BUDGET_CHARS) {
+    treeSplit = true;
+    log(`  context ${contextChars} chars > budget ${CONTEXT_BUDGET_CHARS} → tree-split`);
+    const { scratchpad, cloud_calls } = await treeSplitSummarize(chunks, question);
+    contextForPrompt = `Distilled scratchpad from ${chunks.length} source chunks (too large to fit directly):\n${scratchpad}`;
+    cloudCallsTotal += cloud_calls;
+  } else {
+    contextForPrompt = chunks.map(c => `[chunk @${c.offset}]\n${c.text}`).join("\n\n");
+  }
+
+  // 4. Seed prompt with prior iteration answers (real compounding).
+  //    Not just IDs — the model needs the CONTENT to synthesize.
+  let citationBlock = "";
+  let citationsReceived: string[] = [];
+  if (priorPlaybookIds.length > 0 && priorAnswers.length > 0) {
+    const lines = priorAnswers.map((ans, i) => {
+      const pid = priorPlaybookIds[i]?.slice(0, 12) ?? "unknown";
+      // Trim each prior answer to ~400 chars so we don't blow budget
+      return `[pb:${pid}] iter ${i + 1} answer:\n${ans.slice(0, 400)}\n`;
+    });
+    citationBlock = `\n\n═══ PRIOR ITERATIONS (compounding context) ═══\n${lines.join("\n")}═══ end prior iterations ═══\n\nYour answer MUST cite specific prior iterations using [pb:ID] notation when drawing on them. Synthesis questions require explicit cross-iteration reasoning.`;
+    citationsReceived = priorPlaybookIds.slice();
+  }
+
+  // 5. TASK-LEVEL RETRY LOOP — per J's clarification 2026-04-22.
+  //    Try the task up to MAX_TASK_RETRIES times. Each retry:
+  //    a) Picks a model (normally CLOUD_MODEL; on INJECT_FAIL_ON_ITER,
+  //       cycles through 5 invalid models + 1 valid to force full loop)
+  //    b) Injects prior attempt errors as learning context
+  //    c) If the attempt succeeds (non-empty, >100 chars), loop exits
+  //    d) Otherwise, records failure and tries again with the learning
+  //
+  //    Cap at 6 so we don't spin forever on unsolvable tasks.
+  let result: { content: string; prompt_tokens: number; completion_tokens: number; continuation_retries: number; finish_reason: string } | null = null;
+  let rescueTriggered = false;
+  const taskAttemptHistory: Array<{ n: number; model: string; error: string }> = [];
+  const forceRetries = iteration === INJECT_FAIL_ON_ITER;
+  if (forceRetries) log(`  FORCING TASK-RETRY LOOP — iter ${iteration} will cycle through 5 invalid models + 1 valid`);
+
+  for (let attempt = 1; attempt <= MAX_TASK_RETRIES; attempt++) {
+    const modelForAttempt = forceRetries
+      ? FORCE_RETRY_MODEL_SEQUENCE[attempt - 1]
+      : CLOUD_MODEL;
+    // Compose a prior-attempts learning block for attempts 2+
+    const learningBlock = taskAttemptHistory.length > 0
+      ? `\n\n═══ PRIOR ATTEMPTS THIS TASK (do NOT repeat these failures; adjust approach) ═══\n${taskAttemptHistory.map(a => `Attempt ${a.n} (model ${a.model}) failed: ${a.error.slice(0, 160)}`).join("\n")}\n═══ end prior attempts ═══\n`
+      : "";
+    log(`  task attempt ${attempt}/${MAX_TASK_RETRIES}: model=${modelForAttempt}${learningBlock ? " [with prior-failure context]" : ""}`);
+    try {
+      const r = await generateContinuable({
+        provider: "ollama_cloud",
+        model: modelForAttempt,
+        messages: [
+          { role: "system", content: "You answer questions about the Lakehouse PRD using only the provided source material and prior iteration answers. Be specific. Cite chunk offsets OR [pb:ID] markers. Write a detailed 250-word answer." },
+          { role: "user", content: `Question: ${question}\n\nSource material:\n${contextForPrompt}${citationBlock}${learningBlock}` },
+        ],
+        max_tokens: PRIMARY_MAX_TOKENS,
+        think: false,
+        max_continuations: MAX_CONTINUATIONS,
+      });
+      cloudCallsTotal += 1 + r.continuation_retries;
+      if (r.content && r.content.length > 100) {
+        // Acceptable answer — exit loop
+        result = r;
+        if (attempt > 1) {
+          log(`  task attempt ${attempt} SUCCEEDED (${r.content.length} chars) after ${attempt - 1} prior failures`);
+          rescueTriggered = true;
+        }
+        break;
+      }
+      // Thin response — count as failure with learning signal
+      const err = `thin-answer: ${r.content.length} chars, finish=${r.finish_reason}`;
+      taskAttemptHistory.push({ n: attempt, model: modelForAttempt, error: err });
+      errorsRecovered.push(`attempt ${attempt}: ${err}`);
+    } catch (e) {
+      const err = (e as Error).message;
+      taskAttemptHistory.push({ n: attempt, model: modelForAttempt, error: err });
+      errorsRecovered.push(`attempt ${attempt}: ${err.slice(0, 120)}`);
+      cloudCallsTotal += 1;
+    }
+  }
+
+  // Last-ditch: if all 6 task attempts failed, try the local fallback
+  // once more so we at least return SOMETHING. This is the "don't get
+  // caught in a loop, accept best-so-far" rule J stated explicitly.
+  if (!result) {
+    errorsRecovered.push(`all ${MAX_TASK_RETRIES} task attempts failed — local fallback`);
+    rescueTriggered = true;
+    try {
+      result = await generateContinuable({
+        provider: "ollama",
+        model: "qwen3.5:latest",
+        messages: [{ role: "user", content: `Q: ${question}\n\n${contextForPrompt.slice(0, 4000)}` }],
+        max_tokens: 300,
+        think: false,
+        max_continuations: 2,
+      });
+      cloudCallsTotal += 1 + result.continuation_retries;
+    } catch (e) {
+      // Absolute last resort — fabricate a skeleton result
+      result = {
+        content: `[task failed after ${MAX_TASK_RETRIES} retries + local fallback: ${(e as Error).message}]`,
+        prompt_tokens: 0,
+        completion_tokens: 0,
+        continuation_retries: 0,
+        finish_reason: "error",
+      };
+    }
+  }
+  if (result.content.length === 0) {
+    errorsRecovered.push("even rescue returned empty — last-ditch local fallback");
+    rescueTriggered = true;
+    result = await generateContinuable({
+      provider: "ollama",
+      model: "qwen3.5:latest",
+      messages: [{ role: "user", content: `Q: ${question}\n\n${contextForPrompt.slice(0, 4000)}` }],
+      max_tokens: 300,
+      think: false,
+    });
+    cloudCallsTotal += 1;
+  }
+
+  // 6. Seed playbook with the answer
+  let playbook_id: string | null = null;
+  try {
+    const ts = new Date().toISOString();
+    const seedOp = `TEST: enrich_prd_run_${RUN_NONCE} iter${iteration} in Corpus, PRD`;
+    const r = await fetch(`${GATEWAY}/vectors/playbook_memory/seed`, {
+      method: "POST", headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        operation: seedOp,
+        approach: `q="${question.slice(0, 80)}" context_chars=${contextChars} tree_split=${treeSplit}`,
+        context: result.content.slice(0, 600),
+        endorsed_names: [`iter${iteration}_${RUN_NONCE}`],
+        append: true,
+      }),
+      signal: AbortSignal.timeout(15000),
+    });
+    if (r.ok) {
+      const j: any = await r.json();
+      playbook_id = j.outcome?.playbook_id ?? null;
+    } else {
+      errorsRecovered.push(`seed ${r.status}: ${(await r.text()).slice(0, 100)}`);
+    }
+  } catch (e) {
+    errorsRecovered.push(`seed exception: ${(e as Error).message}`);
+  }
+
+  return {
+    iteration,
+    question,
+    retrieval_top_k: chunks.length,
+    context_chars_before_budget: contextChars,
+    tree_split_fired: treeSplit,
+    cloud_calls_total: cloudCallsTotal,
+    continuation_retries: result.continuation_retries,
+    rescue_triggered: rescueTriggered,
+    task_attempts_made: taskAttemptHistory.length + 1, // +1 for the successful attempt
+    task_retry_history: taskAttemptHistory,
+    playbook_id,
+    tokens_prompt: result.prompt_tokens,
+    tokens_completion: result.completion_tokens,
+    citations_from_prior_iterations: citationsReceived,
+    duration_ms: Date.now() - started,
+    answer_preview: result.content.slice(0, 500),
+    errors_recovered: errorsRecovered,
+  };
+}
+
+async function main() {
+  await mkdir(OUT_DIR, { recursive: true });
+  log(`run nonce: ${RUN_NONCE}`);
+  log(`output dir: ${OUT_DIR}`);
+
+  // ─── Phase 1: load, chunk, embed the PRD ───────────────────────
+  log(`loading PRD from ${PRD_PATH}`);
+  const prd = await readFile(PRD_PATH, "utf8");
+  log(`PRD: ${prd.length} chars, ${prd.split("\n").length} lines`);
+
+  const raw_chunks = chunkText(prd);
+  log(`chunked into ${raw_chunks.length} pieces (size ${CHUNK_SIZE}, overlap ${CHUNK_OVERLAP})`);
+
+  // Embed in batches of 32 to avoid sidecar overload
+  const allChunks: Chunk[] = [];
+  const BATCH = 32;
+  const t0 = Date.now();
+  for (let i = 0; i < raw_chunks.length; i += BATCH) {
+    const batch = raw_chunks.slice(i, i + BATCH);
+    const embs = await embedBatch(batch.map(b => b.text));
+    for (let j = 0; j < batch.length; j++) {
+      allChunks.push({
+        id: hash(batch[j].text),
+        text: batch[j].text,
+        embedding: embs[j].map(x => Number(x)),
+        offset: batch[j].offset,
+      });
+    }
+    log(`  embedded ${allChunks.length}/${raw_chunks.length}`);
+  }
+  log(`embedded all ${allChunks.length} chunks in ${((Date.now() - t0) / 1000).toFixed(1)}s`);
+
+  // ─── Phase 2: 6 iterations ─────────────────────────────────────
+  const results: IterationResult[] = [];
+  const priorIds: string[] = [];
+  const priorAnswers: string[] = [];
+  for (let i = 1; i <= QUESTIONS.length; i++) {
+    const q = QUESTIONS[i - 1];
+    const r = await runIteration(i, q, allChunks, priorIds, priorAnswers);
+    results.push(r);
+    if (r.playbook_id) priorIds.push(r.playbook_id);
+    priorAnswers.push(r.answer_preview);
+    log(`  → iter ${i}: ${r.errors_recovered.length} errors recovered, ${r.continuation_retries} continuations, tree-split=${r.tree_split_fired}, rescue=${r.rescue_triggered}, ${r.duration_ms}ms`);
+    await writeFile(`${OUT_DIR}/iter_${i}.json`, JSON.stringify(r, null, 2));
+  }
+  // Check whether iter 6 actually cited prior pb:IDs in its answer.
+  // Playbook IDs look like `pb-seed-<hex>` so the regex needs to allow
+  // hyphens + letters inside the brackets, not just hex chars.
+  const iter6 = results[5];
+  const citationsHonored = iter6 ? (iter6.answer_preview.match(/\[pb:[\w-]+\]/gi)?.length ?? 0) : 0;
+
+  // ─── Phase 3: summary ──────────────────────────────────────────
+  const summary = {
+    run_nonce: RUN_NONCE,
+    ran_at: new Date().toISOString(),
+    prd_chars: prd.length,
+    prd_chunks: allChunks.length,
+    iterations: results.length,
+    total_cloud_calls: results.reduce((s, r) => s + r.cloud_calls_total, 0),
+    total_continuation_retries: results.reduce((s, r) => s + r.continuation_retries, 0),
+    total_errors_recovered: results.reduce((s, r) => s + r.errors_recovered.length, 0),
+    tree_splits_fired: results.filter(r => r.tree_split_fired).length,
+    rescues_triggered: results.filter(r => r.rescue_triggered).length,
+    iter6_received_prior_ids: results[5]?.citations_from_prior_iterations.length ?? 0,
+    iter6_actually_cited_in_answer: citationsHonored,
+    iter3_task_attempts: results[2]?.task_attempts_made ?? 0,
+    iter3_task_retries: results[2]?.task_retry_history.length ?? 0,
+    max_task_attempts_any_iter: Math.max(...results.map(r => r.task_attempts_made)),
+    total_duration_ms: results.reduce((s, r) => s + r.duration_ms, 0),
+    overall: results.length === 6 && results.every(r => r.playbook_id !== null) ? "PASS" : "PARTIAL",
+  };
+  await writeFile(`${OUT_DIR}/summary.json`, JSON.stringify(summary, null, 2));
+
+  log("");
+  log(`══════ SUMMARY ${summary.overall} ══════`);
+  log(`  6 iterations, ${summary.total_cloud_calls} cloud calls, ${summary.total_errors_recovered} errors recovered`);
+  log(`  tree-splits: ${summary.tree_splits_fired}/6   continuations: ${summary.total_continuation_retries}   rescues: ${summary.rescues_triggered}`);
+  log(`  iter 6 received ${summary.iter6_received_prior_ids} prior IDs, cited ${summary.iter6_actually_cited_in_answer} [pb:...] markers in its answer`);
+  log(`  iter 3 task-retry loop: ${summary.iter3_task_attempts} attempts (${summary.iter3_task_retries} prior-failure retries before success)`);
+  log(`  total duration: ${(summary.total_duration_ms / 1000).toFixed(1)}s`);
+  log("");
+  for (const r of results) {
+    const flags = [
+      r.tree_split_fired ? "tree-split" : "",
+      r.continuation_retries > 0 ? `cont=${r.continuation_retries}` : "",
+      r.rescue_triggered ? "rescued" : "",
+      r.errors_recovered.length > 0 ? `err=${r.errors_recovered.length}` : "",
+    ].filter(Boolean).join(" ");
+    log(`  iter ${r.iteration}: ${r.tokens_prompt}+${r.tokens_completion} tok, ${r.duration_ms}ms ${flags ? `[${flags}]` : ""}`);
+  }
+  log("");
+  log(`artifacts: ${OUT_DIR}/{iter_1..6.json, summary.json}`);
+  process.exit(summary.overall === "PASS" ? 0 : 1);
+}
+
+main().catch(e => { console.error("[enrich] fatal:", e); process.exit(2); });

tests/real-world/hard_task_escalation.ts

@@ -0,0 +1,267 @@
+// Hard-task escalation test. The task is deliberately constructed so
+// that a local 7B model (qwen3.5:latest) will miss at least one of the
+// validation rules. Watch the escalation ladder:
+//   1. qwen3.5:latest  (local 7B)       — likely fails
+//   2. qwen3:latest    (local 7B)       — likely fails differently
+//   3. gpt-oss:20b     (cloud 20B)      — may fail
+//   4. gpt-oss:120b    (cloud 120B)     — should succeed
+//   5. gpt-oss:120b w/ prior-attempt errors injected — retry with context
+//   6. absolute last ditch: return best-so-far with failure annotation
+//
+// Each attempt:
+//   - Calls the model via /v1/chat
+//   - Validates the output against a strict rubric
+//   - On fail: records the specific rubric violations + the partial
+//     output, injects both into the next attempt's prompt as "here's
+//     what's wrong, fix it specifically"
+//   - On success: exit loop
+//
+// Run: bun run tests/real-world/hard_task_escalation.ts
+
+import { writeFile, mkdir } from "node:fs/promises";
+
+const GATEWAY = "http://localhost:3100";
+const OUT_DIR = `/home/profit/lakehouse/tests/real-world/runs/hard_task_${Date.now().toString(36)}`;
+const MAX_ATTEMPTS = 6;
+
+// The hard task. Specific enough that a small model will miss at
+// least one rule. Not purely knowledge-based — it's a code-generation
+// task with strict structural constraints.
+const TASK = `Write a complete Rust async function with the EXACT signature:
+
+    pub async fn check_drift_batched(refs: Vec<DocRef>) -> Result<Vec<String>, String>
+
+It must:
+1. Group refs by tool name (case-insensitive — use .to_ascii_lowercase())
+2. Issue parallel HTTP GET requests to http://localhost:3900/docs/{tool}/diff?since={snippet_hash}
+3. Use reqwest and a JoinSet/Semaphore to cap concurrent in-flight requests at 4
+4. On HTTP 5xx, retry with exponential backoff: sleep 250ms, then 500ms, then 1000ms, then give up on that tool
+5. Parse the response JSON: {"drifted": bool, ...}. Return a Vec<String> of tool names where drifted == true
+6. All errors bubble via ? or Result — NO .unwrap(), NO .expect(), NO panic!()
+7. Include rustdoc /// comments on the function and each internal helper
+
+Assume this struct is already imported:
+
+    pub struct DocRef { pub tool: String, pub snippet_hash: Option<String>, pub version_seen: String }
+
+Output ONLY the Rust code. No prose, no markdown fences, no explanation. Start directly with the /// doc comment.`;
+
+// Escalation ladder — small-local → large-local → cloud → specialist
+// cloud → trillion-param cloud. Corrected 2026-04-22 per J:
+// gpt-oss:20b is LOCAL (ollama list confirms 13 GB on disk), and the
+// final escalation tier should be kimi-k2:1t (the biggest model
+// we have access to on Ollama Cloud).
+const LADDER: Array<{ provider: "ollama" | "ollama_cloud"; model: string; note: string }> = [
+  { provider: "ollama",       model: "qwen3.5:latest", note: "local 7B" },
+  { provider: "ollama",       model: "qwen3:latest",   note: "local 7B (different) " },
+  { provider: "ollama",       model: "gpt-oss:20b",    note: "local 20B" },          // FIXED: local, not cloud
+  { provider: "ollama_cloud", model: "gpt-oss:120b",   note: "cloud 120B" },
+  { provider: "ollama_cloud", model: "devstral-2:123b", note: "cloud 123B (coding specialist)" },
+  // NOTE 2026-04-22 — J wanted Kimi as the last escalation but Kimi
+  // K2.5/K2.6 both return "this model requires a subscription" on our
+  // current Ollama Cloud key. mistral-large-3:675b is the biggest
+  // model actually provisioned on this key (verified via direct curl
+  // to ollama.com/api/generate). Upgrade path: Ollama Cloud Pro →
+  // swap this line to kimi-k2.5 or kimi-k2.6:cloud.
+  { provider: "ollama_cloud", model: "mistral-large-3:675b", note: "cloud 675B (biggest available on current key; kimi-k2.x needs pro subscription)" },
+];
+
+// Validation rubric — the answer must pass all of these to be accepted.
+interface RubricResult {
+  passed: boolean;
+  violations: string[];
+  passed_rules: string[];
+}
+
+function validate(code: string): RubricResult {
+  const violations: string[] = [];
+  const passed: string[] = [];
+
+  const check = (rule: string, ok: boolean) => { ok ? passed.push(rule) : violations.push(rule); };
+
+  check("has pub async fn check_drift_batched signature",
+    /pub\s+async\s+fn\s+check_drift_batched\s*\(/.test(code));
+  check("takes Vec<DocRef> argument",
+    /refs\s*:\s*Vec\s*<\s*DocRef\s*>/.test(code));
+  check("returns Result<Vec<String>, String>",
+    /Result\s*<\s*Vec\s*<\s*String\s*>\s*,\s*String\s*>/.test(code));
+  check("uses reqwest",
+    /\breqwest\b/i.test(code));
+  check("references JoinSet or Semaphore for concurrency",
+    /\bJoinSet\b|\bSemaphore\b/i.test(code));
+  check("bounds concurrency at 4",
+    /\b4\b/.test(code) && (/Semaphore\s*::\s*new\s*\(\s*4\b/.test(code) || /permits\s*:\s*4\b/.test(code) || /limit\s*:\s*4\b/.test(code) || /max\s*:\s*4\b/.test(code) || /capacity\s*:\s*4\b/.test(code)));
+  // Exponential backoff — models express this several ways. Accept
+  // any recognizable doubling pattern starting at 250ms. 2026-04-22:
+  // devstral-2:123b wrote `retry_delay *= 2` which my earlier regex
+  // rejected even though the code is correct. Broadening rubric to
+  // match all idiomatic doubling forms.
+  const hasSeed250 = /Duration\s*::\s*from_millis\s*\(\s*250\b/.test(code)
+                  || /millis\s*\(\s*250\b/.test(code);
+  const hasDoublingPattern = /250\s*\*\s*2/.test(code)         // 250 * 2^n literal
+                          || /<<\s*\d+/.test(code)              // bit-shift
+                          || /\.pow\s*\(/.test(code)            // 2u32.pow(attempt)
+                          || /\*=\s*2\b/.test(code)             // delay *= 2  ← was missing
+                          || /\*\s*2\s*;/.test(code)            // delay = delay * 2;
+                          || /saturating_mul\s*\(\s*2\b/.test(code); // saturating doubling
+  check("has 250ms backoff seed",
+    hasSeed250);
+  check("reaches 500ms backoff (literal or doubling from 250)",
+    /Duration\s*::\s*from_millis\s*\(\s*500\b/.test(code)
+    || /millis\s*\(\s*500\b/.test(code)
+    || (hasSeed250 && hasDoublingPattern));
+  check("reaches 1000ms backoff (literal or doubling to 1000)",
+    /Duration\s*::\s*from_millis\s*\(\s*1000\b/.test(code)
+    || /millis\s*\(\s*1000\b/.test(code)
+    || (hasSeed250 && hasDoublingPattern));
+  check("case-insensitive tool grouping (to_ascii_lowercase)",
+    /to_ascii_lowercase|to_lowercase/.test(code));
+  check("NO .unwrap() — all errors bubble via ?",
+    !/\.unwrap\s*\(\s*\)/.test(code));
+  check("NO .expect(...) — all errors bubble via ?",
+    !/\.expect\s*\(/.test(code));
+  check("NO panic!() / unimplemented!() / todo!()",
+    !/\bpanic!\s*\(|\bunimplemented!\s*\(|\btodo!\s*\(/.test(code));
+  check("has rustdoc /// comments",
+    /\/\/\//.test(code));
+  check("reasonable length (> 500 chars)",
+    code.length > 500);
+
+  return { passed: violations.length === 0, violations, passed_rules: passed };
+}
+
+function log(msg: string) { console.log(`[hard] ${msg}`); }
+
+async function chat(opts: {
+  provider: "ollama" | "ollama_cloud",
+  model: string,
+  prompt: string,
+}): Promise<{ content: string; error?: string }> {
+  try {
+    const r = await fetch(`${GATEWAY}/v1/chat`, {
+      method: "POST", headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        provider: opts.provider,
+        model: opts.model,
+        messages: [{ role: "user", content: opts.prompt }],
+        max_tokens: 2500,
+        temperature: 0.2,
+        think: false,
+      }),
+      signal: AbortSignal.timeout(240000),
+    });
+    if (!r.ok) return { content: "", error: `/v1/chat ${r.status}: ${(await r.text()).slice(0, 300)}` };
+    const j: any = await r.json();
+    return { content: j.choices?.[0]?.message?.content ?? "" };
+  } catch (e) {
+    return { content: "", error: (e as Error).message };
+  }
+}
+
+interface AttemptRecord {
+  n: number;
+  provider: string;
+  model: string;
+  duration_ms: number;
+  content_chars: number;
+  error: string | null;
+  rubric_violations: string[];
+  rubric_passed: string[];
+  accepted: boolean;
+}
+
+function extractCode(raw: string): string {
+  // Strip common fence wrappers
+  const m = raw.match(/```(?:rust)?\s*\n([\s\S]*?)```/);
+  if (m) return m[1].trim();
+  return raw.trim();
+}
+
+async function main() {
+  await mkdir(OUT_DIR, { recursive: true });
+  log(`output: ${OUT_DIR}`);
+  log(`task: ${TASK.slice(0, 120)}...`);
+  log("");
+
+  const attempts: AttemptRecord[] = [];
+  let acceptedCode: string | null = null;
+
+  for (let i = 0; i < MAX_ATTEMPTS; i++) {
+    const n = i + 1;
+    const rung = LADDER[i] ?? LADDER[LADDER.length - 1];
+
+    // Build the prompt: base task + prior failures' learning blocks
+    let priorLearning = "";
+    if (attempts.length > 0) {
+      priorLearning = `\n\n═══ PRIOR ATTEMPTS FAILED. Fix these exact issues: ═══\n`;
+      for (const a of attempts) {
+        priorLearning += `Attempt ${a.n} (${a.provider}/${a.model}, ${a.content_chars} chars) violations:\n`;
+        for (const v of a.rubric_violations) priorLearning += `  - ${v}\n`;
+        if (a.error) priorLearning += `  [error: ${a.error.slice(0, 120)}]\n`;
+      }
+      priorLearning += `═══ end prior attempts ═══\n\nDO NOT repeat the above violations. Address each one explicitly.`;
+    }
+
+    log(`attempt ${n}/${MAX_ATTEMPTS}: ${rung.provider}::${rung.model}${priorLearning ? " [w/ learning]" : ""}`);
+    const t0 = Date.now();
+    const r = await chat({ provider: rung.provider, model: rung.model, prompt: TASK + priorLearning });
+    const dur = Date.now() - t0;
+
+    const code = extractCode(r.content);
+    const rubric = code ? validate(code) : { passed: false, violations: ["empty response"], passed_rules: [] };
+
+    const record: AttemptRecord = {
+      n,
+      provider: rung.provider,
+      model: rung.model,
+      duration_ms: dur,
+      content_chars: code.length,
+      error: r.error ?? null,
+      rubric_violations: rubric.violations,
+      rubric_passed: rubric.passed_rules,
+      accepted: rubric.passed,
+    };
+    attempts.push(record);
+
+    log(`  → ${dur}ms, ${code.length} chars, ${rubric.passed_rules.length} rules passed / ${rubric.violations.length} failed${r.error ? `, err: ${r.error.slice(0, 80)}` : ""}`);
+    for (const v of rubric.violations.slice(0, 5)) log(`       ✗ ${v}`);
+
+    await writeFile(`${OUT_DIR}/attempt_${n}.txt`, code);
+    await writeFile(`${OUT_DIR}/attempt_${n}.json`, JSON.stringify(record, null, 2));
+
+    if (rubric.passed) {
+      log(`  ✅ ACCEPTED on attempt ${n}`);
+      acceptedCode = code;
+      break;
+    }
+  }
+
+  const summary = {
+    task: TASK.slice(0, 200),
+    total_attempts: attempts.length,
+    accepted: acceptedCode !== null,
+    accepted_on_attempt: acceptedCode ? attempts.findIndex(a => a.accepted) + 1 : null,
+    escalation_path: attempts.map(a => `${a.provider}/${a.model}`),
+    per_attempt_pass_counts: attempts.map(a => a.rubric_passed.length),
+    per_attempt_violation_counts: attempts.map(a => a.rubric_violations.length),
+    total_duration_ms: attempts.reduce((s, a) => s + a.duration_ms, 0),
+  };
+  await writeFile(`${OUT_DIR}/summary.json`, JSON.stringify(summary, null, 2));
+
+  log("");
+  log(`═══ RESULT ═══`);
+  log(`attempts: ${summary.total_attempts}`);
+  log(`accepted: ${summary.accepted} ${summary.accepted ? `on attempt ${summary.accepted_on_attempt}` : ""}`);
+  log(`escalation path:`);
+  for (const [i, a] of attempts.entries()) {
+    const mark = a.accepted ? "✅" : "❌";
+    log(`  ${mark} attempt ${i + 1}: ${a.provider}/${a.model} — ${a.rubric_passed.length}/${a.rubric_passed.length + a.rubric_violations.length} rules passed, ${a.duration_ms}ms`);
+  }
+  log("");
+  log(`total time: ${(summary.total_duration_ms / 1000).toFixed(1)}s`);
+  log(`artifacts: ${OUT_DIR}/{attempt_1..N.{txt,json}, summary.json}`);
+
+  process.exit(summary.accepted ? 0 : 1);
+}
+
+main().catch(e => { console.error("[hard] fatal:", e); process.exit(2); });

tests/real-world/nine_consecutive_audits.ts

@@ -0,0 +1,186 @@
+// Nine-consecutive audit runner — empirical test of the predictive-
+// compounding property. Runs the audit pipeline 9 times against the
+// same PR (each time with a new diff from Gitea), captures the
+// verdict + audit_lessons state after each run, and reports whether
+// the KB stabilizes or drifts.
+//
+// What we expect (favorable compounding):
+//   - signature_count grows sublinearly (same patterns recur, so
+//     distinct-signature count stabilizes fast)
+//   - verdict settles on a stable value after run 2-3 (first audit
+//     establishes baseline, rest repeat)
+//   - confidence stays LOW for all signatures (same PR repeatedly)
+//   - NO new recurring findings fire because confidence < 0.3 on
+//     same-PR noise (kb_index rating policy)
+//
+// What would indicate drift (the thing we want to prove DOESN'T happen):
+//   - signature_count grows linearly — each run produces new signatures
+//   - verdict oscillates (block → approve → block ...)
+//   - confidence inflates — kb_index rating escalates on repeated runs
+//
+// Run: bun run tests/real-world/nine_consecutive_audits.ts
+
+import { readFile, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+import { aggregate } from "../../auditor/kb_index.ts";
+import { getPrSnapshot } from "../../auditor/gitea.ts";
+import { auditPr } from "../../auditor/audit.ts";
+
+const REPO = "/home/profit/lakehouse";
+const AUDIT_LESSONS = `${REPO}/data/_kb/audit_lessons.jsonl`;
+const VERDICTS_DIR = `${REPO}/data/_auditor/verdicts`;
+const POLL_INTERVAL_MS = 5_000;
+const RUNS = Number(process.env.LH_AUDIT_RUNS ?? 9);
+const TARGET_PR = Number(process.env.LH_AUDIT_PR ?? 8);
+const SKIP_INFERENCE = process.env.LH_AUDITOR_SKIP_INFERENCE !== "0";
+const RESET_KB = process.env.LH_RESET_KB === "1";
+
+async function waitForVerdict(prNum: number, sha: string, deadlineMs: number): Promise<any> {
+  const short = sha.slice(0, 12);
+  const path = join(VERDICTS_DIR, `${prNum}-${short}.json`);
+  const start = Date.now();
+  while (Date.now() - start < deadlineMs) {
+    try {
+      const raw = await readFile(path, "utf8");
+      return JSON.parse(raw);
+    } catch { /* not yet */ }
+    await Bun.sleep(POLL_INTERVAL_MS);
+  }
+  throw new Error(`no verdict file after ${deadlineMs}ms: ${path}`);
+}
+
+async function captureAggState(): Promise<{ sig_count: number; max_count: number; max_confidence: number; top3: Array<{ sig: string; count: number; conf: number; summary: string }> }> {
+  const agg = await aggregate<any>(AUDIT_LESSONS, {
+    keyFn: (r) => r?.signature,
+    scopeFn: (r) => (r?.pr_number !== undefined ? `pr-${r.pr_number}` : undefined),
+  });
+  const list = Array.from(agg.values()).sort((a, b) => b.count - a.count);
+  const recurring = list.filter(r => r.count >= 2);
+  const recurringMaxCount = recurring.length > 0 ? Math.max(...recurring.map(a => a.count)) : 0;
+  const recurringMaxConf = recurring.length > 0 ? Math.max(...recurring.map(a => a.confidence)) : 0;
+  return {
+    sig_count: list.length,
+    max_count: list[0]?.count ?? 0,
+    max_confidence: recurringMaxConf,
+    recurring_max_count: recurringMaxCount,
+    top3: list.slice(0, 3).map(a => ({
+      sig: a.signature,
+      count: a.count,
+      conf: a.confidence,
+      summary: a.representative_summary.slice(0, 80),
+    })),
+  };
+}
+
+interface RunRecord {
+  run: number;
+  sha: string;
+  verdict_overall: string;
+  findings_total: number;
+  findings_block: number;
+  findings_warn: number;
+  findings_info: number;
+  audit_duration_ms: number;
+  claims_total: number;
+  claims_empirical: number;
+  kb_sig_count_after: number;
+  kb_max_count_after: number;
+  kb_max_confidence_after: number;
+  kb_recurring_max_count: number;
+}
+
+async function main() {
+  console.log(`[nine] target PR: #${TARGET_PR}`);
+  console.log(`[nine] runs:      ${RUNS}`);
+  console.log(`[nine] skip_inference: ${SKIP_INFERENCE}`);
+  console.log(`[nine] reset_kb:  ${RESET_KB}`);
+  console.log(`[nine] audit_lessons.jsonl: ${AUDIT_LESSONS}`);
+
+  if (RESET_KB) {
+    console.log("[nine] clearing audit_lessons.jsonl for clean test...");
+    await writeFile(AUDIT_LESSONS, "");
+  }
+  console.log("");
+
+  const pr = await getPrSnapshot(TARGET_PR);
+  console.log(`[nine] PR #${pr.number}: "${pr.title}" (head=${pr.head_sha.slice(0, 12)})`);
+  console.log(`[nine] files in diff: ${pr.files.length}`);
+  console.log("");
+
+  const baseline = await captureAggState();
+  console.log(`[nine] baseline: sig_count=${baseline.sig_count} max_count=${baseline.max_count} max_conf=${baseline.max_confidence.toFixed(2)}`);
+  console.log("");
+
+  const records: RunRecord[] = [];
+  for (let n = 1; n <= RUNS; n++) {
+    const t0 = Date.now();
+    console.log(`─── run ${n}/${RUNS} ───`);
+
+    const verdict = await auditPr(pr, {
+      dry_run: true,
+      skip_dynamic: true,
+      skip_inference: SKIP_INFERENCE,
+    });
+
+    console.log(`  sha ${verdict.head_sha.slice(0, 12)}`);
+    const after = await captureAggState();
+    const rec: RunRecord = {
+      run: n,
+      sha: verdict.head_sha.slice(0, 12),
+      verdict_overall: String(verdict.overall),
+      findings_total: Number(verdict.metrics?.findings_total ?? 0),
+      findings_block: Number(verdict.metrics?.findings_block ?? 0),
+      findings_warn: Number(verdict.metrics?.findings_warn ?? 0),
+      findings_info: Number(verdict.metrics?.findings_info ?? 0),
+      audit_duration_ms: Number(verdict.metrics?.audit_duration_ms ?? 0),
+      claims_total: Number(verdict.metrics?.claims_total ?? 0),
+      claims_empirical: Number(verdict.metrics?.claims_empirical ?? 0),
+      kb_sig_count_after: after.sig_count,
+      kb_max_count_after: after.max_count,
+      kb_max_confidence_after: after.max_confidence,
+      kb_recurring_max_count: after.recurring_max_count,
+    };
+    records.push(rec);
+    console.log(`  verdict=${rec.verdict_overall} findings=${rec.findings_total} (b=${rec.findings_block} w=${rec.findings_warn})`);
+    console.log(`  kb after: sig=${rec.kb_sig_count_after} max_count=${rec.kb_max_count_after} recurring_max=${rec.kb_recurring_max_count} max_conf=${rec.kb_max_confidence_after.toFixed(2)}`);
+    console.log(`  elapsed: ${((Date.now() - t0) / 1000).toFixed(1)}s`);
+    console.log("");
+  }
+
+  console.log("═══ FINAL ═══");
+  console.log("run | verdict          | find  | block  warn  info | dur_s | kb_sig  max_count  max_conf");
+  for (const r of records) {
+    console.log(
+      `  ${String(r.run).padStart(1)}  | ${r.verdict_overall.padEnd(16)} | ${String(r.findings_total).padStart(4)}  | ${String(r.findings_block).padStart(5)} ${String(r.findings_warn).padStart(5)} ${String(r.findings_info).padStart(5)}  | ${(r.audit_duration_ms / 1000).toFixed(1).padStart(5)} | ${String(r.kb_sig_count_after).padStart(6)}  ${String(r.kb_max_count_after).padStart(9)}  ${r.kb_max_confidence_after.toFixed(2)}`,
+    );
+  }
+
+  console.log("");
+  console.log("═══ COMPOUNDING PROPERTY ═══");
+  const sigDelta = records[records.length - 1].kb_sig_count_after - baseline.sig_count;
+  const maxConf = records[records.length - 1].kb_max_confidence_after;
+  const recurringMax = records[records.length - 1].kb_recurring_max_count;
+  console.log(`  signatures added over ${RUNS} runs: ${sigDelta}`);
+  console.log(`  max recurring count after run ${RUNS}: ${recurringMax} (same-PR recurrences per signature)`);
+  console.log(`  max confidence after run ${RUNS}:    ${maxConf.toFixed(2)} (expect LOW — same-PR should not inflate)`);
+
+  const verdictSet = new Set(records.map(r => r.verdict_overall));
+  if (verdictSet.size === 1) {
+    console.log(`  verdict stable: all ${RUNS} runs returned '${[...verdictSet][0]}' ✓`);
+  } else {
+    console.log(`  verdict oscillated across runs: ${[...verdictSet].join(" | ")} ✗`);
+  }
+
+  if (maxConf < 0.6 && recurringMax < 5) {
+    console.log(`  confidence policy holding: same-PR noise stays below escalation threshold ✓`);
+  } else {
+    console.log(`  ⚠ cross-cutting pattern detected (conf=${maxConf.toFixed(2)}, recurring=${recurringMax}) — kb_index policy escalated`);
+  }
+
+  const jsonOut = `${REPO}/tests/real-world/runs/nine_consecutive_${Date.now().toString(36)}.json`;
+  await Bun.write(jsonOut, JSON.stringify({ target_pr: TARGET_PR, baseline, records }, null, 2));
+  console.log("");
+  console.log(`  report: ${jsonOut}`);
+}
+
+main().catch(e => { console.error("[nine] fatal:", e); process.exit(1); });

tests/real-world/scrum_master_pipeline.ts

@@ -0,0 +1,481 @@
+// Scrum-master orchestrator — pulls git repo source + PRD + a change
+// proposal, chunks everything, hands each code piece to the proven
+// escalation ladder (small-local → big-local → cloud → specialist →
+// biggest) with learning context between attempts. Collects per-file
+// suggestions in a coherent handoff report.
+//
+// What it composes (everything below is already shipped + proven):
+//   - Chunker + embeddings (sidecar /embed, nomic-embed-text)
+//   - In-memory cosine retrieval (top-K PRD + plan chunks per file)
+//   - Escalation ladder (6 tiers, cycling on empty/error/thin-answer)
+//   - Per-attempt learning-context injection (prior failures → prompt)
+//   - Tree-split fallback when combined context exceeds budget
+//   - JSONL output per file + summary
+//
+// Deliberate scope limit: TARGET_FILES is 3 files by default. The
+// pipeline works at larger N, but at ~90s/file × 3 files = 4-5 min,
+// 15 files = 22 min. Bump via env LH_SCRUM_FILES="path1,path2,...".
+//
+// Run: bun run tests/real-world/scrum_master_pipeline.ts
+
+import { readFile, writeFile, mkdir } from "node:fs/promises";
+import { createHash } from "node:crypto";
+
+const GATEWAY = "http://localhost:3100";
+const SIDECAR = "http://localhost:3200";
+const CHUNK_SIZE = 800;
+const CHUNK_OVERLAP = 120;
+const TOP_K_CONTEXT = 5;
+const MAX_ATTEMPTS = 6;
+// Files larger than this get tree-split instead of truncated. Fixes the
+// 6KB false-positive class (model claiming a field is "missing" when
+// it exists past the context cutoff).
+const FILE_TREE_SPLIT_THRESHOLD = 6000;
+const FILE_SHARD_SIZE = 3500;
+// Appended jsonl so auditor's kb_query can surface scrum findings for
+// files touched by a PR under review. Part of cohesion plan Phase C.
+const SCRUM_REVIEWS_JSONL = "/home/profit/lakehouse/data/_kb/scrum_reviews.jsonl";
+const OUT_DIR = `/home/profit/lakehouse/tests/real-world/runs/scrum_${Date.now().toString(36)}`;
+
+const PRD_PATH = "/home/profit/lakehouse/docs/PRD.md";
+// Using CONTROL_PLANE_PRD as the "suggested changes" doc since it
+// describes the Phase 38-44 target architecture and is on main.
+// COHESION_INTEGRATION_PLAN.md is still on PR #7 branch.
+const PROPOSAL_PATH = "/home/profit/lakehouse/docs/CONTROL_PLANE_PRD.md";
+
+// Scoped target: 3 representative source files by default.
+// The scrum-master walks these in order and produces one suggestion
+// set per file. Override via env for a wider sweep.
+const DEFAULT_TARGETS = [
+  "/home/profit/lakehouse/crates/vectord/src/playbook_memory.rs",
+  "/home/profit/lakehouse/crates/vectord/src/doc_drift.rs",
+  "/home/profit/lakehouse/auditor/audit.ts",
+];
+const TARGET_FILES: string[] = process.env.LH_SCRUM_FILES
+  ? process.env.LH_SCRUM_FILES.split(",").map(s => s.trim())
+  : DEFAULT_TARGETS;
+
+const LADDER: Array<{ provider: "ollama" | "ollama_cloud"; model: string; note: string }> = [
+  { provider: "ollama",       model: "qwen3.5:latest",       note: "local 7B" },
+  { provider: "ollama",       model: "qwen3:latest",         note: "local 7B (peer)" },
+  { provider: "ollama",       model: "gpt-oss:20b",          note: "local 20B" },
+  { provider: "ollama_cloud", model: "gpt-oss:120b",         note: "cloud 120B" },
+  { provider: "ollama_cloud", model: "devstral-2:123b",      note: "cloud 123B coding specialist" },
+  { provider: "ollama_cloud", model: "mistral-large-3:675b", note: "cloud 675B last-ditch" },
+];
+
+type Chunk = { id: string; text: string; embedding: number[]; origin: string; offset: number };
+
+interface FileReview {
+  file: string;
+  file_bytes: number;
+  tree_split_fired: boolean;
+  shards_summarized: number;
+  top_prd_chunks: Array<{ origin: string; offset: number; score: number }>;
+  top_proposal_chunks: Array<{ origin: string; offset: number; score: number }>;
+  attempts_made: number;
+  attempts_history: Array<{ n: number; model: string; status: "accepted" | "thin" | "error"; chars: number; error?: string }>;
+  accepted_on: number | null;
+  escalated_to_model: string;
+  suggestions: string;
+  duration_ms: number;
+}
+
+function log(msg: string) { console.log(`[scrum] ${msg}`); }
+
+function cosine(a: number[], b: number[]): number {
+  let dot = 0, na = 0, nb = 0;
+  for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; na += a[i] * a[i]; nb += b[i] * b[i]; }
+  return na && nb ? dot / (Math.sqrt(na) * Math.sqrt(nb)) : 0;
+}
+
+function chunkText(text: string): Array<{ text: string; offset: number }> {
+  const out: Array<{ text: string; offset: number }> = [];
+  for (let i = 0; i < text.length; ) {
+    const end = Math.min(i + CHUNK_SIZE, text.length);
+    const slice = text.slice(i, end).trim();
+    if (slice.length > 60) out.push({ text: slice, offset: i });
+    if (end >= text.length) break;
+    i = end - CHUNK_OVERLAP;
+  }
+  return out;
+}
+
+async function embedBatch(texts: string[]): Promise<number[][]> {
+  const r = await fetch(`${SIDECAR}/embed`, {
+    method: "POST", headers: { "content-type": "application/json" },
+    body: JSON.stringify({ texts }),
+    signal: AbortSignal.timeout(120000),
+  });
+  if (!r.ok) throw new Error(`embed ${r.status}`);
+  return (await r.json() as any).embeddings;
+}
+
+async function chat(opts: {
+  provider: "ollama" | "ollama_cloud",
+  model: string,
+  prompt: string,
+  max_tokens?: number,
+}): Promise<{ content: string; error?: string; prompt_tokens: number; completion_tokens: number }> {
+  try {
+    const r = await fetch(`${GATEWAY}/v1/chat`, {
+      method: "POST", headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        provider: opts.provider,
+        model: opts.model,
+        messages: [{ role: "user", content: opts.prompt }],
+        max_tokens: opts.max_tokens ?? 1500,
+        temperature: 0.2,
+        think: false,
+      }),
+      signal: AbortSignal.timeout(180000),
+    });
+    if (!r.ok) return { content: "", error: `/v1/chat ${r.status}: ${(await r.text()).slice(0, 200)}`, prompt_tokens: 0, completion_tokens: 0 };
+    const j: any = await r.json();
+    return {
+      content: j.choices?.[0]?.message?.content ?? "",
+      prompt_tokens: j.usage?.prompt_tokens ?? 0,
+      completion_tokens: j.usage?.completion_tokens ?? 0,
+    };
+  } catch (e) {
+    return { content: "", error: (e as Error).message, prompt_tokens: 0, completion_tokens: 0 };
+  }
+}
+
+// Accept a file-review answer if it's substantive + structured.
+// We're not validating Rust here — we're validating that the model
+// produced a coherent suggestion set.
+function isAcceptable(answer: string): boolean {
+  if (answer.length < 200) return false;                    // too thin
+  // Must at least try a structured form — numbered list, bullets,
+  // or sections. Models that just hand-wave fail.
+  const hasStructure = /^\s*[-*]\s/m.test(answer)
+                    || /^\s*\d+\.\s/m.test(answer)
+                    || /^\s*#/m.test(answer);
+  return hasStructure;
+}
+
+function retrieveTopK(query_emb: number[], pool: Chunk[], k: number): Chunk[] {
+  return pool
+    .map(c => ({ c, score: cosine(query_emb, c.embedding) }))
+    .sort((a, b) => b.score - a.score)
+    .slice(0, k)
+    .map(x => ({ ...x.c, _score: x.score } as any));
+}
+
+// Tree-split a large file: shard it, summarize each shard against
+// the review question, merge into a scratchpad. Uses cloud because
+// the summarization step needs quality > speed. Returns the
+// scratchpad (full-file distillation) and the cloud-call count.
+async function treeSplitFile(
+  filePath: string,
+  content: string,
+): Promise<{ scratchpad: string; shards: number; cloud_calls: number }> {
+  const shards: Array<{ from: number; to: number; text: string }> = [];
+  for (let i = 0; i < content.length; i += FILE_SHARD_SIZE) {
+    const end = Math.min(i + FILE_SHARD_SIZE, content.length);
+    shards.push({ from: i, to: end, text: content.slice(i, end) });
+  }
+  let scratchpad = "";
+  let cloud_calls = 0;
+  log(`  tree-split: ${content.length} chars → ${shards.length} shards of ${FILE_SHARD_SIZE}`);
+  for (const [si, shard] of shards.entries()) {
+    const prompt = `You are summarizing ONE SHARD of a source file as part of a multi-shard review. File: ${filePath}. Shard ${si + 1}/${shards.length} (bytes ${shard.from}..${shard.to}).
+
+─────── shard source ───────
+${shard.text}
+─────── end shard ───────
+
+Scratchpad of prior shards (if empty, this is shard 1):
+${scratchpad || "(empty)"}
+
+Extract ONLY facts useful for reviewing this file against its PRD: function names + purposes, struct fields + types, invariants, edge cases, TODO markers, error-handling style. Under 150 words. No prose outside the extracted facts.`;
+    const r = await chat({
+      provider: "ollama_cloud",
+      model: "gpt-oss:120b",
+      prompt,
+      max_tokens: 400,
+    });
+    cloud_calls += 1;
+    if (r.content) {
+      scratchpad += `\n--- shard ${si + 1} (bytes ${shard.from}..${shard.to}) ---\n${r.content.trim()}`;
+    }
+  }
+  return { scratchpad, shards: shards.length, cloud_calls };
+}
+
+async function reviewFile(
+  filePath: string,
+  prd_chunks: Chunk[],
+  proposal_chunks: Chunk[],
+): Promise<FileReview> {
+  const t0 = Date.now();
+  log(`file: ${filePath}`);
+  const content = await readFile(filePath, "utf8");
+  const rel = filePath.replace("/home/profit/lakehouse/", "");
+
+  // Build a query embedding from the first ~800 chars of the file
+  // (good enough for topical retrieval).
+  const seed = content.slice(0, 800);
+  const [seedEmb] = await embedBatch([seed]);
+
+  const topPrd = retrieveTopK(seedEmb, prd_chunks, TOP_K_CONTEXT);
+  const topPlan = retrieveTopK(seedEmb, proposal_chunks, TOP_K_CONTEXT);
+  log(`  retrieved ${topPrd.length} PRD chunks + ${topPlan.length} proposal chunks`);
+
+  const contextBlock = [
+    "═══ RELEVANT PRD EXCERPTS ═══",
+    ...topPrd.map(c => `[PRD @${c.offset}]\n${c.text.slice(0, 600)}`),
+    "",
+    "═══ RELEVANT CHANGE PROPOSAL EXCERPTS ═══",
+    ...topPlan.map(c => `[PLAN @${c.offset}]\n${c.text.slice(0, 600)}`),
+  ].join("\n\n");
+
+  // Files bigger than FILE_TREE_SPLIT_THRESHOLD get tree-split.
+  // Summarize each shard to a scratchpad, then review against the
+  // scratchpad instead of the truncated first chunk. Prevents the
+  // false-positive pattern where the model claims a field is
+  // "missing" because it's past the context cutoff.
+  let sourceForPrompt: string;
+  let treeSplitFired = false;
+  let shardsSummarized = 0;
+  let extraCloudCalls = 0;
+  if (content.length > FILE_TREE_SPLIT_THRESHOLD) {
+    treeSplitFired = true;
+    const ts = await treeSplitFile(rel, content);
+    shardsSummarized = ts.shards;
+    extraCloudCalls = ts.cloud_calls;
+    sourceForPrompt = `[FULL-FILE SCRATCHPAD — distilled from ${ts.shards} shards via tree-split]\n${ts.scratchpad}`;
+  } else {
+    sourceForPrompt = content;
+  }
+
+  // Prompt — when tree-split fired, include an explicit instruction
+  // not to claim a field/function is "missing" because the scratchpad
+  // is a distillation not the full file. Attacks the rubric-tuning
+  // concern J called out.
+  const truncationWarning = treeSplitFired
+    ? `\nIMPORTANT: the "source" below is a multi-shard distillation (tree-split across ${shardsSummarized} shards), NOT the full raw file. DO NOT claim any field, function, or feature is "missing" based on its absence from this distillation — the distillation may have elided it. Only call out gaps that appear DIRECTLY contradicted by the PRD excerpts.\n`
+    : "";
+
+  const baseTask = `You are reviewing one source file against the Lakehouse PRD and an active cohesion-integration plan.
+
+FILE: ${rel} (${content.length} bytes${treeSplitFired ? `, tree-split into ${shardsSummarized} shards` : ""})
+${truncationWarning}
+─────── source ───────
+${sourceForPrompt}
+─────── end source ───────
+
+${contextBlock}
+
+Produce a structured review with:
+1. Alignment score (1-10) between this file and the PRD intent
+2. 3-5 concrete suggested changes (bullet points), each naming a specific function/line and what to change
+3. Any gap where this file's behavior contradicts the PRD or the proposal
+
+Respond with markdown. Be specific, not generic. Cite file-region + PRD-chunk-offset when relevant.`;
+
+  const history: FileReview["attempts_history"] = [];
+  let accepted: string | null = null;
+  let acceptedModel = "";
+  let acceptedOn = 0;
+
+  for (let i = 0; i < MAX_ATTEMPTS; i++) {
+    const n = i + 1;
+    const rung = LADDER[i];
+    const learning = history.length > 0
+      ? `\n\n═══ PRIOR ATTEMPTS FAILED. Specific issues to fix: ═══\n${history.map(h => `Attempt ${h.n} (${h.model}, ${h.chars} chars): ${h.status} — ${h.error ?? "thin/unstructured answer"}`).join("\n")}\n═══`
+      : "";
+
+    log(`  attempt ${n}/${MAX_ATTEMPTS}: ${rung.provider}::${rung.model}${learning ? " [w/ learning]" : ""}`);
+    const r = await chat({
+      provider: rung.provider,
+      model: rung.model,
+      prompt: baseTask + learning,
+      max_tokens: 1500,
+    });
+
+    if (r.error) {
+      history.push({ n, model: rung.model, status: "error", chars: 0, error: r.error.slice(0, 180) });
+      log(`    ✗ error: ${r.error.slice(0, 80)}`);
+      continue;
+    }
+    if (!isAcceptable(r.content)) {
+      history.push({ n, model: rung.model, status: "thin", chars: r.content.length, error: `thin/unstructured (${r.content.length} chars)` });
+      log(`    ✗ thin/unstructured (${r.content.length} chars)`);
+      continue;
+    }
+    history.push({ n, model: rung.model, status: "accepted", chars: r.content.length });
+    accepted = r.content;
+    acceptedModel = `${rung.provider}/${rung.model}`;
+    acceptedOn = n;
+    log(`    ✓ ACCEPTED on attempt ${n} (${rung.model}, ${r.content.length} chars)`);
+    break;
+  }
+
+  const review: FileReview = {
+    file: rel,
+    file_bytes: content.length,
+    tree_split_fired: treeSplitFired,
+    shards_summarized: shardsSummarized,
+    top_prd_chunks: topPrd.map(c => ({ origin: c.origin, offset: c.offset, score: (c as any)._score })),
+    top_proposal_chunks: topPlan.map(c => ({ origin: c.origin, offset: c.offset, score: (c as any)._score })),
+    attempts_made: history.length,
+    attempts_history: history,
+    accepted_on: acceptedOn || null,
+    escalated_to_model: acceptedModel,
+    suggestions: accepted ?? "[no acceptable answer after escalation ladder exhausted]",
+    duration_ms: Date.now() - t0,
+  };
+
+  // Append to the shared scrum-reviews jsonl so the auditor's
+  // kb_query check can surface relevant reviews for files in a
+  // PR diff. Cohesion plan Phase C wire.
+  if (accepted) {
+    const { appendFile, mkdir } = await import("node:fs/promises");
+    const { dirname } = await import("node:path");
+    await mkdir(dirname(SCRUM_REVIEWS_JSONL), { recursive: true });
+    const row = {
+      file: rel,
+      reviewed_at: new Date().toISOString(),
+      accepted_model: acceptedModel,
+      accepted_on_attempt: acceptedOn,
+      attempts_made: history.length,
+      tree_split_fired: treeSplitFired,
+      suggestions_preview: accepted.slice(0, 2000),
+      schema_version: 2,
+      scrum_master_reviewed: true,
+    };
+    try {
+      await appendFile(SCRUM_REVIEWS_JSONL, JSON.stringify(row) + "\n");
+    } catch (e) {
+      console.error(`[scrum] failed to append scrum_reviews.jsonl: ${(e as Error).message}`);
+    }
+
+    // Route the accepted review through llm_team's fact extractor so
+    // its entities + relationships land in audit_facts.jsonl alongside
+    // inference-side extractions. Same index, two sources. Tagged
+    // source:"scrum_review" + scrum_master_reviewed:true so downstream
+    // queries can filter by provenance. Reviews shorter than 120
+    // chars are skipped — they're usually one-liners ("LGTM") with
+    // no extractable knowledge.
+    if (accepted.length >= 120 && process.env.LH_SCRUM_SKIP_EXTRACT !== "1") {
+      try {
+        const { extractFacts } = await import("../../auditor/fact_extractor.ts");
+        const ex = await extractFacts(accepted);
+        if (!ex.error || ex.entities.length + ex.facts.length > 0) {
+          const factRow = {
+            pr_number: 0,                   // scrum runs outside a PR scope
+            file: rel,
+            head_sha: "",                    // no SHA scope; scope is the file+timestamp
+            extracted_at: ex.extracted_at,
+            extractor: ex.extractor_model,
+            verifier: ex.verifier_model,
+            llm_team_run_id: ex.llm_team_run_id ?? null,
+            facts: ex.facts,
+            entities: ex.entities,
+            relationships: ex.relationships,
+            verification_preview: ex.verification.slice(0, 400),
+            schema_version: 2,
+            source: "scrum_review",
+            scrum_master_reviewed: true,
+          };
+          const AUDIT_FACTS_JSONL = "/home/profit/lakehouse/data/_kb/audit_facts.jsonl";
+          await appendFile(AUDIT_FACTS_JSONL, JSON.stringify(factRow) + "\n");
+        }
+      } catch (e) {
+        console.error(`[scrum] fact extraction failed for ${rel}: ${(e as Error).message}`);
+      }
+    }
+  }
+
+  return review;
+}
+
+async function loadAndChunk(path: string, origin_tag: string): Promise<Chunk[]> {
+  const text = await readFile(path, "utf8");
+  const raw = chunkText(text);
+  const embs = await embedBatch(raw.map(r => r.text));
+  return raw.map((r, i) => ({
+    id: createHash("sha256").update(r.text).digest("hex").slice(0, 10),
+    text: r.text,
+    embedding: embs[i],
+    origin: origin_tag,
+    offset: r.offset,
+  }));
+}
+
+async function main() {
+  await mkdir(OUT_DIR, { recursive: true });
+  log(`output: ${OUT_DIR}`);
+  log(`targets: ${TARGET_FILES.length} files`);
+
+  log("loading + embedding PRD...");
+  const prd_chunks = await loadAndChunk(PRD_PATH, "PRD");
+  log(`  PRD: ${prd_chunks.length} chunks`);
+
+  log("loading + embedding cohesion plan...");
+  const plan_chunks = await loadAndChunk(PROPOSAL_PATH, "COHESION_PLAN");
+  log(`  plan: ${plan_chunks.length} chunks`);
+
+  log("");
+  log("─── scrum master: walking target files ───");
+
+  const reviews: FileReview[] = [];
+  for (const f of TARGET_FILES) {
+    const review = await reviewFile(f, prd_chunks, plan_chunks);
+    reviews.push(review);
+    await writeFile(
+      `${OUT_DIR}/review_${review.file.replace(/\//g, "_")}.json`,
+      JSON.stringify(review, null, 2),
+    );
+    log(`  → ${review.file}: ${review.accepted_on ? `accepted on ${review.accepted_on} by ${review.escalated_to_model}` : "UNRESOLVED"} (${review.duration_ms}ms)`);
+  }
+
+  // Consolidated scrum-master report
+  const report_md: string[] = [];
+  report_md.push(`# Scrum-master review\n`);
+  report_md.push(`Generated: ${new Date().toISOString()}`);
+  report_md.push(`Files reviewed: ${reviews.length}`);
+  report_md.push(`Total duration: ${(reviews.reduce((s, r) => s + r.duration_ms, 0) / 1000).toFixed(1)}s\n`);
+  for (const r of reviews) {
+    report_md.push(`\n## ${r.file}`);
+    report_md.push(`- **Accepted on attempt:** ${r.accepted_on ?? "NOT resolved after 6 attempts"}`);
+    report_md.push(`- **Escalated to:** \`${r.escalated_to_model || "—"}\``);
+    report_md.push(`- **Total attempts:** ${r.attempts_made}`);
+    if (r.attempts_history.length > 1) {
+      report_md.push(`- **Attempt history:**`);
+      for (const h of r.attempts_history) {
+        report_md.push(`  - ${h.n}: \`${h.model}\` → ${h.status}${h.error ? ` (${h.error.slice(0, 100)})` : ""}`);
+      }
+    }
+    report_md.push(`\n### Suggestions\n\n${r.suggestions}\n`);
+  }
+  await writeFile(`${OUT_DIR}/scrum_report.md`, report_md.join("\n"));
+
+  const summary = {
+    ran_at: new Date().toISOString(),
+    target_count: TARGET_FILES.length,
+    resolved: reviews.filter(r => r.accepted_on !== null).length,
+    total_attempts: reviews.reduce((s, r) => s + r.attempts_made, 0),
+    total_duration_ms: reviews.reduce((s, r) => s + r.duration_ms, 0),
+    per_file: reviews.map(r => ({ file: r.file, accepted_on: r.accepted_on, model: r.escalated_to_model, attempts: r.attempts_made, ms: r.duration_ms })),
+  };
+  await writeFile(`${OUT_DIR}/summary.json`, JSON.stringify(summary, null, 2));
+
+  log("");
+  log("═══ SCRUM REPORT ═══");
+  log(`  files: ${summary.target_count}, resolved: ${summary.resolved}, total attempts: ${summary.total_attempts}`);
+  log(`  total time: ${(summary.total_duration_ms / 1000).toFixed(1)}s`);
+  log("");
+  for (const p of summary.per_file) {
+    const mark = p.accepted_on ? "✓" : "✗";
+    log(`  ${mark} ${p.file.padEnd(60)} attempt ${p.accepted_on ?? "—"}/${p.attempts} ${p.model} ${p.ms}ms`);
+  }
+  log("");
+  log(`report: ${OUT_DIR}/scrum_report.md`);
+
+  process.exit(summary.resolved === summary.target_count ? 0 : 1);
+}
+
+main().catch(e => { console.error("[scrum] fatal:", e); process.exit(2); });