dita_v2+adapters: phantom-position root-cause fixes (UV audit §8)

bingx_direct.submit_intent: (1) empty/orderId-less ack (VST offline-symbol
hole: code==0 + empty data) -> REJECTED, never fabricate fill from
target_size; (2) honor upstream u- client id (intent.metadata
promo_client_id/client_order_id) over self-minted p- id.
rust_backend: EXIT asset-mismatch guard both sync+async paths — EXIT for
asset X while slot holds Y -> NO_OPEN_POSITION/EXIT_ASSET_MISMATCH instead
of asset-blind close (THETA 76bbf8ee cross-slot contamination).
9 regression tests, mutation-RED verified (guard flip).

prod/clean_arch/adapters/bingx_direct.py

@@ -594,7 +594,13 @@ class BingxDirectExecutionAdapter(ExecutionPort):
         _action_char = "e" if intent.action == DecisionAction.ENTER else "x"
         _ts36 = self._base36(int(time.time() * 1000))
         _rand4 = uuid.uuid4().hex[:4]
-        client_order_id = f"p-{_action_char}-{_ts36}-{_rand4}"
+        # UV-FIX 2026-07-11: honor an upstream-minted client id (the u- prefix
+        # from the UV seam) when the intent carries one — venue orders then join
+        # back to exec_journal rows by id, and the sentinel u- check holds.
+        # Same charset/length rules apply (metadata id is trimmed to 40).
+        _meta = intent.metadata or {}
+        _meta_cid = str(_meta.get("client_order_id") or _meta.get("promo_client_id") or "")
+        client_order_id = _meta_cid[:40] if _meta_cid else f"p-{_action_char}-{_ts36}-{_rand4}"
         # DUAL-LEVERAGE TRANSLATION (prod/bingx/leverage.py, SYSTEM BIBLE §6):
         # intent.leverage is the STRATEGY conviction (fractional, 0.5–9.0) and
         # already sized the quantity. At-exchange leverage is derived from it
@@ -653,6 +659,34 @@ class BingxDirectExecutionAdapter(ExecutionPort):
             )
             ack = BingxOrderAck.from_http(ack_payload if isinstance(ack_payload, dict) else {})
             ack_row = dict(unwrap_order_payload(ack_payload)) if isinstance(ack_payload, dict) else {}
+            # UV-FIX 2026-07-11: an accepted order ALWAYS carries the venue's
+            # orderId. VST answers an order on an offline/delisted symbol with
+            # code==0 + EMPTY data (observed BAND/CELR 2026-07-10/11) — the
+            # unwrapped ack is {} and, before this guard, fell through as
+            # "ACKED" while fill_qty below fabricated a full fill from
+            # intent.target_size. The kernel FSM then opened a position that
+            # never existed at the venue (root of the phantom-position pairs).
+            # (code!=0 envelopes raise BingxHttpError inside signed_post and
+            # take the except-branch below; this guard covers the code==0
+            # empty-ack hole, plus any raw envelope leak, belt-and-braces.)
+            _body_code = int(ack_payload.get("code") or 0) if isinstance(ack_payload, dict) else 0
+            _order_handle = (
+                ack_row.get("orderId") or ack_row.get("orderID") or ack_row.get("order_id")
+                if isinstance(ack_row, dict) else None
+            )
+            if _body_code != 0 or not _order_handle:
+                LOGGER.warning(
+                    "order POST venue-dead ack symbol=%s code=%s orderId=%r msg=%.160s"
+                    " — REJECTED (no fabricated fill)",
+                    symbol, _body_code, _order_handle,
+                    str((ack_payload or {}).get("msg") or "") if isinstance(ack_payload, dict) else "",
+                )
+                ack_row = {
+                    "status": "REJECTED",
+                    "msg": "venue returned no orderId (symbol offline/delisted?)",
+                    "symbol": symbol,
+                    "clientOrderId": client_order_id,
+                }
             status = str(ack_row.get("status") or ack.status or "ACKED")
             LOGGER.debug("order ACK: status=%s orderId=%s executedQty=%s side=%s",
                          status, ack_row.get("orderId"), ack_row.get("executedQty"), ack_row.get("side"))
@@ -665,7 +699,7 @@ class BingxDirectExecutionAdapter(ExecutionPort):
                 if value > 0:
                     fill_price = value
                     break
