PINK DITAv2 L0-L2: two-phase persistence + async-fill pump + LIMIT wiring

Execution-infra only (policy stays MARKET; algorithmic integrity untouched).

L0 — two-phase (request->result) persistence (pink_clickhouse.py):
- Split persist_step into persist_request (policy_events + trade_reconstruction
  ORDER_REQUESTED) and persist_result (state snapshot + per-fill lifecycle rows).
- Lifecycle rows (ENTRY_FILLED/EXIT/trade_events/trade_exit_legs) gated on
  evidence of an actual fill (FULL/PARTIAL_FILL event, closed slot, or size drop
  vs _leg_state) -> a resting LIMIT (ACK only) emits no terminal rows.
- Add persist_fill_events: synthesizes a minimal decision/intent from slot+event
  for async fills and routes through persist_result.

L1 — async-fill pump (pink_direct.py):
- PinkDirectRuntime.pump_venue_events(): venue.reconcile() -> kernel.on_venue_event
  (capital settles, FSM advances), persists applied fills; kernel dedups
  duplicates (no double-settle). Called at the start of step().

L2 — LIMIT placement (bingx_direct.py):
- submit_intent now honors _order_type/_limit_price from intent metadata
  (was hardcoded MARKET): LIMIT -> type=LIMIT + price + GTC; MARKET default;
  invalid limit price falls back to MARKET.

Offline: 63 passed (persistence/groundwork/pump/limit-payload/runtime/accounting/
flaws/kernel). MARKET path unchanged; resting LIMIT now correct end-to-end offline.
Live VST validation (L3) pending. BLUE untouched.

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

prod/clean_arch/persistence/pink_clickhouse.py

@@ -24,7 +24,8 @@ from enum import Enum
 from typing import Any, Callable, Mapping, Optional
 
 from prod.clean_arch.dita import AccountProjection, Decision, DecisionAction, Intent, TradeSide, TradeStage
-from prod.clean_arch.dita_v2.contracts import KernelDiagnosticCode, KernelOutcome
+from prod.clean_arch.dita_v2.contracts import KernelDiagnosticCode, KernelEventKind, KernelOutcome
+from prod.clean_arch.dita_v2.contracts import KernelSeverity, TradeStage as KernelStage
 
 Writer = Callable[[str, dict[str, Any]], None]
 
@@ -222,6 +223,71 @@ class PinkClickHousePersistence:
         phase: str = "step",
         market_state: Mapping[str, Any] | None = None,
     ) -> None:
+        """Two-phase persist: log the REQUEST, then log the RESULT.
+
+        REQUEST (:meth:`persist_request`) — the decision/order that was
+        submitted (policy_events + a trade_reconstruction ORDER_REQUESTED row).
+        RESULT (:meth:`persist_result`) — the settled state snapshot plus the
+        per-fill lifecycle rows, gated on *evidence of an actual fill*. A resting
+        LIMIT order (ACK only, no fill) therefore emits state snapshots but no
+        terminal rows; the async-fill pump persists those later via the same
+        result path. The synchronous-MARKET path is unchanged: its FILL event
+        (or the slot's filled/closed state) trips the same gate.
+        """
+        self.persist_request(
+            snapshot=snapshot, decision=decision, intent=intent,
+            phase=phase, market_state=market_state,
+        )
+        self.persist_result(
+            snapshot=snapshot, decision=decision, intent=intent, outcome=outcome,
+            slot_dict=slot_dict, phase=phase, market_state=market_state,
+        )
+
+    def persist_request(
+        self,
+        *,
+        snapshot: Any,
+        decision: Decision,
+        intent: Intent,
+        phase: str = "step",
+        market_state: Mapping[str, Any] | None = None,
+    ) -> None:
+        """Phase 1 — log the requested decision/order (no fill data)."""
+        self._write_policy_event(snapshot, decision, intent, phase=phase)
+        if decision.action in (DecisionAction.ENTER, DecisionAction.EXIT):
+            self._write_trade_reconstruction(
+                snapshot, intent.trade_id,
+                event_type="ORDER_REQUESTED",
+                event_id=f"{intent.trade_id}:request:{decision.action.value.lower()}",
+                payload={
+                    "decision": _decision_summary(decision),
+                    "intent": _intent_summary(intent),
+                    "market_state": _json_safe(market_state or {}),
+                },
+                market_state=market_state,
+            )
+
+    def persist_result(
+        self,
+        *,
+        snapshot: Any,
+        decision: Decision,
+        intent: Intent,
+        outcome: KernelOutcome | None = None,
+        slot_dict: dict[str, Any] | None = None,
+        phase: str = "step",
+        market_state: Mapping[str, Any] | None = None,
+    ) -> None:
+        """Phase 2 — log the settled state + per-fill lifecycle rows.
+
+        The state snapshot rows (account_events, position_state,
+        status_snapshots) always reflect the current slot. The lifecycle rows
+        (ENTRY_FILLED / PARTIAL_EXIT / EXIT / trade_events / trade_exit_legs) are
+        emitted only when a fill is *evidenced* — a FULL/PARTIAL_FILL event in
+        ``outcome.emitted_events``, a closed slot, or a slot whose size dropped
+        vs the last leg snapshot. A resting LIMIT (ACK only) emits no terminal
+        rows here.
+        """
         slot = slot_dict or {}
         stage = (
             TradeStage(decision.stage.value)
@@ -231,12 +297,10 @@ class PinkClickHousePersistence:
         )
         status = self._state_label(slot, phase)
 
