fix: make Firstrade live claim atomic in GCS

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

Author: Pigbibi

application/state_persistence.py

@@ -29,6 +29,7 @@ def _object_store():
 class GcsStateStore:
     bucket: str
     prefix: str = "firstrade-platform"
+    client_factory: Callable[..., Any] | None = None
 
     def __post_init__(self) -> None:
         if not str(self.bucket or "").strip():
@@ -74,8 +75,27 @@ def create_json(self, key: str, payload: dict[str, Any]) -> bool:
         """Atomically create a JSON object; return False if it already exists."""
         uri = self._object_uri(key)
         data = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+        bucket_name, separator, object_name = uri[5:].partition("/")
+        if not separator or not bucket_name or not object_name:
+            raise StatePersistenceError(f"Invalid GCS state URI for {key}")
         try:
-            return bool(_object_store().create_text(uri, data, content_type="application/json"))
+            from google.api_core.exceptions import Conflict, PreconditionFailed
+            from google.cloud import storage
+
+            factory = self.client_factory or storage.Client
+            try:
+                client = factory()
+            except TypeError:
+                client = factory(project=None)
+            blob = client.bucket(bucket_name).blob(object_name)
+            blob.upload_from_string(
+                data,
+                content_type="application/json",
+                if_generation_match=0,
+            )
+            return True
+        except (Conflict, PreconditionFailed):
+            return False
         except Exception as exc:
             raise StatePersistenceError(f"GCS atomic create failed for {key}: {exc}") from exc
 

tests/test_strategy_run_claim.py

@@ -2,6 +2,7 @@
     claim_live_strategy_run,
     strategy_run_claim_key,
 )
+from application.state_persistence import GcsStateStore
 
 
 class AtomicFakeStore:
@@ -31,3 +32,48 @@ def test_live_claim_is_create_only_and_permanent():
     )
     assert store.payloads[key]["stage"] == "PENDING_SUBMISSION"
     assert store.payloads[key]["no_order_submitted"] is True
+
+
+def test_gcs_state_store_create_json_uses_generation_zero_precondition():
+    class FakeBlob:
+        def __init__(self):
+            self.payload = None
+            self.content_type = None
+            self.if_generation_match = None
+
+        def upload_from_string(self, payload, *, content_type, if_generation_match):
+            self.payload = payload
+            self.content_type = content_type
+            self.if_generation_match = if_generation_match
+
+    class FakeBucket:
+        def __init__(self, blob):
+            self._blob = blob
+            self.name = ""
+            self.object_name = ""
+
+        def blob(self, object_name):
+            self.object_name = object_name
+            return self._blob
+
+    class FakeClient:
+        def __init__(self, bucket):
+            self._bucket = bucket
+
+        def bucket(self, name):
+            self._bucket.name = name
+            return self._bucket
+
+    blob = FakeBlob()
+    bucket = FakeBucket(blob)
+    store = GcsStateStore(
+        bucket="state-bucket",
+        prefix="runtime",
+        client_factory=lambda: FakeClient(bucket),
+    )
+
+    assert store.create_json("claims/run.json", {"state": "claimed"}) is True
+    assert bucket.name == "state-bucket"
+    assert bucket.object_name == "runtime/claims/run.json"
+    assert blob.content_type == "application/json"
+    assert blob.if_generation_match == 0