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.
scrum_master: tree-split + scrum_reviews.jsonl writer + truncation warning
2026-04-22 21:18:21 -05:00 · 2026-04-22 21:17:53 -05:00 · 2026-04-22 20:52:42 -05:00 · 2026-04-22 18:50:53 -05:00 · 2026-04-22 17:50:53 -05:00 · 2026-04-22 17:33:24 -05:00
11 changed files with 1372 additions and 510 deletions
@@ -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[] = [
@@ -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); });
@@ -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;
 }
@@ -1,328 +0,0 @@
 // Phase 45 slice 4 — T3 drift-correction synthesis.
 //
 // When a playbook is flagged for doc drift, the auditor / operator
 // wants to know WHAT changed and WHAT the playbook should do
 // differently — not just that something drifted. This module asks
 // gpt-oss:120b on Ollama Cloud to synthesize that explanation and
 // appends it to data/_kb/doc_drift_corrections.jsonl.
 //
 // Fire-and-forget: hooked into /check via tokio::spawn so the
 // response to the operator returns immediately. Cloud failures are
 // logged and dropped — they never block the flag itself.
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 const CLOUD_URL: &str = "https://ollama.com/api/generate";
 const SYNTH_MODEL: &str = "gpt-oss:120b";
 const CALL_TIMEOUT_SECS: u64 = 90;
 const CORRECTIONS_FILE: &str = "/home/profit/lakehouse/data/_kb/doc_drift_corrections.jsonl";
 /// One drift-correction record. Mirrors the shape declared in
 /// docs/CONTROL_PLANE_PRD.md Phase 45 spec.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DriftCorrection {
    pub playbook_id: String,
    pub operation: String,
    pub tool: String,
    pub version_seen: String,
    pub current_snippet_hash: Option<String>,
    pub diff_summary: String,
    pub recommended_action: String,
    pub generated_at: String,
    pub model: String,
 }
 /// Read the Ollama Cloud key from the same sources v1/ollama_cloud.rs
 /// uses. Returns None if not available — caller should skip synthesis.
 pub fn resolve_cloud_key() -> Option<String> {
    if let Ok(k) = std::env::var("OLLAMA_CLOUD_KEY") {
        if !k.trim().is_empty() { return Some(k.trim().to_string()); }
    }
    if let Ok(raw) = std::fs::read_to_string("/root/llm_team_config.json") {
        if let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) {
            if let Some(k) = v.pointer("/providers/ollama_cloud/api_key").and_then(|x| x.as_str()) {
                if !k.trim().is_empty() { return Some(k.trim().to_string()); }
            }
        }
    }
    if let Ok(k) = std::env::var("OLLAMA_CLOUD_API_KEY") {
        if !k.trim().is_empty() { return Some(k.trim().to_string()); }
    }
    None
 }
 /// Ask the cloud model to produce a diff_summary + recommended_action
 /// for a playbook's specific drift. Returns a complete DriftCorrection
 /// ready to append. Caller-provided docs_preview (from the context7
 /// bridge's diff response) is optional but dramatically improves
 /// specificity when available.
 pub async fn synthesize(
    playbook_id: &str,
    operation: &str,
    tool: &str,
    version_seen: &str,
    current_snippet_hash: Option<&str>,
    docs_preview: Option<&str>,
    cloud_key: &str,
 ) -> Result<DriftCorrection, String> {
    let prompt = build_prompt(tool, version_seen, operation, docs_preview);
    let body = serde_json::json!({
        "model": SYNTH_MODEL,
        "prompt": prompt,
        "stream": false,
        "options": { "temperature": 0.2, "num_predict": 600 },
    });
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(CALL_TIMEOUT_SECS))
        .build()
        .map_err(|e| format!("client build: {e}"))?;
    let resp = client
        .post(CLOUD_URL)
        .bearer_auth(cloud_key)
        .json(&body)
        .send()
        .await
        .map_err(|e| format!("cloud unreachable: {e}"))?;
    if !resp.status().is_success() {
        return Err(format!(
            "cloud {}: {}",
            resp.status(),
            resp.text().await.unwrap_or_default().chars().take(200).collect::<String>(),
        ));
    }
    let j: serde_json::Value = resp.json().await.map_err(|e| format!("parse response: {e}"))?;
    let text: &str = j.get("response").and_then(|v| v.as_str()).unwrap_or("");
    let parsed = extract_json_object(text).ok_or_else(|| {
        format!("cloud output not valid JSON — head: {}", text.chars().take(200).collect::<String>())
    })?;
    let diff_summary = parsed
        .get("diff_summary").and_then(|v| v.as_str())
        .unwrap_or("").trim().to_string();
    let recommended_action = parsed
        .get("recommended_action").and_then(|v| v.as_str())
        .unwrap_or("").trim().to_string();
    if diff_summary.is_empty() && recommended_action.is_empty() {
        return Err("cloud returned JSON but diff_summary + recommended_action both empty".into());
    }
    Ok(DriftCorrection {
        playbook_id: playbook_id.to_string(),
        operation: operation.to_string(),
        tool: tool.to_string(),
        version_seen: version_seen.to_string(),
        current_snippet_hash: current_snippet_hash.map(String::from),
        diff_summary,
        recommended_action,
        generated_at: chrono::Utc::now().to_rfc3339(),
        model: SYNTH_MODEL.to_string(),
    })
 }
 /// Append one correction as a JSONL line. Creates the parent dir and
 /// file if missing. Atomic at the line level (single write_all); no
 /// guarantee against concurrent writers interleaving bytes, but
 /// tokio's append mode on Linux is effectively atomic for typical
 /// correction sizes (< 4KB).
 pub async fn append_correction(c: &DriftCorrection) -> Result<(), String> {
    use tokio::fs::{create_dir_all, OpenOptions};
    use tokio::io::AsyncWriteExt;
    let path = std::path::PathBuf::from(CORRECTIONS_FILE);
    if let Some(parent) = path.parent() {
        create_dir_all(parent).await.map_err(|e| format!("mkdir _kb: {e}"))?;
    }
    let mut f = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
        .await
        .map_err(|e| format!("open corrections: {e}"))?;
    let mut line = serde_json::to_string(c).map_err(|e| format!("serialize: {e}"))?;
    line.push('\n');
    f.write_all(line.as_bytes()).await.map_err(|e| format!("write corrections: {e}"))?;
    Ok(())
 }
 /// Spawn a fire-and-forget synthesis + append task. Never blocks the
 /// caller. On failure, logs warn. On success, appends the correction.
 /// The caller (typically a check or scan handler) doesn't need to
 /// await — the operator's response goes out immediately.
 pub fn spawn_synthesize_and_append(
    playbook_id: String,
    operation: String,
    tool: String,
    version_seen: String,
    current_snippet_hash: Option<String>,
    docs_preview: Option<String>,
 ) {
    let Some(key) = resolve_cloud_key() else {
        tracing::warn!(
            target: "drift_synth",
            playbook_id,
            tool,
            "no cloud key — synthesis skipped",
        );
        return;
    };
    tokio::spawn(async move {
        let r = synthesize(
            &playbook_id,
            &operation,
            &tool,
            &version_seen,
            current_snippet_hash.as_deref(),
            docs_preview.as_deref(),
            &key,
        ).await;
        match r {
            Ok(c) => match append_correction(&c).await {
                Ok(_) => tracing::info!(
                    target: "drift_synth",
                    playbook_id = c.playbook_id,
                    tool = c.tool,
                    "drift correction appended",
                ),
                Err(e) => tracing::warn!(target: "drift_synth", "append failed: {e}"),
            },
            Err(e) => tracing::warn!(target: "drift_synth", "synthesis failed: {e}"),
        }
    });
 }
 fn build_prompt(tool: &str, version_seen: &str, operation: &str, preview: Option<&str>) -> String {
    let preview_block = match preview {
        Some(p) if !p.trim().is_empty() => format!(
            "Current docs preview (first ~400 chars from context7):\n---\n{}\n---\n",
            p.chars().take(1000).collect::<String>(),
        ),
        _ => "(no current docs preview available)\n".to_string(),
    };
    format!(
        "A playbook references tool `{tool}` at version_seen=`{version_seen}`. \
 The docs for that tool have drifted upstream and the playbook's captured \
 snippet_hash no longer matches.
 Playbook operation: {operation}
 {preview_block}
 Respond with STRICT JSON ONLY, no prose, no markdown fences. Shape:
 {{
  \"diff_summary\": \"<1-2 sentences describing what meaningful change in the tool's docs is visible in the preview, or state clearly that the preview is insufficient to tell>\",
  \"recommended_action\": \"<1-2 sentences on what the playbook should do differently now, given what you can see>\"
 }}
 If the preview doesn't give you enough information, say so in diff_summary honestly. Don't fabricate a diff."
    )
 }
 fn extract_json_object(text: &str) -> Option<serde_json::Value> {
    // Strip common code-fence wrappers.
    let cleaned = text
        .trim()
        .trim_start_matches("```json")
        .trim_start_matches("```")
        .trim_end_matches("```")
        .trim();
    let bytes = cleaned.as_bytes();
    let mut depth = 0i32;
    let mut start: Option<usize> = None;
    for (i, &b) in bytes.iter().enumerate() {
        match b {
            b'{' => {
                if depth == 0 {
                    start = Some(i);
                }
                depth += 1;
            }
            b'}' => {
                depth -= 1;
                if depth == 0 {
                    if let Some(s) = start {
                        if let Ok(v) = serde_json::from_str::<serde_json::Value>(&cleaned[s..=i]) {
                            return Some(v);
                        }
                    }
                    start = None;
                }
            }
            _ => {}
        }
    }
    None
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn extract_json_finds_first_complete_object() {
        let text = r#"prose here {"a": 1, "b": [1,2]} trailing"#;
        let got = extract_json_object(text).expect("should parse");
        assert_eq!(got["a"], 1);
    }
    #[test]
    fn extract_json_handles_code_fences() {
        let text = "```json\n{\"diff_summary\": \"x\", \"recommended_action\": \"y\"}\n```";
        let got = extract_json_object(text).unwrap();
        assert_eq!(got["diff_summary"], "x");
        assert_eq!(got["recommended_action"], "y");
    }
    #[test]
    fn extract_json_rejects_unclosed_object() {
        let text = "{ \"a\": 1, \"b\": ";
        assert!(extract_json_object(text).is_none());
    }
    #[tokio::test]
    async fn append_correction_creates_parent_dir_and_appends_jsonl() {
        // Use a temp path under /tmp so we don't pollute the real
        // data/_kb. Override via a local fn that takes the path;
        // append_correction is bound to CORRECTIONS_FILE so we write
        // a mirror helper here instead of mocking globals.
        let tmp = std::env::temp_dir().join("lakehouse_drift_synth_test");
        let _ = tokio::fs::remove_dir_all(&tmp).await;
        let nested = tmp.join("sub/dir/doc_drift_corrections.jsonl");
        tokio::fs::create_dir_all(nested.parent().unwrap()).await.unwrap();
        let c = DriftCorrection {
            playbook_id: "pb-1".into(),
            operation: "fill: X in Y, Z".into(),
            tool: "docker".into(),
            version_seen: "24.0.7".into(),
            current_snippet_hash: Some("abc".into()),
            diff_summary: "docs changed default TLS policy".into(),
            recommended_action: "update the playbook step to set TLS explicitly".into(),
            generated_at: "2026-04-22T00:00:00Z".into(),
            model: "gpt-oss:120b".into(),
        };
        let line = serde_json::to_string(&c).unwrap();
        tokio::fs::write(&nested, format!("{line}\n")).await.unwrap();
        // Read back and verify shape
        let back = tokio::fs::read_to_string(&nested).await.unwrap();
        assert!(back.ends_with('\n'), "line should end with newline");
        let parsed: DriftCorrection = serde_json::from_str(back.trim()).unwrap();
        assert_eq!(parsed.playbook_id, "pb-1");
        assert_eq!(parsed.tool, "docker");
        let _ = tokio::fs::remove_dir_all(&tmp).await;
    }
    #[test]
    fn build_prompt_includes_tool_and_operation() {
        let p = build_prompt("terraform", "1.5.0", "fill: Electrician x2 in Cleveland, OH", Some("provider registry changed"));
        assert!(p.contains("terraform"));
        assert!(p.contains("1.5.0"));
        assert!(p.contains("Cleveland"));
        assert!(p.contains("provider registry changed"));
        assert!(p.contains("STRICT JSON"));
    }
    #[test]
    fn build_prompt_handles_missing_preview_honestly() {
        let p = build_prompt("docker", "24.0", "fill: X", None);
        assert!(p.contains("no current docs preview available"));
    }
 }
