Bind Firestore handoff to exact legacy bytes

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

Author: Pigbibi

src/publish.py

@@ -4,6 +4,7 @@
 from copy import deepcopy
 from dataclasses import dataclass
 import hashlib
+import json
 from pathlib import Path
 import re
 from typing import Any
@@ -26,6 +27,10 @@
     "btc_cycle_indicators.json",
 )
 
+LIVE_POOL_LEGACY_EXACT_BYTES_CONTRACT_VERSION = (
+    "qsl.crypto_live_pool_legacy_exact_bytes.v1"
+)
+
 
 @dataclass(frozen=True)
 class PublishSettings:
@@ -304,6 +309,47 @@ def build_firestore_payload(
 ) -> dict[str, Any]:
     symbol_map = dict(artifacts.live_pool_legacy["symbols"])
     symbols = list(symbol_map.keys())
+    exact_bytes = artifacts.live_pool_legacy_path.read_bytes()
+    try:
+        exact_text = exact_bytes.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ValueError("live_pool_legacy.json exact bytes must be valid UTF-8.") from exc
+    try:
+        exact_payload = json.loads(exact_text)
+    except json.JSONDecodeError as exc:
+        raise ValueError("live_pool_legacy.json exact bytes must contain valid JSON.") from exc
+    if not isinstance(exact_payload, dict):
+        raise ValueError("live_pool_legacy.json exact bytes must contain a JSON object.")
+
+    exact_digest = hashlib.sha256(exact_bytes).hexdigest()
+    manifest_entry = artifacts.artifact_manifest.get("artifacts", {}).get(
+        "live_pool_legacy", {}
+    )
+    identity_entry = artifacts.runtime_evidence_identity.get("artifacts", {}).get(
+        "live_pool_legacy", {}
+    )
+    if (
+        manifest_entry.get("sha256") != exact_digest
+        or identity_entry.get("sha256") != exact_digest
+    ):
+        raise ValueError(
+            "live_pool_legacy.json exact bytes digest mismatch with artifact manifest or runtime identity."
+        )
+
+    if (
+        exact_payload.get("as_of_date") != artifacts.as_of_date
+        or exact_payload.get("version") != artifacts.version
+        or exact_payload.get("mode") != settings.mode
+        or exact_payload.get("symbols") != artifacts.live_pool_legacy.get("symbols")
+        or exact_payload.get("symbol_map")
+        != artifacts.live_pool_legacy.get("symbol_map")
+        or list(exact_payload.get("symbols", {})) != symbols
+        or exact_payload.get("symbol_map") != symbol_map
+    ):
+        raise ValueError(
+            "live_pool_legacy.json exact bytes do not match Firestore convenience fields."
+        )
+
     generated_at = pd.Timestamp.now(tz="UTC").isoformat()
     return {
         "as_of_date": artifacts.as_of_date,
@@ -325,6 +371,11 @@ def build_firestore_payload(
         "generated_at": generated_at,
         "source_project": settings.source_project,
         "runtime_evidence_identity": deepcopy(artifacts.runtime_evidence_identity),
+        "live_pool_legacy_exact_bytes": {
+            "contract_version": LIVE_POOL_LEGACY_EXACT_BYTES_CONTRACT_VERSION,
+            "encoding": "utf-8",
+            "utf8_text": exact_text,
+        },
     }
 
 

tests/test_release_contract.py

@@ -10,6 +10,7 @@
 
 from src.publish import (
     PublishSettings,
+    ReleaseArtifacts,
     build_firestore_payload,
     build_release_manifest,
     build_storage_layout,
@@ -31,6 +32,21 @@ def sha256_file(path: Path) -> str:
     return digest.hexdigest()
 
 
+def build_publish_settings() -> PublishSettings:
+    return PublishSettings(
+        enabled=False,
+        dry_run=True,
+        mode="core_major",
+        project_id=None,
+        cloud_bucket=None,
+        cloud_root_prefix="crypto-live-pool-pipelines",
+        firestore_collection="strategy",
+        firestore_document="CRYPTO_LIVE_POOL_ROTATION_LIVE_POOL",
+        source_project="crypto-live-pool-pipelines",
+        upload_current_pointer=False,
+    )
+
+
 class ReleaseContractValidationTests(unittest.TestCase):
     def build_outputs(
         self,
@@ -177,6 +193,21 @@ def build_outputs(
                 },
             )
 
+    def build_runtime_identity_outputs(self, root: Path) -> Path:
+        self.build_outputs(
+            root,
+            include_manifest=True,
+            include_runtime_evidence_identity=True,
+        )
+        return root / "data" / "output"
+
+    def load_publish_context(
+        self, output_dir: Path
+    ) -> tuple[ReleaseArtifacts, PublishSettings, dict[str, object]]:
+        artifacts = load_release_artifacts(output_dir, "core_major")
+        settings = build_publish_settings()
+        return artifacts, settings, build_storage_layout(settings, artifacts)
+
     def test_validate_release_outputs_accepts_consistent_contract(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             root = Path(tmp_dir)
@@ -236,6 +267,110 @@ def test_identity_is_canonical_across_artifact_release_and_firestore(self) -> No
         self.assertEqual(release_manifest["runtime_evidence_identity"], identity)
         self.assertEqual(firestore_payload["runtime_evidence_identity"], identity)
 
+    def test_firestore_payload_preserves_exact_legacy_artifact_bytes(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            output_dir = self.build_runtime_identity_outputs(root)
+            legacy_path = output_dir / "live_pool_legacy.json"
+            legacy_payload = json.loads(legacy_path.read_text(encoding="utf-8"))
+            exact_bytes = ("\n" + json.dumps(legacy_payload, separators=(", ", ": ")) + "\n").encode(
+                "utf-8"
+            )
+            legacy_path.write_bytes(exact_bytes)
+
+            manifest_path = output_dir / "artifact_manifest.json"
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+            exact_digest = hashlib.sha256(exact_bytes).hexdigest()
+            manifest["artifacts"]["live_pool_legacy"]["sha256"] = exact_digest
+            manifest["runtime_evidence_identity"]["artifacts"]["live_pool_legacy"][
+                "sha256"
+            ] = exact_digest
+            write_json(manifest_path, manifest)
+
+            artifacts, settings, storage_layout = self.load_publish_context(output_dir)
+            firestore_payload = build_firestore_payload(
+                settings,
+                artifacts,
+                storage_layout,
+            )
+
+        handoff = firestore_payload["live_pool_legacy_exact_bytes"]
+        self.assertEqual(
+            handoff["contract_version"],
+            "qsl.crypto_live_pool_legacy_exact_bytes.v1",
+        )
+        self.assertEqual(handoff["encoding"], "utf-8")
+        self.assertEqual(handoff["utf8_text"].encode("utf-8"), exact_bytes)
+
+    def test_firestore_payload_rejects_mutated_legacy_bytes_with_unchanged_identity(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            output_dir = self.build_runtime_identity_outputs(root)
+            artifacts, settings, storage_layout = self.load_publish_context(output_dir)
+            with artifacts.live_pool_legacy_path.open("ab") as handle:
+                handle.write(b"\n")
+
+            with self.assertRaisesRegex(ValueError, "digest mismatch"):
+                build_firestore_payload(
+                    settings,
+                    artifacts,
+                    storage_layout,
+                )
+
+    def test_firestore_payload_rejects_top_level_convenience_field_mismatch(self) -> None:
+        for field in ("symbols", "symbol_map"):
+            with self.subTest(field=field), tempfile.TemporaryDirectory() as tmp_dir:
+                root = Path(tmp_dir)
+                output_dir = self.build_runtime_identity_outputs(root)
+                artifacts, settings, storage_layout = self.load_publish_context(output_dir)
+                if field == "symbols":
+                    artifacts.live_pool_legacy[field].pop("TRXUSDT")
+                else:
+                    artifacts.live_pool_legacy[field]["TRXUSDT"] = {
+                        "base_asset": "MISMATCH"
+                    }
+
+                with self.assertRaisesRegex(ValueError, "convenience fields"):
+                    build_firestore_payload(
+                        settings,
+                        artifacts,
+                        storage_layout,
+                    )
+
+    def test_firestore_payload_rejects_invalid_legacy_artifact_bytes(self) -> None:
+        cases = {
+            "invalid_utf8": (b"\xff", "valid UTF-8"),
+            "invalid_json": (b"{", "valid JSON"),
+            "non_object": (b"[]", "JSON object"),
+        }
+        for label, (invalid_bytes, expected_error) in cases.items():
+            with self.subTest(label=label), tempfile.TemporaryDirectory() as tmp_dir:
+                root = Path(tmp_dir)
+                output_dir = self.build_runtime_identity_outputs(root)
+                artifacts, settings, storage_layout = self.load_publish_context(output_dir)
+                artifacts.live_pool_legacy_path.write_bytes(invalid_bytes)
+
+                with self.assertRaisesRegex(ValueError, expected_error):
+                    build_firestore_payload(
+                        settings,
+                        artifacts,
+                        storage_layout,
+                    )
+
+    def test_firestore_payload_rejects_missing_legacy_artifact(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            output_dir = self.build_runtime_identity_outputs(root)
+            artifacts, settings, storage_layout = self.load_publish_context(output_dir)
+            artifacts.live_pool_legacy_path.unlink()
+
+            with self.assertRaises(FileNotFoundError):
+                build_firestore_payload(
+                    settings,
+                    artifacts,
+                    storage_layout,
+                )
+
     def test_artifact_byte_mutation_breaks_runtime_identity_before_publish(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             root = Path(tmp_dir)