Merge nucleic/upbeat-yarn-seal-ekbe into dev

This commit is contained in:
2026-08-01 21:26:31 -07:00
parent 9ecb82c983
commit 8f5acb7755
5 changed files with 1073 additions and 8 deletions
+52
View File
@@ -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
+15 -6
View File
@@ -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)
+717
View File
@@ -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())
+33 -2
View File
@@ -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)
+256
View File
@@ -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()