VIOLET V1c: observe-only launcher + hard guard + divergence monitor + service (DARK)

launch_dolphin_violet.py: own namespaces hard-set (CH dolphin_violet, HZ
DOLPHIN_STATE_VIOLET/PNL, Zinc prefix violet, DOLPHIN-VIOLET-001); own
credentials (BINGX_VIOLET_API_KEY/SECRET) — DARK idle with periodic WARNING
until provisioned; CH preflight SELECT-probes the required tables and NEVER
creates (DDL-before-code); kernel snapshot path repointed away from PINK's
fixed /tmp/.pink_kernel_state.json; mainnet hard-disabled; observe loop
never calls runtime.step(). ObserveOnlyVenue: submit/cancel raise
ObserveOnlyViolation with full attribute delegation — the kernel's
venue-submit-failure rollback converts a refusal into a synthetic REJECT
(slot back to IDLE), proven against the real kernel. FeedDivergenceMonitor:
per-asset scan-vs-venue divergence rows (bookTicker WS via
prod/bingx/market_stream, REST fallback) with stale-mid suppression and
plane seq propagation — the FET 0.2176-vs-0.1878 detector; runs even DARK
(public data). Supervisord [program:dolphin_violet] autostart=false, no
keys in conf by design. Violet package: 42 tests green + V0 gate.

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

prod/clean_arch/violet/divergence.py