@@ -9,7 +9,6 @@ pub mod index_registry;
 pub mod jobs;
 pub mod playbook_memory;
 pub mod doc_drift;
 pub mod drift_synth;
 pub mod promotion;
 pub mod refresh;
 pub mod store;
@@ -136,8 +136,6 @@ pub fn router(state: VectorState) -> Router {
        // Phase 45 slice 3 — doc drift detection + human re-admission.
        .route("/playbook_memory/doc_drift/check/{id}", post(check_doc_drift))
        .route("/playbook_memory/doc_drift/resolve/{id}", post(resolve_doc_drift))
        // Phase 45 slice 4 — batch scan across all active playbooks.
        .route("/playbook_memory/doc_drift/scan", post(scan_doc_drift))
        .with_state(state)
 }
@@ -2567,135 +2565,12 @@ async fn check_doc_drift(
        false
    };
    // Phase 45 slice 4 — fire-and-forget T3 synthesis per drifted tool.
    // Never blocks this response: spawn_synthesize_and_append uses
    // tokio::spawn internally, logs warn on any failure, appends to
    // data/_kb/doc_drift_corrections.jsonl on success. No cloud key
    // present → synthesis is skipped silently.
    if flagged {
        for r in &results {
            if let crate::doc_drift::DriftOutcome::Drifted { current_snippet_hash, source_url: _ } = &r.outcome {
                crate::drift_synth::spawn_synthesize_and_append(
                    entry.playbook_id.clone(),
                    entry.operation.clone(),
                    r.tool.clone(),
                    r.version_seen.clone(),
                    Some(current_snippet_hash.clone()),
                    None, // docs_preview not returned by check_all_refs yet; slice 4.1 extension
                );
            }
        }
    }
    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,
        "synthesis_spawned": any_drifted,
    })))
 }
 /// Phase 45 slice 4 — batch scan across all active playbooks.
 ///
 /// POST /vectors/playbook_memory/doc_drift/scan
 ///   body: { city?: string, state?: string, max_entries?: number }
 ///
 /// Iterates every active (non-retired, non-superseded) playbook that
 /// has at least one doc_ref, optionally filtered by (city, state).
 /// For each: queries the bridge, flags if drifted, fires synthesis.
 ///
 /// NOT paginated — if you have >100 playbooks with doc_refs, this
 /// will take a while (each check is a bridge HTTP call; bridge caches
 /// for 5 min so re-runs are fast). Response reports counts per outcome
 /// so you can tell if an operator should retry.
 #[derive(serde::Deserialize, Default)]
 struct ScanDocDriftRequest {
    #[serde(default)]
    city: Option<String>,
    #[serde(default)]
    state: Option<String>,
    #[serde(default)]
    max_entries: Option<usize>,
 }
 async fn scan_doc_drift(
    State(state): State<VectorState>,
    body: Option<Json<ScanDocDriftRequest>>,
 ) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
    use crate::doc_drift::{check_all_refs, DriftCheckerConfig, DriftOutcome};
    let req = body.map(|Json(r)| r).unwrap_or_default();
    let cfg = DriftCheckerConfig::default();
    let entries = state.playbook_memory.snapshot().await;
    let matches_filter = |e: &crate::playbook_memory::PlaybookEntry| -> bool {
        if e.retired_at.is_some() || e.superseded_at.is_some() { return false; }
        if e.doc_refs.is_empty() { return false; }
        if let Some(c) = &req.city {
            let ec = e.city.as_deref().unwrap_or("");
            if !ec.eq_ignore_ascii_case(c) { return false; }
        }
        if let Some(s) = &req.state {
            let es = e.state.as_deref().unwrap_or("");
            if !es.eq_ignore_ascii_case(s) { return false; }
        }
        true
    };
    let eligible: Vec<_> = entries.iter().filter(|e| matches_filter(e)).collect();
    let limit = req.max_entries.unwrap_or(eligible.len());
    let mut scanned = 0usize;
    let mut drifted_count = 0usize;
    let mut newly_flagged = 0usize;
    let mut unknown_count = 0usize;
    let mut synthesis_spawned = 0usize;
    let mut details: Vec<serde_json::Value> = Vec::new();
    for e in eligible.iter().take(limit) {
        scanned += 1;
        let results = check_all_refs(&cfg, &e.doc_refs).await;
        let any_drifted = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Drifted { .. }));
        let any_unknown = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Unknown { .. }));
        if any_unknown && !any_drifted { unknown_count += 1; }
        if any_drifted {
            drifted_count += 1;
            if state.playbook_memory.flag_doc_drift(&e.playbook_id).await.unwrap_or(false) {
                newly_flagged += 1;
            }
            for r in &results {
                if let DriftOutcome::Drifted { current_snippet_hash, .. } = &r.outcome {
                    crate::drift_synth::spawn_synthesize_and_append(
                        e.playbook_id.clone(),
                        e.operation.clone(),
                        r.tool.clone(),
                        r.version_seen.clone(),
                        Some(current_snippet_hash.clone()),
                        None,
                    );
                    synthesis_spawned += 1;
                }
            }
        }
        details.push(serde_json::json!({
            "playbook_id": e.playbook_id,
            "drifted": any_drifted,
            "unknown": any_unknown,
            "tools": results.iter().map(|r| &r.tool).collect::<Vec<_>>(),
        }));
    }
    Ok(Json(serde_json::json!({
        "scanned": scanned,
        "eligible": eligible.len(),
        "drifted": drifted_count,
        "newly_flagged": newly_flagged,
        "unknown": unknown_count,
        "synthesis_spawned": synthesis_spawned,
        "filter": { "city": req.city, "state": req.state, "max_entries": req.max_entries },
        "details": details,
    })))
 }
