auditor: kb_query surfaces scrum-master reviews for files in PR diff

Wires the cohesion-plan Phase C link: the scrum-master pipeline writes
per-file reviews to data/_kb/scrum_reviews.jsonl on accept; the
auditor now reads that same file and emits one kb_query finding per
scrum review whose `file` matches a path in the PR's diff.

Severity heuristic: attempt 1-3 → info, attempt 4+ → warn. Reaching
the cloud specialist (attempt 4+) means the ladder had to escalate,
which is meaningful signal reviewers should see. Tree-split fired is
also surfaced in the finding summary.

audit.ts now passes pr.files.map(f => f.path) into runKbCheck (the
old signature dropped it on the floor). Also adds auditor/audit_one.ts
— a dry-run CLI for auditing a single PR without posting to Gitea,
useful for verifying check behavior without spamming review comments.

Verified: after writing scrum_reviews for auditor/audit.ts and
mcp-server/observer.ts (both in PR #7), audit_one 7 surfaced both as
info findings with preview + accepted_model + tree_split flag. A
scrum review for playbook_memory.rs (NOT in PR #7) was correctly
filtered out.

auditor/audit.ts

@@ -52,7 +52,7 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
     runStaticCheck(diff),
     opts.skip_dynamic ? Promise.resolve(stubFinding("dynamic", "skipped by options")) : runDynamicCheck(),
     opts.skip_inference ? Promise.resolve(stubFinding("inference", "skipped by options")) : runInferenceCheck(claims, diff),
-    runKbCheck(claims),
+    runKbCheck(claims, pr.files.map(f => f.path)),
   ]);
 
   const allFindings: Finding[] = [

auditor/audit_one.ts

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

auditor/checks/kb_query.ts

@@ -8,6 +8,7 @@
 // What this check reads (all file-backed, append-only or periodic):
 //   data/_kb/outcomes.jsonl         — per-scenario outcomes (kb.ts)
 //   data/_kb/error_corrections.jsonl — fail→succeed deltas on same sig
+//   data/_kb/scrum_reviews.jsonl     — scrum-master accepted reviews
 //   data/_observer/ops.jsonl         — observer ring → disk stream
 //   data/_bot/cycles/*.json          — bot cycle results
 //
@@ -21,10 +22,11 @@ import type { Claim, Finding } from "../types.ts";
 const KB_DIR = "/home/profit/lakehouse/data/_kb";
 const OBSERVER_OPS = "/home/profit/lakehouse/data/_observer/ops.jsonl";
 const BOT_CYCLES_DIR = "/home/profit/lakehouse/data/_bot/cycles";
+const SCRUM_REVIEWS_JSONL = "/home/profit/lakehouse/data/_kb/scrum_reviews.jsonl";
 const TAIL_LINES = 500;
 const MAX_BOT_CYCLE_FILES = 30;
 
-export async function runKbCheck(claims: Claim[]): Promise<Finding[]> {
+export async function runKbCheck(claims: Claim[], prFiles: string[] = []): Promise<Finding[]> {
   const findings: Finding[] = [];
 
   // 1. Recent scenario outcomes: are strong-claim-style phrases showing
@@ -48,6 +50,15 @@ export async function runKbCheck(claims: Claim[]): Promise<Finding[]> {
   const obsFindings = await checkObserverStream();
   findings.push(...obsFindings);
 
+  // 5. Scrum-master reviews — surface prior accepted reviews for any
+  //    file in this PR's diff. Cohesion plan Phase C wire: the
+  //    auditor gets to "borrow" the scrum-master's deeper per-file
+  //    analysis instead of re-doing that work.
+  if (prFiles.length > 0) {
+    const scrumFindings = await checkScrumReviews(prFiles);
+    findings.push(...scrumFindings);
+  }
+
   return findings;
 }
 
@@ -181,3 +192,54 @@ function observerBySource(ops: any[]): string {
   }
   return Object.entries(c).sort((a, b) => b[1] - a[1]).map(([k, v]) => `${k}=${v}`).join(", ") || "empty";
 }
+
+// Scrum-master reviews — the scrum pipeline writes one row per
+// accepted per-file review. We match reviews whose `file` matches
+// any path in the PR's diff, then surface the *preview* + which
+// model the escalation ladder had to reach. If the scrum-master
+// needed the 123B specialist or larger to resolve a file, that's
+// a meaningful signal about the code's complexity — and it's
+// surfaced to the PR without the auditor having to re-run the
+// escalation ladder itself.
+async function checkScrumReviews(prFiles: string[]): Promise<Finding[]> {
+  const rows = await tailJsonl<any>(SCRUM_REVIEWS_JSONL, TAIL_LINES);
+  if (rows.length === 0) return [];
+
+  // Match by exact file OR filename suffix — PR files arrive as
+  // `auditor/audit.ts`-style relative paths; scrum stores the same.
+  const norm = (p: string) => p.replace(/^\/+/, "").replace(/^home\/profit\/lakehouse\//, "");
+  const prSet = new Set(prFiles.map(norm));
+
+  // Keep only the most recent review per file (last-wins).
+  const latestByFile = new Map<string, any>();
+  for (const r of rows) {
+    const f = norm(String(r.file ?? ""));
+    if (!f) continue;
+    if (!prSet.has(f)) continue;
+    latestByFile.set(f, r);
+  }
+  if (latestByFile.size === 0) return [];
+
+  const findings: Finding[] = [];
+  for (const [file, r] of latestByFile) {
+    const model = String(r.accepted_model ?? "?");
+    const attempt = r.accepted_on_attempt ?? "?";
+    const treeSplit = !!r.tree_split_fired;
+    // Heuristic: if the scrum-master had to escalate past attempt 3,
+    // or had to tree-split, that's context the PR reviewer should see.
+    // Severity: info for low-escalation, warn if escalated far up
+    // the ladder (cloud specialist required).
+    const heavyEscalation = Number(attempt) >= 4;
+    const sev: "warn" | "info" = heavyEscalation ? "warn" : "info";
+    findings.push({
+      check: "kb_query",
+      severity: sev,
+      summary: `scrum-master review for \`${file}\` — accepted on attempt ${attempt} by \`${model}\`${treeSplit ? " (tree-split)" : ""}`,
+      evidence: [
+        `reviewed_at: ${r.reviewed_at ?? "?"}`,
+        `preview: ${String(r.suggestions_preview ?? "").slice(0, 300).replace(/\n/g, " ")}`,
+      ],
+    });
+  }
+  return findings;
+}

crates/vectord/Cargo.toml

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

crates/vectord/src/doc_drift.rs

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

crates/vectord/src/lib.rs

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

crates/vectord/src/playbook_memory.rs

@@ -252,11 +252,21 @@ pub struct BoostEntry {
 /// Phase 26 — what happened during an upsert. The seed endpoint
 /// returns this shape so the caller sees whether its write was a new
 /// entry, a merge, or a dedup'd no-op.
+///
+/// SHAPE NOTE: `#[serde(tag = "mode")]` requires struct-like variants —
+/// bare `Added(String)` and `Noop(String)` newtype variants would
+/// panic serialization at runtime. That bug silently 500-ed every
+/// /seed call from Phase 26 (commit 640db8c) until 2026-04-22 when the
+/// auditor's hybrid fixture surfaced it. All three variants are now
+/// struct-like, producing uniform JSON:
+///   {"mode":"added","playbook_id":"pb-..."}
+///   {"mode":"updated","playbook_id":"pb-...","merged_names":[...]}
+///   {"mode":"noop","playbook_id":"pb-..."}
 #[derive(Debug, Clone, Serialize)]
 #[serde(tag = "mode", rename_all = "lowercase")]
 pub enum UpsertOutcome {
     /// New playbook appended. Carries the new playbook_id.
-    Added(String),
+    Added { playbook_id: String },
     /// Existing same-day entry updated. Playbook_id unchanged; names
     /// merged (union, original order preserved, new names appended).
     Updated {
@@ -265,7 +275,7 @@ pub enum UpsertOutcome {
     },
     /// Identical same-day entry already exists; nothing changed.
     /// Returns the stable playbook_id so caller still has a reference.
-    Noop(String),
+    Noop { playbook_id: String },
 }
 
 /// Phase 27 — shape returned from `revise_entry`. Reports both ends of
@@ -388,6 +398,64 @@ impl PlaybookMemory {
         Ok(touched)
     }
 
+    /// Phase 45 slice 3 — stamp `doc_drift_flagged_at` on a playbook.
+    /// Idempotent: if already flagged and not yet reviewed, this is a
+    /// no-op. Callers should check the return value: Ok(true) means we
+    /// made a new flag, Ok(false) means already flagged (unreviewed) or
+    /// playbook not found, Err means persist failed.
+    pub async fn flag_doc_drift(&self, playbook_id: &str) -> Result<bool, String> {
+        let mut touched = false;
+        {
+            let mut state = self.state.write().await;
+            for e in state.entries.iter_mut() {
+                if e.playbook_id == playbook_id {
+                    // Skip if already flagged + unreviewed (avoid
+                    // churning the timestamp on every re-check).
+                    if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                        break;
+                    }
+                    e.doc_drift_flagged_at = Some(chrono::Utc::now().to_rfc3339());
+                    e.doc_drift_reviewed_at = None; // fresh flag clears any stale review
+                    touched = true;
+                    break;
+                }
+            }
+        }
+        if touched {
+            self.persist().await?;
+        }
+        Ok(touched)
+    }
+
+    /// Phase 45 slice 3 — stamp `doc_drift_reviewed_at` on a playbook,
+    /// re-admitting it to boost calculation. Idempotent.
+    pub async fn resolve_doc_drift(&self, playbook_id: &str) -> Result<bool, String> {
+        let mut touched = false;
+        {
+            let mut state = self.state.write().await;
+            for e in state.entries.iter_mut() {
+                if e.playbook_id == playbook_id && e.doc_drift_flagged_at.is_some()
+                    && e.doc_drift_reviewed_at.is_none()
+                {
+                    e.doc_drift_reviewed_at = Some(chrono::Utc::now().to_rfc3339());
+                    touched = true;
+                    break;
+                }
+            }
+        }
+        if touched {
+            self.persist().await?;
+        }
+        Ok(touched)
+    }
+
+    /// Read-only: get a playbook entry by id. Used by the drift-check
+    /// handler to read doc_refs without exposing the full state lock.
+    pub async fn get_entry(&self, playbook_id: &str) -> Option<PlaybookEntry> {
+        let state = self.state.read().await;
+        state.entries.iter().find(|e| e.playbook_id == playbook_id).cloned()
+    }
+
     /// Phase 25 — retire every entry matching (city, state) whose
     /// schema_fingerprint doesn't match the current one. Entries with
     /// no fingerprint (legacy) are skipped — caller can use
@@ -597,7 +665,7 @@ impl PlaybookMemory {
                 drop(state);
                 self.persist().await?;
                 self.rebuild_geo_index().await;
-                Ok(UpsertOutcome::Added(pid))
+                Ok(UpsertOutcome::Added { playbook_id: pid })
             }
             Some(i) => {
                 let mut existing_names_sorted = state.entries[i].endorsed_names.clone();
@@ -605,7 +673,7 @@ impl PlaybookMemory {
                 if existing_names_sorted == new_names_sorted {
                     // NOOP — identical data, just report the existing id
                     let pid = state.entries[i].playbook_id.clone();
-                    Ok(UpsertOutcome::Noop(pid))
+                    Ok(UpsertOutcome::Noop { playbook_id: pid })
                 } else {
                     // UPDATE — merge names (union, stable order).
                     let existing = state.entries.get_mut(i).ok_or("index invalidated")?;
@@ -615,15 +683,45 @@ impl PlaybookMemory {
                     }
                     existing.endorsed_names = merged.clone();
                     existing.timestamp = new_entry.timestamp.clone();
-                    // Keep original playbook_id + embedding + schema fingerprint.
-                    // Refresh embedding only if the caller passed a non-None
-                    // one (indicates the text shape changed).
+                    // Keep original playbook_id. Refresh embedding only
+                    // if the caller passed a non-None one (indicates
+                    // the text shape changed).
                     if new_entry.embedding.is_some() {
                         existing.embedding = new_entry.embedding.clone();
                     }
+                    // Lifecycle / drift-signal fields — refresh only
+                    // when the caller provides a new value. Each was
+                    // previously either ignored entirely (valid_until,
+                    // doc_refs) or only partially handled. Caught by
+                    // the auditor's hybrid fixture on 2026-04-22 when
+                    // re-seeding with fresh doc_refs silently dropped
+                    // them on the UPDATE path.
                     if new_entry.schema_fingerprint.is_some() {
                         existing.schema_fingerprint = new_entry.schema_fingerprint.clone();
                     }
+                    if new_entry.valid_until.is_some() {
+                        existing.valid_until = new_entry.valid_until.clone();
+                    }
+                    if !new_entry.doc_refs.is_empty() {
+                        // Merge by tool (case-insensitive). A new ref
+                        // for the same tool supersedes the old one —
+                        // caller is asserting "this is what I used
+                        // now." Refs for tools the existing entry
+                        // didn't track are appended. This keeps the
+                        // cumulative tool history without letting a
+                        // single re-seed erase it.
+                        let mut merged_refs = existing.doc_refs.clone();
+                        for new_ref in &new_entry.doc_refs {
+                            let hit = merged_refs.iter().position(|r| {
+                                r.tool.eq_ignore_ascii_case(&new_ref.tool)
+                            });
+                            match hit {
+                                Some(idx) => merged_refs[idx] = new_ref.clone(),
+                                None => merged_refs.push(new_ref.clone()),
+                            }
+                        }
+                        existing.doc_refs = merged_refs;
+                    }
                     let pid = existing.playbook_id.clone();
                     drop(state);
                     self.persist().await?;
@@ -745,6 +843,13 @@ impl PlaybookMemory {
             .filter(|e| {
                 if e.retired_at.is_some() { return false; }
                 if e.superseded_at.is_some() { return false; }
+                // Phase 45 slice 3 — entries flagged for doc drift and
+                // not yet human-reviewed are excluded from boost. The
+                // flag is the same strength of exclusion as retirement
+                // for this purpose; reviewing (via /resolve) re-admits.
+                if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                    return false;
+                }
                 if let Some(vu) = &e.valid_until {
                     if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(vu) {
                         if now > parsed.with_timezone(&chrono::Utc) { return false; }
@@ -775,6 +880,13 @@ impl PlaybookMemory {
                 .filter(|e| {
                     if e.retired_at.is_some() { return false; }
                     if e.superseded_at.is_some() { return false; }
+                    // Phase 45 slice 3 — same drift exclusion as the
+                    // non-geo path above. Keeps the two filter paths
+                    // consistent so a flagged entry is invisible to
+                    // BOTH geo-targeted and global boost queries.
+                    if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                        return false;
+                    }
                     if let Some(vu) = &e.valid_until {
                         if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(vu) {
                             if now > parsed.with_timezone(&chrono::Utc) { return false; }
@@ -1600,7 +1712,7 @@ mod upsert_tests {
         let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
         let e = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Alice Smith"]);
         match pm.upsert_entry(e).await.unwrap() {
-            UpsertOutcome::Added(_) => {}
+            UpsertOutcome::Added { .. } => {}
             other => panic!("expected Added, got {:?}", other),
         }
         assert_eq!(pm.entry_count().await, 1);
@@ -1613,7 +1725,7 @@ mod upsert_tests {
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Alice Smith", "Bob Jones"]);
         pm.upsert_entry(e1).await.unwrap();
         let outcome = pm.upsert_entry(e2).await.unwrap();
-        assert!(matches!(outcome, UpsertOutcome::Noop(_)));
+        assert!(matches!(outcome, UpsertOutcome::Noop { .. }));
         // Still exactly one entry, no duplicate from the re-seed.
         assert_eq!(pm.entry_count().await, 1);
     }
@@ -1625,7 +1737,7 @@ mod upsert_tests {
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Alice Smith", "Bob Jones"]);
         let o1 = pm.upsert_entry(e1).await.unwrap();
         let pid = match o1 {
-            UpsertOutcome::Added(p) => p,
+            UpsertOutcome::Added { playbook_id: p } => p,
             other => panic!("expected Added, got {:?}", other),
         };
         let o2 = pm.upsert_entry(e2).await.unwrap();
@@ -1646,7 +1758,7 @@ mod upsert_tests {
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-22", &["Alice Smith"]);
         pm.upsert_entry(e1).await.unwrap();
         let o2 = pm.upsert_entry(e2).await.unwrap();
-        assert!(matches!(o2, UpsertOutcome::Added(_)), "different day → fresh ADD");
+        assert!(matches!(o2, UpsertOutcome::Added { .. }), "different day → fresh ADD");
         assert_eq!(pm.entry_count().await, 2);
     }
 
@@ -1659,9 +1771,200 @@ mod upsert_tests {
         // A new seed on same day should ADD, not merge into the retired one.
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Carol Davis"]);
         let o = pm.upsert_entry(e2).await.unwrap();
-        assert!(matches!(o, UpsertOutcome::Added(_)));
+        assert!(matches!(o, UpsertOutcome::Added { .. }));
         assert_eq!(pm.entry_count().await, 2);
     }
+
+    // Regression test for the auditor-discovered bug: UPDATE branch
+    // silently dropped doc_refs, valid_until, and (partially)
+    // schema_fingerprint from the incoming entry.
+    //
+    // Before the fix: a re-seed with different endorsed_names AND new
+    // doc_refs would merge names but discard the new doc_refs. After:
+    // both sets are unioned (doc_refs merged by tool, same-tool
+    // supersedes).
+
+    fn docref(tool: &str, version: &str) -> DocRef {
+        DocRef {
+            tool: tool.into(),
+            version_seen: version.into(),
+            snippet_hash: Some(format!("hash-{tool}-{version}")),
+            source_url: None,
+            seen_at: "2026-04-22T00:00:00Z".into(),
+        }
+    }
+
+    fn mk_with_docs(op: &str, day: &str, names: &[&str], docs: Vec<DocRef>) -> PlaybookEntry {
+        let mut e = mk(op, day, names);
+        e.doc_refs = docs;
+        e
+    }
+
+    #[tokio::test]
+    async fn update_merges_doc_refs_with_existing_ones() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        // First seed: Alice + Docker doc ref
+        let e1 = mk_with_docs(op, day, &["Alice"], vec![docref("docker", "24.0.7")]);
+        let o1 = pm.upsert_entry(e1).await.unwrap();
+        assert!(matches!(o1, UpsertOutcome::Added { .. }));
+
+        // Second seed on same (op, day) but new names AND new tool ref
+        let e2 = mk_with_docs(op, day, &["Bob"], vec![docref("terraform", "1.5.0")]);
+        let o2 = pm.upsert_entry(e2).await.unwrap();
+        assert!(matches!(o2, UpsertOutcome::Updated { .. }));
+        assert_eq!(pm.entry_count().await, 1); // still one entry, merged
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap.len(), 1);
+        let merged = &snap[0];
+        // Names unioned
+        assert!(merged.endorsed_names.contains(&"Alice".to_string()));
+        assert!(merged.endorsed_names.contains(&"Bob".to_string()));
+        // Both doc_refs preserved (different tools → union)
+        assert_eq!(merged.doc_refs.len(), 2);
+        let tools: Vec<_> = merged.doc_refs.iter().map(|r| r.tool.as_str()).collect();
+        assert!(tools.contains(&"docker"));
+        assert!(tools.contains(&"terraform"));
+    }
+
+    #[tokio::test]
+    async fn update_same_tool_supersedes_older_version() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        let e1 = mk_with_docs(op, day, &["Alice"], vec![docref("docker", "24.0.7")]);
+        pm.upsert_entry(e1).await.unwrap();
+
+        // Same day, different name, SAME tool newer version
+        let e2 = mk_with_docs(op, day, &["Bob"], vec![docref("docker", "25.0.1")]);
+        let o2 = pm.upsert_entry(e2).await.unwrap();
+        assert!(matches!(o2, UpsertOutcome::Updated { .. }));
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap[0].doc_refs.len(), 1, "same tool should supersede, not duplicate");
+        assert_eq!(snap[0].doc_refs[0].version_seen, "25.0.1");
+    }
+
+    #[tokio::test]
+    async fn update_preserves_existing_doc_refs_when_new_entry_has_none() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        let e1 = mk_with_docs(op, day, &["Alice"], vec![docref("docker", "24.0.7")]);
+        pm.upsert_entry(e1).await.unwrap();
+
+        // Second seed doesn't carry doc_refs — shouldn't wipe the existing
+        let e2 = mk(op, day, &["Bob"]); // no doc_refs
+        pm.upsert_entry(e2).await.unwrap();
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap[0].doc_refs.len(), 1, "empty new doc_refs shouldn't wipe existing");
+        assert_eq!(snap[0].doc_refs[0].tool, "docker");
+    }
+
+    // ─── Phase 45 slice 3 — doc_drift flag / resolve / boost exclusion ───
+
+    #[tokio::test]
+    async fn flag_doc_drift_stamps_timestamp_and_persists() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+        let touched = pm.flag_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(touched, "first flag should return true");
+
+        let fetched = pm.get_entry(&e.playbook_id).await.unwrap();
+        assert!(fetched.doc_drift_flagged_at.is_some(), "flag_doc_drift should stamp the timestamp");
+        assert!(fetched.doc_drift_reviewed_at.is_none(), "reviewed_at starts None on fresh flag");
+    }
+
+    #[tokio::test]
+    async fn flag_doc_drift_is_idempotent_on_already_flagged() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-01T00:00:00Z".into());
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+        let touched = pm.flag_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(!touched, "already-flagged, unreviewed entry shouldn't re-stamp");
+    }
+
+    #[tokio::test]
+    async fn resolve_doc_drift_clears_flag_admission_gate() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-21T00:00:00Z".into());
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+
+        let resolved = pm.resolve_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(resolved);
+        let fetched = pm.get_entry(&e.playbook_id).await.unwrap();
+        assert!(fetched.doc_drift_reviewed_at.is_some());
+
+        // Second resolve is a no-op
+        let again = pm.resolve_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(!again);
+    }
+
+    #[tokio::test]
+    async fn boost_excludes_flagged_unreviewed_entries() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let e_clean = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        let mut e_flagged = mk("fill: Welder x1 in Nashville, TN", "2026-04-21", &["Bob"]);
+        e_flagged.doc_drift_flagged_at = Some("2026-04-22T00:00:00Z".into());
+        e_flagged.doc_drift_reviewed_at = None;
+        pm.set_entries(vec![e_clean, e_flagged]).await.unwrap();
+
+        let boosts = pm
+            .compute_boost_for_filtered_with_role(
+                &[1.0, 0.0, 0.0], 10, 0.5,
+                Some(("Nashville", "TN")), None,
+            ).await;
+
+        let keys: Vec<_> = boosts.keys().map(|(_, _, n)| n.clone()).collect();
+        assert!(keys.contains(&"Alice".to_string()));
+        assert!(!keys.contains(&"Bob".to_string()), "flagged+unreviewed entry leaked into boost");
+    }
+
+    #[tokio::test]
+    async fn boost_re_admits_resolved_entries() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-22T00:00:00Z".into());
+        e.doc_drift_reviewed_at = Some("2026-04-22T00:01:00Z".into()); // human reviewed
+        pm.set_entries(vec![e]).await.unwrap();
+
+        let boosts = pm
+            .compute_boost_for_filtered_with_role(
+                &[1.0, 0.0, 0.0], 10, 0.5,
+                Some(("Nashville", "TN")), None,
+            ).await;
+
+        let keys: Vec<_> = boosts.keys().map(|(_, _, n)| n.clone()).collect();
+        assert!(keys.contains(&"Alice".to_string()), "resolved entry should re-enter boost pool");
+    }
+
+    #[tokio::test]
+    async fn update_refreshes_valid_until_when_caller_provides_one() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        // First seed with no valid_until
+        let e1 = mk(op, day, &["Alice"]);
+        pm.upsert_entry(e1).await.unwrap();
+
+        // Second seed with a valid_until deadline
+        let mut e2 = mk(op, day, &["Bob"]);
+        e2.valid_until = Some("2026-05-01T00:00:00Z".into());
+        pm.upsert_entry(e2).await.unwrap();
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap[0].valid_until.as_deref(), Some("2026-05-01T00:00:00Z"));
+    }
 }
 
 #[cfg(test)]

crates/vectord/src/service.rs

@@ -133,6 +133,9 @@ pub fn router(state: VectorState) -> Router {
         .route("/playbook_memory/revise", post(revise_playbook_memory))
         .route("/playbook_memory/history/{id}", get(playbook_memory_history))
         .route("/playbook_memory/status", get(playbook_memory_status))
+        // Phase 45 slice 3 — doc drift detection + human re-admission.
+        .route("/playbook_memory/doc_drift/check/{id}", post(check_doc_drift))
+        .route("/playbook_memory/doc_drift/resolve/{id}", post(resolve_doc_drift))
         .with_state(state)
 }
 
@@ -2500,6 +2503,94 @@ async fn retire_playbook_memory(
         "supply either {playbook_id, reason} or {city, state, current_schema_fingerprint, reason}".into()))
 }
 
+/// Phase 45 slice 3 — POST /playbook_memory/doc_drift/check/{id}
+///
+/// Iterates the playbook's `doc_refs`, asks the context7 bridge whether
+/// each one drifted against the recorded snippet_hash. If any tool
+/// returned `drifted: true`, stamps `doc_drift_flagged_at` on the
+/// entry — which excludes it from boost (via the filter in
+/// `compute_boost_for_filtered_with_role`) until a human reviews and
+/// resolves.
+///
+/// Unknown outcomes (bridge down, tool not in context7, no snippet
+/// hash) are explicitly NOT enough to flag. Only a positive drifted=true
+/// from the bridge flips the flag.
+async fn check_doc_drift(
+    State(state): State<VectorState>,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    use crate::doc_drift::{check_all_refs, DriftCheckerConfig, DriftOutcome};
+
+    let entry = state.playbook_memory.get_entry(&id).await
+        .ok_or((StatusCode::NOT_FOUND, format!("playbook not found: {id}")))?;
+
+    if entry.doc_refs.is_empty() {
+        return Ok(Json(serde_json::json!({
+            "playbook_id": id,
+            "checked_tools": [],
+            "drifted": false,
+            "flagged": false,
+            "reason": "entry has no doc_refs — nothing to check",
+        })));
+    }
+
+    let cfg = DriftCheckerConfig::default();
+    let results = check_all_refs(&cfg, &entry.doc_refs).await;
+
+    let per_tool: Vec<serde_json::Value> = results.iter().map(|r| {
+        let (drifted, current, src, reason) = match &r.outcome {
+            DriftOutcome::Drifted { current_snippet_hash, source_url } =>
+                (true, Some(current_snippet_hash.clone()), source_url.clone(), None),
+            DriftOutcome::Unchanged =>
+                (false, None, None, None),
+            DriftOutcome::Unknown { reason } =>
+                (false, None, None, Some(reason.clone())),
+        };
+        serde_json::json!({
+            "tool": r.tool,
+            "version_seen": r.version_seen,
+            "drifted": drifted,
+            "current_snippet_hash": current,
+            "source_url": src,
+            "unknown_reason": reason,
+        })
+    }).collect();
+
+    let any_drifted = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Drifted { .. }));
+
+    let flagged = if any_drifted {
+        state.playbook_memory.flag_doc_drift(&id).await
+            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("flag: {e}")))?
+    } else {
+        false
+    };
+
+    Ok(Json(serde_json::json!({
+        "playbook_id": id,
+        "checked_tools": results.iter().map(|r| &r.tool).collect::<Vec<_>>(),
+        "drifted": any_drifted,
+        "flagged": flagged,
+        "per_tool": per_tool,
+    })))
+}
+
+/// Phase 45 slice 3 — POST /playbook_memory/doc_drift/resolve/{id}
+///
+/// Human-in-the-loop re-admission. Stamps `doc_drift_reviewed_at`.
+/// Idempotent: returns `resolved: false` if nothing changed (entry
+/// wasn't flagged, already reviewed, or doesn't exist).
+async fn resolve_doc_drift(
+    State(state): State<VectorState>,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    let resolved = state.playbook_memory.resolve_doc_drift(&id).await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("resolve: {e}")))?;
+    Ok(Json(serde_json::json!({
+        "playbook_id": id,
+        "resolved": resolved,
+    })))
+}
+
 /// Phase 27 — request body for `POST /playbook_memory/revise`. Same
 /// shape as a seed request minus `append` (revise is always
 /// append-semantics for a specific parent) plus `parent_id`. The new