@@ -0,0 +1,220 @@
+"""VIOLET V1: scan-vs-venue price divergence monitor (the FET detector).
+
+The 2026-06-11 FET incident entered on a scan price of 0.2176 while the
+venue traded 0.1878 — a 15% phantom divergence the system had no way to
+see. This monitor continuously samples, per asset, the gap between the
+scan-plane price (the alpha-side eigen-scan universe) and the execution
+venue's mid (BingX bookTicker), and journals one row per (scan, asset) to
+``dolphin_violet.violet_feed_divergence``.
+
+Venue mid sources:
+- WS (default): ``prod.bingx.market_stream.BingxMarketStream``
+  bookTicker subscription per symbol;
+- REST fallback (``DOLPHIN_VIOLET_VENUE_MID_MODE=rest``): 1 s poll of the
+  public quote endpoint — kept because the WS client is unexercised
+  in-tree; harden or drop at V2.
+
+Only PUBLIC data — runs even when VIOLET is DARK (no API keys).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, List, Optional
+
+from .clock import PlaneClock, mono_ns
+
+LOGGER = logging.getLogger("violet.divergence")
+
+Sink = Callable[[str, dict], None]
+
+
+def to_bingx_symbol(asset: str) -> str:
+    """BTCUSDT → BTC-USDT (idempotent for already-hyphenated input)."""
+    a = str(asset or "").upper()
+    if "-" in a or not a.endswith("USDT"):
+        return a
+    return f"{a[:-4]}-USDT"
+
+
+def from_bingx_symbol(symbol: str) -> str:
+    """BTC-USDT → BTCUSDT."""
+    return str(symbol or "").upper().replace("-", "")
+
+
+@dataclass
+class _VenueMid:
+    mid: float
+    mono: int
+    seq: int
+
+
+class FeedDivergenceMonitor:
+    """Samples scan-vs-venue divergence and journals it.
+
+    ``on_venue_tick`` is fed by the WS/REST source; ``on_scan`` is called by
+    the launcher's main loop with each data-feed snapshot. A divergence row
+    is emitted only when the venue mid is fresher than the venue plane's
+    staleness budget — a stale mid must never masquerade as live divergence.
+    """
+
+    TABLE = "violet_feed_divergence"
+
+    def __init__(
+        self,
+        *,
+        sink: Sink,
+        scan_clock: PlaneClock,
+        venue_clock: PlaneClock,
+        session_id: str,
+        table: str = TABLE,
+        logger: logging.Logger | None = None,
+    ) -> None:
+        self.sink = sink
+        self.scan_clock = scan_clock
+        self.venue_clock = venue_clock
+        self.session_id = str(session_id)
+        self.table = table
+        self.logger = logger or LOGGER
+        self._mids: Dict[str, _VenueMid] = {}
+        self._stream: Any = None
+        self._stream_task: Optional[asyncio.Task] = None
+        self._rest_task: Optional[asyncio.Task] = None
+        self.rows_emitted = 0
+
+    # ── venue side ───────────────────────────────────────────────────────────
+
+    def on_venue_tick(self, symbol: str, bid: float, ask: float) -> None:
+        try:
+            b, a = float(bid), float(ask)
+        except (TypeError, ValueError):
+            return
+        if not (b > 0 and a > 0):
+            return
+        seq = self.venue_clock.tick()
+        self._mids[from_bingx_symbol(symbol)] = _VenueMid(
+            mid=(b + a) / 2.0, mono=mono_ns(), seq=seq
+        )
+
+    async def _on_ws_frame(self, payload: dict) -> None:
+        """BingX bookTicker frame: {dataType: 'BTC-USDT@bookTicker',
+        data: {... 'b': bid, 'a': ask ...}} (field names per BingX swap WS)."""
+        try:
+            data_type = str(payload.get("dataType", "") or "")
+            if "@bookTicker" not in data_type:
+                return
+            symbol = data_type.split("@", 1)[0]
+            data = payload.get("data") or {}
+            bid = data.get("b") or data.get("bidPrice") or data.get("bestBid")
+            ask = data.get("a") or data.get("askPrice") or data.get("bestAsk")
+            if bid is None or ask is None:
+                return
+            self.on_venue_tick(symbol, float(bid), float(ask))
+        except Exception as exc:  # noqa: BLE001 — never kill the stream
+            self.logger.debug("divergence WS frame parse skipped: %s", exc)
+
+    async def _rest_poll_loop(self, symbols: List[str], rest_base_url: str) -> None:
+        import aiohttp
+
+        url = rest_base_url.rstrip("/") + "/openApi/swap/v2/quote/bookTicker"
+        async with aiohttp.ClientSession() as session:
+            while True:
+                for sym in symbols:
+                    try:
+                        async with session.get(
+                            url, params={"symbol": to_bingx_symbol(sym)},
+                            timeout=aiohttp.ClientTimeout(total=5),
+                        ) as resp:
+                            body = await resp.json(content_type=None)
+                        data = (body or {}).get("data") or {}
+                        # quote payloads use bidPrice/askPrice (REST shape)
+                        book = data.get("book_ticker") or data
+                        bid = book.get("bid_price") or book.get("bidPrice") or book.get("b")
+                        ask = book.get("ask_price") or book.get("askPrice") or book.get("a")
+                        if bid is not None and ask is not None:
+                            self.on_venue_tick(sym, float(bid), float(ask))
+                    except Exception as exc:  # noqa: BLE001
+                        self.logger.debug("divergence REST poll %s failed: %s", sym, exc)
+                await asyncio.sleep(1.0)
+
+    async def start(
+        self,
+        symbols: List[str],
+        *,
+        mode: str = "ws",
+        ws_url: str = "wss://open-api-swap.bingx.com/swap-market",
+        rest_base_url: str = "https://open-api.bingx.com",
+    ) -> None:
+        if mode == "rest":
+            self._rest_task = asyncio.create_task(
+                self._rest_poll_loop(symbols, rest_base_url),
+                name="violet_divergence_rest",
+            )
+            self.logger.info("divergence: REST mid poll started for %d symbols", len(symbols))
+            return
+        from prod.bingx.market_stream import BingxMarketStream
+
+        self._stream = BingxMarketStream(ws_url=ws_url, on_event=self._on_ws_frame)
+        for sym in symbols:
+            self._stream.subscribe(f"{to_bingx_symbol(sym)}@bookTicker")
+        self._stream_task = asyncio.create_task(
+            self._stream.run_forever(), name="violet_divergence_ws"
+        )
+        self.logger.info("divergence: WS bookTicker started for %d symbols", len(symbols))
+
+    async def stop(self) -> None:
+        for task in (self._stream_task, self._rest_task):
+            if task is not None:
+                task.cancel()
+                try:
+                    await task
+                except (asyncio.CancelledError, Exception):
+                    pass
+        self._stream_task = self._rest_task = None
+
+    # ── scan side ─────────────────────────────────────────────────────────────
+
+    def on_scan(self, snapshot: Any) -> int:
+        """Called once per data-feed snapshot. Returns rows emitted."""
+        payload = getattr(snapshot, "scan_payload", None)
+        if not isinstance(payload, dict):
+            return 0
+        assets = payload.get("assets") or []
+        prices = payload.get("asset_prices") or []
+        if not (isinstance(assets, list) and isinstance(prices, list) and assets):
+            return 0
+        scan_seq = self.scan_clock.tick()
+        now = mono_ns()
+        emitted = 0
+        for asset, price in zip(assets, prices):
+            try:
+                scan_price = float(price)
+            except (TypeError, ValueError):
+                continue
+            if scan_price <= 0:
+                continue
+            key = str(asset).upper()
+            vm = self._mids.get(key)
+            if vm is None:
+                continue
+            # Stale venue mid must never masquerade as live divergence.
+            if (now - vm.mono) > self.venue_clock.staleness_budget_ns:
+                continue
+            divergence_bps = (vm.mid - scan_price) / scan_price * 1e4
+            self.sink(self.table, {
+                "ts": int(time.time() * 1000),          # DateTime64(3)
+                "session_id": self.session_id,
+                "asset": key,
+                "scan_price": scan_price,
+                "venue_mid": vm.mid,
+                "divergence_bps": divergence_bps,
+                "scan_seq": int(scan_seq),
+                "venue_seq": int(vm.seq),
+                "mono_ns": int(now),
+            })
+            emitted += 1
+        self.rows_emitted += emitted
+        return emitted

prod/clean_arch/violet/observe_guard.py

@@ -0,0 +1,71 @@
+"""VIOLET observe-only hard guard (Stage V1).
+
+``ObserveOnlyVenue`` wraps any venue adapter and makes order placement
+structurally impossible: ``submit``/``submit_async``/``cancel``/
+``cancel_async`` raise ``ObserveOnlyViolation`` and log CRITICAL. Every
+other attribute (including assignment — e.g. the runtime's
+``venue._kernel_ref = kernel``) delegates to the wrapped adapter, so
+account streams, reconcile reads and connection management work unchanged.
+
+This is the hard guarantee, independent of policy configuration: even if a
+policy step were ever wired into the V1 runtime by mistake, no order can
+reach the exchange.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+LOGGER = logging.getLogger("violet.observe_guard")
+
+_BLOCKED = ("submit", "submit_async", "cancel", "cancel_async")
+_SELF_ATTRS = ("_inner", "_logger")
+
+
+class ObserveOnlyViolation(RuntimeError):
+    """An order-placement call reached the observe-only venue guard."""
+
+
+class ObserveOnlyVenue:
+    """Delegating wrapper that refuses order placement."""
+
+    def __init__(self, inner: Any, logger: logging.Logger | None = None) -> None:
+        object.__setattr__(self, "_inner", inner)
+        object.__setattr__(self, "_logger", logger or LOGGER)
+
+    # -- blocked surface -----------------------------------------------------
+
+    def _refuse(self, name: str, *args: Any, **kwargs: Any) -> None:
+        msg = (
+            f"OBSERVE-ONLY VIOLATION: venue.{name}() called — VIOLET V1 must "
+            f"never place or cancel orders (args={args!r:.200})"
+        )
+        object.__getattribute__(self, "_logger").critical(msg)
+        raise ObserveOnlyViolation(msg)
+
+    def submit(self, *a: Any, **k: Any) -> Any:
+        self._refuse("submit", *a, **k)
+
+    async def submit_async(self, *a: Any, **k: Any) -> Any:
+        self._refuse("submit_async", *a, **k)
+
+    def cancel(self, *a: Any, **k: Any) -> Any:
+        self._refuse("cancel", *a, **k)
+
+    async def cancel_async(self, *a: Any, **k: Any) -> Any:
+        self._refuse("cancel_async", *a, **k)
+
+    # -- delegation ----------------------------------------------------------
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(object.__getattribute__(self, "_inner"), name)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        if name in _SELF_ATTRS:
+            object.__setattr__(self, name, value)
+        else:
+            setattr(object.__getattribute__(self, "_inner"), name, value)
+
+    def __repr__(self) -> str:  # pragma: no cover — debug helper
+        return f"ObserveOnlyVenue({object.__getattribute__(self, '_inner')!r})"

