Cohesion: Python inventory + integration plan + Phase A verdict indexing

Three artifacts in one PR: 1. docs/PYTHON_INVENTORY.md — every .py file in the repo classified: Production (sidecar routers + 3 systemd services), Documented (kb_measure, kb_staffer_report), Manual (one-off tools), Dead (sidecar/sidecar/lab_ui.py + pipeline_lab.py are genuinely not imported anywhere). 2. docs/COHESION_INTEGRATION_PLAN.md — the "smarter DB" loop J called out as missing. Six phases A-F. Phase A ships here; B-F are named + sequenced for follow-up PRs. Each phase adds ONE wire of the loop; no single PR does them all. 3. Phase A wire (auditor verdicts → observer + KB): - auditor/audit.ts: after assembleVerdict, fire-and-forget POST to :3800/event with source="auditor" AND append to data/_kb/outcomes.jsonl with kind="audit". Errors log + drop — the verdict is still on disk at _auditor/verdicts/. - mcp-server/observer.ts: extend source union to include "auditor" | "bot" (was "mcp" | "scenario" only, which silently coerced my first auditor POST to source="scenario"). Accept body.ok OR body.success. Accept body.audit_duration_ms as a fallback for duration_ms. Uses body.one_liner as output_summary when set. Live-verified after observer restart: re-audit PR #6 → verdict=request_changes, 4 findings (1 warn) observer: by_source={'auditor': 1} (previously coerced to 'scenario') _kb/outcomes.jsonl tail: kind=audit sig=pr6-7fe47bab pr=6 overall=request_changes The shape of the loop is now visible to downstream consumers. Phase B (auditor's kb_query check reads these audit rows for history) lands in a follow-up PR. Phase C-F similar. NOT in this PR: - Actually deleting lab_ui.py + pipeline_lab.py (operator decision, called out in the inventory doc) - Cleaning up the 5 overlapping Python scripts (same) - Phases B-F of the cohesion plan (separate PRs per wire) - Integration test that asserts "smarter DB" across runs (Phase F) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-22 17:22:42 -05:00
9 changed files with 300 additions and 1376 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, pr.files.map(f => f.path)),
+    runKbCheck(claims),
  ]);
  const allFindings: Finding[] = [
@@ -80,6 +80,16 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
  await persistVerdict(verdict);
  // Phase A of the cohesion plan (docs/COHESION_INTEGRATION_PLAN.md):
  // make every audit verdict visible to the observer + KB. Enables
  // future Phase B (kb_query sees prior audit history) without a
  // separate backfill. Fire-and-forget: observer/KB failures don't
  // block the Gitea post.
  indexVerdictToObserver(verdict).catch(e =>
    console.error(`[auditor] observer indexing failed: ${(e as Error).message}`));
  appendVerdictToKbOutcomes(verdict).catch(e =>
    console.error(`[auditor] kb outcomes append failed: ${(e as Error).message}`));
  if (!opts.dry_run) {
    await postToGitea(verdict);
  }
@@ -87,6 +97,68 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
  return verdict;
 }
 // Phase A — verdict indexing.
 //
 // Two destinations, both append-only + non-blocking:
 //   1. observer :3800/event — ring buffer + data/_observer/ops.jsonl
 //   2. data/_kb/outcomes.jsonl — same file scenarios write to, with
 //      kind:"audit" so readers can filter
 //
 // Errors log + drop. The verdict is still on disk at
 // _auditor/verdicts/{pr}-{sha}.json; observer + KB are a convenience
 // surface, not a source of truth.
 const OBSERVER_URL = process.env.LH_OBSERVER_URL ?? "http://localhost:3800";
 const KB_OUTCOMES = "/home/profit/lakehouse/data/_kb/outcomes.jsonl";
 async function indexVerdictToObserver(v: Verdict): Promise<void> {
  const payload = {
    source: "auditor",
    event_kind: "audit",
    ok: v.overall === "approve",
    sig_hash: `pr${v.pr_number}-${v.head_sha.slice(0, 8)}`,
    pr_number: v.pr_number,
    head_sha: v.head_sha,
    overall: v.overall,
    one_liner: v.one_liner,
    findings_block: v.metrics.findings_block,
    findings_warn: v.metrics.findings_warn,
    audit_duration_ms: v.metrics.audit_duration_ms,
    audited_at: v.audited_at,
  };
  const r = await fetch(`${OBSERVER_URL}/event`, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify(payload),
    signal: AbortSignal.timeout(3000),
  });
  if (!r.ok) throw new Error(`observer ${r.status}: ${await r.text()}`);
 }
 async function appendVerdictToKbOutcomes(v: Verdict): Promise<void> {
  const { appendFile, mkdir } = await import("node:fs/promises");
  const { dirname } = await import("node:path");
  await mkdir(dirname(KB_OUTCOMES), { recursive: true });
  const row = {
    kind: "audit",
    sig_hash: `pr${v.pr_number}-${v.head_sha.slice(0, 8)}`,
    audited_at: v.audited_at,
    pr_number: v.pr_number,
    head_sha: v.head_sha,
    overall: v.overall,
    one_liner: v.one_liner,
    ok_events: v.overall === "approve" ? 1 : 0,
    total_events: 1,
    findings: {
      block: v.metrics.findings_block,
      warn: v.metrics.findings_warn,
      info: v.metrics.findings_info,
    },
    elapsed_secs: (v.metrics.audit_duration_ms ?? 0) / 1000,
  };
  await appendFile(KB_OUTCOMES, JSON.stringify(row) + "\n");
 }
 async function persistVerdict(v: Verdict): Promise<void> {
  await mkdir(VERDICTS_DIR, { recursive: true });
  const filename = `${v.pr_number}-${v.head_sha.slice(0, 12)}.json`;
@@ -1,68 +0,0 @@
 // 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,7 +8,6 @@
 // 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
 //
@@ -22,11 +21,10 @@ 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[], prFiles: string[] = []): Promise<Finding[]> {
+export async function runKbCheck(claims: Claim[]): Promise<Finding[]> {
  const findings: Finding[] = [];
  // 1. Recent scenario outcomes: are strong-claim-style phrases showing
@@ -50,15 +48,6 @@ export async function runKbCheck(claims: Claim[], prFiles: string[] = []): Promi
  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;
 }
@@ -192,54 +181,3 @@ 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;
 }
