distillation: Phase 4 — dataset export layer

Build the contamination firewall: RAG, SFT, and Preference exporters
that turn scored evidence into clean training datasets without
leaking rejected, unvalidated, hallucinated, or provenance-free
records.

Files (8 new + 4 schema updates):
  scripts/distillation/quarantine.ts      shared QuarantineWriter, 11-reason taxonomy
  scripts/distillation/export_rag.ts      RAG exporter (--include-review opt-in)
  scripts/distillation/export_sft.ts      SFT exporter (--include-partial opt-in, SFT_NEVER constant)
  scripts/distillation/export_preference.ts preference exporter, same task_id pairing
  scripts/distillation/distill.ts         CLI dispatcher (build-evidence/score/export-*)
  tests/distillation/exports.test.ts      15 contamination-firewall tests
  reports/distillation/phase4-export-report.md  acceptance report

Schema field-name alignment with now.md:
  rag_sample.ts        +source_category, exported_at→created_at
  sft_sample.ts        +id, exported_at→created_at, partially_accepted at schema (CLI gates)
  preference_sample.ts +id, source_run_ids→chosen_run_id+rejected_run_id, +created_at

Test metrics: 117 distillation tests pass · 0 fail · 315 expects · 327ms

Real-data export run (1052 scored input rows):
  RAG:        446 exported (351 acc + 95 partial), 606 quarantined
  SFT:        351 exported (all 'accepted'),       701 quarantined
  Preference:  83 pairs exported,                   16 quarantined

CONTAMINATION FIREWALL — verified held on real data:
  - SFT output: 351/351 quality_score='accepted' (ZERO leaked)
  - RAG output: 351 acc + 95 partial (ZERO rejected leaked)
  - Preference: 0 self-pairs (chosen_run_id != rejected_run_id)
  - 536 rejected+needs_human_review records caught at unsafe_sft_category
    gate, exact match to scored-runs forbidden-category total

Defense in depth (the firewall is two layers, not one):
  1. Schema layer (Phase 1): SftSample.quality_score enum forbids
     rejected/needs_human at write time
  2. Exporter layer: SFT_NEVER constant in export_sft.ts checks
     category before synthesis. Even if synthesis produced a row
     with quality_score=rejected, validateSftSample would reject it.

Quarantine reasons (11): missing_provenance, missing_source_run_id,
empty_content, schema_violation, unsafe_sft_category,
unsafe_rag_category, invalid_preference_pairing,
hallucinated_file_path, duplicate_id, self_pairing,
category_disallowed.

Bug surfaced + fixed during testing: module-level evidenceCache
shared state across test runs (tests wipe TMP, cache holds stale
empty Map). Moved cache to per-call scope. Same pattern bit Phase 2
materializer would have hit if its tests had multiple runs sharing
state — preventive fix.

Pairing logic v1: same task_id with category gap. accepted×rejected
preferred, accepted×partially_accepted as fallback. MAX_PAIRS_PER_TASK=5
cap prevents one hot task from dominating. Future: cross-source
pairing (scrum_reviews chosen vs observer_reviews rejected on same
file) to grow dataset beyond 83.

CLI: ./scripts/distill.ts {build-evidence|score|export-rag|export-sft|export-preference|export-all|health}
Flags: --dry-run, --include-partial (SFT only), --include-review (RAG only)

Carry-overs to Phase 5 (Receipts Harness):
- Each exporter currently writes results but no per-stage receipt.json.
  Phase 5 wraps build_evidence_index + score_runs + export_* in a
  withReceipt() helper that captures git_sha + sha256 of inputs/outputs
  + record_counts + validation_pass.
- reports/distillation/latest.md aggregating most-recent run of each stage.

Carry-overs to Phase 3 v2:
- mode_experiments scoring (168 needs_human_review): derive markers from
  validation_results.grounded_fraction
- extraction-class JOIN: distilled_*/audit_facts/observer_escalations
  → JOIN to verdict-bearing parent by task_id

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

scripts/distillation/export_sft.ts

