"""External-cancel + stale-watchdog execution tests (dcy-013 / issue #9). A programmable broker lets each poll return a chosen FillStatus so we can drive full external cancel (-> rollback to holding) and partial-fill-then-cancel (-> ledger record + qty decrement + holding remainder), plus the Alert-only pending stale watchdog. """ from __future__ import annotations import logging from dataclasses import replace from datetime import datetime, timedelta, timezone from trex_core.adapters.ledger.in_memory import InMemoryLedger from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.broker_port import ( BalanceItem, CancelOutcome, CancelState, FillStatus, OrderAck, ) from trex_core.application.execution import ExecutionService from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.application.ledger_outbox import InMemoryLedgerOutbox, LedgerWriter from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository class ProgrammableBroker: """Broker double whose poll_fill response is chosen per call by `responder`.""" def __init__(self): self._seq = 0 # default: full fill (used to drive a buy to holding) self.responder = lambda side: FillStatus( state="filled", filled_qty=10, avg_price=1000.0, execution_id="exec-full", fee=1.0, tax=0.0, ) self.polls = 0 self.balance = {"005930": 10} self.cancel_calls = 0 def name(self): return "stub" def submit_order(self, req): self._seq += 1 return OrderAck(accepted=True, broker_order_id=f"o{self._seq}") def poll_fill(self, broker_order_id, *, side, symbol): self.polls += 1 return self.responder(side) def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): self.cancel_calls += 1 return CancelOutcome(CancelState.REQUESTED) def get_balance(self): # M2 (dcy-020/REQ-062): sell dispatch reads the broker balance. return [BalanceItem(symbol=s, hldg_qty=q) for s, q in self.balance.items() if q > 0] def get_current_price(self, symbol): return None def _wire(broker, *, stale_sec=None): repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") outbox = InMemoryLedgerOutbox() ledger = InMemoryLedger() execution = ExecutionService( svc, repo, broker, ledger_outbox=outbox, stale_pending_threshold_sec=stale_sec, ) writer = LedgerWriter(outbox, ledger) repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents, execution, ledger, writer def _buy_to_holding(intents, execution): sid = intents.submit_buy(BuyIntent( intent_id="b1", qty=10, entry_price=1000, portfolio_id="PF-1", )).slot_id execution.run_once() # buy dispatched + full fill -> holding return sid def test_full_external_cancel_returns_slot_to_holding(): broker = ProgrammableBroker() repo, svc, intents, execution, ledger, writer = _wire(broker) sid = _buy_to_holding(intents, execution) assert repo.fetch_slot(sid).status is SlotState.HOLDING intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) broker.responder = lambda side: FillStatus(state="cancelled") # external cancel r = execution.run_once() # dispatch sell + poll(cancelled) -> rollback slot = repo.fetch_slot(sid) assert slot.status is SlotState.HOLDING # recovered, not stuck pending_sell assert slot.broker_order_id is None # dispatch cleared -> re-sellable assert r.rolled_back == 1 def test_partial_fill_then_cancel_records_and_keeps_remainder(): broker = ProgrammableBroker() repo, svc, intents, execution, ledger, writer = _wire(broker) sid = _buy_to_holding(intents, execution) writer.run_once() # drain the buy ledger item intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) # 4 of 10 sold, then externally cancelled. dcy-021/REQ-062: remainder qty is # the BROKER balance (authority) after the partial fill, not a slot subtraction. broker.balance = {"005930": 6} broker.responder = lambda side: FillStatus( state="partially_filled", filled_qty=4, avg_price=1100.0, execution_id="exec-partial", fee=0.5, tax=0.0, execution_venue="KRX", cancelled=True, ) r = execution.run_once() slot = repo.fetch_slot(sid) assert slot.status is SlotState.HOLDING # remainder stays held assert slot.qty == 6 # broker balance after partial assert slot.avg_price == 1000.0 # cost basis preserved (not sell price) assert slot.broker_order_id is None assert r.filled == 1 and r.ledger_enqueued == 1 writer.run_once() sells = [t for t in ledger.trades() if t.side == "sell"] assert len(sells) == 1 and sells[0].qty == 4 and sells[0].price == 1100.0 def test_stale_pending_watchdog_warns_without_state_change(caplog): broker = ProgrammableBroker() repo, svc, intents, execution, ledger, writer = _wire(broker, stale_sec=60) sid = _buy_to_holding(intents, execution) intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) # order sits accepted (never fills) broker.responder = lambda side: FillStatus(state="accepted") execution.run_once() # dispatch sell, poll -> accepted (fresh, not stale yet) # backdate the dispatched slot's updated_at beyond the threshold s = repo.fetch_slot(sid) repo._slots[sid] = replace(s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120)) with caplog.at_level(logging.WARNING): r = execution.run_once() assert r.stale_pending == 1 assert repo.fetch_slot(sid).status is SlotState.PENDING_SELL # state unchanged assert any("pending order stale" in rec.message for rec in caplog.records) def test_stale_watchdog_disabled_by_default(): broker = ProgrammableBroker() repo, svc, intents, execution, ledger, writer = _wire(broker) # no threshold sid = _buy_to_holding(intents, execution) intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) broker.responder = lambda side: FillStatus(state="accepted") execution.run_once() s = repo.fetch_slot(sid) repo._slots[sid] = replace(s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999)) r = execution.run_once() assert r.stale_pending == 0 # disabled => never warns def test_operator_cancel_is_sent_once_across_repeat_and_restart(): broker = ProgrammableBroker() repo, svc, intents, execution, _, _ = _wire(broker) sid = _buy_to_holding(intents, execution) intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) broker.responder = lambda side: FillStatus(state="accepted") execution.run_once() assert execution.cancel_pending_slot(sid) == "ok" assert execution.cancel_pending_slot(sid) == "cancel_already_recorded" restarted = ExecutionService(svc, repo, broker) assert restarted.cancel_pending_slot(sid) == "cancel_already_recorded" assert broker.cancel_calls == 1 assert repo.fetch_slot(sid).cancel_state == CancelState.REQUESTED.value # ── Regression: first-cancel reservation (retry_before=None) ───────────────── def test_first_cancel_reservation_with_no_retry_before_succeeds(): """First cancel reservation with retry_before=None must reserve exactly once (fix for PostgreSQL IndeterminateDatatype; the in-memory repo mirrors the corrected SQL path).""" broker = ProgrammableBroker() repo, svc, intents, execution, _, _ = _wire(broker) sid = _buy_to_holding(intents, execution) intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) broker.responder = lambda side: FillStatus(state="accepted") execution.run_once() # dispatch # First operator cancel — retry_before=None path result = execution.cancel_pending_slot(sid) assert result == "ok", f"expected ok, got {result}" assert broker.cancel_calls == 1 # Second call must be idempotent (cancel_state already set) result2 = execution.cancel_pending_slot(sid) assert result2 == "cancel_already_recorded" assert broker.cancel_calls == 1 # no second broker call def test_timeout_cancel_first_attempt_with_retry_before_none(): """Timeout cancel's first attempt passes retry_before=None; must call broker once.""" broker = ProgrammableBroker() repo, svc, intents, execution, _, _ = _wire(broker) sid = _buy_to_holding(intents, execution) intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) broker.responder = lambda side: FillStatus( state="accepted", raw={"order_open_confirmed": True}, ) execution.run_once() # dispatch # Simulate timeout: set order_submitted_at well in the past s = repo.fetch_slot(sid) repo._slots[sid] = replace( s, order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120), order_date="20260901", timeout_sec=30, ) execution.run_once() # should trigger timeout cancel assert broker.cancel_calls == 1 # ── Regression: concurrent timeout/operator cancel race — broker call count 1 ─ def test_timeout_operator_race_broker_called_exactly_once(): """Timeout cancel loop and operator cancel_pending_slot race to the same first reservation. The repository CAS guarantees exactly one winner; broker must receive cancel_order exactly once regardless of scheduling order. Uses a threading barrier so both paths enter reserve_cancel_attempt simultaneously, making the race deterministic. """ import threading broker = ProgrammableBroker() repo, _, intents, execution, _, _ = _wire(broker) sid = _buy_to_holding(intents, execution) intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) broker.responder = lambda side: FillStatus( state="accepted", raw={"order_open_confirmed": True}, ) execution.run_once() # dispatch sell # Backdate so timeout_cancel fires in the next run_once() s = repo.fetch_slot(sid) repo._slots[sid] = replace( s, order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120), order_date="20260901", timeout_sec=30, ) barrier = threading.Barrier(2) original_reserve = repo.reserve_cancel_attempt def _barrier_reserve(slot_id, broker_order_id, *, retry_before=None): barrier.wait(timeout=5) return original_reserve(slot_id, broker_order_id, retry_before=retry_before) repo.reserve_cancel_attempt = _barrier_reserve errors = [] def _operator(): try: execution.cancel_pending_slot(sid) except Exception as exc: # noqa: BLE001 errors.append(exc) t = threading.Thread(target=_operator) t.start() # Execution loop's timeout cancel is the second concurrent caller. try: execution.run_once() except Exception as exc: # noqa: BLE001 errors.append(exc) t.join(timeout=5) assert not errors, f"thread errors: {errors}" # CAS ensures exactly one winner — broker cancel must be called once. assert broker.cancel_calls == 1, ( f"expected exactly 1 broker cancel call, got {broker.cancel_calls}" )