ops/systemd/README.md

@@ -0,0 +1,57 @@
+# Lakehouse systemd units
+
+Service definitions for long-running Lakehouse sidecars that aren't the
+Rust gateway itself. The gateway has its own pre-existing unit
+(`lakehouse.service`) that was configured at initial deploy time and
+isn't tracked here.
+
+## Units
+
+| File | Service | Port | Purpose |
+|---|---|---|---|
+| `lakehouse-auditor.service` | `lakehouse-auditor` | n/a | Polls Gitea for open PRs, runs four checks (static / dynamic / inference / KB query), posts commit-status + review comment. Hard-blocks merges when claims aren't backed. |
+| `lakehouse-context7-bridge.service` | `lakehouse-context7-bridge` | `:3900` | HTTP wrapper around context7's public API for Phase 45 doc-drift detection. |
+
+## Install
+
+```bash
+sudo bash ops/systemd/install.sh
+```
+
+Idempotent. Copies units to `/etc/systemd/system/`, reloads, enables + (re)starts both services.
+
+## Operate
+
+```bash
+# Status
+systemctl status lakehouse-auditor
+systemctl status lakehouse-context7-bridge
+
+# Live logs
+journalctl -u lakehouse-auditor -f
+
+# Restart
+systemctl restart lakehouse-auditor
+
+# Stop (won't restart until enable + start again)
+systemctl stop lakehouse-auditor
+```
+
+## Pause the auditor without stopping
+
+```bash
+touch /home/profit/lakehouse/auditor.paused   # skip cycles until removed
+rm    /home/profit/lakehouse/auditor.paused   # resume
+```
+
+## Env toggles on the auditor (edit the unit file, `systemctl daemon-reload`, restart)
+
+```
+LH_AUDITOR_RUN_DYNAMIC=1    # include the hybrid fixture on every audit
+                            # default off — fixture mutates live playbook state
+LH_AUDITOR_SKIP_INFERENCE=1 # skip cloud inference for fast/cheap runs
+```
+
+## Why both services run as root
+
+To match the existing `lakehouse.service` + `mcp-server` + `observer` conventions on this host. Hardening to a dedicated unprivileged user is a follow-up: would need PATH adjustment for `bun`, credential file accessibility (the auditor reads `/home/profit/.git-credentials` which is `0600 profit:profit` — root reads fine, a non-profit non-root user wouldn't).

ops/systemd/install.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+# Install the lakehouse-auditor + lakehouse-context7-bridge systemd units.
+# Idempotent: re-running just reloads + restarts.
+#
+# Usage (as root):
+#   bash ops/systemd/install.sh
+#
+# What it does:
+#   1. Copies *.service to /etc/systemd/system/
+#   2. systemctl daemon-reload
+#   3. systemctl enable --now both services
+#   4. Prints post-install status
+
+set -euo pipefail
+
+UNIT_DIR="$(dirname "$(readlink -f "$0")")"
+TARGET_DIR=/etc/systemd/system
+
+UNITS=(
+  lakehouse-auditor.service
+  lakehouse-context7-bridge.service
+)
+
+if [[ $EUID -ne 0 ]]; then
+  echo "install.sh: must run as root (writes to $TARGET_DIR)" >&2
+  exit 1
+fi
+
+for unit in "${UNITS[@]}"; do
+  src="$UNIT_DIR/$unit"
+  dst="$TARGET_DIR/$unit"
+  if [[ ! -f "$src" ]]; then
+    echo "install.sh: missing source $src" >&2
+    exit 1
+  fi
+  echo "→ copy $unit"
+  install -m 0644 "$src" "$dst"
+done
+
+echo "→ systemctl daemon-reload"
+systemctl daemon-reload
+
+for unit in "${UNITS[@]}"; do
+  echo "→ enable + (re)start $unit"
+  systemctl enable "$unit" >/dev/null
+  systemctl restart "$unit"
+done
+
+echo ""
+echo "─── post-install status ───"
+for unit in "${UNITS[@]}"; do
+  active=$(systemctl is-active "$unit" 2>/dev/null || true)
+  enabled=$(systemctl is-enabled "$unit" 2>/dev/null || true)
+  printf "  %-40s active=%s enabled=%s\n" "$unit" "$active" "$enabled"
+done
+echo ""
+echo "Live logs:   journalctl -u lakehouse-auditor.service -f"
+echo "Pause:       touch /home/profit/lakehouse/auditor.paused"
+echo "Resume:      rm /home/profit/lakehouse/auditor.paused"

ops/systemd/lakehouse-auditor.service

@@ -0,0 +1,34 @@
+[Unit]
+Description=Lakehouse Claim Auditor — polls Gitea for open PRs + hard-blocks placeholder merges
+Documentation=file:///home/profit/lakehouse/auditor/README.md
+After=network.target lakehouse.service
+Wants=lakehouse.service
+
+[Service]
+Type=simple
+WorkingDirectory=/home/profit/lakehouse
+# Runs as root to match the other lakehouse-* services on this host
+# (gateway, mcp-server, observer). The auditor reads the git PAT out
+# of /home/profit/.git-credentials which is 0600 profit:profit —
+# root can read it, which is why the service runs as root.
+# Alternative: run as `profit` and ensure bun binary is on PATH;
+# left for a follow-up hardening PR.
+ExecStart=/home/profit/.bun/bin/bun run /home/profit/lakehouse/auditor/index.ts
+Restart=on-failure
+RestartSec=30
+# Stop responds to SIGTERM cleanly — no in-flight cycle survives a
+# restart; the poller is idempotent so a mid-cycle restart just
+# re-audits from state.json on next start.
+KillSignal=SIGTERM
+TimeoutStopSec=10
+
+# Optional env toggles documented in auditor/index.ts:
+#   LH_AUDITOR_RUN_DYNAMIC=1      — include hybrid fixture on every audit
+#                                   (default off — mutates live playbook)
+#   LH_AUDITOR_SKIP_INFERENCE=1   — skip cloud inference for fast runs
+
+# Pause file — operator can `touch /home/profit/lakehouse/auditor.paused`
+# to skip the next cycle without stopping the service.
+
+[Install]
+WantedBy=multi-user.target

ops/systemd/lakehouse-context7-bridge.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=Lakehouse context7 HTTP bridge — doc-version lookups for Phase 45 drift detection
+Documentation=file:///home/profit/lakehouse/mcp-server/context7_bridge.ts
+After=network.target
+# No hard dependency on gateway — the bridge is self-contained and
+# talks to context7.com directly. Gateway calls INTO the bridge
+# during drift checks but they're decoupled.
+
+[Service]
+Type=simple
+WorkingDirectory=/home/profit/lakehouse
+ExecStart=/home/profit/.bun/bin/bun run /home/profit/lakehouse/mcp-server/context7_bridge.ts
+Restart=on-failure
+RestartSec=15
+# Default port :3900 (overridable via CONTEXT7_BRIDGE_PORT env).
+# If the user moves the bridge to a different port, update the
+# LH_BRIDGE_URL env on lakehouse.service + auditor.service too.
+Environment=CONTEXT7_BRIDGE_PORT=3900
+KillSignal=SIGTERM
+TimeoutStopSec=5
+
+[Install]
+WantedBy=multi-user.target

tests/real-world/enrich_prd_pipeline.ts

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

tests/real-world/hard_task_escalation.ts

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

tests/real-world/scrum_master_pipeline.ts

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