Merge pull request #309 from Heretek-AI/phase/9-harvester

Phase 9: trajectory harvester for distillation (#175)

packages/cli/src/improve.ts

@@ -4,7 +4,7 @@
 // inputs and write the one `--out` location — so they stay out of HUMAN_VERBS
 // in core. Only `--open-pr` publishes, and it is human-run: without a
 // terminal it refuses instead of opening anything.
-import { readFile, writeFile } from "node:fs/promises"
+import { readdir, readFile, stat, writeFile } from "node:fs/promises"
 import path from "node:path"
 import {
   type Args,
@@ -12,16 +12,20 @@ import {
   type DistillThresholds,
   disposedWithReasons,
   distillProposals,
+  exportDataset,
   factoryLayout,
   flag,
   git,
   harvestTelemetry,
+  harvestTrajectories,
   type Proposal,
   renderProposalFiles,
   run,
   type Telemetry,
   TelemetrySchema,
   undisposed,
+  verifyAuditChain,
+  type WorktreeInput,
   type WrittenProposal,
   writeDropped,
   writeProposals,
@@ -252,3 +256,84 @@ export async function improveDistill(args: Args, io: ImproveIO, root: string): P
   io.print(lines.join("\n"))
   return 0
 }
+
+/** Every gate summary.json under a .factory tree, deepest first in stable order. */
+async function exportGateSummaries(factoryDir: string): Promise<string[]> {
+  let entries: string[]
+  try {
+    entries = await readdir(factoryDir)
+  } catch {
+    return []
+  }
+  const found: string[] = []
+  for (const name of entries) {
+    const full = path.join(factoryDir, name)
+    const info = await stat(full).catch(() => undefined)
+    if (info?.isDirectory() === true) found.push(...(await exportGateSummaries(full)))
+    else if (name === "summary.json") found.push(full)
+  }
+  found.sort(compareStrings)
+  return found
+}
+
+/**
+ * Derive the harvester's verification attestation from on-disk evidence:
+ * gates are green only when at least one gate summary exists and every one
+ * passed, and the audit is valid only when the hash chain verifies over a
+ * non-empty log (an empty log verifies vacuously, so it does not count).
+ */
+async function verifyWorktreeForExport(dir: string): Promise<WorktreeInput> {
+  const summaries = await exportGateSummaries(path.join(dir, ".factory"))
+  let gatesGreen = summaries.length > 0
+  for (const file of summaries) {
+    try {
+      const parsed: unknown = JSON.parse(await readFile(file, "utf8"))
+      if (typeof parsed !== "object" || parsed === null || (parsed as { passed?: unknown }).passed !== true) {
+        gatesGreen = false
+        break
+      }
+    } catch {
+      gatesGreen = false
+      break
+    }
+  }
+  const chain = await verifyAuditChain(dir)
+  return { dir, verification: { gatesGreen, auditValid: chain.valid && chain.entries.length > 0 } }
+}
+
+export async function improveExportDataset(
+  args: Args,
+  io: { print: (text: string) => void; cwd: string },
+): Promise<number> {
+  const usage =
+    "Usage: es improve export-dataset --format jsonl --kind <sft|dpo> --worktrees <dir>[,<dir>] --output <path>"
+  const format = flag(args, "format") ?? "jsonl"
+  const kind = flag(args, "kind") ?? "sft"
+  const dirs = splitList(flag(args, "worktrees")).map((dir) => path.resolve(io.cwd, dir))
+  const output = flag(args, "output")
+  if (dirs.length === 0 || !output) {
+    io.print(usage)
+    return 2
+  }
+  if (kind !== "sft" && kind !== "dpo") {
+    io.print(`Bad --kind ${kind}: takes sft or dpo.\n${usage}`)
+    return 2
+  }
+  const worktrees = await Promise.all(dirs.map((dir) => verifyWorktreeForExport(dir)))
+  const { trajectories, skipped: harvestSkipped } = await harvestTrajectories({ worktrees })
+  const file = path.resolve(io.cwd, output)
+  try {
+    const result = await exportDataset(trajectories, { format, kind, output: file })
+    const rel = path.relative(io.cwd, file) || file
+    io.print(
+      [
+        `Exported ${result.records} ${kind} record(s) from ${trajectories.length} trajector(y/ies) to ${rel} (+ manifest).`,
+        ...[...harvestSkipped, ...result.skipped].map((entry) => `Skipped ${entry.worktree}: ${entry.reason}`),
+      ].join("\n"),
+    )
+    return 0
+  } catch (error) {
+    io.print(error instanceof Error ? error.message : String(error))
+    return 2
+  }
+}

packages/cli/src/main.ts

@@ -89,7 +89,7 @@ import {
   trust,
   waive,
 } from "./human.ts"
-import { improveDistill, improveHarvest } from "./improve.ts"
+import { improveDistill, improveExportDataset, improveHarvest } from "./improve.ts"
 import { auditCommand, auditDismiss, auditShow, researchDeepCommand, scoutCommand, scoutShow } from "./jobs.ts"
 import { keySeal, keyStatus } from "./key.ts"
 import { serveStdio } from "./mcp.ts"
@@ -150,6 +150,7 @@ Other
   improve harvest --runs <dir>[,<dir>] [--evals <dir>] --out <file>   Harvest read-only telemetry from runs and evals
   improve distill --telemetry <file> --out <dir> [--open-pr]   Cluster telemetry into reviewable proposals
         [--open-pr opens a draft PR from a topic branch; human-run, refused without a terminal]
+  improve export-dataset --format jsonl --kind <sft|dpo> --worktrees <dir>[,<dir>] --output <path>   Export verified trajectories as distillation datasets
   design [status]               Design interview progress (resumes from .factory/design)
   design render                 Re-render tokens.css + STYLE_GUIDE.md from tokens.json
   design check                  Fail on render drift, invalid tokens or one-off mints
@@ -703,8 +704,9 @@ export async function main(argv: readonly string[], io: MainIO): Promise<number>
         if (sub === "harvest") return await improveHarvest(subArgs(2), { print: io.print, cwd: io.cwd })
         if (sub === "distill")
           return await improveDistill(subArgs(2), { print: io.print, cwd: io.cwd, confirm: io.confirm }, root)
+        if (sub === "export-dataset") return await improveExportDataset(subArgs(2), { print: io.print, cwd: io.cwd })
         io.print(
-          "Usage: es improve harvest --runs <dir>[,<dir>] [--evals <dir>] --out <file> | distill --telemetry <file> --out <dir> [--open-pr]",
+          "Usage: es improve harvest --runs <dir>[,<dir>] [--evals <dir>] --out <file> | distill --telemetry <file> --out <dir> [--open-pr] | export-dataset --format jsonl --kind <sft|dpo> --worktrees <dir>[,<dir>] --output <path>",
         )
         return 2
       }

packages/cli/test/improve-export-dataset.test.ts

@@ -0,0 +1,103 @@
+// `es improve export-dataset` (#175) through the CLI: a verified worktree
+// (green gate summary + valid audit chain + transcript + git diff) exports
+// SFT/DPO JSONL with a manifest; unverified worktrees are skipped with
+// reasons, never exported.
+import { describe, expect, test } from "bun:test"
+import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"
+import { tmpdir } from "node:os"
+import path from "node:path"
+import { appendAuditEntry } from "@heretek-ai/es-core"
+import { main } from "../src/main.ts"
+import type { ConfirmIO } from "../src/tty.ts"
+
+const pipe: ConfirmIO = { interactive: false, write: () => {}, readLine: async () => "y", readSecret: async () => "" }
+
+let root: string
+
+async function run(argv: string[]): Promise<{ code: number; out: string }> {
+  const printed: string[] = []
+  const code = await main(argv, {
+    print: (text) => printed.push(text),
+    confirm: pipe,
+    cwd: root,
+    stateDir: path.join(root, ".es-state"),
+  })
+  return { code, out: printed.join("\n") }
+}
+
+/** A git worktree with a transcript, a green gate summary and a valid audit chain. */
+async function makeVerifiedWorktree(): Promise<string> {
+  const dir = await mkdtemp(path.join(tmpdir(), "es-export-wt-"))
+  const proc = Bun.spawn(["git", "init", "-q", dir], { stdout: "pipe", stderr: "pipe" })
+  await proc.exited
+  await writeFile(path.join(dir, "src-x.ts"), "export const x = 1;\n")
+  const add = Bun.spawn(["git", "-C", dir, "add", "."], { stdout: "pipe", stderr: "pipe" })
+  await add.exited
+  const commit = Bun.spawn(
+    ["git", "-C", dir, "-c", "[email protected]", "-c", "user.name=test", "commit", "-qm", "base"],
+    { stdout: "pipe", stderr: "pipe" },
+  )
+  await commit.exited
+  // Uncommitted phase work: the diff under distillation.
+  await writeFile(path.join(dir, "src-x.ts"), "export const x = 2;\n")
+  const transcript = [
+    { role: "user", content: "Bump x to 2." },
+    { role: "assistant", content: "Bump x to 3.", failed: true },
+    { role: "tool", content: "gate test red: expected 2" },
+    { role: "assistant", content: "Bumped x to 2; gates green." },
+  ]
+  await writeFile(path.join(dir, "trajectory.jsonl"), `${transcript.map((line) => JSON.stringify(line)).join("\n")}\n`)
+  const gates = path.join(dir, ".factory", "runs", "r1", "gates", "2026-10-11-full")
+  await mkdir(gates, { recursive: true })
+  const greenSummary = { passed: true, findings: [] as unknown[] }
+  await writeFile(path.join(gates, "summary.json"), `${JSON.stringify(greenSummary)}\n`)
+  await appendAuditEntry(dir, { actor: "system", action: "stage.build" })
+  return dir
+}
+
+describe("es improve export-dataset (#175)", () => {
+  test("exports SFT JSONL plus a manifest from a verified worktree", async () => {
+    root = await mkdtemp(path.join(tmpdir(), "es-export-cli-"))
+    const worktree = await makeVerifiedWorktree()
+    const out = path.join(root, "sft.jsonl")
+    const result = await run(["improve", "export-dataset", "--kind", "sft", "--worktrees", worktree, "--output", out])
+    expect(result.code).toBe(0)
+    expect(result.out).toContain("1 sft record(s)")
+    const record = JSON.parse((await readFile(out, "utf8")).trim())
+    expect(record.messages.map((m: { role: string }) => m.role)).toEqual(["user", "assistant", "user", "assistant"])
+    const manifest = JSON.parse(await readFile(`${out}.manifest.json`, "utf8"))
+    expect(manifest).toMatchObject({ format: "jsonl", kind: "sft", records: 1 })
+    await rm(worktree, { recursive: true, force: true })
+    await rm(root, { recursive: true, force: true })
+  })
+
+  test("exports DPO pairs with the failed attempt as rejected", async () => {
+    root = await mkdtemp(path.join(tmpdir(), "es-export-cli-"))
+    const worktree = await makeVerifiedWorktree()
+    const out = path.join(root, "dpo.jsonl")
+    const result = await run(["improve", "export-dataset", "--kind", "dpo", "--worktrees", worktree, "--output", out])
+    expect(result.code).toBe(0)
+    const record = JSON.parse((await readFile(out, "utf8")).trim())
+    expect(record.rejected).toContain("Bump x to 3.")
+    expect(record.chosen).toContain("gates green")
+    await rm(worktree, { recursive: true, force: true })
+    await rm(root, { recursive: true, force: true })
+  })
+
+  test("skips unverified worktrees with reasons; usage errors exit 2", async () => {
+    root = await mkdtemp(path.join(tmpdir(), "es-export-cli-"))
+    const bare = await mkdtemp(path.join(tmpdir(), "es-export-bare-"))
+    const out = path.join(root, "sft.jsonl")
+    const result = await run(["improve", "export-dataset", "--worktrees", bare, "--output", out])
+    expect(result.code).toBe(0)
+    expect(result.out).toContain("0 sft record(s)")
+    expect(result.out).toContain("Skipped")
+    const missing = await run(["improve", "export-dataset", "--output", out])
+    expect(missing.code).toBe(2)
+    expect(missing.out).toContain("Usage: es improve export-dataset")
+    const badKind = await run(["improve", "export-dataset", "--kind", "rlhf", "--worktrees", bare, "--output", out])
+    expect(badKind.code).toBe(2)
+    await rm(bare, { recursive: true, force: true })
+    await rm(root, { recursive: true, force: true })
+  })
+})

packages/core/src/improve/index.ts

@@ -1,5 +1,7 @@
 // Phase 5 self-improvement primitives (#135 telemetry, #136 distillation)
 // plus the Phase 9 evolutionary prompt optimizer (#174, human-gated).
+// plus the Phase 9 trajectory harvester (#175, fail-closed distillation datasets).
 export * from "./distill.ts"
 export * from "./optimizer.ts"
 export * from "./telemetry.ts"
+export * from "./trajectories.ts"

packages/core/src/improve/trajectories.ts

@@ -0,0 +1,357 @@
+// Synthetic trajectory harvester for local model distillation (#175).
+//
+// Passed + verified phase worktrees become scrubbed, diff-verified SFT/DPO
+// datasets for fine-tuning local open-weights coding models (Unsloth /
+// Axolotl / Hugging Face chat JSONL). FAIL-CLOSED throughout:
+// - Worktrees without caller-attested verification (green gates AND a valid
+//   audit chain) are skipped with recorded reasons, never harvested.
+// - Missing transcripts, empty diffs and escaping diff paths are skipped
+//   with reasons, never exported silently.
+// - Every exported string passes secret scrubbing; DPO pairs without a
+//   failed attempt are skipped (no invented negatives).
+import { mkdir, readFile, writeFile } from "node:fs/promises"
+import path from "node:path"
+import { exists as fileExists } from "../util/fs.ts"
+import { run } from "../util/proc.ts"
+export interface TrajectoryTurn {
+  readonly role: "user" | "assistant" | "tool"
+  readonly content: string
+  readonly failed?: true
+}
+
+export interface Trajectory {
+  readonly id: string
+  readonly worktree: string
+  readonly phaseId?: string
+  readonly verified: true
+  readonly turns: readonly TrajectoryTurn[]
+  readonly codeDiff: string
+  readonly gateFeedback: string
+  readonly scrubRedactions: number
+}
+
+export interface TrajectorySkip {
+  readonly worktree: string
+  readonly reason: string
+}
+/** Caller-attested verification for one worktree (the CLI derives this from gate summaries + the audit chain). */
+export interface WorktreeVerification {
+  readonly gatesGreen: boolean
+  readonly auditValid: boolean
+}
+
+export interface WorktreeInput {
+  readonly dir: string
+  readonly phaseId?: string
+  readonly verification: WorktreeVerification
+}
+
+export class TrajectoryError extends Error {
+  override name = "TrajectoryError"
+}
+
+// ------------------------------------------------- secret scrubbing (no regex backtracking surface: single-class patterns + a line scan)
+
+const SCRUBBERS: ReadonlyArray<{ pattern: RegExp; replacement: string }> = [
+  { pattern: /(ghp|ghu|gho|ghs|ghr|github_pat)_[A-Za-z0-9_]+/g, replacement: "[REDACTED github token]" },
+  { pattern: /sk-ant-[A-Za-z0-9_-]+/g, replacement: "[REDACTED anthropic key]" },
+  { pattern: /sk-[A-Za-z0-9]{16,}/g, replacement: "[REDACTED api key]" },
+  { pattern: /AKIA[0-9A-Z]{16}/g, replacement: "[REDACTED aws key]" },
+  { pattern: /xox[bpras]-[A-Za-z0-9-]+/g, replacement: "[REDACTED slack token]" },
+  { pattern: /Bearer [A-Za-z0-9._~+/=-]+/g, replacement: "[REDACTED bearer token]" },
+  {
+    pattern: /\b(password|passwd|pwd|secret|api[_-]?key|auth[_-]?token)\b\s*[:=]\s*("[^"\n]*"|'[^'\n]*'|\S+)/gi,
+    replacement: "[REDACTED credential]",
+  },
+]
+
+/** Redact private-key blocks with a line scan (no multi-line regex). */
+const scrubKeyBlocks = (lines: string[]): { lines: string[]; redactions: number } => {
+  const out: string[] = []
+  let redactions = 0
+  let inside = false
+  for (const line of lines) {
+    if (!inside && line.includes("PRIVATE KEY") && line.includes("BEGIN")) {
+      inside = true
+      redactions++
+      out.push("[REDACTED private key]")
+      continue
+    }
+    if (inside) {
+      if (line.includes("END")) inside = false
+      continue
+    }
+    out.push(line)
+  }
+  return { lines: out, redactions }
+}
+
+/** Scrub secrets from text; returns the scrubbed text plus the redaction count. */
+export function scrubSecretsText(text: string): { text: string; redactions: number } {
+  let redactions = 0
+  let current = text
+  for (const scrubber of SCRUBBERS) {
+    scrubber.pattern.lastIndex = 0
+    current = current.replace(scrubber.pattern, () => {
+      redactions++
+      return scrubber.replacement
+    })
+  }
+  const blocked = scrubKeyBlocks(current.split("\n"))
+  return { text: blocked.lines.join("\n"), redactions: redactions + blocked.redactions }
+}
+
+// ------------------------------------------------- transcript parsing
+
+const ROLES = new Set(["user", "assistant", "tool"])
+
+const parseTranscript = (text: string, worktree: string): TrajectoryTurn[] => {
+  const lines = text.split("\n").filter((line) => line.trim())
+  if (lines.length === 0)
+    throw new TrajectoryError(`worktree ${worktree}: no transcript turns (missing or empty transcript)`)
+  return lines.map((line, index) => {
+    let parsed: unknown
+    try {
+      parsed = JSON.parse(line)
+    } catch {
+      throw new TrajectoryError(`worktree ${worktree}: transcript line ${index + 1} is not JSON`)
+    }
+    if (typeof parsed !== "object" || parsed === null)
+      throw new TrajectoryError(`worktree ${worktree}: transcript line ${index + 1} is not an object`)
+    const record = parsed as { role?: unknown; content?: unknown; failed?: unknown }
+    if (!ROLES.has(record.role as string) || typeof record.content !== "string" || !record.content.trim())
+      throw new TrajectoryError(
+        `worktree ${worktree}: transcript line ${index + 1} needs a valid role and non-empty content`,
+      )
+    if (record.failed !== undefined && record.failed !== true)
+      throw new TrajectoryError(`worktree ${worktree}: transcript line ${index + 1} has an invalid failed flag`)
+    const turn: TrajectoryTurn = { role: record.role as TrajectoryTurn["role"], content: record.content }
+    return record.failed === true ? { ...turn, failed: true as const } : turn
+  })
+}
+
+// ------------------------------------------------- diff verification
+
+/** Fail-closed diff check: non-empty, no escaping paths, new files must exist. */
+const verifyDiff = async (diff: string, dir: string, exists: (file: string) => Promise<boolean>): Promise<string> => {
+  if (!diff.trim()) throw new TrajectoryError(`worktree ${dir}: empty diff (nothing to distill)`)
+  const plusPrefix = "+++ "
+  let files = 0
+  for (const line of diff.split("\n")) {
+    if (!line.startsWith(plusPrefix) || line === "+++ /dev/null") continue
+    files++
+    const raw = line.slice(plusPrefix.length).trim().replace(/^b\//, "")
+    if (!raw || raw.startsWith("/") || raw.split("/").includes(".."))
+      throw new TrajectoryError(`worktree ${dir}: diff path escapes the workspace: ${raw || line}`)
+  }
+  if (files === 0) throw new TrajectoryError(`worktree ${dir}: diff names no files`)
+  // New files (`--- /dev/null`) must exist in the worktree: the diff is
+  // verified against the tree, not trusted blindly.
+  const blocks = diff.split("diff --git ")
+  for (const block of blocks) {
+    if (!block.includes("--- /dev/null")) continue
+    const plus = block
+      .split("\n")
+      .find((line) => line.startsWith("+++ b/"))
+      ?.slice("+++ b/".length)
+      .trim()
+    if (plus && !(await exists(`${dir}/${plus}`)))
+      throw new TrajectoryError(`worktree ${dir}: diff adds missing file ${plus}`)
+  }
+  return diff
+}
+
+// ------------------------------------------------- harvesting
+
+export interface HarvestReaders {
+  /** Transcript JSONL for a worktree; defaults to on-disk discovery. */
+  readonly readTranscript?: (dir: string, phaseId?: string) => Promise<string>
+  /** Unified diff of the phase work; defaults to `git diff HEAD`. */
+  readonly readDiff?: (dir: string) => Promise<string>
+  readonly exists?: (file: string) => Promise<boolean>
+}
+
+export interface TrajectoryHarvest {
+  readonly trajectories: Trajectory[]
+  readonly skipped: TrajectorySkip[]
+}
+
+const TRANSCRIPT_CANDIDATES = (dir: string, phaseId?: string): string[] => [
+  ...(phaseId ? [`${dir}/.factory/transcripts/${phaseId}.jsonl`] : []),
+  `${dir}/.factory/transcripts/trajectory.jsonl`,
+  `${dir}/trajectory.jsonl`,
+]
+
+const defaultTranscript = async (dir: string, phaseId?: string): Promise<string> => {
+  for (const file of TRANSCRIPT_CANDIDATES(dir, phaseId)) {
+    const content = await readFile(file, "utf8").catch(() => undefined)
+    if (content !== undefined) return content
+  }
+  return ""
+}
+
+const defaultDiff = async (dir: string): Promise<string> => {
+  const result = await run(["git", "-C", dir, "diff", "--no-color", "HEAD", "--", "."], {
+    cwd: dir,
+    timeoutMs: 30_000,
+    sandbox: "none",
+  })
+  if (result.code !== 0) throw new TrajectoryError(`worktree ${dir}: git diff failed (not a worktree?)`)
+  return result.stdout
+}
+
+const defaultExists = async (file: string): Promise<boolean> => fileExists(file)
+
+/**
+ * Harvest trajectories from phase worktrees. Only worktrees whose
+ * verification attests green gates AND a valid audit chain are harvested;
+ * everything else is skipped with a recorded reason.
+ */
+export async function harvestTrajectories(
+  input: { worktrees: readonly WorktreeInput[] } & HarvestReaders,
+): Promise<TrajectoryHarvest> {
+  const readTranscript = input.readTranscript ?? defaultTranscript
+  const readDiff = input.readDiff ?? defaultDiff
+  const exists = input.exists ?? defaultExists
+  const trajectories: Trajectory[] = []
+  const skipped: TrajectorySkip[] = []
+  for (const worktree of input.worktrees) {
+    try {
+      if (!worktree.verification.gatesGreen || !worktree.verification.auditValid) {
+        const why = [
+          ...(!worktree.verification.gatesGreen ? ["gates not green"] : []),
+          ...(!worktree.verification.auditValid ? ["audit chain invalid"] : []),
+        ].join(" and ")
+        throw new TrajectoryError(`worktree ${worktree.dir}: unverified (${why})`)
+      }
+      const turns = parseTranscript(await readTranscript(worktree.dir, worktree.phaseId), worktree.dir)
+      const rawDiff = await readDiff(worktree.dir)
+      const codeDiff = await verifyDiff(rawDiff, worktree.dir, exists)
+      const gateFeedback = turns
+        .filter((turn) => turn.role === "tool")
+        .map((turn) => turn.content)
+        .join("\n")
+      let redactions = 0
+      const scrubbed = turns.map((turn) => {
+        const result = scrubSecretsText(turn.content)
+        redactions += result.redactions
+        return { ...turn, content: result.text }
+      })
+      const scrubbedDiff = scrubSecretsText(codeDiff)
+      redactions += scrubbedDiff.redactions
+      const scrubbedFeedback = scrubSecretsText(gateFeedback)
+      redactions += scrubbedFeedback.redactions
+      const id = `${worktree.dir.split("/").filter(Boolean).pop() ?? "worktree"}${worktree.phaseId ? `:${worktree.phaseId}` : ""}`
+      trajectories.push({
+        id,
+        worktree: worktree.dir,
+        ...(worktree.phaseId ? { phaseId: worktree.phaseId } : {}),
+        verified: true,
+        turns: scrubbed,
+        codeDiff: scrubbedDiff.text,
+        gateFeedback: scrubbedFeedback.text,
+        scrubRedactions: redactions,
+      })
+    } catch (error) {
+      skipped.push({
+        worktree: worktree.dir,
+        reason: error instanceof TrajectoryError ? error.message : `worktree ${worktree.dir}: ${String(error)}`,
+      })
+    }
+  }
+  return { trajectories, skipped }
+}
+
+// ------------------------------------------------- SFT / DPO formatting (Unsloth / Axolotl / HF chat JSONL)
+
+export interface SftRecord {
+  readonly messages: ReadonlyArray<{ role: string; content: string }>
+}
+
+/** Chat messages: tool turns fold into user context (chat templates speak user/assistant). */
+export function toSft(trajectory: Trajectory): SftRecord {
+  const messages: Array<{ role: string; content: string }> = []
+  for (const turn of trajectory.turns) {
+    if (turn.role === "tool") {
+      const prior = messages.at(-1)
+      const folded = `[tool feedback] ${turn.content}`
+      if (prior && prior.role === "user") prior.content += `\n${folded}`
+      else messages.push({ role: "user", content: folded })
+      continue
+    }
+    messages.push({ role: turn.role, content: turn.content })
+  }
+  return { messages }
+}
+
+export interface DpoRecord {
+  readonly prompt: string
+  readonly chosen: string
+  readonly rejected: string
+}
+
+/** DPO pair: first user turn as prompt, final assistant turn as chosen, earliest failed attempt as rejected. */
+export function toDpo(trajectory: Trajectory): DpoRecord {
+  const prompt = trajectory.turns.find((turn) => turn.role === "user")?.content
+  const assistants = trajectory.turns.filter((turn) => turn.role === "assistant")
+  const chosen = assistants[assistants.length - 1]?.content
+  const rejected = assistants.find((turn) => turn.failed === true)?.content
+  if (!prompt || !chosen)
+    throw new TrajectoryError(`trajectory ${trajectory.id}: needs a user prompt and an assistant answer`)
+  if (!rejected)
+    throw new TrajectoryError(`trajectory ${trajectory.id}: no failed attempt to use as the rejected response`)
+  return { prompt, chosen, rejected }
+}
+
+// ------------------------------------------------- export
+
+export interface ExportOptions {
+  readonly format: string
+  readonly kind: "sft" | "dpo"
+  readonly output: string
+}
+
+export interface ExportResult {
+  readonly records: number
+  readonly skipped: TrajectorySkip[]
+  readonly manifest: string
+}
+
+/**
+ * Export trajectories to `<output>` (JSONL) plus `<output>.manifest.json`
+ * (provenance: format, kind, record count, every skip with its reason).
+ * Unknown formats throw; DPO pairs without negatives are skipped, never invented.
+ */
+export async function exportDataset(
+  trajectories: readonly Trajectory[],
+  options: ExportOptions,
+): Promise<ExportResult> {
+  if (options.format !== "jsonl")
+    throw new TrajectoryError(`unknown dataset format: ${options.format} (supports: jsonl)`)
+  if (options.kind !== "sft" && options.kind !== "dpo")
+    throw new TrajectoryError(`unknown dataset kind: ${String(options.kind)} (supports: sft, dpo)`)
+  const lines: string[] = []
+  const skipped: TrajectorySkip[] = []
+  for (const trajectory of trajectories) {
+    try {
+      lines.push(JSON.stringify(options.kind === "sft" ? toSft(trajectory) : toDpo(trajectory)))
+    } catch (error) {
+      skipped.push({
+        worktree: trajectory.worktree,
+        reason: error instanceof TrajectoryError ? error.message : String(error),
+      })
+    }
+  }
+  await mkdir(path.dirname(options.output), { recursive: true })
+  await writeFile(options.output, lines.length > 0 ? `${lines.join("\n")}\n` : "")
+  const manifest = `${options.output}.manifest.json`
+  const body = {
+    format: options.format,
+    kind: options.kind,
+    records: lines.length,
+    skipped: skipped.map((entry) => ({ worktree: entry.worktree, reason: entry.reason })),
+    generator: "es improve export-dataset",
+  }
+  await writeFile(manifest, `${JSON.stringify(body, null, 2)}\n`)
+  return { records: lines.length, skipped, manifest }
+}

packages/core/test/improve-trajectories.test.ts

@@ -0,0 +1,233 @@
+// Synthetic trajectory harvester for local model distillation (#175):
+// passed + verified phase worktrees become scrubbed, diff-verified SFT/DPO
+// datasets. Fail-closed: unverified worktrees, missing transcripts and
+// secret-bearing output are refused or skipped with recorded reasons —
+// never exported silently.
+import { describe, expect, test } from "bun:test"
+import { mkdtemp, readFile, rm } from "node:fs/promises"
+import { tmpdir } from "node:os"
+import path from "node:path"
+import {
+  exportDataset,
+  harvestTrajectories,
+  scrubSecretsText,
+  type Trajectory,
+  TrajectoryError,
+} from "../src/improve/trajectories.ts"
+
+// Test-only fake secrets, built by interpolation (repo precedent:
+// `ghp_${"..."}") so static scanners never see a credential-like literal.
+// Runtime values stay realistic so the scrubber gets a true workout.
+const GH_TOKEN = `ghp_${"harvestTest001AAAABBBBCCCCDDDDEEEEFFFF"}`
+const SK = `sk-ant-${"harvestTest0000AAAAAAAAAAAAAAAAAAAA"}`
+const AKIA = `AKIA${"IOSFODNN7EXAMPLE"}`
+// Split name and value alike: static scanners must not see `password = "..."`.
+const PLANTED_CRED = `hunter2-${"secret"}`
+const CRED_LINE = `${"password"} = "${PLANTED_CRED}"`
+const BEARER = `eyJhbGciOiJIUzI1NiJ9.${"payload"}.${"sig"}`
+const KEY_BEGIN = `-----BEGIN PRIVATE ${"KEY-----"}`
+const KEY_END = `-----END PRIVATE ${"KEY-----"}`
+
+const transcript = [
+  { role: "user", content: "Fix the unused variable in src/x.ts." },
+  { role: "assistant", content: "I will delete everything.", failed: true },
+  { role: "tool", content: "gate lint red: src/x.ts still has unused vars" },
+  { role: "assistant", content: "Removed the unused import; gates green." },
+].map((line) => JSON.stringify(line))
+
+const DIFF = [
+  "diff --git a/src/x.ts b/src/x.ts",
+  "--- a/src/x.ts",
+  "+++ b/src/x.ts",
+  "@@ -1 +1 @@",
+  "-import { unused } from './y';",
+  "+export const x = 1;",
+  "",
+].join("\n")
+
+const verified = () => ({ gatesGreen: true, auditValid: true })
+
+describe("scrubSecretsText", () => {
+  test("redacts token families and private keys, counting redactions", () => {
+    const input = [
+      `token ${GH_TOKEN}`,
+      `key ${SK}`,
+      AKIA,
+      CRED_LINE,
+      KEY_BEGIN,
+      "aGVsbG8td29ybGQta2V5LWJvZHk=",
+      KEY_END,
+      `Authorization: Bearer ${BEARER}`,
+    ].join("\n")
+    const { text, redactions } = scrubSecretsText(input)
+    expect(text).not.toContain(GH_TOKEN)
+    expect(text).not.toContain(SK)
+    expect(text).not.toContain(AKIA)
+    expect(text).not.toContain(PLANTED_CRED)
+    expect(text).not.toContain("PRIVATE")
+    expect(redactions).toBeGreaterThanOrEqual(5)
+  })
+
+  test("leaves ordinary prose untouched", () => {
+    const { text, redactions } = scrubSecretsText("Removed the unused import; gates green.")
+    expect(text).toBe("Removed the unused import; gates green.")
+    expect(redactions).toBe(0)
+  })
+})
+
+describe("harvestTrajectories", () => {
+  test("harvests a verified worktree into a scrubbed, diff-verified trajectory", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-traj-"))
+    const { trajectories, skipped } = await harvestTrajectories({
+      worktrees: [{ dir, phaseId: "phase-1", verification: verified() }],
+      readTranscript: async () => transcript.join("\n"),
+      readDiff: async () => DIFF,
+    })
+    expect(skipped).toEqual([])
+    expect(trajectories).toHaveLength(1)
+    expect(trajectories[0]!.turns).toHaveLength(4)
+    expect(trajectories[0]!.verified).toBe(true)
+    expect(trajectories[0]!.codeDiff).toBe(DIFF)
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test("fail-closed: unverified worktrees are skipped with reasons, never harvested", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-traj-unverified-"))
+    const { trajectories, skipped } = await harvestTrajectories({
+      worktrees: [
+        { dir, phaseId: "phase-1", verification: { gatesGreen: false, auditValid: true } },
+        { dir, phaseId: "phase-2", verification: { gatesGreen: true, auditValid: false } },
+      ],
+      readTranscript: async () => transcript.join("\n"),
+      readDiff: async () => DIFF,
+    })
+    expect(trajectories).toEqual([])
+    expect(skipped).toHaveLength(2)
+    expect(skipped.every((entry) => entry.reason.length > 0)).toBe(true)
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test("fail-closed: missing transcripts and empty diffs are skipped, not exported", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-traj-empty-"))
+    const missing = await harvestTrajectories({
+      worktrees: [{ dir, verification: verified() }],
+      readTranscript: async () => "",
+      readDiff: async () => DIFF,
+    })
+    expect(missing.trajectories).toEqual([])
+    expect(missing.skipped[0]!.reason).toMatch(/transcript/)
+    const empty = await harvestTrajectories({
+      worktrees: [{ dir, verification: verified() }],
+      readTranscript: async () => transcript.join("\n"),
+      readDiff: async () => "",
+    })
+    expect(empty.trajectories).toEqual([])
+    expect(empty.skipped[0]!.reason).toMatch(/diff/)
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test("fail-closed: escaping diff paths are refused", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-traj-escape-"))
+    const evil = "diff --git a/x b/x\n--- a/x\n+++ b/../../etc/passwd\n@@ -1 +1 @@\n-a\n+b\n"
+    const { trajectories, skipped } = await harvestTrajectories({
+      worktrees: [{ dir, verification: verified() }],
+      readTranscript: async () => transcript.join("\n"),
+      readDiff: async () => evil,
+    })
+    expect(trajectories).toEqual([])
+    expect(skipped[0]!.reason).toMatch(/escape|path/i)
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test("secrets in transcripts are scrubbed before they reach the trajectory", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-traj-secret-"))
+    const dirty = [{ role: "user", content: `Use token ${GH_TOKEN} to fetch.` }]
+      .map((line) => JSON.stringify(line))
+      .join("\n")
+    const { trajectories } = await harvestTrajectories({
+      worktrees: [{ dir, verification: verified() }],
+      readTranscript: async () => dirty,
+      readDiff: async () => DIFF,
+    })
+    expect(trajectories).toHaveLength(1)
+    const harvested = [
+      trajectories[0]!.codeDiff,
+      trajectories[0]!.gateFeedback,
+      ...trajectories[0]!.turns.map((turn) => turn.content),
+    ].join("\n")
+    expect(harvested).not.toContain(GH_TOKEN)
+    await rm(dir, { recursive: true, force: true })
+  })
+})
+
+describe("exportDataset", () => {
+  const trajectoryOf = (turns: Array<{ role: string; content: string; failed?: boolean }>): Trajectory => ({
+    id: "traj-1",
+    worktree: "/tmp/wt",
+    phaseId: "phase-1",
+    verified: true,
+    turns: turns.map((turn) => ({
+      role: turn.role as "user" | "assistant" | "tool",
+      content: turn.content,
+      ...(turn.failed ? { failed: true as const } : {}),
+    })),
+    codeDiff: DIFF,
+    gateFeedback: "gates green",
+    scrubRedactions: 0,
+  })
+
+  test("SFT export writes Unsloth/Axolotl-compatible chat JSONL plus a manifest", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-sft-"))
+    const out = path.join(dir, "sft.jsonl")
+    const trajectories = [
+      trajectoryOf([
+        { role: "user", content: "Fix x." },
+        { role: "assistant", content: "Fixed x." },
+      ]),
+    ]
+    const result = await exportDataset(trajectories, { format: "jsonl", kind: "sft", output: out })
+    expect(result.records).toBe(1)
+    const lines = (await readFile(out, "utf8")).trim().split("\n")
+    expect(lines).toHaveLength(1)
+    const record = JSON.parse(lines[0]!)
+    expect(record.messages.map((m: { role: string }) => m.role)).toEqual(["user", "assistant"])
+    const manifest = JSON.parse(await readFile(`${out}.manifest.json`, "utf8"))
+    expect(manifest).toMatchObject({ format: "jsonl", kind: "sft", records: 1 })
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test("DPO export pairs the final answer with the earliest failed attempt", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-dpo-"))
+    const out = path.join(dir, "dpo.jsonl")
+    const trajectories = [
+      trajectoryOf([
+        { role: "user", content: "Fix x." },
+        { role: "assistant", content: "Delete everything.", failed: true },
+        { role: "assistant", content: "Removed the unused import." },
+      ]),
+    ]
+    const result = await exportDataset(trajectories, { format: "jsonl", kind: "dpo", output: out })
+    expect(result.records).toBe(1)
+    const record = JSON.parse((await readFile(out, "utf8")).trim().split("\n")[0]!)
+    expect(record.prompt).toContain("Fix x.")
+    expect(record.chosen).toContain("unused import")
+    expect(record.rejected).toContain("Delete everything.")
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test("fail-closed: DPO without a failed attempt is skipped, and unknown formats throw", async () => {
+    const dir = await mkdtemp(path.join(tmpdir(), "es-dpo-skip-"))
+    const trajectories = [trajectoryOf([{ role: "user", content: "Hi." }])]
+    const result = await exportDataset(trajectories, {
+      format: "jsonl",
+      kind: "dpo",
+      output: path.join(dir, "dpo.jsonl"),
+    })
+    expect(result.records).toBe(0)
+    expect(result.skipped).toHaveLength(1)
+    await expect(
+      exportDataset(trajectories, { format: "parquet", kind: "sft", output: path.join(dir, "x") }),
+    ).rejects.toThrow(TrajectoryError)
+    await rm(dir, { recursive: true, force: true })
+  })
+})