prod/clean_arch/violet/test_violet_dark_idle.py

@@ -0,0 +1,109 @@
+"""V1: dark-idle behavior — no keys ⇒ idle loop, exec adapter never built."""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+sys.path.insert(0, "/mnt/dolphinng5_predict/nautilus_dolphin")
+
+import pytest
+
+
+def _clear_keys(monkeypatch):
+    monkeypatch.delenv("BINGX_VIOLET_API_KEY", raising=False)
+    monkeypatch.delenv("BINGX_VIOLET_SECRET_KEY", raising=False)
+
+
+def test_keys_present_detection(monkeypatch):
+    import prod.launch_dolphin_violet as lv
+
+    _clear_keys(monkeypatch)
+    assert lv._violet_keys_present() is False
+    monkeypatch.setenv("BINGX_VIOLET_API_KEY", "k")
+    assert lv._violet_keys_present() is False     # secret still missing
+    monkeypatch.setenv("BINGX_VIOLET_SECRET_KEY", "s")
+    assert lv._violet_keys_present() is True
+    monkeypatch.setenv("BINGX_VIOLET_API_KEY", "   ")
+    assert lv._violet_keys_present() is False     # whitespace is absent
+
+
+def test_dark_run_never_builds_exec_adapter(monkeypatch):
+    """Without keys, run() must enter the dark loop and never construct the
+    observe runtime (and therefore never the BingX exec adapter)."""
+    import prod.launch_dolphin_violet as lv
+
+    _clear_keys(monkeypatch)
+    monkeypatch.setenv("DOLPHIN_VIOLET_DARK_DIVERGENCE", "0")  # strict idle
+    monkeypatch.setattr(lv, "_preflight_clickhouse", lambda: [])
+
+    def boom():
+        raise AssertionError("observe runtime must not be built while DARK")
+
+    monkeypatch.setattr(lv, "_build_observe_runtime", boom)
+
+    warnings: list[str] = []
+    monkeypatch.setattr(
+        lv.LOGGER, "warning", lambda msg, *a: warnings.append(msg % a if a else str(msg))
+    )
+
+    async def drive():
+        task = asyncio.ensure_future(lv.run())
+        await asyncio.sleep(0.3)        # boot path reaches the dark loop
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    asyncio.run(drive())
+    assert any("VIOLET DARK" in w for w in warnings), warnings
+
+
+def test_missing_tables_idles_dark_naming_apply_cmd(monkeypatch):
+    import prod.launch_dolphin_violet as lv
+
+    _clear_keys(monkeypatch)
+    monkeypatch.setattr(lv, "_preflight_clickhouse", lambda: ["violet_feed_divergence"])
+    crits: list[str] = []
+    monkeypatch.setattr(
+        lv.LOGGER, "critical", lambda msg, *a: crits.append(msg % a if a else str(msg))
+    )
+
+    async def drive():
+        task = asyncio.ensure_future(lv.run())
+        await asyncio.sleep(0.3)
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    asyncio.run(drive())
+    assert crits and "apply_violet_ddl.py" in crits[0]
+    assert "violet_feed_divergence" in crits[0]
+
+
+def test_violet_env_hard_sets_namespaces_and_snapshot_path(monkeypatch):
+    import prod.launch_dolphin_violet as lv
+    import prod.clean_arch.runtime.pink_direct as pd
+
+    monkeypatch.setenv("DOLPHIN_STATE_MAP", "DOLPHIN_STATE_PINK")  # poisoned env
+    original = pd._KERNEL_STATE_PATH
+    try:
+        lv._apply_violet_env()
+        import os
+
+        assert os.environ["DOLPHIN_STATE_MAP"] == "DOLPHIN_STATE_VIOLET"
+        assert os.environ["DITA_V2_PREFIX"] == "violet"
+        assert os.environ["DOLPHIN_BINGX_ALLOW_MAINNET"] == "0"
+        # kernel snapshot file repointed away from PINK's fixed path
+        assert pd._KERNEL_STATE_PATH == lv._KERNEL_STATE_PATH_VIOLET
+        assert "pink" not in str(pd._KERNEL_STATE_PATH)
+    finally:
+        pd._KERNEL_STATE_PATH = original
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))