@@ -0,0 +1,303 @@
+// export_sft.ts — Phase 4b SFT dataset export. Strict no-leak gates.
+//
+// Default: only category=accepted ships.
+// --include-partial: category in {accepted, partially_accepted} ships.
+// rejected and needs_human_review NEVER ship — schema layer (Phase 1)
+// enforces this AND the exporter filters before validation. Defense
+// in depth.
+//
+// Each SFT row:
+//   instruction = the prompt the executor saw
+//   context     = retrieved context summary (matrix corpora used,
+//                 pathway fingerprints seen, file_path)
+//   response    = the executor's accepted output (evidence.text)
+//
+// Source restriction: SFT only takes records where evidence.text is a
+// real model output (model_role in {executor, applier, reviewer with
+// observer_verdict}). Pure-extraction rows lack a true "instruction"
+// and are quarantined as missing_source_run_id (since they're not
+// instruction→response shape).
+
+import { existsSync, readFileSync, readdirSync, mkdirSync, appendFileSync, statSync } from "node:fs";
+import { resolve, dirname } from "node:path";
+import {
+  SFT_SAMPLE_SCHEMA_VERSION, validateSftSample, type SftSample, type SftQualityScore,
+} from "../../auditor/schemas/distillation/sft_sample";
+import type { ScoredRun, ScoreCategory } from "../../auditor/schemas/distillation/scored_run";
+import type { EvidenceRecord } from "../../auditor/schemas/distillation/evidence_record";
+import { QuarantineWriter } from "./quarantine";
+
+export interface ExportSftOptions {
+  root: string;
+  recorded_at: string;
+  include_partial?: boolean;
+  dry_run?: boolean;
+}
+
+export interface ExportSftResult {
+  scored_files_read: number;
+  records_read: number;
+  records_exported: number;
+  records_quarantined: number;
+  output_path: string;
+  quarantine_summary: string;
+}
+
+const DEFAULT_ROOT = process.env.LH_DISTILL_ROOT ?? "/home/profit/lakehouse";
+
+// Hard non-negotiable: this set never expands. If you find yourself
+// adding "needs_human_review" or "rejected" here, stop — that's the
+// contamination the spec forbids.
+const SFT_NEVER: ScoreCategory[] = ["rejected", "needs_human_review"];
+
+function listScoredRunFiles(root: string): string[] {
+  const out: string[] = [];
+  const dir = resolve(root, "data/scored-runs");
+  if (!existsSync(dir)) return out;
+  for (const yyyy of readdirSync(dir).sort()) {
+    const yp = resolve(dir, yyyy);
+    if (!statSync(yp).isDirectory()) continue;
+    for (const mm of readdirSync(yp).sort()) {
+      const mp = resolve(yp, mm);
+      if (!statSync(mp).isDirectory()) continue;
+      for (const dd of readdirSync(mp).sort()) {
+        const dp = resolve(mp, dd);
+        if (!statSync(dp).isDirectory()) continue;
+        for (const f of readdirSync(dp)) {
+          if (f.endsWith(".jsonl")) out.push(resolve(dp, f));
+        }
+      }
+    }
+  }
+  return out;
+}
+
+function loadEvidenceByRunId(
+  scored_path: string,
+  cache: Map<string, Map<string, EvidenceRecord>>,
+): Map<string, EvidenceRecord> {
+  const evidence_path = scored_path.replace("/scored-runs/", "/evidence/");
+  if (cache.has(evidence_path)) return cache.get(evidence_path)!;
+  const m = new Map<string, EvidenceRecord>();
+  if (!existsSync(evidence_path)) { cache.set(evidence_path, m); return m; }
+  for (const line of readFileSync(evidence_path, "utf8").split("\n")) {
+    if (!line) continue;
+    try { const ev = JSON.parse(line) as EvidenceRecord; m.set(ev.run_id, ev); } catch {}
+  }
+  cache.set(evidence_path, m);
+  return m;
+}
+
+// Synthesize SFT shape from the executed run. For sources where text
+// isn't a model RESPONSE (pure-extraction), this returns null and the
+// caller quarantines.
+function synthesizeSft(
+  scored: ScoredRun,
+  ev: EvidenceRecord,
+  recorded_at: string,
+  sft_id: string,
+): SftSample | null {
+  const text = ev.text ?? "";
+  // Skip extraction-class records — they don't have an instruction→response shape.
+  const role = ev.model_role;
+  if (role !== "executor" && role !== "reviewer" && role !== "applier") return null;
+  if (text.trim().length === 0) return null;
+
+  // Instruction synthesis depends on the source class.
+  const stem = ev.provenance.source_file.replace(/^data\/_kb\//, "").replace(/\.jsonl$/, "");
+  let instruction = "";
+  switch (stem) {
+    case "scrum_reviews":
+      instruction = `Review the file '${ev.source_files?.[0] ?? "<file>"}' against the PRD + change-proposal context. Produce a forensic audit with findings, severity, confidence, patches.`;
+      break;
+    case "mode_experiments":
+      instruction = `Run task_class='${ev.task_id}' for file '${ev.source_files?.[0] ?? "<file>"}'. Produce the mode-runner's expected output shape.`;
+      break;
+    case "auto_apply":
+      instruction = `Auto-apply: emit a 6-line surgical patch for '${ev.source_files?.[0] ?? "<file>"}' based on the latest scrum review findings.`;
+      break;
+    case "audits":
+      instruction = `Audit phase '${ev.task_id.replace(/^phase:/, "")}' and report findings with severity.`;
+      break;
+    case "observer_reviews":
+      instruction = `Observer-review the latest attempt on '${ev.source_files?.[0] ?? "<file>"}'. Verdict: accept | reject | cycle.`;
+      break;
+    case "contract_analyses":
+      instruction = `Analyze contractor '${(ev as any).contractor ?? "<contractor>"}' for permit '${ev.task_id.replace(/^permit:/, "")}'. Recommend with risk markers.`;
+      break;
+    case "outcomes":
+      instruction = `Run scenario; report per-event outcome with citations.`;
+      break;
+    default:
+      instruction = `Source '${stem}' run; produce the appropriate output for this task type.`;
+  }
+
+  // Context — what the model could see. Keep terse.
+  const ctxParts: string[] = [];
+  if (ev.retrieved_context?.matrix_corpora?.length) {
+    ctxParts.push(`matrix=${ev.retrieved_context.matrix_corpora.join(",")}`);
+  }
+  if (typeof ev.retrieved_context?.pathway_fingerprints_seen === "number") {
+    ctxParts.push(`pathway_fingerprints=${ev.retrieved_context.pathway_fingerprints_seen}`);
+  }
+  if (ev.model_name) ctxParts.push(`model=${ev.model_name}`);
+  const context = ctxParts.join(" · ");
+
+  return {
+    schema_version: SFT_SAMPLE_SCHEMA_VERSION,
+    id: sft_id,
+    instruction,
+    context,
+    response: text,
+    source_run_id: scored.evidence_run_id,
+    quality_score: scored.category as SftQualityScore,
+    created_at: recorded_at,
+    provenance: {
+      source_file: scored.provenance.source_file,
+      line_offset: scored.provenance.line_offset,
+      sig_hash: scored.provenance.sig_hash,
+      recorded_at,
+    },
+  };
+}
+
+export async function exportSft(opts: ExportSftOptions): Promise<ExportSftResult> {
+  const { root, recorded_at, include_partial = false, dry_run = false } = opts;
+  const allowed: ScoreCategory[] = include_partial
+    ? ["accepted", "partially_accepted"]
+    : ["accepted"];
+  const out_path = resolve(root, "exports/sft/instruction_response.jsonl");
+  const q = new QuarantineWriter(root, "sft", dry_run);
+
+  let records_read = 0;
+  let records_exported = 0;
+  const seenIds = new Set<string>();
+  const rowsToWrite: string[] = [];
+
+  if (!dry_run && existsSync(out_path)) {
+    for (const line of readFileSync(out_path, "utf8").split("\n")) {
+      if (!line) continue;
+      try { const r = JSON.parse(line); if (r.id) seenIds.add(r.id); } catch {}
+    }
+  }
+
+  const evidenceCache = new Map<string, Map<string, EvidenceRecord>>();
+  const scored_files = listScoredRunFiles(root);
+  for (const sp of scored_files) {
+    const evMap = loadEvidenceByRunId(sp, evidenceCache);
+    const lines = readFileSync(sp, "utf8").split("\n").filter(Boolean);
+    for (let i = 0; i < lines.length; i++) {
+      records_read++;
+      let scored: ScoredRun;
+      try { scored = JSON.parse(lines[i]) as ScoredRun; }
+      catch (e) {
+        q.add({
+          reason: "schema_violation",
+          source_record: { _raw: lines[i].slice(0, 200) },
+          errors: ["scored-run not JSON: " + (e as Error).message.slice(0, 160)],
+          recorded_at,
+        });
+        continue;
+      }
+
+      // CONTAMINATION FIREWALL: any forbidden category goes straight
+      // to quarantine, never reaches the synthesizer.
+      if (SFT_NEVER.includes(scored.category)) {
+        q.add({
+          reason: "unsafe_sft_category",
+          source_record: scored as unknown as Record<string, unknown>,
+          errors: [`category=${scored.category} forbidden in SFT (spec non-negotiable)`],
+          recorded_at,
+          source_provenance: scored.provenance,
+        });
+        continue;
+      }
+      if (!allowed.includes(scored.category)) {
+        q.add({
+          reason: "category_disallowed",
+          source_record: scored as unknown as Record<string, unknown>,
+          errors: [`category=${scored.category} not in [${allowed.join(",")}] (--include-partial=${include_partial})`],
+          recorded_at,
+          source_provenance: scored.provenance,
+        });
+        continue;
+      }
+      if (!scored.provenance?.sig_hash) {
+        q.add({ reason: "missing_provenance", source_record: scored as any, errors: ["provenance missing"], recorded_at });
+        continue;
+      }
+      if (!scored.evidence_run_id) {
+        q.add({ reason: "missing_source_run_id", source_record: scored as any, errors: ["evidence_run_id missing"], recorded_at, source_provenance: scored.provenance });
+        continue;
+      }
+
+      const ev = evMap.get(scored.evidence_run_id);
+      if (!ev) {
+        q.add({ reason: "missing_source_run_id", source_record: scored as any, errors: [`evidence_run_id=${scored.evidence_run_id} not found`], recorded_at, source_provenance: scored.provenance });
+        continue;
+      }
+
+      // ID = sha256(evidence_run_id + sig_hash):16
+      const id_seed = `${scored.evidence_run_id}|${scored.provenance.sig_hash}`;
+      const hasher = new Bun.CryptoHasher("sha256");
+      hasher.update(id_seed);
+      const sft_id = "sft-" + hasher.digest("hex").slice(0, 16);
+      if (seenIds.has(sft_id)) continue;
+
+      const sample = synthesizeSft(scored, ev, recorded_at, sft_id);
+      if (!sample) {
+        q.add({
+          reason: "missing_source_run_id",
+          source_record: { run_id: scored.evidence_run_id, model_role: ev.model_role, has_text: typeof ev.text === "string" && ev.text.length > 0 },
+          errors: ["evidence has no instruction→response shape (extraction-class or empty text)"],
+          recorded_at,
+          source_provenance: scored.provenance,
+        });
+        continue;
+      }
+
+      const v = validateSftSample(sample);
+      if (!v.valid) {
+        q.add({
+          reason: "schema_violation",
+          source_record: sample as unknown as Record<string, unknown>,
+          errors: v.errors,
+          recorded_at,
+          source_provenance: scored.provenance,
+        });
+        continue;
+      }
+
+      seenIds.add(sft_id);
+      rowsToWrite.push(JSON.stringify(v.value));
+      records_exported++;
+    }
+  }
+
+  if (!dry_run && rowsToWrite.length > 0) {
+    mkdirSync(dirname(out_path), { recursive: true });
+    appendFileSync(out_path, rowsToWrite.join("\n") + "\n");
+  }
+
+  return {
+    scored_files_read: scored_files.length,
+    records_read,
+    records_exported,
+    records_quarantined: q.total,
+    output_path: out_path.replace(root + "/", ""),
+    quarantine_summary: q.summary(),
+  };
+}
+
+async function cli() {
+  const dry_run = process.argv.includes("--dry-run");
+  const include_partial = process.argv.includes("--include-partial");
+  const recorded_at = new Date().toISOString();
+  const r = await exportSft({ root: DEFAULT_ROOT, recorded_at, include_partial, dry_run });
+
+  console.log(`[export_sft] read=${r.records_read} exported=${r.records_exported} ${r.quarantine_summary}${dry_run ? " (DRY RUN)" : ""}`);
+  console.log(`[export_sft] output: ${r.output_path}`);
+  if (include_partial) console.log("[export_sft] partially_accepted INCLUDED (--include-partial)");
+}
+
+if (import.meta.main) cli().catch(e => { console.error(e); process.exit(1); });