infra: replace gpt-oss with Ollama Pro + OpenCode Zen across hot paths

Ollama Pro plan went live today (39-model fleet on the same OLLAMA_CLOUD_KEY) and OpenCode Zen was already wired in the gateway but not consumed. Routing every gpt-oss call site to faster / stronger replacements: | Site | gpt-oss → replacement | Why | |---|---|---| | ollama_cloud default | gpt-oss:120b → deepseek-v3.2 | newest DeepSeek revision; live-probed `pong` | | openrouter default | openai/gpt-oss-120b:free → x-ai/grok-4.1-fast | already the scrum LADDER's PRIMARY | | modes.toml staffing_inference | openai/gpt-oss-120b:free → kimi-k2.6 | coding-specialized, on Ollama Pro | | modes.toml doc_drift_check | gpt-oss:120b → gemini-3-flash-preview | speed leader for factual checks | | scrum_master_pipeline tree-split MAP+REDUCE | gpt-oss:120b → gemini-3-flash-preview | latency-dominated path (5-20× per file) | | bot/propose.ts CLOUD_MODEL | gpt-oss:120b → deepseek-v3.2 | same Ollama key, faster | | mcp-server/observer.ts overseer label fallback | gpt-oss:120b → claude-opus-4-7 | matches new overseer model | | crates/gateway/src/execution_loop overseer escalation | ollama_cloud/gpt-oss:120b → opencode/claude-opus-4-7 | frontier reasoning matters here — fires only after local self-correct fails twice; Zen pay-per-token cost is bounded | Verification: - `cargo check -p gateway --tests` — clean - Live probes through localhost:3100/v1/chat: - `opencode/claude-opus-4-7` → "pong" - `gemini-3-flash-preview` (ollama_cloud) → "pong" - `kimi-k2.6` (ollama_cloud) → "pong" - `deepseek-v3.2` (ollama_cloud) → "Pong! 🏓" Notes: - kimi-k2:1t still upstream-broken (HTTP 500 on Ollama Pro probe today, matches yesterday's memory). Replacement table never picks it. - The Rust changes need a `systemctl restart lakehouse.service` to take effect on the running gateway. TS callers reload on next run. - aibridge/src/context.rs still has gpt-oss:{20b,120b} in its window- size lookup table; harmless and kept for callers that pass it explicitly as an override. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Merge PR #11 : distillation v1.0.0 + Phase 42-45 + auditor cross-lineage + staffing cutover
2026-04-28 06:13:48 -05:00 · 2026-04-27 15:55:22 +00:00 · 2026-04-27 10:46:03 -05:00 · 2026-04-27 10:45:38 -05:00 · 2026-04-27 10:45:29 -05:00 · 2026-04-27 08:32:33 -05:00
34 changed files with 1707 additions and 300 deletions
@@ -4093,6 +4093,7 @@ dependencies = [
 "tracing-opentelemetry",
 "tracing-subscriber",
 "truth",
+ "validator",
 "vectord",
 ]