prod/clean_arch/violet/test_violet_divergence.py

@@ -0,0 +1,119 @@
+"""V1: FeedDivergenceMonitor — row shape vs DDL, sign, staleness, seqs."""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+import pytest
+
+from prod.clean_arch.violet.clock import PlaneClock, mono_ns
+from prod.clean_arch.violet.divergence import (
+    FeedDivergenceMonitor,
+    from_bingx_symbol,
+    to_bingx_symbol,
+)
+
+DDL_PATH = Path(
+    "/mnt/dolphinng5_predict/prod/clickhouse/violet/20_violet_feed_divergence.sql"
+)
+
+
+def _mk(sink_rows, venue_budget_ns=2_000_000_000):
+    return FeedDivergenceMonitor(
+        sink=lambda table, row: sink_rows.append((table, row)),
+        scan_clock=PlaneClock("scan", 12_000_000_000),
+        venue_clock=PlaneClock("venue", venue_budget_ns),
+        session_id="sess-test",
+    )
+
+
+def _snapshot(assets, prices):
+    return SimpleNamespace(scan_payload={"assets": assets, "asset_prices": prices})
+
+
+def test_symbol_mapping_round_trip():
+    assert to_bingx_symbol("FETUSDT") == "FET-USDT"
+    assert to_bingx_symbol("FET-USDT") == "FET-USDT"     # idempotent
+    assert from_bingx_symbol("FET-USDT") == "FETUSDT"
+
+
+def test_row_keys_exactly_match_ddl_columns():
+    """Parse the shipped DDL: emitted row keys must equal the column set."""
+    ddl = DDL_PATH.read_text()
+    cols = set(re.findall(r"`(\w+)`", ddl))
+    rows = []
+    m = _mk(rows)
+    m.on_venue_tick("FET-USDT", 0.1877, 0.1879)
+    m.on_scan(_snapshot(["FETUSDT"], [0.2176]))
+    assert len(rows) == 1
+    table, row = rows[0]
+    assert table == "violet_feed_divergence"
+    assert set(row.keys()) == cols, (set(row.keys()) ^ cols)
+
+
+def test_bps_sign_convention_venue_above_scan_positive():
+    rows = []
+    m = _mk(rows)
+    m.on_venue_tick("BTC-USDT", 101.0, 101.0)
+    m.on_scan(_snapshot(["BTCUSDT"], [100.0]))
+    assert rows[0][1]["divergence_bps"] == pytest.approx(100.0)   # +1% = +100bps
+    rows.clear()
+    m.on_venue_tick("BTC-USDT", 99.0, 99.0)
+    m.on_scan(_snapshot(["BTCUSDT"], [100.0]))
+    assert rows[0][1]["divergence_bps"] == pytest.approx(-100.0)
+
+
+def test_fet_incident_magnitude():
+    """The motivating case: scan 0.2176 vs venue ~0.1878 ⇒ ~ -1369 bps."""
+    rows = []
+    m = _mk(rows)
+    m.on_venue_tick("FET-USDT", 0.1877, 0.1879)
+    m.on_scan(_snapshot(["FETUSDT"], [0.2176]))
+    bps = rows[0][1]["divergence_bps"]
+    assert bps == pytest.approx((0.1878 - 0.2176) / 0.2176 * 1e4, rel=1e-6)
+    assert bps < -1300
+
+
+def test_stale_venue_mid_suppressed():
+    rows = []
+    m = _mk(rows, venue_budget_ns=1)              # everything is stale
+    m.on_venue_tick("BTC-USDT", 100.0, 100.0)
+    import time
+
+    time.sleep(0.001)
+    m.on_scan(_snapshot(["BTCUSDT"], [100.0]))
+    assert rows == []                              # no phantom divergence
+
+
+def test_seq_propagation_and_no_mid_no_row():
+    rows = []
+    m = _mk(rows)
+    m.on_venue_tick("BTC-USDT", 100.0, 100.0)
+    m.on_venue_tick("BTC-USDT", 100.2, 100.2)      # venue_seq advances to 2
+    m.on_scan(_snapshot(["BTCUSDT", "ETHUSDT"], [100.0, 2000.0]))
+    assert len(rows) == 1                          # ETH has no venue mid
+    row = rows[0][1]
+    assert row["venue_seq"] == 2
+    assert row["scan_seq"] == 1
+    m.on_scan(_snapshot(["BTCUSDT"], [100.1]))
+    assert rows[-1][1]["scan_seq"] == 2
+
+
+def test_garbage_inputs_ignored():
+    rows = []
+    m = _mk(rows)
+    m.on_venue_tick("BTC-USDT", 0.0, -1.0)         # invalid quotes ignored
+    m.on_venue_tick("BTC-USDT", "x", None)         # type garbage ignored
+    m.on_scan(_snapshot(["BTCUSDT"], ["nan-ish", 0.0]))
+    m.on_scan(SimpleNamespace(scan_payload=None))
+    m.on_scan(SimpleNamespace())                   # no payload attr at all
+    assert rows == []
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))

