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
+// 40KB comfortably fits the consensus models' context windows
-// previously truncated at 15KB causing the reviewer to miss later
+// (deepseek-v3.1 64K, gpt-oss-120b 128K). When the raw PR diff
-// files (gitea.ts, policy.ts) and flag "no Gitea client present" as a
+// exceeds this, we truncate and signal it via curationNote — the
-// block finding when the file was simply outside the truncation window.
+// pr_audit mode runner's matrix retrieval (lakehouse_answers_v1 +
-//
+// arch + symbols) supplies the cross-PR context that tree-split
-// Above this threshold we curate via tree-split rather than truncate,
+// used to synthesize from scratch. Tree-split itself was retired
-// following the scrum_master pattern: shard the diff, summarize each
+// 2026-04-27 (see commit deleting treeSplitDiff/callCloud/SHARD_*).
 // shard against the claim-verification task, merge into a compact
 // scratchpad, then ask the cloud to verify claims against the
 // scratchpad. This gives the cloud full-PR fidelity without bursting
 // its context window (observed failure mode: empty response or
 // unparseable output when prompt exceeds model's comfortable range).
 const MAX_DIFF_CHARS = 40000;
 // Tree-split kicks in above this. 30KB is below MAX_DIFF_CHARS so we
 // curate BEFORE truncation would happen — never lose signal to a hard
 // cut. Shard size is chosen so ~10 shards cover PR #8-size diffs in a
 // reasonable round-trip budget.
 const CURATION_THRESHOLD = 30000;
 const DIFF_SHARD_SIZE = 4500;
 const CALL_TIMEOUT_MS = 120_000;
 // 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,
+    latency_ms: num(body?.latency_ms),
-    enriched_chars: body?.enriched_prompt_chars ?? 0,
+    enriched_chars: num(body?.enriched_prompt_chars),
-    bug_fingerprints: body?.sources?.bug_fingerprints_count ?? 0,
+    bug_fingerprints: num(body?.sources?.bug_fingerprints_count),
-    matrix_kept: body?.sources?.matrix_chunks_kept ?? 0,
+    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 { readFile, writeFile, mkdir, appendFile, stat, realpath } from "node:fs/promises";
-import { existsSync } from "node:fs";
+import { existsSync, realpathSync } from "node:fs";
-import { join, resolve } from "node:path";
+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
+// Cross-lineage alternation. 2026-04-27 J's call: Opus is too
-// to OPUS_MODEL for that audit. 2026-04-27 3-way bake-off (Kimi vs
+// expensive to auto-fire (~$0.30/audit). Kimi K2.6 via Go-sub is
-// Haiku vs Opus on a 32K diff) showed Opus is the only model that
+// effectively free; Haiku 4.5 via Zen is ~$0.04. Alternate between
-// catches cross-file ramifications + escalates `block` severity on
+// them so we get cross-lineage signal (Moonshot vs Anthropic) on
-// real architectural risks. ~5x the spend per audit, only worth it
+// every PR's audit history without burning the budget.
 // when the diff is big enough to have those risks.
 //
