dita_v2(exec): INDETERMINATE submit is not REJECTED — unknown is never flat

A BingX read-timeout/reset/5xx after send means the answer was lost, not that
the order failed. Classify every submit failure by what it PROVES:
NOT_ATTEMPTED / REFUSED -> rollback sound; INDETERMINATE -> point-lookup our
own clientOrderId (read-only, bounded, never a reconcile); unresolved stays
UNKNOWN — no synthetic REJECT, no slot rollback, E-feed FILL settles truth.

- prod/bingx/http.py: BingxHttpError.effect + order_may_exist, 9 raise sites tagged
- adapters/bingx_direct.py: _lookup_own_order_by_client_id (never POSTs)
- dita_v2/venue.py: VenueIndeterminateError(VenuePostAckError) — existing fences catch it
- dita_v2/bingx_venue.py: both submit paths escalate INDETERMINATE receipts
- 14 tests incl. kernel no-rollback invariant + genuine-REFUSED contrast

Suite: 3416 passed, 19 skipped, 3 xfailed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

prod/clean_arch/dita_v2/bingx_venue.py

@@ -45,6 +45,7 @@ from .utils import json_safe
 from .utils import safe_float
 from .venue import VenueAdapter
 from .venue import VenuePostAckError
+from .venue import VenueIndeterminateError
 
 
 def _row_text(row: dict[str, Any], *keys: str, default: str = "") -> str:
@@ -853,6 +854,14 @@ class BingxVenueAdapter(VenueAdapter):
         submitted = replace(intent, target_size=float(legacy.target_size))
         # ── PRE-ACK ── safe to roll back if this raises.
         receipt = self._call_backend("submit_intent", legacy)
+        # ── UNKNOWN OUTCOME ── see submit_async(): the order may be live, so the
+        # kernel must NOT roll back to flat on it.
+        if str(getattr(receipt, "status", "") or "").upper() == "INDETERMINATE":
+            raise VenueIndeterminateError(
+                f"submit outcome UNKNOWN for intent={intent.intent_id} asset={intent.asset} "
+                f"— order may be live at the venue; refusing to claim rejection",
+                receipt=receipt,
+            )
         # ── POINT OF NO RETURN ── order is LIVE; see submit_async().
         try:
             events = self._events_from_submit(submitted, receipt, None, None)
@@ -920,6 +929,17 @@ class BingxVenueAdapter(VenueAdapter):
         # ── PRE-ACK ── an exception here means the venue never took the order.
         # Safe for the caller to roll the FSM back.
         receipt = await self.backend.submit_intent(legacy)
+        # ── UNKNOWN OUTCOME ── the submit timed out / 5xx'd and the adapter could
+        # not establish the truth even after asking the venue about our own
+        # clientOrderId. The order MAY be live. Escalate as indeterminate so the
+        # kernel does NOT roll the slot back to flat (VenueIndeterminateError is a
+        # VenuePostAckError, so the existing no-rollback fences catch it).
+        if str(getattr(receipt, "status", "") or "").upper() == "INDETERMINATE":
+            raise VenueIndeterminateError(
+                f"submit outcome UNKNOWN for intent={intent.intent_id} asset={intent.asset} "
+                f"— order may be live at the venue; refusing to claim rejection",
+                receipt=receipt,
+            )
         # ── POINT OF NO RETURN ── the order is LIVE at the venue from here on.
         # Nothing below may reach the caller as a bare exception: that channel
         # means "no order exists", and the caller rolls back to flat on it.

prod/clean_arch/dita_v2/test_indeterminate_submit.py

