From 726aa5612afad901ad98bd7e5da8d28ad1d5a0bb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 8 Sep 2026 04:37:20 +0800 Subject: [PATCH 1/2] fix(risk): require gs:// account-owner claim store for live Fail closed when non-dry-run would use a local claim backend that cannot fence across instances. Co-Authored-By: Claude Co-authored-by: Cursor --- application/runtime_composer.py | 13 ++++--- tests/test_runtime_composer.py | 63 +++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 4 deletions(-) diff --git a/application/runtime_composer.py b/application/runtime_composer.py index 29caa9d..ea45d2b 100644 --- a/application/runtime_composer.py +++ b/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 diff --git a/tests/test_runtime_composer.py b/tests/test_runtime_composer.py index dc59c05..4b4cc3f 100644 --- a/tests/test_runtime_composer.py +++ b/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://") From a5912774a4cd6118bfd1432dcd003a672907b024 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 8 Sep 2026 05:16:36 +0800 Subject: [PATCH 2/2] test(runtime): align request fixtures with durable claim gate Co-Authored-By: Codex --- tests/test_request_handling.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 68ae63b..1a9ae92 100644 --- a/tests/test_request_handling.py +++ b/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 = []