feat(intents): emit warn alarm on first reconcile_warned rejection

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

tests/application/test_reconcile_warning_blocking.py

@@ -58,6 +58,47 @@ class InMemoryIntentStore:
         self._db[intent_id] = {"status": status, "slot_id": slot_id, "reason": reason}
 
 
+class RecordingIntentStore(InMemoryIntentStore):
+    def __init__(self, trace=None, fail_finalize=False):
+        super().__init__()
+        self._trace = trace
+        self._fail_finalize = fail_finalize
+
+    def finalize(self, intent_id, status, slot_id, reason):
+        if self._trace is not None:
+            self._trace.append("finalize")
+        if self._fail_finalize:
+            raise RuntimeError("finalize failed")
+        super().finalize(intent_id, status, slot_id, reason)
+
+
+class RecordingAlarmEmitter:
+    def __init__(self, *, trace=None, return_value=True, raise_error=False):
+        self.calls = []
+        self._trace = trace
+        self._return_value = return_value
+        self._raise_error = raise_error
+
+    def publish_alarm(self, code, title, *, fields=None, severity="critical"):
+        if self._trace is not None:
+            self._trace.append("publish")
+        self.calls.append((code, title, severity, fields or {}))
+        if self._raise_error:
+            raise RuntimeError("transport down")
+        return self._return_value
+
+
+class DuplicateInFlightIntentStore:
+    def try_insert(self, intent_id, intent_type, payload):
+        return False
+
+    def get(self, intent_id):
+        return None
+
+    def finalize(self, intent_id, status, slot_id, reason):
+        raise AssertionError("finalize must not be called for duplicate_in_flight")
+
+
 def _make_holding_slot(repo, svc, instance_key="default", portfolio_id="pf-A", symbol="005930", qty=10):
     """Helper: creates universe, puts slot in HOLDING state with given qty."""
     repo.reconcile_universe(instance_key, {portfolio_id: symbol})
@@ -272,6 +313,206 @@ def test_intent_service_sell_rejected_when_reconcile_warned():
     assert d.reason == "reconcile_warned"
 
 
