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>

bot/propose.ts

@@ -16,12 +16,14 @@ import type { Gap, Proposal } from "./types.ts";
 // 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. Same upstream model (CLOUD_MODEL gpt-oss:120b on
-// Ollama Cloud) — gateway just owns the routing.
+// 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[]> {

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"
-# Cloud-tier Ollama. Key resolved from OLLAMA_CLOUD_KEY env at gateway
-# boot. Model-prefix routing: "cloud/<model>" auto-routes here
-# (see gateway::v1::resolve_provider).
+default_model = "deepseek-v3.2"
+# Cloud-tier Ollama (Pro plan as of 2026-04-28). Key resolved from
+# OLLAMA_CLOUD_KEY at gateway boot; Pro tier upgraded the account so
+# rate limits + model access widen without a key change. Model-prefix
+# routing: "cloud/<model>" auto-routes here. 39-model fleet now
+# includes deepseek-v3.2, deepseek-v4-{flash,pro}, gemini-3-flash-
+# preview, glm-{5,5.1}, kimi-k2.6, qwen3-coder-next.
+# 2026-04-28: default upgraded gpt-oss:120b → deepseek-v3.2 (newest
+# DeepSeek revision; kimi-k2:1t still upstream-broken with HTTP 500).
 
 [[provider]]
 name = "openrouter"
@@ -38,7 +43,7 @@ base_url = "https://openrouter.ai/api/v1"
 auth = "bearer"
 auth_env = "OPENROUTER_API_KEY"
 auth_fallback_files = ["/home/profit/.env", "/root/llm_team_config.json"]
-default_model = "openai/gpt-oss-120b:free"
+default_model = "x-ai/grok-4.1-fast"
 # Multi-provider gateway. Covers Anthropic, Google, OpenAI, MiniMax,
 # Qwen, Gemma, etc. Key resolved via crates/gateway/src/v1/openrouter.rs
 # resolve_openrouter_key() — env first, then fallback files.

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 KB context — recent outcomes + prior corrections for this
-    /// sig_hash + task_class, across every profile that has run it —
-    /// and (b) the recent log tail. Its output is appended as a
+    /// the cloud overseer (`claude-opus-4-7` via OpenCode Zen) with
+    /// (a) the KB context — recent outcomes + prior corrections for
+    /// this sig_hash + task_class, across every profile that has run
+    /// it — and (b) the recent log tail. Its output is appended as a
     /// `system` role turn so the next executor generation sees it,
     /// AND written to `data/_kb/overseer_corrections.jsonl` so every
     /// future profile activation reads from the same learning pool.
@@ -593,9 +593,16 @@ impl ExecutionLoop {
     /// This is the "pipe to the overviewer" piece from 2026-04-23 —
     /// the overseer is now a first-class KB consumer AND producer, not
     /// a one-shot correction oracle.
+    ///
+    /// 2026-04-28: routed through OpenCode (Zen tier) for Claude Opus
+    /// 4.7. Frontier reasoning matters here because the overseer fires
+    /// only after local self-correction has failed twice — by that
+    /// point we need the strongest reasoning available, not the
+    /// cheapest token. Frequency is low so the Zen pay-per-token cost
+    /// stays bounded.
     async fn escalate_to_overseer(&mut self, turn: u32, reason: &str) -> Result<(), String> {
-        let Some(cloud_key) = self.state.ollama_cloud_key.clone() else {
-            return Err("OLLAMA_CLOUD_KEY not configured — skipping escalation".into());
+        let Some(opencode_key) = self.state.opencode_key.clone() else {
+            return Err("OPENCODE_API_KEY not configured — skipping escalation".into());
         };
 
         let kb = KbContext::load_for(&sig_hash(&self.req), &self.req.task_class).await;
@@ -604,16 +611,18 @@ impl ExecutionLoop {
         let started = std::time::Instant::now();
         let start_time = chrono::Utc::now();
         let chat_req = crate::v1::ChatRequest {
-            model: "gpt-oss:120b".to_string(),
+            model: "claude-opus-4-7".to_string(),
             messages: vec![crate::v1::Message::new_text("user", prompt.clone())],
             temperature: Some(0.1),
             max_tokens: None,
             stream: Some(false),
-            think: Some(true),    // overseer KEEPS thinking (Phase 20 rule)
-            provider: Some("ollama_cloud".into()),
+            // Anthropic models on opencode reject `think` (handled in
+            // the adapter), but we keep the intent flag for parity.
+            think: Some(true),
+            provider: Some("opencode".into()),
         };
-        let resp = crate::v1::ollama_cloud::chat(&cloud_key, &chat_req).await
-            .map_err(|e| format!("ollama_cloud: {e}"))?;
+        let resp = crate::v1::opencode::chat(&opencode_key, &chat_req).await
+            .map_err(|e| format!("opencode: {e}"))?;
         let latency_ms = started.elapsed().as_millis() as u64;
         let end_time = chrono::Utc::now();
         let correction_text: String = resp.choices.into_iter().next()
@@ -633,8 +642,8 @@ impl ExecutionLoop {
         if let Some(lf) = &self.state.langfuse {
             use crate::v1::langfuse_trace::ChatTrace;
             lf.emit_chat(ChatTrace {
-                provider: "ollama_cloud".into(),
-                model: "gpt-oss:120b".into(),
+                provider: "opencode".into(),
+                model: "claude-opus-4-7".into(),
                 input: vec![crate::v1::Message::new_text("user", prompt.clone())],
                 output: correction_text.clone(),
                 prompt_tokens: resp.usage.prompt_tokens,
@@ -650,7 +659,7 @@ impl ExecutionLoop {
 
         // Append to the transcript so the next executor turn sees it.
         self.append(LogEntry::new(
-            turn, "system", "gpt-oss:120b", "overseer_correction",
+            turn, "system", "claude-opus-4-7", "overseer_correction",
             serde_json::json!({
                 "reason": reason,
                 "correction": correction_text,
@@ -672,7 +681,7 @@ impl ExecutionLoop {
             "task_class": self.req.task_class,
             "operation": self.req.operation,
             "reason": reason,
-            "model": "gpt-oss:120b",
+            "model": "claude-opus-4-7",
             "correction": correction_text,
             "applied_at_turn": turn,
             "kb_context_used": kb,

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

mcp-server/observer.ts

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