exec(uv): T1 smart-exec bridge — maker-first entries, gated + dormant

SmartExecBridge: a drop-in for PromotionBridge.try_promote that rests entries as PostOnly GTX
makers (via exec_unified.router policy) instead of always paying the taker cross. DORMANT —
selected only by UV_SMART_EXEC=1 (default off); nothing imports it yet, so the running flight
is untouched. T1 = smallest bug surface (spec §15):
- ENTER -> maker (ACQUIRE): PostOnly LIMIT @ touch; unfilled by next scan tick -> CANCEL/abandon
  ('a missed entry is free' §4-1). No chase, no cross, no requote race.
- EXIT  -> MARKET unchanged (never strand; zero new exit risk in v1).
- the 6s scan tick IS the drive clock: sweep-stale-then-place each promote.
Reuses router.decide + friction side-lane (never blocks a promote); fail-soft (never raises
into the scan loop). 11 tests, 2 mutation litmus RED (entry-maker policy; stale-sweep).

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

prod/clean_arch/violet/uv/exec/smart_bridge.py

@@ -0,0 +1,226 @@
+"""Smart-exec bolt for the PRIME flight (T1 — "better orders, usable right now").
+
+Bolts the unified exec layer's *whether-maker* policy (prod.exec_unified.router) onto the
+PRIME promotion path so entries try to REST as maker (PostOnly GTX) instead of always paying
+the taker cross. It is a drop-in alternative to the naive one-shot MARKET promote in
+``promotion.PromotionBridge`` — selected by the ``UV_SMART_EXEC`` env flag, DORMANT by default
+so the running flight is unaffected until an operator flips it.
+
+T1 policy (deliberately the smallest nonzero bug surface — spec §15 T1):
+  * ENTER → maker (ACQUIRE): PostOnly LIMIT at the reference touch. If it does not fill by the
+    next scan tick it is CANCELLED and abandoned — "a missed entry is free" (§4-1). No chase,
+    no cross, no requote race. The one new state is "resting GTX with a TTL", the one new
+    transition is "TTL → cancel", and both are read off the kernel slot, not invented here.
+  * EXIT → MARKET (unchanged naive behaviour): exits never strand, and adding maker-then-cross
+    to the exit leg is T2+; v1 takes ZERO new risk on the exit side.
+  * The 6 s scan tick IS the drive clock: each promote first sweeps a stale resting entry
+    (cancel if the slot still shows it working past its TTL), THEN places the new one.
+
+WHAT THIS DOES NOT DO (honest negative constraints, §13):
+  * It never changes side/size/asset (reads them off the intent the promotion layer built).
+  * It never crosses on a missed ENTER (abandons). It never delays or vetoes an EXIT.
+  * It writes friction telemetry on a SIDE LANE only (never blocks a promote).
+
+Reuses: prod.exec_unified.router.decide (policy), .friction (telemetry). dita_v2 imported
+lazily so importing this module never drags the kernel in.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import time
+from dataclasses import dataclass, field
+from decimal import Decimal
+from typing import Any, Optional
+
+from prod.exec_unified.contract import ExecutionRequest, Side, UrgencyClass
+from prod.exec_unified.friction import FrictionJournal, FrictionRecord
+from prod.exec_unified.router import ExecutionMethod, decide
+
+LOGGER = logging.getLogger("uv.smart_bridge")
+
+SMART_EXEC_ENV = "UV_SMART_EXEC"
+# Resting entry TTL. The scan tick (~6 s) is the coarse clock; anything older than this at the
+# next tick is treated as expired and swept. Kept >= one tick so a fresh quote gets a full tick.
+DEFAULT_ENTRY_TTL_S = 6.0
+
+
+def smart_exec_enabled() -> bool:
+    """True iff the operator has opted this flight into smart exec. Off ⇒ caller uses the
+    naive promote. Re-read every call (no caching) so it can be toggled without a code change."""
+    return os.environ.get(SMART_EXEC_ENV, "0").strip() == "1"
+
+
+def _side_from_intent(intent: Any) -> Side:
+    """Kernel intent carries POSITION side (LONG/SHORT); the ORDER side we quote is BUY to open
+    a long / sell to open a short (this path is entries only). Verified vs bingx_venue:627."""
+    val = getattr(getattr(intent, "side", None), "value", "") or str(getattr(intent, "side", ""))
+    return Side.BUY if val.upper() == "LONG" else Side.SELL
+
+
+def _is_enter(intent: Any) -> bool:
+    val = getattr(getattr(intent, "action", None), "value", "") or str(getattr(intent, "action", ""))
+    return val.upper() == "ENTER"
+
+
+def urgency_for(intent: Any) -> UrgencyClass:
+    """T1 urgency assignment: entries are patient makers that abandon; everything else (exits)
+    crosses now. This is the ONLY place the T1 policy chooses a class."""
+    return UrgencyClass.ACQUIRE if _is_enter(intent) else UrgencyClass.CATASTROPHIC
+
+
+@dataclass
+class _Resting:
+    """One tracked resting maker (single slot). Enough to sweep it at the next tick."""
+    intent_id: str
+    trade_id: str
+    placed_at: float
+    reference_price: float
+
+
+@dataclass
+class SmartExecBridge:
+    """Drop-in for PromotionBridge.try_promote with T1 maker-first entries.
+
+    Same gate (injected), same fail-soft contract (never raises out of try_promote). Holds the
+    single-slot resting-entry state so it can sweep a stale quote before placing a new one."""
+
+    gate: Any                         # PromotionGate (two-man rule) — reused verbatim
+    kernel: Any                       # DITAv2 ExecutionKernel (.process_intent / .slot)
+    friction: FrictionJournal = field(default_factory=FrictionJournal)
+    entry_ttl_s: float = DEFAULT_ENTRY_TTL_S
+    clock: Any = time.monotonic
+    stats: Any = None                 # optional shared _Stats from promotion (duck-typed)
+    _resting: Optional[_Resting] = None
+
+    # ── slot truth (tolerant of partial kernels) ─────────────────────────────
+    def _slot_stage_size(self) -> tuple[str, Decimal, str]:
+        try:
+            slot = self.kernel.slot(0)
+        except Exception:
+            return "", Decimal(0), ""
+        stage = getattr(getattr(slot, "fsm_state", None), "value", None) or str(getattr(slot, "fsm_state", "") or "")
+        size = Decimal(str(getattr(slot, "size", 0) or 0))
+        tid = str(getattr(slot, "trade_id", "") or "")
+        return str(stage), size, tid
+
+    def _entry_still_working(self, resting: _Resting) -> bool:
+        stage, size, tid = self._slot_stage_size()
+        # unfilled entry ⇒ still in a working/pre-fill stage AND no position size yet, same trade
+        working = stage in ("ENTRY_WORKING", "ORDER_REQUESTED", "ORDER_SENT", "ORDER_ACKED", "IDLE")
+        return working and size <= 0 and (tid == resting.trade_id or tid == "")
+
+    def _sweep_stale_entry(self) -> None:
+        """Cancel a resting maker entry that has outlived its TTL and still hasn't filled. Pump
+        happens inside the kernel on the next process_intent; we only decide to cancel."""
+        r = self._resting
+        if r is None:
+            return
+        expired = (self.clock() - r.placed_at) >= self.entry_ttl_s
+        if not expired:
+            return
+        if self._entry_still_working(r):
+            try:
+                self.kernel.process_intent(self._cancel_intent(r))
+                LOGGER.info("SMART: swept stale resting entry %s (ttl %.1fs)", r.intent_id, self.entry_ttl_s)
+            except Exception as exc:                       # fail-soft: a failed cancel must not block
+                LOGGER.warning("SMART: sweep cancel failed for %s: %r", r.intent_id, exc)
+        self._resting = None
+
+    def _cancel_intent(self, r: _Resting) -> Any:
+        from prod.clean_arch.dita_v2.contracts import KernelCommandType, KernelIntent, TradeSide
+        from datetime import datetime, timezone
+        return KernelIntent(
+            timestamp=datetime.now(timezone.utc), intent_id=f"{r.intent_id}-cxl",
+            trade_id=r.trade_id, slot_id=0, asset="", side=TradeSide.FLAT,
+            action=KernelCommandType.CANCEL, reference_price=0.0, target_size=0.0, leverage=1.0,
+            reason="uv_smart:ttl_abandon", metadata={"smart_exec": True},
+        )
+
+    def _as_maker(self, intent: Any) -> Any:
+        """Return a maker-shaped copy of the promotion-built intent: LIMIT @ touch, PostOnly.
+        Everything else (side/size/asset/ids) is preserved untouched."""
+        import dataclasses
+        meta = dict(getattr(intent, "metadata", {}) or {})
+        meta.update({"smart_exec": True, "tif": "PostOnly", "uv_exec_tier": "T1"})
+        return dataclasses.replace(
+            intent, order_type="LIMIT",
+            limit_price=float(getattr(intent, "reference_price", 0.0) or 0.0),
+            metadata=meta,
+        )
+
+    def try_promote(self, *, decision: dict, ctx: Any = None,
+                    build_intent: Any = None) -> bool:
+        """Gated, fail-soft T1 promote. ``build_intent`` is the promotion layer's
+        ``build_kernel_intent_from_decision`` (injected so this module never imports flight code)."""
+        try:
+            if not self.gate.is_active():
+                self._bump("skipped")
+                return False
+            if not decision.get("has_entry") and not decision.get("is_exit"):
+                # nothing actionable this tick, but still sweep a stale resting quote
+                self._sweep_stale_entry()
+                self._bump("skipped")
+                return False
+
+            intent = build_intent(decision=decision, ctx=ctx, reason="uv_smart")
+            self._sweep_stale_entry()                      # pump-before-place: clear a stale quote first
+
+            u = urgency_for(intent)
+            req = self._request_from(intent, u)
+            routing = decide(req)
+            maker = routing.method is ExecutionMethod.MAKER
+
+            send = self._as_maker(intent) if maker else intent   # taker path = the naive intent verbatim
+            outcome = self.kernel.process_intent(send)
+            dc = getattr(getattr(outcome, "diagnostic_code", ""), "value", None) or str(
+                getattr(outcome, "diagnostic_code", ""))
+            accepted = dc == "OK"
+
+            if maker and accepted:
+                self._resting = _Resting(
+                    intent_id=str(getattr(send, "intent_id", "")),
+                    trade_id=str(getattr(send, "trade_id", "")),
+                    placed_at=self.clock(),
+                    reference_price=float(getattr(send, "reference_price", 0.0) or 0.0),
+                )
+            self._emit_friction(send, req, maker)
+            self._bump("accepted" if accepted else "rejected")
+            LOGGER.info("SMART PROMOTE: %s method=%s -> %s (asset=%s qty=%.4f)",
+                        getattr(send, "intent_id", "?"), routing.method.value, dc,
+                        getattr(send, "asset", "?"), float(getattr(send, "target_size", 0.0) or 0.0))
+            return accepted
+        except Exception as exc:                            # never raise into the scan loop
+            self._bump("errors")
+            LOGGER.error("SMART PROMOTE ERROR: %s: %s", type(exc).__name__, exc)
+            return False
+
+    def _request_from(self, intent: Any, urgency: UrgencyClass) -> ExecutionRequest:
+        return ExecutionRequest(
+            request_id=str(getattr(intent, "intent_id", "") or "uv"),
+            asset=str(getattr(intent, "asset", "") or "BTCUSDT"),
+            side=_side_from_intent(intent),
+            size=Decimal(str(getattr(intent, "target_size", 0) or 0)),
+            urgency=urgency,
+        )
+
+    def _emit_friction(self, intent: Any, req: ExecutionRequest, maker: bool) -> None:
+        ref = Decimal(str(getattr(intent, "reference_price", 0) or 0))
+        if ref <= 0:
+            return                                          # no reference ⇒ nothing to measure (side-lane)
+        self.friction.record(FrictionRecord(
+            request_id=req.request_id, asset=req.asset, side=req.side, urgency=req.urgency,
+            maker=maker, guideline_px=ref, placed_px=ref, fill_px=ref, touch_px=ref,
+            size=req.size, fee_cost=Decimal(0), triage="placed",
+        ))
+
+    def _bump(self, field_name: str) -> None:
+        s = self.stats
+        if s is None:
+            return
+        try:
+            setattr(s, field_name, getattr(s, field_name, 0) + 1)
+            if field_name in ("accepted", "rejected"):
+                s.processed = getattr(s, "processed", 0) + 1
+        except Exception:
+            pass

prod/clean_arch/violet/uv/exec/test_smart_bridge.py

@@ -0,0 +1,193 @@
+"""SmartExecBridge (T1) tests — behaviour, not execution. Runs against a fake DITAv2 kernel
+and the real exec_unified router. Mutation-annotated where a wrong branch moves money.
+
+Home: prod/clean_arch/violet/uv/exec/. Needs prod.exec_unified + dita_v2 importable (both live
+in the repo tree). Run:
+  /home/dolphin/siloqy_env/bin/python3 -m pytest prod/clean_arch/violet/uv/exec/test_smart_bridge.py -q
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from decimal import Decimal
+from types import SimpleNamespace
+
+import pytest
+
+from prod.clean_arch.dita_v2.contracts import KernelCommandType, KernelIntent, TradeSide
+from prod.clean_arch.violet.uv.exec.smart_bridge import (
+    SmartExecBridge,
+    smart_exec_enabled,
+    urgency_for,
+)
+from prod.exec_unified.contract import Side, UrgencyClass
+
+
+# ── fakes ────────────────────────────────────────────────────────────────────
+
+class FakeGate:
+    def __init__(self, active=True):
+        self._active = active
+    def is_active(self):
+        return self._active
+
+
+class FakeKernel:
+    """Records processed intents; slot 0 is settable to mimic the FSM. process_intent returns
+    an OK outcome unless told to boom."""
+    def __init__(self, slot=None, boom=False):
+        self._slot = slot or SimpleNamespace(fsm_state=SimpleNamespace(value="IDLE"),
+                                             size=Decimal(0), trade_id="")
+        self.boom = boom
+        self.processed: list[KernelIntent] = []
+    def slot(self, i):
+        return self._slot
+    def process_intent(self, intent):
+        if self.boom:
+            raise RuntimeError("venue down")
+        self.processed.append(intent)
+        return SimpleNamespace(diagnostic_code=SimpleNamespace(value="OK"))
+    def set_slot(self, stage, size, trade_id=""):
+        self._slot = SimpleNamespace(fsm_state=SimpleNamespace(value=stage),
+                                     size=Decimal(str(size)), trade_id=trade_id)
+
+
+class Clock:
+    def __init__(self, t=1000.0): self.t = t
+    def __call__(self): return self.t
+
+
+def _naive_intent(*, decision, ctx=None, reason="uv_smart"):
+    """Stand-in for promotion.build_kernel_intent_from_decision — a naive MARKET intent."""
+    has_entry = bool(decision.get("has_entry"))
+    return KernelIntent(
+        timestamp=datetime(2026, 7, 15, tzinfo=timezone.utc),
+        intent_id=decision.get("intent_id", "uv-p-1"),
+        trade_id=decision.get("trade_id", "uv-trade-1"),
+        slot_id=0, asset=decision.get("asset", "BTCUSDT"),
+        side=TradeSide.LONG if decision.get("side", "LONG") == "LONG" else TradeSide.SHORT,
+        action=KernelCommandType.ENTER if has_entry else KernelCommandType.EXIT,
+        reference_price=float(decision.get("ref", 100.0)),
+        target_size=float(decision.get("size", 0.001)), leverage=1.0, reason=reason,
+    )
+
+
+def _bridge(**kw):
+    k = FakeKernel(**{a: kw.pop(a) for a in ("slot", "boom") if a in kw})
+    clock = kw.pop("clock", Clock())
+    gate = kw.pop("gate", FakeGate(active=True))
+    return SmartExecBridge(gate=gate, kernel=k, clock=clock, **kw), k, clock
+
+
+def _entry(**kw):
+    d = {"has_entry": True, "side": "LONG", "ref": 100.0, "size": 0.001}
+    d.update(kw); return d
+
+
+def _exit(**kw):
+    d = {"has_entry": False, "is_exit": True, "side": "LONG", "ref": 100.0, "size": 0.001}
+    d.update(kw); return d
+
+
+# ── the gate ─────────────────────────────────────────────────────────────────
+
+def test_gate_off_does_not_touch_the_kernel():
+    b, k, _ = _bridge(gate=FakeGate(active=False))
+    assert b.try_promote(decision=_entry(), build_intent=_naive_intent) is False
+    assert k.processed == []                 # gate closed ⇒ zero venue calls
+
+
+def test_env_flag_default_off():
+    # no UV_SMART_EXEC set ⇒ disabled (the flight uses the naive promote)
+    assert smart_exec_enabled() is False
+
+
+# ── THE T1 core: entries rest as maker, exits cross ──────────────────────────
+
+def test_entry_is_placed_as_postonly_limit_maker():
+    b, k, _ = _bridge()
+    assert b.try_promote(decision=_entry(), build_intent=_naive_intent) is True
+    sent = k.processed[-1]
+    assert sent.order_type == "LIMIT"                        # maker, not MARKET
+    assert sent.limit_price == 100.0                         # at the reference touch
+    assert sent.metadata["tif"] == "PostOnly"               # GTX
+    assert sent.metadata["uv_exec_tier"] == "T1"
+    # Mutation: urgency_for returns CATASTROPHIC for ENTER -> MARKET, order_type != LIMIT -> RED.
+
+
+def test_exit_stays_market_zero_new_risk():
+    b, k, _ = _bridge(slot=SimpleNamespace(fsm_state=SimpleNamespace(value="POSITION_OPEN"),
+                                           size=Decimal("1"), trade_id="uv-trade-1"))
+    assert b.try_promote(decision=_exit(), build_intent=_naive_intent) is True
+    sent = k.processed[-1]
+    assert sent.order_type == "MARKET"                       # exit unchanged — never strand
+    assert "tif" not in sent.metadata                        # no PostOnly on the cross
+
+
+def test_urgency_policy_entry_acquire_exit_catastrophic():
+    enter = _naive_intent(decision=_entry())
+    exit_ = _naive_intent(decision=_exit())
+    assert urgency_for(enter) is UrgencyClass.ACQUIRE
+    assert urgency_for(exit_) is UrgencyClass.CATASTROPHIC
+
+
+def test_entry_side_maps_long_to_buy():
+    b, k, _ = _bridge()
+    b.try_promote(decision=_entry(side="LONG"), build_intent=_naive_intent)
+    # friction row carries the mapped ORDER side; LONG entry = BUY
+    assert b.friction.rows()[-1].side is Side.BUY
+
+
+# ── the drive-clock: sweep a stale unfilled maker at the next tick ────────────
+
+def test_stale_unfilled_entry_is_swept_next_tick():
+    clock = Clock(1000.0)
+    b, k, _ = _bridge(clock=clock, entry_ttl_s=6.0)
+    b.try_promote(decision=_entry(), build_intent=_naive_intent)      # places resting maker
+    assert b._resting is not None
+    # slot still shows the entry working, unfilled; advance past TTL and promote again
+    k.set_slot("ENTRY_WORKING", 0, "uv-trade-1")
+    clock.t = 1010.0                                                   # +10s > 6s TTL
+    b.try_promote(decision={"has_entry": False}, build_intent=_naive_intent)  # idle tick
+    cancels = [i for i in k.processed if i.action is KernelCommandType.CANCEL]
+    assert len(cancels) == 1                                           # swept exactly once
+    assert b._resting is None
+    # Mutation: skip _sweep_stale_entry -> no CANCEL emitted -> RED.
+
+
+def test_filled_entry_is_not_swept():
+    clock = Clock(1000.0)
+    b, k, _ = _bridge(clock=clock, entry_ttl_s=6.0)
+    b.try_promote(decision=_entry(), build_intent=_naive_intent)
+    # the maker FILLED — slot now holds a position; a later tick must NOT cancel it
+    k.set_slot("POSITION_OPEN", Decimal("0.001"), "uv-trade-1")
+    clock.t = 1010.0
+    b.try_promote(decision={"has_entry": False}, build_intent=_naive_intent)
+    assert [i for i in k.processed if i.action is KernelCommandType.CANCEL] == []  # no spurious cancel
+    assert b._resting is None                                          # state cleared, not cancelled
+
+
+def test_fresh_entry_within_ttl_not_swept():
+    clock = Clock(1000.0)
+    b, k, _ = _bridge(clock=clock, entry_ttl_s=6.0)
+    b.try_promote(decision=_entry(), build_intent=_naive_intent)
+    k.set_slot("ENTRY_WORKING", 0, "uv-trade-1")
+    clock.t = 1003.0                                                   # +3s < 6s TTL
+    b.try_promote(decision={"has_entry": False}, build_intent=_naive_intent)
+    assert [i for i in k.processed if i.action is KernelCommandType.CANCEL] == []  # too soon to sweep
+
+
+# ── fail-soft: never raise into the scan loop ────────────────────────────────
+
+def test_try_promote_never_raises_on_kernel_error():
+    b, k, _ = _bridge(boom=True)
+    # kernel.process_intent raises — try_promote must swallow and return False (scan loop lives)
+    assert b.try_promote(decision=_entry(), build_intent=_naive_intent) is False
+
+
+# ── friction is emitted on the side lane ─────────────────────────────────────
+
+def test_friction_row_emitted_on_placement():
+    b, k, _ = _bridge()
+    b.try_promote(decision=_entry(), build_intent=_naive_intent)
+    rows = b.friction.rows()
+    assert len(rows) == 1 and rows[0].maker is True and rows[0].asset == "BTCUSDT"