@@ -0,0 +1,126 @@
 # Cohesion integration plan — the "smarter DB" loop
 **Written:** 2026-04-22, after J flagged that the system has good parts but they don't compose into the self-improving loop promised in the control-plane thesis (`project_control_plane_thesis.md` memory: 0→85% via hyperfocus-then-escalate).
 ## The gap
 Each piece works in isolation. What's not wired:
 ```
 ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
 │   observer  │──✓──│  data/_kb/  │──?──│   auditor   │
 │    :3800    │     │ outcomes.   │     │  kb_query   │
 │             │     │   jsonl     │     │   check     │
 └─────────────┘     └─────────────┘     └─────────────┘
      ▲                                         │
      │?                                        ▼
 ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
 │  auditor    │─────│ data/_audit │     │   cloud     │
 │  verdicts   │     │or/verdicts/ │     │  inference  │
 │             │     │             │     │             │
 └─────────────┘     └─────────────┘     └─────────────┘
                                                │?
                                                ▼
                                         ┌─────────────┐
                                         │ hybrid_srch │
                                         │ playbook_mem│
                                         │  context7   │
                                         └─────────────┘
 ```
 - **✓** solid: observer captures scenarios, writes `_kb/outcomes.jsonl`.
 - **?** missing:
  - auditor verdicts go to `_auditor/verdicts/` — not to observer or KB
  - auditor kb_query reads `_kb/outcomes.jsonl` but doesn't query hybrid_search, playbook_memory, or context7
  - auditor inference sends claim+diff to cloud — but WITHOUT KB neighbors, WITHOUT drift context, WITHOUT tool-aware retrieval
 - Result: every audit is stateless-ish. The DB doesn't get smarter across audits.
 ## The target loop
 ```
 PR opened
  ↓
 auditor fetches diff + claims
  ↓
 auditor enriches with:
  — KB neighbors: past verdicts on similar claim-hashes
  — hybrid_search: playbooks for task classes the diff touches
  — context7 drift status: for any tool named in commit messages
  — MCP tools: agent-style queries if relevant
  ↓
 cloud inference sees: diff + claims + enrichment
  ↓
 verdict posted to Gitea
  ↓
 verdict ALSO persisted to:
  — observer :3800/event (source=auditor)
  — _kb/outcomes.jsonl (kind=audit, sig_hash=pr+sha)
  ↓
 next audit on similar PR:
  — KB neighbors find this prior verdict
  — cloud sees "similar PRs ended in request_changes for reason X"
  — verdict is more calibrated
 ```
 ## Phases of closure
 Building this in order from least-invasive:
 ### Phase A — Verdict indexing (shipped in this PR)
 After every `auditPr()` completes, in addition to persisting to `_auditor/verdicts/`:
 1. POST the verdict to observer `:3800/event` with `source: "auditor"`, `ok: verdict === "approve"`, `event_kind: "audit"`, `sig_hash: <stable hash of pr_number + head_sha>`.
 2. Append a simplified outcome to `data/_kb/outcomes.jsonl` with a special `kind: "audit"` row so the KB surface treats audits alongside scenarios.
 Minimal surgery. Doesn't change the verdict itself. Just makes it visible to downstream consumers.
 **Test:** after this PR lands, re-run the auditor once; observer stats should show `by_source.auditor > 0`; `_kb/outcomes.jsonl` should have one new row per audited SHA.
 ### Phase B — KB query sees auditor history
 Extend `auditor/checks/kb_query.ts` to:
 1. Read `_kb/outcomes.jsonl`, filter `kind === "audit"`, find prior audits with matching `sig_hash` (same PR across SHAs) or similar claim hashes.
 2. Emit finding: "prior N audits on this PR ended in [approve, block, request_changes], last reason was X."
 Gives the auditor memory across re-audits of the same PR.
 ### Phase C — Hybrid search in kb_query
 When a PR touches `crates/vectord/src/playbook_memory.rs` (task-class matching), auditor calls `POST /vectors/hybrid` to find playbooks semantically related to the diff. Surfaces as "N playbooks in production rely on this code path, consider backward-compat."
 Requires:
 - Task-class extractor from diff paths
 - Cloud-free hybrid_search call (we have local Ollama for embeddings)
 ### Phase D — Context7 drift awareness in auditor
 If the diff / commit message names a tool that's in any playbook's `doc_refs`, auditor calls the context7 bridge to check current drift status. Surfaces as "tool X has drifted since N playbooks referenced it; this PR may need to update those."
 ### Phase E — Inference sees enrichment
 The cloud inference check currently sends `diff + claims`. Extend to send `diff + claims + kb_neighbors + drift_context + hybrid_search_matches`. The prompt becomes context-rich — exactly what makes the cloud model competent (per the control-plane thesis).
 ### Phase F — Full integration test
 An auditor self-test that:
 1. Creates a synthetic PR with known-good and known-bad claims
 2. Runs the enriched auditor
 3. Asserts the verdict found the planted issues
 4. Runs a second identical-claim PR
 5. Asserts the SECOND verdict references the FIRST audit (via KB neighbor retrieval)
 6. "Smarter DB" proof: two runs, measurable context gain.
 ## Sequence
 Phase A lands in this PR alongside the inventory. Phases B-F are follow-up PRs, each with their own auditor gate. Order matters: A → B (reads A's output) → C+D (in parallel) → E (consumes B/C/D) → F (asserts E's behavior).
 ## Not in this plan (deliberate)
 - **Rewriting the auditor to use a different architecture.** Current 4-check model stays; we just enrich the checks.
 - **Tuning cloud inference precision.** The false-positive rate is a prompt-engineering concern; this plan is about context enrichment, which is separate.
 - **Branch protection enforcement.** Stays off until Phase F passes.
 The overall bet: this is the "putting it all together coherently" J said was a real problem. Six phases over however many PRs it takes. Each one ships one wire of the loop; no single PR tries to do them all.
