"""Slot transition event emission tests (REQ-021 + REQ-023 + REQ-018 hook).""" from __future__ import annotations import logging import threading import time from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.ledger.in_memory import InMemoryLedger 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, 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 from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository def _wire(transport): repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") outbox = InMemoryLedgerOutbox(instance_key="test-instance") ledger = InMemoryLedger() writer = LedgerWriter(outbox, ledger) repo.reconcile_universe("default", {"PF-1": "SYM"}) 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) except Exception: # noqa: BLE001 pass execution = ExecutionService(svc, repo, PaperBroker(fill_after_polls=0), ledger_outbox=outbox, event_emitter=emitter) return repo, svc, intents, execution, writer def test_fill_emits_slot_filled_event_with_routing_key_and_state(): transport = InMemoryTransport() received: 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.append((rk, p)), stop), daemon=True, ) th.start() time.sleep(0.05) repo, _, intents, execution, _ = _wire(transport) intents.submit_buy(BuyIntent( intent_id="b1", qty=10, strategy_id="A", portfolio_id="PF-1", )) execution.run_once() time.sleep(0.05) stop.set() th.join(timeout=1.0) assert len(received) == 1 rk, payload = received[0] assert rk == "slot_filled_buy.A.005930" # payload carries current state + monotonic sequence (REQ-023 AC-5) assert payload["event_type"] == "slot_filled_buy" assert payload["status"] == SlotState.HOLDING.value assert payload["strategy_id"] == "A" assert payload["symbol"] == "005930" assert "sequence_at" in payload # last-wins ordering token def test_event_for_strategy_a_does_not_reach_strategy_b_queue(): """REQ-021 AC-4: per-strategy routing key binding filters at the broker.""" transport = InMemoryTransport() received_a: list[tuple[str, dict]] = [] received_b: list[tuple[str, dict]] = [] stop = threading.Event() th_a = threading.Thread( target=transport.consume_events_forever, args=("q.A", topology.strategy_event_binding_pattern("A"), lambda rk, p: received_a.append((rk, p)), stop), daemon=True, ) th_b = threading.Thread( target=transport.consume_events_forever, args=("q.B", topology.strategy_event_binding_pattern("B"), lambda rk, p: received_b.append((rk, p)), stop), daemon=True, ) th_a.start() th_b.start() time.sleep(0.05) _, _, intents, execution, _ = _wire(transport) intents.submit_buy(BuyIntent( intent_id="b1", qty=10, strategy_id="A", portfolio_id="PF-1", )) execution.run_once() time.sleep(0.05) stop.set() th_a.join(timeout=1.0) th_b.join(timeout=1.0) assert any(rk == "slot_filled_buy.A.005930" for rk, _ in received_a) assert received_b == [] def test_publish_failure_does_not_break_execution_loop(): """Publish errors must NOT roll back slot transitions or ledger enqueue.""" transport = InMemoryTransport() transport.set_available(False) # all publishes will raise TransportUnavailableError repo, _, intents, execution, _ = _wire(transport) intents.submit_buy(BuyIntent( intent_id="b1", qty=10, strategy_id="A", portfolio_id="PF-1", )) r = execution.run_once() # Slot still transitioned; ledger still enqueued. assert r.filled == 1 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(instance_key="test-instance") 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}" ) def test_sell_fill_prefers_admission_sell_strategy_owner(): emitted: list[tuple[str, str | None]] = [] def emitter(event_type: str, _slot, owner: str | None = None) -> None: emitted.append((event_type, owner)) repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService( svc, repo, InMemoryIntentStore(), resolve_symbol=lambda _pid: "005930", ) outbox = InMemoryLedgerOutbox(instance_key="test-instance") execution = ExecutionService( svc, repo, PaperBroker(fill_after_polls=0), ledger_outbox=outbox, event_emitter=emitter, ) repo.reconcile_universe("default", {"PF-1": "SYM"}) decision = intents.submit_buy(BuyIntent( intent_id="b-owner", qty=10, strategy_id="BUY-A", portfolio_id="PF-1", )) assert decision.slot_id is not None execution.run_once() intents.submit_sell(SellIntent( intent_id="s-owner", slot_id=decision.slot_id, sell_strategy_id="SELL-B", )) pending = repo.fetch_slot(decision.slot_id) assert pending is not None assert pending.sell_strategy_id == "SELL-B" execution.run_once() assert ("slot_filled_sell", "SELL-B") in emitted assert ("slot_filled_sell", "BUY-A") not in emitted execution.run_once() assert emitted.count(("slot_filled_sell", "SELL-B")) == 1 def test_ownerless_sell_admission_persists_legacy_buy_owner(): repo, _svc, intents, execution, _writer = _wire(InMemoryTransport()) decision = intents.submit_buy(BuyIntent( intent_id="b-legacy-owner", qty=10, strategy_id="BUY-A", portfolio_id="PF-1", )) assert decision.slot_id is not None execution.run_once() intents.submit_sell(SellIntent( intent_id="s-legacy-owner", slot_id=decision.slot_id, sell_strategy_id=" ", )) pending = repo.fetch_slot(decision.slot_id) assert pending is not None assert pending.sell_strategy_id == "BUY-A" def test_sell_owner_survives_execution_service_restart(): emitted: list[tuple[str, str | None]] = [] def emitter(event_type: str, _slot, owner: str | None = None) -> None: emitted.append((event_type, owner)) repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService( svc, repo, InMemoryIntentStore(), resolve_symbol=lambda _pid: "005930", ) outbox = InMemoryLedgerOutbox(instance_key="default") repo.reconcile_universe("default", {"PF-1": "SYM"}) broker = PaperBroker(fill_after_polls=0) first_execution = ExecutionService( svc, repo, broker, ledger_outbox=outbox, ) decision = intents.submit_buy(BuyIntent( intent_id="b-restart-owner", qty=10, strategy_id="BUY-A", portfolio_id="PF-1", )) assert decision.slot_id is not None first_execution.run_once() intents.submit_sell(SellIntent( intent_id="s-restart-owner", slot_id=decision.slot_id, sell_strategy_id="SELL-B", )) restarted_execution = ExecutionService( svc, repo, broker, ledger_outbox=outbox, event_emitter=emitter, ) restarted_execution.run_once() assert ("slot_filled_sell", "SELL-B") in emitted class _AlarmRecorder: def __init__(self) -> None: self.calls: list[tuple[str, dict]] = [] def publish_alarm(self, code, _title, *, fields=None, severity="critical"): self.calls.append((code, {**(fields or {}), "severity": severity})) return True def test_ownerless_sell_fill_is_skipped_without_financial_rollback(caplog): emitted: list[tuple[str, str | None]] = [] alarm = _AlarmRecorder() def emitter(event_type: str, _slot, owner: str | None = None) -> None: emitted.append((event_type, owner)) repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService( svc, repo, InMemoryIntentStore(), resolve_symbol=lambda _pid: "005930", ) outbox = InMemoryLedgerOutbox(instance_key="test-instance") execution = ExecutionService( svc, repo, PaperBroker(fill_after_polls=0), ledger_outbox=outbox, event_emitter=emitter, alarm_emitter=alarm, # type: ignore[arg-type] instance_key="default", ) repo.reconcile_universe("default", {"PF-1": "SYM"}) decision = intents.submit_buy(BuyIntent( intent_id="b-ownerless", qty=10, portfolio_id="PF-1", )) assert decision.slot_id is not None execution.run_once() intents.submit_sell(SellIntent(intent_id="s-ownerless", slot_id=decision.slot_id)) with caplog.at_level(logging.ERROR, logger="trex_core.application.execution"): result = execution.run_once() sell_calls = [call for call in emitted if call[0] == "slot_filled_sell"] assert sell_calls == [] assert result.sell_event_owner_missing == 1 assert result.filled == 1 assert result.ledger_enqueued == 1 final_slot = repo.fetch_slot(decision.slot_id) assert final_slot is not None assert final_slot.status is SlotState.INACTIVE assert any("reason=sell_event_owner_missing" in rec.message for rec in caplog.records) assert alarm.calls == [( "sell_event_owner_missing", { "instance_key": "default", "slot_id": str(decision.slot_id), "symbol": "005930", "event_type": "slot_filled_sell", "reason": "sell_event_owner_missing", "severity": "critical", }, )] def test_ownerless_sell_alarm_failure_does_not_rollback_fill(): class RaisingAlarm: def publish_alarm(self, *_args, **_kwargs): raise RuntimeError("alarm unavailable") repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService( svc, repo, InMemoryIntentStore(), resolve_symbol=lambda _pid: "005930", ) outbox = InMemoryLedgerOutbox(instance_key="default") execution = ExecutionService( svc, repo, PaperBroker(fill_after_polls=0), ledger_outbox=outbox, alarm_emitter=RaisingAlarm(), # type: ignore[arg-type] instance_key="default", ) repo.reconcile_universe("default", {"PF-1": "SYM"}) decision = intents.submit_buy(BuyIntent( intent_id="b-ownerless-alarm", qty=10, portfolio_id="PF-1", )) assert decision.slot_id is not None execution.run_once() intents.submit_sell(SellIntent( intent_id="s-ownerless-alarm", slot_id=decision.slot_id, )) result = execution.run_once() final_slot = repo.fetch_slot(decision.slot_id) assert final_slot is not None assert final_slot.status is SlotState.INACTIVE assert result.filled == 1 assert result.ledger_enqueued == 1 assert result.sell_event_owner_missing == 1