JacobLinCool's picture
download
raw
38.5 kB
from __future__ import annotations
import argparse
import importlib.util
import json
import os
import sys
from collections import Counter
from pathlib import Path
from typing import Any
import pytest
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SCRIPTS_ROOT = PROJECT_ROOT / "scripts"
if str(SCRIPTS_ROOT) not in sys.path:
sys.path.insert(0, str(SCRIPTS_ROOT))
import build_tables # noqa: E402
SPEC = importlib.util.spec_from_file_location(
"chebyshev_build_release", SCRIPTS_ROOT / "build_release.py"
)
assert SPEC is not None and SPEC.loader is not None
BUILDER = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = BUILDER
SPEC.loader.exec_module(BUILDER)
IDENTITY = {
"schema_version": "1.0.0",
"experiment_id": "icml2026_chebyshev_v1",
"spec_version": "1.0.0",
"paper_version": BUILDER.EXPECTED_PAPER_VERSION,
"openreview_id": BUILDER.EXPECTED_OPENREVIEW_ID,
}
def write_json(path: Path, value: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(value, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
def declaration(path: Path) -> dict[str, Any]:
return {
"sha256": BUILDER.sha256_file(path),
"bytes": path.stat().st_size,
}
def create_project(root: Path) -> Path:
project = root / "project"
files = {
"README.md": "# Workflow\n",
"pyproject.toml": "[project]\nname='release-test'\nversion='0.0.0'\n",
"uv.lock": "version = 1\n",
".python-version": "3.10\n",
"upstream.lock.json": "{}\n",
"configs/smoke.json": "{}\n",
"scripts/run_reproduction.py": (SCRIPTS_ROOT / "run_reproduction.py").read_text(
encoding="utf-8"
),
"scripts/build_tables.py": "#!/usr/bin/env python3\n# canonical table builder\n",
"scripts/make_figures.py": "#!/usr/bin/env python3\n# interactive figure builder\n",
"scripts/analyze_claims.py": "#!/usr/bin/env python3\n# claim analysis builder\n",
"experiments/test/SPEC.md": "# Spec\n",
"tests/test_smoke.py": "def test_smoke():\n assert True\n",
}
for name, content in files.items():
path = project / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return project
def create_raw_batch(
root: Path,
runner_path: Path,
*,
batch_id: str,
profile: str,
job_id: str,
config_name: str,
checkpoint_name: str = "policy.json",
) -> Path:
batch = root / "raw" / batch_id
inputs = batch / "inputs"
inputs.mkdir(parents=True)
input_contents = {
"SPEC.md": (
PROJECT_ROOT / "experiments" / "icml2026_chebyshev" / "SPEC.md"
).read_text(encoding="utf-8"),
config_name: (PROJECT_ROOT / "configs" / config_name).read_text(
encoding="utf-8"
),
"upstream.lock.json": (PROJECT_ROOT / "upstream.lock.json").read_text(
encoding="utf-8"
),
}
for name, content in input_contents.items():
(inputs / name).write_text(content, encoding="utf-8")
checkpoint = batch / "artifacts" / "checkpoints" / checkpoint_name
checkpoint.parent.mkdir(parents=True)
checkpoint.write_text('{"coefficients": [1.0]}\n', encoding="utf-8")
common = {
**IDENTITY,
"batch_id": batch_id,
"recorded_at": "2026-07-16T00:00:00+00:00",
"status": "success",
}
records = [
{
**common,
"record_type": "proof_certificate",
"claim_id": "claim1",
"task_id": "proof",
},
{
**common,
"record_type": "artifact",
"artifact_path": f"artifacts/checkpoints/{checkpoint_name}",
"artifact_sha256": BUILDER.sha256_file(checkpoint),
"artifact_bytes": checkpoint.stat().st_size,
},
]
records_path = batch / "records.jsonl"
records_path.write_text(
"".join(json.dumps(record, sort_keys=True) + "\n" for record in records),
encoding="utf-8",
)
manifest = {
**IDENTITY,
"batch_id": batch_id,
"status": "success",
"error": None,
"records_sha256": BUILDER.sha256_file(records_path),
"record_counts": dict(
sorted(Counter(record["record_type"] for record in records).items())
),
"runner_sha256": BUILDER.sha256_file(runner_path),
"input_hashes": {
name: BUILDER.sha256_file(inputs / name) for name in sorted(input_contents)
},
"config": {"claims": {"claim1": {"enabled": True}}},
"peak_rss_mb": 800.0 + int(job_id, 16),
"hardware": {
"environment": {"JOB_ID": job_id},
},
}
manifest["config"]["profile"] = profile
write_json(batch / "manifest.json", manifest)
return batch
def create_analysis(root: Path, project: Path, tables: Path) -> Path:
analysis = root / "analysis"
analysis.mkdir()
csv_path = analysis / "claim_results.csv"
markdown_path = analysis / "claim_results.md"
csv_path.write_text("component,verdict\nclaim1,supported\n", encoding="utf-8")
markdown_path.write_text("# Results\n", encoding="utf-8")
tables_manifest = json.loads((tables / "MANIFEST.json").read_text(encoding="utf-8"))
write_json(
analysis / "MANIFEST.json",
{
"status": "success",
"builder_sha256": BUILDER.sha256_file(
project / "scripts" / "analyze_claims.py"
),
"command": [
str((project / "scripts" / "analyze_claims.py").resolve()),
"--tables-root",
str(tables.resolve()),
"--output-root",
str(analysis.resolve()),
],
"source_tables_manifest_sha256": BUILDER.sha256_file(
tables / "MANIFEST.json"
),
"source_table_hashes": {
name: entry["sha256"]
for name, entry in sorted(tables_manifest["tables"].items())
},
"outputs": {
csv_path.name: declaration(csv_path),
markdown_path.name: declaration(markdown_path),
},
},
)
return analysis
def create_reports(root: Path, project: Path, tables: Path) -> Path:
reports = root / "reports"
reports.mkdir()
figures: dict[str, Any] = {}
for key in sorted(BUILDER.EXPECTED_FIGURE_KEYS):
html_path = reports / f"{key}.html"
data_path = reports / f"{key}.csv"
html_path.write_text(f"<!doctype html><p>{key}</p>\n", encoding="utf-8")
data_path.write_text(f"figure,value\n{key},1\n", encoding="utf-8")
figures[key] = {
"status": "generated",
"html": {"path": html_path.name, **declaration(html_path)},
"data": {"path": data_path.name, **declaration(data_path)},
}
tables_manifest = json.loads((tables / "MANIFEST.json").read_text(encoding="utf-8"))
write_json(
reports / "MANIFEST.json",
{
"status": "success",
"builder_sha256": BUILDER.sha256_file(
project / "scripts" / "make_figures.py"
),
"command": [
str((project / "scripts" / "make_figures.py").resolve()),
"--tables-root",
str(tables.resolve()),
"--output-root",
str(reports.resolve()),
],
"source": {
"tables_manifest_sha256": BUILDER.sha256_file(tables / "MANIFEST.json"),
"table_hashes": {
name: entry["sha256"]
for name, entry in sorted(tables_manifest["tables"].items())
},
},
"figures": figures,
},
)
return reports
def create_jobs(root: Path, batches: list[Path]) -> Path:
jobs = root / "jobs"
jobs.mkdir()
batch_by_scope = {
scope: batch
for scope, batch in zip(BUILDER.EXPECTED_JOB_SCOPES, batches, strict=True)
}
attempts: list[dict[str, Any]] = []
serial = 1
for version in BUILDER.EXPECTED_JOB_VERSIONS:
for scope in BUILDER.EXPECTED_JOB_SCOPES:
batch_manifest = json.loads(
(batch_by_scope[scope] / "manifest.json").read_text(encoding="utf-8")
)
if version == "v4":
job_id = batch_manifest["hardware"]["environment"]["JOB_ID"]
status, disposition = "COMPLETED", "validated"
batch_id = batch_manifest["batch_id"]
batch_declaration: dict[str, Any] | None = {
"batch_id": batch_id,
"manifest_sha256": BUILDER.sha256_file(
batch_by_scope[scope] / "manifest.json"
),
"records_sha256": BUILDER.sha256_file(
batch_by_scope[scope] / "records.jsonl"
),
"evidence_url": f"{BUILDER.PUBLIC_BUCKET_URL}/tree/raw/{batch_id}",
}
else:
job_id = f"{serial + 100:024x}"
status, disposition = BUILDER.EXPECTED_HISTORICAL_OUTCOMES[
(version, scope)
]
batch_declaration = None
canceled = status == "CANCELED"
input_sha256 = {
"runner": BUILDER.EXPECTED_RUNNER_SHA256[version],
"config": BUILDER.EXPECTED_CONFIG_SHA256[scope],
"lock": BUILDER.EXPECTED_LOCK_SHA256,
"spec": BUILDER.EXPECTED_SPEC_SHA256,
}
input_base = f"{BUILDER.PUBLIC_BUCKET_URL}/resolve/input/{version}"
config_name = BUILDER.EXPECTED_JOB_CONFIGS[scope]
reproduction_script = " ".join(
(
"set -euo pipefail;",
"tmp=$(mktemp -d);",
f"curl --fail -L {input_base}/run_reproduction.py -o $tmp/run_reproduction.py;",
f"curl --fail -L {input_base}/{config_name} -o $tmp/{config_name};",
f"curl --fail -L {input_base}/upstream.lock.json -o $tmp/upstream.lock.json;",
f"curl --fail -L {input_base}/SPEC.md -o $tmp/SPEC.md;",
(
"printf '"
f"{input_sha256['runner']} $tmp/run_reproduction.py\\n"
f"{input_sha256['config']} $tmp/{config_name}\\n"
f"{input_sha256['lock']} $tmp/upstream.lock.json\\n"
f"{input_sha256['spec']} $tmp/SPEC.md\\n' | sha256sum -c -;"
),
(
f"uv run $tmp/run_reproduction.py --config $tmp/{config_name} "
f"--batch-id rerun-{version}-{scope} --output-root ./runs/raw "
"--lock-path $tmp/upstream.lock.json --spec-path $tmp/SPEC.md;"
),
)
)
if version == "v4":
scheduler_command = [
"bash",
"/artifacts/input/v4/run_hf_job.sh",
config_name,
batch_manifest["batch_id"],
]
else:
scheduler_command = [
"uv",
"run",
f"/artifacts/input/{version}/run_reproduction.py",
"--config",
f"/artifacts/input/{version}/{config_name}",
"--batch-id",
f"hf-{scope}-{version}-test",
]
attempts.append(
{
"version": version,
"scope": scope,
"job_id": job_id,
"job_url": f"https://huggingface.co/jobs/{BUILDER.HF_NAMESPACE}/{job_id}",
"status": status,
"disposition": disposition,
"disposition_reason": (
"Validated raw evidence passed every integrity check."
if disposition == "validated"
else "Historical attempt was audited and intentionally not accepted."
),
"scheduler_command": scheduler_command,
"reproduction_command": ["bash", "-lc", reproduction_script],
"input_sha256": input_sha256,
"hardware": {
"flavor": "t4-medium",
"cpu_count": 8,
"memory_gb": 30,
"storage_gb": 100,
"accelerator": "NVIDIA T4 16GB",
},
"durations": {
"scheduling_seconds": None if canceled else 10,
"running_seconds": None if canceled else 20,
"total_seconds": None if canceled else 30,
},
"cost_usd": None if canceled else 0.01,
"peak_rss_mb": (
batch_manifest["peak_rss_mb"] if version == "v4" else None
),
"batch": batch_declaration,
}
)
serial += 1
write_json(
jobs / "HF_JOBS.json",
{
"schema_version": "1.0.0",
"paper": {
"openreview_id": BUILDER.EXPECTED_OPENREVIEW_ID,
"arxiv_version": BUILDER.EXPECTED_PAPER_VERSION,
},
"evidence_bucket_url": BUILDER.PUBLIC_BUCKET_URL,
"runner": {
"url": BUILDER.PUBLIC_RUNNER_URL_TEMPLATE.format(version="v4"),
"sha256": BUILDER.sha256_file(
batches[0].parents[1]
/ "project"
/ "scripts"
/ "run_reproduction.py"
),
},
"attempts": attempts,
},
)
return jobs
def create_poster(root: Path) -> Path:
poster = root / "poster"
poster.mkdir()
contents: dict[str, bytes] = {
"poster.html": b"<!doctype html><p>Final poster</p>\n",
"poster_embed.html": b"<!doctype html><p>Final embedded poster</p>\n",
"poster_preview.png": b"fake-png-preview\n",
"poster_preview.pdf": b"%PDF-1.4 fake preview\n",
}
for name, payload in contents.items():
(poster / name).write_bytes(payload)
write_json(
poster / "GATE_REPORT.json",
{
"overall": "PASS",
"hard_failures": 0,
"gates": [{"severity": "hard", "status": "PASS"}],
},
)
write_json(
poster / "style_check.json",
{
"status": "PASS",
"rules": [{"severity": "hard", "status": "PASS"}],
},
)
write_json(
poster / "asset_check.json",
{
"status": "WARN",
"checks": [{"severity": "hard", "status": "PASS"}],
},
)
files = {
name: {
**declaration(path),
"role": BUILDER.EXPECTED_POSTER_FILES[name],
}
for name in sorted(BUILDER.EXPECTED_POSTER_FILES)
if (path := poster / name).is_file()
}
write_json(
poster / "POSTER_MANIFEST.json",
{"schema_version": "1.0.0", "status": "success", "files": files},
)
return poster
def create_release_readme(root: Path, jobs: Path) -> Path:
attempts = json.loads((jobs / "HF_JOBS.json").read_text(encoding="utf-8"))[
"attempts"
]
links = "\n".join(
f"- [v4 {attempt['scope']}]({attempt['job_url']})"
for attempt in attempts
if attempt["version"] == "v4"
)
figures = "\n".join(
f"- [{key}](interactive-reports/{key}.html)"
for key in sorted(BUILDER.EXPECTED_FIGURE_KEYS)
)
content = f"""# Chebyshev Policies reproduction release
## Executive summary
This independent reproduction checks every major paper claim with immutable evidence, explicit verdicts, and carefully bounded conclusions for readers.
## Claim-by-claim verdicts
The release separates mathematical verification, empirical replication, arithmetic audits, historical-priority review, and the unavailable physical-hardware experiment.
## How to reproduce
Run the pinned Python workflow described here, then rebuild canonical tables, analysis, figures, and the poster from validated raw evidence.
## Evidence and provenance
- [OpenReview](https://openreview.net/forum?id={BUILDER.EXPECTED_OPENREVIEW_ID})
- [arXiv](https://arxiv.org/abs/{BUILDER.EXPECTED_PAPER_VERSION})
- [Evidence Bucket]({BUILDER.PUBLIC_BUCKET_URL})
- [Jobs inventory](jobs/HF_JOBS.json)
- [Release metadata](RELEASE.json)
{links}
Every accepted result is linked to a completed job and hash-declared batch. Earlier attempts remain documented with their audited disposition.
The `scheduler_command` field is the inspected scheduler argv, while `reproduction_command` is a separate hash-verified rerun recipe.
## Interactive reports
Each interactive report embeds its data and is paired with a canonical CSV, allowing inspection without a notebook or hidden state.
{figures}
- [Poster](poster/poster.html)
## Limitations
The analytical certificate does not establish a global optimum for bounded discrete Gym control, selection and reporting share a grid, and Aero hardware was unavailable.
This document intentionally repeats the core audit contract in plain language: accepted evidence is immutable, derived artifacts are builder-hash anchored, historical claims remain scoped, parameter arithmetic is checked independently, resource limits are reported, and unavailable physical experiments are never replaced with simulation. Readers can follow the links above to inspect every accepted batch, regenerate every table, compare every visualization, and distinguish reproduced findings from corrected or inconclusive findings without relying on undocumented context.
"""
path = root / "RELEASE_README.md"
path.write_text(content, encoding="utf-8")
return path
def create_bundle_inputs(
root: Path, *, checkpoint_name: str = "policy.json"
) -> argparse.Namespace:
project = create_project(root)
raw_roots = [
create_raw_batch(
root,
project / "scripts" / "run_reproduction.py",
batch_id=f"hf-{scope}-full-v4-test",
profile=BUILDER.EXPECTED_BATCH_PROFILES[scope],
job_id=f"{index + 1:024x}",
config_name=BUILDER.EXPECTED_JOB_CONFIGS[scope],
checkpoint_name=(
checkpoint_name if index == 0 else f"{scope}-{checkpoint_name}"
),
)
for index, scope in enumerate(BUILDER.EXPECTED_JOB_SCOPES)
]
tables = root / "tables"
build_tables.build_tables(raw_roots, tables, replace=False)
tables_manifest_path = tables / "MANIFEST.json"
tables_manifest = json.loads(tables_manifest_path.read_text(encoding="utf-8"))
tables_manifest["builder_sha256"] = BUILDER.sha256_file(
project / "scripts" / "build_tables.py"
)
command = [str((project / "scripts" / "build_tables.py").resolve())]
for raw_root in raw_roots:
command.extend(("--raw-root", str(raw_root.resolve())))
command.extend(("--output-root", str(tables.resolve())))
tables_manifest["command"] = command
write_json(tables_manifest_path, tables_manifest)
analysis = create_analysis(root, project, tables)
reports = create_reports(root, project, tables)
jobs = create_jobs(root, raw_roots)
poster = create_poster(root)
release_readme = create_release_readme(root, jobs)
return argparse.Namespace(
project_root=project,
release_readme=release_readme,
raw_root=raw_roots,
jobs_root=jobs,
tables_root=tables,
analysis_root=analysis,
reports_root=reports,
poster_root=poster,
output_root=root / "published" / "v1",
replace=False,
)
@pytest.fixture
def bundle(tmp_path: Path) -> argparse.Namespace:
return create_bundle_inputs(tmp_path)
def test_build_release_is_linked_runnable_and_machine_readable(
bundle: argparse.Namespace,
) -> None:
output = BUILDER.build_release(bundle)
assert (output / "tests" / "test_smoke.py").is_file()
assert (
output / "raw-provenance" / "hf-claims23-full-v4-test" / "manifest.json"
).is_file()
metadata = json.loads((output / "RELEASE.json").read_text(encoding="utf-8"))
assert len(metadata["batches"]) == 3
batch = next(
item
for item in metadata["batches"]
if item["batch_id"] == "hf-claims23-full-v4-test"
)
assert batch["job"]["url"].endswith("/000000000000000000000001")
assert batch["records"]["url"].endswith(
"/resolve/raw/hf-claims23-full-v4-test/records.jsonl"
)
assert batch["checkpoints"][0]["path"].endswith(
"/newly-trained-checkpoints/policy.json"
)
for line in (output / "MANIFEST.sha256").read_text(encoding="utf-8").splitlines():
digest, relative = line.split(" ", maxsplit=1)
assert BUILDER.sha256_file(output / relative) == digest
def test_rejects_stale_table_raw_provenance(bundle: argparse.Namespace) -> None:
manifest_path = bundle.tables_root / "MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["source_batches"][0]["records_sha256"] = "0" * 64
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="records SHA mismatch"):
BUILDER.build_release(bundle)
def test_rejects_analysis_with_mutated_declared_output(
bundle: argparse.Namespace,
) -> None:
with (bundle.analysis_root / "claim_results.md").open(
"a", encoding="utf-8"
) as handle:
handle.write("mutated\n")
with pytest.raises(BUILDER.ReleaseError, match="byte count mismatch"):
BUILDER.build_release(bundle)
def test_rejects_reports_from_different_tables(bundle: argparse.Namespace) -> None:
manifest_path = bundle.reports_root / "MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["source"]["tables_manifest_sha256"] = "0" * 64
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="different canonical tables"):
BUILDER.build_release(bundle)
@pytest.mark.parametrize(
("root_attribute", "manifest_name", "field", "message"),
[
("tables_root", "MANIFEST.json", "builder_sha256", "builder SHA"),
("analysis_root", "MANIFEST.json", "builder_sha256", "builder SHA"),
("reports_root", "MANIFEST.json", "builder_sha256", "builder SHA"),
],
)
def test_rejects_stale_derived_builder_hash(
bundle: argparse.Namespace,
root_attribute: str,
manifest_name: str,
field: str,
message: str,
) -> None:
manifest_path = getattr(bundle, root_attribute) / manifest_name
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest[field] = "0" * 64
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match=message):
BUILDER.build_release(bundle)
@pytest.mark.parametrize(
"root_attribute", ["tables_root", "analysis_root", "reports_root"]
)
def test_rejects_derived_command_for_different_output(
bundle: argparse.Namespace, root_attribute: str
) -> None:
manifest_path = getattr(bundle, root_attribute) / "MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["command"][-1] = "/tmp/stale-output"
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="command does not match"):
BUILDER.build_release(bundle)
def test_requires_all_four_interactive_figure_keys(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.reports_root / "MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["figures"].pop("pendulum_heatmap_difference")
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="missing required figures"):
BUILDER.build_release(bundle)
def test_rejects_required_interactive_figure_marked_missing(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.reports_root / "MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["figures"]["pendulum_heatmap_difference"] = {
"status": "missing",
"reason": "stale evidence",
}
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="was not generated"):
BUILDER.build_release(bundle)
def test_jobs_inventory_requires_exact_nine_attempt_matrix(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["attempts"].pop()
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="exactly nine attempts"):
BUILDER.build_release(bundle)
def test_jobs_inventory_rejects_unvalidated_extra_file(
bundle: argparse.Namespace,
) -> None:
(bundle.jobs_root / "notes.md").write_text("unvalidated\n", encoding="utf-8")
with pytest.raises(BUILDER.ReleaseError, match="file inventory mismatch"):
BUILDER.build_release(bundle)
@pytest.mark.parametrize(
("mutation", "message"),
[
("job_url", "job_url"),
("scheduler_command", "scheduler argv"),
("reproduction_command", "immutable input URLs"),
("input_sha256", "audited public inputs"),
("hardware", "t4-medium hardware"),
("durations", "complete durations"),
("cost", "cost_usd"),
("peak_rss", "peak_rss_mb"),
("disposition", "status/disposition"),
("batch", "does not match validated raw evidence"),
("job_id", "does not match the raw batch manifest"),
],
)
def test_jobs_inventory_rejects_invalid_attempt_evidence(
bundle: argparse.Namespace, mutation: str, message: str
) -> None:
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
attempt = next(
entry
for entry in manifest["attempts"]
if entry["version"] == "v4" and entry["scope"] == "claims23"
)
if mutation == "job_url":
attempt["job_url"] = "https://example.invalid/job"
elif mutation == "scheduler_command":
attempt["scheduler_command"][-1] = "wrong-batch"
elif mutation == "reproduction_command":
attempt["reproduction_command"][-1] = (
"set -euo pipefail; curl --fail claim23_full.json"
)
elif mutation == "input_sha256":
attempt["input_sha256"]["config"] = "0" * 64
elif mutation == "hardware":
attempt["hardware"]["memory_gb"] = 64
elif mutation == "durations":
attempt["durations"]["running_seconds"] = None
elif mutation == "cost":
attempt["cost_usd"] = None
elif mutation == "peak_rss":
attempt["peak_rss_mb"] += 1
elif mutation == "disposition":
attempt["disposition"] = "rejected"
elif mutation == "batch":
attempt["batch"]["records_sha256"] = "0" * 64
elif mutation == "job_id":
replacement = "f" * 24
attempt["job_id"] = replacement
attempt["job_url"] = (
f"https://huggingface.co/jobs/{BUILDER.HF_NAMESPACE}/{replacement}"
)
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match=message):
BUILDER.build_release(bundle)
def test_jobs_inventory_rejects_wrong_historical_disposition(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
attempt = next(
entry
for entry in manifest["attempts"]
if entry["version"] == "v3" and entry["scope"] == "claims23"
)
attempt["disposition"] = "validated"
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="historical status/disposition"):
BUILDER.build_release(bundle)
def test_jobs_inventory_does_not_conflate_scheduler_and_rerun_commands(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
attempt = next(
entry
for entry in manifest["attempts"]
if entry["version"] == "v4" and entry["scope"] == "claim4"
)
attempt["scheduler_command"], attempt["reproduction_command"] = (
attempt["reproduction_command"],
attempt["scheduler_command"],
)
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="scheduler argv"):
BUILDER.build_release(bundle)
def test_jobs_inventory_rejects_non_executable_curl_and_run_description(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
attempt = next(
entry
for entry in manifest["attempts"]
if entry["version"] == "v3" and entry["scope"] == "claims23"
)
attempt["reproduction_command"] = [
"bash",
"-lc",
(
f"curl {BUILDER.PUBLIC_RUNNER_URL_TEMPLATE.format(version='v3')} "
"&& run claim23_full.json"
),
]
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="immutable input URLs"):
BUILDER.build_release(bundle)
def test_jobs_inventory_rejects_unmeasured_historical_peak_rss(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
attempt = next(
entry
for entry in manifest["attempts"]
if entry["version"] == "v2" and entry["scope"] == "claims23"
)
attempt["peak_rss_mb"] = 1
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="must be null for v2/v3"):
BUILDER.build_release(bundle)
def test_jobs_inventory_requires_exact_three_validated_profiles(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.raw_root[2] / "manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["config"]["profile"] = BUILDER.EXPECTED_BATCH_PROFILES["claims23"]
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="Duplicate validated v4 batch"):
BUILDER.build_release(bundle)
def test_release_readme_rejects_arbitrary_text(bundle: argparse.Namespace) -> None:
bundle.release_readme.write_text("# Release\nanything\n", encoding="utf-8")
with pytest.raises(BUILDER.ReleaseError, match="too short"):
BUILDER.build_release(bundle)
@pytest.mark.parametrize(
("old", "new", "message"),
[
(
"## Evidence and provenance",
"## Evidence",
"missing required heading",
),
(
"jobs/HF_JOBS.json",
"jobs/UNKNOWN.json",
"missing required links",
),
(
"immutable evidence",
"__UNRESOLVED_RESULT__ evidence",
"Unresolved release README placeholder",
),
(
"hash-verified rerun recipe",
"another command",
"does not distinguish scheduler evidence",
),
],
)
def test_release_readme_requires_sections_links_and_no_placeholders(
bundle: argparse.Namespace, old: str, new: str, message: str
) -> None:
content = bundle.release_readme.read_text(encoding="utf-8")
assert old in content
bundle.release_readme.write_text(content.replace(old, new, 1), encoding="utf-8")
with pytest.raises(BUILDER.ReleaseError, match=message):
BUILDER.build_release(bundle)
def test_raw_validator_rejects_undeclared_checkpoint(
bundle: argparse.Namespace,
) -> None:
extra = bundle.raw_root[0] / "artifacts" / "checkpoints" / "extra.json"
extra.write_text("{}\n", encoding="utf-8")
with pytest.raises(BUILDER.ReleaseError, match="artifact inventory mismatch"):
BUILDER.build_release(bundle)
def test_release_accepts_only_declared_json_checkpoints(tmp_path: Path) -> None:
args = create_bundle_inputs(tmp_path, checkpoint_name="policy.pt")
with pytest.raises(BUILDER.ReleaseError, match="Model-weight/archive path"):
BUILDER.build_release(args)
def test_rejects_sensitive_or_weight_file_in_source(
bundle: argparse.Namespace,
) -> None:
(bundle.project_root / "scripts" / "model.pt").write_bytes(b"weights")
with pytest.raises(BUILDER.ReleaseError, match="Model-weight/archive path"):
BUILDER.build_release(bundle)
def test_rejects_token_shaped_content_in_allowed_file(
bundle: argparse.Namespace,
) -> None:
token = "hf_" + "a" * 32
token_variable = "_".join(("HF", "TOKEN"))
manifest_path = bundle.jobs_root / "HF_JOBS.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["attempts"][0]["disposition_reason"] = (
f"Unsafe {token_variable}={token} value here"
)
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="Hugging Face token"):
BUILDER.build_release(bundle)
def test_rejects_unresolved_poster_placeholder(bundle: argparse.Namespace) -> None:
poster_path = bundle.poster_root / "poster.html"
poster_path.write_text("<p>__C5_REPRO_RESULT__</p>\n", encoding="utf-8")
manifest_path = bundle.poster_root / "POSTER_MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["files"]["poster.html"].update(declaration(poster_path))
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="Unresolved poster placeholder"):
BUILDER.build_release(bundle)
def test_requires_poster_manifest(bundle: argparse.Namespace) -> None:
(bundle.poster_root / "POSTER_MANIFEST.json").unlink()
with pytest.raises(BUILDER.ReleaseError, match="Expected a real file"):
BUILDER.build_release(bundle)
@pytest.mark.parametrize(
"filename",
[
"poster.html",
"poster_embed.html",
"poster_preview.png",
"poster_preview.pdf",
"GATE_REPORT.json",
"style_check.json",
"asset_check.json",
],
)
def test_poster_manifest_rejects_stale_required_file_hash(
bundle: argparse.Namespace, filename: str
) -> None:
with (bundle.poster_root / filename).open("ab") as handle:
handle.write(b"stale\n")
with pytest.raises(
BUILDER.ReleaseError, match="Declared (byte count|SHA-256) mismatch"
):
BUILDER.build_release(bundle)
def test_poster_manifest_enforces_exact_inventory(bundle: argparse.Namespace) -> None:
(bundle.poster_root / "stale-preview.png").write_bytes(b"stale")
with pytest.raises(BUILDER.ReleaseError, match="file inventory mismatch"):
BUILDER.build_release(bundle)
def test_poster_manifest_requires_every_output_and_gate(
bundle: argparse.Namespace,
) -> None:
manifest_path = bundle.poster_root / "POSTER_MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["files"].pop("poster_preview.pdf")
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match="missing required files"):
BUILDER.build_release(bundle)
@pytest.mark.parametrize(
("filename", "field", "value", "message"),
[
("GATE_REPORT.json", "overall", "FAIL", "aggregate gate report"),
("style_check.json", "status", "FAIL", "style gate"),
("asset_check.json", "status", "FAIL", "asset gate"),
],
)
def test_poster_manifest_rejects_failed_gate_reports(
bundle: argparse.Namespace,
filename: str,
field: str,
value: str,
message: str,
) -> None:
gate_path = bundle.poster_root / filename
gate = json.loads(gate_path.read_text(encoding="utf-8"))
gate[field] = value
write_json(gate_path, gate)
manifest_path = bundle.poster_root / "POSTER_MANIFEST.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["files"][filename].update(declaration(gate_path))
write_json(manifest_path, manifest)
with pytest.raises(BUILDER.ReleaseError, match=message):
BUILDER.build_release(bundle)
def test_rejects_output_nested_in_recursive_source(bundle: argparse.Namespace) -> None:
bundle.output_root = bundle.project_root / "scripts" / "release"
with pytest.raises(BUILDER.ReleaseError, match="must be disjoint"):
BUILDER.build_release(bundle)
def test_atomic_replace_rolls_back_on_publish_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
destination = tmp_path / "release"
staging = tmp_path / "staging"
destination.mkdir()
staging.mkdir()
(destination / "old.txt").write_text("old\n", encoding="utf-8")
(staging / "new.txt").write_text("new\n", encoding="utf-8")
real_replace = os.replace
calls = 0
def fail_second_replace(source: Path, target: Path) -> None:
nonlocal calls
calls += 1
if calls == 2:
raise OSError("injected publish failure")
real_replace(source, target)
monkeypatch.setattr(BUILDER.os, "replace", fail_second_replace)
with pytest.raises(OSError, match="injected publish failure"):
BUILDER.publish_directory(staging, destination, replace=True)
assert (destination / "old.txt").read_text(encoding="utf-8") == "old\n"
assert not any(tmp_path.glob(".release.old-*"))
def test_release_metadata_sorts_batches(tmp_path: Path) -> None:
BUILDER.write_release_metadata(
tmp_path,
[{"batch_id": "z"}, {"batch_id": "a"}],
)
metadata = json.loads((tmp_path / "RELEASE.json").read_text(encoding="utf-8"))
assert [batch["batch_id"] for batch in metadata["batches"]] == ["a", "z"]

Xet Storage Details

Size:
38.5 kB
·
Xet hash:
19a2754de4a6bb2b08f2c92fe5a163aa19e27a378ba735e519272af34eec27ed

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.