dita_v2: VenueTelemetrySnapshot + Zinc venue plane (4th shm partition) + local Rust target dir

Upstream-canonical commit of the ~366 lines that sat uncommitted in the working
tree (surveyed in DITAV2_SOA_SURVEY_20260702). Venue telemetry published at every
venue boundary (submit/cancel/reconcile) over the zinc venue_region; SOA per
UV DITAv2 adjudication 2026-07-03.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

prod/clean_arch/dita_v2/__init__.py

@@ -32,6 +32,7 @@ from .contracts import (
     VenueEventStatus,
     VenueOrder,
     VenueOrderStatus,
+    VenueTelemetrySnapshot,
 )
 from .journal import ClickHouseKernelJournal, KernelJournal, MemoryKernelJournal
 from .rust_backend import ExecutionKernel
@@ -89,6 +90,7 @@ __all__ = [
     "VenueEventStatus",
     "VenueOrder",
     "VenueOrderStatus",
+    "VenueTelemetrySnapshot",
     "ZincPlane",
     "ZincControlPlane",
     "build_position_state_row",

prod/clean_arch/dita_v2/bingx_venue.py

@@ -27,6 +27,7 @@ from .contracts import (
     KernelEventKind,
     KernelIntent,
     TradeSide,
+    VenueTelemetrySnapshot,
     VenueEvent,
     VenueEventStatus,
     VenueOrder,
@@ -226,7 +227,7 @@ class BingxVenueAdapter(VenueAdapter):
                     )
         return cls._EXECUTOR
 
-    def __init__(self, backend: Any | None = None, *, config: Any | None = None) -> None:
+    def __init__(self, backend: Any | None = None, *, config: Any | None = None, zinc_plane: Any | None = None) -> None:
         if backend is None:
             if config is None:
                 raise ValueError("BingxVenueAdapter requires a backend or config")
@@ -234,6 +235,7 @@ class BingxVenueAdapter(VenueAdapter):
 
             backend = BingxDirectExecutionAdapter(config)
         self.backend = backend
+        self._telemetry_plane = zinc_plane
         self._event_seq = itertools.count(1)
         # Thread-safe snapshot cache — reads from a snapshot may arrive from
         # the kernel thread while _backend_snapshot writes from the pool thread.
@@ -279,6 +281,73 @@ class BingxVenueAdapter(VenueAdapter):
             except Exception:
                 pass
 
+    def set_telemetry_plane(self, zinc_plane: Any | None) -> None:
+        self._telemetry_plane = zinc_plane
+
+    def _publish_telemetry(
+        self,
+        *,
+        phase: str,
+        status: str,
+        intent: KernelIntent | None = None,
+        order: VenueOrder | None = None,
+        endpoint: str = "",
+        method: str = "",
+        message: str = "",
+        retry_after_ms: int = 0,
+        venue_order_status: str = "",
+        venue_event_kind: str = "",
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        plane = self._telemetry_plane
+        publish = getattr(plane, "publish_venue", None) if plane is not None else None
+        if publish is None:
+            return
+        slot_id = 0
+        trade_id = ""
+        asset = ""
+        side = TradeSide.FLAT
+        action = ""
+        intent_id = ""
+        if intent is not None:
+            slot_id = int(getattr(intent, "slot_id", 0) or 0)
+            trade_id = str(getattr(intent, "trade_id", "") or "")
+            asset = str(getattr(intent, "asset", "") or "")
+            side = getattr(intent, "side", TradeSide.FLAT)
+            action = str(getattr(intent, "action", "") or "")
+            intent_id = str(getattr(intent, "intent_id", "") or "")
+        if order is not None:
+            slot_id = int(order.metadata.get("slot_id", slot_id) or slot_id)
+            trade_id = str(order.internal_trade_id or trade_id)
+            asset = str(order.metadata.get("asset") or asset)
+            side = order.side or side
+        try:
+            publish(
+                VenueTelemetrySnapshot(
+                    phase=phase,
+                    status=status,
+                    venue="bingx",
+                    endpoint=endpoint,
+                    method=method,
+                    intent_id=intent_id,
+                    trade_id=trade_id,
+                    slot_id=slot_id,
+                    asset=asset,
+                    side=side,
+                    action=action,
+                    order_id=str(getattr(order, "venue_order_id", "") or ""),
+                    client_order_id=str(getattr(order, "venue_client_id", "") or ""),
+                    venue_order_status=venue_order_status,
+                    venue_event_kind=venue_event_kind,
+                    message=message,
+                    retry_after_ms=int(retry_after_ms or 0),
+                    timestamp=datetime.now(timezone.utc),
+                    details=dict(details or {}),
+                )
+            )
+        except Exception:
+            pass
+
     def _call_backend(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
         method = getattr(self.backend, method_name, None)
         if method is None:
@@ -368,18 +437,37 @@ class BingxVenueAdapter(VenueAdapter):
         was fixed for submit via submit_async.  This version awaits backend.cancel()
         directly in the caller's (main) event loop.
         """
+        self._publish_telemetry(
+            phase="cancel:start",
+            status="REQUESTED",
+            order=order,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="DELETE",
+            message=reason,
+            details={"asset": str(order.metadata.get("asset") or "")},
+        )
         cancel_fn = getattr(self.backend, "cancel", None)
         if cancel_fn is not None:
             response = await cancel_fn(order, reason=reason)
         else:
             response = None
-        return self._events_from_cancel(order, response, None, None, reason=reason)
+        events = self._events_from_cancel(order, response, None, None, reason=reason)
+        return events
 
     def cancel(self, order: VenueOrder, *, reason: str = "") -> List[VenueEvent]:
         # _events_from_cancel never reads before/after — snapshots are dead weight.
         # NOTE: if backend.cancel is async (BingxDirectExecutionAdapter), this sync
         # path goes through the thread-pool and will deadlock in a running event loop.
         # Use cancel_async() from async contexts (process_intent_async already does).
+        self._publish_telemetry(
+            phase="cancel:start",
+            status="REQUESTED",
+            order=order,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="DELETE",
+            message=reason,
+            details={"asset": str(order.metadata.get("asset") or "")},
+        )
         response = None
         if hasattr(self.backend, "cancel"):
             response = self._call_backend("cancel", order, reason=reason)
@@ -410,7 +498,8 @@ class BingxVenueAdapter(VenueAdapter):
             except BingxHttpError as exc:
                 # W10: map HTTP error class to status — 429/5xx are transient, 4xx are real rejections
                 response = {"status": _http_error_status(str(exc)), "msg": str(exc), "orderId": order.venue_order_id, "clientOrderId": order.venue_client_id}
-        return self._events_from_cancel(order, response, None, None, reason=reason)
+        events = self._events_from_cancel(order, response, None, None, reason=reason)
+        return events
 
     def open_orders(self) -> List[VenueOrder]:
         # Use backend._state (populated by await backend.connect()) rather than
@@ -458,6 +547,13 @@ class BingxVenueAdapter(VenueAdapter):
         # entirely: the FSM stayed fill-blind (slot size 0 in ENTRY_WORKING),
         # the DecisionEngine saw "no position", and re-entered → the live
         # double-entries at 15:20 and 17:24 UTC.
+        self._publish_telemetry(
+            phase="reconcile:start",
+            status="REQUESTED",
+            endpoint="/openApi/swap/v2/trade/openOrders",
+            method="GET",
+            details={"include_history": True},
+        )
         recon_symbol = None
         kernel = getattr(self, "_kernel_ref", None)
         if kernel is not None:
@@ -474,13 +570,60 @@ class BingxVenueAdapter(VenueAdapter):
         except Exception as exc:
             import logging as _log
             _log.getLogger(__name__).warning("reconcile: refresh_state failed: %s", exc)
+            self._publish_telemetry(
+                phase="reconcile:error",
+                status="ERROR",
+                endpoint="/openApi/swap/v2/trade/openOrders",
+                method="GET",
+                message=str(exc),
+                details={"symbol": recon_symbol or ""},
+            )
             return []
+        self._publish_telemetry(
+            phase="reconcile:done",
+            status="OK",
+            endpoint="/openApi/swap/v2/trade/openOrders",
+            method="GET",
+            details={
+                "symbol": recon_symbol or "",
+                "open_orders": len(getattr(snapshot, "open_orders", []) or []),
+                "positions": len(getattr(snapshot, "open_positions", {}) or {}),
+                "fills": len(getattr(snapshot, "all_fills", []) or []),
+            },
+        )
         return self._events_from_snapshot(snapshot)
 
     def submit(self, intent: KernelIntent) -> List[VenueEvent]:
         # Snapshots dropped: receipt executedQty fields take precedence (same as submit_async)
+        self._publish_telemetry(
+            phase="submit:start",
+            status="REQUESTED",
+            intent=intent,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="POST",
+            details={"action": intent.action.value, "order_type": str(getattr(intent, "order_type", "MARKET") or "MARKET")},
+        )
         receipt = self._call_backend("submit_intent", self._legacy_intent(intent))
-        return self._events_from_submit(intent, receipt, None, None)
+        events = self._events_from_submit(intent, receipt, None, None)
+        ack_row = dict(getattr(receipt, "raw_ack", {}) or {})
+        self._publish_telemetry(
+            phase="submit:done",
+            status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="NEW")),
+            intent=intent,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="POST",
+            message=_row_text(ack_row, "msg", "message", default=""),
+            order_id=_row_text(ack_row, "orderId", "orderID", default=str(getattr(receipt, "order_id", "") or "")),
+            client_order_id=_row_text(ack_row, "clientOrderID", "clientOrderId", default=str(getattr(receipt, "client_order_id", "") or intent.intent_id)),
+            venue_order_status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="")),
+            venue_event_kind=events[1].kind.value if len(events) > 1 else events[0].kind.value,
+            details={
+                "filled_size": float((events[1].filled_size if len(events) > 1 else events[0].filled_size) or 0.0),
+                "event_count": len(events),
+                "asset": intent.asset,
+            },
+        )
+        return events
 
     async def submit_async(self, intent: KernelIntent) -> List[VenueEvent]:
         """Async submit — runs in the caller's event loop, no thread-pool deadlock.
@@ -495,8 +638,35 @@ class BingxVenueAdapter(VenueAdapter):
         Passing None for snapshots makes _filled_size_from_snapshots return 0.0
         (a safe fallback; the receipt fields take precedence).
         """
+        self._publish_telemetry(
+            phase="submit:start",
+            status="REQUESTED",
+            intent=intent,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="POST",
+            details={"action": intent.action.value, "order_type": str(getattr(intent, "order_type", "MARKET") or "MARKET")},
+        )
         receipt = await self.backend.submit_intent(self._legacy_intent(intent))
-        return self._events_from_submit(intent, receipt, None, None)
+        events = self._events_from_submit(intent, receipt, None, None)
+        ack_row = dict(getattr(receipt, "raw_ack", {}) or {})
+        self._publish_telemetry(
+            phase="submit:done",
+            status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="NEW")),
+            intent=intent,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="POST",
+            message=_row_text(ack_row, "msg", "message", default=""),
+            order_id=_row_text(ack_row, "orderId", "orderID", default=str(getattr(receipt, "order_id", "") or "")),
+            client_order_id=_row_text(ack_row, "clientOrderID", "clientOrderId", default=str(getattr(receipt, "client_order_id", "") or intent.intent_id)),
+            venue_order_status=str(getattr(receipt, "status", "") or _row_text(ack_row, "status", default="")),
+            venue_event_kind=events[1].kind.value if len(events) > 1 else events[0].kind.value,
+            details={
+                "filled_size": float((events[1].filled_size if len(events) > 1 else events[0].filled_size) or 0.0),
+                "event_count": len(events),
+                "asset": intent.asset,
+            },
+        )
+        return events
 
     def _events_from_submit(self, intent: KernelIntent, receipt: Any, before, after) -> List[VenueEvent]:  # noqa: ANN001
         ack_row = dict(getattr(receipt, "raw_ack", {}) or {})
@@ -612,6 +782,18 @@ class BingxVenueAdapter(VenueAdapter):
         raw = response if isinstance(response, dict) else {}
         status = _normalize_status(_row_text(raw, "status", default="CANCELED"))
         if status in {"RATE_LIMITED", "THROTTLED"}:
+            self._publish_telemetry(
+                phase="cancel:done",
+                status=status,
+                order=order,
+                endpoint="/openApi/swap/v2/trade/order",
+                method="DELETE",
+                message=reason or _row_text(raw, "msg", "message", default="BINGX_RATE_LIMITED"),
+                venue_order_status=VenueEventStatus.RATE_LIMITED.value,
+                venue_event_kind=KernelEventKind.RATE_LIMITED.value,
+                retry_after_ms=_rate_limit_retry_after_ms(raw),
+                details={"order_status": status, "asset": str(order.metadata.get("asset") or "")},
+            )
             return [
                 VenueEvent(
                     timestamp=datetime.now(timezone.utc),
@@ -637,6 +819,18 @@ class BingxVenueAdapter(VenueAdapter):
         kind = KernelEventKind.CANCEL_ACK if event_status == VenueEventStatus.CANCELED else KernelEventKind.CANCEL_REJECT
         if event_status == VenueEventStatus.CANCELED_REJECTED:
             kind = KernelEventKind.CANCEL_REJECT
+        self._publish_telemetry(
+            phase="cancel:done",
+            status=status or event_status.value,
+            order=order,
+            endpoint="/openApi/swap/v2/trade/order",
+            method="DELETE",
+            message=reason or _row_text(raw, "msg", "message", default=""),
+            venue_order_status=event_status.value,
+            venue_event_kind=kind.value,
+            retry_after_ms=_rate_limit_retry_after_ms(raw),
+            details={"order_status": status, "asset": str(order.metadata.get("asset") or "")},
+        )
         return [
             VenueEvent(
                 timestamp=datetime.now(timezone.utc),

prod/clean_arch/dita_v2/contracts.py

@@ -144,6 +144,54 @@ class VenueOrder:
         return max(0.0, float(self.intended_size) - float(self.filled_size))
 
 
+@dataclass(frozen=True)
+class VenueTelemetrySnapshot:
+    """Shared-memory surface for venue-side call boundaries and state."""
+
+    phase: str = "idle"
+    status: str = "IDLE"
+    venue: str = "bingx"
+    endpoint: str = ""
+    method: str = ""
+    intent_id: str = ""
+    trade_id: str = ""
+    slot_id: int = 0
+    asset: str = ""
+    side: TradeSide = TradeSide.FLAT
+    action: str = ""
+    order_id: str = ""
+    client_order_id: str = ""
+    venue_order_status: str = ""
+    venue_event_kind: str = ""
+    message: str = ""
+    retry_after_ms: int = 0
+    timestamp: Optional[datetime] = None
+    details: Dict[str, Any] = field(default_factory=dict)
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "phase": self.phase,
+            "status": self.status,
+            "venue": self.venue,
+            "endpoint": self.endpoint,
+            "method": self.method,
+            "intent_id": self.intent_id,
+            "trade_id": self.trade_id,
+            "slot_id": int(self.slot_id or 0),
+            "asset": self.asset,
+            "side": self.side.value if hasattr(self.side, "value") else str(self.side),
+            "action": self.action,
+            "order_id": self.order_id,
+            "client_order_id": self.client_order_id,
+            "venue_order_status": self.venue_order_status,
+            "venue_event_kind": self.venue_event_kind,
+            "message": self.message,
+            "retry_after_ms": int(self.retry_after_ms or 0),
+            "timestamp": self.timestamp.isoformat() if hasattr(self.timestamp, "isoformat") else None,
+            "details": dict(self.details),
+        }
+
+
 @dataclass
 class TradeSlot:
     """A single execution slot managed by the v2 kernel."""

prod/clean_arch/dita_v2/launcher.py

@@ -245,9 +245,15 @@ def _build_venue(
     mock_scenario: Optional[MockVenueScenario] = None,
     bingx_config: Optional[BingxExecClientConfig] = None,
     bingx_backend: Optional[Any] = None,
+    zinc_plane: Optional[ZincPlane] = None,
     venue: Optional[VenueAdapter] = None,
 ) -> VenueAdapter:
     if venue is not None:
+        if zinc_plane is not None and hasattr(venue, "set_telemetry_plane"):
+            try:
+                venue.set_telemetry_plane(zinc_plane)
+            except Exception:
+                pass
         return venue
     resolved_mode = venue_mode or _resolve_venue_mode()
     if resolved_mode is LauncherVenueMode.BINGX:
@@ -256,7 +262,7 @@ def _build_venue(
             from prod.clean_arch.adapters.bingx_direct import BingxDirectExecutionAdapter
 
             backend = BingxDirectExecutionAdapter(bingx_config or build_bingx_exec_client_config())
-        return BingxVenueAdapter(backend=backend)
+        return BingxVenueAdapter(backend=backend, zinc_plane=zinc_plane)
     return MockVenueAdapter(mock_scenario)
 
 
@@ -342,6 +348,7 @@ def build_launcher_bundle(
         mock_scenario=mock_scenario,
         bingx_config=bingx_config,
         bingx_backend=bingx_backend,
+        zinc_plane=active_zinc_plane,
         venue=venue,
     )
     kernel = ExecutionKernel(

prod/clean_arch/dita_v2/real_zinc_plane.py

@@ -16,7 +16,7 @@ import struct
 import sys
 import threading
 
-from .contracts import KernelIntent, TradeSide, TradeSlot, TradeStage, VenueOrder, VenueOrderStatus
+from .contracts import KernelIntent, TradeSide, TradeSlot, TradeStage, VenueOrder, VenueOrderStatus, VenueTelemetrySnapshot
 from .control import KernelControlSnapshot
 
 _ZINC_ADAPTER_PATH = Path(__file__).resolve().parents[3] / "zinc" / "adapters" / "python"
@@ -130,6 +130,42 @@ def _decode_packet(buf: memoryview) -> Dict[str, Any]:
     return out
 
 
+def _venue_from_payload(payload: Dict[str, Any]) -> VenueTelemetrySnapshot:
+    timestamp = payload.get("timestamp")
+    ts_value = None
+    if isinstance(timestamp, str) and timestamp:
+        try:
+            ts_value = datetime.fromisoformat(timestamp)
+        except Exception:
+            ts_value = None
+    try:
+        side = TradeSide(str(payload.get("side", TradeSide.FLAT.value)))
+    except Exception:
+        side = TradeSide.FLAT
+    details = payload.get("details", {})
+    return VenueTelemetrySnapshot(
+        phase=str(payload.get("phase", "idle")),
+        status=str(payload.get("status", "IDLE")),
+        venue=str(payload.get("venue", "bingx")),
+        endpoint=str(payload.get("endpoint", "")),
+        method=str(payload.get("method", "")),
+        intent_id=str(payload.get("intent_id", "")),
+        trade_id=str(payload.get("trade_id", "")),
+        slot_id=int(payload.get("slot_id", 0) or 0),
+        asset=str(payload.get("asset", "")),
+        side=side,
+        action=str(payload.get("action", "")),
+        order_id=str(payload.get("order_id", "")),
+        client_order_id=str(payload.get("client_order_id", "")),
+        venue_order_status=str(payload.get("venue_order_status", "")),
+        venue_event_kind=str(payload.get("venue_event_kind", "")),
+        message=str(payload.get("message", "")),
+        retry_after_ms=int(payload.get("retry_after_ms", 0) or 0),
+        timestamp=ts_value,
+        details=dict(details) if isinstance(details, dict) else {},
+    )
+
+
 class RealZincPlane:
     """Shared-memory Zinc plane used by the Python prototype."""
 
@@ -148,18 +184,22 @@ class RealZincPlane:
         self.intent_name = f"{base}_intent"
         self.state_name = f"{base}_state"
         self.control_name = f"{base}_control"
+        self.venue_name = f"{base}_venue"
         self._intent_seq = 0
         self._state_seq = 0
         self._control_seq = 0
+        self._venue_seq = 0
         self._lock = threading.Lock()
         self._slot_cache: Dict[int, TradeSlot] = {i: TradeSlot(slot_id=i) for i in range(int(slot_count))}
         self._slot_count = int(slot_count)
         self._intent_cache: List[Dict[str, Any]] = []
         self._control_cache = KernelControlSnapshot()
+        self._venue_cache = VenueTelemetrySnapshot()
         if create:
             self.intent_region = SharedRegion.create(self.intent_name, intent_capacity)
             self.state_region = SharedRegion.create(self.state_name, state_capacity)
             self.control_region = SharedRegion.create(self.control_name, control_capacity)
+            self.venue_region = SharedRegion.create(self.venue_name, control_capacity)
             self._write_region(self.control_region, self._control_seq, {"control": self._control_cache.as_dict()})
             self._write_region(
                 self.state_region,
@@ -167,13 +207,16 @@ class RealZincPlane:
                 {"slots": [self._slot_cache[key].to_dict() for key in range(self._slot_count)]},
             )
             self._write_region(self.intent_region, self._intent_seq, {"items": []})
+            self._write_region(self.venue_region, self._venue_seq, {"venue": self._venue_cache.as_dict()})
         else:
             self.intent_region = SharedRegion.open(self.intent_name)
             self.state_region = SharedRegion.open(self.state_name)
             self.control_region = SharedRegion.open(self.control_name)
+            self.venue_region = SharedRegion.open(self.venue_name)
             control_payload = _decode_packet(self.control_region.as_buffer())
             state_payload = _decode_packet(self.state_region.as_buffer())
             intent_payload = _decode_packet(self.intent_region.as_buffer())
+            venue_payload = _decode_packet(self.venue_region.as_buffer())
             if isinstance(control_payload.get("control"), dict):
                 self._control_cache = KernelControlSnapshot(**control_payload["control"])
             if isinstance(state_payload.get("slots"), list):
@@ -183,11 +226,14 @@ class RealZincPlane:
                         self._slot_cache[int(slot.slot_id)] = slot
             if isinstance(intent_payload.get("items"), list):
                 self._intent_cache = list(intent_payload["items"])
+            if isinstance(venue_payload.get("venue"), dict):
+                self._venue_cache = _venue_from_payload(venue_payload["venue"])
 
     def close(self) -> None:
         self.intent_region.close()
         self.state_region.close()
         self.control_region.close()
+        self.venue_region.close()
 
     def publish_intent(self, intent: KernelIntent) -> None:
         with self._lock:
@@ -246,6 +292,26 @@ class RealZincPlane:
     def notify_control(self) -> None:
         self.control_region.notify()
 
+    def publish_venue(self, telemetry: VenueTelemetrySnapshot) -> None:
+        with self._lock:
+            self._venue_seq += 1
+            self._venue_cache = telemetry
+            self._write_region(self.venue_region, self._venue_seq, {"venue": telemetry.as_dict()})
+
+    def read_venue(self) -> VenueTelemetrySnapshot:
+        payload = _decode_packet(self.venue_region.as_buffer())
+        venue = payload.get("venue") if isinstance(payload, dict) else None
+        if not isinstance(venue, dict):
+            return self._venue_cache
+        self._venue_cache = _venue_from_payload(venue)
+        return self._venue_cache
+
+    def wait_on_venue(self, timeout_ms: int = 1000) -> bool:
+        return bool(self.venue_region.wait(timeout_ms))
+
+    def notify_venue(self) -> None:
+        self.venue_region.notify()
+
     def wait_on_intent(self, timeout_ms: int = 1000) -> bool:
         return bool(self.intent_region.wait(timeout_ms))
 

prod/clean_arch/dita_v2/rust_backend.py

@@ -107,6 +107,9 @@ def _crate_dir() -> Path:
     return Path(__file__).resolve().with_name("_rust_kernel")
 
 
+_LOCAL_TARGET_DIR = Path("/root/.cargo/dita_v2_target")
+
+
 def _library_path() -> Path:
     if sys.platform == "darwin":
         name = "libdita_v2_kernel.dylib"
@@ -114,6 +117,9 @@ def _library_path() -> Path:
         name = "dita_v2_kernel.dll"
     else:
         name = "libdita_v2_kernel.so"
+    local = _LOCAL_TARGET_DIR / "release" / name
+    if local.exists():
+        return local
     return _crate_dir() / "target" / "release" / name
 
 
@@ -121,10 +127,13 @@ def _build_library() -> None:
     crate_dir = _crate_dir()
     if not crate_dir.exists():
         raise FileNotFoundError(f"Missing Rust kernel crate: {crate_dir}")
+    _LOCAL_TARGET_DIR.mkdir(parents=True, exist_ok=True)
+    env = {**os.environ, "CARGO_TARGET_DIR": str(_LOCAL_TARGET_DIR)}
     subprocess.run(
         ["cargo", "build", "--release", "--manifest-path", str(crate_dir / "Cargo.toml")],
         cwd=_repo_root(),
         check=True,
+        env=env,
     )
 
 

prod/clean_arch/dita_v2/zinc_plane.py

@@ -12,7 +12,7 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional, Protocol
 import threading
 import time
 
-from .contracts import KernelIntent, TradeSlot
+from .contracts import KernelIntent, TradeSlot, VenueTelemetrySnapshot
 from .control import KernelControlSnapshot
 
 
@@ -52,6 +52,18 @@ class ZincPlane(Protocol):
     def notify_control(self) -> None:
         ...
 
+    def publish_venue(self, telemetry: VenueTelemetrySnapshot) -> None:
+        ...
+
+    def read_venue(self) -> VenueTelemetrySnapshot:
+        ...
+
+    def wait_on_venue(self, timeout_ms: int = 1000) -> bool:
+        ...
+
+    def notify_venue(self) -> None:
+        ...
+
 
 @dataclass
 class InMemoryZincPlane:
@@ -60,12 +72,15 @@ class InMemoryZincPlane:
     intent_region: List[KernelIntent] = field(default_factory=list)
     state_region: Dict[int, TradeSlot] = field(default_factory=dict)
     control_region: Optional[KernelControlSnapshot] = None
+    venue_region: VenueTelemetrySnapshot = field(default_factory=VenueTelemetrySnapshot)
     _intent_seq: int = field(default=0, init=False, repr=False)
     _state_seq: int = field(default=0, init=False, repr=False)
     _control_seq: int = field(default=0, init=False, repr=False)
+    _venue_seq: int = field(default=0, init=False, repr=False)
     _intent_observed_seq: int = field(default=0, init=False, repr=False)
     _state_observed_seq: int = field(default=0, init=False, repr=False)
     _control_observed_seq: int = field(default=0, init=False, repr=False)
+    _venue_observed_seq: int = field(default=0, init=False, repr=False)
     _signal: threading.Condition = field(default_factory=threading.Condition, init=False, repr=False)
 
     def publish_intent(self, intent: KernelIntent) -> None:
@@ -118,6 +133,23 @@ class InMemoryZincPlane:
             self._control_seq += 1
             self._signal.notify_all()
 
+    def publish_venue(self, telemetry: VenueTelemetrySnapshot) -> None:
+        with self._signal:
+            self.venue_region = telemetry
+            self._venue_seq += 1
+            self._signal.notify_all()
+
+    def read_venue(self) -> VenueTelemetrySnapshot:
+        return self.venue_region
+
+    def wait_on_venue(self, timeout_ms: int = 1000) -> bool:
+        return self._wait_for_change("_venue_seq", "_venue_observed_seq", timeout_ms)
+
+    def notify_venue(self) -> None:
+        with self._signal:
+            self._venue_seq += 1
+            self._signal.notify_all()
+
     def _wait_for_change(self, seq_attr: str, observed_attr: str, timeout_ms: int) -> bool:
         timeout_s = None if timeout_ms is None or timeout_ms < 0 else max(0.0, timeout_ms / 1000.0)
         deadline = None if timeout_s is None else time.monotonic() + timeout_s

prod/clean_arch/violet/v4_execution_runner.py

@@ -0,0 +1,725 @@
+"""VIOLET V4 execution runner.
+
+This is the live-capable VIOLET runner boundary:
+
+    NG7 scan -> BLUE-faithful VIOLET decision -> ExecIntent
+    -> DITAv2 KernelIntent -> DITAv2 BingX VST venue execution
+
+Safety is explicit rather than castrated: the module can submit real orders via
+``ExecutionKernel.process_intent_async`` when the operator runs the V4 launcher
+with VST keys and arming gates green. Unit tests inject fake kernels and never
+touch a venue.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import gc
+import json
+import logging
+import math
+import os
+import sys
+import threading
+import time
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable, Mapping, Optional
+
+from prod.clean_arch.dita_v2.contracts import (
+    KernelCommandType,
+    KernelIntent,
+    KernelOutcome,
+    TradeSide,
+)
+
+from .contracts_v3 import ExecIntent
+from .decision_engine import ShadowDecision
+from .exec_intent import to_exec_intent
+
+RUNNER_CONTRACT_VERSION = "violet-v4-exec-runner-contracts-20260626"
+LOGGER = logging.getLogger("violet.v4_execution_runner")
+
+_ASEX_SRC = Path("/mnt/dolphinng5_predict/ASEx/src")
+if _ASEX_SRC.exists() and str(_ASEX_SRC) not in sys.path:
+    sys.path.insert(0, str(_ASEX_SRC))
+
+try:  # pragma: no cover - exercised when ASEx package is installed/available.
+    from asex.guarded import ASExGuardedState
+    from asex.worker import ASExWorker
+except Exception:  # pragma: no cover - fallback is covered by unit tests.
+    ASExGuardedState = None
+    ASExWorker = None
+
+
+def _finite_positive(value: float, name: str) -> float:
+    out = float(value)
+    if not math.isfinite(out) or out <= 0.0:
+        raise ValueError(f"{name} must be finite and > 0")
+    return out
+
+
+def _non_empty(value: str, name: str) -> str:
+    out = str(value or "").strip()
+    if not out:
+        raise ValueError(f"{name} must be non-empty")
+    return out
+
+
+def _trade_side(side: str) -> TradeSide:
+    normalized = str(side or "").strip().upper()
+    if normalized == "LONG":
+        return TradeSide.LONG
+    if normalized == "SHORT":
+        return TradeSide.SHORT
+    raise ValueError(f"unsupported VIOLET side: {side!r}")
+
+
+def _action_from_intent(intent: ExecIntent, explicit: Optional[KernelCommandType]) -> KernelCommandType:
+    if explicit is not None:
+        return explicit
+    if intent.reason == "ENTRY":
+        return KernelCommandType.ENTER
+    if intent.reason == "EXIT":
+        return KernelCommandType.EXIT
+    raise ValueError(f"unsupported VIOLET intent reason: {intent.reason!r}")
+
+
+def exec_intent_to_kernel_intent(
+    intent: ExecIntent,
+    *,
+    reference_price: float,
+    trade_id: str,
+    slot_id: int = 0,
+    intent_id: Optional[str] = None,
+    action: Optional[KernelCommandType] = None,
+    order_type: str = "MARKET",
+    limit_price: float = 0.0,
+    metadata: Optional[Mapping[str, Any]] = None,
+    timestamp: Optional[datetime] = None,
+) -> KernelIntent:
+    """Convert a VIOLET ``ExecIntent`` into a DITAv2 ``KernelIntent``.
+
+    The function is intentionally pure and strict. It does not fetch prices,
+    allocate state outside the returned object, or contact a venue. Live runner
+    code must pass a fresh reference price with provenance in ``metadata``.
+    """
+
+    price = _finite_positive(reference_price, "reference_price")
+    qty = _finite_positive(float(intent.qty), "intent.qty")
+    leverage = _finite_positive(float(intent.exchange_leverage), "intent.exchange_leverage")
+    tid = _non_empty(trade_id, "trade_id")
+    oid = _non_empty(intent_id or f"violet-v4-{uuid.uuid4().hex}", "intent_id")
+    order = _non_empty(order_type, "order_type").upper()
+    if order == "LIMIT":
+        _finite_positive(limit_price, "limit_price")
+    elif float(limit_price or 0.0) < 0.0:
+        raise ValueError("limit_price must be >= 0")
+
+    merged_metadata: dict[str, Any] = dict(metadata or {})
+    merged_metadata.update(
+        {
+            "violet_contract_version": RUNNER_CONTRACT_VERSION,
+            "violet_ts_ns": int(intent.ts_ns),
+            "violet_maker_policy": intent.maker_policy,
+            "violet_target_notional": float(intent.target_notional),
+            "violet_reason": intent.reason,
+        }
+    )
+
+    return KernelIntent(
+        timestamp=timestamp or datetime.now(timezone.utc),
+        intent_id=oid,
+        trade_id=tid,
+        slot_id=int(slot_id),
+        asset=_non_empty(intent.asset, "intent.asset"),
+        side=_trade_side(intent.side),
+        action=_action_from_intent(intent, action),
+        reference_price=price,
+        target_size=qty,
+        leverage=leverage,
+        reason=f"violet_v4:{intent.reason.lower()}",
+        metadata=merged_metadata,
+        order_type=order,
+        limit_price=float(limit_price or 0.0),
+    )
+
+
+@dataclass(frozen=True)
+class VioletV4RunnerConfig:
+    """Operator/runtime knobs for the V4 execution runner."""
+
+    session_id: str
+    capital: float = 69_000.0
+    slot_id: int = 0
+    maker_policy: str = "maker_both"
+    order_type: str = "MARKET"
+    limit_price: float = 0.0
+    poll_interval_s: float = 0.25
+    queue_maxsize: int = 4096
+    max_submissions: int = 0
+    max_notional_usdt: float = 0.0
+    assume_vol_ok: bool = False
+    require_arming: bool = True
+    report_dir: str = "/mnt/vp-VIOLET_main/prod/VIOLET_dev/reports"
+    hz_cluster: str = "dolphin"
+    hz_host: str = "localhost:5701"
+    hz_map: str = "DOLPHIN_FEATURES"
+    hz_key: str = "latest_eigen_scan"
+    snapshot_symbol: str = "BTCUSDT"
+
+    @classmethod
+    def from_env(cls) -> "VioletV4RunnerConfig":
+        return cls(
+            session_id=os.environ.get("DOLPHIN_VIOLET_SESSION_ID", uuid.uuid4().hex),
+            capital=float(os.environ.get("DOLPHIN_VIOLET_CAPITAL", "69000")),
+            slot_id=int(os.environ.get("DOLPHIN_VIOLET_SLOT_ID", "0")),
+            maker_policy=os.environ.get("DOLPHIN_VIOLET_MAKER_POLICY", "maker_both"),
+            order_type=os.environ.get("DOLPHIN_VIOLET_ORDER_TYPE", "MARKET"),
+            limit_price=float(os.environ.get("DOLPHIN_VIOLET_LIMIT_PRICE", "0")),
+            poll_interval_s=float(os.environ.get("DOLPHIN_VIOLET_POLL_INTERVAL_SEC", "0.25")),
+            queue_maxsize=int(os.environ.get("DOLPHIN_VIOLET_SCAN_QUEUE_MAX", "4096")),
+            max_submissions=int(os.environ.get("DOLPHIN_VIOLET_MAX_SUBMISSIONS", "0")),
+            max_notional_usdt=float(os.environ.get("DOLPHIN_VIOLET_MAX_NOTIONAL_USDT", "0")),
+            assume_vol_ok=str(os.environ.get("DOLPHIN_VIOLET_ASSUME_VOL_OK", "0")).lower()
+            in {"1", "true", "yes", "on"},
+            require_arming=str(os.environ.get("DOLPHIN_VIOLET_REQUIRE_ARMING", "1")).lower()
+            not in {"0", "false", "no", "off"},
+            report_dir=os.environ.get(
+                "DOLPHIN_VIOLET_REPORT_DIR",
+                "/mnt/vp-VIOLET_main/prod/VIOLET_dev/reports",
+            ),
+            hz_cluster=os.environ.get("HZ_CLUSTER", "dolphin"),
+            hz_host=os.environ.get("HZ_HOST", "localhost:5701"),
+            hz_map=os.environ.get("DOLPHIN_VIOLET_SCAN_MAP", "DOLPHIN_FEATURES"),
+            hz_key=os.environ.get("DOLPHIN_VIOLET_SCAN_KEY", "latest_eigen_scan"),
+            snapshot_symbol=os.environ.get("DOLPHIN_VIOLET_SNAPSHOT_SYMBOL", "BTCUSDT"),
+        )
+
+
+class _LocalGuardedState:
+    """Tiny ASEx-compatible fallback for tests or missing ASEx install."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self.last_scan_number = -1
+        self.scans_seen = 0
+        self.decisions = 0
+        self.submissions = 0
+        self.outcomes = 0
+        self.errors = 0
+        self.last_trade_id = ""
+        self.last_intent_id = ""
+
+    def mutate(self, mutation: Mapping[str, Any]) -> dict[str, Any]:
+        with self._lock:
+            return self._apply(dict(mutation))
+
+    def _apply(self, mutation: dict[str, Any]) -> dict[str, Any]:
+        op = str(mutation.get("op") or "")
+        if op == "scan":
+            sn = int(mutation.get("scan_number") or 0)
+            if sn <= self.last_scan_number:
+                return {"accepted": False, "reason": "duplicate_scan", "last_scan_number": self.last_scan_number}
+            self.last_scan_number = sn
+            self.scans_seen += 1
+            return {"accepted": True, "last_scan_number": self.last_scan_number}
+        if op == "decision":
+            self.decisions += 1
+            return {"accepted": True, "decisions": self.decisions}
+        if op == "submission":
+            self.submissions += 1
+            self.last_trade_id = str(mutation.get("trade_id") or "")
+            self.last_intent_id = str(mutation.get("intent_id") or "")
+            return {"accepted": True, "submissions": self.submissions}
+        if op == "outcome":
+            self.outcomes += 1
+            return {"accepted": True, "outcomes": self.outcomes}
+        if op == "error":
+            self.errors += 1
+            return {"accepted": True, "errors": self.errors}
+        return {"accepted": False, "reason": f"unknown_op:{op}"}
+
+
+if ASExGuardedState is not None:
+
+    class VioletRunnerGuardedState(ASExGuardedState[dict[str, Any], dict[str, Any]]):
+        """ASEx single-writer state for scan/order lifecycle counters."""
+
+        def __init__(self) -> None:
+            super().__init__()
+            self._state = _LocalGuardedState()
+
+        def _validate(self, mutation: dict[str, Any]) -> bool:
+            return str(mutation.get("op") or "") in {
+                "scan",
+                "decision",
+                "submission",
+                "outcome",
+                "error",
+            }
+
+        def _apply(self, mutation: dict[str, Any]) -> dict[str, Any]:
+            return self._state._apply(dict(mutation))
+
+else:
+    VioletRunnerGuardedState = _LocalGuardedState  # type: ignore[assignment]
+
+
+class SerialRunnerState:
+    """Small wrapper around ASExWorker with a deterministic local fallback."""
+
+    def __init__(self) -> None:
+        self.backend = VioletRunnerGuardedState()
+        self.worker = ASExWorker(self.backend) if ASExWorker is not None else None
+
+    def mutate(self, mutation: Mapping[str, Any], *, timeout: float = 5.0) -> dict[str, Any]:
+        payload = dict(mutation)
+        if self.worker is not None:
+            return self.worker.mutate(payload).result(timeout=timeout)
+        return self.backend.mutate(payload)  # type: ignore[attr-defined]
+
+    def close(self) -> None:
+        if self.worker is not None:
+            self.worker.close(timeout=2.0)
+
+
+def _parse_scan_payload(raw: Any) -> dict[str, Any]:
+    if raw is None:
+        return {}
+    if isinstance(raw, str):
+        data = json.loads(raw)
+    elif isinstance(raw, Mapping):
+        data = dict(raw)
+    else:
+        return {}
+    if isinstance(data, dict) and data.get("version") == "NG7":
+        try:
+            from prod.clean_arch.adapters.eigen_scan_normalizer import normalize_ng7_scan
+
+            data = normalize_ng7_scan(data)
+        except Exception:
+            LOGGER.debug("NG7 normalize failed; using raw scan", exc_info=True)
+    return data if isinstance(data, dict) else {}
+
+
+def reference_price_from_scan(payload: Mapping[str, Any], asset: str) -> float:
+    """Extract the decision asset price from an NG7/universe payload."""
+
+    symbol = str(asset or "").upper()
+    assets = payload.get("assets") or []
+    prices = payload.get("asset_prices") or []
+    if isinstance(assets, list) and isinstance(prices, list):
+        for raw_asset, raw_price in zip(assets, prices):
+            if str(raw_asset).upper() != symbol:
+                continue
+            price = _finite_positive(float(raw_price), "reference_price")
+            return price
+
+    result = payload.get("result") if isinstance(payload, Mapping) else None
+    if isinstance(result, Mapping):
+        result_asset = str(result.get("asset") or payload.get("target_asset") or "").upper()
+        if result_asset == symbol:
+            return _finite_positive(float(result.get("price") or payload.get("price")), "reference_price")
+
+    payload_asset = str(payload.get("asset") or payload.get("target_asset") or "").upper()
+    if payload_asset == symbol and payload.get("price") is not None:
+        return _finite_positive(float(payload.get("price")), "reference_price")
+
+    raise ValueError(f"scan payload has no fresh price for {symbol}")
+
+
+def _decision_from_shadow(
+    shadow: dict[str, Any],
+    payload: dict[str, Any],
+    *,
+    scan_number: int,
+    now_ns: int,
+    vel_div: float,
+    vol_ok: bool,
+) -> Optional[ShadowDecision]:
+    """Run the existing BLUE-faithful VIOLET shadow path and return the decision."""
+
+    from .shadow_live_factors import _ensure_ob_engine
+
+    shadow["engine"].observe(payload, scan_number)
+    live_source = shadow.get("live_source")
+    if live_source is None:
+        return None
+
+    ob_engine = _ensure_ob_engine(shadow, payload)
+    live_result = live_source(
+        shadow["client"],
+        scan_history=shadow["scan_history"],
+        selector=shadow["selector"],
+        ob_engine=ob_engine,
+        bar_idx=shadow.get("ob_bar_idx", 0),
+        prior_boost_beta=shadow.get("prior_boost_beta"),
+    )
+    shadow["last_live_source"] = live_result
+    shadow["ob_bar_idx"] = shadow.get("ob_bar_idx", 0) + 1
+    ab = getattr(live_result, "acb_boost", None)
+    bb = getattr(live_result, "acb_beta", None)
+    if ab is not None and bb is not None:
+        shadow["prior_boost_beta"] = (ab, bb)
+
+    decision = shadow["engine"].decide(
+        now_ns=now_ns,
+        scan_number=scan_number,
+        capital=shadow["capital"],
+        vel_div=vel_div,
+        vol_ok=vol_ok,
+        factors=live_result.factors,
+    )
+    if decision is None:
+        return None
+    journal = shadow.get("journal")
+    if journal is not None:
+        journal.journal(decision, mono_ns=now_ns)
+    return decision
+
+
+class HazelcastNG7ScanSource:
+    """Event-listener first, polling fallback source for NG7 scans."""
+
+    def __init__(self, config: VioletV4RunnerConfig) -> None:
+        self.config = config
+        self.client = None
+        self.map = None
+        self.blocking_map = None
+        self.queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=config.queue_maxsize)
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._last_polled_scan = -1
+
+    async def connect(self) -> None:
+        import hazelcast
+
+        self._loop = asyncio.get_running_loop()
+        self.client = hazelcast.HazelcastClient(
+            cluster_name=self.config.hz_cluster,
+            cluster_members=[self.config.hz_host],
+        )
+        self.map = self.client.get_map(self.config.hz_map)
+        self.blocking_map = self.map.blocking()
+
+        def on_entry(event: Any) -> None:
+            try:
+                payload = _parse_scan_payload(getattr(event, "value", None))
+                if not payload:
+                    return
+                loop = self._loop
+                if loop is None:
+                    return
+                loop.call_soon_threadsafe(self._offer_payload, payload)
+            except Exception:
+                LOGGER.debug("scan listener callback failed", exc_info=True)
+
+        self.map.add_entry_listener(
+            include_value=True,
+            key=self.config.hz_key,
+            updated_func=on_entry,
+            added_func=on_entry,
+        )
+
+    def _offer_payload(self, payload: dict[str, Any]) -> None:
+        try:
+            self.queue.put_nowait(payload)
+        except asyncio.QueueFull:
+            try:
+                self.queue.get_nowait()
+            except asyncio.QueueEmpty:
+                pass
+            self.queue.put_nowait(payload)
+
+    async def next_payload(self) -> dict[str, Any]:
+        try:
+            return await asyncio.wait_for(self.queue.get(), timeout=self.config.poll_interval_s)
+        except asyncio.TimeoutError:
+            raw = self.blocking_map.get(self.config.hz_key)
+            payload = _parse_scan_payload(raw)
+            sn = int(payload.get("scan_number") or 0)
+            if sn <= self._last_polled_scan:
+                return {}
+            self._last_polled_scan = sn
+            return payload
+
+    async def close(self) -> None:
+        if self.client is not None:
+            self.client.shutdown()
+
+
+class VioletV4ExecutionRunner:
+    """Live-capable runner: decision path to actual DITAv2 execution."""
+
+    def __init__(
+        self,
+        *,
+        bundle: Any,
+        shadow: dict[str, Any],
+        config: VioletV4RunnerConfig,
+        state: Optional[SerialRunnerState] = None,
+        decision_step: Optional[
+            Callable[[dict[str, Any], dict[str, Any], int, int, float, bool], Optional[ShadowDecision]]
+        ] = None,
+    ) -> None:
+        self.bundle = bundle
+        self.shadow = shadow
+        self.config = config
+        self.state = state or SerialRunnerState()
+        self.decision_step = decision_step or (
+            lambda sh, payload, sn, now_ns, vd, vol_ok: _decision_from_shadow(
+                sh,
+                payload,
+                scan_number=sn,
+                now_ns=now_ns,
+                vel_div=vd,
+                vol_ok=vol_ok,
+            )
+        )
+        self._closed = False
+
+    async def connect(self) -> None:
+        connect = getattr(self.bundle.venue, "connect", None)
+        if connect is not None:
+            result = connect()
+            if asyncio.iscoroutine(result):
+                await result
+
+    def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        try:
+            close = getattr(self.bundle, "close", None)
+            if close is not None:
+                close()
+        finally:
+            self.state.close()
+
+    def _capital(self) -> float:
+        try:
+            cap = float(self.bundle.kernel.account.snapshot.capital)
+            if math.isfinite(cap) and cap > 0.0:
+                return cap
+        except Exception:
+            pass
+        return float(self.config.capital)
+
+    def _maybe_cap_exec_intent(self, intent: ExecIntent, reference_price: float) -> ExecIntent:
+        cap = float(self.config.max_notional_usdt or 0.0)
+        if cap <= 0.0 or float(intent.target_notional) <= cap:
+            return intent
+        return ExecIntent(
+            asset=intent.asset,
+            side=intent.side,
+            qty=cap / reference_price,
+            exchange_leverage=intent.exchange_leverage,
+            maker_policy=intent.maker_policy,
+            target_notional=cap,
+            ts_ns=intent.ts_ns,
+            reason=intent.reason,
+        )
+
+    async def process_scan_payload(self, raw_payload: Any) -> Optional[KernelOutcome]:
+        payload = _parse_scan_payload(raw_payload)
+        if not payload:
+            return None
+        scan_number = int(payload.get("scan_number") or 0)
+        if scan_number <= 0:
+            return None
+        scan_state = self.state.mutate({"op": "scan", "scan_number": scan_number})
+        if not scan_state.get("accepted"):
+            return None
+
+        vel_div = float(payload.get("vel_div") or 0.0)
+        if not math.isfinite(vel_div):
+            vel_div = 0.0
+        vol_ok = bool(payload.get("vol_ok", True))
+        if self.config.assume_vol_ok:
+            vol_ok = True
+        now_ns = int(time.monotonic_ns())
+        decision = self.decision_step(self.shadow, payload, scan_number, now_ns, vel_div, vol_ok)
+        if decision is None:
+            return None
+        self.state.mutate({"op": "decision", "scan_number": scan_number, "asset": decision.asset})
+
+        reference_price = reference_price_from_scan(payload, decision.asset)
+        exec_intent = to_exec_intent(
+            decision,
+            capital=self._capital(),
+            reference_price=reference_price,
+            maker_policy=self.config.maker_policy,
+        )
+        exec_intent = self._maybe_cap_exec_intent(exec_intent, reference_price)
+
+        trade_id = f"VIOLET-{self.config.session_id[:8]}-{scan_number}"
+        intent_id = f"violet-v4-{self.config.session_id[:8]}-{scan_number}"
+        kernel_intent = exec_intent_to_kernel_intent(
+            exec_intent,
+            reference_price=reference_price,
+            trade_id=trade_id,
+            slot_id=self.config.slot_id,
+            intent_id=intent_id,
+            order_type=self.config.order_type,
+            limit_price=self.config.limit_price,
+            metadata={
+                "violet_session_id": self.config.session_id,
+                "scan_number": scan_number,
+                "price_source": "ng7_scan",
+                "raw_notional_before_runner_cap": float(decision.target_exposure),
+            },
+        )
+        submission = self.state.mutate(
+            {"op": "submission", "scan_number": scan_number, "trade_id": trade_id, "intent_id": intent_id}
+        )
+        if self.config.max_submissions and int(submission.get("submissions") or 0) > self.config.max_submissions:
+            LOGGER.critical("max submissions reached; refusing scan=%s trade_id=%s", scan_number, trade_id)
+            return None
+
+        LOGGER.warning(
+            "VIOLET V4 LIVE SUBMIT scan=%s asset=%s side=%s qty=%.8f lev=%.2f ref=%.8f trade=%s",
+            scan_number,
+            kernel_intent.asset,
+            kernel_intent.side.value,
+            kernel_intent.target_size,
+            kernel_intent.leverage,
+            kernel_intent.reference_price,
+            trade_id,
+        )
+        try:
+            outcome = await self.bundle.kernel.process_intent_async(kernel_intent)
+            self.state.mutate(
+                {
+                    "op": "outcome",
+                    "scan_number": scan_number,
+                    "trade_id": trade_id,
+                    "accepted": bool(getattr(outcome, "accepted", False)),
+                    "diagnostic": str(getattr(getattr(outcome, "diagnostic_code", ""), "value", "")),
+                }
+            )
+            return outcome
+        except Exception as exc:
+            self.state.mutate({"op": "error", "scan_number": scan_number, "error": str(exc)})
+            raise
+
+
+def _apply_runtime_optimizations() -> None:
+    if gc.isenabled():
+        gc.disable()
+
+
+def _build_shadow(config: VioletV4RunnerConfig) -> dict[str, Any]:
+    from prod.ch_writer import ch_put_violet
+
+    from .decision_engine import VioletDecisionEngine
+    from .shadow_journal import VioletDecisionJournal
+    from .shadow_live_factors import build_shadow_live_source
+
+    thr = float(os.environ.get("DOLPHIN_VIOLET_ENTRY_VEL_DIV_THRESHOLD", "-0.02"))
+    return {
+        "engine": VioletDecisionEngine(entry_vel_div_threshold=thr),
+        "journal": VioletDecisionJournal(sink=ch_put_violet, session_id=config.session_id),
+        "capital": float(config.capital),
+        **build_shadow_live_source(),
+    }
+
+
+def _build_real_bundle(config: VioletV4RunnerConfig) -> Any:
+    from prod.clean_arch.dita_v2.launcher import build_launcher_bundle
+    from prod.launch_dolphin_violet import (
+        DDL_APPLY_CMD,
+        _apply_violet_env,
+        _preflight_clickhouse,
+        _violet_table_present,
+        build_bingx_exec_client_config,
+    )
+    from .v4_arming import assess_v4_arming, write_v4_arming_report
+
+    _apply_violet_env()
+    missing = _preflight_clickhouse()
+    if missing:
+        raise RuntimeError(f"dolphin_violet tables missing: {missing}; run {DDL_APPLY_CMD}")
+    if not _violet_table_present("violet_decisions"):
+        raise RuntimeError(f"dolphin_violet table missing: violet_decisions; run {DDL_APPLY_CMD}")
+    report = assess_v4_arming(report_dir=config.report_dir, launcher_mode="execution")
+    write_v4_arming_report(report, report_dir=config.report_dir)
+    if config.require_arming and not report.can_arm:
+        raise RuntimeError(f"VIOLET V4 arming refused: {report.reasons}")
+
+    return build_launcher_bundle(
+        venue_mode="BINGX",
+        max_slots=1,
+        prefix="violet",
+        bingx_config=build_bingx_exec_client_config(),
+    )
+
+
+async def run_live(
+    config: Optional[VioletV4RunnerConfig] = None,
+    *,
+    stop_after_first_submission: bool = False,
+) -> None:
+    """Run the actual VIOLET V4 BingX VST execution loop."""
+
+    cfg = config or VioletV4RunnerConfig.from_env()
+    _apply_runtime_optimizations()
+    bundle = _build_real_bundle(cfg)
+    shadow = _build_shadow(cfg)
+    source = HazelcastNG7ScanSource(cfg)
+    runner = VioletV4ExecutionRunner(bundle=bundle, shadow=shadow, config=cfg)
+    await runner.connect()
+    await source.connect()
+    LOGGER.critical(
+        "VIOLET V4 EXECUTION RUNNER ARMED: session=%s venue=BINGX_VST hz=%s[%s]",
+        cfg.session_id,
+        cfg.hz_map,
+        cfg.hz_key,
+    )
+    try:
+        while True:
+            payload = await source.next_payload()
+            if payload:
+                outcome = await runner.process_scan_payload(payload)
+                if stop_after_first_submission and outcome is not None:
+                    return
+            await asyncio.sleep(0)
+    finally:
+        await source.close()
+        runner.close()
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    parser = argparse.ArgumentParser(description="VIOLET V4 live BingX VST execution runner")
+    parser.add_argument("--once", action="store_true", help="process one queued/polled scan then exit")
+    args = parser.parse_args(argv)
+    config = VioletV4RunnerConfig.from_env()
+    if args.once:
+        config = dataclass_replace(config, max_submissions=1)
+    asyncio.run(run_live(config, stop_after_first_submission=args.once))
+    return 0
+
+
+def dataclass_replace(config: VioletV4RunnerConfig, **updates: Any) -> VioletV4RunnerConfig:
+    data = config.__dict__.copy()
+    data.update(updates)
+    return VioletV4RunnerConfig(**data)
+
+
+__all__ = [
+    "HazelcastNG7ScanSource",
+    "RUNNER_CONTRACT_VERSION",
+    "SerialRunnerState",
+    "VioletV4ExecutionRunner",
+    "VioletV4RunnerConfig",
+    "exec_intent_to_kernel_intent",
+    "main",
+    "reference_price_from_scan",
+    "run_live",
+]
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())