@@ -0,0 +1,75 @@
 # Python inventory — what's wired vs dead
 Generated 2026-04-22. Enumerates every `.py` file in the repo and classifies each by actual integration status. Goal: surface the "ambiguous names that don't hook in" J flagged.
 ## Classification
 - **🟢 Production** — running as a systemd service, or imported by code that does
 - **🟡 Documented** — referenced in `docs/PRD.md` / `docs/PHASES.md` / `docs/CONTROL_PLANE_PRD.md` or called from a shell/TS orchestrator
 - **🟠 Manual** — runnable as a one-off utility but not in any automated pipeline
 - **🔴 Dead** — not imported, not referenced, not called anywhere
 ## Results
 ### Sidecar (`sidecar/sidecar/`)
 The Python FastAPI sidecar on `:3200`. Runs as `lakehouse-sidecar.service`.
 | File | Status | Evidence |
 |---|---|---|
 | `main.py` | 🟢 Production | systemd ExecStart: `uvicorn sidecar.sidecar.main:app` |
 | `admin.py` | 🟢 Production | imported by `main.py` as `admin_router` |
 | `embed.py` | 🟢 Production | imported by `main.py` as `embed_router` |
 | `generate.py` | 🟢 Production | imported by `main.py` as `generate_router` |
 | `rerank.py` | 🟢 Production | imported by `main.py` as `rerank_router` |
 | `ollama.py` | 🟢 Production | imported transitively by embed/generate/rerank |
 | `__init__.py` | 🟢 Production | package marker |
 | `lab_ui.py` | 🔴 **Dead** | **not imported by any file in `sidecar/`. `scripts/serve_lab.py` only uses stdlib — doesn't import this.** Mtime 2026-04-16 (pre-session). Safe to delete. |
 | `pipeline_lab.py` | 🔴 **Dead** | **same — not imported anywhere. Mtime 2026-04-16.** Docstring says "iterative embedding/LLM pipeline experimentation" — looks like an abandoned prototype. Safe to delete. |
 ### Scripts (`scripts/`)
 17 files. Only **2** are production (behind systemd), **3** are documented in phase/PRD docs, **12** are unreferenced.
 | File | Status | Evidence |
 |---|---|---|
 | `serve_ui.py` | 🟢 Production | `lakehouse-ui.service` ExecStart |
 | `serve_imagegen.py` | 🟢 Production | `imagegen.service` ExecStart |
 | `serve_lab.py` | 🟢 Production | `pipeline-lab.service` ExecStart |
 | `kb_measure.py` | 🟡 Documented | referenced in `docs/PHASES.md` (Phase 22 aggregator) |
 | `kb_staffer_report.py` | 🟡 Documented | referenced in `docs/PHASES.md` + `docs/PRD.md` (Phase 23), also called by `scripts/run_staffer_demo.sh` |
 | `autonomous_agent.py` | 🟠 Manual | "Autonomous stress-test agent" — no docs refs, no callers. Mtime 2026-04-17. |
 | `copilot.py` | 🟠 Manual | "Staffing Co-Pilot — the anticipation layer" — no docs refs. Writes briefing to `/tmp/copilot_briefing.json` on run. |
 | `generate_demo.py` | 🟠 Manual | "realistic demo datasets" — mtime 2026-03-27, never touched this session |
 | `generate_workers.py` | 🟠 Manual | "worker profiles at scale" — usage example in docstring: `python3 generate_workers.py 100000 > /tmp/workers_100k.csv`. Still useful, just not auto-run. |
 | `lance_tune.py` | 🟠 Manual | "IVF_PQ parameter sweep" — one-shot experiment tool. |
 | `qwen3_plan.py` | 🟠 Manual | "Qwen 3 agent plan — structured test with playbook building" |
 | `quality_eval.py` | 🟠 Manual | "Quality evaluation pipeline — tests whether the system gives CORRECT" |
 | `scale_test.py` | 🟠 Manual | called by `scripts/scale_10m_test.sh` (not in systemd). 2.5M row test. |
 | `staffing_day.py` | 🟠 Manual | "Real-world staffing agency day simulation" — no docs refs. Overlaps conceptually with `tests/multi-agent/scenario.ts`. |
 | `staffing_demo.py` | 🟠 Manual | "Realistic staffing company data generator" — demo fixture. |
 | `staffing_simulation.py` | 🟠 Manual | "multi-agent stress test" — overlaps with `tests/multi-agent/scenario.ts`. |
 | `stress_test.py` | 🟠 Manual | "prove this architecture is sound or find where it breaks" — overlaps with `tests/multi-agent/run_stress.ts`. |
 ## Overlap with TypeScript / Rust equivalents
 Several Python scripts overlap conceptually with newer TS/Rust code. This is the "ambiguous names that don't hook in" pattern — the Python originals were superseded but not removed.
 | Python script | Newer equivalent | Integration drift |
 |---|---|---|
 | `scripts/staffing_simulation.py` | `tests/multi-agent/scenario.ts` | scenario.ts ships + observer-integrated (Phase 24); Python version is unwired. |
 | `scripts/staffing_day.py` | `tests/multi-agent/scenario.ts` | same. |
 | `scripts/stress_test.py` | `tests/multi-agent/run_stress.ts` | TS version is the one the auditor fixture exercises. |
 | `scripts/autonomous_agent.py` | `bot/` + `auditor/` Bun sub-agents | new Bun pattern supersedes. Python version predates the control-plane pivot. |
 | `scripts/qwen3_plan.py` | `tests/multi-agent/agent.ts` | TS agent is the one with Phase 21 continuation + cloud routing. |
 **Recommended cleanup actions (not in this PR — separate decision):**
 1. **Delete the two confirmed dead sidecar files:** `sidecar/sidecar/lab_ui.py`, `sidecar/sidecar/pipeline_lab.py`. Nothing calls them; removal is safe.
 2. **Archive the 5 overlapping scripts** (staffing_simulation, staffing_day, stress_test, autonomous_agent, qwen3_plan) — either move to `scripts/legacy/` with a README explaining they predate the TS versions, or delete if J confirms no future use.
 3. **Keep the one-off utilities** (generate_demo, generate_workers, lance_tune, quality_eval, copilot, scale_test, staffing_demo) — useful as manual tools. Maybe add a `scripts/README.md` documenting what each does and when to run.
 4. **Promote `kb_measure.py` and `kb_staffer_report.py`** to be actually callable from the KB flow — they're documented but only run manually after scenario batches.
 ## Honest next step
 The cleanup above isn't this session's work — it's a subsequent PR. This doc lands first so the state is visible and operator-auditable.
