fix: preserve sell event routing owner

tests/adapters/messaging/test_publish_reliability.py

@@ -101,6 +101,31 @@ class TestPublishRetryOnHalfOpen:
 
         assert reconnect_called["yes"], "_force_reconnect must be called on first failure"
 
+    def test_sell_retry_reuses_identical_payload_and_mandatory_flag(self):
+        t = _make_transport()
+        _inject_connected(t)
+        payload = {
+            "slot_id": "slot-95",
+            "event_type": "slot_filled_sell",
+            "sequence_at": 11,
+        }
+        routing_key = "slot_filled_sell.SELL-B.005930"
+
+        with (
+            patch.object(
+                t,
+                "_do_basic_publish",
+                side_effect=[OSError("first attempt"), None],
+            ) as publish,
+            patch.object(t, "_force_reconnect"),
+        ):
+            t.publish_event(routing_key, payload)
+
+        assert publish.call_count == 2
+        assert publish.call_args_list[0] == publish.call_args_list[1]
+        assert publish.call_args_list[0].args[0] == routing_key
+        assert publish.call_args_list[0].kwargs["mandatory"] is True
+
 
 # ---------------------------------------------------------------------------
 # REQ-091 AC-2: retry also fails → TransportUnavailableError raised

tests/adapters/messaging/test_rabbitmq_transport.py