@@ -0,0 +1,282 @@
+"""INDETERMINATE submit outcomes: "not done" != "tried and failed" != "unknown".
+
+A BingX read timeout / reset / 5xx means the request WAS SENT and the answer was
+lost — the order may be live. Before this class of fix, `bingx_direct` reported such
+a failure to the kernel as a confident REJECTED with fill_qty=0, and the kernel rolled
+the slot back to flat while the venue kept the position. That is an orphan, and it is
+the same disease as the 2026-07-13 post-ack skew, one layer deeper.
+
+The rule these tests protect:
+
+    Rollback is sound ONLY when the effect is provably absent.
+      NOT_ATTEMPTED (never sent)      -> rollback OK
+      REFUSED       (venue said no)   -> rollback OK
+      INDETERMINATE (answer lost)     -> NEVER roll back. Ask the venue about our own
+                                         clientOrderId; if truth cannot be established,
+                                         stay UNKNOWN and let the E-feed FILL settle it.
+
+The contrast tests matter as much as the recovery tests: a fix that refuses to roll
+back on a *genuine* rejection would strand every slot forever.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import itertools
+import threading
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from prod.bingx.http import (
+    BingxHttpError,
+    _effect_of_httpx_error,
+    _effect_of_status,
+)
+from prod.clean_arch.adapters.bingx_direct import BingxDirectExecutionAdapter
+from prod.clean_arch.dita_v2.bingx_venue import BingxVenueAdapter
+from prod.clean_arch.dita_v2.contracts import (
+    KernelCommandType,
+    KernelEventKind,
+    KernelIntent,
+    TradeSide,
+    TradeStage,
+)
+from prod.clean_arch.dita_v2.rust_backend import ExecutionKernel
+from prod.clean_arch.dita_v2.venue import VenueIndeterminateError, VenuePostAckError
+
+
+# ── 1. Classification: what does the failure actually PROVE? ─────────────────
+
+
+def test_read_timeout_is_indeterminate_not_rejection():
+    """The killer case: the request was sent, the answer was lost."""
+    assert _effect_of_httpx_error(httpx.ReadTimeout("timed out")) == BingxHttpError.INDETERMINATE
+
+
+def test_connect_failures_are_not_attempted():
+    """Never left the box -> provably no order -> rollback is sound."""
+    assert _effect_of_httpx_error(httpx.ConnectError("refused")) == BingxHttpError.NOT_ATTEMPTED
+    assert _effect_of_httpx_error(httpx.ConnectTimeout("no route")) == BingxHttpError.NOT_ATTEMPTED
+
+
+def test_5xx_is_indeterminate_and_4xx_is_refused():
+    assert _effect_of_status(500) == BingxHttpError.INDETERMINATE
+    assert _effect_of_status(503) == BingxHttpError.INDETERMINATE
+    assert _effect_of_status(400) == BingxHttpError.REFUSED
+    assert _effect_of_status(429) == BingxHttpError.REFUSED
+
+
+def test_unknown_failure_defaults_to_indeterminate():
+    """When we do not know, we must NOT claim absence."""
+    assert BingxHttpError("mystery").effect == BingxHttpError.INDETERMINATE
+    assert BingxHttpError("mystery").order_may_exist is True
+    assert BingxHttpError("nope", effect=BingxHttpError.REFUSED).order_may_exist is False
+
+
+# ── 2. The point lookup: one order, by our own key, read-only ────────────────
+
+
+class _LookupClient:
+    """Records every call so we can prove the lookup NEVER re-POSTs an order."""
+
+    def __init__(self, get_results):
+        self._get_results = list(get_results)
+        self.gets: list[tuple[str, dict]] = []
+        self.posts: list[tuple[str, dict]] = []
+
+    async def signed_get(self, path, params=None):
+        self.gets.append((path, dict(params or {})))
+        result = self._get_results.pop(0)
+        if isinstance(result, Exception):
+            raise result
+        return result
+
+    async def signed_post(self, path, params=None, **_kw):  # pragma: no cover - must never run
+        self.posts.append((path, dict(params or {})))
+        raise AssertionError("lookup must NEVER submit an order")
+
+
+def _adapter(client) -> BingxDirectExecutionAdapter:
+    adapter = BingxDirectExecutionAdapter.__new__(BingxDirectExecutionAdapter)
+    adapter._client = client
+    return adapter
+
+
+def _order_row(**over):
+    """Shape as signed_get actually delivers it: the envelope is already unwrapped
+    (BingxHttpClient._unwrap_response returns payload["data"]), so a query order
+    lookup yields {"order": {...}}."""
+    row = {"orderId": "venue-1", "clientOrderId": "cid-1", "status": "FILLED",
+           "avgPrice": "100.0", "executedQty": "10.0"}
+    row.update(over)
+    return {"order": row}
+
+
+def test_lookup_finds_the_order_and_never_posts():
+    client = _LookupClient([_order_row()])
+    row = asyncio.run(_adapter(client)._lookup_own_order_by_client_id("TRX-USDT", "cid-1"))
+    assert isinstance(row, dict) and row["orderId"] == "venue-1"
+    assert client.posts == [], "lookup must never re-submit — that is how you double-fill"
+    assert client.gets[0][1]["clientOrderID"] == "cid-1", "must query by OUR idempotency key"
+
+
+def test_lookup_reports_absent_when_venue_has_no_such_order():
+    """Venue answered authoritatively: no order. ONLY now is rollback sound."""
+    client = _LookupClient([{"code": 0, "data": {}}])
+    assert asyncio.run(
+        _adapter(client)._lookup_own_order_by_client_id("TRX-USDT", "cid-1")
+    ) == "ABSENT"
+
+
+def test_lookup_treats_venue_refusal_as_absent():
+    client = _LookupClient([BingxHttpError("order not exist", effect=BingxHttpError.REFUSED)])
+    assert asyncio.run(
+        _adapter(client)._lookup_own_order_by_client_id("TRX-USDT", "cid-1")
+    ) == "ABSENT"
+
+
+def test_lookup_returns_none_when_truth_cannot_be_established():
+    """It is allowed to say "I don't know". It is never allowed to guess."""
+    client = _LookupClient([httpx.ReadTimeout("t")] * 3)
+    result = asyncio.run(
+        _adapter(client)._lookup_own_order_by_client_id("TRX-USDT", "cid-1", attempts=3)
+    )
+    assert result is None, "unresolved must be None — never 'ABSENT', never a fabricated row"
+    assert len(client.gets) == 3, "bounded: exactly `attempts` tries, then give up"
+    assert client.posts == []
+
+
+# ── 3. The venue layer escalates UNKNOWN instead of claiming rejection ───────
+
+
+def _intent(trade_id="ind-1") -> KernelIntent:
+    return KernelIntent(
+        timestamp=datetime.now(timezone.utc),
+        intent_id=f"intent-{trade_id}", trade_id=trade_id, slot_id=0,
+        asset="TRX-USDT", action=KernelCommandType.ENTER, side=TradeSide.SHORT,
+        reason="indeterminate-regression", target_size=10.0, leverage=1.0,
+        reference_price=100.0, exit_leg_ratios=(1.0,), metadata={},
+    )
+
+
+def _receipt(status):
+    return SimpleNamespace(
+        status=status, order_id="", client_order_id="cid-1", price=0.0, quantity=0.0,
+        timestamp=datetime.now(timezone.utc),
+        raw_ack={"status": status, "clientOrderId": "cid-1"},
+    )
+
+
+class _Backend:
+    def __init__(self, status, is_async=False):
+        self._status = status
+        self._is_async = is_async
+        self.submit_count = 0
+
+    def submit_intent(self, _legacy):
+        self.submit_count += 1
+        return _receipt(self._status)
+
+    async def submit_intent_async(self, _legacy):  # pragma: no cover - shim
+        return self.submit_intent(_legacy)
+
+
+class _AsyncBackend(_Backend):
+    async def submit_intent(self, _legacy):  # type: ignore[override]
+        self.submit_count += 1
+        return _receipt(self._status)
+
+
+def _venue(backend) -> BingxVenueAdapter:
+    venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
+    venue.backend = backend
+    venue._event_seq = itertools.count(1)
+    venue._snap_lock = threading.Lock()
+    venue._snapshot_ready = threading.Event()
+    venue._snapshot_ready.set()
+    venue._last_snapshot = None
+    venue._telemetry_plane = None
+    return venue
+
+
+def test_sync_venue_escalates_indeterminate_receipt(monkeypatch):
+    venue = _venue(_Backend("INDETERMINATE"))
+    monkeypatch.setattr(venue, "_publish_telemetry", lambda **_f: None)
+    with pytest.raises(VenueIndeterminateError):
+        venue.submit(_intent())
+
+
+def test_async_venue_escalates_indeterminate_receipt(monkeypatch):
+    venue = _venue(_AsyncBackend("INDETERMINATE"))
+    monkeypatch.setattr(venue, "_publish_telemetry", lambda **_f: None)
+    with pytest.raises(VenueIndeterminateError):
+        asyncio.run(venue.submit_async(_intent()))
+
+
+def test_indeterminate_is_a_post_ack_error_so_existing_fences_catch_it():
+    """Subclassing is load-bearing: the kernel's no-rollback fences key on the base."""
+    assert issubclass(VenueIndeterminateError, VenuePostAckError)
+
+
+# ── 4. THE INVARIANT: the kernel must never conclude "flat" on UNKNOWN ───────
+
+
+def test_kernel_does_not_roll_back_slot_on_indeterminate_submit(monkeypatch):
+    """kernel_believes_flat => venue_is_flat. An UNKNOWN order may be live, so the
+    slot must NOT return to IDLE and NO synthetic REJECT may be emitted."""
+    venue = _venue(_Backend("INDETERMINATE"))
+    monkeypatch.setattr(venue, "_publish_telemetry", lambda **_f: None)
+
+    with ExecutionKernel(max_slots=1, venue=venue) as kernel:
+        outcome = kernel.process_intent(_intent(trade_id="ind-sync"))
+        slot = kernel._get_slot(0)
+
+        kinds = [e.kind for e in outcome.emitted_events]
+        assert KernelEventKind.ORDER_REJECT not in kinds, (
+            "a synthetic REJECT on an UNKNOWN order is the orphan-maker"
+        )
+        assert slot.fsm_state is not TradeStage.IDLE, (
+            "slot rolled back to IDLE while the order may be LIVE at the venue"
+        )
+
+
+def test_kernel_does_not_roll_back_slot_on_indeterminate_submit_async(monkeypatch):
+    venue = _venue(_AsyncBackend("INDETERMINATE"))
+    monkeypatch.setattr(venue, "_publish_telemetry", lambda **_f: None)
+
+    async def exercise():
+        with ExecutionKernel(max_slots=1, venue=venue) as kernel:
+            outcome = await kernel.process_intent_async(_intent(trade_id="ind-async"))
+            return outcome, kernel._get_slot(0)
+
+    outcome, slot = asyncio.run(exercise())
+    kinds = [e.kind for e in outcome.emitted_events]
+    assert KernelEventKind.ORDER_REJECT not in kinds
+    assert slot.fsm_state is not TradeStage.IDLE
+
+
+# ── 5. CONTRAST: a genuine rejection MUST still roll back ────────────────────
+# Over-correcting here would strand every slot in ORDER_REQUESTED forever, which is
+# its own outage. Proving the safe path still works is part of the fix.
+
+
+def test_genuine_rejection_still_rolls_the_slot_back(monkeypatch):
+    class _RejectingBackend:
+        def submit_intent(self, _legacy):
+            raise BingxHttpError("insufficient margin", effect=BingxHttpError.REFUSED)
+
+    venue = _venue(_RejectingBackend())
+    monkeypatch.setattr(venue, "_publish_telemetry", lambda **_f: None)
+
+    with ExecutionKernel(max_slots=1, venue=venue) as kernel:
+        outcome = kernel.process_intent(_intent(trade_id="refused"))
+        slot = kernel._get_slot(0)
+
+        kinds = [e.kind for e in outcome.emitted_events]
+        assert KernelEventKind.ORDER_REJECT in kinds, (
+            "a PROVEN refusal must still produce a REJECT — otherwise the slot strands"
+        )
+        assert slot.fsm_state is TradeStage.IDLE, "provably-no-order must free the slot"

prod/clean_arch/dita_v2/venue.py

@@ -40,6 +40,23 @@ class VenuePostAckError(Exception):
         self.events = events or []
 
 
+class VenueIndeterminateError(VenuePostAckError):
+    """The submit outcome is UNKNOWN: the request was sent, the answer was lost.
+
+    A read timeout / connection reset / 5xx means BingX may have matched the order.
+    The adapter has already asked the venue about our clientOrderId and could not
+    establish the truth, so we are left with genuine uncertainty.
+
+    Deliberately a subclass of VenuePostAckError, because it demands the SAME
+    response: the effect may exist, therefore DO NOT roll the slot back to flat and
+    DO NOT synthesise a REJECT. Leave the slot working and let the E-feed FILL /
+    the account stream settle it. Callers that already fence VenuePostAckError get
+    this behaviour for free.
+
+    Unknown is not flat. It is not failed either. It is unknown.
+    """
+
+
 class VenueAdapter(Protocol):
     """Abstract venue adapter used by the kernel."""