@@ -1,45 +0,0 @@
 // Unit tests for context7_bridge.ts pure helpers (slice 2 backfill).
 // Run: bun test mcp-server/context7_bridge.test.ts
 //
 // The HTTP-bound functions (resolveLibraryId, fetchDocsText,
 // getCurrent) need network access and are exercised by the hybrid
 // fixture on the auditor side — not re-tested here to avoid
 // duplicating what's already integration-covered. This file only
 // exercises the pure, deterministic helpers.
 import { test, expect } from "bun:test";
 import { normalizeTool, hashContent } from "./context7_bridge.ts";
 test("normalizeTool lowercases + trims", () => {
  expect(normalizeTool("Docker")).toBe("docker");
  expect(normalizeTool("  TERRAFORM  ")).toBe("terraform");
  expect(normalizeTool("Next.JS")).toBe("next.js");
 });
 test("normalizeTool preserves internal chars + dashes", () => {
  expect(normalizeTool("react-dom")).toBe("react-dom");
  expect(normalizeTool("@anthropic/sdk")).toBe("@anthropic/sdk");
 });
 test("hashContent is deterministic", () => {
  const a = hashContent("hello world");
  const b = hashContent("hello world");
  expect(a).toBe(b);
 });
 test("hashContent is sensitive to single-char change", () => {
  const a = hashContent("docker run -it ubuntu bash");
  const b = hashContent("docker run -it ubuntu bashh"); // extra h
  expect(a).not.toBe(b);
 });
 test("hashContent returns 16 hex chars", () => {
  const h = hashContent("any input");
  expect(h).toMatch(/^[0-9a-f]{16}$/);
 });
 test("hashContent differs for empty vs whitespace", () => {
  const a = hashContent("");
  const b = hashContent(" ");
  expect(a).not.toBe(b);
 });