@@ -8912,6 +8913,8 @@ dependencies = [
 name = "validator"
 version = "0.1.0"
 dependencies = [
+ "arrow 55.2.0",
+ "parquet 55.2.0",
 "serde",
 "serde_json",
 "thiserror 2.0.18",
@@ -33,37 +33,16 @@ const GATEWAY = process.env.LH_GATEWAY_URL ?? "http://localhost:3100";
 // vendor lineage so consensus + tie-break won't fail-correlate).
 const MODEL = process.env.LH_AUDITOR_REVIEW_MODEL ?? "deepseek-v3.1:671b";
 const TIEBREAKER_MODEL = process.env.LH_AUDITOR_TIEBREAKER_MODEL ?? "x-ai/grok-4.1-fast";
-// SHARD_MODEL retained for the legacy callCloud path (still used by
-// runCloudInference's diagnostic mode), but no longer fired by the
-// main inference flow — tree-split was retired 2026-04-27 in favor of
-// the mode runner's matrix retrieval against lakehouse_answers_v1.
-const SHARD_MODEL = process.env.LH_AUDITOR_SHARD_MODEL ?? "qwen3-coder:480b";
 const N_CONSENSUS = Number(process.env.LH_AUDITOR_CONSENSUS_N ?? 3);
-// Bounded parallelism on the tree-split shard loop. Old behavior was
-// fully serial ("keep gateway load bounded") which made huge PRs take
-// 5+ minutes of curation alone. 6 in flight keeps gateway busy without
-// thrashing it; tunable via env.
-const SHARD_CONCURRENCY = Number(process.env.LH_AUDITOR_SHARD_CONCURRENCY ?? 6);
 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).
+// 40KB comfortably fits the consensus models' context windows
+// (deepseek-v3.1 64K, gpt-oss-120b 128K). When the raw PR diff
+// exceeds this, we truncate and signal it via curationNote — the
+// pr_audit mode runner's matrix retrieval (lakehouse_answers_v1 +
+// arch + symbols) supplies the cross-PR context that tree-split
+// used to synthesize from scratch. Tree-split itself was retired
+// 2026-04-27 (see commit deleting treeSplitDiff/callCloud/SHARD_*).
 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;
 // Mode runner can take longer than a raw /v1/chat call because it does
 // pathway-fingerprint lookup + matrix retrieval + relevance filter
@@ -169,12 +148,16 @@ export async function runInferenceCheck(
  interface Votes { trues: number; falses: number; evidences: string[] }
  const votesByClaim = new Map<number, Votes>();
  const unflaggedByRun: any[][] = [];
-  let totalLatencyMs = 0;
+  // The N=3 consensus calls run via Promise.all — wall-clock is
+  // bounded by the SLOWEST call, not the sum. Pre-2026-04-27 we
+  // summed and reported "Xms total" which double/triple-counted
+  // (Opus self-audit caught it). Use max for accurate wall-clock.
+  let maxLatencyMs = 0;
  let totalEnrichedChars = 0;
  let bugFingerprintsSeen = 0;
  let matrixKeptSeen = 0;
  for (const run of parsedRuns) {
-    totalLatencyMs += run.latency_ms ?? 0;
+    maxLatencyMs = Math.max(maxLatencyMs, run.latency_ms ?? 0);
    totalEnrichedChars += run.enriched_chars ?? 0;
    bugFingerprintsSeen = Math.max(bugFingerprintsSeen, run.bug_fingerprints ?? 0);
    matrixKeptSeen = Math.max(matrixKeptSeen, run.matrix_kept ?? 0);
@@ -199,7 +182,7 @@ export async function runInferenceCheck(
  findings.push({
    check: "inference",
    severity: "info",
-    summary: `pr_audit mode runner completed (model=${MODEL}, consensus=${parsedRuns.length}/${N_CONSENSUS}, ${totalLatencyMs}ms total)${curationNote}`,
+    summary: `pr_audit mode runner completed (model=${MODEL}, consensus=${parsedRuns.length}/${N_CONSENSUS}, ${maxLatencyMs}ms wall-clock)${curationNote}`,
    evidence: [
      `claims voted: ${votesByClaim.size}`,
      `parsed runs: ${parsedRuns.length} / ${N_CONSENSUS}`,
@@ -443,60 +426,28 @@ async function runModeRunnerInference(
      error: "unparseable", diagnostic: (e as Error).message, model,
    };
  }
-  const content: string = body?.response ?? "";
+  const content: string = typeof body?.response === "string" ? body.response : "";
  const parsed = extractJson(content);
+  // Number-coerced extractors so a non-numeric upstream value (string,
+  // null, NaN) collapses to 0 instead of poisoning downstream
+  // arithmetic. Caught 2026-04-27 by kimi_architect self-audit —
+  // optional-chaining + ?? only catches null/undefined, not type drift.
+  const num = (v: unknown): number => {
+    const n = typeof v === "number" ? v : Number(v);
+    return Number.isFinite(n) ? n : 0;
+  };
  return {
    parsed,
-    latency_ms: body?.latency_ms ?? 0,
-    enriched_chars: body?.enriched_prompt_chars ?? 0,
-    bug_fingerprints: body?.sources?.bug_fingerprints_count ?? 0,
-    matrix_kept: body?.sources?.matrix_chunks_kept ?? 0,
+    latency_ms: num(body?.latency_ms),
+    enriched_chars: num(body?.enriched_prompt_chars),
+    bug_fingerprints: num(body?.sources?.bug_fingerprints_count),
+    matrix_kept: num(body?.sources?.matrix_chunks_kept),
    error: parsed ? undefined : "unparseable",
    diagnostic: parsed ? undefined : content.slice(0, 200),
    model,
  };
 }

-// Legacy direct /v1/chat caller — kept for callers outside the
-// pr_audit pipeline. Currently unused after the 2026-04-26 mode-runner
-// rebuild; preserved so we can A/B against the mode runner if a
-// regression surfaces.
-async function runCloudInference(systemMsg: string, userMsg: string, model: string): Promise<{ parsed: any | null; tokens: number; error?: string; diagnostic?: string; model: string }> {
-  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 },
-        ],
-        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 });
@@ -540,111 +491,7 @@ async function extractAndPersistFacts(scratchpad: string, ctx: InferenceContext)
  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 shard summaries + the final verification.
-// Pre-2026-04-26 the shard loop ran serially "to keep gateway load
-// bounded" — turned out to be a bottleneck on PRs with 50+ shards
-// (5+ minutes of curation). Now bounded-parallel via
-// SHARD_CONCURRENCY: in-flight ≤ N at any time, gateway stays calm,
-// wall-clock drops 4-6×.
-//
-// Determinism: each shard summary call uses temp=0 + think=false
-// (same as before), so identical input yields identical scratchpad.
-// Order is preserved by indexed-write into a fixed-length array
-// before string-join, so concurrency doesn't shuffle the scratchpad.
-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");

-  const buildPrompt = (si: number, shard: { from: number; to: number; text: string }): string => [
-    `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");
-
-  // Pre-allocate so we can write back at the original index from
-  // out-of-order completion.
-  const summaries: string[] = new Array(shards.length).fill("");
-  let nextIdx = 0;
-  async function worker() {
-    while (true) {
-      const myIdx = nextIdx++;
-      if (myIdx >= shards.length) return;
-      const r = await callCloud(buildPrompt(myIdx, shards[myIdx]), 400);
-      summaries[myIdx] = r.content;
-    }
-  }
-  const concurrency = Math.max(1, Math.min(SHARD_CONCURRENCY, shards.length));
-  await Promise.all(Array.from({ length: concurrency }, worker));
-
-  let scratchpad = "";
-  for (const [si, shard] of shards.entries()) {
-    const summary = summaries[si];
-    if (summary) {
-      scratchpad += `\n--- shard ${si + 1} (chars ${shard.from}..${shard.to}) ---\n${summary.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: SHARD_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:
@@ -26,15 +26,23 @@
 //
 // Off by default: caller checks LH_AUDITOR_KIMI=1 before invoking.

-import { readFile, writeFile, mkdir, appendFile, stat } from "node:fs/promises";
-import { existsSync } from "node:fs";
-import { join, resolve } from "node:path";
+import { readFile, writeFile, mkdir, appendFile, stat, realpath } from "node:fs/promises";
+import { existsSync, realpathSync } from "node:fs";
+import { dirname, join, resolve } from "node:path";
 import type { Finding, CheckKind } from "../types.ts";

 const GATEWAY = process.env.LH_GATEWAY_URL ?? "http://localhost:3100";
 const KIMI_VERDICTS_DIR = "/home/profit/lakehouse/data/_auditor/kimi_verdicts";
 const KIMI_AUDITS_JSONL = "/home/profit/lakehouse/data/_kb/kimi_audits.jsonl";
 const REPO_ROOT = "/home/profit/lakehouse";
+// Canonicalize at module load — REPO_ROOT itself may be a symlink in
+// some environments (e.g. /home/profit is a bind-mount). Computing
+// once at startup means the per-finding grounding loop can compare
+// realpath(target) against this stable anchor.
+const REPO_ROOT_REAL = (() => {
+  try { return realpathSync(REPO_ROOT); }
+  catch { return REPO_ROOT; }
+})();
 // 15 min budget. Bun's fetch has an intrinsic ~300s limit that our
 // AbortController + setTimeout combo could not override; we use curl
 // via Bun.spawn instead (callKimi below). Curl honors -m for max
@@ -50,37 +58,63 @@ const MAX_PRIOR_FINDINGS = 50;
 // gateway as a fallback for when Ollama Cloud is upstream-broken.
 const KIMI_PROVIDER = process.env.LH_AUDITOR_KIMI_PROVIDER ?? "ollama_cloud";
 const KIMI_MODEL = process.env.LH_AUDITOR_KIMI_MODEL ?? "kimi-k2.6";
-// Big-diff promotion: when the diff exceeds OPUS_THRESHOLD_CHARS, swap
-// to OPUS_MODEL for that audit. 2026-04-27 3-way bake-off (Kimi vs
-// Haiku vs Opus on a 32K diff) showed Opus is the only model that
-// catches cross-file ramifications + escalates `block` severity on
-// real architectural risks. ~5x the spend per audit, only worth it
-// when the diff is big enough to have those risks.
+// Cross-lineage alternation. 2026-04-27 J's call: Opus is too
+// expensive to auto-fire (~$0.30/audit). Kimi K2.6 via Go-sub is
+// effectively free; Haiku 4.5 via Zen is ~$0.04. Alternate between
+// them so we get cross-lineage signal (Moonshot vs Anthropic) on
+// every PR's audit history without burning the budget.
 //
-// Defaults: Haiku for normal diffs (fast, cheap, ~$0.02), Opus for
-// > 100k chars. Disable promotion: set OPUS_THRESHOLD_CHARS very high.
-const OPUS_MODEL = process.env.LH_AUDITOR_KIMI_OPUS_MODEL ?? "claude-opus-4-7";
-const OPUS_PROVIDER = process.env.LH_AUDITOR_KIMI_OPUS_PROVIDER ?? "opencode";
-const OPUS_THRESHOLD_CHARS = Number(process.env.LH_AUDITOR_KIMI_OPUS_THRESHOLD_CHARS) || 100_000;
+// Default: Kimi K2.6 on even audits, Haiku 4.5 on odd. Each PR's
+// audits flip between vendors as new SHAs come in.
+//
+// Frontier models (Opus 4.7, GPT-5.5, Gemini 3.1) are NOT in the
+// auto path. Operator hands distilled findings to a frontier model
+// manually when high-leverage decisions need it. Removing Opus from
+// auto-promotion saves ~$1-3/day on the daemon at our cadence.
+//
+// Override the alternation entirely with LH_AUDITOR_KIMI_MODEL
+// (forces one model regardless of audit count); set
+// LH_AUDITOR_KIMI_ALT_MODEL to the alternate.
+const ALT_MODEL = process.env.LH_AUDITOR_KIMI_ALT_MODEL ?? "claude-haiku-4-5";
+const ALT_PROVIDER = process.env.LH_AUDITOR_KIMI_ALT_PROVIDER ?? "opencode";
+const FORCE_DEFAULT = process.env.LH_AUDITOR_KIMI_MODEL !== undefined && process.env.LH_AUDITOR_KIMI_MODEL !== "";

-function selectModel(diffLen: number): { provider: string; model: string; promoted: boolean } {
-  if (diffLen > OPUS_THRESHOLD_CHARS) {
-    return { provider: OPUS_PROVIDER, model: OPUS_MODEL, promoted: true };
+function selectModel(diffLen: number, auditIndex: number = 0): { provider: string; model: string; promoted: boolean } {
+  // Operator override — env-pinned model wins.
+  if (FORCE_DEFAULT) {
+    return { provider: KIMI_PROVIDER, model: KIMI_MODEL, promoted: false };
+  }
+  // Alternate Kimi (default, even index) ↔ Haiku (alt, odd index).
+  // diffLen kept in the signature for future "big diff → Haiku
+  // anyway" logic; not used yet so we don't auto-burn on big PRs.
+  void diffLen;
+  if (auditIndex % 2 === 1) {
+    return { provider: ALT_PROVIDER, model: ALT_MODEL, promoted: true };
  }
  return { provider: KIMI_PROVIDER, model: KIMI_MODEL, promoted: false };
 }
-// 128K — Kimi K2.6 emits reasoning_content that counts against this
-// budget but isn't surfaced in the OpenAI-shape `content` field.
-// Capping low silently produces empty content with finish_reason=length
-// when reasoning consumes the budget. 128K leaves ample room for both
-// reasoning and visible findings on any audit prompt we throw at it.
-// Override via LH_AUDITOR_KIMI_MAX_TOKENS only if you want to cap cost.
+// Model-aware max_tokens. Different upstream APIs cap at different
+// limits and reject requests that exceed them:
+//   - Anthropic Opus 4.x: 32K output (with extended-output header)
+//   - Anthropic Haiku 4.5: 8K output
+//   - Kimi K2.6 (reasoning): 128K — needs headroom because
+//     reasoning_content counts against the budget
+//   - Default: 16K, conservative middle ground
 //
-// Bug fix 2026-04-27 (caught by Kimi's own self-audit): empty env var
-// like LH_AUDITOR_KIMI_MAX_TOKENS="" used to parse via Number("") → 0
-// because `??` only catches null/undefined. Use `||` so empty string,
-// 0, or NaN all fall back to the default.
-const MAX_TOKENS = Number(process.env.LH_AUDITOR_KIMI_MAX_TOKENS) || 128_000;
+// 2026-04-27 BLOCK from Opus self-audit: the prior single-default of
+// 128K worked silently (Anthropic clamps server-side) but was
+// technically invalid. Per-model caps make it explicit. Override via
+// LH_AUDITOR_KIMI_MAX_TOKENS to force a value (also fixes the empty-
+// env Number("") -> 0 trap by using `||` not `??`).
+const MAX_TOKENS_OVERRIDE = Number(process.env.LH_AUDITOR_KIMI_MAX_TOKENS) || 0;
+function maxTokensFor(model: string): number {
+  if (MAX_TOKENS_OVERRIDE > 0) return MAX_TOKENS_OVERRIDE;
+  if (model.startsWith("claude-opus")) return 32_000;
+  if (model.startsWith("claude-haiku") || model.startsWith("claude-sonnet")) return 8_192;
+  if (model.startsWith("kimi-")) return 128_000;
+  if (model.startsWith("gpt-5") || model.startsWith("o1") || model.startsWith("o3") || model.startsWith("o4")) return 32_000;
+  return 16_000;
+}

 export interface KimiArchitectContext {
  pr_number: number;
@@ -106,22 +140,56 @@ export async function runKimiArchitectCheck(
  ctx: KimiArchitectContext,
 ): Promise<Finding[]> {
  const cachePath = join(KIMI_VERDICTS_DIR, `${ctx.pr_number}-${ctx.head_sha.slice(0, 12)}.json`);
+  const outageSentinel = `${cachePath}.outage`;
+  const OUTAGE_TTL_MS = 10 * 60 * 1000;
+
+  // Outage negative-cache — if upstream failed within OUTAGE_TTL_MS,
+  // skip this audit and return immediately. Prevents the daemon from
+  // hammering a downed Kimi/Anthropic upstream every 90s.
+  if (existsSync(outageSentinel)) {
+    try {
+      const s = await stat(outageSentinel);
+      if (Date.now() - s.mtimeMs < OUTAGE_TTL_MS) {
+        const note = JSON.parse(await readFile(outageSentinel, "utf8"));
+        return [skipFinding(`upstream still down (cached ${Math.round((Date.now() - s.mtimeMs) / 1000)}s ago): ${String(note.reason).slice(0, 160)}`)];
+      }
+    } catch { /* malformed sentinel — fall through to fresh call */ }
+  }

  // Cost cap — return cached findings if a verdict for this exact head
  // SHA was generated within the TTL.
  const cached = await loadCachedVerdict(cachePath);
  if (cached) {
-    const fs2: Finding[] = cached.findings.length > 0
+    return cached.findings.length > 0
      ? cached.findings
      : [{ check: "kimi_architect" as CheckKind, severity: "info", summary: "kimi_architect cached — 0 findings", evidence: [`cache: ${cachePath}`] }];
-    return fs2;
  }

-  const selected = selectModel(diff.length);
+  // Alternate model based on how many audits this PR has had — gives
+  // cross-lineage signal (Kimi/Moonshot ↔ Haiku/Anthropic) on every
+  // PR's audit history. Count is derived from existing kimi_verdicts
+  // files for this PR; cheap O(N_PRs) directory read.
+  let auditIndex = 0;
+  try {
+    const dir = "/home/profit/lakehouse/data/_auditor/kimi_verdicts";
+    if (existsSync(dir)) {
+      const all = require("node:fs").readdirSync(dir) as string[];
+      auditIndex = all.filter((f) => f.startsWith(`${ctx.pr_number}-`)).length;
+    }
+  } catch { /* default 0 — Kimi */ }
+
+  const selected = selectModel(diff.length, auditIndex);
  let response: { content: string; usage: any; finish_reason: string; latency_ms: number };
  try {
    response = await callKimi(buildPrompt(diff, priorFindings, ctx), selected.provider, selected.model);
  } catch (e) {
+    // Negative-cache for 10 min on outage (caught 2026-04-27 by Opus
+    // self-audit): without this, every audit cycle within the 24h
+    // TTL re-calls upstream while it's still down. Use a sentinel
+    // file with mtime check rather than persisting a verdict so the
+    // happy-path cache reader doesn't have to special-case it.
+    const sentinel = `${cachePath}.outage`;
+    try { await writeFile(sentinel, JSON.stringify({ at: new Date().toISOString(), reason: (e as Error).message.slice(0, 200) })); } catch {}
    return [skipFinding(`kimi call failed (${selected.model}): ${(e as Error).message.slice(0, 200)}`)];
  }

@@ -145,17 +213,25 @@ export async function runKimiArchitectCheck(
    grounding,
  };

-  await persistVerdict(cachePath, verdict);
+  // Cache-poisoning guard (caught 2026-04-27 by Opus self-audit):
+  // when parseFindings returns 0 findings (Kimi rambled, prompt too
+  // big, or the markdown shape changed and our regex missed every
+  // block), persisting the empty verdict short-circuits all future
+  // audits in the 24h TTL window with a useless cached "0 findings"
+  // result. Better to leave no cache and re-call upstream next time.
+  // Always append metrics — observability shouldn't depend on whether
+  // findings parsed.
  await appendMetrics(verdict);
-
-  return findings.length > 0
-    ? findings
-    : [{
-        check: "kimi_architect" as CheckKind,
-        severity: "info",
-        summary: `kimi_architect produced 0 ranked findings (${response.finish_reason}, ${verdict.usage.completion_tokens} tokens)`,
-        evidence: [`raw response: ${cachePath}`],
-      }];
+  if (findings.length > 0) {
+    await persistVerdict(cachePath, verdict);
+    return findings;
+  }
+  return [{
+    check: "kimi_architect" as CheckKind,
+    severity: "info",
+    summary: `kimi_architect produced 0 ranked findings (${response.finish_reason}, ${verdict.usage.completion_tokens} tokens) — not cached`,
+    evidence: [`raw saved (no cache): see kimi_audits.jsonl ${verdict.cached_at}`],
+  }];
 }

 async function loadCachedVerdict(path: string): Promise<KimiVerdictFile | null> {
@@ -211,7 +287,7 @@ async function callKimi(prompt: string, provider: string, model: string): Promis
    provider,
    model,
    messages: [{ role: "user", content: prompt }],
-    max_tokens: MAX_TOKENS,
+    max_tokens: maxTokensFor(model),
    temperature: 0.2,
  });
  // curl via Bun.spawn — bypasses Bun fetch's ~300s intrinsic ceiling.
@@ -296,13 +372,41 @@ async function computeGrounding(findings: Finding[]): Promise<{ total: number; v
    const [, relpath, lineStr] = m;
    const line = Number(lineStr);
    if (!line || !relpath) return false;
-    const abs = relpath.startsWith("/") ? relpath : resolve(REPO_ROOT, relpath);
+
+    // Path-traversal guard, two-layer (caught 2026-04-27 by Kimi
+    // self-audits on dd77632 then 2d9cb12).
+    //
+    // Layer 1 (lexical): resolve() normalizes `..` segments. Refuse
+    // any path that doesn't anchor under REPO_ROOT.
+    //
+    // Layer 2 (symlink): even if the lexical path is anchored, it
+    // could be a symlink whose target escapes. realpath() resolves
+    // symlinks; compare the real path against REPO_ROOT_REAL.
+    //
+    // Both layers exist because attackers might bypass either alone:
+    // raw `../etc/passwd` triggers layer 1; a planted symlink at
+    // ./safe-looking-name → /etc/passwd triggers layer 2.
+    const abs = resolve(REPO_ROOT, relpath);
+    if (!abs.startsWith(REPO_ROOT + "/") && abs !== REPO_ROOT) {
+      f.evidence.push(`[grounding: path escapes repo root, refusing]`);
+      return false;
+    }
+
    if (!existsSync(abs)) {
      f.evidence.push("[grounding: file not found]");
      return false;
    }
    try {
-      const lines = (await readFile(abs, "utf8")).split("\n");
+      // Symlink-resolution check before any read. realpath() throws
+      // if the file doesn't exist; existsSync above shields the
+      // common case but a TOCTOU race could still error here — the
+      // outer catch handles it.
+      const realPath = await realpath(abs);
+      if (!realPath.startsWith(REPO_ROOT_REAL + "/") && realPath !== REPO_ROOT_REAL) {
+        f.evidence.push(`[grounding: symlink target escapes repo root, refusing]`);
+        return false;
+      }
+      const lines = (await readFile(realPath, "utf8")).split("\n");
      if (line < 1 || line > lines.length) {
        f.evidence.push(`[grounding: line ${line} > EOF (${lines.length})]`);
        return false;
@@ -325,7 +429,11 @@ async function persistVerdict(path: string, v: KimiVerdictFile): Promise<void> {
 }

 async function appendMetrics(v: KimiVerdictFile): Promise<void> {
-  await mkdir(join(KIMI_AUDITS_JSONL, ".."), { recursive: true });
+  // dirname() instead of join(path, "..") — caught 2026-04-27 by both
+  // Haiku and Opus self-audits. The "/.." idiom resolves correctly
+  // via Node path normalization but is non-idiomatic + breaks if the
+  // path ever has trailing dots.
+  await mkdir(dirname(KIMI_AUDITS_JSONL), { recursive: true });
  await appendFile(KIMI_AUDITS_JSONL, JSON.stringify({
    pr_number: v.pr_number,
    head_sha: v.head_sha,
@@ -77,6 +77,17 @@ export function runStaticCheck(diff: string): Finding[] {
      // Strip the diff prefix (' ' for context, '+' for added).
      const body = (isAdded || line.startsWith(" ")) ? line.slice(1) : line;

+      // Compute the file-level backtick state ENTERING this line.
+      // The state machine sees pattern matches against the right
+      // context: a line that opens a backtick block has its own
+      // pattern checks evaluated under "inside-backtick" semantics
+      // for the portion AFTER the opening tick. Pre-2026-04-27 the
+      // state was updated AFTER the pattern checks, so the FIRST
+      // pattern on a backtick-opening line slipped through with
+      // stale "outside-backtick" semantics. Caught by Kimi self-audit.
+      const stateAtLineStart = inMultilineBacktick;
+      const stateAtLineEnd = updateBacktickState(body, stateAtLineStart);
+
      if (isAdded) {
        const added = body;

@@ -84,11 +95,13 @@ export function runStaticCheck(diff: string): Finding[] {
          for (const { re, why } of BLOCK_PATTERNS) {
            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 (inMultilineBacktick || isInsideQuotedString(added, m.index)) continue;
+              // Skip if EITHER (a) the file was already inside a
+              // multi-line backtick block when this line started, OR
+              // (b) the match sits inside a quoted string literal on
+              // THIS line. The earlier code only checked stateAtLineStart;
+              // now we also check that the match isn't past the
+              // opening backtick of a block that opens on this line.
+              if (stateAtLineStart || isInsideQuotedString(added, m.index)) continue;
              findings.push({
                check: "static",
                severity: "block",
@@ -120,13 +133,8 @@ export function runStaticCheck(diff: string): Finding[] {
        }
      }

-      // Update file-level multi-line backtick state by walking THIS
-      // line's unescaped backticks. Both context and added lines
-      // contribute (they're both in the post-merge file). Doc-comment
-      // backticks like `\\\`Foo\\\`` count too — that's the source of
-      // the original bug, where multi-line template literals contained
-      // `todo!()` references.
-      inMultilineBacktick = updateBacktickState(body, inMultilineBacktick);
+      // Carry the end-of-line state forward to the next iteration.
+      inMultilineBacktick = stateAtLineEnd;
    }

    // "Field added but never read" heuristic — catches exactly the
@@ -213,6 +221,13 @@ function extractNewFieldsWithLine(lines: string[]): Array<{ name: string; lineId
 // Stops the struct-search early if we hit a `}` at zero indent
 // (the previous scope) or another `pub struct` (we left ours).
 function parentStructHasSerdeDerive(lines: string[], fieldLineIdx: number): boolean {
+  // Bounds-check fieldLineIdx (caught 2026-04-27 by Kimi self-audit).
+  // Pre-fix: if fieldLineIdx >= lines.length, the loop ran from a
+  // negative implicit upper bound (fieldLineIdx - 80 could be > 0
+  // even when fieldLineIdx is past EOF) and read undefined slots.
+  // Defensive: bail early on out-of-range input.
+  if (fieldLineIdx < 0 || fieldLineIdx >= lines.length) return false;
+
  let structLineIdx = -1;
  for (let i = fieldLineIdx - 1; i >= 0 && i >= fieldLineIdx - 80; i--) {
    const raw = lines[i];
@@ -24,14 +24,30 @@ const POLL_INTERVAL_MS = 90_000; // 90s — enough budget for audit runs to comp
 const PAUSE_FILE = "/home/profit/lakehouse/auditor.paused";
 const STATE_FILE = "/home/profit/lakehouse/data/_auditor/state.json";

+// Per-PR audit cap. Prevents the daemon from running away on a PR
+// when each push surfaces new findings — operator wants to review
+// in batch, not have the daemon burn budget while they're away.
+// Default 3 audits per PR. Override via LH_AUDITOR_MAX_AUDITS_PER_PR.
+// Set to 0 to disable the cap.
+//
+// Reset (after manual review): edit data/_auditor/state.json and
+// set audit_count_per_pr.<N> = 0 (or delete the key). Daemon picks
+// up the change on the next cycle without restart.
+const MAX_AUDITS_PER_PR = Number(process.env.LH_AUDITOR_MAX_AUDITS_PER_PR) || 3;
+
 interface State {
  // Map: PR number → last-audited head SHA. Lets us dedupe audits
  // across restarts (poller can crash/restart without re-auditing
  // all open PRs from scratch).
  last_audited: Record<string, string>;
+  // Map: PR number → number of audits run on that PR since last reset.
+  // Daemon halts auditing a PR once this hits MAX_AUDITS_PER_PR.
+  // Operator clears the entry to resume.
+  audit_count_per_pr: Record<string, number>;
  started_at: string;
  cycles_total: number;
  cycles_skipped_paused: number;
+  cycles_skipped_capped: number;
  audits_run: number;
  last_cycle_at?: string;
 }
@@ -47,17 +63,21 @@ async function loadState(): Promise<State> {
    return {
      last_audited: s.last_audited ?? {},
      started_at: s.started_at ?? new Date().toISOString(),
+      audit_count_per_pr: s.audit_count_per_pr ?? {},
      cycles_total: s.cycles_total ?? 0,
      cycles_skipped_paused: s.cycles_skipped_paused ?? 0,
+      cycles_skipped_capped: s.cycles_skipped_capped ?? 0,
      audits_run: s.audits_run ?? 0,
      last_cycle_at: s.last_cycle_at,
    };
  } catch {
    return {
      last_audited: {},
+      audit_count_per_pr: {},
      started_at: new Date().toISOString(),
      cycles_total: 0,
      cycles_skipped_paused: 0,
+      cycles_skipped_capped: 0,
      audits_run: 0,
    };
  }
@@ -89,12 +109,38 @@ async function runCycle(state: State): Promise<State> {
  console.log(`[auditor] cycle ${state.cycles_total}: ${prs.length} open PR(s)`);

  for (const pr of prs) {
-    const last = state.last_audited[String(pr.number)];
+    const prKey = String(pr.number);
+    const last = state.last_audited[prKey];
    if (last === pr.head_sha) {
      console.log(`[auditor]   skip PR #${pr.number} (SHA ${pr.head_sha.slice(0, 8)} already audited)`);
      continue;
    }
-    console.log(`[auditor]   audit PR #${pr.number} (${pr.head_sha.slice(0, 8)}) — ${pr.title.slice(0, 60)}`);
+    // Per-head-SHA audit cap. Each new push gets MAX_AUDITS_PER_PR
+    // fresh attempts; the counter auto-resets when the head SHA
+    // changes. Operator only intervenes manually if a single SHA
+    // somehow needs MORE than the cap (rare — usually transient
+    // upstream errors clear themselves inside 3 attempts).
+    //
+    // Reset rule: if `last` exists (we've seen this PR before) AND
+    // pr.head_sha != last, that's a new push. Drop the counter.
+    // The dedup branch above already handles same-SHA → skip, so
+    // we only land here when the SHA actually moved.
+    if (last !== undefined && (state.audit_count_per_pr[prKey] ?? 0) > 0) {
+      const prior_count = state.audit_count_per_pr[prKey];
+      console.log(`[auditor]   PR #${pr.number} new head ${pr.head_sha.slice(0, 8)} (prior ${last.slice(0, 8)}, was ${prior_count}/${MAX_AUDITS_PER_PR}) — resetting cap counter`);
+      state.audit_count_per_pr[prKey] = 0;
+    }
+    const auditedSoFar = state.audit_count_per_pr[prKey] ?? 0;
+    if (MAX_AUDITS_PER_PR > 0 && auditedSoFar >= MAX_AUDITS_PER_PR) {
+      // This branch only fires now if the SAME head SHA somehow
+      // burned MAX audits (transient upstream errors retried that
+      // many times). Operator can clear state.audit_count_per_pr.<N>
+      // = 0 to force one more attempt; otherwise wait for next push.
+      console.log(`[auditor]   skip PR #${pr.number} (same head ${pr.head_sha.slice(0, 8)} burned ${auditedSoFar}/${MAX_AUDITS_PER_PR} — push new code or clear state.json audit_count_per_pr.${prKey})`);
+      state.cycles_skipped_capped += 1;
+      continue;
+    }
+    console.log(`[auditor]   audit PR #${pr.number} (${pr.head_sha.slice(0, 8)}) — ${pr.title.slice(0, 60)} [${auditedSoFar + 1}/${MAX_AUDITS_PER_PR}]`);
    try {
      // Skip dynamic by default: it mutates live playbook state and
      // re-runs on every PR update would pollute quickly. Operator
@@ -106,8 +152,22 @@ async function runCycle(state: State): Promise<State> {
        skip_inference: process.env.LH_AUDITOR_SKIP_INFERENCE === "1",
      });
      console.log(`[auditor]     verdict=${verdict.overall} findings=${verdict.metrics.findings_total} (block=${verdict.metrics.findings_block} warn=${verdict.metrics.findings_warn})`);
-      state.last_audited[String(pr.number)] = pr.head_sha;
+      state.last_audited[prKey] = pr.head_sha;
+      state.audit_count_per_pr[prKey] = auditedSoFar + 1;
      state.audits_run += 1;
+      if (state.audit_count_per_pr[prKey] >= MAX_AUDITS_PER_PR) {
+        console.log(`[auditor]     PR #${pr.number} reached cap (${MAX_AUDITS_PER_PR} audits) — daemon will skip further audits until reset`);
+      }
+      // Persist state immediately after each successful audit so the
+      // increment survives a crash. Pre-2026-04-27 the cycle saved
+      // once at the end (main.ts:140), which lost the count if the
+      // daemon was killed mid-cycle. Fix lifted from kimi_architect's
+      // own audit on this very file. saveState is idempotent + cheap
+      // (one JSON write), so per-audit cost is negligible.
+      try { await saveState(state); }
+      catch (e) {
+        console.error(`[auditor]     saveState mid-cycle failed: ${(e as Error).message} — count held in memory`);
+      }
    } catch (e) {
      console.error(`[auditor]     audit failed: ${(e as Error).message}`);
    }
@@ -13,10 +13,17 @@ import { readFile } from "node:fs/promises";
 import { createHash } from "node:crypto";
 import type { Gap, Proposal } from "./types.ts";

-const SIDECAR_URL = process.env.LH_SIDECAR_URL ?? "http://localhost:3200";
+// Phase 44 migration (2026-04-27): bot/propose.ts now flows through
+// the gateway's /v1/chat instead of hitting the sidecar's /generate
+// directly. /v1/usage tracks the call, Langfuse traces it, observer
+// sees it. Gateway owns the routing.
+//
+// 2026-04-28: gpt-oss:120b → deepseek-v3.2 via Ollama Pro. Newer
+// DeepSeek revision, faster, still on the same OLLAMA_CLOUD_KEY.
+const GATEWAY_URL = process.env.LH_GATEWAY_URL ?? "http://localhost:3100";
 const REPO_ROOT = "/home/profit/lakehouse";
 const PRD_PATH = `${REPO_ROOT}/docs/PRD.md`;
-const CLOUD_MODEL = process.env.LH_BOT_MODEL ?? "gpt-oss:120b";
+const CLOUD_MODEL = process.env.LH_BOT_MODEL ?? "deepseek-v3.2";
 const MAX_TOKENS = 6000;

 export async function findGaps(): Promise<Gap[]> {
@@ -72,13 +79,16 @@ export async function generateProposal(gap: Gap, historySummary: string = ""): P
  sections.push("Propose a small change that addresses this gap. Respond with the JSON object only.");
  const userPrompt = sections.join("\n");

-  const r = await fetch(`${SIDECAR_URL}/generate`, {
+  const r = await fetch(`${GATEWAY_URL}/v1/chat`, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      model: CLOUD_MODEL,
-      system: SYSTEM_PROMPT,
-      prompt: userPrompt,
+      provider: "ollama_cloud",
+      messages: [
+        { role: "system", content: SYSTEM_PROMPT },
+        { role: "user", content: userPrompt },
+      ],
      temperature: 0.2,
      max_tokens: MAX_TOKENS,
      think: false,
@@ -86,10 +96,10 @@ export async function generateProposal(gap: Gap, historySummary: string = ""): P
    signal: AbortSignal.timeout(180000), // cloud T3 can be slow — 3 min
  });
  if (!r.ok) {
-    throw new Error(`sidecar ${r.status}: ${await r.text()}`);
+    throw new Error(`gateway /v1/chat ${r.status}: ${await r.text()}`);
  }
  const j = await r.json() as any;
-  const raw: string = j.text ?? j.response ?? "";
+  const raw: string = j?.choices?.[0]?.message?.content ?? "";
  const usage = j.usage ?? {};
  const tokens = (usage.prompt_tokens ?? 0) + (usage.completion_tokens ?? 0);

@@ -44,7 +44,10 @@ name = "staffing_inference"
 # pattern generalizes beyond code review.
 preferred_mode = "staffing_inference_lakehouse"
 fallback_modes = ["ladder", "consensus", "pipeline"]
-default_model = "openai/gpt-oss-120b:free"
+# 2026-04-28: gpt-oss-120b:free → kimi-k2.6 via Ollama Pro. Coding-
+# specialized, faster than gpt-oss, on the same OLLAMA_CLOUD_KEY so
+# no extra provider hop.
+default_model = "kimi-k2.6"
 matrix_corpus = "workers_500k_v8"

 [[task_class]]
@@ -58,7 +61,9 @@ matrix_corpus = "kb_team_runs_v1"
 name = "doc_drift_check"
 preferred_mode = "drift"
 fallback_modes = ["validator"]
-default_model = "gpt-oss:120b"
+# 2026-04-28: gpt-oss:120b → gemini-3-flash-preview via Ollama Pro.
+# Speed leader on factual checking, same OLLAMA_CLOUD_KEY.
+default_model = "gemini-3-flash-preview"
 matrix_corpus = "distilled_factual_v20260423095819"

 [[task_class]]
@@ -27,10 +27,15 @@ name = "ollama_cloud"
 base_url = "https://ollama.com"
 auth = "bearer"
 auth_env = "OLLAMA_CLOUD_KEY"
-default_model = "gpt-oss:120b"
-# Cloud-tier Ollama. Key resolved from OLLAMA_CLOUD_KEY env at gateway
-# boot. Model-prefix routing: "cloud/<model>" auto-routes here
-# (see gateway::v1::resolve_provider).
+default_model = "deepseek-v3.2"
+# Cloud-tier Ollama (Pro plan as of 2026-04-28). Key resolved from
+# OLLAMA_CLOUD_KEY at gateway boot; Pro tier upgraded the account so
+# rate limits + model access widen without a key change. Model-prefix
+# routing: "cloud/<model>" auto-routes here. 39-model fleet now
+# includes deepseek-v3.2, deepseek-v4-{flash,pro}, gemini-3-flash-
+# preview, glm-{5,5.1}, kimi-k2.6, qwen3-coder-next.
+# 2026-04-28: default upgraded gpt-oss:120b → deepseek-v3.2 (newest
+# DeepSeek revision; kimi-k2:1t still upstream-broken with HTTP 500).

 [[provider]]
 name = "openrouter"
@@ -38,7 +43,7 @@ base_url = "https://openrouter.ai/api/v1"
 auth = "bearer"
 auth_env = "OPENROUTER_API_KEY"
 auth_fallback_files = ["/home/profit/.env", "/root/llm_team_config.json"]
-default_model = "openai/gpt-oss-120b:free"
+default_model = "x-ai/grok-4.1-fast"
 # Multi-provider gateway. Covers Anthropic, Google, OpenAI, MiniMax,
 # Qwen, Gemma, etc. Key resolved via crates/gateway/src/v1/openrouter.rs
 # resolve_openrouter_key() — env first, then fallback files.
@@ -3,10 +3,26 @@ use serde::{Deserialize, Serialize};
 use std::time::Duration;

 /// HTTP client for the Python AI sidecar.
+///
+/// `generate()` has two transport modes:
+/// - When `gateway_url` is None (default), it posts to
+///   `${base_url}/generate` (sidecar direct).
+/// - When `gateway_url` is `Some(url)`, it posts to
+///   `${url}/v1/chat` with `provider="ollama"` so the call appears
+///   in `/v1/usage` and Langfuse traces.
+///
+/// `embed()`, `rerank()`, and admin methods always go direct to the
+/// sidecar — no `/v1` equivalent yet, no point round-tripping.
+///
+/// Phase 44 part 2 (2026-04-27): the gateway URL is wired in by
+/// callers that want observability (vectord modules); it's left
+/// unset by callers that ARE the gateway internals (avoids self-loops
+/// + redundant hops).
 #[derive(Clone)]
 pub struct AiClient {
    client: Client,
    base_url: String,
+    gateway_url: Option<String>,
 }

 // -- Request/Response types --
@@ -86,9 +102,22 @@ impl AiClient {
        Self {
            client,
            base_url: base_url.trim_end_matches('/').to_string(),
+            gateway_url: None,
        }
    }

+    /// Same as `new`, but every `generate()` is routed through
+    /// `${gateway_url}/v1/chat` (provider=ollama) for observability.
+    /// Use this for callers OUTSIDE the gateway. Inside the gateway
+    /// itself, prefer `new()` — calling /v1/chat from /v1/chat works
+    /// (no infinite loop, ollama_arm doesn't use AiClient) but adds
+    /// a wasted localhost hop.
+    pub fn new_with_gateway(base_url: &str, gateway_url: &str) -> Self {
+        let mut c = Self::new(base_url);
+        c.gateway_url = Some(gateway_url.trim_end_matches('/').to_string());
+        c
+    }
+
    pub async fn health(&self) -> Result<serde_json::Value, String> {
        let resp = self.client
            .get(format!("{}/health", self.base_url))
@@ -114,6 +143,13 @@ impl AiClient {
    }

    pub async fn generate(&self, req: GenerateRequest) -> Result<GenerateResponse, String> {
+        if let Some(gw) = self.gateway_url.as_deref() {
+            return self.generate_via_gateway(gw, req).await;
+        }
+        // Direct-sidecar legacy path. Used by gateway internals (so
+        // ollama_arm can call sidecar without a self-loop) and by
+        // any consumer that wants raw transport without /v1/usage
+        // accounting.
        let resp = self.client
            .post(format!("{}/generate", self.base_url))
            .json(&req)
@@ -128,6 +164,59 @@ impl AiClient {
        resp.json().await.map_err(|e| format!("generate parse error: {e}"))
    }

+    /// Phase 44 part 2: route generate() through the gateway's
+    /// /v1/chat with provider="ollama" so the call lands in
+    /// /v1/usage + Langfuse. Translates between the sidecar
+    /// GenerateRequest/Response shape and the OpenAI-compat
+    /// chat shape on the wire.
+    async fn generate_via_gateway(&self, gateway_url: &str, req: GenerateRequest) -> Result<GenerateResponse, String> {
+        let mut messages = Vec::with_capacity(2);
+        if let Some(sys) = &req.system {
+            messages.push(serde_json::json!({"role": "system", "content": sys}));
+        }
+        messages.push(serde_json::json!({"role": "user", "content": req.prompt}));
+        let mut body = serde_json::json!({
+            "messages": messages,
+            "provider": "ollama",
+        });
+        if let Some(m) = &req.model { body["model"] = serde_json::json!(m); }
+        if let Some(t) = req.temperature { body["temperature"] = serde_json::json!(t); }
+        if let Some(mt) = req.max_tokens { body["max_tokens"] = serde_json::json!(mt); }
+        if let Some(th) = req.think { body["think"] = serde_json::json!(th); }
+
+        let resp = self.client
+            .post(format!("{}/v1/chat", gateway_url))
+            .json(&body)
+            .send()
+            .await
+            .map_err(|e| format!("/v1/chat request failed: {e}"))?;
+        if !resp.status().is_success() {
+            let text = resp.text().await.unwrap_or_default();
+            return Err(format!("/v1/chat error: {text}"));
+        }
+        let parsed: serde_json::Value = resp.json().await
+            .map_err(|e| format!("/v1/chat parse error: {e}"))?;
+
+        let text = parsed
+            .pointer("/choices/0/message/content")
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+        let model = parsed.get("model")
+            .and_then(|v| v.as_str())
+            .unwrap_or_else(|| req.model.as_deref().unwrap_or(""))
+            .to_string();
+        let prompt_tokens = parsed.pointer("/usage/prompt_tokens").and_then(|v| v.as_u64());
+        let completion_tokens = parsed.pointer("/usage/completion_tokens").and_then(|v| v.as_u64());
+
+        Ok(GenerateResponse {
+            text,
+            model,
+            tokens_evaluated: prompt_tokens,
+            tokens_generated: completion_tokens,
+        })
+    }
+
    pub async fn rerank(&self, req: RerankRequest) -> Result<RerankResponse, String> {
        let resp = self.client
            .post(format!("{}/rerank", self.base_url))
@@ -195,8 +195,11 @@ pub async fn generate_continuable<G: TextGenerator>(
        let req = make_request(opts, prompt.to_string(), current_max);
        let resp = generator.generate_text(req).await?;
        calls += 1;
-        prompt_tokens = prompt_tokens.saturating_add(resp.tokens_evaluated.unwrap_or(0) as u32);
-        completion_tokens = completion_tokens.saturating_add(resp.tokens_generated.unwrap_or(0) as u32);
+        // u32::try_from saturates at u32::MAX instead of silently
+        // truncating bits when tokens_evaluated/_generated comes back
+        // as a u64 > 4 billion. Caught 2026-04-27 by Opus self-audit.
+        prompt_tokens = prompt_tokens.saturating_add(u32::try_from(resp.tokens_evaluated.unwrap_or(0)).unwrap_or(u32::MAX));
+        completion_tokens = completion_tokens.saturating_add(u32::try_from(resp.tokens_generated.unwrap_or(0)).unwrap_or(u32::MAX));
        if !resp.text.trim().is_empty() {
            combined = resp.text;
            break;
@@ -227,8 +230,8 @@ pub async fn generate_continuable<G: TextGenerator>(
        let req = make_request(opts, cont_prompt, current_max.min(opts.budget_cap));
        let resp = generator.generate_text(req).await?;
        calls += 1;
-        prompt_tokens = prompt_tokens.saturating_add(resp.tokens_evaluated.unwrap_or(0) as u32);
-        completion_tokens = completion_tokens.saturating_add(resp.tokens_generated.unwrap_or(0) as u32);
+        prompt_tokens = prompt_tokens.saturating_add(u32::try_from(resp.tokens_evaluated.unwrap_or(0)).unwrap_or(u32::MAX));
+        completion_tokens = completion_tokens.saturating_add(u32::try_from(resp.tokens_generated.unwrap_or(0)).unwrap_or(u32::MAX));
        combined.push_str(&resp.text);
        continuations += 1;
    }
@@ -13,6 +13,7 @@ ingestd = { path = "../ingestd" }
 vectord = { path = "../vectord" }
 journald = { path = "../journald" }
 truth = { path = "../truth" }
+validator = { path = "../validator" }
 tokio = { workspace = true }
 axum = { workspace = true }
 serde = { workspace = true }
@@ -582,10 +582,10 @@ impl ExecutionLoop {
    /// Phase 20 step (8) — T3 overseer escalation.
    ///
    /// When the local executor/reviewer loop can't self-correct, call
-    /// the cloud overseer (`gpt-oss:120b` via Ollama Cloud) with (a)
-    /// the KB context — recent outcomes + prior corrections for this
-    /// sig_hash + task_class, across every profile that has run it —
-    /// and (b) the recent log tail. Its output is appended as a
+    /// the cloud overseer (`claude-opus-4-7` via OpenCode Zen) with
+    /// (a) the KB context — recent outcomes + prior corrections for
+    /// this sig_hash + task_class, across every profile that has run
+    /// it — and (b) the recent log tail. Its output is appended as a
    /// `system` role turn so the next executor generation sees it,
    /// AND written to `data/_kb/overseer_corrections.jsonl` so every
    /// future profile activation reads from the same learning pool.
@@ -593,9 +593,16 @@ impl ExecutionLoop {
    /// This is the "pipe to the overviewer" piece from 2026-04-23 —
    /// the overseer is now a first-class KB consumer AND producer, not
    /// a one-shot correction oracle.
+    ///
+    /// 2026-04-28: routed through OpenCode (Zen tier) for Claude Opus
+    /// 4.7. Frontier reasoning matters here because the overseer fires
+    /// only after local self-correction has failed twice — by that
+    /// point we need the strongest reasoning available, not the
+    /// cheapest token. Frequency is low so the Zen pay-per-token cost
+    /// stays bounded.
    async fn escalate_to_overseer(&mut self, turn: u32, reason: &str) -> Result<(), String> {
-        let Some(cloud_key) = self.state.ollama_cloud_key.clone() else {
-            return Err("OLLAMA_CLOUD_KEY not configured — skipping escalation".into());
+        let Some(opencode_key) = self.state.opencode_key.clone() else {
+            return Err("OPENCODE_API_KEY not configured — skipping escalation".into());
        };

        let kb = KbContext::load_for(&sig_hash(&self.req), &self.req.task_class).await;
@@ -604,16 +611,18 @@ impl ExecutionLoop {
        let started = std::time::Instant::now();
        let start_time = chrono::Utc::now();
        let chat_req = crate::v1::ChatRequest {
-            model: "gpt-oss:120b".to_string(),
+            model: "claude-opus-4-7".to_string(),
            messages: vec![crate::v1::Message::new_text("user", prompt.clone())],
            temperature: Some(0.1),
            max_tokens: None,
            stream: Some(false),
-            think: Some(true),    // overseer KEEPS thinking (Phase 20 rule)
-            provider: Some("ollama_cloud".into()),
+            // Anthropic models on opencode reject `think` (handled in
+            // the adapter), but we keep the intent flag for parity.
+            think: Some(true),
+            provider: Some("opencode".into()),
        };
-        let resp = crate::v1::ollama_cloud::chat(&cloud_key, &chat_req).await
-            .map_err(|e| format!("ollama_cloud: {e}"))?;
+        let resp = crate::v1::opencode::chat(&opencode_key, &chat_req).await
+            .map_err(|e| format!("opencode: {e}"))?;
        let latency_ms = started.elapsed().as_millis() as u64;
        let end_time = chrono::Utc::now();
        let correction_text: String = resp.choices.into_iter().next()
@@ -633,8 +642,8 @@ impl ExecutionLoop {
        if let Some(lf) = &self.state.langfuse {
            use crate::v1::langfuse_trace::ChatTrace;
            lf.emit_chat(ChatTrace {
-                provider: "ollama_cloud".into(),
-                model: "gpt-oss:120b".into(),
+                provider: "opencode".into(),
+                model: "claude-opus-4-7".into(),
                input: vec![crate::v1::Message::new_text("user", prompt.clone())],
                output: correction_text.clone(),
                prompt_tokens: resp.usage.prompt_tokens,
@@ -650,7 +659,7 @@ impl ExecutionLoop {

        // Append to the transcript so the next executor turn sees it.
        self.append(LogEntry::new(
-            turn, "system", "gpt-oss:120b", "overseer_correction",
+            turn, "system", "claude-opus-4-7", "overseer_correction",
            serde_json::json!({
                "reason": reason,
                "correction": correction_text,
@@ -672,7 +681,7 @@ impl ExecutionLoop {
            "task_class": self.req.task_class,
            "operation": self.req.operation,
            "reason": reason,
-            "model": "gpt-oss:120b",
+            "model": "claude-opus-4-7",
            "correction": correction_text,
            "applied_at_turn": turn,
            "kb_context_used": kb,
@@ -95,8 +95,35 @@ async fn main() {
        tracing::warn!("workspace rebuild: {e}");
    }

-    // AI sidecar client
-    let ai_client = aibridge::client::AiClient::new(&config.sidecar.url);
+    // AI sidecar clients — Phase 44 part 3 (2026-04-27).
+    //
+    // Two flavors of the same client:
+    // - `ai_client_direct` posts directly to ${sidecar}/generate. Used
+    //   inside the gateway by V1State + the legacy /ai proxy. These
+    //   call sites are themselves the implementation of /v1/chat
+    //   (or its sidecar shim), so routing them through /v1/chat
+    //   would self-loop.
+    // - `ai_client_observable` posts via ${gateway}/v1/chat with
+    //   provider="ollama". Used by vectord modules (autotune agent,
+    //   /vectors service) so their LLM calls land in /v1/usage and
+    //   Langfuse traces. Adds one localhost HTTP hop per call (~ms);
+    //   accepted for the observability gain.
+    //
+    // The gateway can call its own /v1/chat over localhost during
+    // boot's transient period because we don't fire any LLM calls
+    // until the listener is up — the observable client is just
+    // configured here, not exercised.
+    let ai_client_direct = aibridge::client::AiClient::new(&config.sidecar.url);
+    let gateway_self_url = format!("http://{}:{}", config.gateway.host, config.gateway.port);
+    let ai_client_observable = aibridge::client::AiClient::new_with_gateway(
+        &config.sidecar.url,
+        &gateway_self_url,
+    );
+    // Backwards-compat alias for the (many) existing references in this file.
+    // Defaults to direct so the existing wiring (V1State, /ai proxy)
+    // keeps its non-self-loop transport. New vectord wiring below
+    // explicitly uses ai_client_observable.
+    let ai_client = ai_client_direct.clone();

    // Vector service components — built before the router because both the
    // /vectors service AND ingestd need the agent handle to enqueue triggers.
@@ -134,7 +161,9 @@ async fn main() {
        agent_cfg,
        vectord::agent::AgentDeps {
            store: store.clone(),
-            ai_client: ai_client.clone(),
+            // Observable: autotune agent's LLM calls go through
+            // /v1/chat for /v1/usage + Langfuse visibility.
+            ai_client: ai_client_observable.clone(),
            catalog: registry.clone(),
            index_registry: index_reg.clone(),
            hnsw_store: hnsw.clone(),
@@ -189,7 +218,9 @@ async fn main() {
        }))
        .nest("/vectors", vectord::service::router(vectord::service::VectorState {
            store: store.clone(),
-            ai_client: ai_client.clone(),
+            // Observable: /vectors service's LLM calls (RAG, summary,
+            // playbook synthesis, etc.) flow through /v1/chat.
+            ai_client: ai_client_observable.clone(),
            job_tracker: vectord::jobs::JobTracker::new(),
            index_registry: index_reg.clone(),
            hnsw_store: hnsw,
@@ -295,6 +326,30 @@ async fn main() {
                }
                k
            },
+            validate_workers: {
+                // Load workers_500k.parquet snapshot for /v1/validate.
+                // Path overridable via LH_WORKERS_PARQUET env. Missing
+                // file is non-fatal — validators run schema/PII checks
+                // unaffected; only worker-existence checks fail clean.
+                let path_str = std::env::var("LH_WORKERS_PARQUET")
+                    .unwrap_or_else(|_| "/home/profit/lakehouse/data/datasets/workers_500k.parquet".into());
+                let path = std::path::Path::new(&path_str);
+                if path.exists() {
+                    match validator::staffing::parquet_lookup::load_workers_parquet(path) {
+                        Ok(lookup) => {
+                            tracing::info!("v1: workers parquet loaded from {} — /v1/validate worker-existence checks enabled", path_str);
+                            lookup
+                        }
+                        Err(e) => {
+                            tracing::warn!("v1: workers parquet at {} unreadable ({e}) — /v1/validate worker-existence checks will fail Consistency", path_str);
+                            std::sync::Arc::new(validator::InMemoryWorkerLookup::new())
+                        }
+                    }
+                } else {
+                    tracing::warn!("v1: workers parquet at {} not found — /v1/validate worker-existence checks will fail Consistency", path_str);
+                    std::sync::Arc::new(validator::InMemoryWorkerLookup::new())
+                }
+            },
            // Phase 40 early deliverable — Langfuse trace emitter.
            // Defaults match mcp-server/tracing.ts conventions so
            // gateway traces land in the same staffing project.
@@ -0,0 +1,313 @@
+//! /v1/iterate — the Phase 43 PRD's "generate → validate → correct → retry" loop.
+//!
+//! Closes the "0→85% with iteration" thesis structurally. A caller
+//! posts a prompt + artifact kind + validation context; the gateway:
+//!   1. Generates a JSON artifact via /v1/chat (any provider/model)
+//!   2. Extracts the JSON object from the model output
+//!   3. Validates via /v1/validate (FillValidator / EmailValidator /
+//!      PlaybookValidator with the shared WorkerLookup)
+//!   4. On ValidationError, appends the error to the prompt and
+//!      retries up to `max_iterations` (default 3)
+//!   5. Returns the accepted artifact + Report on success, OR the
+//!      attempt history + final error on max-iter exhaustion
+//!
+//! Internal calls go via HTTP loopback to localhost:gateway_port so
+//! the same /v1/usage tracking and Langfuse traces apply. A small
+//! latency cost (~1-3ms per loopback hop) for clean separation of
+//! concerns and observability.
+//!
+//! 2026-04-27 Phase 43 v3 part 3: this endpoint makes the iteration
+//! loop a first-class lakehouse capability rather than a per-caller
+//! re-implementation. Staffing executors, agent loops, and future
+//! validators all reach the same code path.
+
+use axum::{extract::State, http::StatusCode, response::IntoResponse, Json};
+use serde::{Deserialize, Serialize};
+
+const DEFAULT_MAX_ITERATIONS: u32 = 3;
+const LOOPBACK_TIMEOUT_SECS: u64 = 240;
+
+#[derive(Deserialize)]
+pub struct IterateRequest {
+    /// "fill" | "email" | "playbook" — picks which validator runs.
+    pub kind: String,
+    /// The prompt to seed generation. Validation errors from prior
+    /// attempts are appended on retry.
+    pub prompt: String,
+    /// Provider/model passed through to /v1/chat. e.g. "ollama_cloud"
+    /// + "kimi-k2.6", or "opencode" + "claude-haiku-4-5".
+    pub provider: String,
+    pub model: String,
+    /// Optional system prompt — sent to /v1/chat as the system message.
+    #[serde(default)]
+    pub system: Option<String>,
+    /// Validation context (target_count, city, state, role, client_id
+    /// for fills; candidate_id for emails). Forwarded to /v1/validate.
+    #[serde(default)]
+    pub context: Option<serde_json::Value>,
+    /// Cap on iteration count. Defaults to 3 per the Phase 43 PRD.
+    #[serde(default)]
+    pub max_iterations: Option<u32>,
+    /// Forwarded to /v1/chat. Defaults to 0.2 if unset.
+    #[serde(default)]
+    pub temperature: Option<f64>,
+    /// Forwarded to /v1/chat. Defaults to 4096 if unset.
+    #[serde(default)]
+    pub max_tokens: Option<u32>,
+}
+
+#[derive(Serialize)]
+pub struct IterateAttempt {
+    pub iteration: u32,
+    pub raw: String,
+    pub status: AttemptStatus,
+}
+
+#[derive(Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum AttemptStatus {
+    /// Model output didn't contain extractable JSON.
+    NoJson,
+    /// JSON extracted but failed validation; carries the error.
+    ValidationFailed { error: serde_json::Value },
+    /// Validation passed (last attempt's terminal status).
+    Accepted,
+}
+
+#[derive(Serialize)]
+pub struct IterateResponse {
+    pub artifact: serde_json::Value,
+    pub validation: serde_json::Value,
+    pub iterations: u32,
+    pub history: Vec<IterateAttempt>,
+}
+
+#[derive(Serialize)]
+pub struct IterateFailure {
+    pub error: String,
+    pub iterations: u32,
+    pub history: Vec<IterateAttempt>,
+}
+
+pub async fn iterate(
+    State(state): State<super::V1State>,
+    Json(req): Json<IterateRequest>,
+) -> impl IntoResponse {
+    let max_iter = req.max_iterations.unwrap_or(DEFAULT_MAX_ITERATIONS).max(1);
+    let temperature = req.temperature.unwrap_or(0.2);
+    let max_tokens = req.max_tokens.unwrap_or(4096);
+    let mut history: Vec<IterateAttempt> = Vec::with_capacity(max_iter as usize);
+    let mut current_prompt = req.prompt.clone();
+
+    let client = match reqwest::Client::builder()
+        .timeout(std::time::Duration::from_secs(LOOPBACK_TIMEOUT_SECS))
+        .build() {
+        Ok(c) => c,
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, format!("client build: {e}")).into_response(),
+    };
+    // Self-loopback to the gateway port. Carries gateway internal
+    // calls through /v1/chat + /v1/validate so /v1/usage tracks them.
+    let gateway = "http://127.0.0.1:3100";
+
+    for iteration in 0..max_iter {
+        // ── Generate ──
+        let mut messages = Vec::with_capacity(2);
+        if let Some(sys) = &req.system {
+            messages.push(serde_json::json!({"role": "system", "content": sys}));
+        }
+        messages.push(serde_json::json!({"role": "user", "content": current_prompt}));
+        let chat_body = serde_json::json!({
+            "messages": messages,
+            "provider": req.provider,
+            "model": req.model,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+        });
+        let raw = match call_chat(&client, gateway, &chat_body).await {
+            Ok(r) => r,
+            Err(e) => return (StatusCode::BAD_GATEWAY, format!("/v1/chat hop failed at iter {iteration}: {e}")).into_response(),
+        };
+
+        // ── Extract JSON ──
+        let artifact = match extract_json(&raw) {
+            Some(a) => a,
+            None => {
+                history.push(IterateAttempt {
+                    iteration,
+                    raw: raw.chars().take(2000).collect(),
+                    status: AttemptStatus::NoJson,
+                });
+                current_prompt = format!(
+                    "{}\n\nYour previous attempt did not contain a JSON object. Reply with ONLY a valid JSON object matching the requested artifact shape.",
+                    req.prompt,
+                );
+                continue;
+            }
+        };
+
+        // ── Validate ──
+        let validate_body = serde_json::json!({
+            "kind": req.kind,
+            "artifact": artifact,
+            "context": req.context.clone().unwrap_or(serde_json::Value::Null),
+        });
+        match call_validate(&client, gateway, &validate_body).await {
+            Ok(report) => {
+                history.push(IterateAttempt {
+                    iteration,
+                    raw: raw.chars().take(2000).collect(),
+                    status: AttemptStatus::Accepted,
+                });
+                return (StatusCode::OK, Json(IterateResponse {
+                    artifact,
+                    validation: report,
+                    iterations: iteration + 1,
+                    history,
+                })).into_response();
+            }
+            Err(err) => {
+                let err_summary = err.to_string();
+                history.push(IterateAttempt {
+                    iteration,
+                    raw: raw.chars().take(2000).collect(),
+                    status: AttemptStatus::ValidationFailed {
+                        error: serde_json::to_value(&err_summary).unwrap_or(serde_json::Value::Null),
+                    },
+                });
+                // Append validation feedback to prompt for next iter.
+                // The model sees concrete failure mode + retries with
+                // corrective context. This is the "observer correction"
+                // in Phase 43 PRD shape, simplified — the validator
+                // itself IS the observer for now.
+                current_prompt = format!(
+                    "{}\n\nPrior attempt failed validation:\n{}\n\nFix the specific issue above and respond with a corrected JSON object.",
+                    req.prompt, err_summary,
+                );
+                continue;
+            }
+        }
+    }
+
+    (StatusCode::UNPROCESSABLE_ENTITY, Json(IterateFailure {
+        error: format!("max iterations reached ({max_iter}) without passing validation"),
+        iterations: max_iter,
+        history,
+    })).into_response()
+}
+
+async fn call_chat(client: &reqwest::Client, gateway: &str, body: &serde_json::Value) -> Result<String, String> {
+    let resp = client.post(format!("{gateway}/v1/chat"))
+        .json(body)
+        .send()
+        .await
+        .map_err(|e| format!("chat hop: {e}"))?;
+    let status = resp.status();
+    if !status.is_success() {
+        let body = resp.text().await.unwrap_or_default();
+        return Err(format!("chat {}: {}", status, body.chars().take(300).collect::<String>()));
+    }
+    let parsed: serde_json::Value = resp.json().await.map_err(|e| format!("chat parse: {e}"))?;
+    Ok(parsed.pointer("/choices/0/message/content")
+        .and_then(|v| v.as_str())
+        .unwrap_or("")
+        .to_string())
+}
+
+async fn call_validate(client: &reqwest::Client, gateway: &str, body: &serde_json::Value) -> Result<serde_json::Value, String> {
+    let resp = client.post(format!("{gateway}/v1/validate"))
+        .json(body)
+        .send()
+        .await
+        .map_err(|e| format!("validate hop: {e}"))?;
+    let status = resp.status();
+    let parsed: serde_json::Value = resp.json().await.map_err(|e| format!("validate parse: {e}"))?;
+    if status.is_success() {
+        Ok(parsed)
+    } else {
+        // The /v1/validate endpoint returns a ValidationError JSON
+        // on 422; surface its structure verbatim so the prompt-
+        // appending step gets specific failure detail.
+        Err(serde_json::to_string(&parsed).unwrap_or_else(|_| format!("validation {} (unparseable body)", status)))
+    }
+}
+
+/// Extract the first JSON object from a model's output. Handles
+/// fenced code blocks (```json ... ```), bare braces, and stray
+/// prose around the JSON. Returns None on no extractable object.
+fn extract_json(raw: &str) -> Option<serde_json::Value> {
+    // Try fenced first.
+    let candidates: Vec<String> = {
+        let mut out = vec![];
+        let mut s = raw;
+        while let Some(start) = s.find("```") {
+            let after = &s[start + 3..];
+            // Skip optional language tag (json, etc.)
+            let body_start = after.find('\n').map(|n| n + 1).unwrap_or(0);
+            let body = &after[body_start..];
+            if let Some(end) = body.find("```") {
+                out.push(body[..end].trim().to_string());
+                s = &body[end + 3..];
+            } else { break; }
+        }
+        out
+    };
+    for c in &candidates {
+        if let Ok(v) = serde_json::from_str::<serde_json::Value>(c) {
+            if v.is_object() { return Some(v); }
+        }
+    }
+    // Fall back to outermost {...} balance.
+    let bytes = raw.as_bytes();
+    let mut depth = 0i32;
+    let mut start: Option<usize> = None;
+    for (i, &b) in bytes.iter().enumerate() {
+        match b {
+            b'{' => { if start.is_none() { start = Some(i); } depth += 1; }
+            b'}' => {
+                depth -= 1;
+                if depth == 0 {
+                    if let Some(s) = start {
+                        let slice = &raw[s..=i];
+                        if let Ok(v) = serde_json::from_str::<serde_json::Value>(slice) {
+                            if v.is_object() { return Some(v); }
+                        }
+                        start = None;
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+    None
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extract_json_from_fenced_block() {
+        let raw = "Here's my answer:\n```json\n{\"fills\": [{\"candidate_id\": \"W-1\"}]}\n```\nDone.";
+        let v = extract_json(raw).unwrap();
+        assert!(v.get("fills").is_some());
+    }
+
+    #[test]
+    fn extract_json_from_bare_braces() {
+        let raw = "Here you go: {\"fills\": [{\"candidate_id\": \"W-2\"}]}";
+        let v = extract_json(raw).unwrap();
+        assert!(v.get("fills").is_some());
+    }
+
+    #[test]
+    fn extract_json_returns_none_on_no_object() {
+        assert!(extract_json("just prose, no json").is_none());
+    }
+
+    #[test]
+    fn extract_json_picks_first_balanced() {
+        let raw = "{\"a\":1} then {\"b\":2}";
+        let v = extract_json(raw).unwrap();
+        assert_eq!(v.get("a").and_then(|v| v.as_i64()), Some(1));
+    }
+}
@@ -18,6 +18,8 @@ pub mod gemini;
 pub mod claude;
 pub mod kimi;
 pub mod opencode;
+pub mod validate;
+pub mod iterate;
 pub mod langfuse_trace;
 pub mod mode;
 pub mod respond;
@@ -68,6 +70,15 @@ pub struct V1State {
    /// `OPENCODE_API_KEY` env or `/etc/lakehouse/opencode.env`. None =
    /// provider="opencode" calls 503.
    pub opencode_key: Option<String>,
+    /// Shared WorkerLookup loaded once at startup from
+    /// workers_500k.parquet (path: LH_WORKERS_PARQUET env, default
+    /// data/datasets/workers_500k.parquet). Used by /v1/validate to
+    /// run FillValidator/EmailValidator with worker-existence checks.
+    /// Falls back to an empty InMemoryWorkerLookup if the file is
+    /// missing — validators still run schema/PII checks but every
+    /// worker-existence check fails (Consistency error), which is
+    /// the correct behavior when the roster isn't configured.
+    pub validate_workers: std::sync::Arc<dyn validator::WorkerLookup>,
    /// Phase 40 early deliverable — Langfuse client. None = tracing
    /// disabled (keys missing or container unreachable). Traces are
    /// fire-and-forget: never block the response path.
@@ -107,6 +118,9 @@ pub fn router(state: V1State) -> Router {
        .route("/mode", post(mode::route))
        .route("/mode/list", get(mode::list))
        .route("/mode/execute", post(mode::execute))
+        .route("/validate", post(validate::validate))
+        .route("/iterate", post(iterate::iterate))
+        .route("/health", get(health))
        .with_state(state)
 }

@@ -555,6 +569,43 @@ async fn usage(State(state): State<V1State>) -> impl IntoResponse {
    Json(snapshot)
 }

+/// Production operational health endpoint.
+///
+/// `/v1/health` reports per-subsystem status as a JSON object so an
+/// operator (or the lakehouse-auditor service, or a load balancer)
+/// can verify the gateway is fully booted, has its provider keys
+/// loaded, the worker roster is hot, and Langfuse is reachable.
+/// Returns 200 always — fields are observed-state, not pass/fail
+/// gates. A monitoring tool should evaluate the booleans + counts
+/// against its own thresholds.
+async fn health(State(state): State<V1State>) -> impl IntoResponse {
+    // Honest worker count via WorkerLookup::len. Production switchover
+    // verification: after swapping workers_500k.parquet → real Chicago
+    // data and restarting, this number should match the row count of
+    // the new file. 0 means the file was missing / unreadable / had a
+    // schema mismatch and the gateway booted with the empty fallback.
+    let workers_count = state.validate_workers.len();
+    let providers_configured = serde_json::json!({
+        "ollama_cloud": state.ollama_cloud_key.is_some(),
+        "openrouter": state.openrouter_key.is_some(),
+        "kimi": state.kimi_key.is_some(),
+        "opencode": state.opencode_key.is_some(),
+        "gemini": state.gemini_key.is_some(),
+        "claude": state.claude_key.is_some(),
+    });
+    let langfuse_configured = state.langfuse.is_some();
+    let usage_snapshot = state.usage.read().await.clone();
+    Json(serde_json::json!({
+        "status": "ok",
+        "workers_count": workers_count,
+        "workers_loaded": workers_count > 0,
+        "providers_configured": providers_configured,
+        "langfuse_configured": langfuse_configured,
+        "usage_total_requests": usage_snapshot.requests,
+        "usage_by_provider": usage_snapshot.by_provider.keys().collect::<Vec<_>>(),
+    }))
+}
+
 // Phase 38 is stateless — no session persistence yet. Return an empty
 // list in OpenAI-ish shape so clients that probe this endpoint don't
 // 404. Real session state lands in Phase 41 with the profile-system
@@ -0,0 +1,82 @@
+//! /v1/validate — gateway-side artifact validation endpoint.
+//!
+//! Phase 43 v3 part 2: makes the validator crate network-callable.
+//! Any caller (scrum loop, test harness, future agent) can POST a
+//! generated artifact and get back a Report (success) or
+//! ValidationError (failure with structured field/reason).
+//!
+//! Request shape:
+//!   POST /v1/validate
+//!   {
+//!     "kind": "fill" | "email" | "playbook",
+//!     "artifact": { ... },
+//!     "context": { ... }   // optional — folded into artifact._context
+//!   }
+//!
+//! Response on success: 200 + Report JSON
+//! Response on failure: 422 + ValidationError JSON
+//! Response on bad request: 400 + plain-text error
+//!
+//! The shared WorkerLookup is loaded once at gateway startup from
+//! workers_500k.parquet (path configurable via LH_WORKERS_PARQUET
+//! env, defaults to data/datasets/workers_500k.parquet). Falls back
+//! to an empty InMemoryWorkerLookup if the file is missing — the
+//! validators will still run schema/length/PII checks but worker-
+//! existence checks will all fail (Consistency error), which is the
+//! correct behavior when the roster isn't configured.
+
+use axum::{extract::State, http::StatusCode, response::IntoResponse, Json};
+use serde::Deserialize;
+use validator::{
+    Artifact, Validator, ValidationError,
+    staffing::{
+        fill::FillValidator,
+        email::EmailValidator,
+        playbook::PlaybookValidator,
+    },
+};
+
+#[derive(Deserialize)]
+pub struct ValidateRequest {
+    /// `"fill" | "email" | "playbook"` — picks which validator runs.
+    pub kind: String,
+    /// The artifact JSON (free-form; shape depends on `kind`).
+    pub artifact: serde_json::Value,
+    /// Optional context bag — merged into `artifact._context` so the
+    /// validator can read fields like `target_count`, `city`,
+    /// `client_id`, `candidate_id` without callers having to embed
+    /// `_context` in the artifact themselves.
+    #[serde(default)]
+    pub context: Option<serde_json::Value>,
+}
+
+pub async fn validate(
+    State(state): State<super::V1State>,
+    Json(req): Json<ValidateRequest>,
+) -> impl IntoResponse {
+    // Merge context into artifact under `_context` so validators can
+    // pull contract metadata uniformly.
+    let mut artifact_value = req.artifact;
+    if let Some(ctx) = req.context {
+        if let Some(obj) = artifact_value.as_object_mut() {
+            obj.insert("_context".to_string(), ctx);
+        }
+    }
+
+    // Dispatch.
+    let workers = state.validate_workers.clone();
+    let result: Result<validator::Report, ValidationError> = match req.kind.as_str() {
+        "fill" => FillValidator::new(workers).validate(&Artifact::FillProposal(artifact_value)),
+        "email" => EmailValidator::new(workers).validate(&Artifact::EmailDraft(artifact_value)),
+        "playbook" => PlaybookValidator.validate(&Artifact::Playbook(artifact_value)),
+        other => return (
+            StatusCode::BAD_REQUEST,
+            format!("unknown kind '{other}' — expected fill | email | playbook"),
+        ).into_response(),
+    };
+
+    match result {
+        Ok(report) => (StatusCode::OK, Json(report)).into_response(),
+        Err(e) => (StatusCode::UNPROCESSABLE_ENTITY, Json(e)).into_response(),
+    }
+}
@@ -9,3 +9,7 @@ serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+# Parquet loader for ParquetWorkerLookup (Phase 43 v3 — production
+# WorkerLookup backed by workers_500k.parquet snapshot).
+arrow = { workspace = true }
+parquet = { workspace = true }
@@ -130,6 +130,16 @@ pub struct WorkerRecord {
 /// hold an in-memory snapshot, not perform per-call I/O.
 pub trait WorkerLookup: Send + Sync {
    fn find(&self, candidate_id: &str) -> Option<WorkerRecord>;
+    /// Number of workers in the snapshot. Default 0 for impls that
+    /// genuinely don't know (e.g. a future SQL-backed lookup that
+    /// counts on demand). InMemoryWorkerLookup overrides with the
+    /// HashMap size; ParquetWorkerLookup constructs an
+    /// InMemoryWorkerLookup so it inherits the override. Used by
+    /// /v1/health to report data-load status during production
+    /// switchover (the Chicago dataset replaces synthetic test data;
+    /// the health endpoint is how operators verify the new file
+    /// loaded correctly without restart-and-pray).
+    fn len(&self) -> usize { 0 }
 }

 /// HashMap-backed lookup. Used by validator unit tests + as a
@@ -165,4 +175,7 @@ impl WorkerLookup for InMemoryWorkerLookup {
    fn find(&self, candidate_id: &str) -> Option<WorkerRecord> {
        self.rows.get(candidate_id).cloned()
    }
+    fn len(&self) -> usize {
+        self.rows.len()
+    }
 }
@@ -6,3 +6,4 @@
 pub mod fill;
 pub mod email;
 pub mod playbook;
+pub mod parquet_lookup;
@@ -0,0 +1,165 @@
+//! Production WorkerLookup backed by a workers_500k.parquet snapshot.
+//!
+//! Loads the full roster into memory at startup (one-shot). 500K rows
+//! at ~150 bytes per WorkerRecord ≈ 75 MB resident — fine for any
+//! production lakehouse process. Refresh is intentionally
+//! caller-driven (call `from_parquet` again to rebuild) rather than
+//! automatic — operators decide when staffing data has changed enough
+//! to justify the few-second reload.
+//!
+//! Schema mapping (workers_500k.parquet → WorkerRecord):
+//!   worker_id (int64)     → candidate_id = "W-{id}"
+//!   name (string)         → name
+//!   role (string)         → role
+//!   city (string)         → city
+//!   state (string)        → state
+//!   availability (double) → status: "active" if >0 else "inactive"
+//!
+//! No status column on workers_500k, so we derive from availability —
+//! the floor convention used elsewhere in the lakehouse staffing
+//! pipeline. Workers with availability=0.0 are treated as inactive
+//! (vacation, suspended, etc.). Once the Track-A.B `_safe` view ships
+//! with proper `status`, switch this loader to read it directly.
+//!
+//! Blacklist join is not done here — caller is expected to populate
+//! `blacklisted_clients` from a separate source (Phase 43 PRD says
+//! `client_blacklist` table; not yet defined). Default empty.
+
+use crate::{InMemoryWorkerLookup, WorkerLookup, WorkerRecord};
+use parquet::file::reader::{FileReader, SerializedFileReader};
+use parquet::record::Field;
+use std::fs::File;
+use std::path::Path;
+use std::sync::Arc;
+
+#[derive(Debug, thiserror::Error)]
+pub enum LookupLoadError {
+    #[error("opening parquet at {path}: {source}")]
+    Open { path: String, #[source] source: std::io::Error },
+    #[error("parsing parquet at {path}: {source}")]
+    Parse { path: String, #[source] source: parquet::errors::ParquetError },
+    #[error("missing required column {column}")]
+    MissingColumn { column: String },
+    #[error("row {row}: {reason}")]
+    BadRow { row: usize, reason: String },
+}
+
+/// Build an `InMemoryWorkerLookup` from a workers_500k-shaped parquet
+/// file. Returned as `Arc<dyn WorkerLookup>` to drop into validator
+/// constructors.
+pub fn load_workers_parquet(path: &Path) -> Result<Arc<dyn WorkerLookup>, LookupLoadError> {
+    let file = File::open(path).map_err(|e| LookupLoadError::Open {
+        path: path.display().to_string(),
+        source: e,
+    })?;
+    let reader = SerializedFileReader::new(file).map_err(|e| LookupLoadError::Parse {
+        path: path.display().to_string(),
+        source: e,
+    })?;
+
+    // Validate schema covers what we need before iterating rows.
+    let schema = reader.metadata().file_metadata().schema();
+    let column_names: Vec<&str> = schema.get_fields().iter().map(|f| f.name()).collect();
+    for required in &["worker_id", "name", "role", "city", "state", "availability"] {
+        if !column_names.contains(required) {
+            return Err(LookupLoadError::MissingColumn { column: (*required).to_string() });
+        }
+    }
+
+    let row_iter = reader.get_row_iter(None).map_err(|e| LookupLoadError::Parse {
+        path: path.display().to_string(),
+        source: e,
+    })?;
+
+    let mut records: Vec<WorkerRecord> = Vec::with_capacity(reader.metadata().file_metadata().num_rows() as usize);
+    let mut row_idx = 0usize;
+    for row_result in row_iter {
+        let row = row_result.map_err(|e| LookupLoadError::Parse {
+            path: path.display().to_string(),
+            source: e,
+        })?;
+        let mut worker_id: Option<i64> = None;
+        let mut name: Option<String> = None;
+        let mut role: Option<String> = None;
+        let mut city: Option<String> = None;
+        let mut state: Option<String> = None;
+        let mut availability: f64 = 0.0;
+        for (col_name, field) in row.get_column_iter() {
+            match (col_name.as_str(), field) {
+                ("worker_id", Field::Long(v)) => worker_id = Some(*v),
+                ("worker_id", Field::Int(v)) => worker_id = Some(*v as i64),
+                ("name", Field::Str(v)) => name = Some(v.clone()),
+                ("role", Field::Str(v)) => role = Some(v.clone()),
+                ("city", Field::Str(v)) => city = Some(v.clone()),
+                ("state", Field::Str(v)) => state = Some(v.clone()),
+                ("availability", Field::Double(v)) => availability = *v,
+                ("availability", Field::Float(v)) => availability = *v as f64,
+                _ => { /* extra columns ignored */ }
+            }
+        }
+        let id = worker_id.ok_or_else(|| LookupLoadError::BadRow {
+            row: row_idx,
+            reason: "worker_id missing or non-integer".into(),
+        })?;
+        let nm = name.ok_or_else(|| LookupLoadError::BadRow {
+            row: row_idx,
+            reason: "name missing".into(),
+        })?;
+        records.push(WorkerRecord {
+            candidate_id: format!("W-{id}"),
+            name: nm,
+            // status derived from availability (workers_500k has no
+            // status column). 0.0 → inactive, >0.0 → active.
+            status: if availability > 0.0 { "active".into() } else { "inactive".into() },
+            city,
+            state,
+            role,
+            blacklisted_clients: vec![],
+        });
+        row_idx += 1;
+    }
+
+    tracing::info!(
+        target: "validator.parquet_lookup",
+        rows = records.len(),
+        path = %path.display(),
+        "loaded workers parquet snapshot"
+    );
+
+    Ok(Arc::new(InMemoryWorkerLookup::from_records(records)))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::path::PathBuf;
+
+    /// Smoke test against the live workers_500k.parquet on disk.
+    /// Skipped automatically if the file isn't present (CI / sparse
+    /// checkouts) so the test suite stays portable.
+    #[test]
+    fn load_real_workers_500k() {
+        let path = PathBuf::from("/home/profit/lakehouse/data/datasets/workers_500k.parquet");
+        if !path.exists() {
+            eprintln!("skip: {} not present", path.display());
+            return;
+        }
+        let lookup = load_workers_parquet(&path).expect("load");
+        // Basic shape: at least one worker resolves and has the
+        // expected fields populated.
+        let probe = lookup.find("W-1");
+        assert!(probe.is_some(), "W-1 should exist in 500K-row parquet");
+        let w = probe.unwrap();
+        assert!(!w.name.is_empty(), "name should be populated");
+        assert!(w.status == "active" || w.status == "inactive");
+        assert!(w.role.is_some());
+        assert!(w.city.is_some());
+        assert!(w.state.is_some());
+    }
+
+    #[test]
+    fn missing_file_returns_error() {
+        let r = load_workers_parquet(Path::new("/nonexistent.parquet"));
+        assert!(matches!(r, Err(LookupLoadError::Open { .. })));
+    }
+}
@@ -146,6 +146,11 @@ pub fn router(state: VectorState) -> Router {
        // 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))
+        // Phase 45 closure (2026-04-27) — batch scan across all active
+        // playbooks. Operator runs this on a schedule (cron or manual);
+        // each newly-detected drift writes a row to
+        // data/_kb/doc_drift_corrections.jsonl for downstream review.
+        .route("/playbook_memory/doc_drift/scan", post(scan_doc_drift))
        // Pathway memory — consensus-designed sidecar (2026-04-24).
        // scrum_master_pipeline POSTs /pathway/insert at the end of each
        // review, calls /pathway/query before running the ladder for a
@@ -2539,6 +2544,119 @@ async fn check_doc_drift(
    })))
 }

+/// Phase 45 closure (2026-04-27) — POST /playbook_memory/doc_drift/scan
+///
+/// Iterates all active playbooks (non-retired, has doc_refs), runs
+/// drift check against context7 for each, flags drifted entries via
+/// PlaybookMemory::flag_doc_drift, and appends a row to
+/// data/_kb/doc_drift_corrections.jsonl for each drift detected.
+///
+/// Returns aggregate stats so an operator can see at-a-glance how
+/// many playbooks drifted and which tools moved.
+///
+/// Honors entries already flagged: they're counted in `already_flagged`
+/// (no double-flag, no duplicate corrections.jsonl row).
+async fn scan_doc_drift(
+    State(state): State<VectorState>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    use crate::doc_drift::{check_all_refs, DriftCheckerConfig, DriftOutcome};
+
+    let entries = state.playbook_memory.snapshot().await;
+    let now = chrono::Utc::now().to_rfc3339();
+    let cfg = DriftCheckerConfig::default();
+
+    let mut scanned = 0usize;
+    let mut newly_flagged = 0usize;
+    let mut already_flagged = 0usize;
+    let mut skipped_no_refs = 0usize;
+    let mut skipped_retired = 0usize;
+    let mut tool_counts: std::collections::HashMap<String, usize> = Default::default();
+    let mut corrections_rows: Vec<String> = vec![];
+
+    for e in entries.iter() {
+        if e.retired_at.is_some() { skipped_retired += 1; continue; }
+        if e.doc_refs.is_empty() { skipped_no_refs += 1; continue; }
+        if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+            already_flagged += 1;
+            continue;
+        }
+        scanned += 1;
+        let results = check_all_refs(&cfg, &e.doc_refs).await;
+        let drifted_tools: Vec<&str> = results.iter()
+            .filter(|r| matches!(r.outcome, DriftOutcome::Drifted { .. }))
+            .map(|r| r.tool.as_str())
+            .collect();
+        if drifted_tools.is_empty() { continue; }
+
+        // Flag the entry.
+        let flagged = state.playbook_memory.flag_doc_drift(&e.playbook_id).await
+            .unwrap_or(false);
+        if flagged { newly_flagged += 1; }
+        for t in &drifted_tools {
+            *tool_counts.entry(t.to_string()).or_insert(0) += 1;
+        }
+
+        // Build corrections.jsonl row — one per drifted playbook with
+        // the tool list inline. Downstream consumers (overview model,
+        // operator dashboard) read this to decide reviews + revisions.
+        let row = serde_json::json!({
+            "playbook_id": e.playbook_id,
+            "scanned_at": now,
+            "drifted_tools": drifted_tools,
+            "per_tool": results.iter().map(|r| {
+                let (drifted, current, src) = match &r.outcome {
+                    DriftOutcome::Drifted { current_snippet_hash, source_url } =>
+                        (true, Some(current_snippet_hash.clone()), source_url.clone()),
+                    _ => (false, None, None),
+                };
+                serde_json::json!({
+                    "tool": r.tool, "version_seen": r.version_seen,
+                    "drifted": drifted, "current_snippet_hash": current, "source_url": src,
+                })
+            }).collect::<Vec<_>>(),
+            "recommended_action": "review-and-resolve",
+        });
+        corrections_rows.push(row.to_string());
+    }
+
+    // Persist corrections.jsonl row(s) for the operator/overview model.
+    if !corrections_rows.is_empty() {
+        let path = std::path::PathBuf::from("/home/profit/lakehouse/data/_kb/doc_drift_corrections.jsonl");
+        if let Some(parent) = path.parent() {
+            if let Err(e) = tokio::fs::create_dir_all(parent).await {
+                tracing::warn!(target: "vectord.doc_drift", "create_dir_all {parent:?}: {e}");
+            }
+        }
+        let body = corrections_rows.join("\n") + "\n";
+        if let Err(e) = tokio::fs::OpenOptions::new()
+            .create(true).append(true).open(&path).await
+        {
+            tracing::warn!(target: "vectord.doc_drift", "open {path:?}: {e}");
+        } else {
+            use tokio::io::AsyncWriteExt;
+            match tokio::fs::OpenOptions::new().create(true).append(true).open(&path).await {
+                Ok(mut f) => {
+                    if let Err(e) = f.write_all(body.as_bytes()).await {
+                        tracing::warn!(target: "vectord.doc_drift", "append {path:?}: {e}");
+                    }
+                }
+                Err(e) => tracing::warn!(target: "vectord.doc_drift", "reopen {path:?}: {e}"),
+            }
+        }
+    }
+
+    Ok(Json(serde_json::json!({
+        "scanned_at": now,
+        "scanned": scanned,
+        "newly_flagged": newly_flagged,
+        "already_flagged": already_flagged,
+        "skipped_retired": skipped_retired,
+        "skipped_no_refs": skipped_no_refs,
+        "drifted_by_tool": tool_counts,
+        "corrections_written": corrections_rows.len(),
+    })))
+}
+
 /// Phase 45 slice 3 — POST /playbook_memory/doc_drift/resolve/{id}
 ///
 /// Human-in-the-loop re-admission. Stamps `doc_drift_reviewed_at`.
@@ -0,0 +1,24 @@
+{
+  "name": "candidates_safe",
+  "base_dataset": "candidates",
+  "columns": [
+    "candidate_id",
+    "first_name",
+    "city",
+    "state",
+    "skills",
+    "years_experience",
+    "status"
+  ],
+  "row_filter": "status != 'blocked'",
+  "column_redactions": {
+    "candidate_id": {
+      "kind": "mask",
+      "keep_prefix": 3,
+      "keep_suffix": 2
+    }
+  },
+  "created_at": "2026-04-27T15:42:00Z",
+  "created_by": "j",
+  "description": "PII-free candidate projection — drops last_name, email, phone, hourly_rate_usd. candidate_id masked (keep first 3, last 2). Visible to recruiter / mode-runner agents."
+}
@@ -0,0 +1,26 @@
+{
+  "name": "jobs_safe",
+  "base_dataset": "job_orders",
+  "columns": [
+    "job_order_id",
+    "client_id",
+    "title",
+    "vertical",
+    "status",
+    "city",
+    "state",
+    "zip",
+    "bill_rate",
+    "pay_rate"
+  ],
+  "column_redactions": {
+    "client_id": {
+      "kind": "mask",
+      "keep_prefix": 3,
+      "keep_suffix": 2
+    }
+  },
+  "created_at": "2026-04-27T15:42:00Z",
+  "created_by": "j",
+  "description": "Job-order projection with client_id masked. Drops description (often quotes client names verbatim, no text-scrubber available). bill_rate / pay_rate kept — commercial info, not PII per staffing PRD."
+}
@@ -0,0 +1,22 @@
+{
+  "name": "workers_safe",
+  "base_dataset": "workers_500k",
+  "columns": [
+    "worker_id",
+    "role",
+    "city",
+    "state",
+    "skills",
+    "certifications",
+    "archetype",
+    "reliability",
+    "responsiveness",
+    "engagement",
+    "compliance",
+    "availability"
+  ],
+  "column_redactions": {},
+  "created_at": "2026-04-27T15:42:00Z",
+  "created_by": "j",
+  "description": "PII-free worker projection — drops name, email, phone, zip, communications, resume_text. resume_text + communications carry verbatim PII (full names) and there's no in-view text scrubber, so they're dropped wholesale. Skills + certifications + scores carry the matching signal for staffing inference. Source for workers_500k_v9 vector corpus rebuild."
+}
@@ -349,6 +349,24 @@
  - Per-type endpoints: `/profiles/retrieval`, `/profiles/memory`, `/profiles/observer`
  - `profile_type` field on ModelProfile
  - Guard fix: automated scrumaudit.py finds real issues
+- [x] **Phase 42: Truth Layer** (2026-04-27 closure verified)
+  - `crates/truth/{lib,staffing,devops,loader}.rs`
+  - Staffing rules populated; devops scaffold by design
+  - `/v1/context` serves task_classes + rules; 37 tests green
+- [x] **Phase 43: Validation Pipeline** (2026-04-27)
+  - `crates/validator/` real validators + WorkerLookup + ParquetWorkerLookup
+  - 500K-row workers_500k.parquet loaded at gateway boot
+  - `POST /v1/validate` + `POST /v1/iterate` (the 0→85% loop)
+  - 33 validator tests green
+- [x] **Phase 44: Caller Migration** (2026-04-27)
+  - TS callers + aibridge::AiClient::new_with_gateway opt-in
+  - Vectord routed through /v1/chat for autotune + RAG
+  - scripts/check_phase44_callers.sh CI guard
+- [x] **Phase 45: Doc-Drift Detection** (2026-04-27)
+  - DocRef + doc_drift module + context7 bridge
+  - /doc_drift/check + /scan + /resolve endpoints
+  - data/_kb/doc_drift_corrections.jsonl writes
+  - boost exclusion of unreviewed drift-flagged entries
 - [ ] Fine-tuned domain models (Phase 25+)
 - [ ] Multi-node query distribution (only if ceilings bite)

@@ -550,13 +550,19 @@ async function analyzeErrors() {
    `[${f.endpoint}] ${f.input_summary}: ${f.error}`
  ).join("\n");

-  // Ask local model to diagnose
+  // Ask local model to diagnose. Phase 44 migration (2026-04-27):
+  // /v1/chat instead of legacy /ai/generate so /v1/usage tracks the
+  // call + Langfuse traces it. Same upstream model (qwen2.5 local).
  try {
-    const resp = await fetch(`${LAKEHOUSE}/ai/generate`, {
+    const resp = await fetch(`${LAKEHOUSE}/v1/chat`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
-        prompt: `You are a system reliability observer. Analyze these recent failures and suggest fixes:
+        model: "qwen2.5",
+        provider: "ollama",
+        messages: [{
+          role: "user",
+          content: `You are a system reliability observer. Analyze these recent failures and suggest fixes:

 ${errorSummary}

@@ -566,14 +572,15 @@ For each error:
 3. Should this be added to the playbook as a "don't do this"?

 Be specific and actionable. Under 200 words.`,
-        model: "qwen2.5",
+        }],
        max_tokens: 400,
        temperature: 0.2,
      }),
    });
-    const analysis = await resp.json();
-    if (analysis.text) {
-      console.error(`[observer] Error analysis:\n${analysis.text}`);
+    const analysis = await resp.json() as any;
+    const analysisText = analysis?.choices?.[0]?.message?.content ?? "";
+    if (analysisText) {
+      console.error(`[observer] Error analysis:\n${analysisText}`);
      // Log the analysis as a playbook entry
      await fetch(`${GATEWAY}/log`, {
        method: "POST",
@@ -581,7 +588,7 @@ Be specific and actionable. Under 200 words.`,
        body: JSON.stringify({
          operation: `error_analysis: ${failures.length} failures`,
          approach: "LLM-analyzed error patterns",
-          result: analysis.text.slice(0, 500),
+          result: analysisText.slice(0, 500),
          context: errorSummary.slice(0, 500),
        }),
      });
@@ -762,7 +769,7 @@ async function tailOverseerCorrections(): Promise<number> {
    try { row = JSON.parse(line); } catch { continue; }
    const op: ObservedOp = {
      timestamp: row.created_at ?? new Date().toISOString(),
-      endpoint: `overseer:${row.model ?? "gpt-oss:120b"}`,
+      endpoint: `overseer:${row.model ?? "claude-opus-4-7"}`,
      input_summary: `${row.task_class ?? "?"}: ${row.reason ?? "escalation"}`,
      // Correction itself is neither success nor failure — it's a
      // mitigation attempt. We mark success=true so analyzeErrors
@@ -0,0 +1,58 @@
+#!/usr/bin/env bash
+# Phase 44 caller-migration guard. Fails if any non-adapter file
+# fetches the sidecar's /generate endpoint or hits Ollama Cloud's
+# /api/generate directly. Adapter files (gateway provider crate +
+# the sidecar's own Python implementation) are exempt.
+#
+# Run:    ./scripts/check_phase44_callers.sh
+# CI:     fail-loud (exits non-zero on regression)
+# Watch:  pre-commit hook — invoke from .git/hooks/pre-commit
+
+set -e
+cd "$(dirname "$0")/.."
+
+FORBIDDEN_TS=$(grep -rEln "fetch\([^)]*[/\$]generate" \
+    --include="*.ts" \
+    --exclude-dir=node_modules \
+    --exclude-dir=target \
+    --exclude-dir=.git \
+    . 2>/dev/null | grep -v "^\./sidecar/" || true)
+
+FORBIDDEN_RS=$(grep -rEln "post\([^)]*\"\.?/generate\"" \
+    --include="*.rs" \
+    --exclude-dir=target \
+    . 2>/dev/null | \
+    grep -vE "^\./crates/(gateway|aibridge)/" || true)
+
+# Ollama Cloud /api/generate outside the gateway adapter. Match only
+# when the URL appears in an actual fetch/post call (not in a comment).
+# Tightened 2026-04-27 — pre-tightening regex flagged prose mentions.
+FORBIDDEN_CLOUD=$(grep -rEln "(fetch|client\.post)\([^)]*api/generate" \
+    --include="*.ts" --include="*.rs" \
+    --exclude-dir=node_modules \
+    --exclude-dir=target \
+    --exclude-dir=.git \
+    . 2>/dev/null | \
+    grep -vE "^\./(crates/gateway|sidecar)" || true)
+
+ANY_FAIL=0
+if [ -n "$FORBIDDEN_TS" ]; then
+    echo "❌ Direct sidecar /generate calls (migrate to /v1/chat):"
+    echo "$FORBIDDEN_TS" | sed 's/^/   /'
+    ANY_FAIL=1
+fi
+if [ -n "$FORBIDDEN_RS" ]; then
+    echo "❌ Direct Rust /generate post() calls (migrate via gateway adapter):"
+    echo "$FORBIDDEN_RS" | sed 's/^/   /'
+    ANY_FAIL=1
+fi
+if [ -n "$FORBIDDEN_CLOUD" ]; then
+    echo "❌ Direct Ollama Cloud /api/generate (migrate to gateway provider=ollama_cloud):"
+    echo "$FORBIDDEN_CLOUD" | sed 's/^/   /'
+    ANY_FAIL=1
+fi
+
+if [ $ANY_FAIL -eq 0 ]; then
+    echo "✅ Phase 44 caller-migration: clean (no direct /generate outside adapters)"
+fi
+exit $ANY_FAIL
@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""
+build_fill_events.py — Decision A from the synthetic-data gap report.
+
+Walks tests/multi-agent/scenarios/*.json (43 client-day scenarios) and
+data/_playbook_lessons/*.json (64 retrospective outcomes) and emits a
+single normalized fill_events.parquet at data/datasets/fill_events.parquet.
+
+Pure deterministic normalization — no LLM, no new data. Each scenario
+event becomes one row. Lesson outcomes augment scenario events with
+success/fail counts where (client, date, city, state) matches.
+
+Reproducibility: identical inputs → bit-identical output. event_id is
+SHA1(client|date|role|at|city) truncated to 16 hex chars; rows are
+sorted by event_id before write so re-runs produce the same parquet.
+"""
+
+import hashlib
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+REPO = Path(__file__).resolve().parents[2]
+SCENARIO_DIR = REPO / "tests" / "multi-agent" / "scenarios"
+LESSONS_DIR = REPO / "data" / "_playbook_lessons"
+OUT_PATH = REPO / "data" / "datasets" / "fill_events.parquet"
+
+
+def event_id(client: str, date: str, role: str, at: str, city: str) -> str:
+    h = hashlib.sha1(f"{client}|{date}|{role}|{at}|{city}".encode()).hexdigest()
+    return h[:16]
+
+
+def load_lessons() -> dict:
+    """Returns map of (client, date) → outcome dict."""
+    out: dict = {}
+    for path in sorted(LESSONS_DIR.glob("*.json")):
+        try:
+            d = json.loads(path.read_text())
+        except json.JSONDecodeError:
+            continue
+        client = d.get("client")
+        date = d.get("date")
+        if not client or not date:
+            continue
+        out[(client, date)] = {
+            "outcome_events_total": d.get("events_total"),
+            "outcome_events_ok": d.get("events_ok"),
+            "outcome_checkpoint_count": d.get("checkpoint_count"),
+            "outcome_model": d.get("model"),
+            "outcome_cloud": d.get("cloud"),
+            "outcome_lesson_path": str(path.relative_to(REPO)),
+        }
+    return out
+
+
+def load_scenarios(lessons: dict) -> list[dict]:
+    rows: list[dict] = []
+    for path in sorted(SCENARIO_DIR.glob("scen_*.json")):
+        try:
+            d = json.loads(path.read_text())
+        except json.JSONDecodeError:
+            continue
+        client = d.get("client")
+        date = d.get("date")
+        contract = d.get("contract") or {}
+        events = d.get("events") or []
+        if not client or not date or not events:
+            continue
+        outcome = lessons.get((client, date), {})
+        for event in events:
+            role = event.get("role") or ""
+            at = event.get("at") or ""
+            city = event.get("city") or ""
+            state = event.get("state") or ""
+            rows.append({
+                "event_id": event_id(client, date, role, at, city),
+                "source_file": str(path.relative_to(REPO)),
+                "source_kind": "scenario",
+                "client": client,
+                "date": date,
+                "city": city,
+                "state": state,
+                "role": role,
+                "count": int(event.get("count") or 0),
+                "kind": event.get("kind") or "",
+                "at": at,
+                "shift_start": event.get("shift_start") or "",
+                "contract_deadline": contract.get("deadline"),
+                "contract_budget_per_hour_max": contract.get("budget_per_hour_max"),
+                "contract_local_bonus_per_hour": contract.get("local_bonus_per_hour"),
+                "contract_local_bonus_radius_mi": contract.get("local_bonus_radius_mi"),
+                "contract_fill_requirement": contract.get("fill_requirement"),
+                "outcome_events_total": outcome.get("outcome_events_total"),
+                "outcome_events_ok": outcome.get("outcome_events_ok"),
+                "outcome_checkpoint_count": outcome.get("outcome_checkpoint_count"),
+                "outcome_model": outcome.get("outcome_model"),
+                "outcome_cloud": outcome.get("outcome_cloud"),
+                "outcome_lesson_path": outcome.get("outcome_lesson_path"),
+            })
+    return rows
+
+
+def main() -> int:
+    lessons = load_lessons()
+    rows = load_scenarios(lessons)
+    if not rows:
+        print("no rows produced — scenario dir empty?", file=sys.stderr)
+        return 1
+    rows.sort(key=lambda r: r["event_id"])
+
+    schema = pa.schema([
+        ("event_id", pa.string()),
+        ("source_file", pa.string()),
+        ("source_kind", pa.string()),
+        ("client", pa.string()),
+        ("date", pa.string()),
+        ("city", pa.string()),
+        ("state", pa.string()),
+        ("role", pa.string()),
+        ("count", pa.int32()),
+        ("kind", pa.string()),
+        ("at", pa.string()),
+        ("shift_start", pa.string()),
+        ("contract_deadline", pa.string()),
+        ("contract_budget_per_hour_max", pa.int32()),
+        ("contract_local_bonus_per_hour", pa.int32()),
+        ("contract_local_bonus_radius_mi", pa.int32()),
+        ("contract_fill_requirement", pa.string()),
+        ("outcome_events_total", pa.int32()),
+        ("outcome_events_ok", pa.int32()),
+        ("outcome_checkpoint_count", pa.int32()),
+        ("outcome_model", pa.string()),
+        ("outcome_cloud", pa.bool_()),
+        ("outcome_lesson_path", pa.string()),
+    ])
+    table = pa.Table.from_pylist(rows, schema=schema)
+
+    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    pq.write_table(table, OUT_PATH, compression="snappy")
+
+    matched = sum(1 for r in rows if r["outcome_events_total"] is not None)
+    print(f"fill_events.parquet written: {OUT_PATH.relative_to(REPO)}")
+    print(f"  rows: {len(rows)}")
+    print(f"  scenarios: {len({r['source_file'] for r in rows})}")
+    print(f"  with outcome: {matched}")
+    print(f"  unique (client,date): {len({(r['client'], r['date']) for r in rows})}")
+    print(f"  generated_at: {datetime.now(timezone.utc).isoformat(timespec='seconds')}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+# build_workers_v9.sh — Decision B (corpus rebuild side).
+#
+# Rebuilds workers_500k_v9 vector corpus from workers_safe view rather
+# than the raw workers_500k table. Closes the PII enforcement gap
+# (verified 2026-04-27 that v8 was built directly from raw — LLM saw
+# names/emails/phones/resume_text for every staffing query).
+#
+# Run as a background job — embedding 500K chunks took ~4 min for v8
+# of 50K rows; v9 of 500K rows will be 30+ min. Do not block on this.
+#
+# Usage:
+#     ./scripts/staffing/build_workers_v9.sh
+#     LH_GATEWAY=http://localhost:3100 ./scripts/staffing/build_workers_v9.sh
+#
+# After it completes:
+#   - Verify via: curl /vectors/indexes/workers_500k_v9 | jq
+#   - Flip config/modes.toml `staffing_inference` matrix_corpus to v9
+#   - Restart gateway to pick up the modes.toml change
+
+set -euo pipefail
+
+GATEWAY="${LH_GATEWAY:-http://localhost:3100}"
+
+# The /vectors/index endpoint accepts {name, sql, embed_model, ...}.
+# SQL pulls from workers_safe (see data/_catalog/views/workers_safe.json)
+# so the embedded text never contained raw PII by construction.
+#
+# Concatenated text is what gets embedded — keep it short enough that
+# 500K rows × N chunks fits in disk + memory budgets but still carries
+# the match signal (role, location, skills, scores).
+
+BODY=$(cat <<'JSON'
+{
+  "name": "workers_500k_v9",
+  "sql": "SELECT CAST(worker_id AS VARCHAR) AS doc_id, CONCAT(role, ' in ', city, ', ', state, '. Skills: ', COALESCE(skills, ''), '. Certifications: ', COALESCE(certifications, ''), '. Archetype: ', COALESCE(archetype, ''), '. Scores — reliability ', CAST(reliability AS VARCHAR), ', responsiveness ', CAST(responsiveness AS VARCHAR), ', availability ', CAST(availability AS VARCHAR), '.') AS text FROM workers_safe",
+  "embed_model": "nomic-embed-text",
+  "chunk_size": 500,
+  "overlap": 50,
+  "source_dataset": "workers_safe",
+  "bucket": "primary"
+}
+JSON
+)
+
+echo "POSTing /vectors/index → workers_500k_v9 (background job)..."
+curl -sS -X POST "${GATEWAY}/vectors/index" \
+  -H 'content-type: application/json' \
+  -d "$BODY"
+echo
+echo "Job started. Monitor progress:"
+echo "  curl ${GATEWAY}/vectors/indexes/workers_500k_v9 | jq"
+echo "  watch -n 5 'curl -s ${GATEWAY}/vectors/jobs | jq'"
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""
+fixup_phone_type.py — Decision D from the synthetic-data gap report.
+
+Converts workers_500k.parquet `phone` column from int64 → string. Phones
+in this dataset are 11-digit US numbers (1 + area + 7), e.g. 13122277740.
+Stored as int64, the column compares fine numerically but breaks join
+keys with string-typed phone columns elsewhere (formatted "+1...", or
+loaded from a CSV).
+
+Backs up the original to workers_500k.parquet.bak-<date> before write.
+Idempotent: detects when the fix has already been applied and exits 0.
+
+Usage:
+    python3 scripts/staffing/fixup_phone_type.py
+"""
+
+import datetime as dt
+import shutil
+import sys
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.compute as pc
+import pyarrow.parquet as pq
+
+REPO = Path(__file__).resolve().parents[2]
+TARGET = REPO / "data" / "datasets" / "workers_500k.parquet"
+
+
+def main() -> int:
+    if not TARGET.exists():
+        print(f"missing: {TARGET}", file=sys.stderr)
+        return 1
+
+    table = pq.read_table(TARGET)
+    phone_field = table.schema.field("phone")
+    if phone_field.type == pa.string():
+        print(f"phone is already string — no-op")
+        return 0
+
+    today = dt.date.today().isoformat()
+    backup = TARGET.with_suffix(f".parquet.bak-{today}")
+    if not backup.exists():
+        shutil.copy2(TARGET, backup)
+        print(f"backup: {backup.relative_to(REPO)}")
+
+    phone_str = pc.cast(table["phone"], pa.string())
+    new_table = table.set_column(
+        table.schema.get_field_index("phone"),
+        pa.field("phone", pa.string()),
+        phone_str,
+    )
+
+    pq.write_table(new_table, TARGET, compression="snappy")
+    rounds_trip = pq.read_table(TARGET, columns=["phone"])
+    sample = rounds_trip["phone"].slice(0, 3).to_pylist()
+    print(f"wrote: {TARGET.relative_to(REPO)}")
+    print(f"phone type: {rounds_trip.schema.field('phone').type}")
+    print(f"sample: {sample}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -161,16 +161,17 @@ const TOOL_SCHEMA = `Available tools (call by emitting JSON like: {"tool": "name
 // ─── AGENT LOOP ───

 async function callAgent(messages: Array<{role: string; content: string}>): Promise<string> {
-  // think:false disables hidden reasoning so all generated tokens go to
-  // visible response. qwen3.5:latest defaults to thinking and silently
-  // burns the token budget otherwise.
-  const r = await fetch(`${SIDECAR}/generate`, {
+  // Phase 44 migration (2026-04-27): /v1/chat instead of direct sidecar
+  // /generate so /v1/usage tracks the call, Langfuse traces it.
+  // think:false still disables hidden reasoning so generated tokens
+  // go to visible response — qwen3.5:latest defaults to thinking.
+  const r = await fetch(`${GATEWAY}/v1/chat`, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      model: AGENT_MODEL,
-      prompt: messages.map(m => `${m.role.toUpperCase()}:\n${m.content}`).join("\n\n") + "\n\nASSISTANT:\n",
-      stream: false,
+      provider: "ollama",
+      messages,
      max_tokens: 1500,
      think: false,
    }),
@@ -178,7 +179,7 @@ async function callAgent(messages: Array<{role: string; content: string}>): Prom
  });
  if (!r.ok) throw new Error(`agent ${r.status}: ${(await r.text()).slice(0, 200)}`);
  const j: any = await r.json();
-  return String(j.text ?? j.response ?? "").trim();
+  return String(j?.choices?.[0]?.message?.content ?? "").trim();
 }

 function extractToolCall(response: string): { tool: string; args: any } | null {
@@ -372,26 +372,34 @@ export async function generate(model: string, prompt: string, opts: {
  bypass_budget?: boolean;
  think?: boolean;
 } = {}): Promise<string> {
+  // Phase 44 migration (2026-04-27): was hitting `${SIDECAR}/generate`
+  // directly, bypassing the gateway's /v1/usage accounting + Langfuse
+  // tracing. Now flows through /v1/chat with provider="ollama" so
+  // every local call is observable + auditable. Sidecar transport is
+  // unchanged — gateway just owns the call.
  assertContextBudget(model, prompt, {
    system: opts.system,
    max_tokens: opts.max_tokens,
    bypass: opts.bypass_budget,
  });
+  const messages: Array<{ role: string; content: string }> = [];
+  if (opts.system) messages.push({ role: "system", content: opts.system });
+  messages.push({ role: "user", content: prompt });
  const body: Record<string, any> = {
    model,
-    prompt,
+    messages,
+    provider: "ollama",
    temperature: opts.temperature ?? 0.3,
    max_tokens: opts.max_tokens ?? 800,
  };
-  if (opts.system) body.system = opts.system;
  if (opts.think !== undefined) body.think = opts.think;
-  const r = await http<any>("POST", `${SIDECAR}/generate`, body);
-  const text = typeof r.text === "string" ? r.text : "";
+  const r = await http<any>("POST", `${GATEWAY}/v1/chat`, body);
+  const text = r?.choices?.[0]?.message?.content ?? "";
  // Do NOT throw on empty. Thinking models (gpt-oss, qwen3.5) burn the
  // max_tokens budget on hidden reasoning and emit "" when budget was
  // too tight. generateContinuable detects empty + continues with more
  // budget. Callers that expected non-empty can check themselves.
-  return text;
+  return typeof text === "string" ? text : "";
 }

 // Cloud generate — routes through the lakehouse gateway's /v1/chat
@@ -1143,9 +1143,15 @@ Format each as a code-fenced block with the byte offset within the shard:
 EXACT LINE OF SOURCE — DO NOT PARAPHRASE, DO NOT TRUNCATE
 \`\`\`
 Pick the most reviewer-relevant lines: route definitions (e.g. \`@app.route(...)\`), function signatures, security-sensitive calls (auth/SQL/exec/template/secrets), hardcoded credentials/defaults, exception handlers, sensitive imports. The reviewer will REFUSE to act on any claim not backed by a verbatim anchor — so anchors are how you prove findings are real.`;
+    // 2026-04-28: gpt-oss:120b → gemini-3-flash-preview via Ollama
+    // Pro. Tree-split MAP fires once per shard (potentially 5-20×
+    // per file), so latency dominates total scrum time. Gemini 3
+    // flash returns shard digests substantially faster than the old
+    // 120B free model while staying strong enough for byte-anchored
+    // extraction.
    const r = await chat({
      provider: "ollama_cloud",
-      model: "gpt-oss:120b",
+      model: "gemini-3-flash-preview",
      prompt,
      max_tokens: 900,
    });
@@ -1195,9 +1201,14 @@ COPY EVERY anchor block from the piece notes IN ORDER, character-perfect. DO NOT

 Output the anchor blocks under their original \`\`\`@offset...\`\`\` fences, each on its own with a blank line between. The reviewer rejects findings that don't quote a string from this anchors block, so completeness here directly determines review quality.`;

+  // 2026-04-28: gpt-oss:120b → gemini-3-flash-preview via Ollama
+  // Pro. The reducer runs once per file (vs once per shard for MAP)
+  // but on a much larger context (all shard digests stacked), so
+  // throughput per token still matters. Same model as MAP for
+  // consistency in tree-split outputs.
  const reduced = await chat({
    provider: "ollama_cloud",
-    model: "gpt-oss:120b",
+    model: "gemini-3-flash-preview",
    prompt: reducePrompt,
    max_tokens: 2400,
  });