diff --git a/CLAUDE.md b/CLAUDE.md index 2e433d68..36aa7897 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -36,7 +36,10 @@ Builds, calibrations, and releases run outside PR CI, need gated Hugging Face data and credentials, and cannot run from forks. Release publication is a deliberate human step (`tools/publish_release.sh` → `microcosm-publish-release`), gated by `tools/preflight_us_release_gates.py`; -see README "Releasing & alerts". Never publish or promote artifacts as a side +see README "Releasing & alerts". Publication also refuses a release whose +build recorded staging telemetry that never reached its repo +(`--allow-missing-staging` overrides); a build that declared `--no-staging` +publishes without the flag. Never publish or promote artifacts as a side effect of another task. ## Root journals are history, not state diff --git a/README.md b/README.md index ca829be3..e9fbeeb4 100644 --- a/README.md +++ b/README.md @@ -43,8 +43,15 @@ progress JSON is uploaded to `policyengine/populace-us-staging` while the build runs (best-effort — a missing token or failed upload never fails the build), so every candidate shows up on the staging dashboard before it is published. Disable with `--no-staging`, or point elsewhere with `--staging-repo-id` / -`POPULACE_STAGING_REPO_ID`. The build manifest records the staging run id, and -`microcosm-publish-release` warns when publishing a release that has none: +`POPULACE_STAGING_REPO_ID`. An *empty* `POPULACE_STAGING_REPO_ID` is ignored +rather than read as off, and staging with no destination at all is an argparse +error — `--no-staging` is the only way a build produces no telemetry. + +The build manifest records what staging did: the run id, the destination, and +how many files actually reached it, or an explicit `enabled: false` for a +declared opt-out. `microcosm-publish-release` **refuses** a release whose build +meant to stage and delivered nothing (`--allow-missing-staging` overrides); a +declared `--no-staging` build publishes without the flag: ```bash python tools/build_us_fiscal_refresh_release.py \ diff --git a/changelog.d/staging-env-563.fixed.md b/changelog.d/staging-env-563.fixed.md new file mode 100644 index 00000000..cc3129f4 --- /dev/null +++ b/changelog.d/staging-env-563.fixed.md @@ -0,0 +1 @@ +Treat an empty POPULACE_STAGING_REPO_ID or POPULACE_STAGING_PREFIX as unset rather than as off, make --no-staging the only way a build produces no staging telemetry, record in the build manifest whether staging was skipped deliberately or meant to run and delivered nothing, refuse to publish a release in the latter case (--allow-missing-staging overrides), and mark a staging run failed when the build does not finish. diff --git a/packages/microcosm-build/src/microcosm/build/__init__.py b/packages/microcosm-build/src/microcosm/build/__init__.py index 02060675..8c17f556 100644 --- a/packages/microcosm-build/src/microcosm/build/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/__init__.py @@ -132,6 +132,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None: run_source_stage, ) from microcosm.build.staging import ( # noqa: E402 - after the compat gate + DEFAULT_STAGING_PREFIX, LATEST_STAGING_POINTER, RUNS_INDEX, STAGING_SCHEMA_VERSION, @@ -144,6 +145,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None: "BlockingMode", "CountrySpec", "DonorSpec", + "DEFAULT_STAGING_PREFIX", "EvidenceContext", "FitWeightRecord", "FunctionBinding", diff --git a/packages/microcosm-build/src/microcosm/build/staging.py b/packages/microcosm-build/src/microcosm/build/staging.py index e91c4e1e..c8067b39 100644 --- a/packages/microcosm-build/src/microcosm/build/staging.py +++ b/packages/microcosm-build/src/microcosm/build/staging.py @@ -23,6 +23,7 @@ STAGING_SCHEMA_VERSION = 1 LATEST_STAGING_POINTER = "latest_staging.json" RUNS_INDEX = "runs.json" +DEFAULT_STAGING_PREFIX = "runs" def _now() -> str: @@ -71,7 +72,7 @@ class StagingTelemetry: candidate_release_id: str run_dir: Path | str repo_id: str | None = None - path_prefix: str = "runs" + path_prefix: str = DEFAULT_STAGING_PREFIX api: Any = None upload_interval_seconds: float = 30.0 started_at: str = field(default_factory=_now) @@ -79,8 +80,16 @@ class StagingTelemetry: def __post_init__(self) -> None: self.run_dir = Path(self.run_dir) self.run_dir.mkdir(parents=True, exist_ok=True) + # Normalize here rather than at the caller: a blank or slash-only + # prefix would otherwise put run files at the repo root, where the + # dashboard's runs/ paths cannot find them. Covers the CLI + # flag, the environment, and programmatic callers in one place. + self.path_prefix = self.path_prefix.strip().strip("/").strip() + if not self.path_prefix: + self.path_prefix = DEFAULT_STAGING_PREFIX self._last_upload_at = 0.0 self._upload_failures = 0 + self._upload_successes = 0 self._calibration_events: list[dict[str, Any]] = [] self._artifacts: dict[str, dict[str, Any]] = {} self._progress: dict[str, Any] = { @@ -97,8 +106,22 @@ def __post_init__(self) -> None: @property def repo_run_prefix(self) -> str: - prefix = self.path_prefix.strip("/") - return f"{prefix}/{self.run_id}" if prefix else self.run_id + # path_prefix is normalized non-empty at construction, so there is no + # root-level fallback here: writing runs to the repo root is the + # failure this class now refuses, not an alternative layout. + return f"{self.path_prefix}/{self.run_id}" + + @property + def uploads_succeeded(self) -> int: + """How many files actually reached the staging repo. + + Zero on a run that was configured to upload but never managed to -- + no write token, revoked access, a Hub outage. Uploads are best-effort + and never fail the build, so this is the only signal separating a run + that staged from one that merely intended to. + """ + + return self._upload_successes def _api(self): if self.api is not None: @@ -125,6 +148,7 @@ def _upload_file(self, local: Path, path_in_repo: str) -> None: repo_type="dataset", ) self._upload_failures = 0 + self._upload_successes += 1 except Exception as exc: self._upload_failures += 1 print( diff --git a/packages/microcosm-build/tests/test_staging.py b/packages/microcosm-build/tests/test_staging.py index 248a18ae..5f9f6f46 100644 --- a/packages/microcosm-build/tests/test_staging.py +++ b/packages/microcosm-build/tests/test_staging.py @@ -128,3 +128,61 @@ def hf_hub_download(self, **kwargs): err = capsys.readouterr().err assert "staging upload" in err assert "disabling staging uploads" in err + # Nothing reached the repo, and the run says so. A configured destination + # is not evidence of delivery. + assert telemetry.uploads_succeeded == 0 + + +def test_uploads_succeeded_counts_files_that_reached_the_repo(tmp_path): + api = FakeApi() + telemetry = StagingTelemetry( + run_id="run-c", + candidate_release_id="run-c", + run_dir=tmp_path / "run-c", + repo_id="org/staging", + api=api, + upload_interval_seconds=0.0, + ) + + telemetry.stage("target_compilation", force_upload=True) + + assert telemetry.uploads_succeeded == len(api.uploads) + assert telemetry.uploads_succeeded > 0 + + +def test_blank_path_prefix_falls_back_to_the_default(tmp_path): + # A blank or slash-only prefix would put run files at the repo root, where + # the dashboard's runs/ paths cannot find them. + for blank in ("", " ", "/", " / "): + telemetry = StagingTelemetry( + run_id="run-e", + candidate_release_id="run-e", + run_dir=tmp_path / "run-e", + path_prefix=blank, + ) + assert telemetry.path_prefix == "runs" + assert telemetry.repo_run_prefix == "runs/run-e" + + +def test_path_prefix_is_trimmed_but_otherwise_respected(tmp_path): + telemetry = StagingTelemetry( + run_id="run-f", + candidate_release_id="run-f", + run_dir=tmp_path / "run-f", + path_prefix=" /candidate-runs/ ", + ) + assert telemetry.path_prefix == "candidate-runs" + assert telemetry.repo_run_prefix == "candidate-runs/run-f" + + +def test_uploads_succeeded_is_zero_for_a_local_only_run(tmp_path): + telemetry = StagingTelemetry( + run_id="run-d", + candidate_release_id="run-d", + run_dir=tmp_path / "run-d", + ) + + telemetry.stage("target_compilation", force_upload=True) + + assert telemetry.uploads_succeeded == 0 + assert (tmp_path / "run-d" / "progress.json").is_file() diff --git a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py index 1c4c4a62..59fb2084 100644 --- a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py @@ -146,7 +146,9 @@ def test_certified_release_dir_refusal_precedes_all_side_effects() -> None: builder = _load_builder_module() tree = ast.parse(Path(builder.__file__).read_text()) main_fn = next( - n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "main" + n + for n in ast.walk(tree) + if isinstance(n, ast.FunctionDef) and n.name == "_main" ) refusals = [ n @@ -226,7 +228,9 @@ def test_ssi_delivery_fences_are_passed_on_the_dense_arm_only() -> None: builder = _load_builder_module() tree = ast.parse(Path(builder.__file__).read_text()) main_fn = next( - n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "main" + n + for n in ast.walk(tree) + if isinstance(n, ast.FunctionDef) and n.name == "_main" ) calls = [ n @@ -256,7 +260,9 @@ def test_delivery_gate_result_reaches_the_manifest_gates_block() -> None: builder = _load_builder_module() tree = ast.parse(Path(builder.__file__).read_text()) main_fn = next( - n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "main" + n + for n in ast.walk(tree) + if isinstance(n, ast.FunctionDef) and n.name == "_main" ) manifest_calls = [ n @@ -323,7 +329,9 @@ def visit_Call(self, node): main_calls = [ (node, stack) for node, stack in calls - if any(isinstance(anc, ast.FunctionDef) and anc.name == "main" for anc in stack) + if any( + isinstance(anc, ast.FunctionDef) and anc.name == "_main" for anc in stack + ) ] assert len(main_calls) == 1 call_node, stack = main_calls[0] @@ -354,7 +362,7 @@ def visit_Call(self, node): # evidence (release-dir reuse, microcosm#568 round 2) before the # certified dataset write. main_fn = next( - anc for anc in stack if isinstance(anc, ast.FunctionDef) and anc.name == "main" + anc for anc in stack if isinstance(anc, ast.FunctionDef) and anc.name == "_main" ) def _is_bound_cleanup_for(node): @@ -1494,7 +1502,7 @@ def test_builder_pool_release_identity_is_manifest_authenticated() -> None: "invented-release", {"publication_run_id": "fixture-publication"}, ) - assert "_assert_pool_release_id_value" in builder.main.__code__.co_names + assert "_assert_pool_release_id_value" in builder._main.__code__.co_names def test_builder_rejects_replaced_authenticated_pool_h5_at_first_consumer( @@ -1582,7 +1590,7 @@ def test_authenticated_pool_h5_consumers_use_one_returned_identity() -> None: assert "shutil.copy2(base_h5" not in source assert 'pool_h5.get("sha256")' not in source - main_source = inspect.getsource(builder.main) + main_source = inspect.getsource(builder._main) diagnostics_source = inspect.getsource( builder._write_release_calibration_diagnostics ) @@ -1602,7 +1610,7 @@ def test_builder_reconciles_exact_k_count_before_any_release_write() -> None: import inspect builder = _load_builder_module() - source = inspect.getsource(builder.main) + source = inspect.getsource(builder._main) count_gate = source.index("assert_exact_k_realized_count(ladder_outcome") for later_write in ( @@ -1809,7 +1817,7 @@ def test_maximum_microsim_batch_size_defaults_and_overrides(monkeypatch) -> None def test_staging_repo_can_default_from_environment(monkeypatch) -> None: builder = _load_builder_module() - monkeypatch.setenv("POPULACE_STAGING_REPO_ID", "policyengine/populace-us-staging") + monkeypatch.setenv("POPULACE_STAGING_REPO_ID", "policyengine/populace-us-canary") monkeypatch.setenv("POPULACE_STAGING_PREFIX", "candidate-runs") monkeypatch.setattr( sys, @@ -1825,10 +1833,48 @@ def test_staging_repo_can_default_from_environment(monkeypatch) -> None: args = builder._parse_args() - assert args.staging_repo_id == "policyengine/populace-us-staging" + assert args.staging_repo_id == "policyengine/populace-us-canary" assert args.staging_prefix == "candidate-runs" +def test_empty_staging_environment_does_not_disable_staging(monkeypatch) -> None: + builder = _load_builder_module() + monkeypatch.setenv("POPULACE_STAGING_REPO_ID", "") + monkeypatch.setenv("POPULACE_STAGING_PREFIX", " ") + monkeypatch.setattr( + sys, + "argv", + [ + "build_us_fiscal_refresh_release.py", + "--ledger-facts", + "facts.jsonl", + "--out", + "release", + ], + ) + + args = builder._parse_args() + + assert args.staging_repo_id == builder.STAGING_REPO_ID + assert args.staging_prefix == builder.DEFAULT_STAGING_PREFIX + + +def test_env_default_treats_blank_as_unset_and_trims(monkeypatch) -> None: + builder = _load_builder_module() + + monkeypatch.delenv("POPULACE_TEST_ENV_DEFAULT", raising=False) + assert builder._env_default("POPULACE_TEST_ENV_DEFAULT", "fallback") == "fallback" + + monkeypatch.setenv("POPULACE_TEST_ENV_DEFAULT", "") + assert builder._env_default("POPULACE_TEST_ENV_DEFAULT", "fallback") == "fallback" + + monkeypatch.setenv("POPULACE_TEST_ENV_DEFAULT", " ") + assert builder._env_default("POPULACE_TEST_ENV_DEFAULT", "fallback") == "fallback" + + monkeypatch.setenv("POPULACE_TEST_ENV_DEFAULT", " org/repo ") + assert builder._env_default("POPULACE_TEST_ENV_DEFAULT", "fallback") == "org/repo" + + def test_soi_indicator_rows_flag_positive_component_items() -> None: builder = _load_builder_module() @@ -3900,6 +3946,8 @@ def fake_sha256(path): class LiveTelemetry: run_id = "live-telemetry-test" + repo_id = "policyengine/populace-us-staging" + uploads_succeeded = 3 def stage(self, stage, **details): captured.setdefault("telemetry_events", []).append(("stage", stage)) @@ -8656,7 +8704,7 @@ def test_pool_owned_fiscal_transforms_are_guarded_for_prepared_pool_input() -> N main_fn = next( node for node in ast.walk(tree) - if isinstance(node, ast.FunctionDef) and node.name == "main" + if isinstance(node, ast.FunctionDef) and node.name == "_main" ) def call_name(call: ast.Call) -> str | None: @@ -9315,7 +9363,8 @@ def test_staging_telemetry_defaults_on_and_no_staging_disables(tmp_path, monkeyp ], ) args = module._parse_args() - assert args.staging_repo_id == "policyengine/populace-us-staging" + assert module.STAGING_REPO_ID == "policyengine/populace-us-staging" + assert args.staging_repo_id == module.STAGING_REPO_ID assert not args.no_staging def namespace(no_staging: bool) -> SimpleNamespace: @@ -9334,6 +9383,7 @@ def namespace(no_staging: bool) -> SimpleNamespace: ) assert telemetry is not None assert telemetry.run_id == "rel-1" + assert telemetry.repo_id is None # --no-staging wins even when a staging destination is configured. assert ( @@ -9344,6 +9394,153 @@ def namespace(no_staging: bool) -> SimpleNamespace: ) +def test_blank_staging_repo_id_is_refused_at_parse_time(monkeypatch, capsys) -> None: + module = _load_builder_module() + argv = [ + "--ledger-facts", + "facts.jsonl", + "--out", + "release", + "--staging-repo-id", + "", + ] + + with pytest.raises(SystemExit) as excinfo: + module._parse_args(argv) + + assert excinfo.value.code == 2 + assert "--no-staging" in capsys.readouterr().err + + +def test_blank_staging_repo_id_is_accepted_with_a_local_staging_dir( + tmp_path, +) -> None: + module = _load_builder_module() + argv = [ + "--ledger-facts", + "facts.jsonl", + "--out", + "release", + "--staging-repo-id", + "", + "--staging-dir", + str(tmp_path / "stage"), + ] + + args = module._parse_args(argv) + + assert args.staging_repo_id == "" + assert args.staging_dir == tmp_path / "stage" + + +def test_a_crashed_build_marks_its_staging_run_failed(monkeypatch) -> None: + module = _load_builder_module() + recorded: list[BaseException] = [] + + class Telemetry: + def fail(self, error): + recorded.append(error) + + monkeypatch.setattr(module, "_ACTIVE_TELEMETRY", Telemetry()) + monkeypatch.setattr( + module, + "_main", + lambda argv=None: (_ for _ in ()).throw(RuntimeError("build exploded")), + ) + + with pytest.raises(RuntimeError, match="build exploded"): + module.main() + + assert [str(error) for error in recorded] == ["build exploded"] + + +def test_crash_reporting_never_replaces_the_real_traceback(monkeypatch, capsys) -> None: + module = _load_builder_module() + + class ExplodingTelemetry: + def fail(self, error): + raise RuntimeError("telemetry itself is broken") + + monkeypatch.setattr(module, "_ACTIVE_TELEMETRY", ExplodingTelemetry()) + monkeypatch.setattr( + module, + "_main", + lambda argv=None: (_ for _ in ()).throw(ValueError("the real failure")), + ) + + with pytest.raises(ValueError, match="the real failure"): + module.main() + + assert "could not record the staging run as failed" in capsys.readouterr().err + + +def test_staging_telemetry_clears_any_previous_active_run(tmp_path) -> None: + module = _load_builder_module() + args = SimpleNamespace( + no_staging=False, + staging_dir=tmp_path / "stage", + staging_repo_id=None, + staging_run_id=None, + staging_prefix=module.DEFAULT_STAGING_PREFIX, + staging_upload_interval_seconds=60.0, + ) + module._staging_telemetry(args, release_root=tmp_path, release_id="rel-1") + assert module._ACTIVE_TELEMETRY is not None + + args.no_staging = True + assert ( + module._staging_telemetry(args, release_root=tmp_path, release_id="rel-2") + is None + ) + assert module._ACTIVE_TELEMETRY is None + + +def test_staging_manifest_block_distinguishes_opt_out_from_delivery() -> None: + module = _load_builder_module() + + assert module._staging_manifest_block(None) == { + "enabled": False, + "reason": "--no-staging", + } + + class Delivered: + run_id = "rel-1" + repo_id = "policyengine/populace-us-staging" + uploads_succeeded = 7 + + assert module._staging_manifest_block(Delivered()) == { + "enabled": True, + "run_id": "rel-1", + "repo_id": "policyengine/populace-us-staging", + "uploads_succeeded": 7, + } + + class Undelivered: + run_id = "rel-2" + repo_id = None + uploads_succeeded = 0 + + block = module._staging_manifest_block(Undelivered()) + assert block["enabled"] is True + assert block["uploads_succeeded"] == 0 + assert block["repo_id"] is None + + +def test_staging_telemetry_refuses_a_destinationless_namespace(tmp_path) -> None: + module = _load_builder_module() + args = SimpleNamespace( + no_staging=False, + staging_dir=None, + staging_repo_id="", + staging_run_id=None, + staging_prefix=module.DEFAULT_STAGING_PREFIX, + staging_upload_interval_seconds=60.0, + ) + + with pytest.raises(ValueError, match="no destination"): + module._staging_telemetry(args, release_root=tmp_path, release_id="rel-1") + + # --------------------------------------------------------------------------- # #299 / #217: per-reform materialization checkpoint resume + cache-key safety. # @@ -9952,7 +10149,7 @@ def test_main_runs_cross_register_and_take_up_contract_preflights() -> None: are looked up by name inside ``main``). """ builder = _load_builder_module() - called = set(builder.main.__code__.co_names) + called = set(builder._main.__code__.co_names) for preflight in ( "assert_release_input_coverage_manifest_current", "us_register_consistency_gate", @@ -10044,7 +10241,7 @@ def test_release_h5_write_sits_between_batched_raise_and_smoke() -> None: import inspect builder = _load_builder_module() - source = inspect.getsource(builder.main) + source = inspect.getsource(builder._main) tree = ast.parse(source) batched_raises: list[int] = [] diff --git a/packages/microcosm-data/src/microcosm/data/publish_cli.py b/packages/microcosm-data/src/microcosm/data/publish_cli.py index 16b3f5e5..d24d0979 100644 --- a/packages/microcosm-data/src/microcosm/data/publish_cli.py +++ b/packages/microcosm-data/src/microcosm/data/publish_cli.py @@ -10,12 +10,13 @@ from microcosm.data.release import publish_release -def _staging_missing(release_dir: Path) -> bool: - """True if the release's build manifest records no staging telemetry. +def _staging_undelivered(release_dir: Path) -> bool: + """True if a build that should have staged has nothing to show for it. - Releases are expected to publish staging runs while building (the builder - now stages by default); a missing block means the build predates that or - was run with --no-staging. Non-fatal — surfaced as a warning at publish. + The gate is scoped by the presence of the ``staging`` key. Builders with + no staging path are untouched, while an explicit ``enabled: false`` is a + legitimate opt-out. A present but empty block, or an enabled run with no + successful upload, records intended staging that never arrived. """ path = release_dir / "build_manifest.json" if not path.exists(): @@ -24,7 +25,14 @@ def _staging_missing(release_dir: Path) -> bool: manifest = json.loads(path.read_text()) except (OSError, ValueError): return False - return not manifest.get("staging") + if not isinstance(manifest, dict) or "staging" not in manifest: + return False + staging = manifest["staging"] + if not isinstance(staging, dict) or not staging: + return True + if staging.get("enabled") is False: + return False + return not staging.get("uploads_succeeded") def _reform_validation_skipped(release_dir: Path) -> bool: @@ -118,6 +126,17 @@ def main(argv: list[str] | None = None) -> int: "release never silently ships blank OBBBA validation." ), ) + parser.add_argument( + "--allow-missing-staging", + action="store_true", + help=( + "Publish even if the build recorded no staging telemetry, or " + "recorded staging that never uploaded anything. Off by default so " + "a release that never appeared on the staging dashboard is not " + "shipped unnoticed. A declared --no-staging build publishes " + "without this flag." + ), + ) args = parser.parse_args(argv) if args.tag_only and not args.no_latest: @@ -138,14 +157,20 @@ def main(argv: list[str] | None = None) -> int: ) return 1 - if _staging_missing(Path(args.release_dir)): + if not args.allow_missing_staging and _staging_undelivered(Path(args.release_dir)): print( - "warning: this release's build_manifest records no staging " - "telemetry — the build ran with --no-staging or predates " - "staging-by-default, so it will not appear on the staging " - "dashboard.", + "refusing to publish: this release's build_manifest records no " + "delivered staging telemetry, so the build never appeared on the " + "staging dashboard and there is no pre-publication review of it. " + "Either the staging destination was lost mid-build (uploads " + "self-disable after repeated failures — check the build machine's " + "Hugging Face write token), or the manifest predates staging " + "provenance. Rebuild with staging reaching its repo, or pass " + "--allow-missing-staging to publish anyway. A build that declared " + "--no-staging publishes without the flag.", file=sys.stderr, ) + return 1 pointer = publish_release( Path(args.release_dir), diff --git a/packages/microcosm-data/tests/test_publish_guard.py b/packages/microcosm-data/tests/test_publish_guard.py index 76106650..2832fd99 100644 --- a/packages/microcosm-data/tests/test_publish_guard.py +++ b/packages/microcosm-data/tests/test_publish_guard.py @@ -3,7 +3,11 @@ import pytest -from microcosm.data.publish_cli import _reform_validation_skipped, main +from microcosm.data.publish_cli import ( + _reform_validation_skipped, + _staging_undelivered, + main, +) def _write_rv(release_dir: Path, *, out_of_sample_simulated: bool | None) -> None: @@ -45,28 +49,143 @@ def _stub_publish(monkeypatch): return cli -def test_publish_warns_when_build_manifest_has_no_staging( - tmp_path, capsys, monkeypatch -): - (tmp_path / "build_manifest.json").write_text( - json.dumps({"build_id": "x", "staging": None}) +def test_allow_incomplete_reform_validation_publishes(tmp_path, capsys, monkeypatch): + _write_rv(tmp_path, out_of_sample_simulated=False) + cli = _stub_publish(monkeypatch) + monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) + rc = cli.main([str(tmp_path), "--allow-incomplete-reform-validation"]) + assert rc == 0 + assert "refusing to publish" not in capsys.readouterr().err + + +def _write_bm(release_dir: Path, manifest: dict) -> None: + (release_dir / "build_manifest.json").write_text(json.dumps(manifest)) + + +def test_staging_undelivered_reads_the_manifest_not_the_country(tmp_path): + assert _staging_undelivered(tmp_path) is False + + # No staging key: a builder with no staging path, e.g. the ACS local-area + # product. Key presence scopes the gate without an exception list. + _write_bm(tmp_path, {"build_id": "x", "dataset": {"kind": "acs_local_area"}}) + assert _staging_undelivered(tmp_path) is False + + # Present and empty: the pre-provenance shape, or a lost destination. + _write_bm(tmp_path, {"build_id": "x", "staging": None}) + assert _staging_undelivered(tmp_path) is True + _write_bm(tmp_path, {"build_id": "x", "staging": {}}) + assert _staging_undelivered(tmp_path) is True + + # A declared opt-out is a statement, not a gap. + _write_bm( + tmp_path, + {"build_id": "x", "staging": {"enabled": False, "reason": "--no-staging"}}, + ) + assert _staging_undelivered(tmp_path) is False + + # Enabled but nothing landed: what a run without a write token looks like. + _write_bm( + tmp_path, + { + "build_id": "x", + "staging": { + "enabled": True, + "run_id": "r", + "uploads_succeeded": 0, + }, + }, + ) + assert _staging_undelivered(tmp_path) is True + + _write_bm( + tmp_path, + { + "build_id": "x", + "staging": { + "enabled": True, + "run_id": "r", + "uploads_succeeded": 9, + }, + }, + ) + assert _staging_undelivered(tmp_path) is False + + +def test_publish_refused_when_staging_never_delivered(tmp_path, capsys, monkeypatch): + _write_bm( + tmp_path, + { + "build_id": "x", + "staging": { + "enabled": True, + "run_id": "r", + "uploads_succeeded": 0, + }, + }, + ) + monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) + rc = main([str(tmp_path)]) + assert rc == 1 + assert "refusing to publish" in capsys.readouterr().err + + +def test_publish_refused_when_staging_block_is_null(tmp_path, capsys, monkeypatch): + _write_bm(tmp_path, {"build_id": "x", "staging": None}) + monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) + rc = main([str(tmp_path)]) + assert rc == 1 + assert "refusing to publish" in capsys.readouterr().err + + +def test_publish_allowed_for_a_declared_no_staging_build(tmp_path, capsys, monkeypatch): + _write_bm( + tmp_path, + {"build_id": "x", "staging": {"enabled": False, "reason": "--no-staging"}}, ) cli = _stub_publish(monkeypatch) monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) rc = cli.main([str(tmp_path)]) - assert rc == 0 # warning, not a refusal - assert "no staging telemetry" in capsys.readouterr().err + assert rc == 0 + assert "refusing to publish" not in capsys.readouterr().err -def test_publish_silent_when_staging_recorded(tmp_path, capsys, monkeypatch): - (tmp_path / "build_manifest.json").write_text( - json.dumps({"build_id": "x", "staging": {"run_id": "rel-1", "repo_id": "r/s"}}) +def test_publish_allowed_when_the_builder_has_no_staging_path( + tmp_path, capsys, monkeypatch +): + _write_bm(tmp_path, {"build_id": "microcosm-us-local-x", "dataset": {}}) + cli = _stub_publish(monkeypatch) + monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) + rc = cli.main([str(tmp_path), "--no-latest"]) + assert rc == 0 + assert "refusing to publish" not in capsys.readouterr().err + + +def test_allow_missing_staging_escape_hatch_publishes(tmp_path, capsys, monkeypatch): + _write_bm(tmp_path, {"build_id": "x", "staging": None}) + cli = _stub_publish(monkeypatch) + monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) + rc = cli.main([str(tmp_path), "--allow-missing-staging"]) + assert rc == 0 + assert "refusing to publish" not in capsys.readouterr().err + + +def test_publish_proceeds_when_staging_delivered(tmp_path, capsys, monkeypatch): + _write_bm( + tmp_path, + { + "build_id": "x", + "staging": { + "enabled": True, + "run_id": "r", + "uploads_succeeded": 9, + }, + }, ) cli = _stub_publish(monkeypatch) monkeypatch.delenv("SLACK_WEBHOOK_POPULACE_US", raising=False) rc = cli.main([str(tmp_path)]) assert rc == 0 - assert "no staging telemetry" not in capsys.readouterr().err + assert "refusing to publish" not in capsys.readouterr().err def test_tag_only_cli_forwards_no_main_publication_mode(tmp_path, monkeypatch): diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 1a1f73cb..b6fea134 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -63,7 +63,7 @@ ) from microcosm.build.ledger_artifact import load_ledger_consumer_artifact from microcosm.build.source_runtime import SourceRuntimeConfig, run_source_stage -from microcosm.build.staging import StagingTelemetry +from microcosm.build.staging import DEFAULT_STAGING_PREFIX, StagingTelemetry from microcosm.build.us_runtime import ( ASEC_2023_WEEKS_UNEMPLOYED_SOURCE_SHA256, CONGRESSIONAL_DISTRICT_VINTAGE_CROSSWALK_SHA256_ATTR, @@ -291,6 +291,7 @@ PERIOD = 2024 REPO_ID = "policyengine/populace-us" +STAGING_REPO_ID = "policyengine/populace-us-staging" DATASET_FILENAME = "populace_us_2024.h5" CALIBRATION_FILENAME = "populace_us_2024_calibration.npz" FINAL_HOUSEHOLD_WEIGHTS_FILENAME = "final_household_weights.npy" @@ -730,6 +731,19 @@ def _automatic_gc_suspended(): "hours; defaults False (PolicyEngine/microcosm#249)." ), } + + +def _env_default(name: str, default: str) -> str: + """Return a trimmed environment override, treating blank as unset. + + ``os.environ.get(name, default)`` falls back only when the variable is + absent. An exported empty string otherwise silently defeats the staging + default. For staging, only ``--no-staging`` should turn telemetry off. + """ + + return os.environ.get(name, "").strip() or default + + US_ACA_MARKETPLACE_STAGE = "aca_marketplace_inputs" US_ACA_SOURCE_OUTPUT_COLUMNS = US_HEALTH_INPUT_NONCONSTANT_COLUMNS US_ACA_REPORTED_SUBSIDIZED_ANCHOR = ( @@ -1435,14 +1449,13 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: ) parser.add_argument( "--staging-repo-id", - default=os.environ.get( - "POPULACE_STAGING_REPO_ID", "policyengine/populace-us-staging" - ), + default=_env_default("POPULACE_STAGING_REPO_ID", STAGING_REPO_ID), help=( "Hugging Face dataset repo to upload staging telemetry to while " "the build runs. On by default (uploads are best-effort and never " "fail the build); override with POPULACE_STAGING_REPO_ID or " - "disable with --no-staging." + "disable with --no-staging. An empty POPULACE_STAGING_REPO_ID is " + "ignored rather than read as off." ), ) parser.add_argument( @@ -1471,7 +1484,7 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: ) parser.add_argument( "--staging-prefix", - default=os.environ.get("POPULACE_STAGING_PREFIX", "runs"), + default=_env_default("POPULACE_STAGING_PREFIX", DEFAULT_STAGING_PREFIX), help=( "Repo prefix for staging run artifacts. Defaults to " "POPULACE_STAGING_PREFIX or runs." @@ -1509,6 +1522,15 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: "--refit-l2-lambda requires the sparse L0+refit default dataset; " "--dense-default-dataset has no refit stage (use --l2-lambda)." ) + if not args.no_staging and not args.staging_dir and not args.staging_repo_id: + # Staging with nowhere to write is a configuration error, not a quiet + # skip. Turning staging off is an explicit decision, never a side + # effect of a blank repo id. + parser.error( + "--staging-repo-id is empty and no --staging-dir is set, so staging " + "telemetry would silently do nothing. Pass --no-staging to skip " + "staging deliberately, or --staging-dir for a local-only run." + ) ladder_values = ( args.exact_k, args.exact_k_pi_hi, @@ -8083,19 +8105,54 @@ def _assert_exact_k_original_pool_alignment( ) +#: The staging run for the build in flight, so the entry point can mark it +#: failed on the way out. The build body hands its telemetry object down a +#: large call stack; a module-level handle avoids threading a second copy back +#: up purely for the failure path. +_ACTIVE_TELEMETRY: StagingTelemetry | None = None + + +def _staging_manifest_block(telemetry: StagingTelemetry | None) -> dict[str, object]: + """Record what staging did and distinguish an opt-out from non-delivery. + + Uploads are best-effort and self-disable after repeated failures, so a + configured destination is not evidence that anything reached it. + """ + + if telemetry is None: + return {"enabled": False, "reason": "--no-staging"} + return { + "enabled": True, + "run_id": telemetry.run_id, + "repo_id": telemetry.repo_id, + "uploads_succeeded": telemetry.uploads_succeeded, + } + + def _staging_telemetry( args: argparse.Namespace, *, release_root: Path, release_id: str, ) -> StagingTelemetry | None: + global _ACTIVE_TELEMETRY + # Each call establishes the current run, so a handle from a previous one + # can never be marked failed in place of this build's. + _ACTIVE_TELEMETRY = None if args.no_staging: return None if not args.staging_dir and not args.staging_repo_id: - return None + # The parser rejects this combination, so reaching it means a caller + # built the namespace directly. Returning None here would reinstate + # the silent skip the parser guard exists to prevent. + raise ValueError( + "staging is enabled but has no destination: staging_repo_id is " + "empty and staging_dir is unset. Set no_staging to skip staging, " + "or give a staging_dir for a local-only run." + ) run_id = args.staging_run_id or release_id run_dir = args.staging_dir or release_root / "staging" / "runs" / run_id - return StagingTelemetry( + _ACTIVE_TELEMETRY = StagingTelemetry( run_id=run_id, candidate_release_id=release_id, run_dir=run_dir, @@ -8103,6 +8160,7 @@ def _staging_telemetry( path_prefix=args.staging_prefix, upload_interval_seconds=args.staging_upload_interval_seconds, ) + return _ACTIVE_TELEMETRY class _TerminalBatchTelemetry: @@ -8175,6 +8233,30 @@ def _print_build_result( def main(argv: Sequence[str] | None = None) -> None: + """Run the build and mark its staging run failed if it does not finish. + + An uncaught build error previously left the dashboard status at ``running`` + forever. SIGKILL remains outside the reach of an in-process handler; this + closes the ordinary exception and termination half of the gap. + """ + + try: + _main(argv) + except BaseException as error: + if _ACTIVE_TELEMETRY is not None: + try: + _ACTIVE_TELEMETRY.fail(error) + except Exception as telemetry_error: # pragma: no cover - defensive + # A failing failure-report must not replace the real traceback. + print( + "warning: could not record the staging run as failed: " + f"{type(telemetry_error).__name__}: {telemetry_error}", + file=sys.stderr, + ) + raise + + +def _main(argv: Sequence[str] | None = None) -> None: args = _parse_args(argv) if _git_dirty(): raise SystemExit("Refusing to build a release from a dirty git worktree.") @@ -11308,11 +11390,7 @@ def main(argv: Sequence[str] | None = None) -> None: ledger_artifact=ledger_artifact.provenance(), default_dataset=default_dataset, medicaid_enrollment_substitutions=medicaid_enrollment_substitutions, - staging=( - {"run_id": telemetry.run_id, "repo_id": args.staging_repo_id} - if telemetry is not None - else None - ), + staging=_staging_manifest_block(telemetry), dataset_key=dataset_key, dataset_filename=dataset_filename, calibration_key=calibration_key,