From 8f5acb77555f470d3dae2688f59b3953ddc0c269 Mon Sep 17 00:00:00 2001 From: Nucleic Date: Sat, 1 Aug 2026 21:26:31 -0700 Subject: [PATCH] Merge nucleic/upbeat-yarn-seal-ekbe into dev --- README.md | 52 ++ label_nucleic_prompts.py | 21 +- rebuild_sol_high.py | 717 ++++++++++++++++++++++++++++ tests/test_label_nucleic_prompts.py | 35 +- tests/test_rebuild_sol_high.py | 256 ++++++++++ 5 files changed, 1073 insertions(+), 8 deletions(-) create mode 100644 rebuild_sol_high.py create mode 100644 tests/test_rebuild_sol_high.py diff --git a/README.md b/README.md index b3bd929..71bc5a5 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,58 @@ The frozen test set is the synthetic JSONL plus the 87 classifiable records in records are excluded because `general` is deliberately not a model label. The exact membership and hashes are locked in `data/dataset-v1-manifest.json`. +## Full Sol-high label reset + +`rebuild_sol_high.py` is the end-to-end workflow for intentionally invalidating the old +labels and rebuilding both classifiers from one teacher. It freezes and relabels every +canonical public prompt, every shipped fixture, local Nucleic first-prompt history, and +the pinned SWE-chat candidates with `gpt-5.6-sol` at high reasoning effort. The teacher +settings are not configurable in this workflow. Batches default to eight prompts and +failed structured responses are retried up to ten times. + +The reset is staged below the ignored `.artifacts/sol-high-reset/` directory. Snapshot +hashes prevent a resumed run from silently mixing input revisions or teacher settings. +Run these commands from the repository root with the purpose-classifier environment: + +```bash +"$PY" ml/purpose-classifier/rebuild_sol_high.py snapshot + +"$PY" ml/purpose-classifier/rebuild_sol_high.py label \ + --python "$PY" + +"$PY" ml/purpose-classifier/rebuild_sol_high.py status +``` + +Rerunning `label` resumes every existing decision log and does not relabel completed +lines. Once `status` reports `"complete": true`, promote the staged result explicitly: + +```bash +"$PY" ml/purpose-classifier/rebuild_sol_high.py promote \ + --confirm overwrite-all-labels-with-sol-high +``` + +Promotion first validates and curates the complete staged population. It then replaces +the two canonical source files, regenerates all round-two mirrors, relabels shipped +fixtures (teacher-rejected fixtures become the runtime `general` fallback), rebuilds the +frozen public evaluation split, and replaces `.artifacts/dataset-v1/` with the combined +training dataset. Nucleic history and SWE-chat are deduplicated against public evaluation +data and added to training only; `vague-eval` records remain optimization exclusions. +The prior files are moved into a timestamped ignored backup before replacement. + +The old human/semantic review assertions are marked superseded because they were tied to +the invalidated label population. Consequently, Sol-high v2 evaluation measures agreement +with the new teacher and is not directly comparable with the old v1 scores. + +Print the two clean, from-pretrained-base training commands after promotion: + +```bash +"$PY" ml/purpose-classifier/rebuild_sol_high.py train-commands +``` + +The purpose-lite command starts from the pinned MiniLM revision. The purpose-deep command +starts from the pinned ModernBERT base variant; neither command supplies a prior classifier +checkpoint or continuation flag. + ## Label local Nucleic history `export_nucleic_prompts.py` extracts the first `userText` event from every local diff --git a/label_nucleic_prompts.py b/label_nucleic_prompts.py index 8d17b92..f086ca8 100644 --- a/label_nucleic_prompts.py +++ b/label_nucleic_prompts.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Filter and label exported Nucleic prompts with GPT-5.6 Terra via Codex.""" +"""Filter and label prompt-only sources with a configurable Codex teacher.""" from __future__ import annotations @@ -31,8 +31,8 @@ DEFAULT_OUTPUT = ( / ".artifacts" / "nucleic-history-first-prompts.labeled.jsonl" ) -MODEL = "gpt-5.6-terra" -REASONING_EFFORT = "low" +DEFAULT_MODEL = "gpt-5.6-terra" +DEFAULT_REASONING_EFFORT = "low" STATE_SCHEMA_VERSION = 1 DEFAULT_BATCH_SIZE = 40 DEFAULT_BATCH_CHARS = 80_000 @@ -443,6 +443,8 @@ def invoke_codex( *, codex: str, codex_isolation: str, + model: str, + reasoning_effort: str, max_prompt_chars: int, timeout_seconds: int, max_attempts: int, @@ -487,9 +489,9 @@ def invoke_codex( "--skip-git-repo-check", *exec_isolation_args, "--model", - MODEL, + model, "--config", - f'model_reasoning_effort="{REASONING_EFFORT}"', + f'model_reasoning_effort="{reasoning_effort}"', "--output-schema", str(schema_path), "--output-last-message", @@ -677,7 +679,8 @@ def label(args: argparse.Namespace) -> dict[str, int]: print( f"input={len(lines)} prefiltered={len(mechanical_rejections)} " f"resumed={len(states) - len(new_mechanical)} pending={len(pending)} " - f"batches={len(batch_list)} model={MODEL}", + f"batches={len(batch_list)} model={args.model} " + f"reasoning={args.reasoning_effort}", flush=True, ) @@ -691,6 +694,8 @@ def label(args: argparse.Namespace) -> dict[str, int]: batch, codex=args.codex, codex_isolation=args.codex_isolation, + model=args.model, + reasoning_effort=args.reasoning_effort, max_prompt_chars=args.max_prompt_chars, timeout_seconds=args.timeout_seconds, max_attempts=args.max_attempts, @@ -750,6 +755,8 @@ def build_parser() -> argparse.ArgumentParser: help="Audit JSONL for discarded lines (default: derived from --output)", ) parser.add_argument("--codex", default="codex", help="Codex CLI executable") + parser.add_argument("--model", default=DEFAULT_MODEL) + parser.add_argument("--reasoning-effort", default=DEFAULT_REASONING_EFFORT) parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE) parser.add_argument("--batch-chars", type=int, default=DEFAULT_BATCH_CHARS) parser.add_argument( @@ -803,6 +810,8 @@ def main(argv: Sequence[str] | None = None) -> int: parser.error("--timeout-seconds must be positive") if args.max_attempts <= 0: parser.error("--max-attempts must be positive") + if not args.model.strip() or not args.reasoning_effort.strip(): + parser.error("--model and --reasoning-effort must be non-empty") try: metrics = label(args) diff --git a/rebuild_sol_high.py b/rebuild_sol_high.py new file mode 100644 index 0000000..85c8ce1 --- /dev/null +++ b/rebuild_sol_high.py @@ -0,0 +1,717 @@ +#!/usr/bin/env python3 +"""Rebuild every purpose label with the pinned Sol-high teacher. + +The workflow is intentionally staged: + +1. ``snapshot`` freezes every prompt-bearing input below an ignored artifact directory. +2. ``label`` runs resumable Sol-high labeling jobs with small batches. +3. ``status`` proves whether every frozen input line has a terminal decision. +4. ``promote`` validates and curates the complete result before replacing canonical data. +5. ``train-commands`` prints from-base commands for purpose-lite and purpose-deep. + +Promotion is the only destructive phase and requires an explicit confirmation flag. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import hashlib +import json +import os +import shutil +import subprocess +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Sequence + +import prepare_data +import prepare_history_experiment +from purpose_data import ( + DataError, + file_sha256, + jsonl_bytes, + load_jsonl, + validate_source_record, + write_json, + write_jsonl, +) + + +SCRIPT_DIR = Path(__file__).resolve().parent +REPOSITORY_ROOT = SCRIPT_DIR.parent.parent +DATA_DIR = SCRIPT_DIR / "data" +FIXTURES = ( + REPOSITORY_ROOT + / "Tests" + / "NucleicCoreTests" + / "Fixtures" + / "purpose-prompts.json" +) +STAGE_DIR = SCRIPT_DIR / ".artifacts" / "sol-high-reset" +HISTORY_INPUT = SCRIPT_DIR / ".artifacts" / "nucleic-history-first-prompts.unlabeled.jsonl" +SWE_INPUT = SCRIPT_DIR / ".artifacts" / "swe-chat" / "candidates.jsonl" +PUBLIC_SOURCES = ( + DATA_DIR / "purpose-prompts.jsonl", + DATA_DIR / "purpose-prompts-round2.jsonl", +) +MODEL = "gpt-5.6-sol" +REASONING_EFFORT = "high" +WORKFLOW_SCHEMA = 1 +CONFIRMATION = "overwrite-all-labels-with-sol-high" +PUBLIC_DATASET_DESTINATION = SCRIPT_DIR / ".artifacts" / "dataset-sol-high-v2-public" +COMBINED_DATASET_DESTINATION = SCRIPT_DIR / ".artifacts" / "dataset-v1" + + +@dataclass(frozen=True) +class Job: + name: str + input: Path + output: Path + state: Path + rejects_or_audit: Path + script: Path + + +def canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _line_count(path: Path) -> int: + return len(path.read_text(encoding="utf-8").splitlines()) + + +def _relative(path: Path) -> str: + try: + return str(path.resolve().relative_to(REPOSITORY_ROOT)) + except ValueError: + return str(path.resolve()) + + +def jobs(stage: Path) -> tuple[Job, ...]: + def general(name: str) -> Job: + output = stage / f"{name}.labeled.jsonl" + return Job( + name=name, + input=stage / f"{name}.unlabeled.jsonl", + output=output, + state=stage / f"{name}.labeled.state.jsonl", + rejects_or_audit=stage / f"{name}.labeled.rejects.jsonl", + script=SCRIPT_DIR / "label_nucleic_prompts.py", + ) + + swe_output = stage / "swe.labeled.jsonl" + return ( + general("public"), + general("fixtures"), + general("history"), + Job( + name="swe", + input=stage / "swe.candidates.jsonl", + output=swe_output, + state=stage / "swe.labeled.state.jsonl", + rejects_or_audit=stage / "swe.labeled.audit.jsonl", + script=SCRIPT_DIR / "label_swe_chat_prompts.py", + ), + ) + + +def _read_json_array(path: Path) -> list[dict[str, Any]]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise DataError(f"{path}: cannot read JSON: {exc}") from exc + if not isinstance(value, list) or not all(isinstance(item, dict) for item in value): + raise DataError(f"{path}: expected an array of objects") + return value + + +def _prompt(value: dict[str, Any], location: str) -> str: + prompt = value.get("prompt") + if not isinstance(prompt, str) or not prompt.strip() or "\x00" in prompt: + raise DataError(f"{location}: invalid prompt") + return prompt + + +def _snapshot_jsonl(source: Path, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary = tempfile.mkstemp( + prefix=f".{destination.name}.", dir=destination.parent + ) + try: + with source.open("rb") as input_handle, os.fdopen( + descriptor, "wb" + ) as output_handle: + shutil.copyfileobj(input_handle, output_handle) + output_handle.flush() + os.fsync(output_handle.fileno()) + os.replace(temporary, destination) + except Exception: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + raise + + +def snapshot( + *, + stage: Path, + history_input: Path, + swe_input: Path, + overwrite_stage: bool, +) -> dict[str, Any]: + if stage.exists() and any(stage.iterdir()): + if not overwrite_stage: + raise DataError( + f"{stage}: stage is not empty; use --overwrite-stage intentionally" + ) + shutil.rmtree(stage) + stage.mkdir(parents=True, exist_ok=True) + + for path in (*PUBLIC_SOURCES, FIXTURES, history_input, swe_input): + if not path.is_file(): + raise DataError(f"{path}: required relabel input is missing") + + public_rows: list[dict[str, Any]] = [] + public_map: list[dict[str, Any]] = [] + for source in PUBLIC_SOURCES: + for line, value in enumerate(load_jsonl(source), 1): + prompt = _prompt(value, f"{source}:{line}") + public_rows.append({"prompt": prompt, "sessionID": f"{source.name}:{line}"}) + public_map.append({"source": source.name, "sourceLine": line}) + write_jsonl(stage / "public.unlabeled.jsonl", public_rows) + write_jsonl(stage / "public-map.jsonl", public_map) + + fixture_rows = _read_json_array(FIXTURES) + write_jsonl( + stage / "fixtures.unlabeled.jsonl", + ( + { + "prompt": _prompt(value, f"{FIXTURES}:{line}"), + "sessionID": f"fixture:{line}", + } + for line, value in enumerate(fixture_rows, 1) + ), + ) + _snapshot_jsonl(history_input, stage / "history.unlabeled.jsonl") + _snapshot_jsonl(swe_input, stage / "swe.candidates.jsonl") + + inputs = { + path.name: {"records": _line_count(path), "sha256": file_sha256(path)} + for path in ( + stage / "public.unlabeled.jsonl", + stage / "public-map.jsonl", + stage / "fixtures.unlabeled.jsonl", + stage / "history.unlabeled.jsonl", + stage / "swe.candidates.jsonl", + ) + } + config = { + "schemaVersion": WORKFLOW_SCHEMA, + "teacher": {"model": MODEL, "reasoningEffort": REASONING_EFFORT}, + "inputs": inputs, + "sourceInputs": { + _relative(path): { + "records": _line_count(path), + "sha256": file_sha256(path), + } + for path in (*PUBLIC_SOURCES, history_input, swe_input) + } + | { + _relative(FIXTURES): { + "records": len(fixture_rows), + "sha256": file_sha256(FIXTURES), + } + }, + } + write_json(stage / "workflow.json", config) + return config + + +def load_config(stage: Path) -> dict[str, Any]: + path = stage / "workflow.json" + try: + config = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise DataError(f"{path}: run snapshot first ({exc})") from exc + if config.get("schemaVersion") != WORKFLOW_SCHEMA: + raise DataError(f"{path}: unsupported workflow schema") + if config.get("teacher") != {"model": MODEL, "reasoningEffort": REASONING_EFFORT}: + raise DataError(f"{path}: workflow is not locked to {MODEL}/{REASONING_EFFORT}") + for name, expected in config.get("inputs", {}).items(): + candidate = stage / name + if not candidate.is_file() or file_sha256(candidate) != expected.get("sha256"): + raise DataError(f"{candidate}: frozen workflow input changed") + if _line_count(candidate) != expected.get("records"): + raise DataError(f"{candidate}: frozen workflow record count changed") + return config + + +def label( + *, + stage: Path, + python: str, + codex: str, + batch_size: int, + max_attempts: int, +) -> None: + load_config(stage) + for job in jobs(stage): + command = [ + python, + str(job.script), + "--input", + str(job.input), + "--output", + str(job.output), + "--state", + str(job.state), + "--model", + MODEL, + "--reasoning-effort", + REASONING_EFFORT, + "--batch-size", + str(batch_size), + "--max-attempts", + str(max_attempts), + "--codex", + codex, + ] + if job.name == "swe": + command += ["--audit", str(job.rejects_or_audit)] + else: + command += ["--rejects", str(job.rejects_or_audit)] + if job.state.exists(): + command.append("--resume") + print(f"\n==> {job.name}: {'resuming' if job.state.exists() else 'starting'}", flush=True) + completed = subprocess.run(command, check=False) + if completed.returncode != 0: + raise DataError( + f"{job.name} labeling failed with exit code {completed.returncode}; " + "rerun label to resume" + ) + + +def _load_terminal_states(job: Job) -> list[dict[str, Any]]: + if not job.state.is_file(): + raise DataError(f"{job.state}: missing labeling state") + states = load_jsonl(job.state) + expected = _line_count(job.input) + if len(states) != expected: + raise DataError(f"{job.name}: {len(states)}/{expected} input lines have terminal decisions") + raw_inputs = job.input.read_text(encoding="utf-8").splitlines() + by_line: dict[int, dict[str, Any]] = {} + for index, state in enumerate(states, 1): + line = state.get("sourceLine") + if not isinstance(line, int) or not 1 <= line <= expected or line in by_line: + raise DataError(f"{job.state}:{index}: invalid or duplicate sourceLine") + if state.get("status") not in {"labeled", "rejected"}: + raise DataError(f"{job.state}:{index}: decision is not terminal") + expected_hash = _sha256_bytes(raw_inputs[line - 1].encode("utf-8")) + if state.get("sourceLineHash") != expected_hash: + raise DataError(f"{job.state}:{index}: frozen input line hash changed") + if state["status"] == "labeled": + record = state.get("record") + if not isinstance(record, dict): + raise DataError(f"{job.state}:{index}: labeled state has no record") + validate_source_record(record, f"{job.state}:{index}") + by_line[line] = state + return [by_line[line] for line in range(1, expected + 1)] + + +def status(stage: Path) -> dict[str, Any]: + load_config(stage) + result: dict[str, Any] = {} + for job in jobs(stage): + expected = _line_count(job.input) + decided = _line_count(job.state) if job.state.exists() else 0 + labeled = rejected = 0 + if job.state.exists(): + for value in load_jsonl(job.state): + labeled += value.get("status") == "labeled" + rejected += value.get("status") == "rejected" + result[job.name] = { + "input": expected, + "decided": decided, + "labeled": labeled, + "rejected": rejected, + "complete": decided == expected, + } + result["complete"] = all(value["complete"] for value in result.values()) + return result + + +def _records_from_states(states: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + return [state["record"] for state in states if state["status"] == "labeled"] + + +def _write_fixture_candidate( + path: Path, + frozen_input: Path, + states: Sequence[dict[str, Any]], +) -> None: + source = load_jsonl(frozen_input) + if len(source) != len(states): + raise DataError("fixture snapshot no longer matches fixture decisions") + values = [] + for original, state in zip(source, states): + purpose = state["record"]["purpose"] if state["status"] == "labeled" else "general" + values.append( + { + "prompt": _prompt(original, f"{frozen_input}:{len(values) + 1}"), + "purpose": purpose, + } + ) + write_json(path, values) + + +def _rewrite_public_manifest_paths( + manifest: dict[str, Any], + sources: Sequence[Path], + fixtures: Path, + frozen: Path, +) -> None: + manifest["datasetVersion"] = "purpose-dataset-sol-high-v2" + for entry, destination in zip(manifest["sources"], sources): + entry["path"] = _relative(destination) + manifest["frozenEval"]["syntheticPath"] = _relative(frozen) + manifest["frozenEval"]["shippedFixturesPath"] = _relative(fixtures) + + +def _candidate_generation_manifest( + public_records: dict[str, list[dict[str, Any]]], +) -> dict[str, Any]: + path = DATA_DIR / "generation-manifest.json" + value = json.loads(path.read_text(encoding="utf-8")) + value["labeling"] = { + "schemaVersion": 1, + "model": MODEL, + "reasoningEffort": REASONING_EFFORT, + "date": dt.date.today().isoformat(), + "scope": "all canonical public prompts; rejected junk removed", + "files": { + name: { + "records": len(records), + "sha256": _sha256_bytes(jsonl_bytes(records)), + } + for name, records in public_records.items() + }, + } + return value + + +def _candidate_curation_review() -> dict[str, Any]: + path = DATA_DIR / "curation-review-v1.json" + value = json.loads(path.read_text(encoding="utf-8")) + semantic_review = value.get("semanticDuplicateReview") + if isinstance(semantic_review, dict): + semantic_review["status"] = "superseded-by-sol-high-relabel" + semantic_review["decision"] = ( + "The prior semantic decisions were tied to the invalidated label population; " + "the reset uses deterministic lexical curation only." + ) + review = value.get("humanLabelAndDifficultyReview") + if isinstance(review, dict): + review.update( + { + "status": "superseded-by-sol-high-relabel", + "decisionDate": dt.date.today().isoformat(), + "decisionBasis": ( + "The dataset owner invalidated the original labels and required " + "a complete Sol-high relabel." + ), + "decision": ( + "Every supervised target was regenerated with gpt-5.6-sol at " + "high reasoning effort." + ), + } + ) + value["datasetVersion"] = "purpose-dataset-sol-high-v2" + return value + + +def _replace_transaction(candidates: Sequence[tuple[Path, Path]], backup_root: Path) -> None: + backup_root.mkdir(parents=True, exist_ok=False) + moved_backups: list[tuple[Path, Path]] = [] + promoted: list[Path] = [] + try: + for candidate, destination in candidates: + destination.parent.mkdir(parents=True, exist_ok=True) + backup = backup_root / _relative(destination) + if destination.exists(): + backup.parent.mkdir(parents=True, exist_ok=True) + os.replace(destination, backup) + moved_backups.append((backup, destination)) + os.replace(candidate, destination) + promoted.append(destination) + except Exception: + for destination in reversed(promoted): + if destination.is_dir(): + shutil.rmtree(destination) + elif destination.exists(): + destination.unlink() + for backup, destination in reversed(moved_backups): + destination.parent.mkdir(parents=True, exist_ok=True) + os.replace(backup, destination) + raise + + +def promote(stage: Path, confirmation: str | None) -> dict[str, Any]: + if confirmation != CONFIRMATION: + raise DataError(f"promotion requires --confirm {CONFIRMATION}") + load_config(stage) + state_by_name = {job.name: _load_terminal_states(job) for job in jobs(stage)} + promotion = stage / "promotion" + if promotion.exists(): + shutil.rmtree(promotion) + promotion.mkdir(parents=True) + + public_map = load_jsonl(stage / "public-map.jsonl") + public_records: dict[str, list[dict[str, Any]]] = { + path.name: [] for path in PUBLIC_SOURCES + } + round2_by_line: dict[int, dict[str, Any]] = {} + for mapping, state in zip(public_map, state_by_name["public"]): + if state["status"] != "labeled": + continue + name = mapping.get("source") + if name not in public_records: + raise DataError(f"public map contains unknown source {name!r}") + record = state["record"] + public_records[name].append(record) + if name == "purpose-prompts-round2.jsonl": + round2_by_line[mapping["sourceLine"]] = record + + candidate_data = promotion / "data" + candidate_data.mkdir() + candidate_sources = [] + for destination in PUBLIC_SOURCES: + candidate = candidate_data / destination.name + write_jsonl(candidate, public_records[destination.name]) + candidate_sources.append(candidate) + for batch in range(1, 16): + first = (batch - 1) * 200 + 1 + write_jsonl( + candidate_data / f"round2-{batch:02d}.jsonl", + ( + round2_by_line[line] + for line in range(first, first + 200) + if line in round2_by_line + ), + ) + + candidate_fixtures = promotion / "purpose-prompts.json" + _write_fixture_candidate( + candidate_fixtures, + stage / "fixtures.unlabeled.jsonl", + state_by_name["fixtures"], + ) + combined_real = promotion / "combined-real.labeled.jsonl" + real_records = _records_from_states( + state_by_name["history"] + ) + _records_from_states(state_by_name["swe"]) + write_jsonl(combined_real, real_records) + + public_dataset = promotion / "dataset-public" + candidate_frozen = promotion / "frozen-test-v1.jsonl" + candidate_split_manifest = promotion / "dataset-v1-manifest.json" + public_manifest = prepare_data.prepare( + sources=candidate_sources, + fixtures_path=candidate_fixtures, + output_dir=public_dataset, + frozen_test_path=candidate_frozen, + manifest_path=candidate_split_manifest, + refresh_frozen_test=True, + seed=prepare_data.DEFAULT_SEED, + near_duplicate_threshold=0.92, + # The old semantic-review decisions were made against the label population + # that this reset explicitly invalidates. Lexical curation remains deterministic. + curation_review_path=None, + ) + _rewrite_public_manifest_paths( + public_manifest, + PUBLIC_SOURCES, + FIXTURES, + DATA_DIR / "frozen-test-v1.jsonl", + ) + write_json(public_dataset / "manifest.json", public_manifest) + write_json(candidate_split_manifest, public_manifest) + + combined_dataset = promotion / "dataset-combined" + combined_manifest = prepare_history_experiment.prepare( + base_dataset=public_dataset, + history_path=combined_real, + fixtures_path=candidate_fixtures, + output_dir=combined_dataset, + near_duplicate_threshold=0.92, + overwrite_output=False, + ) + combined_manifest["experimentVersion"] = "all-sol-high-training-augmentation-v2" + combined_manifest["teacher"] = {"model": MODEL, "reasoningEffort": REASONING_EFFORT} + combined_manifest["policy"]["historyUsage"] = ( + "Nucleic history and SWE-chat are training-only" + ) + combined_manifest["sources"]["baseDataset"]["path"] = _relative( + PUBLIC_DATASET_DESTINATION + ) + for split, entry in combined_manifest["sources"]["baseDataset"]["splits"].items(): + entry["path"] = _relative(PUBLIC_DATASET_DESTINATION / f"{split}.jsonl") + combined_manifest["sources"]["history"]["path"] = _relative( + stage / "combined-real.labeled.jsonl" + ) + combined_manifest["sources"]["fixtures"]["path"] = _relative(FIXTURES) + write_json(combined_dataset / "manifest.json", combined_manifest) + + candidate_generation = promotion / "generation-manifest.json" + candidate_review = promotion / "curation-review-v1.json" + write_json(candidate_generation, _candidate_generation_manifest(public_records)) + write_json(candidate_review, _candidate_curation_review()) + + replacements: list[tuple[Path, Path]] = [] + replacements.extend( + (candidate_data / path.name, path) for path in PUBLIC_SOURCES + ) + replacements.extend( + ( + candidate_data / f"round2-{batch:02d}.jsonl", + DATA_DIR / f"round2-{batch:02d}.jsonl", + ) + for batch in range(1, 16) + ) + replacements.extend( + [ + (candidate_fixtures, FIXTURES), + (candidate_frozen, DATA_DIR / "frozen-test-v1.jsonl"), + (candidate_split_manifest, DATA_DIR / "dataset-v1-manifest.json"), + (candidate_generation, DATA_DIR / "generation-manifest.json"), + (candidate_review, DATA_DIR / "curation-review-v1.json"), + (public_dataset, PUBLIC_DATASET_DESTINATION), + (combined_dataset, COMBINED_DATASET_DESTINATION), + ] + ) + backup_root = ( + stage + / "backups" + / dt.datetime.now(dt.UTC).strftime("%Y%m%dT%H%M%SZ") + ) + _replace_transaction(replacements, backup_root) + promoted_real = stage / "combined-real.labeled.jsonl" + os.replace(combined_real, promoted_real) + result = { + "teacher": {"model": MODEL, "reasoningEffort": REASONING_EFFORT}, + "publicLabeled": sum(len(value) for value in public_records.values()), + "realLabeled": len(real_records), + "combinedTrain": combined_manifest["outputs"]["train"]["records"], + "validation": combined_manifest["outputs"]["validation"]["records"], + "test": combined_manifest["outputs"]["test"]["records"], + "backup": _relative(backup_root), + } + write_json(stage / "promotion-result.json", result) + return result + + +def train_commands(python: str) -> str: + dataset = _relative(COMBINED_DATASET_DESTINATION) + lite = _relative(SCRIPT_DIR / "outputs" / "purpose-lite-sol-high-v2") + deep = _relative(SCRIPT_DIR / "outputs" / "purpose-deep-sol-high-v2-base") + separator = " " + "\\" + "\n " + lite_command = separator.join( + ( + f'"{python}" ml/purpose-classifier/train.py', + f"--dataset-dir {dataset}", + f"--output-dir {lite}", + "--overwrite-output", + ) + ) + deep_command = separator.join( + ( + f'"{python}" ml/purpose-classifier/train_deep_mlx.py', + "--variant base", + f"--dataset-dir {dataset}", + f"--output-dir {deep}", + "--overwrite-output", + ) + ) + return f"{lite_command}\n\n{deep_command}" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--stage-dir", type=Path, default=STAGE_DIR) + subparsers = parser.add_subparsers(dest="command", required=True) + + snapshot_parser = subparsers.add_parser("snapshot", help="freeze every relabel input") + snapshot_parser.add_argument("--history-input", type=Path, default=HISTORY_INPUT) + snapshot_parser.add_argument("--swe-input", type=Path, default=SWE_INPUT) + snapshot_parser.add_argument("--overwrite-stage", action="store_true") + + label_parser = subparsers.add_parser("label", help="run or resume all Sol-high jobs") + label_parser.add_argument("--python", default=sys.executable) + label_parser.add_argument("--codex", default="codex") + label_parser.add_argument("--batch-size", type=int, default=8) + label_parser.add_argument("--max-attempts", type=int, default=10) + + subparsers.add_parser("status", help="report completeness without changing data") + promote_parser = subparsers.add_parser( + "promote", help="curate and replace canonical labels" + ) + promote_parser.add_argument("--confirm") + commands_parser = subparsers.add_parser( + "train-commands", help="print clean from-base training commands" + ) + commands_parser.add_argument("--python", default="$PY") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + stage = args.stage_dir.expanduser().resolve() + try: + if args.command == "snapshot": + result = snapshot( + stage=stage, + history_input=args.history_input.expanduser().resolve(), + swe_input=args.swe_input.expanduser().resolve(), + overwrite_stage=args.overwrite_stage, + ) + print(json.dumps(result, indent=2, sort_keys=True)) + elif args.command == "label": + if args.batch_size <= 0 or args.max_attempts <= 0: + parser.error("--batch-size and --max-attempts must be positive") + label( + stage=stage, + python=args.python, + codex=args.codex, + batch_size=args.batch_size, + max_attempts=args.max_attempts, + ) + print(json.dumps(status(stage), indent=2, sort_keys=True)) + elif args.command == "status": + print(json.dumps(status(stage), indent=2, sort_keys=True)) + elif args.command == "promote": + print(json.dumps(promote(stage, args.confirm), indent=2, sort_keys=True)) + else: + print(train_commands(args.python)) + except ( + DataError, + OSError, + UnicodeError, + json.JSONDecodeError, + subprocess.SubprocessError, + ValueError, + ) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_label_nucleic_prompts.py b/tests/test_label_nucleic_prompts.py index ef0c90e..bddfebd 100644 --- a/tests/test_label_nucleic_prompts.py +++ b/tests/test_label_nucleic_prompts.py @@ -22,9 +22,10 @@ from pathlib import Path args = sys.argv[1:] model_index = args.index("--model") -if args[model_index + 1] != "gpt-5.6-terra": +if args[model_index + 1] != os.environ.get("EXPECTED_MODEL", "gpt-5.6-terra"): raise SystemExit("wrong model") -if 'model_reasoning_effort="low"' not in args: +expected_effort = os.environ.get("EXPECTED_REASONING_EFFORT", "low") +if f'model_reasoning_effort="{expected_effort}"' not in args: raise SystemExit("wrong reasoning effort") prompt = sys.stdin.read() @@ -95,6 +96,12 @@ class LabelNucleicPromptsTests(unittest.TestCase): ) -> subprocess.CompletedProcess[str]: environment = os.environ.copy() environment["FAKE_CODEX_LOG"] = str(root / "calls.log") + if "--model" in extra: + environment["EXPECTED_MODEL"] = extra[extra.index("--model") + 1] + if "--reasoning-effort" in extra: + environment["EXPECTED_REASONING_EFFORT"] = extra[ + extra.index("--reasoning-effort") + 1 + ] return subprocess.run( [ sys.executable, @@ -224,6 +231,30 @@ class LabelNucleicPromptsTests(unittest.TestCase): ) self.assertIn("pending=0", second.stdout) + def test_custom_model_and_reasoning_effort_are_forwarded(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + fake_codex = self.make_fake_codex(root) + input_path = root / "unlabeled.jsonl" + input_path.write_text( + json.dumps({"prompt": "Write the migration note"}) + "\n", + encoding="utf-8", + ) + + result = self.run_script( + root, + input_path, + root / "labeled.jsonl", + fake_codex, + "--model", + "gpt-5.6-sol", + "--reasoning-effort", + "high", + ) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertIn("model=gpt-5.6-sol reasoning=high", result.stdout) + def test_head_tail_excerpt_is_bounded(self): prompt = "HEAD" + ("x" * 4_000) + "TAIL" excerpt = label_nucleic_prompts.excerpt_for_labeling(prompt, 1_000) diff --git a/tests/test_rebuild_sol_high.py b/tests/test_rebuild_sol_high.py new file mode 100644 index 0000000..8a92c73 --- /dev/null +++ b/tests/test_rebuild_sol_high.py @@ -0,0 +1,256 @@ +import json +import hashlib +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +MODULE_DIR = Path(__file__).resolve().parents[1] +if str(MODULE_DIR) not in sys.path: + sys.path.insert(0, str(MODULE_DIR)) + +import purpose_data +import rebuild_sol_high + + +def source_record(prompt: str, purpose: str = "backendImpl") -> dict: + return { + "prompt": prompt, + "purpose": purpose, + "secondary": None, + "mixed": False, + "difficulty": 0.4, + "slice": "core", + "lang": "en", + } + + +def raw_hash(raw: str) -> str: + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +class RebuildSolHighTests(unittest.TestCase): + def test_snapshot_locks_all_inputs_and_status_starts_incomplete(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + data = root / "data" + data.mkdir() + first = data / "first.jsonl" + second = data / "second.jsonl" + purpose_data.write_jsonl(first, [source_record("Build the API")]) + purpose_data.write_jsonl(second, [source_record("Fix the layout", "frontendImpl")]) + fixtures = root / "fixtures.json" + purpose_data.write_json( + fixtures, + [{"prompt": "Explain this module", "purpose": "review"}], + ) + history = root / "history.jsonl" + purpose_data.write_jsonl(history, [{"prompt": "Add a cache"}]) + swe = root / "swe.jsonl" + purpose_data.write_jsonl(swe, [{"prompt": "Debug the crash"}]) + stage = root / "stage" + + with mock.patch.object( + rebuild_sol_high, "PUBLIC_SOURCES", (first, second) + ), mock.patch.object(rebuild_sol_high, "FIXTURES", fixtures): + config = rebuild_sol_high.snapshot( + stage=stage, + history_input=history, + swe_input=swe, + overwrite_stage=False, + ) + result = rebuild_sol_high.status(stage) + + self.assertEqual( + config["teacher"], + {"model": "gpt-5.6-sol", "reasoningEffort": "high"}, + ) + self.assertEqual(result["public"]["input"], 2) + self.assertEqual(result["fixtures"]["input"], 1) + self.assertFalse(result["complete"]) + + with (stage / "history.unlabeled.jsonl").open("a", encoding="utf-8") as handle: + handle.write('{"prompt":"changed"}\n') + with self.assertRaisesRegex(purpose_data.DataError, "input changed"): + rebuild_sol_high.load_config(stage) + + def test_status_requires_one_terminal_decision_per_input_line(self): + with tempfile.TemporaryDirectory() as directory: + stage = Path(directory) + inputs = {} + for job in rebuild_sol_high.jobs(stage): + purpose_data.write_jsonl(job.input, [{"prompt": f"{job.name} prompt"}]) + inputs[job.input.name] = { + "records": 1, + "sha256": purpose_data.file_sha256(job.input), + } + state = { + "schemaVersion": 1, + "sourceLine": 1, + "sourceLineHash": "unused-by-status", + "promptHash": "unused-by-status", + "sessionID": None, + "status": "labeled", + "reason": None, + "record": source_record(f"{job.name} prompt"), + } + purpose_data.write_jsonl(job.state, [state]) + purpose_data.write_jsonl(stage / "public-map.jsonl", []) + purpose_data.write_json( + stage / "workflow.json", + { + "schemaVersion": 1, + "teacher": { + "model": "gpt-5.6-sol", + "reasoningEffort": "high", + }, + "inputs": inputs, + "sourceInputs": {}, + }, + ) + + result = rebuild_sol_high.status(stage) + + self.assertTrue(result["complete"]) + self.assertTrue( + all( + result[name]["labeled"] == 1 + for name in ("public", "fixtures", "history", "swe") + ) + ) + + def test_training_commands_are_from_base_for_both_tiers(self): + commands = rebuild_sol_high.train_commands("$PY") + + self.assertIn("train.py", commands) + self.assertIn("train_deep_mlx.py", commands) + self.assertIn("--variant base", commands) + self.assertEqual(commands.count("--dataset-dir"), 2) + self.assertNotIn("--resume-from", commands) + self.assertNotIn("--model ", commands) + + def test_complete_stage_promotes_public_and_combined_datasets(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + data = root / "ml" / "purpose-classifier" / "data" + data.mkdir(parents=True) + script_dir = data.parent + first = data / "purpose-prompts.jsonl" + second = data / "purpose-prompts-round2.jsonl" + records = [ + source_record( + f"Implement sample endpoint number {index} with stable pagination", + purpose_data.LABELS[index % len(purpose_data.LABELS)], + ) + for index in range(80) + ] + purpose_data.write_jsonl(first, records[:40]) + purpose_data.write_jsonl(second, records[40:]) + purpose_data.write_json( + data / "generation-manifest.json", + { + "schemaVersion": 1, + "canonicalFiles": [first.name, second.name], + "derivedBatchFiles": [], + "generations": [], + "limitations": [], + }, + ) + purpose_data.write_json( + data / "curation-review-v1.json", + { + "schemaVersion": 1, + "datasetVersion": "purpose-dataset-v1", + "semanticDuplicateReview": {"status": "complete"}, + "humanLabelAndDifficultyReview": {"status": "complete"}, + }, + ) + fixtures = root / "Tests" / "Fixtures" / "purpose-prompts.json" + purpose_data.write_json( + fixtures, + [{"prompt": "Plan the cache migration", "purpose": "planning"}], + ) + history = root / "history.jsonl" + purpose_data.write_jsonl(history, [{"prompt": "Add a private cache layer"}]) + swe = root / "swe.jsonl" + purpose_data.write_jsonl(swe, [{"prompt": "Diagnose a unique worker crash"}]) + stage = script_dir / ".artifacts" / "sol-high-reset" + public_dataset = script_dir / ".artifacts" / "dataset-public" + combined_dataset = script_dir / ".artifacts" / "dataset-v1" + + patches = ( + mock.patch.object(rebuild_sol_high, "REPOSITORY_ROOT", root), + mock.patch.object(rebuild_sol_high, "DATA_DIR", data), + mock.patch.object(rebuild_sol_high, "FIXTURES", fixtures), + mock.patch.object(rebuild_sol_high, "PUBLIC_SOURCES", (first, second)), + mock.patch.object( + rebuild_sol_high, + "PUBLIC_DATASET_DESTINATION", + public_dataset, + ), + mock.patch.object( + rebuild_sol_high, + "COMBINED_DATASET_DESTINATION", + combined_dataset, + ), + ) + with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + rebuild_sol_high.snapshot( + stage=stage, + history_input=history, + swe_input=swe, + overwrite_stage=False, + ) + for job in rebuild_sol_high.jobs(stage): + input_lines = job.input.read_text(encoding="utf-8").splitlines() + states = [] + for line, raw in enumerate(input_lines, 1): + prompt = json.loads(raw)["prompt"] + original = next( + (item for item in records if item["prompt"] == prompt), + None, + ) + record = original or source_record(prompt) + states.append( + { + "schemaVersion": 1, + "sourceLine": line, + "sourceLineHash": raw_hash(raw), + "promptHash": "test", + "sessionID": None, + "status": "labeled", + "reason": None, + "record": record, + } + ) + purpose_data.write_jsonl(job.state, states) + + result = rebuild_sol_high.promote( + stage, + rebuild_sol_high.CONFIRMATION, + ) + + self.assertEqual(80, result["publicLabeled"]) + self.assertEqual(2, result["realLabeled"]) + self.assertTrue((combined_dataset / "train.jsonl").is_file()) + self.assertGreater( + len(purpose_data.load_jsonl(combined_dataset / "train.jsonl")), + len(purpose_data.load_jsonl(public_dataset / "train.jsonl")), + ) + promoted_manifest = json.loads( + (data / "dataset-v1-manifest.json").read_text(encoding="utf-8") + ) + self.assertEqual( + "purpose-dataset-sol-high-v2", + promoted_manifest["datasetVersion"], + ) + + def test_promotion_requires_exact_confirmation(self): + with self.assertRaisesRegex(purpose_data.DataError, "promotion requires"): + rebuild_sol_high.promote(Path("unused"), None) + + +if __name__ == "__main__": + unittest.main()