-        self._write_policy_event(snapshot, decision, intent, phase=phase)
         self._write_account_event(snapshot, decision, intent, stage=stage, slot_dict=slot)
         self._write_position_state(snapshot, decision, intent, slot_dict=slot, stage=stage, status=status, market_state=market_state)
         self._write_status_snapshot(snapshot, decision, intent, slot_dict=slot, phase=phase)
 
-        # Emit anomaly for diagnostic codes (except OK).
         if outcome is not None and outcome.diagnostic_code != KernelDiagnosticCode.OK:
             self._write_anomaly(
                 snapshot, decision, intent,
@@ -246,38 +310,56 @@ class PinkClickHousePersistence:
             )
 
         if outcome is None:
-            # Decision-only step (HOLD, no execution).
+            # Decision-only step (HOLD): state snapshot already written.
             return
 
+        events = tuple(outcome.emitted_events or ())
+        has_fill_evt = any(
+            e.kind in (KernelEventKind.FULL_FILL, KernelEventKind.PARTIAL_FILL)
+            for e in events
+        )
+        slot_closed = bool(slot.get("closed", False))
+        cur_size = _safe_float(slot.get("size", 0.0), 0.0)
+        slot_open = (not slot_closed) and cur_size > 0.0
+
         if decision.action == DecisionAction.ENTER:
-            # Reset per-trade leg deltas: a fresh position starts with zero
-            # realized PnL and the full initial size remaining.
-            self._leg_state[intent.trade_id] = {
-                "prev_realized": 0.0,
-                "prev_size": _safe_float(
-                    slot.get("initial_size", slot.get("size", 0.0)), 0.0
-                ) or _safe_float(intent.target_size, 0.0),
-                "prev_leg_id": "",
-            }
-            self._write_trade_reconstruction(
-                snapshot, intent.trade_id,
-                event_type="ENTRY_FILLED",
-                event_id=f"{intent.trade_id}:entry",
-                payload={
-                    "decision": _decision_summary(decision),
-                    "intent": _intent_summary(intent),
-                    "outcome": _outcome_summary(outcome),
-                    "slot": slot,
-                    "market_state": _json_safe(market_state or {}),
-                },
-                market_state=market_state,
-            )
+            # Emit ENTRY_FILLED only once the entry is actually filled (fill event
+            # or an open slot). A resting LIMIT entry emits nothing here.
+            if has_fill_evt or slot_open:
+                self._leg_state[intent.trade_id] = {
+                    "prev_realized": 0.0,
+                    "prev_size": _safe_float(
+                        slot.get("initial_size", slot.get("size", 0.0)), 0.0
+                    ) or _safe_float(intent.target_size, 0.0),
+                    "prev_leg_id": "",
+                }
+                self._write_trade_reconstruction(
+                    snapshot, intent.trade_id,
+                    event_type="ENTRY_FILLED",
+                    event_id=f"{intent.trade_id}:entry",
+                    payload={
+                        "decision": _decision_summary(decision),
+                        "intent": _intent_summary(intent),
+                        "outcome": _outcome_summary(outcome),
+                        "slot": slot,
+                        "market_state": _json_safe(market_state or {}),
+                    },
+                    market_state=market_state,
+                )
             return
 
         if decision.action != DecisionAction.EXIT:
             return
 
