diff --git a/changelog.d/1292-us-calibration-epoch-thinning.changed.md b/changelog.d/1292-us-calibration-epoch-thinning.changed.md new file mode 100644 index 0000000000..0cd0c6255f --- /dev/null +++ b/changelog.d/1292-us-calibration-epoch-thinning.changed.md @@ -0,0 +1 @@ +US builds now forward calibration epochs to their staging bundle and hosted telemetry the way UK builds do: each solve's first and last epoch and every tenth epoch, with the stride growing so no run forwards more than 2,400 epochs on the stride beside those first and last epochs. Previously every epoch was forwarded. UK builds now also forward each solve's first epoch, one more row per solve, so the forwarded epochs fall at every boundary between solves and a dashboard that starts a new solve when the epoch falls no longer merges solves of ten or fewer epochs. The rule is defined once in `microcosm.build.calibration_epoch_telemetry` and both countries use it. diff --git a/docs/uk-full-build-graph.md b/docs/uk-full-build-graph.md index ed179cdca5..c046c7f208 100644 --- a/docs/uk-full-build-graph.md +++ b/docs/uk-full-build-graph.md @@ -116,7 +116,7 @@ The graph evaluates every declared UK gate once; `uk.full.gates.calibrated.gate_ The terminal graph node writes unsigned `certification.json` from those identified artifacts and any declared native or matched-size comparisons. It is non-gating (microcosm#1301): no release step reads it, a scorecard that was not supplied is `not_requested`, one whose identities hold but whose measured limits miss is `diagnostic_failed`, and a holdout that was not requested is not a readiness failure. `build.json` is the completion marker that binds both `candidate.json` and the certification artifact. Physical output bytes are checked against their declared artifacts. Bundle publication writes the completion marker last and rolls back handled failures or interrupts. A process kill or power loss can leave an absent completion marker; a directory without a valid bound marker is not a completed build. -A non-dry dense run is wrapped in the rowwise tool's operational envelope: the Logbook attempt (a `uk-local-candidate` row spooled under `/logbook-spool` on every terminal outcome, chained through `--logbook-prev-row-digest` or `POPULACE_LOGBOOK_PREV_ROW_DIGEST`, with an error receipt on failure), version 3 staging telemetry with the run's sampling evidence (`sample: {"mode": "full"}` on the f100 rung, null below it, judged on the effective fraction), stage events around each graph phase and per-epoch `calibration_progress` rows from the dense solve and, with `--dataset-households`, the size search and refit (rows tagged with their `phase`), and the staged-dataset delivery of the published bundle under `staged//` in the private repository. `--staging-local-only`, `--no-staging`, `--staging-read-back` and `--no-staged-dataset` behave as in [UK staging operations](uk-staging-operations.md). Dry runs plan without solving or writing and record no Logbook row. +A non-dry dense run is wrapped in the rowwise tool's operational envelope: the Logbook attempt (a `uk-local-candidate` row spooled under `/logbook-spool` on every terminal outcome, chained through `--logbook-prev-row-digest` or `POPULACE_LOGBOOK_PREV_ROW_DIGEST`, with an error receipt on failure), version 3 staging telemetry with the run's sampling evidence (`sample: {"mode": "full"}` on the f100 rung, null below it, judged on the effective fraction), stage events around each graph phase and `calibration_progress` rows for each solve's first and last epoch and every tenth epoch (the rule every build shares, in `microcosm.build.calibration_epoch_telemetry`) from the dense solve and, with `--dataset-households`, the size search and refit (rows tagged with their `phase`), and the staged-dataset delivery of the published bundle under `staged//` in the private repository. `--staging-local-only`, `--no-staging`, `--staging-read-back` and `--no-staged-dataset` behave as in [UK staging operations](uk-staging-operations.md). Dry runs plan without solving or writing and record no Logbook row. Structural failures stop export. The maintained local statistical failure policy may still export an unreleasable diagnostic candidate with a nonzero process status. Missing evidence remains explicit. `--release-candidate` applies the maintained strictness and solve settings and requires the UK gate signing key and local targets, because it signs the local gate report; it does not publish or authorize a release. Fixture acceptance proves graph behavior. Measured incumbent comparison evidence is an optional diagnostic supplied with `--native-scorecard` (and, at the requested k, `--matched-size-scorecard`); the certification node verifies their candidate and output identities before recording their assessment. diff --git a/packages/microcosm-build/src/microcosm/build/calibration_epoch_telemetry.py b/packages/microcosm-build/src/microcosm/build/calibration_epoch_telemetry.py new file mode 100644 index 0000000000..286eabea38 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/calibration_epoch_telemetry.py @@ -0,0 +1,70 @@ +"""Which calibration epochs a build forwards to its telemetry. + +The calibrator reports one ``calibration_epoch`` event per epoch. Every build, +UK and US, forwards the same subset to its staging bundle and its telemetry +emitter (microcosm#1292): the first and last epoch of each solve and every +``CALIBRATION_EPOCH_EVERY``-th epoch, so the loss curve and every solve's +starting and final loss survive at a tenth of the rows. + +Each solve's epoch counter restarts at 1, and forwarding that first epoch +keeps the restart visible: the forwarded epochs fall at every boundary +between solves, as the full per-epoch stream does. A reader that starts a +new solve when the epoch falls (the calibration-diagnostics build monitor +does) therefore sees every solve, even when ``--epochs`` is no longer than +the stride. + +A UK size run at 2,000 epochs solves the dense pool, up to ten full-length L0 +probes and the refit: about 24,000 epochs. One forwarded epoch is one row in +the staging bundle's ``calibration_progress.json`` and one event in its +``events.ndjson``, both under the staging contract's 5 MiB remote cap, so the +stride also grows with the run: no run forwards more than +``CALIBRATION_MAX_EPOCH_ROWS`` epochs on the stride, whatever ``--epochs`` +says. Each solve's first and last epoch can add one row each, so a run of +``solves`` solves forwards at most ``CALIBRATION_MAX_EPOCH_ROWS + 2 * solves`` +rows. +""" + +from __future__ import annotations + +from collections.abc import Mapping + +__all__ = [ + "CALIBRATION_EPOCH_EVERY", + "CALIBRATION_MAX_EPOCH_ROWS", + "calibration_epoch_stride", + "forwards_calibration_epoch", +] + +CALIBRATION_EPOCH_EVERY = 10 +CALIBRATION_MAX_EPOCH_ROWS = 2400 + + +def calibration_epoch_stride(total_epochs: int) -> int: + """The forwarding stride for a run whose solves total ``total_epochs``. + + At least ``CALIBRATION_EPOCH_EVERY``, and large enough that the epochs on + the stride stay within ``CALIBRATION_MAX_EPOCH_ROWS``; each solve's first + and last epoch are forwarded beside them. A caller that cannot know its + exact solve count (a budget search may stop early) passes the most epochs + the run can solve. + """ + + return max( + CALIBRATION_EPOCH_EVERY, + -(-int(total_epochs) // CALIBRATION_MAX_EPOCH_ROWS), + ) + + +def forwards_calibration_epoch(event: Mapping[str, object], *, every: int) -> bool: + """Whether a build forwards this calibrator report at stride ``every``. + + Only ``calibration_epoch`` reports are forwarded: each solve's first and + last epoch and every ``every``-th epoch. A solve's epoch counter restarts + at 1, so a selection and its refit each keep their first and final epoch, + and the forwarded epochs fall where the refit begins. + """ + + if event.get("kind") != "calibration_epoch": + return False + epoch = int(event["epoch"]) + return epoch == 1 or epoch % every == 0 or epoch == int(event["epochs"]) diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/rowwise_staging.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/rowwise_staging.py index f7d6584899..8ff62fe8dc 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/rowwise_staging.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/rowwise_staging.py @@ -22,6 +22,11 @@ from pathlib import Path from typing import Any +from microcosm.build.calibration_epoch_telemetry import ( + CALIBRATION_EPOCH_EVERY, + calibration_epoch_stride, + forwards_calibration_epoch, +) from microcosm.build.run_outcome import ( UNRECORDED_GATE_BLOCK, GateBlock, @@ -55,8 +60,6 @@ __all__ = [ "STAGED_DATASET_PHASES", - "STAGING_EPOCH_EVERY", - "STAGING_MAX_EPOCH_ROWS", "STAGING_UPLOAD_INTERVAL_SECONDS", "add_staging_artifact", "block_staging_run_bundle", @@ -85,16 +88,6 @@ # and loses uploads (the v20 national run did); five minutes keeps a # 1,500-epoch run well inside it. STAGING_UPLOAD_INTERVAL_SECONDS = 300.0 -# The staging bundle keeps one row per forwarded epoch in -# calibration_progress.json and one event in events.ndjson, both under the -# contract's 5 MiB remote cap. A size run at 2,000 epochs solves the dense -# pool, up to ten full-length L0 probes and the refit: about 24,000 epochs, -# which would breach the cap mid-run. Forwarding every tenth epoch and the -# last epoch of each phase keeps the loss curve and stays near 0.8 MB. -STAGING_EPOCH_EVERY = 10 -# ...and never more than this many forwarded epochs per run, whatever --epochs -# says: the stride grows with the run so the cap holds by construction. -STAGING_MAX_EPOCH_ROWS = 2400 STAGED_DATASET_PHASES = { "uploaded": "dataset_staged", "already_staged": "dataset_staged", @@ -102,8 +95,6 @@ "skipped": "dataset_stage_skipped", } _STAGING_UPLOAD_INTERVAL_SECONDS = STAGING_UPLOAD_INTERVAL_SECONDS -_STAGING_EPOCH_EVERY = STAGING_EPOCH_EVERY -_STAGING_MAX_EPOCH_ROWS = STAGING_MAX_EPOCH_ROWS _STAGED_DATASET_PHASES = STAGED_DATASET_PHASES _ACTIVE_EMITTER: LocalTelemetryEmitter | None = None @@ -354,18 +345,17 @@ def stage_sample( def staging_epoch_every(args: argparse.Namespace) -> int: - """The epoch stride that keeps the forwarded rows under the row budget. + """The shared calibration epoch stride over this run's solves. A dense run solves once; a size run solves the pool, up to ``budget_iters`` - full-length probes and the refit. The stride is at least - ``STAGING_EPOCH_EVERY`` and grows so at most ``STAGING_MAX_EPOCH_ROWS`` - epochs are forwarded, keeping ``calibration_progress.json`` and + full-length probes and the refit. The stride grows with that total, as + :func:`~microcosm.build.calibration_epoch_telemetry.calibration_epoch_stride` + defines for every build, keeping ``calibration_progress.json`` and ``events.ndjson`` under the contract's 5 MiB cap for any ``--epochs``. """ solves = 1 if args.dataset_households is None else 2 + BUDGET_ITERS - total = int(args.epochs) * solves - return max(STAGING_EPOCH_EVERY, -(-total // STAGING_MAX_EPOCH_ROWS)) + return calibration_epoch_stride(int(args.epochs) * solves) _staging_epoch_every = staging_epoch_every @@ -374,10 +364,14 @@ def staging_epoch_every(args: argparse.Namespace) -> int: def thinned_epochs( sink: Callable[[Mapping[str, Any]], None], *, - every: int = STAGING_EPOCH_EVERY, + every: int = CALIBRATION_EPOCH_EVERY, withhold_loss: bool = False, ) -> Callable[[dict[str, object]], None]: - """Forward every ``every``-th epoch and each phase's last epoch to ``sink``. + """Forward the shared rule's calibration epochs to ``sink``. + + :func:`~microcosm.build.calibration_epoch_telemetry.forwards_calibration_epoch` + selects each phase's first and last epoch and every ``every``-th epoch, as + for every build. The kernel flags probe epochs with ``budget_search: True``; the staging contract records that field as an integer or null, so the flag becomes 1 @@ -394,11 +388,7 @@ def thinned_epochs( def callback(event: dict[str, object]) -> None: nonlocal disabled - if disabled or event.get("kind") != "calibration_epoch": - return - epoch = int(event["epoch"]) - epochs = int(event["epochs"]) - if epoch % every != 0 and epoch != epochs: + if disabled or not forwards_calibration_epoch(event, every=every): return forwarded = dict(event) if withhold_loss: diff --git a/packages/microcosm-build/tests/engine_free/shared/test_calibration_epoch_telemetry.py b/packages/microcosm-build/tests/engine_free/shared/test_calibration_epoch_telemetry.py new file mode 100644 index 0000000000..d0cfc0c525 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_calibration_epoch_telemetry.py @@ -0,0 +1,117 @@ +"""The calibration epochs every build forwards to its telemetry (microcosm#1292).""" + +import pytest + +from microcosm.build.calibration_epoch_telemetry import ( + CALIBRATION_EPOCH_EVERY, + CALIBRATION_MAX_EPOCH_ROWS, + calibration_epoch_stride, + forwards_calibration_epoch, +) + + +def _forwarded_epochs(epochs: int, *, every: int) -> list[int]: + return [ + epoch + for epoch in range(1, epochs + 1) + if forwards_calibration_epoch( + { + "kind": "calibration_epoch", + "epoch": epoch, + "epochs": epochs, + "loss": 1.0, + }, + every=every, + ) + ] + + +def test_calibration_epoch_rule_forwards_each_solve_first_last_and_every_tenth_epoch(): + assert CALIBRATION_EPOCH_EVERY == 10 + assert _forwarded_epochs(25, every=CALIBRATION_EPOCH_EVERY) == [1, 10, 20, 25] + # A last epoch on the stride is forwarded once. + assert _forwarded_epochs(30, every=CALIBRATION_EPOCH_EVERY) == [1, 10, 20, 30] + # A solve shorter than the stride still reports its first and final loss. + assert _forwarded_epochs(5, every=CALIBRATION_EPOCH_EVERY) == [1, 5] + # A one-epoch solve's first epoch is its last, forwarded once. + assert _forwarded_epochs(1, every=CALIBRATION_EPOCH_EVERY) == [1] + assert _forwarded_epochs(40, every=13) == [1, 13, 26, 39, 40] + + +def test_forwarded_calibration_epochs_fall_at_every_solve_boundary(): + """A reader can tell consecutive solves apart from the forwarded rows alone. + + A selection and its refit run the same number of epochs, and each solve's + epoch counter restarts at 1. The calibration-diagnostics build monitor + starts a new solver pass when a row's epoch is lower than the row before + it (or its epoch total differs), so the forwarded epochs must fall at each + boundary, as the full per-epoch stream does, even when both solves are + shorter than the stride. + """ + + boundaries_without_a_fall = [] + for every in (CALIBRATION_EPOCH_EVERY, 13): + for epochs in range(2, 41): + # Both solves forward the same epochs: the refit's first forwarded + # row directly follows the selection's last. + forwarded = _forwarded_epochs(epochs, every=every) + if forwarded[0] >= forwarded[-1]: + boundaries_without_a_fall.append((epochs, every)) + + assert boundaries_without_a_fall == [] + + +def test_calibration_epoch_rule_ignores_other_progress_events(): + # Budget-search reports share the progress callback but are not epochs; + # the closing report carries no epoch at all. + assert not forwards_calibration_epoch( + {"kind": "budget_probe", "epoch": 10, "epochs": 10}, every=10 + ) + assert not forwards_calibration_epoch({"kind": "budget_search_done"}, every=10) + + +@pytest.mark.parametrize( + ("total_epochs", "expected_stride"), + [ + (1, 10), + (3_000, 10), + (24_000, 10), + (24_001, 11), + (30_000, 13), + (120_000, 50), + ], +) +def test_calibration_epoch_stride_grows_to_keep_the_strided_epochs_under_the_cap( + total_epochs, expected_stride +): + assert CALIBRATION_MAX_EPOCH_ROWS == 2_400 + stride = calibration_epoch_stride(total_epochs) + assert stride == expected_stride + assert total_epochs // stride <= CALIBRATION_MAX_EPOCH_ROWS + + +@pytest.mark.parametrize( + ("epochs_per_solve", "solves", "expected_rows"), + [ + # The default US run: an L0 selection and its refit. + (1_500, 2, 302), + # The default UK size run: the dense solve, ten budget-search probes + # and the refit, 2,400 epochs on the stride and each solve's first. + (2_000, 12, 2_412), + (15_000, 2, 2_310), + (30_000, 1, 2_309), + # Last epochs off the stride took even the rule without first epochs + # past a flat 2,400 rows (to 2,401 here). + (3_428, 7, 2_408), + ], +) +def test_forwarded_rows_stay_within_the_cap_plus_two_per_solve( + epochs_per_solve, solves, expected_rows +): + """The stride caps the strided epochs; each solve adds its first and last.""" + + every = calibration_epoch_stride(epochs_per_solve * solves) + forwarded_rows = solves * len(_forwarded_epochs(epochs_per_solve, every=every)) + + assert forwarded_rows == expected_rows + assert forwarded_rows <= CALIBRATION_MAX_EPOCH_ROWS + 2 * solves diff --git a/packages/microcosm-build/tests/engine_free/uk/test_uk_rowwise_candidate.py b/packages/microcosm-build/tests/engine_free/uk/test_uk_rowwise_candidate.py index 86ce48b765..371b31e9a0 100644 --- a/packages/microcosm-build/tests/engine_free/uk/test_uk_rowwise_candidate.py +++ b/packages/microcosm-build/tests/engine_free/uk/test_uk_rowwise_candidate.py @@ -7,6 +7,7 @@ """ # ruff: noqa: F403, F405 +from microcosm.build.calibration_epoch_telemetry import CALIBRATION_MAX_EPOCH_ROWS from test_support.microcosm_build.uk_rowwise_candidate import * @@ -567,17 +568,73 @@ def stride(epochs, households): assert stride(2, None) == 10 and stride(2, 300) == 10 assert stride(2000, None) == 10 # dense + ten probes + refit at 2,000 epochs: 24,000 epochs -> 2,400 rows + # on the stride, plus each of the twelve solves' first epoch assert stride(2000, 55000) == 10 assert stride(10000, 55000) == 50 assert stride(100000, None) == 42 for epochs, households in ((2000, 55000), (10000, 55000), (100000, None)): solves = 1 if households is None else 2 + rowwise_cli.BUDGET_ITERS assert ( - epochs * solves / stride(epochs, households) - <= builder._STAGING_MAX_EPOCH_ROWS + epochs * solves / stride(epochs, households) <= CALIBRATION_MAX_EPOCH_ROWS ) +def test_uk_thinned_epochs_forward_each_solve_first_epoch(): + """A UK size run no longer than the stride keeps its solves apart. + + The dense solve, each budget-search probe and the refit all run + ``--epochs`` and restart the epoch counter at 1. Forwarding each solve's + first epoch beside its last makes the epoch fall at every boundary, which + is how the calibration-diagnostics build monitor starts a new solver pass + (microcosm#1292). + """ + + epochs = 5 + forwarded_rows: list[dict[str, object]] = [] + observer = rowwise_staging.thinned_epochs( + forwarded_rows.append, + every=rowwise_staging.staging_epoch_every( + SimpleNamespace(epochs=epochs, dataset_households=55000) + ), + ) + # The dense solve's reports carry no phase; probe epochs are flagged. + solve_contexts = [ + {}, + {"phase": "size_search", "budget_search": True}, + {"phase": "size_search", "budget_search": True}, + {"phase": "size_refit"}, + ] + + for solve_context in solve_contexts: + for epoch in range(1, epochs + 1): + observer( + { + **solve_context, + "kind": "calibration_epoch", + "epoch": epoch, + "epochs": epochs, + "loss": 1.0 / epoch, + } + ) + + assert [(row.get("phase"), row["epoch"]) for row in forwarded_rows] == [ + (solve_context.get("phase"), epoch) + for solve_context in solve_contexts + for epoch in (1, epochs) + ] + # The staging contract records the probe flag as 1 or null. + assert [row.get("budget_search") for row in forwarded_rows] == [ + None, + None, + 1, + 1, + 1, + 1, + None, + None, + ] + + def test_remote_dataset_staging_is_refused_up_front_without_credential_or_repo( monkeypatch, tmp_path, capsys ): diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py index a1d494147c..9373d815ed 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py @@ -13755,6 +13755,8 @@ def namespace(no_staging: bool) -> SimpleNamespace: staging_run_id=None, staging_prefix=args.staging_prefix, staging_upload_interval_seconds=60.0, + epochs=module.DEFAULT_US_FISCAL_CALIBRATION_EPOCHS, + dense_default_dataset=False, ) progress = module._build_progress( @@ -13804,6 +13806,191 @@ def stage(self, *args, **kwargs): ] +def _solver_epoch_events(phase: str, epochs: int) -> list[dict[str, object]]: + """One solve's per-epoch reports, as the calibrator emits them.""" + + return [ + { + "phase": phase, + "kind": "calibration_epoch", + "epoch": epoch, + "epochs": epochs, + "loss": 100.0 / epoch, + } + for epoch in range(1, epochs + 1) + ] + + +def _forwarded_calibration_events(emitter: _TestEmitter) -> list[dict[str, object]]: + return [recorded[1] for recorded in emitter.events if recorded[0] == "calibration"] + + +def test_us_build_progress_forwards_each_solve_first_final_and_every_tenth_epoch( + tmp_path, +) -> None: + """The US forwards the UK's epochs: each solve's first and last, every tenth. + + The solver reports every epoch (microcosm#1292). An L0 selection and its + refit are two solves whose epoch counters each restart at 1, so each + keeps its own first and final epoch, and both the hosted emitter and the + staging bundle receive the same thinned rows. + """ + + module = _load_builder_module() + emitter = _TestEmitter() + run_dir = tmp_path / "stage" + progress = module._BuildProgress( + run_id="rel-1", + emitter=emitter, + staging_bundle=module.StagingRunBundleWriter( + run_id="rel-1", + candidate_release_id="rel-1", + run_dir=run_dir, + ), + ) + solver_reports = [ + *_solver_epoch_events("l0_selection", 25), + {"kind": "budget_probe", "budget_iteration": 1}, + *_solver_epoch_events("post_l0_refit", 25), + ] + + for report in solver_reports: + progress.calibration_progress(report) + + expected_rows = [ + (phase, epoch, 100.0 / epoch) + for phase in ("l0_selection", "post_l0_refit") + for epoch in (1, 10, 20, 25) + ] + forwarded = _forwarded_calibration_events(emitter) + assert [ + (event.get("phase"), event.get("epoch"), event.get("loss")) + for event in forwarded + ] == expected_rows + staged_rows = json.loads((run_dir / "calibration_progress.json").read_text())[ + "events" + ] + assert [(row["epoch"], row["epochs"], row["loss"]) for row in staged_rows] == [ + (epoch, 25, loss) for _phase, epoch, loss in expected_rows + ] + latest = json.loads((run_dir / "progress.json").read_text())["calibration"] + assert (latest["epoch"], latest["epochs"], latest["loss"]) == (25, 25, 4.0) + + +@pytest.mark.parametrize("epochs", [10, 5]) +@pytest.mark.parametrize( + "solve_phases", + [("l0_selection", "post_l0_refit"), ("l0_selection", "exact_k_refit")], + ids=["l0_refit", "exact_k_below_the_pool"], +) +def test_us_short_run_forwards_each_solve_first_epoch_so_the_epoch_falls_at_the_refit( + tmp_path, solve_phases, epochs +) -> None: + """A run no longer than the stride still shows both of its solves. + + With ``--epochs`` of 10 or fewer, no epoch but the last is a multiple of + the stride, so without each solve's first epoch an L0 run, or an exact-k + run whose k is below the pool, forwarded ``(epochs, epochs)`` twice. The + calibration-diagnostics build monitor starts a new solver pass only when + the epoch falls or the epoch total changes, so it merged the selection and + its refit into one pass with no start time it could back-date + (microcosm#1292). + """ + + module = _load_builder_module() + emitter = _TestEmitter() + run_dir = tmp_path / "stage" + args = SimpleNamespace( + no_staging=False, + staging_dir=run_dir, + staging_repo_id=None, + staging_run_id=None, + staging_prefix=module.DEFAULT_STAGING_PREFIX, + staging_upload_interval_seconds=60.0, + epochs=epochs, + dense_default_dataset=False, + ) + progress = module._build_progress( + args, + release_root=tmp_path, + release_id="rel-1", + emitter=emitter, + ) + + for phase in solve_phases: + for report in _solver_epoch_events(phase, epochs): + progress.calibration_progress(report) + + assert [ + (event["phase"], event["epoch"]) + for event in _forwarded_calibration_events(emitter) + ] == [(phase, epoch) for phase in solve_phases for epoch in (1, epochs)] + staged_epochs = [ + row["epoch"] + for row in json.loads((run_dir / "calibration_progress.json").read_text())[ + "events" + ] + ] + assert staged_epochs == [1, epochs, 1, epochs] + refit_first_row = 2 + assert staged_epochs[refit_first_row] < staged_epochs[refit_first_row - 1] + + +@pytest.mark.parametrize( + ("epochs", "dense_default_dataset", "expected_stride", "expected_rows"), + [ + # The default L0 selection and refit: 2 x 1,500 epochs. + (1_500, False, 10, 302), + # 2 x 15,000 epochs would be 3,000 rows at the base stride. + (15_000, False, 13, 2_310), + # A single dense solve of the same total length. + (30_000, True, 13, 2_309), + ], +) +def test_us_build_progress_stride_keeps_long_runs_under_the_row_cap( + tmp_path, epochs, dense_default_dataset, expected_stride, expected_rows +) -> None: + """The stride grows with the run's solves, as the UK's does. + + No run forwards more than the UK's 2,400 epochs on the stride plus each + solve's first and last epoch, whatever ``--epochs`` says (microcosm#1292). + """ + + module = _load_builder_module() + emitter = _TestEmitter() + args = SimpleNamespace( + no_staging=True, + staging_dir=None, + staging_repo_id=None, + staging_run_id=None, + staging_prefix=module.DEFAULT_STAGING_PREFIX, + staging_upload_interval_seconds=60.0, + epochs=epochs, + dense_default_dataset=dense_default_dataset, + ) + progress = module._build_progress( + args, + release_root=tmp_path, + release_id="rel-1", + emitter=emitter, + ) + phases = ("dense",) if dense_default_dataset else ("l0_selection", "post_l0_refit") + + for phase in phases: + for report in _solver_epoch_events(phase, epochs): + progress.calibration_progress(report) + + forwarded = _forwarded_calibration_events(emitter) + assert len(forwarded) == expected_rows + assert len(forwarded) <= 2_400 + 2 * len(phases) + for phase in phases: + assert [event["epoch"] for event in forwarded if event["phase"] == phase] == [ + epoch + for epoch in range(1, epochs + 1) + if epoch == 1 or epoch % expected_stride == 0 or epoch == epochs + ] + + def test_blank_staging_repo_id_is_refused_at_parse_time(monkeypatch, capsys) -> None: module = _load_builder_module() argv = [ @@ -13893,6 +14080,8 @@ def test_build_progress_replaces_any_previous_active_run(tmp_path) -> None: staging_run_id=None, staging_prefix=module.DEFAULT_STAGING_PREFIX, staging_upload_interval_seconds=60.0, + epochs=module.DEFAULT_US_FISCAL_CALIBRATION_EPOCHS, + dense_default_dataset=False, ) first = module._build_progress( args, @@ -13958,6 +14147,8 @@ def test_staging_bundle_refuses_a_destinationless_namespace(tmp_path) -> None: staging_run_id=None, staging_prefix=module.DEFAULT_STAGING_PREFIX, staging_upload_interval_seconds=60.0, + epochs=module.DEFAULT_US_FISCAL_CALIBRATION_EPOCHS, + dense_default_dataset=False, ) with pytest.raises(ValueError, match="no destination"): diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index be88543d84..44bebf56c4 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -54,6 +54,11 @@ import numpy as np import pandas as pd +from microcosm.build.calibration_epoch_telemetry import ( + CALIBRATION_EPOCH_EVERY, + calibration_epoch_stride, + forwards_calibration_epoch, +) from microcosm.build.gates import ( GateResult, TargetFitRequirement, @@ -11504,8 +11509,25 @@ def _assert_exact_k_original_pool_alignment( _ACTIVE_EMITTER: LocalTelemetryEmitter | None = None +def _calibration_epoch_every(args: argparse.Namespace) -> int: + """The shared calibration epoch stride over this build's solves. + + ``--dense-default-dataset`` solves once; the L0 and exact-k paths solve a + selection and then its refit, each for ``--epochs`` (an exact-k ladder + whose ``k`` is the whole pool solves once, so two is the most it solves). + """ + + solves = 1 if args.dense_default_dataset else 2 + return calibration_epoch_stride(int(args.epochs) * solves) + + class _BuildProgress: - """Report progress to an emitter and, independently, a staging bundle.""" + """Report progress to an emitter and, independently, a staging bundle. + + Both receive the same calibration epochs: each solve's first and last and + every ``epoch_every``-th epoch, the rule the UK build follows too + (microcosm#1292). + """ def __init__( self, @@ -11514,11 +11536,13 @@ def __init__( emitter: LocalTelemetryEmitter, staging_bundle: StagingRunBundleWriter | None, staging_opt_out_reason: str | None = None, + epoch_every: int = CALIBRATION_EPOCH_EVERY, ) -> None: self.run_id = run_id self.emitter = emitter self.staging_bundle = staging_bundle self.staging_opt_out_reason = staging_opt_out_reason + self.epoch_every = epoch_every @property def repo_id(self) -> str | None: @@ -11557,6 +11581,8 @@ def stage( ) def calibration_progress(self, event: dict[str, object]) -> None: + if not forwards_calibration_epoch(event, every=self.epoch_every): + return self.emitter.transition_calibration_progress(event) if self.staging_bundle is not None: self.staging_bundle.calibration_progress(event) @@ -12249,12 +12275,14 @@ def _build_progress( # can never be marked failed in place of this build's. _ACTIVE_PROGRESS = None run_id = run_id or args.staging_run_id or release_id + epoch_every = _calibration_epoch_every(args) if args.no_staging: _ACTIVE_PROGRESS = _BuildProgress( run_id=run_id, emitter=emitter, staging_bundle=None, staging_opt_out_reason="--no-staging", + epoch_every=epoch_every, ) return _ACTIVE_PROGRESS if not args.staging_dir and not args.staging_repo_id: @@ -12279,6 +12307,7 @@ def _build_progress( run_id=run_id, emitter=emitter, staging_bundle=staging_bundle, + epoch_every=epoch_every, ) return _ACTIVE_PROGRESS @@ -12790,6 +12819,7 @@ def _main(argv: Sequence[str] | None = None) -> int | None: emitter=_ACTIVE_EMITTER, staging_bundle=None, staging_opt_out_reason="dry run", + epoch_every=_calibration_epoch_every(args), ) _ACTIVE_PROGRESS = telemetry if telemetry is not None: diff --git a/tools/route_a/README.md b/tools/route_a/README.md index 86eb1733d7..ec20bfa76d 100644 --- a/tools/route_a/README.md +++ b/tools/route_a/README.md @@ -212,10 +212,16 @@ uploads reach the dashboard's staging repository, the records for its Build progress tab include the release's run ID and candidate release ID, current status/stage, timestamps, messages and stage details from `progress.json`, the event history from `events.ndjson`, and calibration epoch/loss/budget-search information from -`calibration_progress.json`. The builder emits stages for loading the base, -source enrichments, target compilation, calibration, export/gates, diagnostics -and manifests, and uploads attached diagnostics as they are written. Completion -records status `passed`; reported failures record status `failed`. +`calibration_progress.json`. That file and the hosted telemetry carry each +solve's first and last epoch and every tenth epoch between them, the rule the +UK build follows too (`microcosm.build.calibration_epoch_telemetry`); the +stride grows past ten only when the run's solves total more than 24,000 +epochs. Because each solve's rows start at epoch 1, the epoch falls wherever +the refit begins, even when `--epochs` is ten or fewer. The builder emits +stages for loading the base, source enrichments, target compilation, +calibration, export/gates, diagnostics and manifests, and uploads attached +diagnostics as they are written. Completion records status `passed`; reported +failures record status `failed`. Telemetry is written locally under `/staging/runs//` and uploaded under