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>

Cargo.lock

@@ -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",

auditor/checks/inference.ts

@@ -426,14 +426,22 @@ 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,

auditor/checks/kimi_architect.ts

@@ -26,8 +26,8 @@
 //
 // 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 { dirname, join, resolve } from "node:path";
 import type { Finding, CheckKind } from "../types.ts";
 
@@ -35,6 +35,14 @@ 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
@@ -364,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;

auditor/checks/static.ts

@@ -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];

auditor/index.ts

@@ -115,13 +115,28 @@ async function runCycle(state: State): Promise<State> {
       console.log(`[auditor]   skip PR #${pr.number} (SHA ${pr.head_sha.slice(0, 8)} already audited)`);
       continue;
     }
-    // Per-PR audit cap — once a PR has been audited MAX_AUDITS_PER_PR
+    // Per-head-SHA audit cap. Each new push gets MAX_AUDITS_PER_PR
-    // times, halt further audits until the operator manually clears
+    // fresh attempts; the counter auto-resets when the head SHA
-    // audit_count_per_pr[<N>] in state.json. Prevents runaway burn
+    // changes. Operator only intervenes manually if a single SHA
-    // when each fix surfaces new findings.
+    // 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) {
-      console.log(`[auditor]   skip PR #${pr.number} (capped at ${auditedSoFar}/${MAX_AUDITS_PER_PR} audits — clear state.json audit_count_per_pr.${prKey} to resume)`);
+      // 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;
     }
@@ -143,6 +158,16 @@ async function runCycle(state: State): Promise<State> {
       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}`);
     }

bot/propose.ts

@@ -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);
 

config/modes.toml

@@ -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]]

config/providers.toml

@@ -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.

crates/aibridge/src/client.rs

@@ -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))

crates/gateway/Cargo.toml

@@ -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 }

crates/gateway/src/execution_loop/mod.rs

@@ -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,

crates/gateway/src/main.rs

@@ -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.

crates/gateway/src/v1/iterate.rs

@@ -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));
+    }
+}

crates/gateway/src/v1/mod.rs

@@ -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

crates/gateway/src/v1/validate.rs

@@ -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(),
+    }
+}

crates/validator/Cargo.toml

@@ -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 }

crates/validator/src/lib.rs

@@ -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()
+    }
 }

crates/validator/src/staffing/mod.rs

@@ -6,3 +6,4 @@
 pub mod fill;
 pub mod email;
 pub mod playbook;
+pub mod parquet_lookup;

crates/validator/src/staffing/parquet_lookup.rs

@@ -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 { .. })));
+    }
+}

crates/vectord/src/service.rs

@@ -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`.

data/_catalog/views/candidates_safe.json

@@ -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."
+}

data/_catalog/views/jobs_safe.json

@@ -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."
+}

data/_catalog/views/workers_safe.json

@@ -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."
+}

docs/PHASES.md

@@ -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)
 

mcp-server/observer.ts

@@ -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

scripts/check_phase44_callers.sh

@@ -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

scripts/staffing/build_fill_events.py

@@ -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())

scripts/staffing/build_workers_v9.sh

@@ -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'"

scripts/staffing/fixup_phone_type.py

@@ -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())

tests/agent_test/agent_harness.ts

@@ -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 {

tests/multi-agent/agent.ts

@@ -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

tests/real-world/scrum_master_pipeline.ts

@@ -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,
   });