prod/clean_arch/violet/test_violet_observe_guard.py

@@ -0,0 +1,121 @@
+"""V1: ObserveOnlyVenue — refusal + delegation contract."""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+sys.path.insert(0, "/mnt/dolphinng5_predict/nautilus_dolphin")
+
+import pytest
+
+from prod.clean_arch.violet.observe_guard import ObserveOnlyVenue, ObserveOnlyViolation
+
+
+class _Inner:
+    def __init__(self):
+        self.connected = False
+        self.calls = []
+
+    def submit(self, intent):
+        self.calls.append(("submit", intent))
+        return ["should-never-happen"]
+
+    async def submit_async(self, intent):
+        self.calls.append(("submit_async", intent))
+        return ["should-never-happen"]
+
+    def cancel(self, order, reason=""):
+        self.calls.append(("cancel", order))
+        return []
+
+    async def cancel_async(self, order, reason=""):
+        self.calls.append(("cancel_async", order))
+        return []
+
+    async def connect(self):
+        self.connected = True
+
+    async def subscribe(self):
+        if False:  # pragma: no cover — generator shape
+            yield None
+
+
+@pytest.mark.asyncio
+async def test_submit_and_cancel_raise_and_never_reach_inner():
+    inner = _Inner()
+    guard = ObserveOnlyVenue(inner)
+    with pytest.raises(ObserveOnlyViolation):
+        guard.submit({"x": 1})
+    with pytest.raises(ObserveOnlyViolation):
+        guard.cancel(object())
+    with pytest.raises(ObserveOnlyViolation):
+        await guard.submit_async({})
+    with pytest.raises(ObserveOnlyViolation):
+        await guard.cancel_async(object())
+    assert inner.calls == []
+
+
+@pytest.mark.asyncio
+async def test_attribute_get_delegates():
+    inner = _Inner()
+    guard = ObserveOnlyVenue(inner)
+    assert guard.connected is False
+    await guard.connect()
+    assert inner.connected is True and guard.connected is True
+
+
+def test_attribute_set_delegates_to_inner():
+    """The runtime does `venue._kernel_ref = kernel` — assignment must land
+    on the wrapped adapter, not the guard."""
+    inner = _Inner()
+    guard = ObserveOnlyVenue(inner)
+    sentinel = object()
+    guard._kernel_ref = sentinel
+    assert getattr(inner, "_kernel_ref") is sentinel
+    assert not hasattr(type(guard), "_kernel_ref")
+
+
+def test_wrapped_mock_venue_full_kernel_drive_never_submits():
+    """Real ExecutionKernel over a wrapped MOCK venue: an ENTER intent is
+    refused at the venue boundary; the kernel's venue-submit-failure
+    rollback converts the refusal into a synthetic REJECT (slot returns to
+    IDLE) and the inner venue NEVER sees the order."""
+    from prod.clean_arch.dita_v2.launcher import build_launcher_bundle
+    from prod.clean_arch.dita_v2.contracts import (
+        KernelCommandType,
+        KernelIntent,
+        TradeSide,
+        TradeStage,
+    )
+    from datetime import datetime, timezone
+
+    bundle = build_launcher_bundle(venue_mode="MOCK", max_slots=1)
+    kernel = bundle.kernel
+    inner = kernel.venue
+    submitted = []
+    orig_submit = inner.submit
+    inner.submit = lambda *a, **k: submitted.append(a) or orig_submit(*a, **k)
+    kernel.venue = ObserveOnlyVenue(inner)
+
+    intent = KernelIntent(
+        timestamp=datetime.now(timezone.utc),
+        intent_id="i-1", trade_id="T-OBS-1", slot_id=0,
+        asset="BTCUSDT", side=TradeSide.SHORT,
+        action=KernelCommandType.ENTER,
+        reference_price=100.0, target_size=1.0, leverage=1.0,
+        exit_leg_ratios=(1.0,), reason="test", metadata={},
+        stage=TradeStage.INTENT_CREATED,
+    )
+    outcome = kernel.process_intent(intent)
+    assert submitted == []                       # nothing reached the venue
+    assert kernel.slot(0).is_free()              # FSM rolled back to IDLE
+    assert any(
+        "VENUE_SUBMIT_ERROR" in str(e.reason)
+        for e in outcome.emitted_events
+    ), "expected the synthetic REJECT from the guard refusal"
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))