Publish UK Orrery graphs with the existing local build service

README.md

@@ -46,6 +46,11 @@ until you restart the kernel, load it by path with
 
 ## Build progress and staging run files
 
+UK build completion automatically exports Orrery execution graphs, attempts
+immutable publication before Hugging Face artifact staging, and preserves
+durable retry jobs and receipt snapshots. See
+[graph publication](docs/orrery-publication.md) for configuration and retry.
+
 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

changelog.d/1150-publication-failures.fixed.md

@@ -0,0 +1 @@
+Keep a failed UK build's graph evidence out of the output directory so an earlier successful bundle stays intact, and keep graph delivery running when a preserved directory is removed or the local queue reports an error.

changelog.d/1150-publication-types.changed.md

@@ -0,0 +1 @@
+Define explicit types for publication inventories, results, and staged files, and return a frozen record when reserving a graph upload.

changelog.d/1150.added.md

@@ -0,0 +1 @@
+Export UK build execution graphs automatically, queue immutable authenticated publication with durable retries, and include graph evidence and receipt snapshots in staged artifacts.

docs/agent-guide.md

@@ -14,6 +14,12 @@ the PEP 420 namespace `microcosm.<x>`: `frame`, `fit`, `calibrate`, `build`,
 
 ## Commands
 
+UK build completion exports and stages Orrery graph evidence. Publication uses
+an independent worker inside the existing local service and a durable queue;
+see [configuration and retry](orrery-publication.md). No new producer secret or
+environment variable is required. Keep graph export/publication tests independent
+of the hosted runs app and collector; use synthetic files and fake HTTP.
+
 ```bash
 uv sync --all-packages   # set up the whole workspace
 uv sync --all-packages --locked --extra us  # US engine environment

docs/orrery-publication.md

@@ -0,0 +1,70 @@
+# UK graph publication
+
+UK full, dense, size and national builds export `graph.orrery.json` from the
+saved graph schema and recorded execution evidence. They preserve the explicit
+public-safe file inventory in the attempt directory, attempt publication before
+Hugging Face artifact staging, and register graph/evidence files and an
+`orrery.publication.json` receipt snapshot in the artifact manifest.
+
+The existing local service has two independent workers: live telemetry and
+completed graph publication. Slow uploads cannot delay socket acknowledgement
+or telemetry delivery. This change does not introduce another generic runtime,
+rename the telemetry package, or add compatibility wrappers.
+
+Publication IDs do not depend on telemetry run IDs. Jobs use SQLAlchemy and
+Alembic in the existing local SQLite database. Pending jobs survive missing or
+rejected credentials, process restarts and telemetry event pruning. Retry
+leases prevent a stale worker from replacing another worker's result.
+The exact source bytes must remain available until delivery completes.
+
+The authenticated API creates upload permissions, receives exact files using
+scoped signed URLs, then finalizes an immutable publication. Signed uploads
+never receive the collector bearer credential. The API rejects different bytes
+under an existing publication ID. No population H5 or content-store arrays are
+included; only allowlisted graph declarations, execution receipts and reviewed
+aggregate summaries can be published publicly.
+
+## Configuration
+
+| Setting | Source and consumer | Provisioning |
+| --- | --- | --- |
+| `HF_TOKEN` / `HUGGINGFACE_TOKEN` | Existing operator credential, exchanged by the service for a short-lived collector session | Existing environment or Hugging Face login; never committed |
+| Collector origin | Tracked `PRODUCTION_COLLECTOR_URL` in the existing service | No producer variable or destination override |
+| Runs origin | Tracked `DEFAULT_GRAPH_PUBLICATION_ORIGIN` | No new producer variable |
+| `XDG_CACHE_HOME` | Optional existing cache-location setting | Defaults to the user's cache directory |
+
+The separately deployed runs application retains its Vercel Blob configuration.
+Authenticated production publication also requires
+PolicyEngine/calibration-diagnostics#208 deployed, and the runs application's
+publication API. Producer tests do not contact either hosted service.
+
+No new Microcosm runtime variable or secret is required. Tests isolate ambient
+credentials and caches and use fake HTTP or explicit loopback services.
+
+`--staging-local-only`, `--no-staging` and `--no-staged-dataset` export locally
+without remote submission. `--publish-orrery` explicitly enables submission;
+`--no-publish-orrery` disables it independently. A publication failure does not
+replace the dataset's validation result or prevent HF artifact staging.
+
+Failed builds preserve recorded graph evidence before temporary-file cleanup.
+They publish it, with an `orrery.failure.json` marker, to `orrery-failure/` in
+the attempt evidence directory named by `failure.json`, and leave the output
+directory, including an earlier successful bundle, unchanged. An interrupted
+build preserves the same local evidence and queue entry but does not wait for
+an upload or stage to Hugging Face.
+
+## Retry
+
+```bash
+microcosm-publish-graph --publication-id PUBLICATION_ID \
+  --directory /path/to/attempt/orrery-publication
+```
+
+The preserved directory contains `orrery.upload.json` and the exact files.
+Retries update a separate `publication.status.json`, never the original
+HF receipt snapshot. Pending jobs also resume when a subsequent local service
+starts. A job whose preserved directory has been removed stays pending with
+error code `preserved_files_missing` until its files are restored to the same
+directory; it does not stop delivery of other jobs. Moving this
+service into the `microcosm-emitter` package is a separate, deferred consumer
+change; this implementation does not require that package.

packages/microcosm-build/pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "pandas>=2",
     "scipy>=1.13",
     "huggingface-hub>=0.20",
+    "httpx>=0.27,<1",
     "microcosm-frame>=0.1,<0.2",
     "microcosm-graph>=0.1,<0.2",
     "microcosm-fit>=0.1,<0.2",
@@ -50,6 +51,7 @@ us = [
 uk = ["policyengine-uk>=2.123.0", "microcosm-data>=0.1,<0.2", "h5py>=3", "tables>=3", "openpyxl>=3.1"]
 
 [project.scripts]
+microcosm-publish-graph = "microcosm.build.graph_publication_cli:main"
 microcosm-build-uk = "microcosm.build.uk_runtime.full_build_cli:main"
 microcosm-export-us-l0-refit-h5 = "microcosm.build.us_runtime.l0_refit_export:main"
 

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

@@ -6,11 +6,20 @@
 import tempfile
 from collections.abc import Mapping
 from pathlib import Path
+from typing import TypedDict
 
 from .trace import sha256_file
 
 
-def file_artifact(path: str | Path) -> dict[str, object]:
+class FileArtifact(TypedDict):
+    """The measured identity of a regular file."""
+
+    filename: str
+    sha256: str
+    size_bytes: int
+
+
+def file_artifact(path: str | Path) -> FileArtifact:
     """Bind one regular file without loading its entire payload into memory."""
     source = Path(path)
     if not source.is_file():
@@ -27,7 +36,7 @@ def file_artifact(path: str | Path) -> dict[str, object]:
     return {"filename": source.name, "sha256": digest, "size_bytes": after.st_size}
 
 
-def materialize_bytes(payload: bytes, path: str | Path) -> dict[str, object]:
+def materialize_bytes(payload: bytes, path: str | Path) -> FileArtifact:
     """Write deterministic bytes atomically, including after a graph cache hit."""
     if not isinstance(payload, bytes):
         raise TypeError("Artifact payload must be immutable bytes.")
@@ -76,7 +85,7 @@ def publish_staged_bundle(
     *,
     completion_role: str = "manifest",
     expected: Mapping[str, Mapping[str, object]] | None = None,
-) -> dict[str, dict[str, object]]:
+) -> dict[str, FileArtifact]:
     """Publish a validated bundle with rollback and the completion marker last.
 
     The marker is absent while files change. Handled failures, including

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

@@ -0,0 +1,58 @@
+"""Retry preserved graph uploads without registering a telemetry run."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from microcosm.build.telemetry_emitter import default_spool_path
+from microcosm.build.telemetry_emitter_service.auth import (
+    CollectorSession,
+    validated_origin,
+)
+from microcosm.build.telemetry_emitter_service.constants import PRODUCTION_COLLECTOR_URL
+from microcosm.build.telemetry_emitter_service.graph_publication import (
+    DEFAULT_GRAPH_PUBLICATION_ORIGIN,
+    GraphPublicationDelivery,
+    GraphPublicationQueue,
+)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--spool", type=Path, default=default_spool_path())
+    parser.add_argument("--publication-id", required=True)
+    parser.add_argument(
+        "--directory",
+        type=Path,
+        help="Optional preserved bundle containing orrery.upload.json; re-enqueues after moving machines.",
+    )
+    parser.add_argument("--origin", default=DEFAULT_GRAPH_PUBLICATION_ORIGIN)
+    args = parser.parse_args(argv)
+    queue = GraphPublicationQueue(args.spool)
+    if args.directory is not None:
+        inventory = json.loads((args.directory / "orrery.upload.json").read_text())
+        if inventory["publication_id"] != args.publication_id:
+            parser.error("The preserved inventory has a different publication ID.")
+        queue.enqueue(args.directory, inventory)
+    if queue.receipt(args.publication_id) is None:
+        parser.error("Publication job not found; supply its preserved directory.")
+    queue.retry(args.publication_id)
+    session = CollectorSession(validated_origin(PRODUCTION_COLLECTOR_URL))
+    delivery = GraphPublicationDelivery(
+        queue,
+        credential=session.credential,
+        origin=args.origin,
+        invalidate_credential=session.invalidate,
+    )
+    # Explicit retries process this job only, not unrelated pending jobs.
+    delivery.flush_once(publication_id=args.publication_id)
+    receipt = queue.receipt(args.publication_id)
+    assert receipt is not None, "A queued graph publication must have a receipt."
+    print(json.dumps(receipt, sort_keys=True))
+    return 0 if receipt["status"] == "published" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

@@ -0,0 +1,60 @@
+"""Explicit data contracts for immutable graph publication."""
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal, NotRequired, TypedDict
+
+type PublicationFileRole = Literal[
+    "graph",
+    "schema",
+    "index",
+    "declaration",
+    "manifest",
+    "binding",
+    "summaries",
+    "upstream",
+]
+
+
+type GraphJobStatus = Literal["pending", "published"]
+
+
+class PublicationFile(TypedDict):
+    """One public file, identified by its exact bytes and allowed role."""
+
+    name: str
+    role: PublicationFileRole
+    bytes: int
+    sha256: str
+
+
+class PublicationInventory(TypedDict):
+    """The versioned file inventory submitted to the publication API."""
+
+    version: Literal[1]
+    publication_id: str
+    files: list[PublicationFile]
+
+
+class PublicationReceipt(TypedDict):
+    """Local publication result, including retry and build-snapshot fields."""
+
+    version: Literal[1]
+    publication_id: str | None
+    status: Literal["pending", "published", "skipped", "failed", "no_evidence"]
+    error_code: str | None
+    url: NotRequired[str | None]
+    attempts: NotRequired[int]
+    inventory_sha256: NotRequired[str]
+    preserved_directory: NotRequired[str]
+    recorded_execution: NotRequired[bool]
+
+
+@dataclass(frozen=True)
+class ClaimedGraphPublication:
+    """A job reserved until lease_until; the timestamp also identifies its lease."""
+
+    publication_id: str
+    directory: Path
+    inventory: PublicationInventory
+    lease_until: float

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

@@ -0,0 +1 @@
+

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

@@ -25,7 +25,7 @@
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
-from typing import Any
+from typing import Any, Literal, TypedDict, cast
 
 from microcosm.build.staging_storage import HuggingFaceDatasetStorage
 
@@ -36,6 +36,7 @@
     "STAGED_MANIFEST_FILENAME",
     "STAGED_MANIFEST_SCHEMA",
     "StagedDatasetBundle",
+    "StagedDatasetDelivery",
     "StagedDatasetError",
     "StagedFile",
     "delivery_covers_bundle",
@@ -90,6 +91,30 @@ class StagedDatasetError(ValueError):
     """A staged-dataset bundle or delivery record is malformed."""
 
 
+class StagedFileDigest(TypedDict):
+    """The exact bytes registered for one staged file."""
+
+    sha256: str
+    bytes: int
+
+
+class StagedDatasetDelivery(TypedDict):
+    """Validated result of staging a dataset or graph-evidence bundle."""
+
+    contract_version: Literal[1]
+    mode: Literal["local_and_remote", "local_only", "disabled"]
+    repository: str | None
+    prefix: str | None
+    run_id: str | None
+    revision: str | None
+    status: Literal["uploaded", "already_staged", "failed", "skipped"]
+    error_code: (
+        Literal["UPLOAD_FAILED", "REMOTE_DIFFERS", "REPOSITORY_UNAVAILABLE"] | None
+    )
+    opt_out_reason: str | None
+    files: dict[str, StagedFileDigest]
+
+
 def sha256_file(path: Path | str) -> str:
     digest = hashlib.sha256()
     with Path(path).open("rb") as stream:
@@ -353,7 +378,9 @@ def parse_sha256sums(text: str) -> list[tuple[str, str]]:
     return entries
 
 
-def validate_staged_dataset_delivery(payload: Mapping[str, Any]) -> dict[str, Any]:
+def validate_staged_dataset_delivery(
+    payload: Mapping[str, Any],
+) -> StagedDatasetDelivery:
     """Validate and normalize a staged-dataset delivery record."""
 
     if not isinstance(payload, Mapping):
@@ -423,7 +450,7 @@ def validate_staged_dataset_delivery(payload: Mapping[str, Any]) -> dict[str, An
             or files
         ):
             raise StagedDatasetError("a disabled staged dataset records nothing else.")
-        return record
+        return cast(StagedDatasetDelivery, record)
     if reason is not None:
         raise StagedDatasetError(
             "only a disabled staged dataset carries an opt-out reason."
@@ -439,7 +466,7 @@ def validate_staged_dataset_delivery(payload: Mapping[str, Any]) -> dict[str, An
             )
         if error_code is not None:
             raise StagedDatasetError("a local-only staged dataset has no error code.")
-        return record
+        return cast(StagedDatasetDelivery, record)
     if repository is None:
         raise StagedDatasetError("a remote staged dataset names its repository.")
     if status == "skipped":
@@ -451,12 +478,12 @@ def validate_staged_dataset_delivery(payload: Mapping[str, Any]) -> dict[str, An
             raise StagedDatasetError(
                 "a failed staged dataset carries an error code and no revision."
             )
-        return record
+        return cast(StagedDatasetDelivery, record)
     if error_code is not None:
         raise StagedDatasetError("a delivered staged dataset carries no error code.")
     if status == "uploaded" and revision is None:
         raise StagedDatasetError("an uploaded staged dataset records its revision.")
-    return record
+    return cast(StagedDatasetDelivery, record)
 
 
 def delivery_covers_bundle(
@@ -483,7 +510,7 @@ def delivery_covers_bundle(
     )
 
 
-def disabled_staged_dataset(reason: str) -> dict[str, Any]:
+def disabled_staged_dataset(reason: str) -> StagedDatasetDelivery:
     """Explicit evidence for a deliberate staged-dataset opt-out."""
 
     if not isinstance(reason, str) or not reason.strip():
@@ -506,7 +533,7 @@ def disabled_staged_dataset(reason: str) -> dict[str, Any]:
 
 def local_only_staged_dataset(
     bundle: StagedDatasetBundle, *, prefix: str = DEFAULT_STAGED_DATASET_PREFIX
-) -> dict[str, Any]:
+) -> StagedDatasetDelivery:
     """Evidence for a bundle verified and inventoried on disk but not uploaded."""
 
     return validate_staged_dataset_delivery(
@@ -535,7 +562,7 @@ def stage_bundle(
     storage: HuggingFaceDatasetStorage,
     prefix: str = DEFAULT_STAGED_DATASET_PREFIX,
     message: str | None = None,
-) -> dict[str, Any]:
+) -> StagedDatasetDelivery:
     """Upload the bundle and its sidecars in one commit; record, never raise.
 
     Idempotent on the outputs' digests: a remote ``staged_manifest.json`` whose