-            if fill_price <= 0 and self._state is not None:
+            if fill_price <= 0 and status != "REJECTED" and self._state is not None:
                 fill_price = next(
                     (float(row.get("markPrice") or row.get("avgPrice") or 0.0)
                      for row in self._state.open_positions.values()
@@ -688,8 +722,14 @@ class BingxDirectExecutionAdapter(ExecutionPort):
         # BingX REST ACK does not include commission. WS FILL_SETTLED will deliver
         # the actual fee later and update the fee_source to "WS_SETTLED".
         # Until then, log an estimate so CH rows are never blank on this field.
-        fill_qty = float(ack_row.get("executedQty") or ack_row.get("filledQty") or
+        # UV-FIX 2026-07-11: never fabricate a fill for a rejected order — the
-                         getattr(intent, "target_size", 0.0) or 0.0)
+        # target_size fallback applies only to accepted MARKET acks whose fill
+        # arrives later via WS (BingX REST ack omits executedQty on those).
+        if status in ("REJECTED", "RATE_LIMITED"):
+            fill_qty = 0.0
+        else:
+            fill_qty = float(ack_row.get("executedQty") or ack_row.get("filledQty") or
+                             getattr(intent, "target_size", 0.0) or 0.0)
         if is_limit:
             # LIMIT orders *may* rest and fill as maker — optimistic estimate.
             fee_rate = 0.0002  # BingX perpetuals maker fee 0.02%

prod/clean_arch/dita_v2/rust_backend.py

@@ -761,6 +761,36 @@ class ExecutionKernel:
                     control=self.control,
                 )
 
+    def _exit_asset_mismatch_outcome(self, intent: KernelIntent) -> Optional[KernelOutcome]:
+        """UV-FIX 2026-07-11: an EXIT must name the asset its slot holds.
+
+        All UV promotion intents share slot 0. During the 2026-07-10/11
+        phantom-pair incident, an EXIT for asset X arriving while the slot
+        held asset Y was accepted asset-blind — closing Y's venue position
+        and orphaning Y's own later exit (NO_OPEN_POSITION). Reject the
+        mismatch at the kernel boundary instead.
+        """
+        if intent.action != KernelCommandType.EXIT:
+            return None
+        slot = self._get_slot(int(intent.slot_id))
+        open_asset = str(getattr(slot, "asset", "") or "")
+        if slot.is_open() and open_asset and open_asset != intent.asset:
+            return KernelOutcome(
+                accepted=False,
+                slot_id=int(intent.slot_id),
+                trade_id=intent.trade_id,
+                state=slot.fsm_state,
+                diagnostic_code=KernelDiagnosticCode.NO_OPEN_POSITION,
+                severity=KernelSeverity.WARNING,
+                details={
+                    "reason": "EXIT_ASSET_MISMATCH",
+                    "slot_asset": open_asset,
+                    "intent_asset": intent.asset,
+                    "intent_id": intent.intent_id,
+                },
+            )
+        return None
+
     def process_intent(self, intent: KernelIntent) -> KernelOutcome:
         self.zinc_plane.publish_intent(intent)
         if not (0 <= int(intent.slot_id) < self.max_slots):
@@ -795,6 +825,9 @@ class ExecutionKernel:
                     "asset": intent.asset,
                 },
             )
+        _mismatch = self._exit_asset_mismatch_outcome(intent)
+        if _mismatch is not None:
+            return _mismatch
         payload = _intent_to_payload(intent)
         result = _get_rust().process_intent(
             self._backend,
@@ -943,6 +976,9 @@ class ExecutionKernel:
                 details={"reason": "INVALID_INTENT", "field": name, "value": str(value),
                          "intent_id": intent.intent_id, "action": intent.action.value, "asset": intent.asset},
             )