@@ -5,6 +5,7 @@ queue/routing keys to avoid colliding with other tests or runs.
 """
 from __future__ import annotations
 
+import json
 import os
 import threading
 import time
@@ -149,6 +150,63 @@ def test_event_publish_and_per_strategy_consume(mq_url):
         _drain_queue(mq_url, queue_b)
 
 
+def test_sell_event_retry_preserves_payload_and_strategy_route(mq_url, monkeypatch):
+    """AC-0039: live retry keeps identity and uses mandatory confirmed publish."""
+    suffix = uuid.uuid4().hex[:8]
+    owner = f"SELL-{suffix}"
+    other_owner = f"OTHER-{suffix}"
+    queue = topology.strategy_event_queue_name(owner)
+    other_queue = topology.strategy_event_queue_name(other_owner)
+    routing_key = topology.event_routing_key("slot_filled_sell", owner, "005930")
+    payload = {
+        "slot_id": f"slot-{suffix}",
+        "event_type": "slot_filled_sell",
+        "sequence_at": 7,
+    }
+    publisher = RabbitMQTransport(mq_url, timeout=5.0)
+    attempts: list[tuple[str, bytes, bool]] = []
+    try:
+        publisher.declare_topology()
+        publisher.declare_strategy_queue(
+            queue, topology.strategy_event_binding_pattern(owner),
+        )
+        publisher.declare_strategy_queue(
+            other_queue, topology.strategy_event_binding_pattern(other_owner),
+        )
+        original_publish = publisher._do_basic_publish
+
+        def fail_first_then_publish(rk: str, body: bytes, *, mandatory: bool) -> None:
+            attempts.append((rk, body, mandatory))
+            if len(attempts) == 1:
+                raise OSError("forced first-attempt failure")
+            original_publish(rk, body, mandatory=mandatory)
+
+        monkeypatch.setattr(publisher, "_do_basic_publish", fail_first_then_publish)
+        publisher.publish_event(routing_key, payload)
+
+        assert len(attempts) == 2
+        assert attempts[0] == attempts[1]
+        assert attempts[0][0] == routing_key
+        assert attempts[0][2] is True
+        assert json.loads(attempts[0][1]) == payload
+
+        publisher._ensure_connected()
+        channel = publisher._channel
+        assert channel is not None
+        method, _props, body = channel.basic_get(queue, auto_ack=True)
+        assert method is not None
+        assert body is not None
+        assert json.loads(body) == payload
+        other_method, _other_props, _other_body = channel.basic_get(
+            other_queue, auto_ack=True,
+        )
+        assert other_method is None
+    finally:
+        publisher.close()
+        _drain_queue(mq_url, queue)
+        _drain_queue(mq_url, other_queue)
+
+
 def test_call_rpc_timeout(mq_url):
     queue = f"trex.test.rpc.timeout.{uuid.uuid4().hex[:8]}"
     client = RabbitMQTransport(mq_url, timeout=5.0)

tests/adapters/persistence/test_postgresql_repository.py

@@ -63,6 +63,38 @@ def test_buy_intent_persists_transition(pg_clean):
     assert slot.status is SlotState.PENDING_BUY and slot.symbol == "005930"
 
 
+def test_sell_admission_owner_survives_postgresql_repository_restart(pg_clean):
+    repo, svc, intents = _wire(pg_clean)
+    repo.reconcile_universe(INST, _SYMS)
+    decision = intents.submit_buy(BuyIntent(
+        intent_id="pg-owner-buy",
+        qty=10,
+        strategy_id="BUY-A",
+        portfolio_id="PF-1",
+    ))
+    assert decision.slot_id is not None
+    svc.transit_slot(
+        decision.slot_id,
+        SlotState.PENDING_BUY,
+        SlotState.HOLDING,
+        reason="filled",
+        actor="test",
+        qty=10.0,
+        avg_price=1000.0,
+    )
+    sell = intents.submit_sell(SellIntent(
+        intent_id="pg-owner-sell",
+        slot_id=decision.slot_id,
+        sell_strategy_id="SELL-B",
+    ))
+    assert sell.admitted
+
+    reloaded = PostgreSQLSlotRepository(pg_clean).fetch_slot(decision.slot_id)
+    assert reloaded is not None
+    assert reloaded.status is SlotState.PENDING_SELL
+    assert reloaded.sell_strategy_id == "SELL-B"
+
+
 def test_buy_intent_idempotent_on_pg(pg_clean):
     repo, svc, intents = _wire(pg_clean)
     repo.reconcile_universe(INST, _SYMS)

tests/application/test_execution_cancel.py

@@ -65,7 +65,7 @@ class ProgrammableBroker:
         return None
 
 
-def _wire(broker, *, stale_sec=None):
+def _wire(broker, *, stale_sec=None, event_emitter=None):
     repo = InMemorySlotRepository()
     svc = SlotService(repo)
     intents = IntentService(svc, repo, InMemoryIntentStore(),
@@ -74,7 +74,7 @@ def _wire(broker, *, stale_sec=None):
     ledger = InMemoryLedger()
     execution = ExecutionService(
         svc, repo, broker, ledger_outbox=outbox,
-        stale_pending_threshold_sec=stale_sec,
+        stale_pending_threshold_sec=stale_sec, event_emitter=event_emitter,
     )
     writer = LedgerWriter(outbox, ledger)
     repo.reconcile_universe("default", {"PF-1": "SYM"})
@@ -104,6 +104,64 @@ def test_full_external_cancel_returns_slot_to_holding():
     assert r.rolled_back == 1
 
 
+def test_zero_fill_cancel_preserves_owner_for_ownerless_resell():
+    broker = ProgrammableBroker()
+    repo, _svc, intents, execution, _ledger, _writer = _wire(broker)
+    sid = _buy_to_holding(intents, execution)
+    assert sid is not None
+
+    intents.submit_sell(SellIntent(
+        intent_id="s-owner-1", slot_id=sid, sell_strategy_id="SELL-B",
+    ))
+    broker.responder = lambda side: FillStatus(state="cancelled")
+    execution.run_once()
+
+    holding = repo.fetch_slot(sid)
+    assert holding is not None
+    assert holding.status is SlotState.HOLDING
+    assert holding.sell_strategy_id == "SELL-B"
+
+    intents.submit_sell(SellIntent(intent_id="s-owner-2", slot_id=sid))
+    pending = repo.fetch_slot(sid)
+    assert pending is not None
+    assert pending.status is SlotState.PENDING_SELL
+    assert pending.sell_strategy_id == "SELL-B"
+
+
+
+def test_partial_sell_cancel_routes_with_sell_owner_and_preserves_it():
+    emitted: list[tuple[str, str | None]] = []
+
+    def emitter(event_type, _slot, owner=None):
+        emitted.append((event_type, owner))
+
+    broker = ProgrammableBroker()
+    repo, _svc, intents, execution, _ledger, _writer = _wire(
+        broker, event_emitter=emitter,
+    )
+    sid = _buy_to_holding(intents, execution)
+    assert sid is not None
+    intents.submit_sell(SellIntent(
+        intent_id="s-partial-owner", slot_id=sid, sell_strategy_id="SELL-B",
+    ))
+    broker.balance["005930"] = 6
+    broker.responder = lambda side: FillStatus(
+        state="partially_filled",
+        filled_qty=4,
+        avg_price=1100.0,
+        execution_id="exec-partial-owner",
+        cancelled=True,
+    )
+
+    execution.run_once()
+
+    holding = repo.fetch_slot(sid)
+    assert holding is not None
+    assert holding.status is SlotState.HOLDING
+    assert holding.sell_strategy_id == "SELL-B"
+    assert ("slot_filled_sell", "SELL-B") in emitted
+
+
 def test_unresolved_poll_observability_is_sanitized_and_deduplicated(caplog):
     broker = ProgrammableBroker()
     repo, _svc, intents, execution, _ledger, _writer = _wire(broker)

tests/application/test_messaging_event_publisher.py

@@ -1,6 +1,7 @@
 """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
