Merge pull request #211 from QuantStrategyLab/codex/binance-r5-atomic-claim-20260905

fix: stop retries after uncertain earn submissions

Author: Pigbibi

infra/binance_runtime.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from runtime_support import ExecutionIntegrityError
+
 
 def resolve_runtime_btc_snapshot(
     runtime,
@@ -98,6 +100,8 @@ def ensure_asset_available_runtime(
                 if not runtime.dry_run:
                     sleep_fn(3)
                 return True
+    except ExecutionIntegrityError:
+        raise
     except Exception:
         runtime_notify_fn(
             runtime,
@@ -175,6 +179,8 @@ def manage_usdt_earn_buffer_runtime(
                         effect_type="earn_redeem",
                     )
                     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,

runtime_support.py

@@ -29,6 +29,14 @@
 _ORDER_SUBMISSION_UNKNOWN = "SUBMISSION_UNKNOWN"
 _ORDER_SUBMISSION_TERMINAL = "TERMINAL"
 _ORDER_CLIENT_ID_PREFIX = "QSL_"
+_EARN_METHOD_BY_EFFECT_TYPE = {
+    "earn_redeem": "redeem_simple_earn_flexible_product",
+    "earn_subscribe": "subscribe_simple_earn_flexible_product",
+}
+_EARN_SUCCESS_ID_BY_METHOD = {
+    "redeem_simple_earn_flexible_product": "redeemId",
+    "subscribe_simple_earn_flexible_product": "purchaseId",
+}
 _LAST_API_CALL_TS: float = 0.0
 RUNTIME_EVIDENCE_CONTRACT_VERSION = "qsl.runtime_evidence_aggregate.v1"
 RECONCILIATION_STATUSES = frozenset({"MISSING", "MATCHED", "MISMATCHED"})
@@ -610,11 +618,15 @@ def _load_order_submission_state(runtime):
         if set(record) != {"state"}:
             raise StatePersistenceError("submission_state_invalid") from None
     elif status == _ORDER_SUBMISSION_UNKNOWN:
-        if set(record) != {"state", "identity_sha256", "symbol"}:
-            raise StatePersistenceError("submission_state_invalid") from None
         if not _is_sha256(record.get("identity_sha256")):
             raise StatePersistenceError("submission_state_invalid") from None
-        if not re.fullmatch(r"[A-Z0-9]{3,30}", str(record.get("symbol") or "")):
+        if set(record) == {"state", "identity_sha256", "symbol"}:
+            if not re.fullmatch(r"[A-Z0-9]{3,30}", str(record.get("symbol") or "")):
+                raise StatePersistenceError("submission_state_invalid") from None
+        elif set(record) == {"state", "identity_sha256", "method_name"}:
+            if str(record.get("method_name") or "") not in _EARN_SUCCESS_ID_BY_METHOD:
+                raise StatePersistenceError("submission_state_invalid") from None
+        else:
             raise StatePersistenceError("submission_state_invalid") from None
     else:
         raise StatePersistenceError("submission_state_invalid") from None
@@ -665,6 +677,14 @@ def _is_order_transport_uncertainty(exc):
     )
 
 
+def _is_confirmed_earn_success(method_name, response):
+    response_id_field = _EARN_SUCCESS_ID_BY_METHOD.get(str(method_name))
+    if not response_id_field or not isinstance(response, Mapping) or response.get("success") is not True:
+        return False
+    response_id = response.get(response_id_field)
+    return isinstance(response_id, int) and not isinstance(response_id, bool) and response_id > 0
+
+
 def _reconcile_uncertain_order(client, symbol, identity_sha256):
     if not symbol or not identity_sha256:
         raise OrderReconciliationError("order_reconciliation_uncertain") from None
@@ -705,13 +725,20 @@ def runtime_call_client(runtime, report, *, method_name, payload, effect_type,
         raise RuntimeError("runtime.client is not configured")
 
     is_order_call = str(effect_type or "").startswith("order_")
+    earn_method = _EARN_METHOD_BY_EFFECT_TYPE.get(str(effect_type or ""))
+    is_earn_call = earn_method == str(method_name)
+    if str(effect_type or "").startswith("earn_") and not is_earn_call:
+        raise StatePersistenceError("submission_state_invalid") from None
+    is_funding_mutation = is_order_call or is_earn_call
     client_payload = dict(payload)
     trade_state = None
     identity_sha256 = None
-    if is_order_call:
+    if is_funding_mutation:
         trade_state, submission_record = _load_order_submission_state(runtime)
         submission_status = submission_record["state"]
         if submission_status == _ORDER_SUBMISSION_UNKNOWN:
+            if "method_name" in submission_record or is_earn_call:
+                raise OrderReconciliationError("order_reconciliation_uncertain") from None
             current_payload, current_identity_sha256 = _ensure_order_logical_identity(runtime, method_name, payload)
             if current_identity_sha256 != submission_record["identity_sha256"]:
                 raise OrderReconciliationError("order_reconciliation_intent_mismatch") from None
@@ -730,19 +757,64 @@ def runtime_call_client(runtime, report, *, method_name, payload, effect_type,
                 trade_state,
                 {"state": _ORDER_SUBMISSION_RESERVED},
             )
-        client_payload, identity_sha256 = _ensure_order_logical_identity(runtime, method_name, payload)
-        association = _build_order_request_association(method_name, client_payload)
-        symbol = association["symbol"]
-        _persist_order_submission_state(
-            runtime,
-            trade_state,
-            {
+        if is_order_call:
+            client_payload, identity_sha256 = _ensure_order_logical_identity(runtime, method_name, payload)
+            association = _build_order_request_association(method_name, client_payload)
+            symbol = association["symbol"]
+            unknown_record = {
                 "state": _ORDER_SUBMISSION_UNKNOWN,
                 "identity_sha256": identity_sha256,
                 "symbol": symbol,
-            },
+            }
+        else:
+            _unused_order_payload, identity_sha256 = _ensure_order_logical_identity(runtime, method_name, payload)
+            unknown_record = {
+                "state": _ORDER_SUBMISSION_UNKNOWN,
+                "identity_sha256": identity_sha256,
+                "method_name": str(method_name),
+            }
+        _persist_order_submission_state(
+            runtime,
+            trade_state,
+            unknown_record,
         )
         record_order_submission_attempt(report)
+    if is_earn_call:
+        _rate_limit_pause()
+        try:
+            response = getattr(runtime.client, method_name)(**client_payload)
+        except Exception as exc:
+            if _is_order_transport_uncertainty(exc):
+                record_order_transport_uncertainty(report)
+            record_side_effect(
+                runtime,
+                report,
+                effect_type=f"{effect_type}_failed",
+                target=method_name,
+                payload={"payload": dict(client_payload), "reason": "order_reconciliation_uncertain", "retries": 0},
+                executed=False,
+            )
+            raise OrderReconciliationError("order_reconciliation_uncertain") from None
+        if not _is_confirmed_earn_success(method_name, response):
+            record_side_effect(
+                runtime,
+                report,
+                effect_type=f"{effect_type}_failed",
+                target=method_name,
+                payload={"payload": dict(client_payload), "reason": "order_reconciliation_uncertain", "retries": 0},
+                executed=False,
+            )
+            raise OrderReconciliationError("order_reconciliation_uncertain") from None
+        record_side_effect(
+            runtime,
+            report,
+            effect_type=effect_type,
+            target=method_name,
+            payload=dict(client_payload),
+            executed=True,
+        )
+        _persist_order_submission_state(runtime, trade_state, {"state": _ORDER_SUBMISSION_TERMINAL})
+        return response
     _rate_limit_pause()
     retries_used = max_retries
     for attempt in range(max_retries + 1):

tests/test_binance_runtime_infra.py

@@ -8,6 +8,7 @@
     resolve_runtime_btc_snapshot,
     resolve_runtime_trend_indicators,
 )
+from runtime_support import ExecutionIntegrityError
 
 
 class BinanceRuntimeInfraTests(unittest.TestCase):
@@ -153,6 +154,51 @@ def get_simple_earn_flexible_product_list(self, *, asset):
         self.assertEqual(observed["calls"][0][0], "subscribe_simple_earn_flexible_product")
         self.assertEqual(len(observed["logs"]), 1)
 
+    def test_earn_integrity_errors_are_not_swallowed(self):
+        failure = ExecutionIntegrityError("order_reconciliation_uncertain")
+        observed = {"notifications": [], "logs": [], "sleeps": []}
+
+        class RedemptionClient:
+            def get_asset_balance(self, *, asset):
+                return {"free": "2.0"}
+
+            def get_simple_earn_flexible_product_position(self, *, asset):
+                return {"rows": [{"productId": "earn-1", "totalAmount": "5.0"}]}
+
+        with self.assertRaises(ExecutionIntegrityError):
+            ensure_asset_available_runtime(
+                SimpleNamespace(client=RedemptionClient(), dry_run=False),
+                {"redemption_subscription_intents": []},
+                "ETH",
+                3.0,
+                [],
+                runtime_call_client_fn=lambda *_args, **_kwargs: (_ for _ in ()).throw(failure),
+                append_log_fn=lambda _buffer, message: observed["logs"].append(message),
+                runtime_notify_fn=lambda _runtime, _report, text: observed["notifications"].append(text),
+                translate_fn=lambda key, **_kwargs: key,
+                sleep_fn=lambda seconds: observed["sleeps"].append(seconds),
+            )
+
+        class SubscriptionClient:
+            def get_asset_balance(self, *, asset):
+                return {"free": "150.0"}
+
+            def get_simple_earn_flexible_product_list(self, *, asset):
+                return {"rows": [{"productId": "earn-1"}]}
+
+        with self.assertRaises(ExecutionIntegrityError):
+            manage_usdt_earn_buffer_runtime(
+                SimpleNamespace(client=SubscriptionClient()),
+                {"redemption_subscription_intents": []},
+                100.0,
+                [],
+                runtime_call_client_fn=lambda *_args, **_kwargs: (_ for _ in ()).throw(failure),
+                append_log_fn=lambda _buffer, message: observed["logs"].append(message),
+                translate_fn=lambda key, **_kwargs: key,
+            )
+
+        self.assertEqual(observed, {"notifications": [], "logs": [], "sleeps": []})
+
     def test_ensure_runtime_client_marks_report_aborted_after_retries(self):
         runtime = SimpleNamespace(client=None, api_key="key", api_secret="secret")
         report = {"status": "ok"}

tests/test_cycle_service.py

@@ -9,6 +9,7 @@
 from application.execution_service import execute_trend_buys
 from infra.binance_runtime import ensure_runtime_client
 from runtime_support import (
+    ExecutionIntegrityError,
     ExecutionRuntime,
     append_report_error,
     build_execution_report,
@@ -18,11 +19,29 @@
 
 
 class CycleServiceTests(unittest.TestCase):
-    def _run_funds_cycle(self, execution_permitted, *, post_execution_permitted=None, snapshot_error=False, state=None):
+    def _run_funds_cycle(
+        self,
+        execution_permitted,
+        *,
+        post_execution_permitted=None,
+        snapshot_error=False,
+        state=None,
+        earn_failure=False,
+    ):
         events = []
         state = {} if state is None else state
         post_execution_permitted = execution_permitted if post_execution_permitted is None else post_execution_permitted
         allocation_permissions = [execution_permitted, post_execution_permitted]
+
+        def manage_earn(*_args, **_kwargs):
+            events.append("earn")
+            if earn_failure:
+                raise ExecutionIntegrityError("order_reconciliation_uncertain")
+
+        def send_periodic_status(*_args, **_kwargs):
+            if earn_failure:
+                events.append("periodic")
+
         runtime = SimpleNamespace(
             dry_run=False,
             now_utc=SimpleNamespace(strftime=lambda fmt: "20260905" if "%d" in fmt else "2026-09-05"),
@@ -82,8 +101,8 @@ def _run_funds_cycle(self, execution_permitted, *, post_execution_permitted=None
             run_daily_circuit_breaker=lambda *_args: events.append("circuit_breaker") or False,
             execute_trend_rotation=lambda *_args, **_kwargs: events.append("trend") or 185.0,
             execute_btc_dca_cycle=lambda *_args: events.append("dca") or 185.0,
-            manage_usdt_earn_buffer_runtime=lambda *_args, **_kwargs: events.append("earn"),
-            maybe_send_periodic_btc_status_report=lambda *_args, **_kwargs: None,
+            manage_usdt_earn_buffer_runtime=manage_earn,
+            maybe_send_periodic_btc_status_report=send_periodic_status,
             runtime_set_trade_state=lambda *_args, **_kwargs: events.append("state_write"),
             append_report_error=lambda *_args, **_kwargs: None,
             runtime_notify=lambda *_args, **_kwargs: None,
@@ -127,6 +146,14 @@ def test_post_trade_risk_veto_blocks_dca_and_earn_actions(self):
         self.assertNotIn("dca", events)
         self.assertNotIn("earn", events)
 
+    def test_earn_integrity_error_stops_final_state_write_and_later_cycle_actions(self):
+        report, events = self._run_funds_cycle(True, earn_failure=True)
+
+        self.assertEqual(report["status"], "error")
+        self.assertIn("earn", events)
+        self.assertNotIn("periodic", events)
+        self.assertNotIn("state_write", events)
+
     def test_research_cycle_settings_require_dry_run(self):
         runtime = SimpleNamespace(
             dry_run=False,