+        _mismatch = self._exit_asset_mismatch_outcome(intent)
+        if _mismatch is not None:
+            return _mismatch
         # ── Rust FSM (sync, atomic, μs-fast — no await here) ─────────────────
         payload = _intent_to_payload(intent)
         result = _get_rust().process_intent(

prod/clean_arch/dita_v2/test_uv_seam_fixes.py

@@ -0,0 +1,179 @@
+"""Regression tests for the 2026-07-11 UV phantom-position fixes.
+
+Covers (root causes from UV_TESTNET_TRADE_AUDIT_20260710.md §8):
+  - kernel: EXIT for asset X while the slot holds asset Y is rejected
+    (EXIT_ASSET_MISMATCH -> NO_OPEN_POSITION), not executed asset-blind
+  - kernel: a matching-asset EXIT still closes normally (no over-block)
+  - bingx_direct: HTTP-200 + {"code": nonzero} business-reject -> status
+    REJECTED with NO fabricated fill (fill_qty must be 0)
+  - bingx_direct: metadata client id (u- prefix) is propagated to the venue
+    order payload instead of the self-minted p- id
+
+Mutation litmus: inverting the asset comparison in
+_exit_asset_mismatch_outcome, or dropping the business-reject guard in
+submit_intent, goes RED here.
+
+Run:
+    python -m pytest test_uv_seam_fixes.py -v
+"""
+from __future__ import annotations
+
+import asyncio
+import sys
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+import pytest
+
+from prod.clean_arch.dita_v2.contracts import (
+    KernelCommandType,
+    KernelDiagnosticCode,
+    KernelIntent,
+    TradeSide,
+)
+from prod.clean_arch.dita_v2.mock_venue import MockVenueAdapter, MockVenueScenario
+from prod.clean_arch.dita_v2.rust_backend import ExecutionKernel
+
+
+def _intent(action=KernelCommandType.ENTER, trade_id="t1", asset="TRX-USDT",
+            size=10.0, metadata=None) -> KernelIntent:
+    return KernelIntent(
+        timestamp=datetime.now(timezone.utc),
+        intent_id=trade_id,
+        trade_id=trade_id,
+        slot_id=0,
+        asset=asset,
+        action=action,
+        side=TradeSide.SHORT,
+        reason="test",
+        target_size=size,
+        leverage=1.0,
+        reference_price=100.0,
+        exit_leg_ratios=(1.0,),
+        metadata=metadata or {},
+    )
+
+
+# ── kernel: EXIT asset-mismatch guard ─────────────────────────────────────────
+
+class TestExitAssetMismatch:
+    def _open_position(self, kernel):
+        out = kernel.process_intent(_intent(action=KernelCommandType.ENTER,
+                                            trade_id="t1", asset="TRX-USDT"))
+        assert out.accepted, f"test setup: ENTER should open, got {out.diagnostic_code}"
+        return out
+
+    def test_exit_for_other_asset_is_rejected(self):
+        kernel = ExecutionKernel(max_slots=1, venue=MockVenueAdapter(scenario=MockVenueScenario()))
+        self._open_position(kernel)
+        out = kernel.process_intent(_intent(action=KernelCommandType.EXIT,
+                                            trade_id="t2", asset="BAND-USDT"))
+        assert out.accepted is False
+        assert out.diagnostic_code is KernelDiagnosticCode.NO_OPEN_POSITION
+        assert out.details.get("reason") == "EXIT_ASSET_MISMATCH"
+        assert out.details.get("slot_asset") == "TRX-USDT"
+        assert out.details.get("intent_asset") == "BAND-USDT"
+
+    def test_exit_for_matching_asset_still_closes(self):
+        kernel = ExecutionKernel(max_slots=1, venue=MockVenueAdapter(scenario=MockVenueScenario()))
+        self._open_position(kernel)
+        out = kernel.process_intent(_intent(action=KernelCommandType.EXIT,
+                                            trade_id="t1", asset="TRX-USDT"))
+        assert out.accepted is True, f"matching-asset EXIT must pass, got {out.diagnostic_code}"
+
+    def test_mismatch_guard_untouched_when_slot_idle(self):
+        """EXIT on an empty slot keeps the kernel's own NO_OPEN_POSITION path."""
+        kernel = ExecutionKernel(max_slots=1, venue=MockVenueAdapter(scenario=MockVenueScenario()))
+        out = kernel.process_intent(_intent(action=KernelCommandType.EXIT,
+                                            trade_id="t9", asset="BAND-USDT"))
+        assert out.accepted is False  # idle slot: rejected by FSM, not by the guard
+        assert out.details.get("reason") != "EXIT_ASSET_MISMATCH"
+
+    def test_async_path_has_same_guard(self):
+        kernel = ExecutionKernel(max_slots=1, venue=MockVenueAdapter(scenario=MockVenueScenario()))
+        self._open_position(kernel)
+        out = asyncio.run(kernel.process_intent_async(
+            _intent(action=KernelCommandType.EXIT, trade_id="t3", asset="XLM-USDT")))
+        assert out.accepted is False
+        assert out.details.get("reason") == "EXIT_ASSET_MISMATCH"
+
+
+# ── bingx_direct: business-reject + client id propagation ────────────────────
+
+def _make_adapter(post_response):
+    from prod.clean_arch.adapters import bingx_direct as bd
+    adapter = bd.BingxDirectExecutionAdapter.__new__(bd.BingxDirectExecutionAdapter)
+    adapter._client = MagicMock()
+    adapter._client.signed_post = AsyncMock(return_value=post_response)
+    adapter._state = None
+    adapter._config = MagicMock()
+    adapter._config.default_leverage = 1
+    adapter._config.exchange_leverage_cap = 3
+    adapter._ensure_leverage = AsyncMock(return_value=None)
+    adapter._instrument_venue_symbol = lambda a: f"{a[:-4]}-USDT"
+    adapter._format_quantity = lambda a, q: str(q)
+    adapter._format_price = lambda a, p: str(p)
+    adapter._s2_tasks = {}
+    adapter._refresh_state_background = AsyncMock(return_value=None)
+    adapter._refresh_exchange_state = AsyncMock(return_value=None)
+    return adapter, bd
+
+
+def _intent_for_adapter(metadata=None):
+    from prod.clean_arch.dita import Intent, TradeSide as TS, DecisionAction
+    i = MagicMock(spec=Intent)
+    i.asset = "BANDUSDT"
+    i.action = DecisionAction.ENTER
+    i.side = TS.SHORT
+    i.target_size = 2919.77
+    i.leverage = 1.0
+    i.metadata = metadata or {}
+    return i
+
+
+# signed_post returns the UNWRAPPED envelope `data` (code!=0 raises
+# BingxHttpError inside the client). VST's offline-symbol hole: code==0
+# with EMPTY data — observed BAND/CELR 2026-07-10/11.
+
+def test_empty_ack_yields_rejected_receipt_with_zero_fill():
+    adapter, bd = _make_adapter(None)  # code==0, data null → unwrapped None
+    receipt = asyncio.run(adapter.submit_intent(_intent_for_adapter()))
+    assert receipt.status == "REJECTED"
+    assert float(receipt.raw_ack.get("executedQty") or 0.0) == 0.0, \
+        "a venue-rejected order must never carry a fabricated fill"
+    assert receipt.price == 0.0
+
+
+def test_ack_without_order_id_is_rejected():
+    adapter, bd = _make_adapter({"order": {"status": "NEW"}})  # no orderId
+    receipt = asyncio.run(adapter.submit_intent(_intent_for_adapter()))
+    assert receipt.status == "REJECTED"
+
+
+def test_accepted_order_keeps_ack_semantics():
+    adapter, bd = _make_adapter(
+        {"order": {"orderId": "O77", "status": "FILLED",
+                   "executedQty": "2919.77", "avgPrice": "0.17"}})
+    receipt = asyncio.run(adapter.submit_intent(_intent_for_adapter()))
+    assert receipt.status == "FILLED"
+    assert receipt.price == pytest.approx(0.17)
+    assert float(receipt.raw_ack.get("executedQty")) == pytest.approx(2919.77)
+
+
+def test_metadata_client_id_propagates_to_order_payload():
+    adapter, bd = _make_adapter({"order": {"orderId": "O78", "status": "NEW"}})
+    uid = "u-86ada259_uv_promo_a7d4ddf"
+    receipt = asyncio.run(adapter.submit_intent(
+        _intent_for_adapter(metadata={"promo_client_id": uid})))
+    payload = adapter._client.signed_post.call_args[0][1]
+    assert payload["clientOrderId"] == uid
+    assert receipt.client_order_id == uid
+
+
+def test_without_metadata_falls_back_to_p_mint():
+    adapter, bd = _make_adapter({"order": {"orderId": "O79", "status": "NEW"}})
+    asyncio.run(adapter.submit_intent(_intent_for_adapter()))
+    payload = adapter._client.signed_post.call_args[0][1]
+    assert payload["clientOrderId"].startswith("p-e-")