From a9e88f9770ea24b0823c17c321d2c4086d6f54f7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 26 Aug 2026 04:19:14 +0000 Subject: [PATCH] feat(intents): emit warn alarm on first reconcile_warned rejection Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../test_reconcile_warning_blocking.py | 241 ++++++++++++++++++ tests/runtime/test_daemon_resolve_symbol.py | 13 + trex_core/application/intents.py | 63 +++++ trex_core/runtime/daemon.py | 1 + 4 files changed, 318 insertions(+) diff --git a/tests/application/test_reconcile_warning_blocking.py b/tests/application/test_reconcile_warning_blocking.py index 4a2783a..f57f0e6 100644 --- a/tests/application/test_reconcile_warning_blocking.py +++ b/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() diff --git a/tests/runtime/test_daemon_resolve_symbol.py b/tests/runtime/test_daemon_resolve_symbol.py index e2b1bb9..8500150 100644 --- a/tests/runtime/test_daemon_resolve_symbol.py +++ b/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 diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index d881483..2f7cea4 100644 --- a/trex_core/application/intents.py +++ b/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).""" diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 29fc8dc..f1f68ea 100644 --- a/trex_core/runtime/daemon.py +++ b/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