+def test_reconcile_warned_buy_emits_alarm_with_allowlisted_fields_after_finalize():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    trace = []
+    intent_store = RecordingIntentStore(trace=trace)
+    alarms = RecordingAlarmEmitter(trace=trace)
+
+    repo.reconcile_universe("default", {"pf-A": "005930"})
+    slot = repo.find_slot_by_portfolio("default", "pf-A")
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
+
+    intent_svc = IntentService(
+        svc, repo, intent_store,
+        instance_key="default",
+        warn_store=warn_store,
+        alarm_emitter=alarms,
+    )
+    d = intent_svc.submit_buy(BuyIntent(
+        intent_id="b-alert-001",
+        slot_id=slot.slot_id,
+        entry_price=100.0,
+        qty=10.0,
+        strategy_id="stg-buy",
+    ))
+    assert d.status == "rejected"
+    assert d.reason == "reconcile_warned"
+    assert trace == ["finalize", "publish"]
+    assert len(alarms.calls) == 1
+    code, title, severity, fields = alarms.calls[0]
+    assert code == "strategy_intent_rejected"
+    assert severity == "warn"
+    assert title
+    assert fields["instance_key"] == "default"
+    assert fields["intent_id"] == "b-alert-001"
+    assert fields["intent_type"] == "buy"
+    assert fields["reason"] == "reconcile_warned"
+    assert fields["broker_submitted"] is False
+    assert fields["slot_id"] == str(slot.slot_id)
+    assert fields["portfolio_id"] == "pf-A"
+    assert fields["symbol"] == "005930"
+    assert fields["strategy_id"] == "stg-buy"
+    for blocked in ("qty", "entry_price", "limit_price", "venue", "extra", "sell_strategy_id"):
+        assert blocked not in fields
+    text = repr((title, fields))
+    for forbidden in ("account", "token", "password", "secret", "dsn", "connection", "raw_payload"):
+        assert forbidden not in text.lower()
+
+
+def test_reconcile_warned_sell_emits_alarm_with_normalized_strategy_id():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    intent_store = InMemoryIntentStore()
+    alarms = RecordingAlarmEmitter()
+
+    slot = _make_holding_slot(repo, svc, qty=10, portfolio_id="pf-A", symbol="005930")
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
+
+    intent_svc = IntentService(
+        svc, repo, intent_store,
+        instance_key="default",
+        warn_store=warn_store,
+        alarm_emitter=alarms,
+    )
+    d = intent_svc.submit_sell(SellIntent(
+        intent_id="s-alert-001",
+        slot_id=slot.slot_id,
+        reason="strategy_exit",
+        sell_strategy_id="stg-sell",
+    ))
+    assert d.status == "rejected"
+    assert d.reason == "reconcile_warned"
+    assert len(alarms.calls) == 1
+    _, _, severity, fields = alarms.calls[0]
+    assert severity == "warn"
+    assert fields["intent_type"] == "sell"
+    assert fields["strategy_id"] == "stg-sell"
+    assert fields["slot_id"] == str(slot.slot_id)
+
+
+def test_reconcile_warned_alarm_dedup_by_intent_id_across_service_recreation():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    intent_store = InMemoryIntentStore()
+    alarms = RecordingAlarmEmitter()
+
+    repo.reconcile_universe("default", {"pf-A": "005930"})
+    slot = repo.find_slot_by_portfolio("default", "pf-A")
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
+
+    s1 = IntentService(
+        svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
+    )
+    s2 = IntentService(
+        svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
+    )
+    first = s1.submit_buy(BuyIntent(intent_id="b-dedup", slot_id=slot.slot_id))
+    second = s1.submit_buy(BuyIntent(intent_id="b-dedup", slot_id=slot.slot_id))
+    third = s2.submit_buy(BuyIntent(intent_id="b-dedup", slot_id=slot.slot_id))
+    assert first.reason == "reconcile_warned"
+    assert second.reason == "reconcile_warned"
+    assert third.reason == "reconcile_warned"
+    assert len(alarms.calls) == 1
+
+
+def test_reconcile_warned_alarm_emits_per_distinct_intent_id_on_same_slot():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    intent_store = InMemoryIntentStore()
+    alarms = RecordingAlarmEmitter()
+
+    repo.reconcile_universe("default", {"pf-A": "005930"})
+    slot = repo.find_slot_by_portfolio("default", "pf-A")
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
+
+    intent_svc = IntentService(
+        svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
+    )
+    d1 = intent_svc.submit_buy(BuyIntent(intent_id="b-uniq-1", slot_id=slot.slot_id))
+    d2 = intent_svc.submit_buy(BuyIntent(intent_id="b-uniq-2", slot_id=slot.slot_id))
+    assert d1.reason == "reconcile_warned"
+    assert d2.reason == "reconcile_warned"
+    assert len(alarms.calls) == 2
+
+
+def test_reconcile_warned_alarm_not_published_for_duplicate_in_flight_or_other_decisions():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    alarms = RecordingAlarmEmitter()
+
+    duplicate = IntentService(
+        svc, repo, DuplicateInFlightIntentStore(), instance_key="default", alarm_emitter=alarms
+    )
+    d_dup = duplicate.submit_buy(BuyIntent(intent_id="dup", portfolio_id="pf-A"))
+    assert d_dup.reason == "duplicate_in_flight"
+    assert alarms.calls == []
+
+    repo.reconcile_universe("default", {"pf-A": "005930"})
+    slot = repo.find_slot_by_portfolio("default", "pf-A")
+    normal = IntentService(
+        svc, repo, InMemoryIntentStore(), instance_key="default", alarm_emitter=alarms
+    )
+    admitted = normal.submit_buy(BuyIntent(intent_id="admitted", slot_id=slot.slot_id, qty=1, entry_price=100))
+    assert admitted.status == "admitted"
+    invalid = normal.submit_buy(BuyIntent(intent_id="bad-venue", slot_id=slot.slot_id, venue="BAD"))
+    assert invalid.reason == "invalid_venue"
+    repo.reconcile_universe("default", {"pf-A": "005930", "pf-B": "000660"})
+    slot_b = repo.find_slot_by_portfolio("default", "pf-B")
+    capped = IntentService(
+        svc, repo, InMemoryIntentStore(), instance_key="default", max_active_slots=0, alarm_emitter=alarms
+    )
+    full = capped.submit_buy(BuyIntent(intent_id="full", slot_id=slot_b.slot_id))
+    assert full.reason == "capacity_full"
+    assert alarms.calls == []
+
+
+def test_reconcile_warned_alarm_finalize_failure_and_publish_failure_are_isolated():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    repo.reconcile_universe("default", {"pf-A": "005930"})
+    slot = repo.find_slot_by_portfolio("default", "pf-A")
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
+
+    alarms = RecordingAlarmEmitter()
+    failing_store = RecordingIntentStore(fail_finalize=True)
+    failing_finalize = IntentService(
+        svc, repo, failing_store,
+        instance_key="default",
+        warn_store=warn_store,
+        alarm_emitter=alarms,
+    )
+    try:
+        failing_finalize.submit_buy(BuyIntent(intent_id="b-finalize-fail", slot_id=slot.slot_id))
+        assert False, "expected finalize failure"
+    except RuntimeError:
+        pass
+    assert alarms.calls == []
+
+    soft_fail_alarm = RecordingAlarmEmitter(return_value=False)
+    raising_alarm = RecordingAlarmEmitter(raise_error=True)
+    non_blocking_1 = IntentService(
+        svc, repo, InMemoryIntentStore(), instance_key="default", warn_store=warn_store, alarm_emitter=soft_fail_alarm
+    )
+    non_blocking_2 = IntentService(
+        svc, repo, InMemoryIntentStore(), instance_key="default", warn_store=warn_store, alarm_emitter=raising_alarm
+    )
+    d1 = non_blocking_1.submit_buy(BuyIntent(intent_id="b-alarm-false", slot_id=slot.slot_id))
+    d2 = non_blocking_2.submit_buy(BuyIntent(intent_id="b-alarm-exc", slot_id=slot.slot_id))
+    d3 = IntentService(
+        svc, repo, InMemoryIntentStore(), instance_key="default", warn_store=warn_store, alarm_emitter=None
+    ).submit_buy(BuyIntent(intent_id="b-no-emitter", slot_id=slot.slot_id))
+    assert d1.reason == "reconcile_warned"
+    assert d2.reason == "reconcile_warned"
+    assert d3.reason == "reconcile_warned"
+
+
 # REQ-104 AC-3: no warn, no pause => buy/sell admitted normally
 def test_intent_service_no_warn_admitted_normally():
     repo = InMemorySlotRepository()