@@ -36,8 +36,7 @@ interface CachedDoc {
 const cache = new Map<string, { entry: CachedDoc; at: number }>();
-// Exported for unit testing — pure functions that the bridge composes.
+function normalizeTool(s: string): string {
 export function normalizeTool(s: string): string {
  return s.trim().toLowerCase();
 }
@@ -70,7 +69,7 @@ async function fetchDocsText(libraryId: string): Promise<string> {
  return await r.text();
 }
-export function hashContent(s: string): string {
+function hashContent(s: string): string {
  return createHash("sha256").update(s).digest("hex").slice(0, 16);
 }
@@ -106,11 +105,6 @@ function jsonResponse(body: unknown, status: number = 200): Response {
  });
 }
 // Only start the HTTP server when this file is run as the entry
 // point. When imported by tests (or future callers), the pure
 // helpers are exposed without starting a second server on the same
 // port (systemd service lakehouse-context7-bridge already owns it).
 if (import.meta.main) {
 Bun.serve({
  port: PORT,
  hostname: "0.0.0.0",
@@ -182,4 +176,3 @@ Bun.serve({
 });
 console.log(`[context7-bridge] listening on :${PORT} (context7 base: ${CONTEXT7_BASE})`);
 } // end of if (import.meta.main) guard
@@ -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); });
@@ -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); });
@@ -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); });
Author	SHA1	Message	Date
profit	dc01ba0a3b	auditor: kb_query surfaces scrum-master reviews for files in PR diff lakehouse/auditor 2 blocking issues: unimplemented!() macro call in tests/real-world/hard_task_escalation.ts 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.	2026-04-22 21:18:21 -05:00
root	89d188074b	scrum_master: tree-split + scrum_reviews.jsonl writer + truncation warning Extends the scrum-master pipeline to handle input overflow on large source files (>6KB). Previously, the review prompt truncated the file to first-chunk, which caused false-positive "field is missing" findings whenever the actual field was past the cutoff. Now each file >FILE_TREE_SPLIT_THRESHOLD (6000) is sharded at FILE_SHARD_SIZE (3500), each shard summarized via gpt-oss:120b cloud, and the distillations merged into a scratchpad. The review then runs against the scratchpad with an explicit truncation-awareness clause in the prompt: "DO NOT claim any field, function, or feature is 'missing' based on its absence from this distillation." Also writes each accepted review as a JSONL row to data/_kb/scrum_reviews.jsonl (file, reviewed_at, accepted_model, accepted_on_attempt, attempts_made, tree_split_fired, preview). This is the source the auditor's kb_query reads to surface per-file scrum reviews on PRs that touch those files (cohesion plan Phase C). Verified: scrum review of 92KB playbook_memory.rs → 27 shards via cloud → distilled scratchpad → qwen3.5 local 7B accepted on attempt 1 (5931 chars). Tree-split fires, jsonl row appended, output file contains structured suggestions.	2026-04-22 21:17:53 -05:00
profitandClaude Opus 4.7	a7aba31935	tests/real-world: scrum-master pipeline — composes everything we built The orchestrator J described: pulls git repo source + PRD + suggested-changes doc, chunks them, hands each code piece through the proven escalation ladder with learning context, collects per-file suggestions in a consolidated handoff report. Composes ONLY already-shipped primitives — no new core code: - chunker with 800-char / 120-overlap windows - sidecar /embed for real nomic-embed-text embeddings - in-memory cosine retrieval for top-5 PRD + top-5 proposal chunks per target file - escalation ladder (qwen3.5 → qwen3 → gpt-oss:20b → gpt-oss:120b → devstral-2:123b → mistral-large-3:675b) - per-attempt learning-context injection (prior failures as "do not repeat" block) - acceptance rubric (length ≥ 200 chars + structured form) Live-run (tests/real-world/runs/scrum_moatqkee/): targets: 3 files - crates/vectord/src/playbook_memory.rs (920 lines) - crates/vectord/src/doc_drift.rs (163 lines) - auditor/audit.ts (170 lines) resolved: 3/3 on attempt 1 by qwen3.5:latest local 7B total duration: 111.7s output: scrum_report.md + per-file JSON Sample from scrum_report.md (playbook_memory.rs review): - Alignment score: 9/10 vs PRD Phase 19 - 4 concrete change suggestions naming specific lines + PLAN/PRD chunk offsets - 3 gap analyses with PRD-reference citations Honest findings from this run: 1. Local 7B handled review-style tasks first-try. The escalation ladder infrastructure is live but didn't fire — review is an easier task shape than strict code-generation (see hard_task test which needed devstral-2 specialist). 2. 6KB file-truncation caused one false positive: model claimed playbook_memory.rs lacks a `doc_refs` field, but that field exists past the 6KB cutoff. Trade-off between context-size and review-depth needs tuning per file. 3. Chunk-offset citations are real: model output includes `[PRD @27880]` and `[PLAN @16320]` which map to the actual byte offsets of retrieved context chunks. Auditor pattern could adopt this for traceable claims. This is the scrum-master-handoff shape J asked for: repo + PRD + proposal → chunk → retrieve → escalate → consolidate → human-reviewable markdown report Not shipping: per-PR diff analysis, open-PR integration, Gitea posting of suggestions. Those compose the same primitives differently — this proves the core pattern. Env override: LH_SCRUM_FILES=path1,path2,... to target a different file set. Default 3 files keeps runtime ~2min. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>	2026-04-22 20:52:42 -05:00
profitandClaude Opus 4.7	540c493ff1	tests/real-world: hard-task escalation — prove the ladder solves tasks local can't J asked (2026-04-22): construct a task the local model provably can't complete, then watch the escalation + retry + cloud pipeline actually solve it. The task: generate a Rust async function with 15 specific structural rules (exact signature, bounded concurrency, exponential backoff 250/500/1000ms, NO .unwrap(), rustdoc comments, etc.). Small enough to fit in one response but strict enough that one rule violation = not accepted. Fits Rust + async + concurrency + error-handling — across the hardest dimensions for 7B models. Escalation ladder (corrected per J — kimi-k2.x requires Ollama Cloud Pro subscription which J's key lacks; mistral-large-3:675b is the biggest provisioned model): 1. qwen3.5:latest (local 7B) 2. qwen3:latest (local 7B) 3. gpt-oss:20b (local 20B) 4. gpt-oss:120b (cloud 120B) 5. devstral-2:123b (cloud 123B coding specialist) 6. mistral-large-3:675b (cloud 675B — biggest available) Each attempt gets PRIOR failures' rubric violations injected as learning context. Loop caps at MAX_ATTEMPTS=6. Live run (runs/hard_task_moapd3g3/): attempt 1: qwen3.5:latest 11/15 — missed concurrency + some constraints attempt 2: qwen3:latest 11/15 — different misses after learning attempt 3: gpt-oss:20b 0/1 — empty response (local model dead-end) attempt 4: gpt-oss:120b 0/1 — empty (heavy learning context may confuse) attempt 5: devstral-2:123b 15/15 ✅ ACCEPTED after 10.4s attempt 6: (not reached) Total: 5 attempts, 145.6s, coding-specialist succeeded. Honest findings from the run: - Pipeline works: escalated through 4 distinct model tiers, injected learning, bounded at 6, graceful failure surfaces. - Learning injection doesn't always help general-purpose models — gpt-oss:120b returned empty when given heavy prior-failure context (attempt 4). The coding specialist (devstral) worked better because the task is domain-aligned. - Local 7B came within 4 rules of success first-try (11/15) — not bad for the scale, but specific constraints like "EXACT signature" and "bounded concurrency at 4" are where small models slip. - Kimi K2.5/K2.6 both require a paid subscription on our current Ollama Cloud key — verified via direct ollama.com curl. Swap to kimi once subscription lands. Also includes a rubric bug-fix caught in the run: the regex for "reaches 500/1000ms backoff" originally required literal constants, but devstral-2:123b wrote idiomatic `retry_delay = 2;` which doubles 250 → 500 → 1000 correctly. Broadened rubric to recognize `= 2`, bit-shift, `.pow()`, and literal forms. Without this the ladder would have false-failed on semantically-correct code. Files: tests/real-world/hard_task_escalation.ts (270 LOC) tests/real-world/runs/hard_task_moapd3g3/ attempt_{1..5}.txt — raw model outputs (last successful) attempt_{1..5}.json — per-attempt rubric verdict + error summary.json — ladder summary What this PROVES that no prior test did: - Task-level retry ESCALATES across distinct model capabilities (not just same model retried) - Bigger and more-specialized models ACTUALLY solve what smaller ones can't — the ladder works by design, not by luck - The subscription boundary (Kimi K2.x) is a real operational constraint, not a code issue - Rubric engineering is its own discipline — a strict-but-wrong validator can reject correct code; shipping the test harness required tuning against actual model outputs Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>	2026-04-22 18:50:53 -05:00
profit	6d6a306d4e	tests/real-world: add task-level 6-retry loop (per J 2026-04-22) Two distinct retry loops now both cap at 6 and serve different purposes: 1. Per-cloud-call continuation (Phase 21 primitive) — when a single cloud call returns empty or truncated, stitches up to 6 continuation calls. Handles output-overflow. 2. Per-TASK retry (this commit) — when the whole task errors (500/404, thin answer, etc.), retries the full task up to 6 times. Each retry gets PRIOR ATTEMPTS' failures injected into the prompt as learning context, so attempt N+1 is informed by what N failed at. Handles error-recovery with compounding context. Both loops fired on iter 3 of the stress run, proving them independent and composable: FORCING TASK-RETRY LOOP — iter 3 will cycle through 5 invalid models + 1 valid attempt 1/6: model=deliberately-invalid-model-attempt-1 /v1/chat 502: ollama.com 404: model not found attempt 2/6: [with prior-failure context] ... (5 failures total, each with the full chain of prior errors) attempt 6/6: model=gpt-oss:20b [with prior-failure context] continuation retry 1..6 (empty responses) SUCCEEDED after 5 prior failures (441 chars) What J was asking to prove: "I expect it to retry the process six times to build on the knowledge database... when an error is legitimately triggered that it will go through six times... without getting caught in a loop" Proof: - 6/6 attempts fired on the FORCED iteration - Each retry embedded the preceding attempts' errors as "do not repeat" context - Hard cap at MAX_TASK_RETRIES (6) prevents infinite loops - Last-ditch local fallback exists if all 6 still fail - Other iterations succeed on attempt 1 — the loop ONLY fires when errors are legitimately triggered Stress run totals (runs/moan4h71/): 6/6 iterations complete, 58 cloud calls, 306s end-to-end tree-splits: 6/6 continuations: 10 rescues: 2 iter 3: 8197+2800 tok, 6 task attempts, 6 continuation retries local stored summary + per-iter JSON for inspection What this proves that prior stress runs did NOT: - Error-recovery at task granularity is live, not aspirational - Compounding failure context flows between retries as text - Loop bound is enforced; runaway cases aren't possible - Two retry mechanisms compose without deadlock (continuation inside task-retry inside tree-split) Follow-ups worth doing (separate PRs): - Persist retry-history to observer :3800 so cross-run learning sees the failure patterns - Route retries through /vectors/hybrid to surface similar prior errors from the real KB (currently only in-memory across one iteration) - Fix citation regex in summary — iter 6 received 5 prior IDs but counter shows 0 (regex needs to tolerate hyphens in IDs)	2026-04-22 17:50:53 -05:00
profitandClaude Opus 4.7	4458c94f45	tests/real-world: enrich_prd_pipeline — architecture stress test Real end-to-end test of the Lakehouse pipeline at scale. Runs the PRD (63 KB, 901 lines → 93 chunks) through 6 iterations with cloud inference, intentional failure injection, and tight context budget to force every Phase 21 primitive to fire. What the test exercises: - Sidecar /embed for 93 chunks (nomic-embed-text) - In-memory cosine retrieval for top-K per iteration - Tree-split (shard → summarize → scratchpad → merge) when context chunks exceed the 4000-char budget - Scratchpad truncation to keep compounding context bounded - Cloud inference via /v1/chat provider=ollama_cloud (gpt-oss:120b) - Injected primary-cloud failure on iter 3 (invalid model name) + rescue with gpt-oss:20b — proves catch-and-retry isn't dead code - Playbook seeding per iteration (real HTTP against gateway) - Prior-iteration answer injection for compounding (not just IDs — the first version passed IDs only and the model ignored them) Live run results (tests/real-world/runs/moamj810/): 6/6 iterations complete, 42 cloud calls total, 245s end-to-end tree-splits: 6/6 (every iter overflowed 4K budget) continuations: 0 (no responses hit max_tokens) rescues: 1 (iter 3 injected failure → gpt-oss:20b → valid answer) iter 6 answer explicitly cites [pb:pb-seed-82e1] — compounding real scratchpad truncation fired on iter 6 as designed What this PROVES: - Tree-split primitives work under real context pressure, not just in unit tests. The 4000-char budget forced every iteration to shard 12 chunks → 6 shards → scratchpad → final answer. - Rescue on primary failure is wired and produces answers from a weaker model rather than erroring out. - Compounding context injection works: iter 6's prompt had the 5 prior answers in its citation block, and the cloud model acknowledged at least one via [pb:...] notation. - The existence claims in Phase 21 (continuation + tree-split) are backed by executable evidence, not just unit tests. What this DOESN'T prove (deliberate — scoped for follow-up): - Continuation retries (no iter hit max_tokens in this run; would need a harder prompt or lower max_tokens to force) - Real integration with /vectors/hybrid endpoint (test does in-memory cosine instead, bypassing gateway vector surface) - Observer consumption of these runs (nothing posted to :3800 during the test — adding that is Phase A integration, handled separately) Files: tests/real-world/enrich_prd_pipeline.ts (333 LOC) tests/real-world/runs/moamj810/{iter_1..6.json, summary.json} — artifacts from the stress run, committed for inspection Follow-ups worth doing: 1. Lower max_tokens / harder prompt to force continuation path 2. Route retrieval through /vectors/hybrid for real Phase 19 boost 3. POST per-iteration summary to observer :3800 so runs accumulate like scenario runs do Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>	2026-04-22 17:33:24 -05:00