Reuse validated legacy snapshot for upload

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

Author: Pigbibi

src/publish.py

@@ -434,6 +434,8 @@ def upload_release_artifacts(
     settings: PublishSettings,
     artifacts: ReleaseArtifacts,
     storage_layout: dict[str, Any],
+    *,
+    live_pool_legacy_exact_bytes: bytes,
 ) -> None:
     if settings.dry_run:
         return
@@ -462,9 +464,14 @@ def upload_release_artifacts(
             files[opt_file] = opt_path
     for filename, local_path in files.items():
         object_info = storage_layout["objects"][filename]
-        store.write_bytes(object_info["release_uri"], local_path.read_bytes())
+        upload_bytes = (
+            live_pool_legacy_exact_bytes
+            if filename == "live_pool_legacy.json"
+            else local_path.read_bytes()
+        )
+        store.write_bytes(object_info["release_uri"], upload_bytes)
         if settings.upload_current_pointer:
-            store.write_bytes(object_info["current_uri"], local_path.read_bytes())
+            store.write_bytes(object_info["current_uri"], upload_bytes)
 
 
 def publish_firestore_summary(settings: PublishSettings, firestore_payload: dict[str, Any]) -> None:
@@ -514,6 +521,9 @@ def run_release_publish(
     artifacts = load_release_artifacts(config["paths"].output_dir, settings.mode)
     storage_layout = build_storage_layout(settings, artifacts)
     firestore_payload = build_firestore_payload(settings, artifacts, storage_layout)
+    live_pool_legacy_exact_bytes = firestore_payload["live_pool_legacy_exact_bytes"][
+        "utf8_text"
+    ].encode("utf-8")
     manifest = build_release_manifest(settings, artifacts, storage_layout, firestore_payload)
     manifest_path = write_release_manifest(artifacts.output_dir, manifest)
     validation = assert_release_outputs(
@@ -533,8 +543,18 @@ def run_release_publish(
         or firestore_payload.get("runtime_evidence_identity") != artifacts.runtime_evidence_identity
     ):
         raise ValueError("Runtime evidence identity changed before publish.")
+    if (
+        live_pool_legacy_exact_bytes
+        != artifacts.live_pool_legacy_path.read_bytes()
+    ):
+        raise ValueError("Validated live_pool_legacy.json exact bytes changed before publish.")
 
-    upload_release_artifacts(settings, artifacts, storage_layout)
+    upload_release_artifacts(
+        settings,
+        artifacts,
+        storage_layout,
+        live_pool_legacy_exact_bytes=live_pool_legacy_exact_bytes,
+    )
     publish_firestore_summary(settings, firestore_payload)
 
     return {

tests/test_release_contract.py

@@ -5,6 +5,7 @@
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 import pandas as pd
 
@@ -16,6 +17,7 @@
     build_storage_layout,
     ensure_publish_preflight,
     load_release_artifacts,
+    upload_release_artifacts,
 )
 from src.release_contract import validate_release_outputs
 
@@ -302,6 +304,66 @@ def test_firestore_payload_preserves_exact_legacy_artifact_bytes(self) -> None:
         self.assertEqual(handoff["encoding"], "utf-8")
         self.assertEqual(handoff["utf8_text"].encode("utf-8"), exact_bytes)
 
+    def test_upload_reuses_validated_legacy_snapshot_after_path_mutation(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            output_dir = self.build_runtime_identity_outputs(root)
+            artifacts = load_release_artifacts(output_dir, "core_major")
+            settings = PublishSettings(
+                enabled=True,
+                dry_run=False,
+                mode="core_major",
+                project_id="test-project",
+                cloud_bucket="test-bucket",
+                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=True,
+            )
+            storage_layout = build_storage_layout(settings, artifacts)
+            firestore_payload = build_firestore_payload(
+                settings,
+                artifacts,
+                storage_layout,
+            )
+            validated_bytes = firestore_payload["live_pool_legacy_exact_bytes"][
+                "utf8_text"
+            ].encode("utf-8")
+            artifacts.live_pool_legacy_path.write_bytes(b'{"mutated": true}\n')
+
+            uploaded: dict[str, bytes] = {}
+
+            class FakeStore:
+                def write_bytes(self, uri: str, payload: bytes) -> None:
+                    uploaded[uri] = payload
+
+            with patch(
+                "quant_platform_kit.cloud.get_object_store",
+                return_value=FakeStore(),
+            ):
+                upload_release_artifacts(
+                    settings,
+                    artifacts,
+                    storage_layout,
+                    live_pool_legacy_exact_bytes=validated_bytes,
+                )
+
+        legacy_objects = storage_layout["objects"]["live_pool_legacy.json"]
+        release_bytes = uploaded[legacy_objects["release_uri"]]
+        current_bytes = uploaded[legacy_objects["current_uri"]]
+        manifest_digest = artifacts.artifact_manifest["artifacts"][
+            "live_pool_legacy"
+        ]["sha256"]
+        identity_digest = artifacts.runtime_evidence_identity["artifacts"][
+            "live_pool_legacy"
+        ]["sha256"]
+        self.assertEqual(release_bytes, validated_bytes)
+        self.assertEqual(current_bytes, validated_bytes)
+        self.assertEqual(hashlib.sha256(validated_bytes).hexdigest(), manifest_digest)
+        self.assertEqual(hashlib.sha256(release_bytes).hexdigest(), manifest_digest)
+        self.assertEqual(identity_digest, manifest_digest)
+
     def test_firestore_payload_rejects_mutated_legacy_bytes_with_unchanged_identity(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             root = Path(tmp_dir)