fix(risk): require gs:// claim store for live account-owner fence

application/runtime_composer.py

@@ -276,6 +276,14 @@ def build_rebalance_config(
         plugin_error_lines = tuple(build_plugin_error_lines(strategy_plugin_error))
         fractional_buy_execution = fractional_buy_execution_enabled(self.strategy_profile)
         notional_buy_compat_mode = dca_compat_mode_enabled(self.strategy_profile)
+        execution_state_store = build_execution_marker_store_from_env(
+            env_reader=self.env_reader,
+            gcp_project_id=self.project_id,
+        )
+        if not self.dry_run_only and not str(execution_state_store.cloud_prefix_uri or "").startswith("gs://"):
+            raise RuntimeError(
+                "LongBridge live execution requires a gs:// execution state URI for atomic claims"
+            )
         return LongBridgeRebalanceConfig(
             limit_sell_discount=self.limit_sell_discount,
             limit_buy_premium=self.limit_buy_premium,
@@ -305,10 +313,7 @@ def build_rebalance_config(
                 dry_run_only=self.dry_run_only,
                 account_scope=self.account_region,
             ),
-            execution_state_store=build_execution_marker_store_from_env(
-                env_reader=self.env_reader,
-                gcp_project_id=self.project_id,
-            ),
+            execution_state_store=execution_state_store,
             execution_state_account_scope=self.account_region,
             physical_account_id=_resolve_configured_physical_account_id(
                 env_reader=self.env_reader

tests/test_request_handling.py

@@ -775,6 +775,7 @@ def attach_strategy_plugin_report(self, *_args, **_kwargs):
         self.assertNotIn("probe failed", notification["compact_text"])
         self.assertLessEqual(len(notification["compact_text"]), 3500)
 
+    @patch.dict(os.environ, {"LONGBRIDGE_EXECUTION_STATE_CLOUD_URI": "gs://unit-test-execution/claims"})
     def test_run_strategy_emits_structured_runtime_events(self):
         module = load_module()
         observed = []
@@ -792,6 +793,7 @@ def test_run_strategy_emits_structured_runtime_events(self):
         )
         self.assertTrue(all(run_id == "run-001" for run_id, _event, _fields in observed))
 
+    @patch.dict(os.environ, {"LONGBRIDGE_EXECUTION_STATE_CLOUD_URI": "gs://unit-test-execution/claims"})
     def test_run_strategy_market_hours_tuple_without_error_does_not_warn(self):
         module = load_module()
         observed = []
@@ -808,6 +810,20 @@ def test_run_strategy_market_hours_tuple_without_error_does_not_warn(self):
             [event for event, _fields in observed],
         )
 
+    @patch.dict(os.environ, {}, clear=True)
+    def test_run_strategy_without_durable_claim_store_never_enters_cycle(self):
+        module = load_module()
+        cycles = []
+        events = []
+        module.is_market_open_now = lambda **_kwargs: True
+        module.run_rebalance_cycle = lambda **_kwargs: cycles.append("called")
+        module.emit_runtime_log = lambda _context, event, **_kwargs: events.append(event)
+
+        self.assertFalse(module.run_strategy())
+        self.assertEqual(cycles, [])
+        self.assertIn("strategy_cycle_failed", events)
+        self.assertNotIn("strategy_cycle_completed", events)
+
     def test_run_strategy_error_notification_is_compact_and_bounded(self):
         module = load_module()
         observed = {}
@@ -942,6 +958,7 @@ def fake_dispatch(signals, **kwargs):
         self.assertIs(observed["alerts"][0][1]["notification_settings"], module.RUNTIME_SETTINGS)
         self.assertIsNotNone(observed["alerts"][0][1]["state_settings"])
 
+    @patch.dict(os.environ, {"LONGBRIDGE_EXECUTION_STATE_CLOUD_URI": "gs://unit-test-execution/claims"})
     def test_run_strategy_force_runs_when_market_closed(self):
         module = load_module()
         observed = []
@@ -1055,6 +1072,7 @@ def build_rebalance_config(
 
         self.assertEqual(observed["notification_title_key"], "dry_run_title")
 
+    @patch.dict(os.environ, {"LONGBRIDGE_EXECUTION_STATE_CLOUD_URI": "gs://unit-test-execution/claims"})
     def test_run_strategy_persists_machine_readable_report(self):
         module = load_module()
         observed_reports = []

tests/test_runtime_composer.py

@@ -1,5 +1,6 @@
 import sys
 from pathlib import Path
+from dataclasses import replace
 from types import SimpleNamespace
 
 
@@ -157,3 +158,65 @@ def fake_cycle_sender(**kwargs):
     assert config.execution_state_account_scope == "HK"
     assert config.execution_state_store.cloud_prefix_uri == "gs://bucket/runtime-reports"
     assert config.account_identity_policy.is_configured is False
+
+
+def test_runtime_composer_parks_live_without_durable_execution_claim_backend():
+    class MinimalComposer(LongBridgeRuntimeComposer):
+        pass
+
+    fields = {
+        "project_id": "project-1",
+        "secret_name": "secret-1",
+        "token_refresh_threshold_days": 30,
+        "account_prefix": "HK",
+        "account_region": "HK",
+        "strategy_profile": "profile",
+        "strategy_display_name": "Profile",
+        "strategy_display_name_localized": "Profile",
+        "strategy_domain": "us_equity",
+        "notify_lang": "en",
+        "tg_token": None,
+        "tg_chat_id": None,
+        "managed_symbols": (),
+        "benchmark_symbol": "QQQ",
+        "signal_effective_after_trading_days": 1,
+        "separator": "-",
+        "limit_sell_discount": 0.99,
+        "limit_buy_premium": 1.01,
+        "order_poll_interval_sec": 1,
+        "order_poll_max_attempts": 1,
+        "safe_haven_cash_substitute_threshold_usd": 0.0,
+        "min_order_notional_usd": 1.0,
+        "dry_run_only": False,
+        "broker_adapters": SimpleNamespace(),
+        "strategy_adapters": SimpleNamespace(
+            translator=lambda key, **_kwargs: key,
+            build_strategy_plugin_notification_lines=lambda _signals: (),
+            build_strategy_plugin_error_notification_lines=lambda _error: (),
+        ),
+        "estimate_max_purchase_quantity_fn": lambda *_a, **_k: 0.0,
+        "fetch_order_status_fn": lambda *_a, **_k: None,
+        "fetch_token_from_secret_fn": lambda *_a, **_k: "",
+        "refresh_token_if_needed_fn": lambda *_a, **_k: "",
+        "build_contexts_fn": lambda *_a, **_k: (None, None),
+        "run_id_builder": lambda: "run",
+        "event_logger": lambda *_a, **_k: {},
+        "report_builder": lambda *_a, **_k: None,
+        "report_persister": lambda *_a, **_k: None,
+        "translator": lambda key, **_kwargs: key,
+        "env_reader": lambda _name, default="": default,
+        "sleeper": lambda _seconds: None,
+    }
+    composer = MinimalComposer(**fields)
+
+    try:
+        composer.build_rebalance_config()
+    except RuntimeError as exc:
+        assert "requires a gs:// execution state URI" in str(exc)
+    else:
+        raise AssertionError("live execution must fail closed without durable atomic claims")
+
+    dry = replace(composer, dry_run_only=True)
+    dry_config = dry.build_rebalance_config()
+    assert dry_config.dry_run_only is True
+    assert not str(dry_config.execution_state_store.cloud_prefix_uri or "").startswith("gs://")