fix: hand off lifecycle preflight baselines

.github/workflows/reusable-drift-check.yml

@@ -16,6 +16,10 @@ on:
         required: false
         type: string
         default: "3.11"
+      quant_platform_kit_ref:
+        required: false
+        type: string
+        default: "main"
       ai_gateway_service_url:
         required: false
         type: string
@@ -31,6 +35,10 @@ on:
         required: false
         type: string
         default: ""
+      lifecycle_preflight_artifact:
+        required: false
+        type: string
+        default: ""
     secrets:
       codex_audit_service_url:
         required: false
@@ -131,7 +139,7 @@ jobs:
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/QuantPlatformKit
-          ref: ${{ github.workflow_sha }}
+          ref: ${{ inputs.quant_platform_kit_ref }}
           path: external/QuantPlatformKit
 
       - name: Set up Python
@@ -146,6 +154,13 @@ jobs:
           python -m pip install -e . pandas
           python -m pip install --no-deps -e external/QuantPlatformKit
 
+      - name: Download lifecycle preflight artifact
+        if: inputs.lifecycle_preflight_artifact != ''
+        uses: actions/download-artifact@v4
+        with:
+          name: ${{ inputs.lifecycle_preflight_artifact }}
+          path: ${{ env.LIFECYCLE_LOCAL_ROOT }}
+
       - name: Build lifecycle performance snapshots
         run: quant-lifecycle monitor --domain ${{ inputs.strategy_domain }}
 

src/quant_platform_kit/strategy_lifecycle/performance_store.py

@@ -102,13 +102,21 @@ def _cloud_path(self, key: str) -> str:
         clean = _clean_key(key)
         return f"{prefix}/{clean}" if prefix else clean
 
+    def _cloud_uri(self, key: str) -> str:
+        return f"gs://{self.cloud_bucket}/{self._cloud_path(key)}"
+
+    def _cloud_key(self, uri: str) -> str:
+        bucket_prefix = f"gs://{self.cloud_bucket}/"
+        path = uri[len(bucket_prefix) :] if uri.startswith(bucket_prefix) else uri
+        prefix = self.cloud_prefix.strip("/")
+        return path[len(prefix) + 1 :] if prefix and path.startswith(f"{prefix}/") else path
+
     def _read_cloud_json(self, key: str) -> dict[str, Any] | None:
         if not self.cloud_bucket:
             return None
         try:
             store = self._object_store()
-            path = self._cloud_path(key)
-            raw = store.read_bytes(self.cloud_bucket, path)
+            raw = store.read_bytes(self._cloud_uri(key))
             data = json.loads(raw.decode("utf-8"))
             return data if isinstance(data, Mapping) else None
         except Exception:
@@ -118,15 +126,17 @@ def _write_cloud_json(self, key: str, payload: Mapping[str, Any]) -> None:
         if not self.cloud_bucket:
             return
         store = self._object_store()
-        path = self._cloud_path(key)
-        store.write_bytes(self.cloud_bucket, path, json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8"))
+        store.write_bytes(
+            self._cloud_uri(key),
+            json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8"),
+        )
 
     def _list_cloud_keys(self, prefix: str) -> list[str]:
         if not self.cloud_bucket:
             return []
         try:
             store = self._object_store()
-            return store.list_keys(self.cloud_bucket, self._cloud_path(prefix))
+            return [self._cloud_key(uri) for uri in store.list(self._cloud_uri(prefix))]
         except Exception:
             return []
 

tests/test_lifecycle_performance_store.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 from unittest.mock import patch
 
+from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
 from quant_platform_kit.strategy_lifecycle.performance_store import (
     DEFAULT_LOCAL_ROOT,
     PerformanceStore,
@@ -26,6 +27,53 @@ def test_from_env_uses_default_local_root_without_override(self) -> None:
             store = PerformanceStore.from_env()
         self.assertEqual(store.local_root, DEFAULT_LOCAL_ROOT)
 
+    def test_cloud_backtest_storage_uses_uri_object_store_contract(self) -> None:
+        class UriObjectStore:
+            def __init__(self) -> None:
+                self.objects: dict[str, bytes] = {}
+
+            def read_bytes(self, uri: str) -> bytes:
+                return self.objects[uri]
+
+            def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str:
+                self.objects[uri] = data
+                return uri
+
+            def list(self, prefix: str) -> list[str]:
+                return sorted(uri for uri in self.objects if uri.startswith(prefix))
+
+        with tempfile.TemporaryDirectory() as tmp:
+            object_store = UriObjectStore()
+            store = PerformanceStore(
+                cloud_bucket="lifecycle-bucket",
+                cloud_prefix="production",
+                local_root=Path(tmp),
+            )
+            result = BacktestResult(
+                strategy_profile="global_etf_rotation",
+                domain="us_equity",
+                param_set_id="baseline",
+                params={},
+                param_version=1,
+                sharpe_ratio=1.0,
+                calmar_ratio=1.0,
+                max_drawdown=-0.1,
+                cagr=0.2,
+                volatility=0.2,
+                win_rate=0.55,
+            )
+            with patch(
+                "quant_platform_kit.strategy_lifecycle.performance_store.get_object_store",
+                return_value=object_store,
+            ):
+                store.save_backtest_result(result)
+                keys = store._list_cloud_keys("backtest/us_equity/global_etf_rotation/")
+                payload = store._read_cloud_json(keys[0])
+
+        self.assertEqual(len(keys), 1)
+        self.assertTrue(keys[0].startswith("backtest/us_equity/global_etf_rotation/"))
+        self.assertEqual(payload["strategy_profile"], "global_etf_rotation")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_reusable_drift_workflow.py

@@ -15,9 +15,11 @@ def test_reusable_drift_workflow_enforces_lifecycle_preflight() -> None:
     assert "snapshot_repository_token:" in workflow
     assert "snapshot_checkout_path:" in workflow
     assert "ai_gateway_service_url:" in workflow
+    assert "quant_platform_kit_ref:" in workflow
     assert "lifecycle_performance_bucket:" in workflow
     assert "caller_event_name:" in workflow
     assert "caller_pr_head_repository:" in workflow
+    assert "lifecycle_preflight_artifact:" in workflow
     assert "codex_audit_service_url:" in workflow
     assert 'python-version: ${{ inputs.python_version }}' in workflow
     assert "LIFECYCLE_PERFORMANCE_BUCKET: ${{ inputs.lifecycle_performance_bucket || vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }}" in workflow
@@ -38,7 +40,10 @@ def test_reusable_drift_workflow_enforces_lifecycle_preflight() -> None:
     assert 'repository: ${{ inputs.snapshot_repository }}' in workflow
     assert 'path: ${{ inputs.snapshot_checkout_path }}' in workflow
     assert 'token: ${{ secrets.snapshot_repository_token || github.token }}' in workflow
-    assert 'ref: ${{ github.workflow_sha }}' in workflow
+    assert 'ref: ${{ inputs.quant_platform_kit_ref }}' in workflow
+    assert "Download lifecycle preflight artifact" in workflow
+    assert "actions/download-artifact@v4" in workflow
+    assert "name: ${{ inputs.lifecycle_preflight_artifact }}" in workflow
     assert 'GH_TOKEN: ${{ github.token }}' in workflow
     assert 'os.environ["CODEX_AUDIT_ORG"] = owner' in workflow
     assert 'os.environ["CODEX_AUDIT_ORCHESTRATOR_REPO"] = repository' in workflow