From 7cd38fee8a6b0bcd745e082be3f3296c3f221e7d Mon Sep 17 00:00:00 2001 From: Owen Price Skelly <21372141+OwenPriceSkelly@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:51:54 -0500 Subject: [PATCH] smoke-test: persist each checkpoint's result to the manifest as it lands Results used to accumulate in memory and reach manifest.json only after the last checkpoint and custom-weights leg. Two 12 h Delta nightly runs on 2026-08-13 (SLURM 21119813, 21119858) hit their wall-time limit mid-run and left nothing: no stamps on disk, no push. Each outcome (verification record plus its weight capture) is now written to {root}/manifest.json the moment it is known, via one short locked read-modify-write per checkpoint, so a killed run keeps every verified/failed stamp it reached. The custom-weights legs persist the same way; they still take passed_keys/baselines/base_captures from the in-memory main loop. The backend push stays a single end-of-run push (the backend replaces the whole manifest), with a new optional --push-every N that also pushes after every N results. --no-push still suppresses every push. JSON/text output, checkpoint-first selection, and the shared-install clusters projection are unchanged. Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 2 +- docs/nightly-smoke-testing.md | 15 +++ rootstock/cli.py | 12 ++ rootstock/commands/smoke_test.py | 150 +++++++++++++---------- tests/cli/test_smoke_test.py | 167 ++++++++++++++++++++++++++ tests/cli/test_smoke_test_clusters.py | 1 + 6 files changed, 285 insertions(+), 62 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 4fb1e92..7ede96c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -28,7 +28,7 @@ rootstock add [--kwarg key=val ...] [--device cuda] [--no-verify # (re-loads a same-family checkpoint's cached weights and compares results). # Suitable for nightly cron. --cluster names the machine (required on shared # installs like sophia/polaris; add/sync take it too). -rootstock smoke-test [--env ENV] [--checkpoint CKPT] [--device cuda] [--cluster NAME] [--json] +rootstock smoke-test [--env ENV] [--checkpoint CKPT] [--device cuda] [--cluster NAME] [--json] [--push-every N] # Show status (per-checkpoint verified/stale grid; --json for machine-readable) rootstock status [--root ] [--json] diff --git a/docs/nightly-smoke-testing.md b/docs/nightly-smoke-testing.md index 4a2215b..202f26c 100644 --- a/docs/nightly-smoke-testing.md +++ b/docs/nightly-smoke-testing.md @@ -9,6 +9,16 @@ automated smoke-test rotation. pass each) and pushes the refreshed `manifest.json` to the backend. Running it on a schedule keeps each cluster's manifest honest about checkpoint health. +Each checkpoint's result is written to `manifest.json` the moment it is known +(a short locked read-modify-write per checkpoint), so a job the scheduler +kills at its wall-time limit keeps every verified/failed stamp it reached. +The backend push is a separate step that happens once, after the last +checkpoint. To bound how much a killed run leaves unpushed, add +`--push-every N`: the manifest is then also pushed after every N results +(each push replaces the whole manifest server-side, so an intermediate push +is simply a snapshot of the stamps so far). Either way, stamps a killed run +left on disk ride along with the next successful push from that install. + User-supplied weights (`:custom` checkpoints) are never smoke-tested — there is nothing registered to verify. Every use is a fresh load through the env's `setup_from_path` hook, so a bad file simply fails at server start. @@ -60,6 +70,11 @@ Two recipes with placeholders: start time sits in state `W`, waiting — not `Q`), and - the manifest landed — `rootstock status` shows fresh `verified_at` times. + If the job hit its wall-time limit before pushing, `rootstock status` still + shows the stamps it reached (they are written per checkpoint); only the + push is missing. Raise the walltime in the recipe, or pass `--push-every N` + in the `rootstock smoke-test` invocation so partial runs reach the backend. + ## How the self-scheduling works - **Resubmit-first.** The job queues its next run *before* the smoke-test, so a diff --git a/rootstock/cli.py b/rootstock/cli.py index ba7e500..059bfff 100644 --- a/rootstock/cli.py +++ b/rootstock/cli.py @@ -582,6 +582,18 @@ def main(): action="store_true", help="Don't push manifest to backend", ) + smoke_parser.add_argument( + "--push-every", + type=int, + default=0, + metavar="N", + help=( + "Also push the manifest after every N checkpoint results (default: " + "only once, at the end). Each result is written to manifest.json as " + "soon as it is known regardless; this only bounds how much a run " + "killed at its wall-time limit leaves unpushed. Ignored with --no-push." + ), + ) smoke_parser.set_defaults(func=cmd_smoke_test) # status command diff --git a/rootstock/commands/smoke_test.py b/rootstock/commands/smoke_test.py index 76f06ee..5e7f3f9 100644 --- a/rootstock/commands/smoke_test.py +++ b/rootstock/commands/smoke_test.py @@ -317,6 +317,55 @@ def _dominant_weights_file(files: list[dict] | None) -> str | None: return largest["path"] +def _persist_outcome( + root: Path, + cluster: str, + device: str, + env_name: str, + ckpt_name: str, + verification: VerificationRecord, + weight_files: list[dict] | None, + fetched_at: str | None, + progress, +) -> None: + """Write one checkpoint's outcome to ``{root}/manifest.json`` now. + + Each result lands as soon as it is known — one short locked + read-modify-write per checkpoint — so a run the scheduler kills at its + wall-time limit still leaves the verified/failed stamps it reached (two + 12 h Delta runs on 2026-08-13 timed out and left nothing). The manifest + is re-read *inside* the lock: verification runs are minutes long and a + co-maintainer may have written in between, so the copy loaded before the + loop is never saved. + """ + with manifest_lock(root): + fresh = load_manifest(root) + if fresh is None: + return + env_record = fresh.environments.get(env_name) + if env_record is None: + return # env removed while we were testing + # Created on first pass when the resolved env has no record of an id + # it now hosts (a variant tested checkpoint-first for the first time, + # #208; a ':custom' entry, which nothing else ever writes). The + # donor's fetch stamp rides along so a canonical record never reads + # "verified but never fetched". + ckpt_record = env_record.checkpoints.setdefault(ckpt_name, CheckpointInfo()) + if ckpt_record.fetched_at is None and fetched_at is not None: + ckpt_record.fetched_at = fetched_at + ckpt_record.verifications[cluster] = verification + apply_weights_record( + ckpt_record, weight_files, label=f"{env_name}/{ckpt_name}", progress=progress + ) + save_manifest(fresh, root) + + +def _verification_for(ok: bool, err: str | None, device: str) -> VerificationRecord: + if ok: + return VerificationRecord(verified_at=now_iso(), verified_device=device) + return VerificationRecord(last_error=f"smoke-test: {err}") + + def cmd_smoke_test(args) -> int: root: Path = get_root_or_exit(args) cache_root = resolve_cache_root(root) @@ -326,6 +375,7 @@ def cmd_smoke_test(args) -> int: verify_timeout = args.verify_timeout json_out = args.json no_push = args.no_push + push_every = getattr(args, "push_every", 0) or 0 if ckpt_filter is not None and env_filter is None: print("Error: --checkpoint requires --env", file=sys.stderr) @@ -385,12 +435,19 @@ def cmd_smoke_test(args) -> int: n_failed = 0 total_start = time.monotonic() - # Verification runs are long (minutes per checkpoint) and must not hold - # the manifest lock. Record outcomes here and apply them to a *freshly - # loaded* manifest in one locked cycle afterwards — mutating the manifest - # loaded before the loop and saving it at the end would silently revert - # anything a co-maintainer wrote in between. - outcomes: list[tuple[str, str, bool, str | None, list[dict] | None, str | None]] = [] + # Every outcome is written to manifest.json the moment it is known (see + # _persist_outcome), so a run killed mid-way keeps what it reached. The + # backend push is separate: once at the end by default, or additionally + # after every --push-every recorded outcomes. Each push replaces the + # whole manifest server-side, so an intermediate push is simply a + # snapshot of the stamps so far. + n_recorded = 0 + progress = None if json_out else print + + def _maybe_push() -> None: + if no_push or push_every <= 0 or n_recorded % push_every: + return + update_and_push_manifest(root, quiet=True, push=True) # The custom legs reuse the canonical loop's work: its results are the # comparison baseline, and its fresh weight capture names the file on @@ -421,22 +478,31 @@ def cmd_smoke_test(args) -> int: ) weight_files = read_weights_capture(capture_path) elapsed = time.monotonic() - start - outcomes.append((env_name, ckpt_name, ok, err, weight_files, ckpt.fetched_at)) if (env_name, ckpt_name) in base_keys: baselines[(env_name, ckpt_name)] = run_results base_captures[(env_name, ckpt_name)] = weight_files + verification = _verification_for(ok, err, device) + _persist_outcome( + root, + cluster, + device, + env_name, + ckpt_name, + verification, + weight_files, + ckpt.fetched_at, + progress, + ) + n_recorded += 1 # Mirror the outcome onto the working copy so verified_current in the # report reflects this run. + ckpt.verifications[cluster] = verification if ok: passed_keys.add((env_name, ckpt_name)) - ckpt.verifications[cluster] = VerificationRecord( - verified_at=now_iso(), verified_device=device - ) n_passed += 1 else: - ckpt.verifications[cluster] = VerificationRecord(last_error=f"smoke-test: {err}") n_failed += 1 results.append( @@ -458,10 +524,10 @@ def cmd_smoke_test(args) -> int: if not ok: line += f" {err}" print(line) + _maybe_push() # Custom-weights legs (#200): re-load each base's cached weights file via # the weights= path and require agreement with the canonical run. - custom_outcomes: list[tuple[str, str, bool, str | None]] = [] for leg in custom_legs: key = (leg.env_name, leg.base) if key not in passed_keys: @@ -504,17 +570,20 @@ def cmd_smoke_test(args) -> int: if mismatch is not None: ok, err = False, f"weights= run diverges from {leg.base}: {mismatch}" elapsed = time.monotonic() - start - custom_outcomes.append((leg.env_name, leg.custom_id, ok, err)) + verification = _verification_for(ok, err, device) + # ':custom' entries are never fetched and carry no weights of their + # own: no fetch stamp, no weight capture. + _persist_outcome( + root, cluster, device, leg.env_name, leg.custom_id, verification, None, None, progress + ) + n_recorded += 1 env = manifest.environments[leg.env_name] ckpt = env.checkpoints.setdefault(leg.custom_id, CheckpointInfo()) + ckpt.verifications[cluster] = verification if ok: - ckpt.verifications[cluster] = VerificationRecord( - verified_at=now_iso(), verified_device=device - ) n_passed += 1 else: - ckpt.verifications[cluster] = VerificationRecord(last_error=f"smoke-test: {err}") n_failed += 1 results.append( @@ -540,55 +609,14 @@ def cmd_smoke_test(args) -> int: if not ok: line += f" {err}" print(line) + _maybe_push() if not json_out: for env_name, custom_id, reason in custom_skips: print(f"{env_name}/{custom_id:<24} [SKIP] {reason}") - with manifest_lock(root): - fresh = load_manifest(root) - if fresh is not None: - for env_name, ckpt_name, ok, err, weight_files, fetched_at in outcomes: - env_record = fresh.environments.get(env_name) - if env_record is None: - continue # env removed while we were testing - # Created on first pass when the resolved env has no record of - # an id it now hosts (a variant tested checkpoint-first for - # the first time, #208); the donor's fetch stamp rides along - # so the record never reads "verified but never fetched". - ckpt_record = env_record.checkpoints.setdefault(ckpt_name, CheckpointInfo()) - if ckpt_record.fetched_at is None: - ckpt_record.fetched_at = fetched_at - if ok: - ckpt_record.verifications[cluster] = VerificationRecord( - verified_at=now_iso(), verified_device=device - ) - else: - ckpt_record.verifications[cluster] = VerificationRecord( - last_error=f"smoke-test: {err}" - ) - apply_weights_record( - ckpt_record, - weight_files, - label=f"{env_name}/{ckpt_name}", - progress=None if json_out else print, - ) - for env_name, custom_id, ok, err in custom_outcomes: - env_record = fresh.environments.get(env_name) - if env_record is None: - continue # env removed while we were testing - # Created on first pass — ':custom' entries are never fetched, - # so nothing else ever writes them into the manifest. - ckpt_record = env_record.checkpoints.setdefault(custom_id, CheckpointInfo()) - if ok: - ckpt_record.verifications[cluster] = VerificationRecord( - verified_at=now_iso(), verified_device=device - ) - else: - ckpt_record.verifications[cluster] = VerificationRecord( - last_error=f"smoke-test: {err}" - ) - save_manifest(fresh, root) + # Every stamp is already on disk; this is the one push (plus the usual + # state refresh) that carries them to the backend. update_and_push_manifest(root, quiet=True, push=not no_push) total_elapsed = time.monotonic() - total_start diff --git a/tests/cli/test_smoke_test.py b/tests/cli/test_smoke_test.py index 90a83a9..093730c 100644 --- a/tests/cli/test_smoke_test.py +++ b/tests/cli/test_smoke_test.py @@ -76,6 +76,7 @@ class _Args: args.root = str(root) args.no_push = overrides.get("no_push", True) args.cluster = overrides.get("cluster") + args.push_every = overrides.get("push_every", 0) return args @@ -236,6 +237,143 @@ def test_smoke_test_empty_selection_returns_0(tmp_path, monkeypatch): assert rc == 0 +class _Killed(BaseException): + """Stand-in for the scheduler killing the job (SIGTERM/SIGKILL at the + wall-time limit) — not an Exception, so nothing in the command path + could swallow it.""" + + +def _verify_killing_at(victim: str, verdicts: dict[str, tuple[bool, str | None]] | None = None): + """A verify_checkpoint stand-in that dies when asked for ``victim``, + exactly as a wall-time kill would strike mid-verification.""" + verdicts = verdicts or {} + + def fake_verify(root, env_name, checkpoint, device, setup_kwargs, **_): + if checkpoint == victim: + raise _Killed() + return verdicts.get(checkpoint, (True, None)) + + return fake_verify + + +def test_smoke_test_persists_each_result_before_the_next_checkpoint(populated_root, monkeypatch): + """Two 12 h Delta runs (SLURM 21119813 / 21119858, 2026-08-13) hit their + wall-time limit and left no trace: results only reached the manifest + after the last checkpoint. Each outcome must land on disk as soon as it + is known, so a killed run keeps the stamps it reached.""" + monkeypatch.setattr( + smoke_module, + "verify_checkpoint", + _verify_killing_at( + "uma-s-1p1", verdicts={"mace-mp-0-medium": (False, "RuntimeError: bad")} + ), + ) + pushes: list[dict] = [] + monkeypatch.setattr( + smoke_module, "update_and_push_manifest", lambda *a, **kw: pushes.append(kw) or True + ) + + with pytest.raises(_Killed): + cmd_smoke_test(_make_args(populated_root)) + + m = load_manifest(populated_root) + small = m.environments["mace"].checkpoints["mace-mp-0-small"] + medium = m.environments["mace"].checkpoints["mace-mp-0-medium"] + uma = m.environments["uma"].checkpoints["uma-s-1p1"] + assert small.verification("test").verified_at is not None + assert "smoke-test: RuntimeError: bad" == medium.verification("test").last_error + assert uma.verification("test").verified_at is None # never got its turn + assert uma.verification("test").last_error is None + assert pushes == [] # the kill also pre-empted the single end-of-run push + + +def test_smoke_test_persists_weight_files_per_checkpoint(populated_root, monkeypatch): + files = [{"path": "cache/fake/model.bin", "size": 9_000_000}] + + def fake_verify( + root, env_name, checkpoint, device, setup_kwargs, *, weights_capture_path=None, **_ + ): + if checkpoint == "mace-mp-0-medium": + raise _Killed() + Path(weights_capture_path).write_text(json.dumps({"files": files})) + return True, None + + monkeypatch.setattr(smoke_module, "verify_checkpoint", fake_verify) + with pytest.raises(_Killed): + cmd_smoke_test(_make_args(populated_root)) + + small = load_manifest(populated_root).environments["mace"].checkpoints["mace-mp-0-small"] + assert small.weight_files == files + assert small.weights_recorded_at is not None + + +def test_smoke_test_does_not_clobber_concurrent_manifest_writes(populated_root, monkeypatch): + """Each per-checkpoint write re-reads the manifest under the lock, so a + co-maintainer's change made while a verification was running survives.""" + + def fake_verify(root, env_name, checkpoint, device, setup_kwargs, **_): + if checkpoint == "mace-mp-0-medium": + m = load_manifest(populated_root) + m.environments["uma"].checkpoints["uma-s-1p2p1"] = CheckpointInfo( + fetched_at="2026-03-01T00:00:00Z" + ) + save_manifest(m, populated_root) + return True, None + + monkeypatch.setattr(smoke_module, "verify_checkpoint", fake_verify) + assert cmd_smoke_test(_make_args(populated_root)) == 0 + + m = load_manifest(populated_root) + assert "uma-s-1p2p1" in m.environments["uma"].checkpoints + assert m.environments["mace"].checkpoints["mace-mp-0-small"].verification("test").verified_at + + +def test_smoke_test_pushes_exactly_once_by_default(populated_root, monkeypatch): + monkeypatch.setattr(smoke_module, "verify_checkpoint", lambda *a, **kw: (True, None)) + pushes: list[dict] = [] + monkeypatch.setattr( + smoke_module, "update_and_push_manifest", lambda *a, **kw: pushes.append(kw) or True + ) + + assert cmd_smoke_test(_make_args(populated_root, no_push=False)) == 0 + assert pushes == [{"quiet": True, "push": True}] + + +def test_smoke_test_push_every_adds_intermediate_pushes(populated_root, monkeypatch): + """--push-every N pushes after every N results and still once at the + end; each push is a whole-manifest snapshot of the stamps so far.""" + monkeypatch.setattr(smoke_module, "verify_checkpoint", lambda *a, **kw: (True, None)) + snapshots: list[int] = [] + + def fake_push(root, quiet, push): + m = load_manifest(root) + stamped = sum( + 1 + for env in m.environments.values() + for c in env.checkpoints.values() + if c.verification("test").verified_at + ) + snapshots.append(stamped) + return True + + monkeypatch.setattr(smoke_module, "update_and_push_manifest", fake_push) + + # 3 checkpoints selected: pushes after the 2nd, then the final one. + assert cmd_smoke_test(_make_args(populated_root, no_push=False, push_every=2)) == 0 + assert snapshots == [2, 3] + + +def test_smoke_test_no_push_suppresses_push_every(populated_root, monkeypatch): + monkeypatch.setattr(smoke_module, "verify_checkpoint", lambda *a, **kw: (True, None)) + pushes: list[dict] = [] + monkeypatch.setattr( + smoke_module, "update_and_push_manifest", lambda *a, **kw: pushes.append(kw) or True + ) + + assert cmd_smoke_test(_make_args(populated_root, no_push=True, push_every=1)) == 0 + assert pushes == [{"quiet": True, "push": False}] + + # --------------------------------------------------------------------------- # Custom-weights legs (#200): each ':custom' entry re-loads a # same-family checkpoint's cached weights via weights= and must agree. @@ -417,6 +555,35 @@ def test_custom_leg_divergence_fails_and_records_error(custom_root, monkeypatch) assert base.verification("test").verified_at is not None +def test_custom_leg_result_is_persisted_before_the_next_leg(custom_root, monkeypatch): + """The custom legs run after every canonical checkpoint, i.e. latest in + the job — exactly where a wall-time kill lands. Each leg's stamp must be + on disk before the next leg starts.""" + calls: list[dict] = [] + inner = _fake_verify_factory(calls) + + def fake_verify(root, env_name, checkpoint, device, setup_kwargs, **kw): + if checkpoint == "mace-off:custom": + raise _Killed() + return inner(root, env_name, checkpoint, device, setup_kwargs, **kw) + + monkeypatch.setattr(smoke_module, "verify_checkpoint", fake_verify) + with pytest.raises(_Killed): + cmd_smoke_test(_make_args(custom_root)) + + m = load_manifest(custom_root) + assert m.environments["uma"].checkpoints["uma:custom"].verification("test").verified_at + assert m.environments["mace"].checkpoints["mace:custom"].verification("test").verified_at + assert "mace-off:custom" not in m.environments["mace"].checkpoints + # Every canonical baseline landed too. + for env, ckpt in ( + ("uma", "uma-s-1p1"), + ("mace", "mace-mp-0-small"), + ("mace", "mace-off23-small"), + ): + assert m.environments[env].checkpoints[ckpt].verification("test").verified_at + + def test_custom_leg_skipped_when_baseline_fails(custom_root, monkeypatch, capsys): calls: list[dict] = [] monkeypatch.setattr( diff --git a/tests/cli/test_smoke_test_clusters.py b/tests/cli/test_smoke_test_clusters.py index e5fc9be..d7cf6f8 100644 --- a/tests/cli/test_smoke_test_clusters.py +++ b/tests/cli/test_smoke_test_clusters.py @@ -118,6 +118,7 @@ class _Args: args.root = str(root) args.no_push = True args.cluster = overrides.get("cluster") + args.push_every = 0 return args