/** * @copyright Sister Software * @license AGPL-3.0 * @author Teffen Ellis, et al. * * Build a parquet slice from the DeepSeek-generated kryptonite JSONL and emit the corpus-v0.4.0 * MANIFEST. corpus-v0.4.0 is a pure adapter-addition revision: it points at every slice from * v0.3.0 plus the new kryptonite slice(s). No v0.3.0 bytes are touched or re-shuffled. * * See docs/engineering/reference/CORPUS_V0_4_0_GENERATION.md for the why; that doc also pins the * DeepSeek model version + prompt versions used to produce the JSONL. * * Invoke via `mailwoman corpus slice kryptonite \ * --jsonl /data/corpus/versioned/v0.4.0/kryptonite/canonical-kryptonite.jsonl \ * --base-manifest /data/corpus/versioned/v0.3.0/corpus-v0.3.0/MANIFEST.json \ * --out-dir /data/corpus/versioned/v0.4.0` */ import { pathExists, readLocalJSONFile } from "@mailwoman/core/fs/readers" import { writeLocalJSONFile, writeLocalTextFile, makeDirectories } from "@mailwoman/core/fs/writers" import { join } from "path-ts" import { JSONSpliterator } from "spliterator" import type { CanonicalRow, LabeledRow } from "#types" import { alignRow, PARQUET_COLUMNS, ROW_GROUP_SIZE, SLICE_COMPRESSION, writeSlices } from "#utils" import type { SliceManifest } from "#utils" export interface SliceKryptoniteOptions { jsonl: string baseManifest: string outDir: string /** * Default `"0.4.0"`. */ corpusVersion?: string /** * Default `"deepseek-kryptonite"`. */ source?: string } async function* canonicalRows(jsonl: string, corpusVersion: string): AsyncIterable { for await (const raw of JSONSpliterator.fromAsync>(jsonl)) { // Strip sidecar underscore-prefixed fields the generator left behind for debugging. const components = raw["components"] as Record yield { raw: raw["raw"] as string, components, country: (raw["country"] as string) ?? "US", locale: (raw["locale"] as string) ?? undefined, source: (raw["source"] as string) ?? "deepseek-kryptonite", source_id: raw["source_id"] as string, corpus_version: corpusVersion, license: (raw["license"] as string) ?? "Synthetic (DeepSeek-v4-flash, AGPL-compatible)", synth: raw["synth"] as CanonicalRow["synth"], } } } async function* labeledRows(jsonl: string, corpusVersion: string, quarantineLog: string[]): AsyncIterable { for await (const row of canonicalRows(jsonl, corpusVersion)) { const result = alignRow(row) if (result.kind === "labeled") { yield result.row } else { quarantineLog.push(`${row.source_id}\t${result.row.reason}`) } } } export async function buildKryptoniteSlice( options: SliceKryptoniteOptions, report?: (line: string) => void ): Promise { const corpusVersion = options.corpusVersion ?? "0.4.0" const source = options.source ?? "deepseek-kryptonite" if (!(await pathExists(options.jsonl))) throw new Error(`jsonl not found: ${options.jsonl}`) if (!(await pathExists(options.baseManifest))) throw new Error(`base-manifest not found: ${options.baseManifest}`) await makeDirectories(options.outDir) const quarantine: string[] = [] const newManifest = await writeSlices( { train: labeledRows(options.jsonl, corpusVersion, quarantine) }, { outputDir: options.outDir, corpusVersion } ) report?.( `wrote ${newManifest.total_rows} rows into ${newManifest.slices.length} slice(s); ` + `quarantined ${quarantine.length}` ) if (quarantine.length) { const qPath = join(options.outDir, `corpus-v${corpusVersion}`, "quarantine-kryptonite.tsv") await writeLocalTextFile(quarantine.join("\n") + "\n", qPath) report?.(`quarantine log → ${qPath}`) } // Stamp the new slice's source field for audit.ts (which prefers slice.source over // first_source_id-prefix inference). Without this, deepseek-kryptonite IDs would have // to match a prefix in KNOWN_SOURCE_PREFIXES — we add it there too as a belt-and-braces. for (const sh of newManifest.slices) { sh.source = source } // Compose the final corpus-v0.4.0 manifest: every slice from base + the new slice(s). const base = await readLocalJSONFile(options.baseManifest) const combined: SliceManifest = { corpus_version: corpusVersion, schema: PARQUET_COLUMNS, rows_per_slice: base.rows_per_slice, row_group_size: base.row_group_size ?? ROW_GROUP_SIZE, slices: [...base.slices, ...newManifest.slices], counts: { train: base.counts.train + (newManifest.counts.train ?? 0), val: base.counts.val, test: base.counts.test, }, total_rows: base.total_rows + newManifest.total_rows, } // Stamp source on the legacy v0.3.0 slices too, so audit's slice.source path is the // authoritative one. v0.3.0 slices mix sources; we use the first_source_id-prefix // inference for them (audit.ts will re-derive on its own when slice.source is absent). const combinedPath = join(options.outDir, `corpus-v${corpusVersion}`, "MANIFEST.json") await writeLocalJSONFile(combined, combinedPath) report?.(`wrote combined manifest → ${combinedPath}`) report?.(` total_rows=${combined.total_rows} (base=${base.total_rows}, added=${newManifest.total_rows})`) report?.(` slices=${combined.slices.length} (base=${base.slices.length}, added=${newManifest.slices.length})`) report?.(` compression=${SLICE_COMPRESSION}`) }