"""Order execution service (REQ-013, REQ-014, REQ-018, dcy-003, dcy-004).

A single background execution loop (invariant A single writer, invariant B
single broker session) that, per tick:
  dispatch: claim pending slots with broker_order_id IS NULL (option A queue),
            submit the order, record broker_order_id (idempotency guard).
            dcy-020/REQ-062 (BAT-030): a SELL is dispatched with the broker
            balance quantity (hldg_qty) as the single source of truth — never
            the slot's mirror qty. Broker flat (qty 0) -> skip + WARN
            (divergence surfaces via reconcile, REQ-062 A).
  poll:     observe fills on dispatched pending slots and transition the slot
            (pending_buy->holding / pending_sell->inactive), or roll back on reject.
            On a successful fill, enqueue a LedgerWorkItem (slot transition is
            kept ON the critical path; the ledger write is eventually-consistent
            via a separate writer — REQ-019).
reconcile() re-polls dispatched slots on restart so a crash after dispatch
converges without duplicate orders (REQ-014).

All slot writes go through SlotService (advisory lock). The broker is touched
only here, never by the domain or the HTTP API. The ledger outbox is the
single producer for ledger work items (NFR-3, single record subject).
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from typing import Callable, Optional

from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.broker_port import (
    BalanceItem,
    BrokerPort,
    CancelOutcome,
    CancelState,
    OrderRequest,
)
from trex_core.application.ledger_outbox import LedgerOutbox
from trex_core.application.ledger_port import LedgerTrade
from trex_core.domain.pricing import ceil_to_tick, floor_to_tick
from trex_core.domain.slot import SlotService, SlotState, SlotStateMismatchError
from trex_core.runtime.logging_setup import set_correlation_id

_log = logging.getLogger(__name__)

# REQ-021/023: event emission hook. Runtime wires a callable that publishes
# slot transition events through MessagingTransport. Application layer stays
# transport-agnostic (NFR-1). Slot type is duck-typed to keep the application
# layer free of domain re-imports (the slot is just used for attribute access).
#
# The optional third argument `routing_strategy_id` is a hint for the routing
# key when the slot's own strategy_id has been cleared by the transition (e.g.
# PENDING_SELL -> INACTIVE clears strategy_id on the slot row — REQ-057
# _apply_transition_fields). The emitter MUST prefer this hint over
# `slot.strategy_id` so that slot_filled_sell events reach the correct
# per-strategy queue instead of being NO_ROUTEd to the DLQ.
SlotEventEmitter = Callable[
    [str, object, Optional[str]], None
]  # (event_type, slot, routing_strategy_id) -> None

_MAX_DISPATCH_PER_TICK = 64


@dataclass
class ExecutionResult:
    dispatched: int = 0
    filled: int = 0
    rolled_back: int = 0
    ledger_enqueued: int = 0  # REQ-019: ledger work items appended this tick
    ledger_skipped: int = 0  # REQ-018 AC-4: fills without portfolio_id
    stale_pending: int = 0  # dcy-013 / REQ-014 FR-3: pending orders over threshold
    sell_skipped_no_balance: int = 0  # dcy-020 / REQ-062: broker flat -> sell skipped
    timeout_cancelled: int = 0  # dcy-022 / REQ-069: cancel_order sent on timeout


class ExecutionService:
    def __init__(
        self,
        slot_service: SlotService,
        repo,
        broker: BrokerPort,
        ledger_outbox: Optional[LedgerOutbox] = None,
        event_emitter: Optional[SlotEventEmitter] = None,
        stale_pending_threshold_sec: Optional[int] = None,
        instance_key: str = "default",
        alarm_emitter: Optional[AlarmEmitter] = None,
        balance_reader: Optional[Callable[[], list[BalanceItem]]] = None,
    ) -> None:
        self._svc = slot_service
        self._repo = repo
        self._broker = broker
        self._ledger_outbox = ledger_outbox
        self._event_emitter = event_emitter
        # dcy-020/REQ-056 (BAT-030): dispatch/poll are instance-scoped so two
        # daemons sharing a DB never touch each other's orders.
        self._instance_key = instance_key
        # dcy-013 / REQ-014 (issue #9): Alert-only pending-order stale watchdog.
        # <=0 or None disables it. When set, a dispatched-pending slot whose
        # last update is older than the threshold is surfaced (WARN + counter)
        # without changing its state — the backstop for a cancel-detection miss.
        self._stale_threshold_sec = (
            stale_pending_threshold_sec
            if stale_pending_threshold_sec and stale_pending_threshold_sec > 0
            else None
        )
        # dcy-033 / REQ-100: best-effort alarm emitter (None = no alarms sent).
        self._alarm_emitter = alarm_emitter
        # dcy-041 / GitHub #45: pending-sell quantity may use a bounded
        # account snapshot only through this injected reader. Reconciliation
        # remains wired directly to the live broker.
        self._sell_dispatch_balance_reader = balance_reader or broker.get_balance

    # ------------------------------------------------------------
    def run_once(self) -> ExecutionResult:
        r = ExecutionResult()
        self._dispatch(r)
        self._poll(r)
        return r

    def reconcile(self) -> ExecutionResult:
        """Restart recovery: re-poll dispatched-but-pending slots (REQ-014).

        Undispatched pending slots (broker_order_id IS NULL) are simply picked
        up by the next dispatch, so a crash before dispatch causes no duplicate.
        """
        r = ExecutionResult()
        self._poll(r)
        return r

    # ------------------------------------------------------------
    def _dispatch(self, r: ExecutionResult) -> None:
        sell_balance: Optional[dict[str, int]] = None  # lazy, once per tick
        for slot in self._repo.list_undispatched_pending(
            self._instance_key, _MAX_DISPATCH_PER_TICK
        ):
            side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
            if side == "sell":
                # dcy-020 / REQ-062 (BAT-030, D3): the sell quantity is the
                # BROKER balance (hldg_qty) — the slot's mirror qty is never
                # trusted. Broker flat -> skip dispatch + WARN; the divergence
                # surfaces via reconcile_core_holdings (REQ-062 A). No mirror
                # fallback by design.
                if sell_balance is None:
                    sell_balance = self._sell_dispatch_balance_map()
                bal = sell_balance.get(slot.symbol or "", 0)
                # dcy-021 / REQ-062 (D4): full sell (slot.qty is None) => broker
                # balance; partial sell => min(order qty, broker balance). Broker
                # flat -> skip + WARN (divergence surfaces via reconcile).
                qty = bal if slot.qty is None else min(int(slot.qty), bal)
                if qty <= 0:
                    r.sell_skipped_no_balance += 1
                    _log.warning(
                        "sell dispatch skipped: slot=%s symbol=%s broker balance is 0 "
                        "(divergence — no order submitted; operator/reconcile attention)",
                        slot.slot_id,
                        slot.symbol,
                    )
                    # dcy-033 / REQ-100: CRITICAL alarm — PENDING_SELL may stick.
                    if self._alarm_emitter is not None:
                        self._alarm_emitter.publish_alarm(
                            "sell_skipped_no_balance",
                            f"매도 dispatch skip: 브로커 잔고 0 (slot={slot.slot_id} symbol={slot.symbol})",
                            fields={"slot_id": str(slot.slot_id), "symbol": slot.symbol},
                        )
                    continue
            else:
                qty = int(slot.qty or 0)
            # dcy-012 / REQ-013: side-agnostic — use the slot's pending-order
            # limit_price only. A sell never reuses a buy limit (issue #10):
            # limit_price is cleared on entering HOLDING, so a market sell
            # carries None -> order_type "01".
            # dcy-022 / REQ-070: slippage_pct adjusts the effective limit price
            # when entry_price is set. No slippage when limit is None (market
            # order — no reference price).
            # dcy-030 / REQ-070 CHG-20260616-01: snap the slippage limit to a
            # valid KRX tick (REQ-083). Direction = cap-respecting: buy floors
            # to tick (never exceeds the slippage ceiling), sell ceils to tick
            # (never breaches the floor). A band<1-tick guard keeps the order
            # fillable at the entry's own valid tick (no under-fill).
            limit = int(slot.limit_price) if slot.limit_price else None
            if limit is not None and slot.slippage_pct:
                base = float(slot.limit_price)
                if side == "buy":
                    raw = base * (1.0 + slot.slippage_pct / 100.0)
                    limit = max(floor_to_tick(raw), ceil_to_tick(base))
                else:
                    raw = base * (1.0 - slot.slippage_pct / 100.0)
                    limit = min(ceil_to_tick(raw), floor_to_tick(base))
                _log.debug(
                    "slippage applied: slot=%s side=%s base_limit=%s pct=%.4f "
                    "raw=%.2f effective=%s (tick-snapped)",
                    slot.slot_id,
                    side,
                    slot.limit_price,
                    slot.slippage_pct,
                    raw,
                    limit,
                )
            # REQ-084/085: correlation id (intent→order→fill join key = slot_id)
            # so submit/fill/reject logs for one slot share a key.
            set_correlation_id(str(slot.slot_id))
            ack = self._broker.submit_order(
                OrderRequest(
                    slot_id=str(slot.slot_id),
                    side=side,
                    symbol=slot.symbol or "",
                    qty=qty,
                    limit_price=limit,
                    order_type="00" if limit is not None else "01",
                    venue=slot.venue,
                )
            )
            if ack.accepted and ack.broker_order_id:
                self._repo.mark_dispatched(
                    slot.slot_id,
                    ack.broker_order_id,
                    venue=ack.venue,
                )
                r.dispatched += 1
                if ack.reason and ack.reason.startswith("submit_outcome_unknown:"):
                    _log.warning(
                        "order submission outcome unknown event=order_submit_unknown "
                        "side=%s symbol=%s slot_id=%s qty=%s limit_price=%s "
                        "automatic_retry=blocked reason=%s",
                        side,
                        slot.symbol,
                        slot.slot_id,
                        qty,
                        limit,
                        ack.reason,
                    )
                else:
                    # REQ-085: order-result logging — tailable submit line.
                    _log.info(
                        "order submitted event=order_submit side=%s symbol=%s slot_id=%s "
                        "qty=%s limit_price=%s order_type=%s",
                        side,
                        slot.symbol,
                        slot.slot_id,
                        qty,
                        limit,
                        "00" if limit is not None else "01",
                    )
            else:
                # REQ-085/087: reject is visible (event key) and ERROR level
                # (no more silent rollback — issue #28 surfaced too late).
                _log.error(
                    "order rejected event=order_reject side=%s symbol=%s slot_id=%s "
                    "qty=%s reason=%s",
                    side,
                    slot.symbol,
                    slot.slot_id,
                    qty,
                    ack.reason or "order_rejected",
                )
                self._rollback(slot, reason=ack.reason or "order_rejected")
                r.rolled_back += 1

    def _poll(self, r: ExecutionResult) -> None:
        now = datetime.now(tz=timezone.utc)
        for slot in self._repo.list_dispatched_pending(self._instance_key):
            side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
            set_correlation_id(str(slot.slot_id))  # REQ-084: same key as submit
            fs = self._broker.poll_fill(slot.broker_order_id, side=side, symbol=slot.symbol or "")
            resolved_order_id = (
                fs.raw.get("resolved_broker_order_id") if isinstance(fs.raw, dict) else None
            )
            if resolved_order_id and resolved_order_id != slot.broker_order_id:
                if not self._repo.replace_dispatch(
                    slot.slot_id,
                    str(slot.broker_order_id),
                    str(resolved_order_id),
                ):
                    _log.error(
                        "resolved broker order id persistence failed "
                        "event=order_submit_resolution_conflict slot_id=%s symbol=%s",
                        slot.slot_id,
                        slot.symbol,
                    )
                    continue
                slot = replace(slot, broker_order_id=str(resolved_order_id))
                _log.info(
                    "order submission resolved event=order_submit_resolved "
                    "side=%s symbol=%s slot_id=%s",
                    side,
                    slot.symbol,
                    slot.slot_id,
                )
            if fs.is_filled:
                self._on_filled(slot, fs, side, r)
                r.filled += 1
            elif fs.is_partial_cancel:
                # dcy-013 / REQ-013 FR-2: partial fill then external cancel.
                self._on_partial_cancel(slot, fs, side, r)
                r.filled += 1
            elif fs.is_terminal_reject:
                # rejected or full external cancel (dcy-013 FR-1) -> rollback.
                # REQ-085/087: surface the reject (event key + ERROR level).
                _log.error(
                    "order rejected event=order_reject side=%s symbol=%s slot_id=%s "
                    "reason=order_%s",
                    side,
                    slot.symbol,
                    slot.slot_id,
                    fs.state,
                )
                self._rollback(slot, reason=f"order_{fs.state}")
                r.rolled_back += 1
            else:
                # accepted / partially_filled (not cancelled) -> leave for next
                # tick, but surface if it has been stuck too long (FR-3), and
                # send a cancel if the slot-level timeout is exceeded (REQ-069).
                self._check_stale_pending(slot, now, r)
                self._try_timeout_cancel(slot, now, side, r)

    # ------------------------------------------------------------
    def _balance_map(self) -> dict:
        """Current broker balance as {symbol: hldg_qty} (REQ-062 ground truth)."""
        return {b.symbol: int(b.hldg_qty) for b in self._broker.get_balance()}

    def _sell_dispatch_balance_map(self) -> dict:
        """Balance used only to size a not-yet-submitted pending sell."""
        return {
            b.symbol: int(b.hldg_qty)
            for b in self._sell_dispatch_balance_reader()
        }

    def _resolve_fill_price(self, fs, slot) -> tuple[float | None, bool]:
        """Resolve the fill price from FillStatus, with limit_price backfill.

        REQ-095 / REQ-096: KIS may return avg_price=None or 0.0 on a snapshot
        race. Rather than silently storing 0 we try the order's limit_price as
        a reasonable approximation, and surface the substitution via WARN so
        ops can audit the cost-basis estimate.

        Returns:
            (fill_price, was_backfilled)
            fill_price=None  → skip signal: no resolvable price (strong WARN
                               already emitted). Caller must not transition slot
                               or write ledger — slot stays PENDING so the
                               watchdog / stale_pending path fires for ops.
        """
        avg = fs.avg_price
        if avg is not None and float(avg) > 0:
            return float(avg), False
        # avg_price None or ≤0: attempt limit_price backfill (REQ-095 AC-1/2)
        limit = slot.limit_price
        if limit is not None and float(limit) > 0:
            _log.warning(
                "fill price backfill: slot=%s symbol=%s broker_order_id=%s "
                "avg_price=%s backfill_price=%s reason=avg_price_backfill",
                slot.slot_id,
                slot.symbol,
                slot.broker_order_id,
                avg,
                limit,
            )
            return float(limit), True
        # limit_price also absent/≤0: skip + strong WARN (REQ-095 AC-6 / REQ-096 AC-2)
        # Slot stays PENDING — watchdog/stale_pending fires for operator attention.
        _log.warning(
            "fill price unresolvable: slot=%s symbol=%s broker_order_id=%s "
            "avg_price=%s limit_price=%s reason=fill_price_unresolvable "
            "(slot stays PENDING — operator attention required)",
            slot.slot_id,
            slot.symbol,
            slot.broker_order_id,
            avg,
            limit,
        )
        # dcy-033 / REQ-100: CRITICAL alarm — slot stuck in PENDING.
        if self._alarm_emitter is not None:
            self._alarm_emitter.publish_alarm(
                "fill_price_unresolvable",
                f"체결가 미해결: 슬롯 PENDING 고착 위험 (slot={slot.slot_id} symbol={slot.symbol})",
                fields={
                    "slot_id": str(slot.slot_id),
                    "symbol": slot.symbol,
                    "broker_order_id": slot.broker_order_id,
                },
            )
        return None, False

    def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None:
        broker_order_id = slot.broker_order_id  # captured BEFORE clear_dispatch
        # Capture strategy_id BEFORE the transition clears it.
        # PENDING_SELL -> INACTIVE clears strategy_id on the slot row
        # (_apply_transition_fields REQ-057), so we must capture it here to
        # supply as a routing hint for the slot_filled_sell event — otherwise the
        # routing key uses "_" and the message is NO_ROUTEd to the DLQ instead
        # of reaching trex.events.strategy.{strategy_id} (M2 REL-009 feedback).
        routing_strategy_id = slot.strategy_id
        # REQ-095: resolve fill price with limit_price backfill guard.
        fill_price, _was_backfilled = self._resolve_fill_price(fs, slot)
        if fill_price is None:
            return  # skip: unresolvable price — WARN already emitted, slot stays PENDING
        try:
            if slot.status is SlotState.PENDING_BUY:
                # dcy-021/REQ-066: held qty = broker balance (authority, includes
                # the just-filled increase); avg_price = weighted average. For an
                # initial buy slot.avg_price is None (=0) so new_avg = fill price.
                total = self._balance_map().get(slot.symbol or "", 0)
                filled = float(fs.filled_qty)
                new_qty = float(total) if total > 0 else filled
                old_qty = max(0.0, new_qty - filled)
                old_avg = float(slot.avg_price) if slot.avg_price else 0.0
                new_avg = (
                    (old_qty * old_avg + filled * fill_price) / new_qty
                    if new_qty > 0
                    else fill_price
                )
                updated = self._svc.transit_slot(
                    slot.slot_id,
                    SlotState.PENDING_BUY,
                    SlotState.HOLDING,
                    reason="buy_filled",
                    actor="execution_loop",
                    avg_price=new_avg,
                    qty=new_qty,
                )
                event_type = "slot_filled_buy"
            else:  # PENDING_SELL — dcy-021/REQ-066: remainder>0 stays holding,
                #     full exit -> inactive. Held qty = broker balance.
                total = self._balance_map().get(slot.symbol or "", 0)
                if total > 0:
                    updated = self._svc.transit_slot(
                        slot.slot_id,
                        SlotState.PENDING_SELL,
                        SlotState.HOLDING,
                        reason="sell_filled_partial",
                        actor="execution_loop",
                        qty=float(total),  # avg_price (cost basis) preserved
                    )
                else:
                    updated = self._svc.transit_slot(
                        slot.slot_id,
                        SlotState.PENDING_SELL,
                        SlotState.INACTIVE,
                        reason="sell_filled",
                        actor="execution_loop",
                    )
                event_type = "slot_filled_sell"
        except SlotStateMismatchError:
            return  # slot moved concurrently; next tick reconciles
        # REQ-085: order-result logging — tailable fill line (price/qty/fee/tax).
        _log.info(
            "order filled event=order_fill side=%s symbol=%s slot_id=%s "
            "fill_price=%s filled_qty=%s fee=%s tax=%s",
            side,
            slot.symbol,
            slot.slot_id,
            fs.avg_price,
            fs.filled_qty,
            getattr(fs, "fee", None),
            getattr(fs, "tax", None),
        )
        # REQ-018: enqueue ledger work item AFTER the slot transition is durable.
        # Slot transition is the critical path; ledger write is async (REQ-019).
        self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price)
        # REQ-021/023: emit event AFTER transition + ledger enqueue (single
        # record subject). Best-effort: a publish failure must not roll back.
        # Pass routing_strategy_id so sell fills route correctly even after
        # strategy_id is cleared from the slot row (slot_filled_sell NO_ROUTE fix).
        self._emit_event(event_type, updated, routing_strategy_id)
        self._repo.clear_dispatch(slot.slot_id)  # done with this order

    def _on_partial_cancel(self, slot, fs, side: str, r: ExecutionResult) -> None:
        """Partial fill then external cancel (dcy-013 / REQ-013 FR-2, OQ-1 A).

        Record the filled portion to the ledger ONCE, decrement the held qty,
        and return the slot to holding with the remainder — so the remaining
        position can be re-sold and no over-sell / double-record occurs (NFR-3).
        avg_price (cost basis) is preserved on a sell; a buy keeps only the
        portion that actually filled.
        """
        broker_order_id = slot.broker_order_id  # captured BEFORE clear_dispatch
        # Capture strategy_id BEFORE the transition (same reason as _on_filled:
        # PENDING_SELL -> INACTIVE clears strategy_id, causing NO_ROUTE).
        routing_strategy_id = slot.strategy_id
        # REQ-095: resolve fill price with limit_price backfill guard.
        fill_price, _was_backfilled = self._resolve_fill_price(fs, slot)
        if fill_price is None:
            return  # skip: unresolvable price — WARN already emitted, slot stays PENDING
        filled = float(fs.filled_qty)
        try:
            if slot.status is SlotState.PENDING_BUY:
                # dcy-021: held qty = broker balance; avg_price weighted average
                # (consistent with _on_filled). Only the filled portion recorded.
                total = self._balance_map().get(slot.symbol or "", 0)
                new_qty = float(total) if total > 0 else filled
                old_qty = max(0.0, new_qty - filled)
                old_avg = float(slot.avg_price) if slot.avg_price else 0.0
                new_avg = (
                    (old_qty * old_avg + filled * fill_price) / new_qty
                    if new_qty > 0
                    else fill_price
                )
                updated = self._svc.transit_slot(
                    slot.slot_id,
                    SlotState.PENDING_BUY,
                    SlotState.HOLDING,
                    reason="buy_partial_cancel",
                    actor="execution_loop",
                    avg_price=new_avg,
                    qty=new_qty,
                )
                event_type = "slot_filled_buy"
            else:  # PENDING_SELL: remainder = broker balance, cost basis unchanged.
                total = self._balance_map().get(slot.symbol or "", 0)
                if total > 0:
                    updated = self._svc.transit_slot(
                        slot.slot_id,
                        SlotState.PENDING_SELL,
                        SlotState.HOLDING,
                        reason="sell_partial_cancel",
                        actor="execution_loop",
                        qty=float(total),
                    )
                else:
                    updated = self._svc.transit_slot(
                        slot.slot_id,
                        SlotState.PENDING_SELL,
                        SlotState.INACTIVE,
                        reason="sell_partial_cancel",
                        actor="execution_loop",
                    )
                event_type = "slot_filled_sell"
        except SlotStateMismatchError:
            return  # slot moved concurrently; next tick reconciles
        self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price)
        self._emit_event(event_type, updated, routing_strategy_id)
        self._repo.clear_dispatch(slot.slot_id)

    def _check_stale_pending(self, slot, now: datetime, r: ExecutionResult) -> None:
        """Alert-only: surface a dispatched-pending slot stuck past the threshold.

        dcy-013 / REQ-014 FR-3 — state is NOT changed (auto-remediation is a
        follow-up). This is the backstop for a cancel-detection miss (R-1).
        """
        if self._stale_threshold_sec is None or slot.updated_at is None:
            return
        age = (now - slot.updated_at).total_seconds()
        if age >= self._stale_threshold_sec:
            r.stale_pending += 1
            _log.warning(
                "pending order stale: slot=%s status=%s broker_order_id=%s "
                "age_sec=%.0f threshold_sec=%d (no state change — ops attention)",
                slot.slot_id,
                slot.status.value,
                slot.broker_order_id,
                age,
                self._stale_threshold_sec,
            )

    def _try_timeout_cancel(self, slot, now: datetime, side: str, r: ExecutionResult) -> None:
        """Send cancel_order when the slot-level timeout is exceeded (dcy-022/REQ-069).

        Best-effort: a failure only logs a warning — the slot is NOT changed here.
        Actual rollback happens on the next poll tick when the broker confirms the
        cancel (is_terminal_reject) or partial fill then cancel (is_partial_cancel).
        This design is safe across the cancel-fill race: if the cancel races with a
        fill, the next poll sees is_filled and follows the normal fill path (FR-4).
        """
        if str(slot.broker_order_id or "").startswith("submission-unknown:"):
            return
        if slot.cancel_state is not None:
            return
        if slot.timeout_sec is None or slot.updated_at is None:
            return
        age = (now - slot.updated_at).total_seconds()
        if age < slot.timeout_sec:
            return
        qty = int(slot.qty or 0)
        # Write UNKNOWN before external I/O. If the process dies after KIS
        # receives the request but before the response is persisted, restart
        # polls the order instead of issuing a blind second cancellation.
        if not self._repo.mark_cancel_outcome(
            slot.slot_id,
            str(slot.broker_order_id),
            CancelState.UNKNOWN.value,
            only_if_unset=True,
        ):
            _log.error(
                "cancel attempt reservation conflict: slot=%s broker_order_id=%s",
                slot.slot_id,
                slot.broker_order_id,
            )
            return
        try:
            cancel_kwargs = {
                "side": side,
                "symbol": slot.symbol or "",
                "qty": qty,
            }
            if slot.venue != "KRX":
                cancel_kwargs["venue"] = slot.venue
            outcome = self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs)
        except Exception:  # noqa: BLE001 - adapter failure becomes durable unknown
            outcome = CancelOutcome(CancelState.UNKNOWN, "adapter_exception")
            _log.warning(
                "timeout cancel outcome unknown: slot=%s broker_order_id=%s",
                slot.slot_id,
                slot.broker_order_id,
            )
        persisted = self._repo.mark_cancel_outcome(
            slot.slot_id,
            str(slot.broker_order_id),
            outcome.state.value,
        )
        if not persisted:
            _log.error(
                "cancel outcome persistence conflict: slot=%s broker_order_id=%s "
                "outcome=%s",
                slot.slot_id,
                slot.broker_order_id,
                outcome.state.value,
            )
            return
        if outcome.is_requested:
            r.timeout_cancelled += 1
        _log.warning(
            "timeout cancel result: slot=%s status=%s broker_order_id=%s "
            "age_sec=%.0f timeout_sec=%d outcome=%s",
            slot.slot_id,
            slot.status.value,
            slot.broker_order_id,
            age,
            slot.timeout_sec,
            outcome.state.value,
        )

    def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None:
        if self._event_emitter is None:
            return
        try:
            self._event_emitter(event_type, slot, routing_strategy_id)
        except Exception:  # noqa: BLE001 - event publish failure is non-critical
            return

    def _record_ledger(
        self,
        slot,
        fs,
        side: str,
        broker_order_id: str,
        r: ExecutionResult,
        *,
        resolved_price: float | None = None,
    ) -> None:
        """Append a LedgerWorkItem for the just-filled trade (REQ-018 + REQ-019).

        resolved_price: pre-computed fill price from _resolve_fill_price (REQ-095 KD-3).
            When provided, it takes priority over fs.avg_price so that a backfilled
            limit_price is consistently used in both the slot avg and the ledger record.
            Defaults to None for backward compatibility (falls back to fs.avg_price).
        """
        if self._ledger_outbox is None:
            return
        if not slot.portfolio_id:
            # dcy-011 / REQ-018 (issue #11): ledger skip must NOT be silent.
            # Slot transition still proceeded; surface for ops (FR-4).
            r.ledger_skipped += 1
            _log.warning(
                "ledger skip: slot=%s symbol=%s side=%s reason=missing_portfolio_id "
                "(trade NOT recorded to portfolio-hub)",
                slot.slot_id,
                slot.symbol,
                side,
            )
            return
        if fs.execution_id is None:
            # Broker contract regression — slot already transitioned; surface for ops.
            r.ledger_skipped += 1
            _log.warning(
                "ledger skip: slot=%s portfolio_id=%s side=%s reason=missing_execution_id "
                "(trade NOT recorded to portfolio-hub)",
                slot.slot_id,
                slot.portfolio_id,
                side,
            )
            return
        # REQ-095 KD-3: use pre-resolved price when available; fall back to
        # fs.avg_price for any direct callers that do not pass resolved_price.
        price = (
            resolved_price
            if resolved_price is not None
            else (float(fs.avg_price) if fs.avg_price is not None else 0.0)
        )
        trade = LedgerTrade(
            asset_id=str(slot.portfolio_id),
            order_id=str(broker_order_id),
            execution_id=str(fs.execution_id),
            side=side,
            qty=float(fs.filled_qty),
            price=price,
            fee=float(fs.fee),
            tax=float(fs.tax),
            executed_at=datetime.now(tz=timezone.utc),
            venue=slot.venue,
        )
        self._ledger_outbox.enqueue(trade)
        r.ledger_enqueued += 1

    def cancel_pending_slot(self, slot_id) -> str:
        """Operator backstop: send cancel_order for a pending slot (dcy-022/REQ-071).

        Returns "ok" (request accepted), "cancel_rejected", "cancel_unknown",
        "not_pending", "not_dispatched", or "slot_not_found". Does NOT
        transition the slot — fill polling alone confirms cancelled/filled.
        """
        from uuid import UUID

        sid = slot_id if isinstance(slot_id, UUID) else UUID(str(slot_id))
        slot = self._repo.fetch_slot(sid)
        if slot is None:
            return "slot_not_found"
        if slot.status not in (SlotState.PENDING_BUY, SlotState.PENDING_SELL):
            return "not_pending"
        if not slot.broker_order_id:
            # Claimed but not yet dispatched — execution loop will pick up or rollback.
            return "not_dispatched"
        if slot.cancel_state is not None:
            return "cancel_already_recorded"
        side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
        qty = int(slot.qty or 0)
        try:
            cancel_kwargs = {
                "side": side,
                "symbol": slot.symbol or "",
                "qty": qty,
            }
            if slot.venue != "KRX":
                cancel_kwargs["venue"] = slot.venue
            # Durable unknown reservation closes the crash window between the
            # outbound request and saving its immediate result.
            if not self._repo.mark_cancel_outcome(
                slot.slot_id,
                str(slot.broker_order_id),
                CancelState.UNKNOWN.value,
                only_if_unset=True,
            ):
                return "cancel_already_recorded"
            outcome = self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs)
        except Exception:  # noqa: BLE001 - adapter failure becomes durable unknown
            outcome = CancelOutcome(CancelState.UNKNOWN, "adapter_exception")
            _log.warning(
                "operator cancel outcome unknown: slot=%s broker_order_id=%s",
                slot.slot_id,
                slot.broker_order_id,
            )
        if not self._repo.mark_cancel_outcome(
            slot.slot_id,
            str(slot.broker_order_id),
            outcome.state.value,
        ):
            return "cancel_state_conflict"
        _log.warning(
            "operator cancel result: slot=%s status=%s broker_order_id=%s outcome=%s",
            slot.slot_id,
            slot.status.value,
            slot.broker_order_id,
            outcome.state.value,
        )
        if outcome.state is CancelState.REQUESTED:
            return "ok"
        if outcome.state is CancelState.REJECTED:
            return "cancel_rejected"
        return "cancel_unknown"

    def _rollback(self, slot, *, reason: str) -> None:
        total = self._balance_map().get(slot.symbol or "", 0)
        try:
            if slot.status is SlotState.PENDING_BUY:
                if total > 0:
                    # dcy-021: increase rejected — the prior holding is intact
                    # (broker still holds), so return to holding, not inactive.
                    self._svc.transit_slot(
                        slot.slot_id,
                        SlotState.PENDING_BUY,
                        SlotState.HOLDING,
                        reason=reason,
                        actor="execution_loop",
                        qty=float(total),
                    )
                    self._repo.clear_dispatch(slot.slot_id)
                else:  # initial buy rejected -> inactive (clears broker_order_id)
                    self._svc.transit_slot(
                        slot.slot_id,
                        SlotState.PENDING_BUY,
                        SlotState.INACTIVE,
                        reason=reason,
                        actor="execution_loop",
                    )
            else:  # PENDING_SELL rejected -> back to holding (broker authoritative qty)
                self._svc.transit_slot(
                    slot.slot_id,
                    SlotState.PENDING_SELL,
                    SlotState.HOLDING,
                    reason=reason,
                    actor="execution_loop",
                    qty=float(total) if total > 0 else None,
                )
                self._repo.clear_dispatch(slot.slot_id)
        except SlotStateMismatchError:
            return