Add always-on build telemetry delivery

README.md

@@ -44,16 +44,23 @@ until you restart the kernel, load it by path with
 `load_country_spec(Path(...))` (always re-read), or call
 `microcosm.build.country_spec._load_packaged_country_spec.cache_clear()`.
 
-## Staging build telemetry
-
-US fiscal refresh builds emit pre-release staging telemetry **by default**:
-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`. 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.
+## Build progress and staging run files
+
+Supported US and UK build commands always start the local telemetry emitter
+service. It reports live progress to the hosted collector when the operator's
+existing Hugging Face login is accepted; otherwise it retains the events
+locally and the build continues. This live event path is independent of the
+staging run files described below.
+
+US fiscal refresh builds also write pre-release staging run files **by
+default**. 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 these files with `--no-staging`, or point them
+elsewhere with `--staging-repo-id` / `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` does not disable
+the local telemetry emitter service.
 
 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
@@ -75,8 +82,8 @@ The UK commands (`tools/build_uk_frs_spine.py`, a shim over the package's
 `uk_runtime.spine_build`, and `microcosm-build-uk` / `tools/build_uk_full.py`,
 whose `--release-role` builds either the national or the dense line;
 `tools/build_uk_rowwise_candidate.py` is a stub over the same driver) stage
-version 2 telemetry to `policyengine/populace-uk-staging` under the same
-switch. The build command also **stages the finished dataset bundle** it built,
+version 2 staging run files to `policyengine/populace-uk-staging` under the
+same switch. The build command also **stages the finished dataset bundle** it built,
 national, dense or exact-count, under `staged/<run_id>/` in the
 private `policyengine/populace-uk-private` repository so the team can inspect
 it without publishing it: `releases/` and `latest.json` are untouched, the
@@ -203,7 +210,8 @@ weights, calling the release tool's own gate functions (none is
 re-implemented), and exits `1` on any certain failure, `2` on AT-RISK only,
 and `0` when clean. An argparse error also exits `2` but writes no report, so
 the wrapper returns `64` whenever no report was written. It writes only its
-report: nothing under `--out`, no staging telemetry, no receipts. A base or
+report: nothing under `--out`, no staging run files, no receipts. The local
+telemetry emitter service still reports dry-run progress. A base or
 donor that the config does not name locally is still downloaded, into the same
 caches the release uses. A refusal before the stop point becomes the report's
 certain failure. A crash while grading is reported as the dry run's own error,

changelog.d/1099-telemetry-failure-reporting.fixed.md

@@ -0,0 +1,3 @@
+Report failed UK builds and active stages as failed, isolate telemetry in automated tests, and keep local telemetry setup errors from aborting builds. Rename the initial telemetry migration module so wheel validation accepts it without changing its revision identifier.
+
+Report nonzero US dry-run results as failures, delay telemetry delivery retries during shutdown, and retain reaped workers' CPU usage in the Linux fallback sampler.

changelog.d/always-on-telemetry-emitter.added.md

@@ -0,0 +1 @@
+Microcosm builds now start a local telemetry emitter service that reports authenticated live progress, process-tree CPU and memory, heartbeats, US engine-batch progress, and UK graph-node progress without making collector network requests in the build process. PolicyEngine members use their existing Hugging Face login automatically, while builds without an accepted organization credential remain local-only. The staging JSON writers are now named as run-bundle writers and operate independently from the hosted event emitter. SQLAlchemy ORM sessions now manage the local event spool, and Alembic applies and records its schema migrations.

conftest.py

@@ -10,6 +10,9 @@
 import importlib.util
 import sys
 from pathlib import Path
+from urllib.parse import urlsplit
+
+import pytest
 
 _ROOT = str(Path(__file__).resolve().parent)
 if _ROOT not in sys.path:
@@ -65,3 +68,82 @@ def pytest_collection_modifyitems(items) -> None:
             item.add_marker("integration")
         if group.engine_module is not None:
             item.add_marker(f"requires_{group.country}")
+
+
+@pytest.fixture(autouse=True)
+def fake_telemetry_emitters(monkeypatch, tmp_path):
+    """Default every workspace test to in-memory telemetry, never production."""
+    from huggingface_hub import constants as hf_constants
+
+    from microcosm.build import telemetry_emitter
+    from microcosm.build.telemetry_emitter_service import collector
+    from test_support.microcosm_build.telemetry import FakeTelemetryEmitter
+
+    for name in ("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN"):
+        monkeypatch.delenv(name, raising=False)
+    isolated_hf = tmp_path / "telemetry-hf"
+    token_path = isolated_hf / "token"
+    monkeypatch.setenv("HF_HOME", str(isolated_hf))
+    monkeypatch.setenv("HF_TOKEN_PATH", str(token_path))
+    monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "telemetry-cache"))
+    # HF constants may have been imported before this fixture ran.
+    monkeypatch.setattr(hf_constants, "HF_HOME", str(isolated_hf))
+    monkeypatch.setattr(hf_constants, "HF_TOKEN_PATH", str(token_path))
+
+    emitters = []
+
+    def start(_cls, **kwargs):
+        fields = telemetry_emitter.TelemetryRun.__dataclass_fields__
+        run = telemetry_emitter.TelemetryRun(
+            **{name: value for name, value in kwargs.items() if name in fields}
+        )
+        emitter = FakeTelemetryEmitter(run)
+        emitters.append(emitter)
+        return emitter
+
+    monkeypatch.setattr(
+        telemetry_emitter.LocalTelemetryEmitter, "start", classmethod(start)
+    )
+    # Reset any cached handle left by a previously loaded build entrypoint.
+    for module in tuple(sys.modules.values()):
+        if isinstance(
+            getattr(module, "_ACTIVE_EMITTER", None),
+            telemetry_emitter.LocalTelemetryEmitter,
+        ):
+            monkeypatch.setattr(module, "_ACTIVE_EMITTER", None)
+
+    real_post = collector._http_post
+
+    def local_post(url, *args, **kwargs):
+        parsed = urlsplit(url)
+        collector._development_collector_url(f"{parsed.scheme}://{parsed.netloc}")
+        return real_post(url, *args, **kwargs)
+
+    monkeypatch.setattr(collector, "_http_post", local_post)
+    yield emitters
+    for emitter in emitters:
+        emitter.close()
+
+
+@pytest.fixture
+def real_local_telemetry(monkeypatch, fake_telemetry_emitters):
+    """Permit actual startup only with an explicit loopback collector."""
+    from microcosm.build import telemetry_emitter
+    from microcosm.build.telemetry_emitter_service.collector import (
+        _development_collector_url,
+    )
+
+    def start(cls, **kwargs):
+        _development_collector_url(kwargs.get("development_collector_url") or "")
+        return _REAL_TELEMETRY_START(cls, **kwargs)
+
+    monkeypatch.setattr(
+        telemetry_emitter.LocalTelemetryEmitter, "start", classmethod(start)
+    )
+
+
+# Save the underlying method before per-test patches so imported aliases and
+# class references all share the same default fake and explicit local opt-in.
+from microcosm.build.telemetry_emitter import LocalTelemetryEmitter  # noqa: E402
+
+_REAL_TELEMETRY_START = LocalTelemetryEmitter.start.__func__

docs/agent-guide.md

@@ -78,6 +78,13 @@ dependency set under `tools/orrery-contract/`; the pytest test generates a
 document through Microcosm's public Python API and requires Orrery's public
 parser to accept it. It performs no browser rendering.
 
+Workspace tests use an in-memory telemetry emitter by default. The shared
+fixture isolates Hugging Face credentials and cache paths and blocks collector
+HTTP requests outside loopback. Tests of actual emitter startup must request
+`real_local_telemetry` and supply an explicit loopback development collector;
+that opt-in retains credential isolation. Other build tests can inspect
+`fake_telemetry_emitters` without creating sockets or service processes.
+
 New commits to a PR cancel older unfinished CI runs for that same PR.
 Each main-push run has a unique concurrency group, so all main-push runs
 remain independent and can finish validating their merged changes.

packages/microcosm-build/pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.13"
 license = { text = "MIT" }
 authors = [{ name = "PolicyEngine" }]
 dependencies = [
+    "alembic>=1.13.3,<2",
     "numpy>=1.26",
     "pandas>=2",
     "scipy>=1.13",
@@ -24,6 +25,7 @@ dependencies = [
     "pyyaml>=6",
     "jsonschema>=4.23,<5",
     "referencing>=0.35,<1",
+    "sqlalchemy>=2,<3",
 ]
 
 [project.optional-dependencies]

packages/microcosm-build/src/microcosm/build/__init__.py

@@ -190,7 +190,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None:
     LATEST_STAGING_POINTER,
     RUNS_INDEX,
     STAGING_SCHEMA_VERSION,
-    StagingTelemetry,
+    StagingRunBundleWriter,
 )
 
 __version__ = "0.1.0"
@@ -233,7 +233,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None:
     "LATEST_STAGING_POINTER",
     "RUNS_INDEX",
     "STAGING_SCHEMA_VERSION",
-    "StagingTelemetry",
+    "StagingRunBundleWriter",
     "TargetCoverageRequirement",
     "TargetFitRequirement",
     "ACCEPTED_CONSUMER_ARTIFACT_SCHEMA_VERSIONS",

packages/microcosm-build/src/microcosm/build/staging.py

@@ -58,8 +58,8 @@ def _write_json(path: Path, payload: dict[str, Any]) -> None:
 
 
 @dataclass
-class StagingTelemetry:
-    """Write and optionally upload build-run telemetry.
+class StagingRunBundleWriter:
+    """Write and optionally upload a build's staging run bundle.
 
     Args:
         run_id: Stable id for this build attempt.

packages/microcosm-build/src/microcosm/build/staging_cli.py

@@ -1,4 +1,4 @@
-"""Country-neutral command-line options for staging telemetry."""
+"""Country-neutral command-line options for staging run bundles."""
 
 from __future__ import annotations
 
@@ -27,9 +27,9 @@ def add_staging_arguments(
 
     ``default_upload_interval_seconds`` lets a long-running command choose a
     slower best-effort upload cadence: the Hub allows about 128 commits per
-    hour per repository and every telemetry cycle is up to eight single-file
-    commits, so a multi-hour solve at the 30-second default exhausts the
-    budget and loses uploads (the UK rowwise driver runs at 300).
+    hour per repository and each run-bundle upload cycle performs up to eight
+    single-file commits, so a multi-hour solve at the 30-second default
+    exhausts the budget and loses uploads (the UK rowwise driver runs at 300).
     """
 
     parser.add_argument(
@@ -42,7 +42,7 @@ def add_staging_arguments(
         default=repository.repo_id(os.environ),
         help=(
             "Access-controlled Hugging Face dataset repository for best-effort "
-            "telemetry delivery."
+            "staging run-bundle delivery."
         ),
     )
     parser.add_argument(
@@ -68,7 +68,10 @@ def add_staging_arguments(
     mode.add_argument(
         "--no-staging",
         action="store_true",
-        help="Deliberately disable staging telemetry for this build.",
+        help=(
+            "Disable the staging run bundle for this build; hosted telemetry "
+            "remains active."
+        ),
     )
     parser.add_argument(
         "--staging-read-back",
@@ -107,8 +110,8 @@ def add_staged_dataset_arguments(
     The finished bundle follows the staging mode switch: ``--no-staging``
     keeps nothing, ``--staging-local-only`` keeps the bundle and its sidecars
     on disk, and the default uploads it to ``repository`` under
-    ``staged/<run_id>/``. ``--no-staged-dataset`` runs telemetry alone: the
-    bundle is neither inventoried nor uploaded.
+    ``staged/<run_id>/``. ``--no-staged-dataset`` retains only the staging run
+    files: the dataset bundle is neither inventoried nor uploaded.
     """
 
     parser.add_argument(
@@ -125,7 +128,7 @@ def add_staged_dataset_arguments(
         "--no-staged-dataset",
         action="store_true",
         help=(
-            "Run staging telemetry alone: the finished bundle is neither "
+            "Keep only the staging run files: the finished bundle is neither "
             "inventoried nor uploaded (--no-staging already disables both)."
         ),
     )

packages/microcosm-build/src/microcosm/build/staging_v2.py

@@ -705,8 +705,8 @@ def _reject_record_collections(self, value: Any) -> None:
                 self._reject_record_collections(item)
 
 
-class StagingTelemetryV2:
-    """Record, validate, persist, and optionally upload telemetry version 2."""
+class StagingRunBundleWriterV2:
+    """Record, validate, persist, and optionally upload a version 2 run bundle."""
 
     def __init__(
         self,
@@ -743,7 +743,8 @@ def __init__(
             raise StagingContractError("pipeline_version must be non-empty.")
         if delivery_mode == "disabled":
             raise StagingContractError(
-                "Do not construct telemetry for disabled staging; record an opt-out."
+                "Do not construct a staging run bundle when staging is disabled; "
+                "record an opt-out."
             )
         if delivery_mode == "local_and_remote" and not (repo_id or "").strip():
             raise StagingContractError(