@@ -218,3 +219,213 @@ def test_sell_fill_dlq_sentinel_absent_when_strategy_id_set():
     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

tests/domain/slot/test_service_sell.py

@@ -34,7 +34,7 @@ def _holding_slot(service, repo):
     slot = _inactive_slot(repo)
     pending = service.transit_slot(
         slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
-        reason="buy", actor="test", symbol="005930",
+        reason="buy", actor="test", symbol="005930", strategy_id="BUY-A",
     )
     return service.transit_slot(
         pending.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
@@ -49,6 +49,7 @@ class TestSell:
         service.sell(holding.slot_id, actor="safety-sell", reason="stop_loss")
         updated = repo.fetch_slot(holding.slot_id)
         assert updated.status is SlotState.PENDING_SELL
+        assert updated.sell_strategy_id == "BUY-A"
 
     def test_sell_records_transition_log(self, service, repo):
         """sell() appends a transition log entry."""

tests/runtime/test_sell_event_owner_routing.py

@@ -0,0 +1,60 @@
+"""Runtime SELL event owner routing defenses (REQ-0011 / AC-0039)."""
+from __future__ import annotations
+
+from typing import Any, cast
+from uuid import uuid4
+
+from trex_core.domain.slot import Slot, SlotState
+from trex_core.runtime.daemon import Daemon
+
+
+class _Publisher:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, dict]] = []
+        self.failure: Exception | None = None
+
+    def publish_event(self, routing_key: str, payload: dict) -> None:
+        if self.failure is not None:
+            raise self.failure
+        self.calls.append((routing_key, payload))
+
+
+def _daemon_with_publisher() -> tuple[Daemon, _Publisher]:
+    daemon = object.__new__(Daemon)
+    publisher = _Publisher()
+    daemon.event_publisher = cast(Any, publisher)
+    return daemon, publisher
+
+
+def test_runtime_emitter_blocks_ownerless_sell_without_synthetic_owner(caplog):
+    daemon, publisher = _daemon_with_publisher()
+    slot = Slot(slot_id=uuid4(), status=SlotState.INACTIVE, symbol="005930")
+
+    daemon._make_event_emitter()("slot_filled_sell", slot, None)
+
+    assert publisher.calls == []
+    assert "reason=sell_event_owner_missing" in caplog.text
+
+
+def test_runtime_emitter_routes_sell_with_explicit_owner():
+    daemon, publisher = _daemon_with_publisher()
+    slot = Slot(slot_id=uuid4(), status=SlotState.INACTIVE, symbol="005930")
+
+    daemon._make_event_emitter()("slot_filled_sell", slot, "SELL-B")
+
+    assert len(publisher.calls) == 1
+    assert publisher.calls[0][0] == "slot_filled_sell.SELL-B.005930"
+
+
+def test_runtime_publish_failure_log_excludes_raw_exception(caplog):
+    daemon, publisher = _daemon_with_publisher()
+    publisher.failure = RuntimeError("SENTINEL_SECRET_CONNECTION_TEXT")
+    slot = Slot(slot_id=uuid4(), status=SlotState.INACTIVE, symbol="005930")
+
+    daemon._make_event_emitter()("slot_filled_sell", slot, "SELL-B")
+
+    assert publisher.calls == []
+    assert "reason=event_publish_failed" in caplog.text
+    assert "exc_type=RuntimeError" in caplog.text
+    assert "SENTINEL_SECRET_CONNECTION_TEXT" not in caplog.text
+    assert not any(record.exc_info for record in caplog.records)