From 153a563f6df051cb2e766920409898bfad627d56 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 14:50:29 +0000 Subject: [PATCH] feat(alerts): publish reconciliation quantity mismatches to trex-notifier (req-0007/ac-0024) Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../test_reconcile_notification.py | 841 ++++++++++++++++++ ...26_create_reconcile_notification_state.sql | 16 + .../portfolio/reconcile_notification_store.py | 122 +++ trex_core/application/alarm_emitter.py | 19 +- trex_core/application/portfolio.py | 159 +++- trex_core/application/reconciliation.py | 353 +++++++- 6 files changed, 1458 insertions(+), 52 deletions(-) create mode 100644 tests/application/test_reconcile_notification.py create mode 100644 trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql create mode 100644 trex_core/adapters/portfolio/reconcile_notification_store.py diff --git a/tests/application/test_reconcile_notification.py b/tests/application/test_reconcile_notification.py new file mode 100644 index 0000000..5769b29 --- /dev/null +++ b/tests/application/test_reconcile_notification.py @@ -0,0 +1,841 @@ +"""Reconciliation mismatch notification lifecycle tests (req-0007, ac-0024). + +Verifies B-case and C-case detected/changed/resolved alarm publication with +durable deduplication via ReconcileNotificationStore, including: + - first detection + - unchanged suppression (within and across simulated restarts) + - quantity change (changed event) + - multi-portfolio C-case aggregation (one event per symbol) + - B-case slot identity isolation + - B sync resolution + - C verify resolution + - publish failure: fingerprint not advanced, retry next detection + - resolved publish failure: warning clear preserved, pending_resolved set + - notification-state persistence failure: business path not aborted + - duplicate resolved: idempotent + - sensitive-field absence +""" +from __future__ import annotations + +from unittest.mock import MagicMock +from uuid import uuid4 + +from trex_core.adapters.messaging.in_memory import InMemoryTransport +from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.broker_port import BalanceItem +from trex_core.application.portfolio import ( + InMemoryReconcileNotificationStore, + InMemoryReconcileWarningStore, + PortfolioItem, + SnapshotStore, +) +from trex_core.application.reconciliation import ( + _c_identity_key, + reconcile_core_holdings, + reconcile_positions, + sync_slot_to_broker, + verify_slot_reconcile, +) +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository +from trex_core.domain.slot.models import Slot + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +INSTANCE = "test-instance" + + +def _make_emitter() -> tuple[AlarmEmitter, InMemoryTransport]: + transport = InMemoryTransport() + return AlarmEmitter(transport), transport + + +def _make_failing_emitter() -> AlarmEmitter: + """AlarmEmitter backed by a transport that always raises.""" + transport = InMemoryTransport() + transport.set_available(False) + return AlarmEmitter(transport) + + +def _alarm_codes(transport: InMemoryTransport) -> list[str]: + return [payload["code"] for _, payload in transport.alarm_events()] + + +def _alarm_fields_for(transport: InMemoryTransport, code: str) -> list[dict]: + return [ + payload["fields"] + for _, payload in transport.alarm_events() + if payload["code"] == code + ] + + +def _make_slot( + status: SlotState = SlotState.HOLDING, + qty: float = 10.0, + symbol: str = "005930", + pid: str = "pf-A", +) -> Slot: + return Slot( + slot_id=uuid4(), + status=status, + symbol=symbol, + qty=qty, + portfolio_id=pid, + instance_key=INSTANCE, + ) + + +def _snapshot(portfolio_id: str, asset_qty: float, symbol: str = "005930") -> SnapshotStore: + store = SnapshotStore() + store.swap([PortfolioItem( + portfolio_id=portfolio_id, + symbol=symbol, + cash_balance=0.0, + asset_quantity=asset_qty, + )]) + return store + + +class FakeBroker: + def __init__(self, balances: dict[str, int]): + self._b = balances + + def name(self) -> str: + return "fake" + + def submit_order(self, req): ... + def poll_fill(self, *args, **kw): ... + def cancel_order(self, *args, **kw): ... + def get_current_price(self, sym): return None + + def get_balance(self) -> list[BalanceItem]: + return [BalanceItem(symbol=sym, hldg_qty=qty) + for sym, qty in self._b.items() if qty > 0] + + +def _repo_with_slot(slot: Slot) -> InMemorySlotRepository: + repo = InMemorySlotRepository() + repo.add_slot(slot) + return repo + + +def _run_reconcile_positions( + hub_qty: int, + broker_qty: int, + symbol: str = "005930", + pid: str = "pf-A", + emitter: AlarmEmitter | None = None, + notification_store=None, + warn_store=None, + slot_repo=None, +) -> dict: + store = _snapshot(pid, float(hub_qty), symbol=symbol) + broker = FakeBroker({symbol: broker_qty}) + return reconcile_positions( + store=store, + broker=broker, + instance_key=INSTANCE, + alarm_emitter=emitter, + notification_store=notification_store, + warn_store=warn_store, + slot_repo=slot_repo, + ) + + +# --------------------------------------------------------------------------- +# B-case: reconcile_core_holdings +# --------------------------------------------------------------------------- + +def test_b_case_first_detection_publishes_detected(): + """New B-case slot mismatch → detected event published once.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + slot_service = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, + slot_service=slot_service, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, + notification_store=ns, + ) + + codes = _alarm_codes(transport) + assert codes == ["reconcile_mismatch_detected"] + state = ns.get_state(str(slot.slot_id)) + assert state is not None + assert state["lifecycle_state"] == "notified" + assert state["quantity_a"] == 10 + assert state["quantity_b"] == 7 + + +def test_b_case_unchanged_suppression(): + """Repeated tick with same qty tuple → no second event.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + slot_service = SlotService(repo) + + for _ in range(3): + reconcile_core_holdings( + slot_repo=repo, + slot_service=slot_service, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, + notification_store=ns, + ) + + assert len(transport.alarm_events()) == 1 + + +def test_b_case_process_restart_suppression(): + """After restart (new store seeded with old fingerprint), unchanged tick is suppressed.""" + emitter1, transport1 = _make_emitter() + ns1 = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter1, notification_store=ns1, + ) + assert len(transport1.alarm_events()) == 1 + + # Simulate restart: new emitter + new in-memory store, but seed it from ns1 state. + emitter2, transport2 = _make_emitter() + ns2 = InMemoryReconcileNotificationStore() + state = ns1.get_state(slot_id_str) + ns2.set_state(slot_id_str, **{k: v for k, v in state.items() if k != "identity_key"}) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter2, notification_store=ns2, + ) + assert len(transport2.alarm_events()) == 0 # suppressed + + +def test_b_case_quantity_change_publishes_changed(): + """B-case same slot, different qty tuple → changed event.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, notification_store=ns, + ) + + # Now broker qty changes → different fingerprint + # We need to update the slot qty in-place to simulate a changed qty scenario. + # Manually update notification state to simulate a different prior fingerprint. + ns.set_state(str(slot.slot_id), case_code="B", instance_key=INSTANCE, + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="notified") + + # New broker qty: 5 instead of 7 + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 5}), + instance_key=INSTANCE, + alarm_emitter=emitter, notification_store=ns, + ) + + codes = _alarm_codes(transport) + assert "reconcile_mismatch_changed" in codes + state = ns.get_state(str(slot.slot_id)) + assert state["quantity_b"] == 5 + + +def test_b_case_separate_slots_no_fingerprint_mixing(): + """Different B-case slot identities have independent fingerprints.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot_a = _make_slot(qty=10.0, symbol="005930", pid="pf-A") + slot_b = _make_slot(qty=20.0, symbol="000660", pid="pf-B") + repo = InMemorySlotRepository() + repo.add_slot(slot_a) + repo.add_slot(slot_b) + ss = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7, "000660": 15}), + instance_key=INSTANCE, + alarm_emitter=emitter, notification_store=ns, + ) + + # Both should get detected events + codes = _alarm_codes(transport) + assert codes.count("reconcile_mismatch_detected") == 2 + + state_a = ns.get_state(str(slot_a.slot_id)) + state_b = ns.get_state(str(slot_b.slot_id)) + assert state_a is not None + assert state_b is not None + # Fingerprints are independent + assert state_a["quantity_a"] == 10 + assert state_b["quantity_a"] == 20 + assert state_a["identity_key"] != state_b["identity_key"] + + +def test_b_case_publish_failure_fingerprint_not_advanced(): + """Transport failure → fingerprint not advanced → retry on next detection.""" + failing_emitter = _make_failing_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=failing_emitter, notification_store=ns, + ) + + # Fingerprint must not be advanced + assert ns.get_state(str(slot.slot_id)) is None + + # Next detection should try again (detected, not changed) + emitter2, transport2 = _make_emitter() + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter2, notification_store=ns, + ) + assert _alarm_codes(transport2) == ["reconcile_mismatch_detected"] + + +def test_b_case_no_notification_without_store(): + """Backward compat: no notification_store → no crash, no notification.""" + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + emitter, transport = _make_emitter() + + decisions = reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, + # no notification_store + ) + assert str(slot.slot_id) in decisions + assert len(transport.alarm_events()) == 0 + + +# --------------------------------------------------------------------------- +# B-case resolved: sync_slot_to_broker +# --------------------------------------------------------------------------- + +def test_b_sync_resolution_publishes_resolved(): + """operator sync clears warning → resolved event published.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="B") + + # Seed notification state (simulating prior detection) + ns.set_state(slot_id_str, case_code="B", instance_key=INSTANCE, + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="notified") + + sync_slot_to_broker( + slot_id=slot_id_str, + slot_repo=repo, + slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=emitter, + notification_store=ns, + ) + + codes = _alarm_codes(transport) + assert "reconcile_mismatch_resolved" in codes + # State cleared after resolved + assert ns.get_state(slot_id_str) is None + + +def test_b_sync_resolved_publish_failure_warning_clear_preserved(): + """Resolved publish fails → warning clear is preserved, pending_resolved set.""" + failing_emitter = _make_failing_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="B") + ns.set_state(slot_id_str, case_code="B", instance_key=INSTANCE, + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="notified") + + result = sync_slot_to_broker( + slot_id=slot_id_str, + slot_repo=repo, + slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=failing_emitter, + notification_store=ns, + ) + + # Warning clear preserved + assert result["warning_cleared"] is True + assert not ws.is_warned(slot_id_str) + # pending_resolved set for retry + state = ns.get_state(slot_id_str) + assert state is not None + assert state["lifecycle_state"] == "pending_resolved" + + +def test_b_sync_resolved_retry_on_next_reconcile(): + """After pending_resolved set, next reconcile pass retries resolved publish.""" + ns = InMemoryReconcileNotificationStore() + slot_id_str = "some-slot-id" + # Manually set pending_resolved state (warning already cleared) + ns.set_state(slot_id_str, case_code="B", instance_key=INSTANCE, + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="pending_resolved") + + emitter, transport = _make_emitter() + slot = _make_slot(qty=7.0) # qty matches broker now — not in active mismatch set + repo = _repo_with_slot(slot) + ss = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, notification_store=ns, + ) + + codes = _alarm_codes(transport) + assert "reconcile_mismatch_resolved" in codes + assert ns.get_state(slot_id_str) is None + + +def test_b_sync_duplicate_resolved_idempotent(): + """No notification state → resolved is a no-op (idempotent).""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="B") + # No notification state seeded — simulate never-detected + + sync_slot_to_broker( + slot_id=slot_id_str, + slot_repo=repo, + slot_service=ss, + broker=FakeBroker({"005930": 7}), # qty 10 != 7 so sync updates + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=emitter, + notification_store=ns, + ) + + # resolved alarm must not be published when no prior state + resolved = [c for c in _alarm_codes(transport) if c == "reconcile_mismatch_resolved"] + assert len(resolved) == 0 + + +# --------------------------------------------------------------------------- +# C-case: reconcile_positions +# --------------------------------------------------------------------------- + +def test_c_case_first_detection_publishes_detected(): + """New C-case symbol aggregate mismatch → detected event published once.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + + _run_reconcile_positions( + hub_qty=10, broker_qty=7, + emitter=emitter, notification_store=ns, + ) + + assert _alarm_codes(transport) == ["reconcile_mismatch_detected"] + c_key = _c_identity_key(INSTANCE, "005930") + state = ns.get_state(c_key) + assert state is not None + assert state["case_code"] == "C" + assert state["quantity_a"] == 10 + assert state["quantity_b"] == 7 + + +def test_c_case_multi_portfolio_single_event(): + """C-case multiple portfolios for same symbol → exactly one detected event.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + # pf-A has 6, pf-B has 4 → hub total 10 ≠ broker 7 + store = SnapshotStore() + store.swap([ + PortfolioItem(portfolio_id="pf-A", symbol="005930", cash_balance=0, asset_quantity=6.0), + PortfolioItem(portfolio_id="pf-B", symbol="005930", cash_balance=0, asset_quantity=4.0), + ]) + broker = FakeBroker({"005930": 7}) + + reconcile_positions( + store=store, + broker=broker, + instance_key=INSTANCE, + alarm_emitter=emitter, + notification_store=ns, + ) + + assert len(transport.alarm_events()) == 1 + assert _alarm_codes(transport) == ["reconcile_mismatch_detected"] + # Only one state entry for the symbol + c_key = _c_identity_key(INSTANCE, "005930") + state = ns.get_state(c_key) + assert state is not None + assert state["quantity_a"] == 10 # 6 + 4 + + +def test_c_case_unchanged_suppression(): + """Repeated C-case ticks with same qty tuple → no second event.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + + for _ in range(3): + _run_reconcile_positions(hub_qty=10, broker_qty=7, emitter=emitter, notification_store=ns) + + assert len(transport.alarm_events()) == 1 + + +def test_c_case_quantity_change_publishes_changed(): + """C-case qty tuple changes → changed event.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + + _run_reconcile_positions(hub_qty=10, broker_qty=7, emitter=emitter, notification_store=ns) + assert _alarm_codes(transport) == ["reconcile_mismatch_detected"] + + # Broker qty changes + _run_reconcile_positions(hub_qty=10, broker_qty=5, emitter=emitter, notification_store=ns) + codes = _alarm_codes(transport) + assert "reconcile_mismatch_changed" in codes + + +def test_c_case_publish_failure_fingerprint_not_advanced(): + """Transport failure → fingerprint not advanced → retry on next detection.""" + failing_emitter = _make_failing_emitter() + ns = InMemoryReconcileNotificationStore() + + _run_reconcile_positions( + hub_qty=10, broker_qty=7, emitter=failing_emitter, notification_store=ns, + ) + + c_key = _c_identity_key(INSTANCE, "005930") + assert ns.get_state(c_key) is None + + # Retry with working emitter + emitter2, transport2 = _make_emitter() + _run_reconcile_positions(hub_qty=10, broker_qty=7, emitter=emitter2, notification_store=ns) + assert _alarm_codes(transport2) == ["reconcile_mismatch_detected"] + + +def test_c_case_no_notification_without_store(): + """Backward compat: no alarm_emitter/notification_store → no crash.""" + decisions = _run_reconcile_positions(hub_qty=10, broker_qty=7) + assert "pf-A" in decisions + + +# --------------------------------------------------------------------------- +# C-case resolved: verify_slot_reconcile +# --------------------------------------------------------------------------- + +def test_c_verify_resolution_publishes_resolved(): + """operator verify clears C-case warning → resolved event published.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="C") + + # Seed notification state + c_key = _c_identity_key(INSTANCE, "005930") + ns.set_state(c_key, case_code="C", instance_key=INSTANCE, + symbol="005930", quantity_a=8, quantity_b=10, lifecycle_state="notified") + + store = _snapshot("pf-A", 10.0) # hub 10 == broker 10 → match + + verify_slot_reconcile( + portfolio_id="pf-A", + store=store, + broker=FakeBroker({"005930": 10}), + slot_repo=repo, + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=emitter, + notification_store=ns, + ) + + assert "reconcile_mismatch_resolved" in _alarm_codes(transport) + assert ns.get_state(c_key) is None + + +def test_c_verify_resolved_publish_failure_warning_clear_preserved(): + """Resolved publish fails → warning clear preserved, pending_resolved set.""" + failing_emitter = _make_failing_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="C") + + c_key = _c_identity_key(INSTANCE, "005930") + ns.set_state(c_key, case_code="C", instance_key=INSTANCE, + symbol="005930", quantity_a=8, quantity_b=10, lifecycle_state="notified") + + store = _snapshot("pf-A", 10.0) + result = verify_slot_reconcile( + portfolio_id="pf-A", + store=store, + broker=FakeBroker({"005930": 10}), + slot_repo=repo, + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=failing_emitter, + notification_store=ns, + ) + + # Warning cleared + assert result["result"] == "cleared" + assert not ws.is_warned(slot_id_str) + # pending_resolved set + state = ns.get_state(c_key) + assert state is not None + assert state["lifecycle_state"] == "pending_resolved" + + +def test_c_verify_duplicate_resolved_idempotent(): + """No prior notification state → verify resolved is a no-op.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="C") + + store = _snapshot("pf-A", 10.0) + verify_slot_reconcile( + portfolio_id="pf-A", + store=store, + broker=FakeBroker({"005930": 10}), + slot_repo=repo, + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=emitter, + notification_store=ns, + ) + + resolved = [c for c in _alarm_codes(transport) if c == "reconcile_mismatch_resolved"] + assert len(resolved) == 0 + + +def test_c_verify_resolved_retry_on_next_reconcile(): + """pending_resolved → next reconcile_positions pass retries resolved publish.""" + ns = InMemoryReconcileNotificationStore() + c_key = _c_identity_key(INSTANCE, "005930") + ns.set_state(c_key, case_code="C", instance_key=INSTANCE, + symbol="005930", quantity_a=8, quantity_b=10, lifecycle_state="pending_resolved") + + emitter, transport = _make_emitter() + # Simulate match (hub == broker) so no active mismatch + _run_reconcile_positions(hub_qty=10, broker_qty=10, emitter=emitter, notification_store=ns) + + assert "reconcile_mismatch_resolved" in _alarm_codes(transport) + assert ns.get_state(c_key) is None + + +# --------------------------------------------------------------------------- +# Notification-state persistence failure: business path not aborted +# --------------------------------------------------------------------------- + +def test_notification_state_persistence_failure_does_not_abort_reconcile(): + """notification_store raises → reconciliation completes normally.""" + bad_ns = MagicMock() + bad_ns.get_state.side_effect = RuntimeError("db down") + bad_ns.set_state.side_effect = RuntimeError("db down") + + emitter, _transport = _make_emitter() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + + decisions = reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, + notification_store=bad_ns, + ) + + # Business path: decisions produced, no crash + assert str(slot.slot_id) in decisions + assert decisions[str(slot.slot_id)] == "qty_warning" + + +def test_notification_state_persistence_failure_c_case_does_not_abort_reconcile(): + """C-case notification_store raises → reconcile_positions completes normally.""" + bad_ns = MagicMock() + bad_ns.get_state.side_effect = RuntimeError("db down") + bad_ns.list_states_by_lifecycle.side_effect = RuntimeError("db down") + + emitter, _transport = _make_emitter() + decisions = _run_reconcile_positions( + hub_qty=10, broker_qty=7, emitter=emitter, notification_store=bad_ns, + ) + assert "pf-A" in decisions + + +# --------------------------------------------------------------------------- +# Sensitive-field absence +# --------------------------------------------------------------------------- + +def test_b_case_alarm_fields_no_sensitive_data(): + """B-case alarm fields contain only allowlist values — no account/credential/DSN.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, notification_store=ns, + ) + + events = transport.alarm_events() + assert len(events) == 1 + _, payload = events[0] + fields = payload["fields"] + + # Allowlist: instance_key, symbol, case_code, broker_quantity, slot_quantity, slot_id + allowed = {"instance_key", "symbol", "case_code", "broker_quantity", "slot_quantity", "slot_id"} + extra = set(fields.keys()) - allowed + assert extra == set(), f"Unexpected fields: {extra}" + + # No sensitive values + all_values = " ".join(str(v) for v in fields.values()) + for forbidden in ("account", "credential", "token", "dsn", "postgresql", "amqp", "secret"): + assert forbidden not in all_values.lower(), f"Sensitive token '{forbidden}' found in fields" + + +def test_c_case_alarm_fields_no_sensitive_data(): + """C-case alarm fields contain only allowlist values.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + + _run_reconcile_positions(hub_qty=10, broker_qty=7, emitter=emitter, notification_store=ns) + + _, payload = transport.alarm_events()[0] + fields = payload["fields"] + + allowed = {"instance_key", "symbol", "case_code", "broker_quantity", "hub_quantity"} + extra = set(fields.keys()) - allowed + assert extra == set(), f"Unexpected fields: {extra}" + + # hub_quantity present, slot_quantity absent + assert "hub_quantity" in fields + assert "slot_quantity" not in fields + + +def test_resolved_alarm_fields_b_case(): + """Resolved B-case alarm fields include allowlist values only.""" + emitter, transport = _make_emitter() + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + ws = InMemoryReconcileWarningStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="B") + ns.set_state(slot_id_str, case_code="B", instance_key=INSTANCE, + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="notified") + + sync_slot_to_broker( + slot_id=slot_id_str, + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=emitter, notification_store=ns, + ) + + resolved_events = [ + (rk, p) for rk, p in transport.alarm_events() + if p["code"] == "reconcile_mismatch_resolved" + ] + assert len(resolved_events) == 1 + _, payload = resolved_events[0] + fields = payload["fields"] + allowed = {"instance_key", "symbol", "case_code", "broker_quantity", "slot_quantity", "slot_id"} + extra = set(fields.keys()) - allowed + assert extra == set(), f"Unexpected fields: {extra}" + + +# --------------------------------------------------------------------------- +# Existing behavior preservation +# --------------------------------------------------------------------------- + +def test_b_case_existing_alarm_emitter_still_works_without_notification_store(): + """AlarmEmitter existing callers: no notification_store → no regression.""" + emitter, transport = _make_emitter() + slot = _make_slot(qty=10.0) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + ws = InMemoryReconcileWarningStore() + + reconcile_core_holdings( + slot_repo=repo, slot_service=ss, + broker=FakeBroker({"005930": 7}), + instance_key=INSTANCE, + alarm_emitter=emitter, + warn_store=ws, + ) + + # No alarm from B-case without notification_store (only existing A-case alarm path unchanged) + assert len(transport.alarm_events()) == 0 + assert ws.is_warned(str(slot.slot_id)) + + +def test_publish_alarm_returns_bool(): + """AlarmEmitter.publish_alarm returns True on success, False on failure.""" + emitter, _ = _make_emitter() + result = emitter.publish_alarm("test_code", "Test title") + assert result is True + + failing = _make_failing_emitter() + result2 = failing.publish_alarm("test_code", "Test title") + assert result2 is False diff --git a/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql b/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql new file mode 100644 index 0000000..7593e74 --- /dev/null +++ b/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql @@ -0,0 +1,16 @@ +-- req-0007 / ac-0024: durable mismatch notification lifecycle state. +-- PK: identity_key (slot_id TEXT for B-case; "{instance_key}\x00C\x00{symbol}" for C-case). +-- lifecycle_state: 'notified' | 'pending_resolved'. +-- ON CONFLICT DO UPDATE: replay-safe; re-running migration is a no-op. +CREATE TABLE IF NOT EXISTS reconcile_notification_state ( + identity_key TEXT PRIMARY KEY, + case_code TEXT NOT NULL CHECK (case_code IN ('B', 'C')), + instance_key TEXT NOT NULL, + symbol TEXT NOT NULL, + quantity_a INTEGER NOT NULL, + quantity_b INTEGER NOT NULL, + lifecycle_state TEXT NOT NULL + CHECK (lifecycle_state IN ('notified', 'pending_resolved')), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/trex_core/adapters/portfolio/reconcile_notification_store.py b/trex_core/adapters/portfolio/reconcile_notification_store.py new file mode 100644 index 0000000..2d752ff --- /dev/null +++ b/trex_core/adapters/portfolio/reconcile_notification_store.py @@ -0,0 +1,122 @@ +"""PostgreSQL-backed durable reconcile notification lifecycle store (req-0007, ac-0024). + +Tracks per-identity B/C-case mismatch notification fingerprints so that +duplicate alarms are suppressed across scheduler ticks and process restarts, +and pending resolved transitions can be retried. + +PK: identity_key (slot_id for B-case; "{instance_key}\\x00C\\x00{symbol}" for C-case). +lifecycle_state: 'notified' | 'pending_resolved'. +""" +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from trex_core.adapters.persistence.postgresql import _require_psycopg + + +class PostgreSQLReconcileNotificationStore: + """`reconcile_notification_state` table CRUD (req-0007, ac-0024).""" + + def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: + self._psycopg = _require_psycopg(psycopg_module) + self._dsn = dsn + + def get_state(self, identity_key: str) -> dict | None: + """Return state dict or None.""" + with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur: + cur.execute( + "SELECT identity_key, case_code, instance_key, symbol, " + " quantity_a, quantity_b, lifecycle_state " + "FROM reconcile_notification_state " + "WHERE identity_key = %s LIMIT 1", + (identity_key,), + ) + row = cur.fetchone() + if row is None: + return None + return { + "identity_key": row[0], + "case_code": row[1], + "instance_key": row[2], + "symbol": row[3], + "quantity_a": row[4], + "quantity_b": row[5], + "lifecycle_state": row[6], + } + + def set_state( + self, + identity_key: str, + *, + case_code: str, + instance_key: str, + symbol: str, + quantity_a: int, + quantity_b: int, + lifecycle_state: str, + ) -> None: + """Upsert notification state.""" + now = datetime.now(tz=UTC) + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO reconcile_notification_state " + "(identity_key, case_code, instance_key, symbol, " + " quantity_a, quantity_b, lifecycle_state, created_at, updated_at) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) " + "ON CONFLICT (identity_key) DO UPDATE " + "SET case_code = EXCLUDED.case_code, " + " instance_key = EXCLUDED.instance_key, " + " symbol = EXCLUDED.symbol, " + " quantity_a = EXCLUDED.quantity_a, " + " quantity_b = EXCLUDED.quantity_b, " + " lifecycle_state = EXCLUDED.lifecycle_state, " + " updated_at = EXCLUDED.updated_at", + ( + identity_key, + case_code, + instance_key, + symbol, + quantity_a, + quantity_b, + lifecycle_state, + now, + now, + ), + ) + conn.commit() + + def clear_state(self, identity_key: str) -> None: + """Delete state row (called after successful resolved publish).""" + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "DELETE FROM reconcile_notification_state WHERE identity_key = %s", + (identity_key,), + ) + conn.commit() + + def list_states_by_lifecycle(self, lifecycle_state: str) -> list[dict]: + """Return all state dicts with the given lifecycle_state.""" + with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur: + cur.execute( + "SELECT identity_key, case_code, instance_key, symbol, " + " quantity_a, quantity_b, lifecycle_state " + "FROM reconcile_notification_state " + "WHERE lifecycle_state = %s", + (lifecycle_state,), + ) + rows = cur.fetchall() + return [ + { + "identity_key": row[0], + "case_code": row[1], + "instance_key": row[2], + "symbol": row[3], + "quantity_a": row[4], + "quantity_b": row[5], + "lifecycle_state": row[6], + } + for row in rows + ] diff --git a/trex_core/application/alarm_emitter.py b/trex_core/application/alarm_emitter.py index a045c8e..37b2a7a 100644 --- a/trex_core/application/alarm_emitter.py +++ b/trex_core/application/alarm_emitter.py @@ -17,11 +17,10 @@ where no broker connection exists. In that case publish_alarm() logs and skips. from __future__ import annotations import logging -from datetime import datetime, timezone, timedelta -from typing import Optional +from datetime import datetime, timedelta, timezone -from trex_core.application.messaging import MessagingTransport, TransportUnavailableError from trex_core.adapters.messaging.topology import alarm_routing_key +from trex_core.application.messaging import MessagingTransport, TransportUnavailableError _log = logging.getLogger(__name__) @@ -41,7 +40,7 @@ class AlarmEmitter: - None-transport guard for non-daemon entry points """ - def __init__(self, transport: Optional[MessagingTransport]) -> None: + def __init__(self, transport: MessagingTransport | None) -> None: self._transport = transport def publish_alarm( @@ -49,9 +48,9 @@ class AlarmEmitter: code: str, title: str, *, - fields: Optional[dict] = None, + fields: dict | None = None, severity: str = "critical", - ) -> None: + ) -> bool: """Publish a single alarm envelope. Best-effort: never re-raises. Args: @@ -60,13 +59,16 @@ class AlarmEmitter: exception bodies — callers must sanitize. fields: Optional structured context dict (opaque, default {}). severity: 'critical' | 'warn' | 'info'. Default 'critical'. + + Returns: + True if the alarm was successfully published, False otherwise. """ if self._transport is None: _log.info( "alarm skipped (no transport): code=%s title=%s", code, title, ) - return + return False routing_key = alarm_routing_key(severity, _SOURCE) occurred_at = datetime.now(tz=_KST).isoformat() @@ -81,13 +83,16 @@ class AlarmEmitter: } try: self._transport.publish_alarm_event(routing_key, payload) + return True except TransportUnavailableError as exc: _log.error( "alarm publish skipped (transport unavailable): code=%s: %s", code, exc, ) + return False except Exception: # noqa: BLE001 — alarm must never crash the caller _log.error( "alarm publish failed: code=%s title=%s", code, title, exc_info=True, ) + return False diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index d4723ee..f9bfbf6 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -17,10 +17,9 @@ from __future__ import annotations import threading from dataclasses import dataclass, field, replace -from datetime import datetime, timezone +from datetime import UTC, datetime from enum import Enum -from typing import Optional, Protocol - +from typing import Protocol # ---------------------------------------------------------------- portfolio-hub @@ -38,10 +37,10 @@ class PortfolioItem: symbol: str cash_balance: float asset_quantity: float # integer shares semantically; float for tolerance - active_strategy_id: Optional[str] = None + active_strategy_id: str | None = None strategy_params: dict = field(default_factory=dict) # opaque, pass-through # REQ-116 additive fields (BAT-055) — safe defaults for legacy hub responses - strategy_name: Optional[str] = None # portfolio-hub strategy_name; None if absent + strategy_name: str | None = None # portfolio-hub strategy_name; None if absent markets: list = field(default_factory=list) # portfolio-hub markets list; [] if absent @@ -63,7 +62,7 @@ class RealizedPnlItem: Minimum required: symbol, realized_pnl, sell_qty. Additional fields preserved for completeness. """ - symbol: Optional[str] + symbol: str | None realized_pnl: float sell_qty: float cost_basis_per_unit: float = 0.0 @@ -78,7 +77,7 @@ class RealizedPnlResult: Returned by get_realized_pnl port method. """ items: list[RealizedPnlItem] - cost_basis_from: Optional[str] # ISO date string or None + cost_basis_from: str | None # ISO date string or None data_completeness_warning: bool # True when portfolio-hub signals incomplete history total_realized_pnl: float @@ -96,7 +95,7 @@ class BrokerCredentialPayload: account_ref: str app_key: str app_secret: str - credential_version: Optional[int] = None + credential_version: int | None = None @dataclass(frozen=True) @@ -106,7 +105,7 @@ class TradingContext: Returned by fetch_broker_credential() in a single atomic read to avoid double HTTP calls. """ - broker: Optional[BrokerCredentialPayload] + broker: BrokerCredentialPayload | None trading_enabled: bool @@ -132,8 +131,8 @@ class PortfolioHubReadPort(Protocol): from_dt: str, to_dt: str, *, - symbol: Optional[str] = None, - ) -> "RealizedPnlResult": + symbol: str | None = None, + ) -> RealizedPnlResult: """GET /v1/portfolios/{portfolio_id}/pnl/realized — WAC realized PnL (REQ-109). Args: @@ -151,7 +150,7 @@ class PortfolioHubReadPort(Protocol): """ ... - def fetch_broker_credential(self) -> "TradingContext": + def fetch_broker_credential(self) -> TradingContext: """GET /v1/user_info — broker credential + trading_enabled (REQ-112). Returns a TradingContext with: @@ -181,7 +180,7 @@ class SnapshotAsset: """A portfolio item augmented with overlays (warning, sequence).""" item: PortfolioItem warning: bool = False - updated_at: Optional[datetime] = None + updated_at: datetime | None = None class SnapshotStore: @@ -198,14 +197,14 @@ class SnapshotStore: def __init__(self) -> None: self._lock = threading.Lock() self._assets: dict[str, SnapshotAsset] = {} - self._degraded: Optional[DegradedReason] = None - self._degraded_message: Optional[str] = None - self._last_refresh_at: Optional[datetime] = None + self._degraded: DegradedReason | None = None + self._degraded_message: str | None = None + self._last_refresh_at: datetime | None = None # --- universe --- def swap(self, items: list[PortfolioItem]) -> None: """Replace the entire universe atomically (immutable swap, Decision 6).""" - now = datetime.now(tz=timezone.utc) + now = datetime.now(tz=UTC) with self._lock: prior_warnings = {pid for pid, a in self._assets.items() if a.warning} new_assets: dict[str, SnapshotAsset] = {} @@ -225,7 +224,7 @@ class SnapshotStore: with self._lock: return list(self._assets.values()) - def get(self, portfolio_id: str) -> Optional[SnapshotAsset]: + def get(self, portfolio_id: str) -> SnapshotAsset | None: with self._lock: return self._assets.get(portfolio_id) @@ -256,7 +255,7 @@ class SnapshotStore: self._assets[portfolio_id] = replace( a, item=replace(a.item, cash_balance=cash_balance, asset_quantity=asset_quantity), - updated_at=datetime.now(tz=timezone.utc), + updated_at=datetime.now(tz=UTC), ) # --- degraded signal (REQ-028 AC-2) --- @@ -270,11 +269,11 @@ class SnapshotStore: self._degraded = None self._degraded_message = None - def degraded_state(self) -> tuple[Optional[DegradedReason], Optional[str]]: + def degraded_state(self) -> tuple[DegradedReason | None, str | None]: with self._lock: return self._degraded, self._degraded_message - def last_refresh_at(self) -> Optional[datetime]: + def last_refresh_at(self) -> datetime | None: with self._lock: return self._last_refresh_at @@ -297,7 +296,7 @@ class PauseStore(Protocol): class InMemoryPauseStore: """Process-local pause store for tests + CI.""" - def __init__(self, initial: Optional[set[str]] = None) -> None: + def __init__(self, initial: set[str] | None = None) -> None: self._lock = threading.Lock() self._paused: set[str] = set(initial or set()) @@ -379,10 +378,10 @@ class ReconcileWarningStore(Protocol): def is_warned(self, slot_id: str) -> bool: ... - def list_warned(self) -> "set[str]": + def list_warned(self) -> set[str]: ... - def get_warning_info(self, slot_id: str) -> "Optional[dict]": + def get_warning_info(self, slot_id: str) -> dict | None: """Return {"portfolio_id", "case_code", "actor", "reason"} or None. Used by verify to determine whether the warning is case_code='C' @@ -391,7 +390,7 @@ class ReconcileWarningStore(Protocol): """ ... - def list_c_case_warned(self) -> "set[str]": + def list_c_case_warned(self) -> set[str]: """Return set of slot_id strings that have an active case_code='C' warning. AC-0023: used by strategy slots RPC to exclude C-case mismatched holding slots. @@ -432,27 +431,119 @@ class InMemoryReconcileWarningStore: with self._lock: return slot_id in self._warned - def list_warned(self) -> "set[str]": + def list_warned(self) -> set[str]: with self._lock: return set(self._warned.keys()) - def get_warning_info(self, slot_id: str) -> "Optional[dict]": + def get_warning_info(self, slot_id: str) -> dict | None: with self._lock: info = self._warned.get(slot_id) return dict(info) if info is not None else None - def list_c_case_warned(self) -> "set[str]": + def list_c_case_warned(self) -> set[str]: with self._lock: return {sid for sid, info in self._warned.items() if info.get("case_code") == "C"} +class ReconcileNotificationStore(Protocol): + """Durable mismatch notification lifecycle store (req-0007, ac-0024). + + Tracks the last successfully published fingerprint for each B/C-case identity + to suppress duplicate alarms across scheduler ticks and process restarts. + + Identity keys: + B-case: slot_id (UUID string) + C-case: `"{instance_key}\\x00C\\x00{symbol}"` (null-separated to avoid collisions) + + lifecycle_state values: + "notified" — detected or changed alarm was published for this fingerprint + "pending_resolved" — warning was cleared but resolved publish failed; retry pending + """ + + def get_state(self, identity_key: str) -> dict | None: + """Return state dict or None. + + State dict keys: identity_key, case_code, instance_key, symbol, + quantity_a, quantity_b, lifecycle_state. + """ + ... + + def set_state( + self, + identity_key: str, + *, + case_code: str, + instance_key: str, + symbol: str, + quantity_a: int, + quantity_b: int, + lifecycle_state: str, + ) -> None: + """Upsert notification state (call after successful publish).""" + ... + + def clear_state(self, identity_key: str) -> None: + """Remove state (call after successful resolved publish).""" + ... + + def list_states_by_lifecycle(self, lifecycle_state: str) -> list[dict]: + """Return all state dicts with the given lifecycle_state.""" + ... + + +class InMemoryReconcileNotificationStore: + """Process-local notification lifecycle store for tests + CI (req-0007).""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._states: dict[str, dict] = {} + + def get_state(self, identity_key: str) -> dict | None: + with self._lock: + s = self._states.get(identity_key) + return dict(s) if s is not None else None + + def set_state( + self, + identity_key: str, + *, + case_code: str, + instance_key: str, + symbol: str, + quantity_a: int, + quantity_b: int, + lifecycle_state: str, + ) -> None: + with self._lock: + self._states[identity_key] = { + "identity_key": identity_key, + "case_code": case_code, + "instance_key": instance_key, + "symbol": symbol, + "quantity_a": quantity_a, + "quantity_b": quantity_b, + "lifecycle_state": lifecycle_state, + } + + def clear_state(self, identity_key: str) -> None: + with self._lock: + self._states.pop(identity_key, None) + + def list_states_by_lifecycle(self, lifecycle_state: str) -> list[dict]: + with self._lock: + return [ + dict(s) for s in self._states.values() + if s["lifecycle_state"] == lifecycle_state + ] + + def slot_to_dict( s, *, - snapshot_store: "Optional[SnapshotStore]" = None, + snapshot_store: SnapshotStore | None = None, watchdog_store=None, - pause_store: "Optional[PauseStore]" = None, - warn_store: "Optional[ReconcileWarningStore]" = None, + pause_store: PauseStore | None = None, + warn_store: ReconcileWarningStore | None = None, watchdog_timeout_sec: int = 120, ) -> dict: """Merged slot row for trex.slots.req (REQ-073/074). @@ -479,9 +570,9 @@ def slot_to_dict( # hub authority: asset_quantity (hub primary) + cash_balance # REQ-116: strategy_name / strategy_params / markets also from PortfolioItem - asset_qty: Optional[float] = None - cash_balance: Optional[float] = None - strategy_name_val: Optional[str] = None + asset_qty: float | None = None + cash_balance: float | None = None + strategy_name_val: str | None = None strategy_params_val: dict = {} markets_val: list = [] if snapshot_store is not None and s.portfolio_id: diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index d0c23ab..48fc582 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -19,13 +19,16 @@ durable store is additive and Optional (None => warn-log-only, backward compat). BAT-051 / REQ-105: sync_slot_to_broker — operator CLI sync path. BAT-051 / REQ-106: verify_slot_reconcile — operator CLI verify path (C-only, read-centric). + +req-0007 / ac-0024: B/C-case mismatch alarms (detected, changed, resolved) are +published via AlarmEmitter with durable deduplication through ReconcileNotificationStore. """ from __future__ import annotations import logging import math import re -from typing import Optional, cast +from typing import cast from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import BrokerPort @@ -47,7 +50,7 @@ _COST_BASIS_REASON_CODES = frozenset({ _COST_BASIS_LEGACY_SOURCE_ALIASES = frozenset({"avg_cost", "wac", "cost_basis"}) -def _sanitize_cost_basis_reason(reason: object) -> Optional[str]: +def _sanitize_cost_basis_reason(reason: object) -> str | None: """Keep public diagnostics in the bounded REQ-123 reason taxonomy.""" if not isinstance(reason, str): return None @@ -62,7 +65,7 @@ def _normalize_cost_basis_diagnostics( value: object, source: object, reason: object, -) -> tuple[Optional[float], str, Optional[str]]: +) -> tuple[float | None, str, str | None]: """Normalize diagnostic-port output before it reaches reconciliation results.""" safe_reason = _sanitize_cost_basis_reason(reason) if source == "canonical": @@ -89,7 +92,7 @@ def _normalize_cost_basis_diagnostics( return normalized_value, normalized_source, safe_reason -def _normalize_legacy_cost_basis(value: object) -> tuple[Optional[float], str, Optional[str]]: +def _normalize_legacy_cost_basis(value: object) -> tuple[float | None, str, str | None]: """Make the diagnostic-free compatibility seam obey safe WAC semantics.""" normalized_value, _, _ = _normalize_cost_basis_diagnostics(value, "legacy", "invalid") if normalized_value is None: @@ -97,14 +100,233 @@ def _normalize_legacy_cost_basis(value: object) -> tuple[Optional[float], str, O return normalized_value, "legacy", None +# --------------------------------------------------------------------------- +# req-0007 / ac-0024: Notification lifecycle helpers +# --------------------------------------------------------------------------- + +def _c_identity_key(instance_key: str, symbol: str) -> str: + """Stable C-case notification identity key (null-byte separated to avoid collisions).""" + return f"{instance_key}\x00C\x00{symbol}" + + +def _build_alarm_fields( + *, + case_code: str, + instance_key: str, + symbol: str, + quantity_a: int, + quantity_b: int, + slot_id: str | None = None, +) -> dict: + """Build sanitized alarm context fields (ac-0024 allowlist). + + Excluded: account number/hash, credential, token, DSN, raw payload, exception. + """ + fields: dict = { + "instance_key": instance_key, + "symbol": symbol, + "case_code": case_code, + "broker_quantity": quantity_b, + } + if case_code == "B": + fields["slot_quantity"] = quantity_a + if slot_id is not None: + fields["slot_id"] = slot_id + else: + fields["hub_quantity"] = quantity_a + return fields + + +def _notify_mismatch( + alarm_emitter: AlarmEmitter, + notification_store, + *, + identity_key: str, + case_code: str, + instance_key: str, + symbol: str, + quantity_a: int, + quantity_b: int, + slot_id: str | None = None, +) -> None: + """Publish detected or changed alarm; advance fingerprint only on publish success. + + Suppresses if the identity fingerprint is unchanged from the last notified state. + Retries on every detection if publish previously failed (fingerprint not advanced). + """ + try: + state = notification_store.get_state(identity_key) + except Exception: # noqa: BLE001 + _log.exception( + "reconcile_notification: get_state failed for %s — skipping notify", identity_key, + ) + return + + if state is not None and state.get("lifecycle_state") == "notified": + if state.get("quantity_a") == quantity_a and state.get("quantity_b") == quantity_b: + return # unchanged — suppress + event_type = "changed" + else: + # No prior state, OR pending_resolved (new mismatch after failed resolved) → re-detect + event_type = "detected" + + fields = _build_alarm_fields( + case_code=case_code, + instance_key=instance_key, + symbol=symbol, + quantity_a=quantity_a, + quantity_b=quantity_b, + slot_id=slot_id, + ) + published = alarm_emitter.publish_alarm( + f"reconcile_mismatch_{event_type}", + f"reconciliation mismatch {event_type}: {case_code}-case symbol={symbol}", + fields=fields, + ) + + if published: + try: + notification_store.set_state( + identity_key, + case_code=case_code, + instance_key=instance_key, + symbol=symbol, + quantity_a=quantity_a, + quantity_b=quantity_b, + lifecycle_state="notified", + ) + except Exception: # noqa: BLE001 + # Fingerprint write failure: at-least-once duplicate on next detection. + _log.exception( + "reconcile_notification: fingerprint persist failed for %s — " + "at-least-once duplicate possible on next detection", + identity_key, + ) + + +def _notify_resolved( + alarm_emitter: AlarmEmitter, + notification_store, + *, + identity_key: str, + case_code: str, + instance_key: str, + symbol: str, + slot_id: str | None = None, +) -> None: + """Publish resolved alarm; clear or mark pending_resolved on publish failure. + + Warning clear is always preserved by the caller regardless of this function's outcome. + - Publish success: clear notification state. + - Publish failure + state writable: mark pending_resolved for retry. + - State persistence failure: log loss risk (business path unaffected). + """ + try: + state = notification_store.get_state(identity_key) + except Exception: # noqa: BLE001 + _log.exception( + "reconcile_notification: get_state failed for %s — skipping resolved notify", + identity_key, + ) + return + + if state is None: + return # idempotent: already resolved or never detected + + quantity_a = state["quantity_a"] + quantity_b = state["quantity_b"] + + fields = _build_alarm_fields( + case_code=case_code, + instance_key=instance_key, + symbol=symbol, + quantity_a=quantity_a, + quantity_b=quantity_b, + slot_id=slot_id, + ) + published = alarm_emitter.publish_alarm( + "reconcile_mismatch_resolved", + f"reconciliation mismatch resolved: {case_code}-case symbol={symbol}", + fields=fields, + ) + + if published: + try: + notification_store.clear_state(identity_key) + except Exception: # noqa: BLE001 + # At-least-once: retry may produce duplicate resolved event. + _log.exception( + "reconcile_notification: clear_state failed for %s — " + "at-least-once duplicate resolved possible", + identity_key, + ) + else: + # Publish failed — persist pending_resolved so the next opportunity retries. + try: + notification_store.set_state( + identity_key, + case_code=case_code, + instance_key=instance_key, + symbol=symbol, + quantity_a=quantity_a, + quantity_b=quantity_b, + lifecycle_state="pending_resolved", + ) + except Exception: # noqa: BLE001 + # State persistence failed: warning clear is kept; resolved event may be lost. + _log.error( + "reconcile_notification: resolved publish failed AND state persist failed " + "for %s — resolved event may be lost", + identity_key, + ) + + +def _retry_pending_resolved( + alarm_emitter: AlarmEmitter, + notification_store, + *, + case_code: str, + active_identity_keys: set[str], +) -> None: + """Retry pending_resolved publishes for identities no longer actively mismatched. + + Called at the end of each reconcile pass to drain the pending_resolved queue + for identities that have since been cleared by operator sync/verify. + """ + try: + pending = notification_store.list_states_by_lifecycle("pending_resolved") + except Exception: # noqa: BLE001 + _log.exception( + "reconcile_notification: list_states_by_lifecycle failed — skipping pending retry", + ) + return + + for state in pending: + if state.get("case_code") != case_code: + continue + identity_key = state["identity_key"] + if identity_key in active_identity_keys: + continue # still mismatched — do not retry resolved yet + _notify_resolved( + alarm_emitter, + notification_store, + identity_key=identity_key, + case_code=state["case_code"], + instance_key=state["instance_key"], + symbol=state["symbol"], + slot_id=identity_key if case_code == "B" else None, + ) + + def reconcile_core_holdings( *, slot_repo, slot_service, broker: BrokerPort, instance_key: str, - alarm_emitter: Optional[AlarmEmitter] = None, - warn_store: Optional[ReconcileWarningStore] = None, + alarm_emitter: AlarmEmitter | None = None, + warn_store: ReconcileWarningStore | None = None, + notification_store=None, ) -> dict[str, str]: """dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding vs broker divergence and set durable reconcile-warning for B case. @@ -122,6 +344,7 @@ def reconcile_core_holdings( → set by reconcile_positions warn_store: Optional — when None, WARN-log-only (backward compat, REQ-103 AC-5). + notification_store: Optional — when None, alarm deduplication is skipped (backward compat). Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'| 'managed_non_holding_divergence'} for telemetry. """ @@ -130,6 +353,7 @@ def reconcile_core_holdings( } decisions: dict[str, str] = {} matched: set[str] = set() + active_b_identity_keys: set[str] = set() # req-0007: track active B mismatches for retry for slot in slot_repo.list_holding_slots(instance_key): sym = slot.symbol matched.add(sym) @@ -157,6 +381,7 @@ def reconcile_core_holdings( elif slot.qty is not None and int(slot.qty) != bq: # B case (REQ-103) — qty mismatch slot_id_str = str(slot.slot_id) decisions[slot_id_str] = "qty_warning" + active_b_identity_keys.add(slot_id_str) # REQ-107: structured WARNING log for B case. _log.warning( "reconcile_warning set: slot_id=%s portfolio_id=%s symbol=%s " @@ -177,6 +402,24 @@ def reconcile_core_holdings( _log.exception( "failed to set reconcile_warning for slot %s", slot.slot_id ) + # req-0007 / ac-0024: publish B-case detected/changed alarm (deduped). + if alarm_emitter is not None and notification_store is not None: + try: + _notify_mismatch( + alarm_emitter, + notification_store, + identity_key=slot_id_str, + case_code="B", + instance_key=instance_key, + symbol=sym, + quantity_a=int(slot.qty), + quantity_b=bq, + slot_id=slot_id_str, + ) + except Exception: # noqa: BLE001 + _log.exception( + "reconcile_core_holdings: notify_mismatch failed for slot=%s", slot_id_str, + ) else: decisions[str(slot.slot_id)] = "ok" for sym, bq in broker_qty.items(): # unmatched broker position @@ -196,6 +439,17 @@ def reconcile_core_holdings( "unmanaged position: symbol=%s qty=%d instance_key=%s", sym, bq, instance_key, ) + # req-0007 / ac-0024: retry pending_resolved for B-case identities no longer mismatched. + if alarm_emitter is not None and notification_store is not None: + try: + _retry_pending_resolved( + alarm_emitter, + notification_store, + case_code="B", + active_identity_keys=active_b_identity_keys, + ) + except Exception: # noqa: BLE001 + _log.exception("reconcile_core_holdings: pending_resolved retry failed") return decisions @@ -205,7 +459,9 @@ def reconcile_positions( broker: BrokerPort, slot_repo=None, instance_key: str = "default", - warn_store: Optional[ReconcileWarningStore] = None, + warn_store: ReconcileWarningStore | None = None, + alarm_emitter: AlarmEmitter | None = None, + notification_store=None, ) -> dict[str, str]: """Run one reconciliation pass (REQ-031 + REQ-103 BAT-050). @@ -216,6 +472,10 @@ def reconcile_positions( (ledger/hub qty vs broker) also set durable reconcile-warning on the corresponding slot (portfolio_id -> slot lookup via slot_repo). When either is None, falls back to in-memory warning only (backward compat, REQ-103 AC-5). + + req-0007 / ac-0024: when alarm_emitter and notification_store are provided, + C-case detected/changed alarms are published per-symbol (one aggregate event + per symbol regardless of portfolio count). """ broker_qty: dict[str, int] = { item.symbol: int(item.hldg_qty) for item in broker.get_balance() @@ -230,6 +490,7 @@ def reconcile_positions( by_symbol.setdefault(sym, []).append((a.item.portfolio_id, a.item.asset_quantity)) decisions: dict[str, str] = {} + active_c_identity_keys: set[str] = set() # req-0007: track active C mismatches for retry for symbol, entries in by_symbol.items(): hub_total = int(sum(qty for _, qty in entries)) broker_total = broker_qty.get(symbol, 0) @@ -277,6 +538,41 @@ def reconcile_positions( decisions[pid] = "cleared" else: decisions[pid] = "ok" + + # req-0007 / ac-0024: publish one C-case detected/changed alarm per symbol + # (aggregate mismatch lifecycle — not per portfolio). + if alarm_emitter is not None and notification_store is not None: + c_key = _c_identity_key(instance_key, symbol) + if not match: + active_c_identity_keys.add(c_key) + try: + _notify_mismatch( + alarm_emitter, + notification_store, + identity_key=c_key, + case_code="C", + instance_key=instance_key, + symbol=symbol, + quantity_a=hub_total, + quantity_b=broker_total, + ) + except Exception: # noqa: BLE001 + _log.exception( + "reconcile_positions: notify_mismatch failed for symbol=%s", symbol, + ) + + # req-0007 / ac-0024: retry pending_resolved for C-case identities no longer mismatched. + if alarm_emitter is not None and notification_store is not None: + try: + _retry_pending_resolved( + alarm_emitter, + notification_store, + case_code="C", + active_identity_keys=active_c_identity_keys, + ) + except Exception: # noqa: BLE001 + _log.exception("reconcile_positions: pending_resolved retry failed") + return decisions @@ -291,9 +587,11 @@ def sync_slot_to_broker( slot_service, broker: BrokerPort, instance_key: str, - warn_store: "ReconcileWarningStore", + warn_store: ReconcileWarningStore, cost_basis_port=None, # Optional[PortfolioHubCostBasisPort] dry_run: bool = False, + alarm_emitter: AlarmEmitter | None = None, + notification_store=None, ) -> dict: """Sync one slot's qty to the broker ground truth (REQ-105, BAT-051). @@ -373,9 +671,9 @@ def sync_slot_to_broker( avg_price = slot.avg_price seeded_avg_price = False cost_basis_source = "none" - cost_basis_reason: Optional[str] = None + cost_basis_reason: str | None = None - def _set_cost_basis_unavailable_message() -> tuple[str, Optional[str]]: + def _set_cost_basis_unavailable_message() -> tuple[str, str | None]: if pid: return "cost_basis_unavailable", f"inactive slot {slot_id} portfolio_id={pid}" return "cost_basis_unavailable", "inactive slot without portfolio_id" @@ -559,6 +857,22 @@ def sync_slot_to_broker( ) except Exception: # noqa: BLE001 _log.exception("sync: failed to clear_warned for slot=%s", slot_id) + # req-0007 / ac-0024: publish B-case resolved alarm (best-effort, warning clear kept). + if alarm_emitter is not None and notification_store is not None: + try: + _notify_resolved( + alarm_emitter, + notification_store, + identity_key=slot_id, + case_code="B", + instance_key=instance_key, + symbol=sym, + slot_id=slot_id, + ) + except Exception: # noqa: BLE001 + _log.exception( + "sync: resolved notify failed for slot=%s — warning clear kept", slot_id, + ) return { "slot_id": slot_id, "symbol": sym, "portfolio_id": pid, @@ -583,7 +897,9 @@ def verify_slot_reconcile( broker: BrokerPort, slot_repo, instance_key: str, - warn_store: "ReconcileWarningStore", + warn_store: ReconcileWarningStore, + alarm_emitter: AlarmEmitter | None = None, + notification_store=None, ) -> dict: """Re-check C-case reconcile-warning for a portfolio (REQ-106, BAT-051). @@ -693,6 +1009,21 @@ def verify_slot_reconcile( "broker_qty": broker_qty, "reason": "warn_store error during C-case clear — warnings NOT fully cleared", } + # req-0007 / ac-0024: publish C-case resolved alarm (best-effort, warning clear kept). + if alarm_emitter is not None and notification_store is not None: + try: + _notify_resolved( + alarm_emitter, + notification_store, + identity_key=_c_identity_key(instance_key, sym), + case_code="C", + instance_key=instance_key, + symbol=sym, + ) + except Exception: # noqa: BLE001 + _log.exception( + "verify: resolved notify failed for symbol=%s — warning clear kept", sym, + ) return { "portfolio_id": portfolio_id, "slot_id": slot_id_str,