-        partial = slot.get("closed", False) is False and slot.get("size", 0) > 0
+        # An exit leg is evidenced by a fill event, a closed slot, or a drop in
+        # remaining size vs the previous leg snapshot. A resting LIMIT exit (no
+        # size change) emits nothing until the async-fill pump observes the fill.
+        prev_size = _safe_float(self._leg_state.get(intent.trade_id, {}).get("prev_size", 0.0), 0.0)
+        exit_filled = has_fill_evt or slot_closed or (prev_size - cur_size > 1e-12)
+        if not exit_filled:
+            return
+
+        partial = (not slot_closed) and cur_size > 0.0
         # One trade_exit_legs row per exit leg (partial or final), BLUE-schema
         # compatible so PINK multi-exit trades reconcile against the same table.
         self._write_trade_exit_leg(snapshot, decision, intent, slot, outcome)
@@ -295,9 +377,63 @@ class PinkClickHousePersistence:
             market_state=market_state,
         )
         # Terminal trade event.
-        if slot.get("closed", False):
+        if slot_closed:
             self._write_trade_event(snapshot, decision, intent, slot, outcome, market_state=market_state)
 
+    def persist_fill_events(
+        self,
+        *,
+        snapshot: Any,
+        events: Any,
+        slot_dict: dict[str, Any] | None = None,
+        market_state: Mapping[str, Any] | None = None,
+    ) -> None:
+        """Persist a late (async) venue fill drained by the runtime pump.
+
+        There is no fresh policy decision for an async fill, so we synthesize a
+        minimal Decision/Intent from the post-fill slot + event and route it
+        through :meth:`persist_result`. Direction (ENTER vs EXIT) is inferred
+        from the slot: a closed slot or a drop in remaining size vs the last leg
+        snapshot is an EXIT; otherwise an opening fill is an ENTER. Capital
+        authority remains the kernel — this only logs the settled result.
+        """
+        slot = slot_dict or {}
+        event_list = tuple(events or ())
+        trade_id = str(slot.get("trade_id") or "")
+        asset = str(slot.get("asset") or "")
+        side = self._slot_side(slot)
+        closed = bool(slot.get("closed", False))
+        cur_size = self._slot_size(slot)
+        leverage = _safe_float(slot.get("leverage", 1.0), 1.0)
+        price = next((float(getattr(e, "price", 0.0) or 0.0) for e in event_list if getattr(e, "price", 0.0)), 0.0) or self._slot_entry_price(slot)
+        prev_size = _safe_float(self._leg_state.get(trade_id, {}).get("prev_size", 0.0), 0.0)
+        is_exit = closed or (prev_size > 0.0 and cur_size < prev_size - 1e-12)
+        action = DecisionAction.EXIT if is_exit else DecisionAction.ENTER
+        ts = getattr(snapshot, "timestamp", datetime.now(timezone.utc))
+
+        decision = Decision(
+            timestamp=ts, decision_id=trade_id or "async", asset=asset, action=action,
+            side=side, reason="ASYNC_FILL", confidence=0.0, velocity_divergence=0.0,
+            irp_alignment=0.0, reference_price=price, target_size=cur_size,
+            leverage=leverage, stage=TradeStage.POSITION_UPDATED, metadata={},
+        )
+        intent = Intent(
+            timestamp=ts, trade_id=trade_id, decision_id=trade_id or "async", asset=asset,
+            action=action, side=side, reason="ASYNC_FILL", target_size=cur_size,
+            leverage=leverage, reference_price=price, confidence=0.0,
+            exit_leg_ratios=tuple(slot.get("exit_leg_ratios", (1.0,)) or (1.0,)), metadata={},
+        )
+        outcome = KernelOutcome(
+            accepted=True, slot_id=int(slot.get("slot_id", 0) or 0), trade_id=trade_id,
+            state=KernelStage.CLOSED if closed else KernelStage.POSITION_OPEN,
+            diagnostic_code=KernelDiagnosticCode.OK, severity=KernelSeverity.INFO,
+            transitions=(), emitted_events=event_list, details={"origin": "async_fill_pump"},
+        )
+        self.persist_result(
+            snapshot=snapshot, decision=decision, intent=intent, outcome=outcome,
+            slot_dict=slot, phase="async_fill", market_state=market_state,
+        )
+
     def persist_recovery_state(
         self,
         *,