distillation: full SFT export port — closes OPEN #2 fully
Follow-up to b216b7e (which shipped the SFT export substrate). This
commit ports the synthesis logic, completing the migration:
- SynthesizeSft(scored, ev, recordedAt, sftID) → *SftSample
Mirrors the Rust synthesizeSft byte-for-byte. Returns nil for
extraction-class records + empty-text records (same skip
semantics as Rust).
- LoadEvidenceByRunID(scoredPath, cache) reads the paired evidence
JSONL (path derived by /scored-runs/ → /evidence/ replacement).
Per-call cache so multiple scored-runs files in the same dir
don't reload the same evidence.
- buildInstruction maps source_file stem → per-class instruction
template. All 8 templates (scrum_reviews, mode_experiments,
auto_apply, audits, observer_reviews, contract_analyses,
outcomes, default) match Rust output exactly so a/b validation
between runtimes can diff JSONL byte-for-byte.
- stemFromSourceFile strips data/_kb/ prefix + .jsonl suffix.
- ExportSft now writes data/distilled/sft/sft_export.jsonl with
the synthesized samples (DryRun=true skips file write).
Per-class templates verified by 8-case sub-test:
- scrum_reviews → "Review the file '...' against the PRD..."
- mode_experiments → "Run task_class='...' for file..."
- auto_apply → "Auto-apply: emit a 6-line surgical patch..."
- audits with phase: prefix → strips to bare phase name
- observer_reviews → "Observer-review the latest attempt..."
- contract_analyses with permit: prefix → strips to permit ID
- outcomes → "Run scenario; report per-event outcome..."
- unknown source → "Source 'X' run; produce the appropriate output"
Caveat documented inline: contract_analyses uses ev.metadata.contractor
in Rust to produce "Analyze contractor 'X' for permit 'Y'" when
present. Go's EvidenceRecord doesn't carry a free-form metadata bag
yet, so we always emit the no-contractor form. Operators needing
contractor-aware instructions can extend EvidenceRecord with an
explicit Metadata field (separate ADR).
Test additions (5 new):
- TestSynthesizeSft_PerSourceClass: 8 sub-cases, one per template
- TestSynthesizeSft_RejectsExtraction: extraction-role records skipped
- TestSynthesizeSft_RejectsEmptyText: empty/whitespace text skipped
- TestSynthesizeSft_ContextAssembly: matrix + pathway + model
context string formatting matches Rust " · " join
- TestExportSft_FullPort_WritesJSONL: end-to-end fixture, asserts
output contains expected instruction + omits firewalled records
Pre-existing TestExportSft_PartialPort_FirewallFires renamed +
updated to TestExportSft_FirewallFiresBeforeEvidenceLoad — reflects
the new contract that records passing the firewall but lacking
evidence land in "not-instructable" rather than being silently
exported. Honest semantics shift documented in the test.
OPEN #2 now fully closed (was: substrate-only). The synthesis path
no longer requires the Rust pipeline to be invoked — Go-side
operators can run the full distillation export end-to-end.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
@@ -164,51 +164,236 @@ func LoadScoredRunsFromFile(path string) ([]ScoredRun, int, error) {
|
||||
return out, skipped, nil
|
||||
}
|
||||
|
||||
// ExportSft is the partial port. Lists scored-run files, loads
|
||||
// each, applies the contamination firewall, and reports counts.
|
||||
// What's NOT yet ported (deliberate, separate wave):
|
||||
// - Evidence-record loading + cache (loadEvidenceByRunId).
|
||||
// - synthesizeSft — the actual instruction/input/response
|
||||
// synthesis logic. ~80 lines of TS in scripts/distillation/export_sft.ts.
|
||||
// - Quarantine writer integration (write rejected records to
|
||||
// a quarantine JSONL for operator review).
|
||||
// - File output (write SFT JSONL to data/distilled/sft/).
|
||||
// LoadEvidenceByRunID reads {evidence_path}'s JSONL and returns a
|
||||
// map[run_id]EvidenceRecord. evidence_path is derived by replacing
|
||||
// /scored-runs/ with /evidence/ in the scored-runs path — the
|
||||
// convention the Rust pipeline uses to colocate evidence + scores.
|
||||
//
|
||||
// Returning a non-nil result with RecordsExported=0 is intentional
|
||||
// pre-synthesis — operators calling this on the Go side will see
|
||||
// the count of records that PASSED the firewall and would have
|
||||
// been exported by a complete implementation. RecordsQuarantined
|
||||
// reflects records BLOCKED by the firewall.
|
||||
// Cache is the per-call memoization map: callers iterating multiple
|
||||
// scored-runs files in the same directory benefit from shared
|
||||
// loads. Pass nil to disable caching (useful in tests).
|
||||
func LoadEvidenceByRunID(scoredPath string, cache map[string]map[string]EvidenceRecord) (map[string]EvidenceRecord, error) {
|
||||
evidencePath := strings.Replace(scoredPath, "/scored-runs/", "/evidence/", 1)
|
||||
if cache != nil {
|
||||
if hit, ok := cache[evidencePath]; ok {
|
||||
return hit, nil
|
||||
}
|
||||
}
|
||||
out := make(map[string]EvidenceRecord)
|
||||
data, err := os.ReadFile(evidencePath)
|
||||
if os.IsNotExist(err) {
|
||||
if cache != nil {
|
||||
cache[evidencePath] = out
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, line := range strings.Split(string(data), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var r EvidenceRecord
|
||||
if err := json.Unmarshal([]byte(line), &r); err != nil {
|
||||
continue // tolerate malformed lines per Rust convention
|
||||
}
|
||||
out[r.RunID] = r
|
||||
}
|
||||
if cache != nil {
|
||||
cache[evidencePath] = out
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SynthesizeSft turns a (ScoredRun, EvidenceRecord) pair into an
|
||||
// SftSample. Returns nil when the record is unsuitable for SFT:
|
||||
// - model_role isn't executor/reviewer/applier (extraction-class
|
||||
// records have no instruction→response shape)
|
||||
// - text is empty (nothing to learn from)
|
||||
//
|
||||
// Tests/contracts that synthesis port must preserve:
|
||||
// - SftNever firewall fires before any other validation
|
||||
// - Sort order matches Rust (file walk + record order within file)
|
||||
// - Empty root dir returns zero-counts, not error
|
||||
// Mirrors the Rust synthesizeSft exactly so byte-identical SFT
|
||||
// JSONL output is achievable across both runtimes (operators
|
||||
// running a/b validation between Rust + Go pipelines depend on this).
|
||||
func SynthesizeSft(scored ScoredRun, ev EvidenceRecord, recordedAt, sftID string) *SftSample {
|
||||
if ev.ModelRole != RoleExecutor && ev.ModelRole != RoleReviewer && ev.ModelRole != RoleApplier {
|
||||
return nil
|
||||
}
|
||||
text := ev.Text
|
||||
if strings.TrimSpace(text) == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
stem := stemFromSourceFile(ev.Provenance.SourceFile)
|
||||
firstSourceFile := ""
|
||||
if len(ev.SourceFiles) > 0 {
|
||||
firstSourceFile = ev.SourceFiles[0]
|
||||
}
|
||||
if firstSourceFile == "" {
|
||||
firstSourceFile = "<file>"
|
||||
}
|
||||
instruction := buildInstruction(stem, ev.TaskID, firstSourceFile)
|
||||
|
||||
// Context — what the model could see. Keep terse; matches Rust's
|
||||
// space-separated " · " join.
|
||||
var ctxParts []string
|
||||
if ev.RetrievedContext != nil {
|
||||
if len(ev.RetrievedContext.MatrixCorpora) > 0 {
|
||||
ctxParts = append(ctxParts, "matrix="+strings.Join(ev.RetrievedContext.MatrixCorpora, ","))
|
||||
}
|
||||
if ev.RetrievedContext.PathwayFingerprintsSeen > 0 {
|
||||
ctxParts = append(ctxParts, fmt.Sprintf("pathway_fingerprints=%d", ev.RetrievedContext.PathwayFingerprintsSeen))
|
||||
}
|
||||
}
|
||||
if ev.ModelName != "" {
|
||||
ctxParts = append(ctxParts, "model="+ev.ModelName)
|
||||
}
|
||||
context := strings.Join(ctxParts, " · ")
|
||||
|
||||
return &SftSample{
|
||||
SchemaVersion: SftSampleSchemaVersion,
|
||||
ID: sftID,
|
||||
Instruction: instruction,
|
||||
Context: context,
|
||||
Response: text,
|
||||
SourceRunID: scored.EvidenceRunID,
|
||||
QualityScore: SftQualityScore(scored.Category),
|
||||
CreatedAt: recordedAt,
|
||||
Provenance: Provenance{
|
||||
SourceFile: scored.Provenance.SourceFile,
|
||||
LineOffset: scored.Provenance.LineOffset,
|
||||
SigHash: scored.Provenance.SigHash,
|
||||
RecordedAt: recordedAt,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// stemFromSourceFile mirrors the Rust path-cleanup: strip the
|
||||
// `data/_kb/` prefix and the `.jsonl` suffix. Used to dispatch
|
||||
// the per-source-class instruction templates below.
|
||||
func stemFromSourceFile(path string) string {
|
||||
s := strings.TrimPrefix(path, "data/_kb/")
|
||||
s = strings.TrimSuffix(s, ".jsonl")
|
||||
return s
|
||||
}
|
||||
|
||||
// buildInstruction returns the per-source-class instruction
|
||||
// template. Mirrors the Rust switch statement byte-for-byte so
|
||||
// trained-model behavior is unchanged across runtimes.
|
||||
func buildInstruction(stem, taskID, firstSourceFile string) string {
|
||||
switch stem {
|
||||
case "scrum_reviews":
|
||||
return fmt.Sprintf("Review the file '%s' against the PRD + change-proposal context. Produce a forensic audit with findings, severity, confidence, patches.", firstSourceFile)
|
||||
case "mode_experiments":
|
||||
return fmt.Sprintf("Run task_class='%s' for file '%s'. Produce the mode-runner's expected output shape.", taskID, firstSourceFile)
|
||||
case "auto_apply":
|
||||
return fmt.Sprintf("Auto-apply: emit a 6-line surgical patch for '%s' based on the latest scrum review findings.", firstSourceFile)
|
||||
case "audits":
|
||||
phase := strings.TrimPrefix(taskID, "phase:")
|
||||
return fmt.Sprintf("Audit phase '%s' and report findings with severity.", phase)
|
||||
case "observer_reviews":
|
||||
return fmt.Sprintf("Observer-review the latest attempt on '%s'. Verdict: accept | reject | cycle.", firstSourceFile)
|
||||
case "contract_analyses":
|
||||
permit := strings.TrimPrefix(taskID, "permit:")
|
||||
// The Rust version reads ev.metadata.contractor when present.
|
||||
// Go's EvidenceRecord doesn't carry a free-form metadata bag
|
||||
// today, so we always emit the no-contractor form. Operators
|
||||
// who need contractor-aware instructions can extend this once
|
||||
// the metadata field lands in EvidenceRecord (separate ADR).
|
||||
return fmt.Sprintf("Analyze permit '%s'. Recommend with risk markers.", permit)
|
||||
case "outcomes":
|
||||
return "Run scenario; report per-event outcome with citations."
|
||||
default:
|
||||
return fmt.Sprintf("Source '%s' run; produce the appropriate output for this task type.", stem)
|
||||
}
|
||||
}
|
||||
|
||||
// ExportSft is the now-COMPLETE port (not just substrate). Walks
|
||||
// scored-runs files, loads paired evidence, applies the firewall,
|
||||
// synthesizes SFT samples for surviving records, writes the
|
||||
// combined JSONL output.
|
||||
//
|
||||
// Behavior:
|
||||
// - DryRun=true: counts but doesn't write files.
|
||||
// - IncludePartial=true: emit samples even when synthesis returns
|
||||
// "skip" reasons that aren't on the firewall (model_role
|
||||
// mismatch, empty text). Default false.
|
||||
// - Empty Root or missing scored-runs dir: returns zero-counts,
|
||||
// no error.
|
||||
//
|
||||
// Output JSONL format: one SftSample per line, UTF-8, LF-terminated,
|
||||
// matching the Rust convention so a/b validation between runtimes
|
||||
// can diff the files directly.
|
||||
func ExportSft(opts ExportSftOptions) (ExportSftResult, error) {
|
||||
if opts.RecordedAt == "" {
|
||||
return ExportSftResult{}, errors.New("distillation: ExportSft requires RecordedAt")
|
||||
}
|
||||
res := ExportSftResult{
|
||||
OutputPath: filepath.Join(opts.Root, "data", "distilled", "sft", "sft_partial.jsonl"),
|
||||
QuarantineSummary: "synthesis not yet ported — see internal/distillation/sft_export.go header",
|
||||
OutputPath: filepath.Join(opts.Root, "data", "distilled", "sft", "sft_export.jsonl"),
|
||||
}
|
||||
files, err := ListScoredRunFiles(opts.Root)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("list scored runs: %w", err)
|
||||
}
|
||||
res.ScoredFilesRead = len(files)
|
||||
|
||||
evCache := make(map[string]map[string]EvidenceRecord)
|
||||
var output []byte
|
||||
skippedNotInstructable := 0
|
||||
|
||||
for _, f := range files {
|
||||
runs, _, err := LoadScoredRunsFromFile(f)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
evidence, err := LoadEvidenceByRunID(f, evCache)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
res.RecordsRead += len(runs)
|
||||
for _, r := range runs {
|
||||
for i, r := range runs {
|
||||
if IsSftNever(r.Category) {
|
||||
res.RecordsQuarantined++
|
||||
continue
|
||||
}
|
||||
// Synthesis would happen here. Pre-port: count as
|
||||
// "would-export" for the firewall-passing records.
|
||||
ev, ok := evidence[r.EvidenceRunID]
|
||||
if !ok {
|
||||
// Evidence missing — without it we can't synthesize.
|
||||
// IncludePartial=true would let us emit a sample with
|
||||
// just scored-side fields, but the contract says
|
||||
// SftSample.Response must be non-empty (no evidence
|
||||
// = no text), so skip regardless.
|
||||
skippedNotInstructable++
|
||||
continue
|
||||
}
|
||||
sftID := fmt.Sprintf("%s:%d", filepath.Base(f), i)
|
||||
sample := SynthesizeSft(r, ev, opts.RecordedAt, sftID)
|
||||
if sample == nil {
|
||||
skippedNotInstructable++
|
||||
continue
|
||||
}
|
||||
if !opts.DryRun {
|
||||
line, err := json.Marshal(sample)
|
||||
if err != nil {
|
||||
skippedNotInstructable++
|
||||
continue
|
||||
}
|
||||
output = append(output, line...)
|
||||
output = append(output, '\n')
|
||||
}
|
||||
res.RecordsExported++
|
||||
}
|
||||
}
|
||||
|
||||
res.QuarantineSummary = fmt.Sprintf("firewall=%d, not-instructable=%d", res.RecordsQuarantined, skippedNotInstructable)
|
||||
if !opts.DryRun && len(output) > 0 {
|
||||
if err := os.MkdirAll(filepath.Dir(res.OutputPath), 0o755); err != nil {
|
||||
return res, fmt.Errorf("mkdir output: %w", err)
|
||||
}
|
||||
if err := os.WriteFile(res.OutputPath, output, 0o644); err != nil {
|
||||
return res, fmt.Errorf("write output: %w", err)
|
||||
}
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user