@@ -37,7 +37,11 @@ interface ObservedOp {
  // Phase 24 — optional provenance so error analyzer and playbook
  // builder can differentiate MCP-layer ops from scenario-sourced
  // events. Scenarios set source="scenario" + staffer_id + sig_hash.
-  source?: "mcp" | "scenario";
+  // Phase A (cohesion plan 2026-04-22) — auditor verdicts also post
  // to :3800/event with source="auditor"; stats endpoint counts them
  // separately so KB readers can see audit cadence alongside
  // scenario cadence.
  source?: "mcp" | "scenario" | "auditor" | "bot";
  staffer_id?: string;
  sig_hash?: string;
  event_kind?: string;
@@ -253,15 +257,30 @@ function startHttpListener() {
      }
      if (req.method === "POST" && url.pathname === "/event") {
        return req.json().then((body: any) => {
          // Accept caller-provided source when present and valid.
          // Unrecognized sources fall back to "scenario" for
          // backward-compat with older callers.
          const allowedSource = ["mcp", "scenario", "auditor", "bot"] as const;
          const bodySrc = typeof body.source === "string" ? body.source : "";
          const source = (allowedSource as readonly string[]).includes(bodySrc)
            ? (bodySrc as typeof allowedSource[number])
            : "scenario";
          // Phase A: auditor+bot POST with ok=true|false (not success)
          // for symmetry with the kind_of of their domain. Accept either.
          const success = typeof body.success === "boolean" ? body.success
                        : typeof body.ok === "boolean" ? body.ok
                        : false;
          const op: ObservedOp = {
            timestamp: body.timestamp ?? new Date().toISOString(),
-            endpoint: body.endpoint ?? "scenario:fill",
+            endpoint: body.endpoint ?? (source === "auditor" ? "auditor:verdict"
                                       : source === "bot" ? "bot:cycle"
                                       : "scenario:fill"),
            input_summary: body.input_summary ?? `${body.event_kind ?? "?"} ${body.role ?? "?"}×${body.count ?? "?"} in ${body.city ?? "?"}, ${body.state ?? "?"}`,
-            success: !!body.success,
+            success,
-            duration_ms: Number(body.duration_ms ?? 0),
+            duration_ms: Number(body.duration_ms ?? body.audit_duration_ms ?? 0),
-            output_summary: body.output_summary ?? (body.success ? "filled" : (body.error ?? "failed")),
+            output_summary: body.output_summary ?? body.one_liner ?? (success ? "ok" : (body.error ?? "failed")),
            error: body.error,
-            source: "scenario",
+            source,
            staffer_id: body.staffer_id,
            sig_hash: body.sig_hash,
            event_kind: body.event_kind,
@@ -1,528 +0,0 @@
 // 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); });
@@ -1,267 +0,0 @@
 // 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); });
@@ -1,443 +0,0 @@
 // 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); });