Thin US calibration epoch telemetry to the UK's rule

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.

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 `<out>/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/<run_id>/` 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 `<out>/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/<run_id>/` 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.
 

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"])

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,25 +88,13 @@
 # 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",
     "failed": "dataset_stage_failed",
     "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:

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

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
 ):