diff --git a/tests/application/test_messaging_event_publisher.py b/tests/application/test_messaging_event_publisher.py index ecd57a8..03a3bd5 100644 --- a/tests/application/test_messaging_event_publisher.py +++ b/tests/application/test_messaging_event_publisher.py @@ -9,7 +9,7 @@ from trex_core.adapters.messaging import topology from trex_core.adapters.messaging.in_memory import InMemoryTransport from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.execution import ExecutionService -from trex_core.application.intents import BuyIntent, IntentService +from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.application.ledger_outbox import InMemoryLedgerOutbox, LedgerWriter from trex_core.application.messaging_handlers import slot_transition_event_payload from trex_core.domain.slot import SlotService, SlotState @@ -27,8 +27,11 @@ def _wire(transport): writer = LedgerWriter(outbox, ledger) repo.reconcile_universe("default", {"PF-1": "SYM"}) - def emitter(event_type: str, slot) -> None: - rk = topology.event_routing_key(event_type, slot.strategy_id or "_", slot.symbol or "_") + def emitter(event_type: str, slot, routing_strategy_id: str | None = None) -> None: + # Use routing_strategy_id hint when supplied (e.g. sell fills clear + # slot.strategy_id via INACTIVE transition — NO_ROUTE fix). + effective_sid = routing_strategy_id or getattr(slot, "strategy_id", None) or "_" + rk = topology.event_routing_key(event_type, effective_sid, slot.symbol or "_") payload = slot_transition_event_payload(slot, event_type=event_type) try: transport.publish_event(rk, payload) @@ -124,3 +127,94 @@ def test_publish_failure_does_not_break_execution_loop(): assert r.ledger_enqueued == 1 pending = [s for s in repo.list_slots("default") if s.status is SlotState.HOLDING] assert len(pending) == 1 + + +def test_sell_fill_emits_slot_filled_sell_with_correct_strategy_routing(): + """Regression: slot_filled_sell must route to strategy queue even after + PENDING_SELL->INACTIVE clears strategy_id from the slot row. + + M2 REL-009 feedback: slot_filled_sell was NO_ROUTEd (DLQ) because + _on_filled used the post-transition slot (strategy_id=None) for the + routing key, producing `slot_filled_sell._.{symbol}` which does not match + the `*.{strategy_id}.#` binding pattern of trex.events.strategy.{sid}. + """ + transport = InMemoryTransport() + received_sell: list[tuple[str, dict]] = [] + stop = threading.Event() + th = threading.Thread( + target=transport.consume_events_forever, + args=("trex.events.strategy.A", topology.strategy_event_binding_pattern("A"), + lambda rk, p: received_sell.append((rk, p)), stop), + daemon=True, + ) + th.start() + time.sleep(0.05) + + repo, svc, intents, execution, _ = _wire(transport) + # Buy with strategy_id="A" + d = intents.submit_buy(BuyIntent( + intent_id="b1", qty=10, strategy_id="A", portfolio_id="PF-1", + )) + execution.run_once() # buy fill -> holding + slot = repo.fetch_slot(d.slot_id) + assert slot.status is SlotState.HOLDING + assert slot.strategy_id == "A" + + # Sell — strategy_id_for_emit must be captured before INACTIVE clears it. + intents.submit_sell(SellIntent(intent_id="s1", slot_id=d.slot_id)) + execution.run_once() # sell fill -> inactive + + time.sleep(0.05) + stop.set() + th.join(timeout=1.0) + + # The slot_filled_sell event must reach the strategy A queue. + sell_events = [(rk, p) for rk, p in received_sell if "sell" in rk] + assert len(sell_events) == 1, ( + f"Expected 1 slot_filled_sell event on strategy-A queue, got {len(sell_events)}. " + f"All received: {[rk for rk, _ in received_sell]}" + ) + rk, payload = sell_events[0] + # Routing key must use strategy_id="A", not "_" (NO_ROUTE sentinel). + assert rk == "slot_filled_sell.A.005930", ( + f"NO_ROUTE regression: routing key is '{rk}', expected 'slot_filled_sell.A.005930'" + ) + assert payload["event_type"] == "slot_filled_sell" + + +def test_sell_fill_dlq_sentinel_absent_when_strategy_id_set(): + """Routing key must NOT contain '_' as strategy_id token when a real + strategy_id was set on the slot (no DLQ sentinel in the key).""" + transport = InMemoryTransport() + all_published: list[str] = [] + + def emitter(event_type: str, slot, routing_strategy_id: str | None = None) -> None: + effective_sid = routing_strategy_id or getattr(slot, "strategy_id", None) or "_" + rk = topology.event_routing_key(event_type, effective_sid, slot.symbol or "_") + all_published.append(rk) + payload = slot_transition_event_payload(slot, event_type=event_type) + try: + transport.publish_event(rk, payload) + except Exception: # noqa: BLE001 + pass + + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService(svc, repo, InMemoryIntentStore(), + resolve_symbol=lambda pid: "005930") + from trex_core.application.ledger_outbox import InMemoryLedgerOutbox + outbox = InMemoryLedgerOutbox() + execution = ExecutionService(svc, repo, PaperBroker(fill_after_polls=0), + ledger_outbox=outbox, event_emitter=emitter) + repo.reconcile_universe("default", {"PF-1": "SYM"}) + + d = intents.submit_buy(BuyIntent(intent_id="b1", qty=10, strategy_id="orb-v1", portfolio_id="PF-1")) + execution.run_once() + intents.submit_sell(SellIntent(intent_id="s1", slot_id=d.slot_id)) + execution.run_once() + + # Neither buy nor sell events should use "_" as strategy token. + bad_keys = [rk for rk in all_published if "._.".encode() in rk.encode() or rk.split(".")[1] == "_"] + assert bad_keys == [], ( + f"DLQ-sentinel routing keys found (strategy_id=None leak): {bad_keys}" + ) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 4a9f4c5..f5ff3f5 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -41,7 +41,14 @@ _log = logging.getLogger(__name__) # 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). -SlotEventEmitter = Callable[[str, object], None] # (event_type, slot) -> None +# +# 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 @@ -288,6 +295,13 @@ class ExecutionService: 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: @@ -342,7 +356,9 @@ class ExecutionService: 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. - self._emit_event(event_type, updated) + # 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: @@ -355,6 +371,9 @@ class ExecutionService: 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: @@ -395,7 +414,7 @@ class ExecutionService: 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) + 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: @@ -454,11 +473,11 @@ class ExecutionService: slot.slot_id, slot.broker_order_id, ) - def _emit_event(self, event_type: str, slot) -> None: + 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) + self._event_emitter(event_type, slot, routing_strategy_id) except Exception: # noqa: BLE001 - event publish failure is non-critical return diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 4ad33b7..17755b8 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -264,10 +264,18 @@ class Daemon: # ------------------------------------------------------------ event hook def _make_event_emitter(self): - def emit(event_type: str, slot) -> None: + def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None: payload = slot_transition_event_payload(slot, event_type=event_type) + # routing_strategy_id is supplied when the transition clears + # slot.strategy_id (e.g. PENDING_SELL -> INACTIVE for slot_filled_sell). + # Without this hint the routing key would use "_" and the message would + # be NO_ROUTEd to the DLQ instead of reaching the strategy queue + # (M2 REL-009 feedback / slot_filled NO_ROUTE fix). + effective_strategy_id = ( + routing_strategy_id or getattr(slot, "strategy_id", None) or "_" + ) rk = topology.event_routing_key( - event_type, slot.strategy_id or "_", slot.symbol or "_", + event_type, effective_strategy_id, slot.symbol or "_", ) try: self.event_publisher.publish_event(rk, payload)