fix(execution): serialize live cycles with persistent state owner (#212)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

application/cycle_service.py

@@ -9,7 +9,10 @@
 from quant_platform_kit.strategy_lifecycle.performance_monitor import try_record_platform_execution
 from application.execution_receipt_adapter import attach_execution_receipt_from_report
 from runtime_logging import RuntimeLogContext, emit_runtime_log
-from runtime_support import append_report_error, finalize_notification_delivery
+from runtime_support import (
+    append_report_error, finalize_notification_delivery, acquire_runtime_state_owner,
+    release_runtime_state_owner, reconcile_runtime_cash_effects, ExecutionIntegrityError,
+)
 
 
 def execute_strategy_cycle(
@@ -58,7 +61,11 @@ def execute_strategy_cycle(
             "Runtime target disables standard execution; monitoring continues and all order/state-write calls are suppressed."
         )
 
+    state_healthy = False
     try:
+        if not acquire_runtime_state_owner(runtime):
+            report["execution_blocked_reason"] = "state_owner_busy"
+            return report
         if not ensure_runtime_client(runtime, report):
             return report
 
@@ -67,6 +74,9 @@ def execute_strategy_cycle(
             return report
 
         state, trend_pool_resolution, runtime_trend_universe, allow_new_trend_entries = cycle_state
+        runtime.trade_state = state
+        if state.get("order_submission", {}).get("state") == "SUBMISSION_UNKNOWN":
+            raise ExecutionIntegrityError("order_reconciliation_uncertain")
         append_trend_pool_source_logs(log_buffer, trend_pool_resolution, allow_new_trend_entries)
 
         report["upstream_pool_symbols"] = list(runtime_trend_universe.keys())
@@ -87,6 +97,7 @@ def execute_strategy_cycle(
         btc_snapshot = market_snapshot["btc_snapshot"]
         trend_indicators = market_snapshot["trend_indicators"]
 
+        state_healthy = True
         allocation = compute_portfolio_allocation(
             runtime,
             runtime_trend_universe,
@@ -155,6 +166,11 @@ def execute_strategy_cycle(
         )
         if fuel_status != "ready":
             report["execution_blocked_reason"] = f"bnb_fuel_{fuel_status}"
+            if fuel_status == "filled_pending_snapshot":
+                reconcile_runtime_cash_effects(runtime, state)
+                runtime_set_trade_state(runtime, report, state, reason="cash_reconciliation")
+            else:
+                state_healthy = False
             return report
 
         u_total = execute_trend_rotation(
@@ -243,16 +259,24 @@ def execute_strategy_cycle(
         )
 
         state["last_balance_snapshot"] = build_balance_snapshot(runtime_trend_universe, balances, u_total)
+        reconcile_runtime_cash_effects(runtime, state)
         runtime_set_trade_state(runtime, report, state, reason="cycle_complete")
 
     except Exception:
+        state_healthy = False
         report["status"] = "error"
         append_report_error(report, "cycle_execution_failed", stage="execute_cycle")
         try:
             runtime_notify(runtime, report, f"{translate_fn('system_crash')}\ncycle_execution_failed")
         except Exception:
             pass
     finally:
+        if state_healthy and getattr(runtime, "state_owner_held", False):
+            try:
+                release_runtime_state_owner(runtime)
+            except ExecutionIntegrityError:
+                report["status"] = "error"
+                append_report_error(report, "state_owner_release_uncertain", stage="state_release")
         report["log_lines"] = list(log_buffer)
         finalize_notification_delivery(report)
         attach_execution_receipt_from_report(report)

infra/binance_runtime.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from math import isfinite
+
 from runtime_support import ExecutionIntegrityError
 
 
@@ -92,23 +94,22 @@ def ensure_asset_available_runtime(
                     method_name="redeem_simple_earn_flexible_product",
                     payload={"productId": product_id, "amount": redeem_amt},
                     effect_type="earn_redeem",
+                    accounting_asset=asset,
                 )
                 append_log_fn(
                     log_buffer,
                     translate_fn("execution_spot_short_redeeming_from_earn", asset=asset, amount=redeem_amt),
                 )
                 if not runtime.dry_run:
                     sleep_fn(3)
+                    observed_free = float(runtime.client.get_asset_balance(asset=asset)["free"])
+                    if not isfinite(observed_free) or observed_free + 1e-8 < spot_free + redeem_amt:
+                        raise ExecutionIntegrityError("earn_reconciliation_pending")
                 return True
     except ExecutionIntegrityError:
         raise
     except Exception:
-        runtime_notify_fn(
-            runtime,
-            report,
-            f"{translate_fn('redeem_failed')} {asset}\n"
-            f"{translate_fn('error_label')}: asset_availability_failed",
-        )
+        raise ExecutionIntegrityError("asset_availability_failed") from None
     return False
 
 
@@ -153,7 +154,12 @@ def manage_usdt_earn_buffer_runtime(
                     method_name="subscribe_simple_earn_flexible_product",
                     payload={"productId": product_id, "amount": excess},
                     effect_type="earn_subscribe",
+                    accounting_asset=asset,
                 )
+                if not runtime.dry_run:
+                    observed_free = float(runtime.client.get_asset_balance(asset=asset)["free"])
+                    if not isfinite(observed_free) or observed_free < 0 or observed_free > spot_free - excess + 1e-8:
+                        raise ExecutionIntegrityError("earn_reconciliation_pending")
                 append_log_fn(log_buffer, translate_fn("cash_manager_subscribed_to_earn", amount=excess))
         elif spot_free < target_buffer - 5.0:
             shortfall = round(target_buffer - spot_free, 4)
@@ -177,15 +183,17 @@ def manage_usdt_earn_buffer_runtime(
                         method_name="redeem_simple_earn_flexible_product",
                         payload={"productId": product_id, "amount": redeem_amt},
                         effect_type="earn_redeem",
+                        accounting_asset=asset,
                     )
+                    if not runtime.dry_run:
+                        observed_free = float(runtime.client.get_asset_balance(asset=asset)["free"])
+                        if not isfinite(observed_free) or observed_free + 1e-8 < spot_free + redeem_amt:
+                            raise ExecutionIntegrityError("earn_reconciliation_pending")
                     append_log_fn(log_buffer, translate_fn("cash_manager_redeeming_to_spot", amount=redeem_amt))
     except ExecutionIntegrityError:
         raise
     except Exception:
-        append_log_fn(
-            log_buffer,
-            translate_fn("usdt_earn_buffer_maintenance_failed", error="earn_buffer_maintenance_failed"),
-        )
+        raise ExecutionIntegrityError("earn_buffer_maintenance_failed") from None
 
 
 def ensure_runtime_client(

live_services.py

@@ -26,9 +26,9 @@ def get_state_doc_ref(*, collection="strategy", document="MULTI_ASSET_STATE"):
     return get_firestore_client().collection(collection).document(document)
 
 
-def load_trade_state(*, normalize_fn, default_state_factory, normalize=True, collection="strategy", document="MULTI_ASSET_STATE"):
+def load_trade_state(*, normalize_fn, default_state_factory, normalize=True, collection="strategy", document="MULTI_ASSET_STATE", store=None):
     try:
-        payload = _get_document_store().get(collection=collection, document_id=document)
+        payload = (store if store is not None else _get_document_store()).get(collection=collection, document_id=document)
         if payload is not None:
             return normalize_fn(payload) if normalize else payload
         return default_state_factory() if normalize else {}
@@ -37,16 +37,60 @@ def load_trade_state(*, normalize_fn, default_state_factory, normalize=True, col
         return None
 
 
-def save_trade_state(data, *, normalize_fn, collection="strategy", document="MULTI_ASSET_STATE"):
+def save_trade_state(data, *, normalize_fn, collection="strategy", document="MULTI_ASSET_STATE", store=None):
     try:
         persisted_state = normalize_fn(data)
-        _get_document_store().set(collection=collection, document_id=document, data=persisted_state)
+        (store if store is not None else _get_document_store()).set(collection=collection, document_id=document, data=persisted_state)
         return True
     except Exception:
         print(t("firestore_write_failed", error="state_persistence_failed"))
         return False
 
 
+def bind_trade_state_access(*, normalize_fn, default_state_factory,
+                            collection="strategy", document="MULTI_ASSET_STATE"):
+    """Bind this runtime's state and persistent owner to the same Firestore backend."""
+    store = _get_document_store()
+
+    def load(normalize=True):
+        return load_trade_state(normalize_fn=normalize_fn, default_state_factory=default_state_factory,
+                                normalize=normalize, collection=collection, document=document, store=store)
+
+    def save(data):
+        return save_trade_state(data, normalize_fn=normalize_fn, collection=collection, document=document, store=store)
+
+    def owner_document():
+        return store.client.collection(collection).document(document + "__owner")
+
+    def claim(owner_id):
+        from google.api_core.exceptions import AlreadyExists
+        if not isinstance(owner_id, str) or not owner_id.strip():
+            raise ValueError("state_owner_required")
+        try:
+            owner_document().create({"owner_id": owner_id}, retry=None)
+        except AlreadyExists:
+            return False
+        return True
+
+    def release(owner_id):
+        from google.cloud import firestore
+        if not isinstance(owner_id, str) or not owner_id.strip():
+            raise ValueError("state_owner_required")
+        ref = owner_document()
+
+        @firestore.transactional
+        def delete_owned(transaction):
+            snapshot = ref.get(transaction=transaction, retry=None)
+            if not snapshot.exists or snapshot.to_dict().get("owner_id") != owner_id:
+                return False
+            transaction.delete(ref)
+            return True
+
+        return delete_owned(store.client.transaction(max_attempts=1))
+
+    return load, save, claim, release
+
+
 def send_tg_msg(token, chat_id, text):
     message = build_telegram_message(text)
     receipt = {

main.py

@@ -29,6 +29,7 @@
     manage_usdt_earn_buffer as qpk_manage_usdt_earn_buffer,
 )
 from live_services import (
+    bind_trade_state_access,
     get_firestore_client as live_get_firestore_client,
     get_state_doc_ref as live_get_state_doc_ref,
     send_tg_msg as live_send_tg_msg,
@@ -766,6 +767,12 @@ def build_live_runtime(now_utc=None):
         state_writer=set_trade_state,
         notifier=lambda **kwargs: send_tg_msg(kwargs["token"], kwargs["chat_id"], kwargs["text"]),
     )
+    if not runtime.dry_run and runtime.standard_execution_permitted:
+        runtime.state_loader, runtime.state_writer, runtime.state_owner_claim, runtime.state_owner_release = bind_trade_state_access(
+            normalize_fn=normalize_trade_state, default_state_factory=build_default_state,
+        )
+        runtime.fuel_symbol = BNB_FUEL_SYMBOL
+        runtime.fuel_asset = BNB_FUEL_ASSET
     _activate_execution_strategy_runtime(runtime.strategy_profile)
     return runtime