feat(release): bind producer provenance

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/export.py

@@ -2,6 +2,8 @@
 
 import hashlib
 from pathlib import Path
+import re
+import subprocess
 from typing import Any
 
 import pandas as pd
@@ -13,6 +15,12 @@
 DEFAULT_STRATEGY_PROFILE = "crypto_live_pool_rotation"
 DEFAULT_ARTIFACT_TYPE = "live_pool"
 DEFAULT_ARTIFACT_CONTRACT_VERSION = "crypto_live_pool_rotation.live_pool.v1"
+REQUIRED_IDENTITY_ARTIFACTS = (
+    "live_pool",
+    "live_pool_legacy",
+    "latest_ranking",
+    "latest_universe",
+)
 
 
 def _sha256_file(path: Path) -> str:
@@ -23,6 +31,49 @@ def _sha256_file(path: Path) -> str:
     return digest.hexdigest()
 
 
+def resolve_clean_source_revision(repo_root: Path | None = None) -> str:
+    """Resolve the commit whose clean tracked tree is executing this producer."""
+    root = Path(repo_root) if repo_root is not None else Path(__file__).resolve().parents[1]
+    try:
+        revision = subprocess.run(
+            ["git", "rev-parse", "--verify", "HEAD^{commit}"],
+            cwd=root,
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.strip()
+        subprocess.run(
+            ["git", "diff", "--quiet", "HEAD", "--"],
+            cwd=root,
+            check=True,
+        )
+        subprocess.run(
+            ["git", "diff", "--cached", "--quiet", "HEAD", "--"],
+            cwd=root,
+            check=True,
+        )
+    except (OSError, subprocess.CalledProcessError) as exc:
+        raise RuntimeError("Cannot bind artifact bytes to a clean producer commit.") from exc
+    if not re.fullmatch(r"[0-9a-f]{40}", revision):
+        raise RuntimeError("Producer HEAD did not resolve to a lowercase 40-character commit.")
+    return revision
+
+
+def _normalize_input_timestamp(value: Any, *, as_of_date: str) -> str:
+    timestamp = pd.Timestamp(value)
+    if pd.isna(timestamp):
+        raise ValueError("input_timestamp must be a finite panel date.")
+    if timestamp.tzinfo is None:
+        timestamp = timestamp.tz_localize("UTC")
+    else:
+        timestamp = timestamp.tz_convert("UTC")
+    timestamp = timestamp.normalize()
+    expected = pd.Timestamp(as_of_date).tz_localize("UTC")
+    if timestamp != expected:
+        raise ValueError("input_timestamp must equal live_pool.as_of_date at UTC midnight.")
+    return timestamp.strftime("%Y-%m-%dT00:00:00Z")
+
+
 def export_latest_universe(panel: pd.DataFrame, output_dir: str | Any, as_of_date: pd.Timestamp) -> dict[str, Any]:
     """Export the latest dynamic universe to JSON."""
     snapshot = panel.xs(as_of_date, level="date")
@@ -170,6 +221,7 @@ def build_strategy_artifact_manifest(
     artifact_type: str = DEFAULT_ARTIFACT_TYPE,
     contract_version: str = DEFAULT_ARTIFACT_CONTRACT_VERSION,
     source_project: str = "crypto-live-pool-pipelines",
+    input_timestamp: Any,
     generated_at: Any | None = None,
 ) -> dict[str, Any]:
     """Build the profile-aware artifact manifest consumed by downstream runtimes."""
@@ -190,8 +242,8 @@ def build_strategy_artifact_manifest(
     artifacts = {}
     for artifact_name, filename in artifact_files.items():
         path = output_path / filename
-        if not path.exists():
-            continue
+        if not path.is_file():
+            raise FileNotFoundError(f"Required identity artifact is missing: {path}")
         artifacts[artifact_name] = {
             "path": filename,
             "sha256": _sha256_file(path),
@@ -209,6 +261,16 @@ def build_strategy_artifact_manifest(
     mode = str(live_pool.get("mode", "")).strip()
     version = str(live_pool.get("version", "")).strip()
     source_project_text = str(live_pool.get("source_project") or source_project)
+    input_timestamp_text = _normalize_input_timestamp(input_timestamp, as_of_date=as_of_date)
+    runtime_evidence_identity = {
+        "strategy_profile": str(strategy_profile),
+        "mode": mode,
+        "source_revision": resolve_clean_source_revision(),
+        "input_timestamp": input_timestamp_text,
+        "artifact_contract": str(contract_version),
+        "artifact_version": version,
+        "artifacts": {name: dict(artifacts[name]) for name in REQUIRED_IDENTITY_ARTIFACTS},
+    }
     return {
         "manifest_type": "strategy_artifact",
         "contract_version": str(contract_version),
@@ -225,6 +287,7 @@ def build_strategy_artifact_manifest(
         "generated_at": generated_at_text,
         "primary_artifact": "live_pool",
         "artifacts": artifacts,
+        "runtime_evidence_identity": runtime_evidence_identity,
     }
 
 
@@ -250,6 +313,7 @@ def export_strategy_artifact_manifest(
     artifact_type: str = DEFAULT_ARTIFACT_TYPE,
     contract_version: str = DEFAULT_ARTIFACT_CONTRACT_VERSION,
     source_project: str = "crypto-live-pool-pipelines",
+    input_timestamp: Any,
 ) -> dict[str, Any]:
     manifest = build_strategy_artifact_manifest(
         output_dir=output_dir,
@@ -258,6 +322,7 @@ def export_strategy_artifact_manifest(
         artifact_type=artifact_type,
         contract_version=contract_version,
         source_project=source_project,
+        input_timestamp=input_timestamp,
     )
     write_json(Path(output_dir) / "artifact_manifest.json", manifest)
     return manifest

src/pipeline.py

@@ -388,6 +388,7 @@ def build_live_pool_outputs(
     else:
         panel = panel.join(result.predictions, how="left")
     panel = build_final_scores(panel, config)
+    input_timestamp = resolve_scoring_input_timestamp(panel, score_mask)
 
     output_dir = config["paths"].output_dir
     export_latest_universe(panel, output_dir, latest_date)
@@ -418,6 +419,7 @@ def build_live_pool_outputs(
         output_dir=output_dir,
         live_pool=live_payload,
         source_project=source_project,
+        input_timestamp=input_timestamp,
     )
 
     # Compute and export BTC cycle indicators for crypto DCA strategies
@@ -455,3 +457,14 @@ def build_live_pool_outputs(
         "ml_backend": result.ml_backend,
         "universe_mode": resolved_mode,
     }
+
+
+def resolve_scoring_input_timestamp(panel: pd.DataFrame, score_mask: Any) -> pd.Timestamp:
+    """Return the maximum date among rows admitted to final scoring/export."""
+    scoring_rows = panel.loc[score_mask]
+    if scoring_rows.empty:
+        raise ValueError("No panel rows were admitted to final scoring/export.")
+    dates = pd.to_datetime(scoring_rows.index.get_level_values("date"), errors="coerce")
+    if dates.isna().any():
+        raise ValueError("Final scoring/export rows contain an invalid date.")
+    return pd.Timestamp(dates.max()).normalize()

src/publish.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import os
+from copy import deepcopy
 from dataclasses import dataclass
+import hashlib
 from pathlib import Path
+import re
 from typing import Any
 
 import pandas as pd
@@ -53,6 +56,7 @@ class ReleaseArtifacts:
     live_pool: dict[str, Any]
     live_pool_legacy: dict[str, Any]
     artifact_manifest: dict[str, Any]
+    runtime_evidence_identity: dict[str, Any]
 
 
 def parse_bool(value: Any, default: bool = False) -> bool:
@@ -117,6 +121,59 @@ def _require_file(path: Path) -> None:
         raise FileNotFoundError(f"Required release artifact is missing: {path}")
 
 
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _validate_runtime_evidence_identity(
+    *,
+    identity: Any,
+    artifact_manifest: dict[str, Any],
+    live_pool: dict[str, Any],
+    paths: dict[str, Path],
+) -> dict[str, Any]:
+    if not isinstance(identity, dict):
+        raise ValueError("artifact_manifest.json runtime_evidence_identity must be an object.")
+    expected_artifacts = {
+        "live_pool": paths["live_pool.json"],
+        "live_pool_legacy": paths["live_pool_legacy.json"],
+        "latest_ranking": paths["latest_ranking.csv"],
+        "latest_universe": paths["latest_universe.json"],
+    }
+    manifest_artifacts = artifact_manifest.get("artifacts")
+    identity_artifacts = identity.get("artifacts")
+    if not isinstance(manifest_artifacts, dict) or set(manifest_artifacts) != set(expected_artifacts):
+        raise ValueError("artifact_manifest.json must bind exactly the four release artifacts.")
+    if not isinstance(identity_artifacts, dict) or identity_artifacts != manifest_artifacts:
+        raise ValueError("runtime_evidence_identity artifacts must equal artifact_manifest.json artifacts.")
+    for name, path in expected_artifacts.items():
+        entry = manifest_artifacts.get(name)
+        if (
+            not isinstance(entry, dict)
+            or entry.get("path") != path.name
+            or entry.get("sha256") != _sha256_file(path)
+        ):
+            raise ValueError(f"Runtime identity digest mismatch for {name}.")
+    if identity.get("strategy_profile") != artifact_manifest.get("strategy_profile"):
+        raise ValueError("Runtime identity strategy_profile mismatch.")
+    if identity.get("mode") != live_pool.get("mode"):
+        raise ValueError("Runtime identity mode mismatch.")
+    if identity.get("artifact_contract") != artifact_manifest.get("contract_version"):
+        raise ValueError("Runtime identity artifact_contract mismatch.")
+    if identity.get("artifact_version") != live_pool.get("version"):
+        raise ValueError("Runtime identity artifact_version mismatch.")
+    if not re.fullmatch(r"[0-9a-f]{40}", str(identity.get("source_revision", ""))):
+        raise ValueError("Runtime identity source_revision is invalid.")
+    expected_timestamp = f"{live_pool.get('as_of_date')}T00:00:00Z"
+    if identity.get("input_timestamp") != expected_timestamp:
+        raise ValueError("Runtime identity input_timestamp mismatch.")
+    return deepcopy(identity)
+
+
 def load_release_artifacts(output_dir: Path | str, mode: str) -> ReleaseArtifacts:
     output_path = Path(output_dir)
     paths = {name: output_path / name for name in REQUIRED_OUTPUT_FILES}
@@ -147,6 +204,12 @@ def load_release_artifacts(output_dir: Path | str, mode: str) -> ReleaseArtifact
         raise ValueError("live_pool_legacy.json must contain a non-empty symbols mapping.")
 
     version = build_release_version(as_of_date, mode)
+    runtime_evidence_identity = _validate_runtime_evidence_identity(
+        identity=artifact_manifest.get("runtime_evidence_identity"),
+        artifact_manifest=artifact_manifest,
+        live_pool=live_pool,
+        paths=paths,
+    )
     return ReleaseArtifacts(
         as_of_date=as_of_date,
         version=version,
@@ -161,6 +224,7 @@ def load_release_artifacts(output_dir: Path | str, mode: str) -> ReleaseArtifact
         live_pool=live_pool,
         live_pool_legacy=live_pool_legacy,
         artifact_manifest=artifact_manifest,
+        runtime_evidence_identity=runtime_evidence_identity,
     )
 
 
@@ -260,6 +324,7 @@ def build_firestore_payload(
         "artifact_contract_version": str(artifacts.artifact_manifest.get("contract_version", "")),
         "generated_at": generated_at,
         "source_project": settings.source_project,
+        "runtime_evidence_identity": deepcopy(artifacts.runtime_evidence_identity),
     }
 
 
@@ -269,6 +334,8 @@ def build_release_manifest(
     storage_layout: dict[str, Any],
     firestore_payload: dict[str, Any],
 ) -> dict[str, Any]:
+    if firestore_payload.get("runtime_evidence_identity") != artifacts.runtime_evidence_identity:
+        raise ValueError("Firestore runtime_evidence_identity must equal artifact_manifest.json.")
     return {
         "version": artifacts.version,
         "mode": settings.mode,
@@ -284,6 +351,7 @@ def build_release_manifest(
             "live_pool_legacy": storage_layout["objects"]["live_pool_legacy.json"],
             "artifact_manifest": storage_layout["objects"]["artifact_manifest.json"],
         },
+        "runtime_evidence_identity": deepcopy(artifacts.runtime_evidence_identity),
         "firestore": {
             "collection": settings.firestore_collection,
             "document": settings.firestore_document,
@@ -392,8 +460,15 @@ def run_release_publish(
         max_age_days=max_age_days,
         require_manifest=True,
         require_artifact_manifest=True,
+        require_runtime_evidence_identity=True,
         require_freshness=require_freshness,
     )
+    artifacts = load_release_artifacts(artifacts.output_dir, settings.mode)
+    if (
+        manifest.get("runtime_evidence_identity") != artifacts.runtime_evidence_identity
+        or firestore_payload.get("runtime_evidence_identity") != artifacts.runtime_evidence_identity
+    ):
+        raise ValueError("Runtime evidence identity changed before publish.")
 
     upload_release_artifacts(settings, artifacts, storage_layout)
     publish_firestore_summary(settings, firestore_payload)