tests/runtime/test_daemon_resolve_symbol.py

@@ -7,6 +7,8 @@ daemon wiring was never run until the REL-004 live smoke timed out every buy.
 """
 from __future__ import annotations
 
+from pathlib import Path
+
 from trex_core.application.portfolio import PortfolioItem, SnapshotStore
 from trex_core.runtime.daemon import Daemon
 
@@ -35,3 +37,14 @@ def test_resolve_symbol_reads_item_symbol():
 def test_resolve_symbol_unknown_returns_none():
     holder = _StoreHolder(_store())
     assert Daemon._resolve_symbol(holder, "pf-missing") is None
+
+
+def test_daemon_wires_shared_alarm_emitter_into_intent_service():
+    src = (
+        Path(__file__).resolve().parents[2]
+        / "trex_core"
+        / "runtime"
+        / "daemon.py"
+    ).read_text(encoding="utf-8")
+    assert "self.intent_service = IntentService(" in src
+    assert "alarm_emitter=self.alarm_emitter" in src

trex_core/application/intents.py

@@ -28,6 +28,7 @@ from decimal import Decimal
 from typing import Any, Callable, Optional
 from uuid import UUID
 
+from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.ports import IntentStore
 from trex_core.domain.slot import (
     SlotAdmissionRejectedError,
@@ -115,6 +116,7 @@ class IntentService:
         max_active_slots: Optional[int] = None,
         default_order_timeout_sec: Optional[int] = None,
         warn_store=None,  # Optional[ReconcileWarningStore] — BAT-050 / REQ-104
+        alarm_emitter: Optional[AlarmEmitter] = None,
         entry_allowed_fn: Optional[Callable[[], bool]] = None,
         buy_guard_fn: Optional[Callable[[BuyIntent], Optional[str]]] = None,
     ) -> None:
@@ -141,6 +143,8 @@ class IntentService:
         # BAT-050 / REQ-104: reconcile-warning gate store (Optional).
         # When None, gate is skipped (backward compat, NFR-4).
         self._warn_store = warn_store
+        # Optional best-effort alarm publisher for reconcile_warned rejections.
+        self._alarm_emitter = alarm_emitter
         # dcy-036 / REQ-114: day-start entry gate (Optional callable → bool).
         # When None or returns True, gate passes (backward compat / opt-in=False).
         # When returns False, BuyIntent is rejected with reason 'day_start_load_failed'.
@@ -178,6 +182,22 @@ class IntentService:
         self._intents.finalize(
             intent.intent_id, decision.status, decision.slot_id, decision.reason
         )
+        if decision.status == "rejected" and decision.reason == "reconcile_warned":
+            slot = (
+                self._repo.fetch_slot(intent.slot_id)
+                if intent.slot_id is not None
+                else (
+                    self._repo.find_slot_by_portfolio(self._instance_key, intent.portfolio_id)
+                    if intent.portfolio_id
+                    else None
+                )
+            )
+            self._publish_reconcile_warned_rejection_alarm(
+                intent_id=intent.intent_id,
+                intent_type="buy",
+                strategy_id=intent.strategy_id,
+                slot=slot,
+            )
         return decision
 
     def submit_sell(self, intent: SellIntent) -> IntentDecision:
@@ -196,8 +216,51 @@ class IntentService:
         self._intents.finalize(
             intent.intent_id, decision.status, decision.slot_id, decision.reason
         )
+        if decision.status == "rejected" and decision.reason == "reconcile_warned":
+            self._publish_reconcile_warned_rejection_alarm(
+                intent_id=intent.intent_id,
+                intent_type="sell",
+                strategy_id=intent.sell_strategy_id,
+                slot=self._repo.fetch_slot(intent.slot_id),
+            )
         return decision
 
+    def _publish_reconcile_warned_rejection_alarm(
+        self,
+        *,
+        intent_id: str,
+        intent_type: str,
+        strategy_id: Optional[str],
+        slot: Any,
+    ) -> None:
+        if self._alarm_emitter is None:
+            return
+        fields: dict[str, Any] = {
+            "instance_key": self._instance_key,
+            "intent_id": intent_id,
+            "intent_type": intent_type,
+            "reason": "reconcile_warned",
+            "broker_submitted": False,
+        }
+        if slot is not None:
+            if getattr(slot, "slot_id", None) is not None:
+                fields["slot_id"] = str(slot.slot_id)
+            if getattr(slot, "portfolio_id", None):
+                fields["portfolio_id"] = slot.portfolio_id
+            if getattr(slot, "symbol", None):
+                fields["symbol"] = slot.symbol
+        if strategy_id:
+            fields["strategy_id"] = strategy_id
+        try:
+            self._alarm_emitter.publish_alarm(
+                "strategy_intent_rejected",
+                "Strategy intent rejected by reconcile warning gate",
+                fields=fields,
+                severity="warn",
+            )
+        except Exception:  # noqa: BLE001 - alarm path must not affect intent decision
+            pass
+
     # ------------------------------------------------------------
     def _dedup(self, intent_id: str, intent_type: str, payload: dict) -> Optional[IntentDecision]:
         """Return prior decision if intent_id already seen; else None (and reserve it)."""

trex_core/runtime/daemon.py

@@ -299,6 +299,7 @@ class Daemon:
             max_active_slots=config.max_active_slots,  # dcy-020: asset-centric gate
             default_order_timeout_sec=config.default_order_timeout_sec,  # dcy-022
             warn_store=self.warn_store,  # BAT-050 / REQ-104: reconcile-warning gate
+            alarm_emitter=self.alarm_emitter,  # req-0008 / ac-0025
             entry_allowed_fn=entry_allowed_fn,  # dcy-036 / REQ-114: day-start gate
             buy_guard_fn=(
                 self._account_snapshot_buy_guard