-// Defaults: Haiku for normal diffs (fast, cheap, ~$0.02), Opus for
+// Default: Kimi K2.6 on even audits, Haiku 4.5 on odd. Each PR's
-// > 100k chars. Disable promotion: set OPUS_THRESHOLD_CHARS very high.
+// audits flip between vendors as new SHAs come in.
-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";
+// Frontier models (Opus 4.7, GPT-5.5, Gemini 3.1) are NOT in the
-const OPUS_THRESHOLD_CHARS = Number(process.env.LH_AUDITOR_KIMI_OPUS_THRESHOLD_CHARS) || 100_000;
+// 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 } {
+function selectModel(diffLen: number, auditIndex: number = 0): { provider: string; model: string; promoted: boolean } {
-  if (diffLen > OPUS_THRESHOLD_CHARS) {
+  // Operator override — env-pinned model wins.
-    return { provider: OPUS_PROVIDER, model: OPUS_MODEL, promoted: true };
+  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
+// Model-aware max_tokens. Different upstream APIs cap at different
-// budget but isn't surfaced in the OpenAI-shape `content` field.
+// limits and reject requests that exceed them:
-// Capping low silently produces empty content with finish_reason=length
+//   - Anthropic Opus 4.x: 32K output (with extended-output header)
-// when reasoning consumes the budget. 128K leaves ample room for both
+//   - Anthropic Haiku 4.5: 8K output
-// reasoning and visible findings on any audit prompt we throw at it.
+//   - Kimi K2.6 (reasoning): 128K — needs headroom because
-// Override via LH_AUDITOR_KIMI_MAX_TOKENS only if you want to cap cost.
+//     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
+// 2026-04-27 BLOCK from Opus self-audit: the prior single-default of
-// like LH_AUDITOR_KIMI_MAX_TOKENS="" used to parse via Number("") → 0
+// 128K worked silently (Anthropic clamps server-side) but was
-// because `??` only catches null/undefined. Use `||` so empty string,
+// technically invalid. Per-model caps make it explicit. Override via
-// 0, or NaN all fall back to the default.
+// LH_AUDITOR_KIMI_MAX_TOKENS to force a value (also fixes the empty-
-const MAX_TOKENS = Number(process.env.LH_AUDITOR_KIMI_MAX_TOKENS) || 128_000;
+// 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,16 +213,24 @@ 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);
-
+  if (findings.length > 0) {
-  return findings.length > 0
+    await persistVerdict(cachePath, verdict);
-    ? findings
+    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)`,
+    summary: `kimi_architect produced 0 ranked findings (${response.finish_reason}, ${verdict.usage.completion_tokens} tokens) — not cached`,
-        evidence: [`raw response: ${cachePath}`],
+    evidence: [`raw saved (no cache): see kimi_audits.jsonl ${verdict.cached_at}`],
  }];
 }
@@ -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 —
+              // Skip if EITHER (a) the file was already inside a
-              // this is how rubric files (tests/real-world/*, prompt
+              // multi-line backtick block when this line started, OR
-              // templates) legitimately reference the patterns they
+              // (b) the match sits inside a quoted string literal on
-              // guard against, without actually executing them.
+              // THIS line. The earlier code only checked stateAtLineStart;
-              if (inMultilineBacktick || isInsideQuotedString(added, m.index)) continue;
+              // 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
+      // Carry the end-of-line state forward to the next iteration.
-      // line's unescaped backticks. Both context and added lines
+      inMultilineBacktick = stateAtLineEnd;
      // 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);
    }
    // "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,
+      provider: "ollama_cloud",
-      prompt: userPrompt,
+      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"
+default_model = "deepseek-v3.2"
-# Cloud-tier Ollama. Key resolved from OLLAMA_CLOUD_KEY env at gateway
+# Cloud-tier Ollama (Pro plan as of 2026-04-28). Key resolved from
-# boot. Model-prefix routing: "cloud/<model>" auto-routes here
+# OLLAMA_CLOUD_KEY at gateway boot; Pro tier upgraded the account so
-# (see gateway::v1::resolve_provider).
+# 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);
+        // u32::try_from saturates at u32::MAX instead of silently
-        completion_tokens = completion_tokens.saturating_add(resp.tokens_generated.unwrap_or(0) as u32);
+        // 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);
+        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(resp.tokens_generated.unwrap_or(0) as u32);
+        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 cloud overseer (`claude-opus-4-7` via OpenCode Zen) with
-    /// the KB context — recent outcomes + prior corrections for this
+    /// (a) the KB context — recent outcomes + prior corrections for
-    /// sig_hash + task_class, across every profile that has run it —
+    /// this sig_hash + task_class, across every profile that has run
-    /// and (b) the recent log tail. Its output is appended as a
+    /// 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 {
+        let Some(opencode_key) = self.state.opencode_key.clone() else {
-            return Err("OLLAMA_CLOUD_KEY not configured — skipping escalation".into());
+            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)
+            // Anthropic models on opencode reject `think` (handled in
-            provider: Some("ollama_cloud".into()),
+            // 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
+        let resp = crate::v1::opencode::chat(&opencode_key, &chat_req).await
-            .map_err(|e| format!("ollama_cloud: {e}"))?;
+            .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(),
+                provider: "opencode".into(),
-                model: "gpt-oss:120b".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
+    // AI sidecar clients — Phase 44 part 3 (2026-04-27).
-    let ai_client = aibridge::client::AiClient::new(&config.sidecar.url);
+    //
    // 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();
+    const analysis = await resp.json() as any;
-    if (analysis.text) {
+    const analysisText = analysis?.choices?.[0]?.message?.content ?? "";
-      console.error(`[observer] Error analysis:\n${analysis.text}`);
+    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
+  // Phase 44 migration (2026-04-27): /v1/chat instead of direct sidecar
-  // visible response. qwen3.5:latest defaults to thinking and silently
+  // /generate so /v1/usage tracks the call, Langfuse traces it.
-  // burns the token budget otherwise.
+  // think:false still disables hidden reasoning so generated tokens
-  const r = await fetch(`${SIDECAR}/generate`, {
+  // 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",
+      provider: "ollama",
-      stream: false,
+      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 r = await http<any>("POST", `${GATEWAY}/v1/chat`, body);
-  const text = typeof r.text === "string" ? r.text : "";
+  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,
  });