Merge pull request #73 from fourblocks2020/copilot/feat-alerts-publish-reconciliation-quantity-mismat

feat(alerts): publish B/C reconciliation quantity mismatches to trex.alerts with durable deduplication

tests/adapters/persistence/test_migration_and_drift.py

@@ -27,6 +27,7 @@ _EXPECTED_TABLES = {
     "snapshot_pause",     # BAT-008 / REQ-032
     "broker_account_snapshot",  # dcy-041 / GitHub #45
     "ledger_skip",             # GitHub #68 skipped-ledger durable audit
+    "reconcile_notification_state",  # req-0007 / ac-0024
 }
 
 

tests/adapters/portfolio/test_reconcile_notification_store.py

@@ -0,0 +1,213 @@
+"""PostgreSQLReconcileNotificationStore integration tests (req-0007, ac-0024).
+
+Verifies migration 026 schema creation (idempotent replay), full CRUD
+round-trip, and lifecycle semantics against a real PostgreSQL 16 instance.
+
+Gated on TREX_POSTGRESQL_TEST_DSN — skipped in unit-test-only CI.
+"""
+from __future__ import annotations
+
+import pytest
+
+from trex_core.application.reconciliation import _c_identity_key
+
+
+@pytest.fixture
+def pg_notify_store(pg_clean):
+    from trex_core.adapters.portfolio.reconcile_notification_store import (
+        PostgreSQLReconcileNotificationStore,
+    )
+    return PostgreSQLReconcileNotificationStore(pg_clean)
+
+
+# ---------------------------------------------------------------------------
+# Migration 026 schema
+# ---------------------------------------------------------------------------
+
+class TestMigration026:
+    def test_table_exists_after_migration(self, pg_clean):
+        """Migration 026 creates reconcile_notification_state (IF NOT EXISTS)."""
+        import psycopg
+        with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT table_name FROM information_schema.tables "
+                "WHERE table_schema = 'public' AND table_name = 'reconcile_notification_state'"
+            )
+            assert cur.fetchone() is not None, "reconcile_notification_state table not found"
+
+    def test_table_has_expected_columns(self, pg_clean):
+        """Migration 026 creates all required columns."""
+        import psycopg
+        with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = 'public' AND table_name = 'reconcile_notification_state'"
+            )
+            cols = {r[0] for r in cur.fetchall()}
+        expected = {
+            "identity_key", "case_code", "instance_key", "symbol",
+            "quantity_a", "quantity_b", "lifecycle_state", "created_at", "updated_at",
+        }
+        assert expected <= cols, f"Missing columns: {expected - cols}"
+
+    def test_migration_is_idempotent(self, pg_clean):
+        """Running migration 026 twice does not fail (IF NOT EXISTS guard)."""
+        from trex_core.adapters.persistence.migrate import apply_migrations
+        # pg_clean already ran apply_migrations once; run again to verify idempotency.
+        apply_migrations(pg_clean)  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# CRUD round-trip
+# ---------------------------------------------------------------------------
+
+class TestCRUDRoundTrip:
+    def test_set_and_get_state_b_case(self, pg_notify_store):
+        """set_state → get_state round-trip for B-case (slot_id identity)."""
+        pg_notify_store.set_state(
+            "slot-uuid-001",
+            case_code="B",
+            instance_key="inst-A",
+            symbol="005930",
+            quantity_a=10,
+            quantity_b=7,
+            lifecycle_state="notified",
+        )
+        state = pg_notify_store.get_state("slot-uuid-001")
+        assert state is not None
+        assert state["case_code"] == "B"
+        assert state["instance_key"] == "inst-A"
+        assert state["symbol"] == "005930"
+        assert state["quantity_a"] == 10
+        assert state["quantity_b"] == 7
+        assert state["lifecycle_state"] == "notified"
+
+    def test_set_and_get_state_c_case(self, pg_notify_store):
+        """set_state → get_state round-trip for C-case (SHA-256 digest identity)."""
+        c_key = _c_identity_key("inst-A", "005930")
+        pg_notify_store.set_state(
+            c_key,
+            case_code="C",
+            instance_key="inst-A",
+            symbol="005930",
+            quantity_a=50,
+            quantity_b=42,
+            lifecycle_state="notified",
+        )
+        state = pg_notify_store.get_state(c_key)
+        assert state is not None
+        assert state["case_code"] == "C"
+        assert state["symbol"] == "005930"
+        assert state["quantity_a"] == 50
+
+    def test_set_state_upsert(self, pg_notify_store):
+        """set_state is an upsert — second call with same key updates the row."""
+        pg_notify_store.set_state(
+            "slot-uuid-002",
+            case_code="B",
+            instance_key="inst-A",
+            symbol="005930",
+            quantity_a=10,
+            quantity_b=7,
+            lifecycle_state="notified",
+        )
+        pg_notify_store.set_state(
+            "slot-uuid-002",
+            case_code="B",
+            instance_key="inst-A",
+            symbol="005930",
+            quantity_a=10,
+            quantity_b=9,  # quantity changed
+            lifecycle_state="notified",
+        )
+        state = pg_notify_store.get_state("slot-uuid-002")
+        assert state["quantity_b"] == 9
+
+    def test_get_state_returns_none_for_missing_key(self, pg_notify_store):
+        assert pg_notify_store.get_state("does-not-exist") is None
+
+    def test_clear_state_removes_row(self, pg_notify_store):
+        """clear_state deletes the row; subsequent get_state returns None."""
+        pg_notify_store.set_state(
+            "slot-uuid-003",
+            case_code="B",
+            instance_key="inst-A",
+            symbol="005930",
+            quantity_a=10,
+            quantity_b=7,
+            lifecycle_state="notified",
+        )
+        pg_notify_store.clear_state("slot-uuid-003")
+        assert pg_notify_store.get_state("slot-uuid-003") is None
+
+    def test_list_states_by_lifecycle(self, pg_notify_store):
+        """list_states_by_lifecycle returns only rows with the given state."""
+        pg_notify_store.set_state(
+            "slot-A",
+            case_code="B", instance_key="inst-A", symbol="005930",
+            quantity_a=10, quantity_b=7, lifecycle_state="notified",
+        )
+        pg_notify_store.set_state(
+            "slot-B",
+            case_code="B", instance_key="inst-A", symbol="005930",
+            quantity_a=20, quantity_b=15, lifecycle_state="pending_resolved",
+        )
+        notified = pg_notify_store.list_states_by_lifecycle("notified")
+        pending = pg_notify_store.list_states_by_lifecycle("pending_resolved")
+        assert len(notified) == 1
+        assert notified[0]["identity_key"] == "slot-A"
+        assert len(pending) == 1
+        assert pending[0]["identity_key"] == "slot-B"
+
+    def test_durability_across_store_instances(self, pg_clean):
+        """State survives a simulated process restart (new store instance, same DSN)."""
+        from trex_core.adapters.portfolio.reconcile_notification_store import (
+            PostgreSQLReconcileNotificationStore,
+        )
+        a = PostgreSQLReconcileNotificationStore(pg_clean)
+        a.set_state(
+            "slot-restart",
+            case_code="B", instance_key="inst-A", symbol="005930",
+            quantity_a=10, quantity_b=7, lifecycle_state="notified",
+        )
+        # "restart" — fresh store instance
+        b = PostgreSQLReconcileNotificationStore(pg_clean)
+        state = b.get_state("slot-restart")
+        assert state is not None
+        assert state["lifecycle_state"] == "notified"
+
+    def test_c_identity_keys_for_distinct_inputs_are_distinct(self, pg_notify_store):
+        """Two distinct (instance_key, symbol) pairs must yield distinct DB rows."""
+        key_a = _c_identity_key("inst-A", "005930")
+        key_b = _c_identity_key("inst-A", "005931")
+        assert key_a != key_b
+        pg_notify_store.set_state(
+            key_a, case_code="C", instance_key="inst-A", symbol="005930",
+            quantity_a=10, quantity_b=8, lifecycle_state="notified",
+        )
+        pg_notify_store.set_state(
+            key_b, case_code="C", instance_key="inst-A", symbol="005931",
+            quantity_a=20, quantity_b=15, lifecycle_state="notified",
+        )
+        state_a = pg_notify_store.get_state(key_a)
+        state_b = pg_notify_store.get_state(key_b)
+        assert state_a["symbol"] == "005930"
+        assert state_b["symbol"] == "005931"
+
+    def test_crafted_separator_collision_absent(self, pg_notify_store):
+        """Crafted-separator inputs ('a\\x1fC','b') vs ('a','C\\x1fb') map to distinct rows."""
+        key_a = _c_identity_key("a\x1fC", "b")
+        key_b = _c_identity_key("a", "C\x1fb")
+        assert key_a != key_b
+        pg_notify_store.set_state(
+            key_a, case_code="C", instance_key="a\x1fC", symbol="b",
+            quantity_a=1, quantity_b=2, lifecycle_state="notified",
+        )
+        pg_notify_store.set_state(
+            key_b, case_code="C", instance_key="a", symbol="C\x1fb",
+            quantity_a=3, quantity_b=4, lifecycle_state="notified",
+        )
+        state_a = pg_notify_store.get_state(key_a)
+        state_b = pg_notify_store.get_state(key_b)
+        assert state_a["instance_key"] == "a\x1fC"
+        assert state_b["instance_key"] == "a"

tests/application/test_reconcile_notification.py

@@ -0,0 +1,932 @@
+"""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
+
+
+# ---------------------------------------------------------------------------
+# Regression tests for ac-0024 blocker fixes
+# ---------------------------------------------------------------------------
+
+def test_c_identity_key_is_injective_and_postgres_safe():
+    """_c_identity_key must be injective (no collisions) and PostgreSQL TEXT-safe.
+
+    Specifically verifies the crafted-separator collision that bit the \\x1f
+    implementation: ('a\\x1fC', 'b') and ('a', 'C\\x1fb') previously produced
+    the same key and suppressed the second detected event.
+    """
+    key = _c_identity_key("my-instance", "005930")
+    assert "\x00" not in key, f"identity key contains NUL: {key!r}"
+    # Different symbols → different keys
+    assert key != _c_identity_key("my-instance", "005931")
+    # Different instances → different keys
+    assert key != _c_identity_key("other-instance", "005930")
+    # Crafted-separator collision must not exist
+    crafted_a = _c_identity_key("a\x1fC", "b")
+    crafted_b = _c_identity_key("a", "C\x1fb")
+    assert crafted_a != crafted_b, (
+        f"Separator-crafted collision: both produce {crafted_a!r}"
+    )
+
+
+def test_retry_pending_resolved_ignores_other_instance_states():
+    """_retry_pending_resolved must not touch pending states belonging to a different instance."""
+    from trex_core.application.reconciliation import _retry_pending_resolved
+
+    ns = InMemoryReconcileNotificationStore()
+    other_slot_id = "slot-other-instance"
+    # pending_resolved belonging to *another* instance
+    ns.set_state(other_slot_id, case_code="B", instance_key="other-instance",
+                 symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="pending_resolved")
+
+    emitter, transport = _make_emitter()
+    _retry_pending_resolved(
+        emitter, ns,
+        case_code="B",
+        instance_key=INSTANCE,          # current instance — should NOT see other-instance states
+        active_identity_keys=set(),
+    )
+
+    # No resolved alarm published for other instance's state
+    codes = _alarm_codes(transport)
+    assert "reconcile_mismatch_resolved" not in codes
+    # State preserved (not consumed by wrong instance)
+    assert ns.get_state(other_slot_id) is not None
+
+
+def test_b_sync_does_not_publish_resolved_when_clear_warned_fails():
+    """Resolved alarm must NOT be published when clear_warned() raises (warning still active)."""
+    from trex_core.application.reconciliation import sync_slot_to_broker
+
+    class FailingWarnStore(InMemoryReconcileWarningStore):
+        def clear_warned(self, slot_id):
+            raise RuntimeError("DB unavailable")
+
+    ns = InMemoryReconcileNotificationStore()
+    slot = _make_slot(qty=10.0)
+    slot_id_str = str(slot.slot_id)
+    repo = _repo_with_slot(slot)
+    ss = SlotService(repo)
+    ws = FailingWarnStore()
+    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")
+
+    emitter, transport = _make_emitter()
+    result = sync_slot_to_broker(
+        slot_id=slot_id_str,
+        slot_repo=repo,
+        slot_service=ss,
+        broker=FakeBroker({"005930": 7}),   # qty matches → clear attempt made
+        instance_key=INSTANCE,
+        warn_store=ws,
+        alarm_emitter=emitter,
+        notification_store=ns,
+    )
+
+    # warning_cleared is False because clear_warned raised
+    assert result["warning_cleared"] is False
+    # No resolved alarm should have been published
+    codes = _alarm_codes(transport)
+    assert "reconcile_mismatch_resolved" not in codes
+    # notification state untouched
+    state = ns.get_state(slot_id_str)
+    assert state is not None
+    assert state["lifecycle_state"] == "notified"

tests/conftest.py

@@ -26,6 +26,7 @@ _PG_TABLES = (
     "slot_transition_log",
     "daily_symbol_entries",
     "daily_slot",
+    "reconcile_notification_state",
 )
 
 

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; "C:<sha256_hex>" length-prefix-digest 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()
+);

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 TEXT for B-case; "C:<sha256_hex>" length-prefix-digest 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
+            ]

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

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:

trex_core/application/reconciliation.py

@@ -19,13 +19,17 @@ 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 hashlib
 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 +51,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 +66,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 +93,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 +101,255 @@ 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:
+    """Return an injective, PostgreSQL TEXT-safe key for a (instance_key, symbol) C-case pair.
+
+    Uses a length-prefixed SHA-256 hex digest so that *any* inputs — including
+    those containing separator characters — produce distinct keys with no
+    collision risk.  The prefix ``C:`` makes the key case-code-scoped and
+    human-scannable in the DB; the digest is 64 hex chars (256-bit).
+
+    Injectivity: the raw string fed to SHA-256 is ``"<N>:<instance_key>:<symbol>"``
+    where ``N = len(instance_key)``.  The leading ``N:`` uniquely delimits
+    ``instance_key`` from the rest, so two distinct ``(instance_key, symbol)``
+    pairs always produce different raw strings (and thus different digests).
+
+    Example: _c_identity_key("a\\x1fC", "b") != _c_identity_key("a", "C\\x1fb")
+    """
+    raw = f"{len(instance_key)}:{instance_key}:{symbol}"
+    digest = hashlib.sha256(raw.encode()).hexdigest()
+    return f"C:{digest}"
+
+
+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,
+    instance_key: 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.
+
+    Only processes states belonging to ``instance_key`` so that a shared
+    PostgreSQL store used by multiple scheduler instances does not let instance A
+    re-publish resolved events for instance B's pending states.
+    """
+    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
+        if state.get("instance_key") != instance_key:
+            continue  # belongs to a different scheduler instance — do not touch
+        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 +367,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 +376,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 +404,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 +425,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 +462,18 @@ 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",
+                instance_key=instance_key,
+                active_identity_keys=active_b_identity_keys,
+            )
+        except Exception:  # noqa: BLE001
+            _log.exception("reconcile_core_holdings: pending_resolved retry failed")
     return decisions
 
 
@@ -205,7 +483,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 +496,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 +514,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 +562,42 @@ 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",
+                instance_key=instance_key,
+                active_identity_keys=active_c_identity_keys,
+            )
+        except Exception:  # noqa: BLE001
+            _log.exception("reconcile_positions: pending_resolved retry failed")
+
     return decisions
 
 
@@ -291,9 +612,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 +696,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 +882,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 only after warning is cleared.
+        if warning_cleared and 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 +922,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 +1034,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,

trex_core/application/snapshot_scheduler.py

@@ -22,6 +22,7 @@ from trex_core.application.portfolio import (
     DegradedReason,
     PortfolioHubReadPort,
     PortfolioHubUnavailableError,
+    ReconcileNotificationStore,
     ReconcileWarningStore,
     SnapshotStore,
     TradingContext,
@@ -49,6 +50,7 @@ class SnapshotScheduler:
         instance_key: str = "default",
         alarm_emitter: Optional[AlarmEmitter] = None,
         warn_store: Optional[ReconcileWarningStore] = None,
+        notification_store: Optional[ReconcileNotificationStore] = None,
         credential_fetch_enabled: bool = False,
         kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None,
         credential_refresh: Optional[Callable[[], object]] = None,
@@ -71,6 +73,8 @@ class SnapshotScheduler:
         self._alarm_emitter = alarm_emitter
         # BAT-050 / REQ-103: durable reconcile-warning store (Optional).
         self._warn_store = warn_store
+        # req-0007 / ac-0024: durable notification deduplication store (Optional).
+        self._notification_store = notification_store
         # dcy-036 / REQ-113/114/115: credential fetch opt-in + break-glass detection.
         # credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch.
         self._credential_fetch_enabled: bool = credential_fetch_enabled
@@ -312,6 +316,8 @@ class SnapshotScheduler:
                 slot_repo=self._slot_repo,
                 instance_key=self._instance_key,
                 warn_store=self._warn_store,
+                alarm_emitter=self._alarm_emitter,
+                notification_store=self._notification_store,
             )
         except Exception:  # noqa: BLE001
             _log.exception("reconciliation pass failed")
@@ -329,6 +335,7 @@ class SnapshotScheduler:
                         broker=self._broker, instance_key=self._instance_key,
                         alarm_emitter=self._alarm_emitter,  # dcy-033 / REQ-100
                         warn_store=self._warn_store,  # BAT-050 / REQ-103
+                        notification_store=self._notification_store,  # req-0007 / ac-0024
                     )
                 except Exception:  # noqa: BLE001
                     _log.exception("core-holding divergence reconcile failed")

trex_core/runtime/cli.py

@@ -707,6 +707,9 @@ def _cmd_sync(args, cfg, confirm) -> int:
     from trex_core.adapters.portfolio.reconcile_warning_store import (
         PostgreSQLReconcileWarningStore,
     )
+    from trex_core.adapters.portfolio.reconcile_notification_store import (
+        PostgreSQLReconcileNotificationStore,
+    )
     from trex_core.application.reconciliation import sync_slot_to_broker
 
     if not args.portfolio_id and not args.all:
@@ -745,6 +748,9 @@ def _cmd_sync(args, cfg, confirm) -> int:
         return _broker_read_failure(exc, args.json)
 
     warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)
+    notification_store = PostgreSQLReconcileNotificationStore(cfg.dsn)
+    # req-0007 / ac-0024: alarm emitter for resolved notifications.
+    alarm_emitter_sync, _alarm_transport_sync = _default_alarm_emitter(cfg)
 
     # Build slot_repo + slot_service
     from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository
@@ -789,22 +795,27 @@ def _cmd_sync(args, cfg, confirm) -> int:
 
     results = []
     exit_code = 0
-    for sid in slot_ids:
-        res = sync_slot_to_broker(
-            slot_id=sid,
-            slot_repo=slot_repo,
-            slot_service=slot_service,
-            broker=broker_port,
-            instance_key=instance,
-            warn_store=warn_store,
-            cost_basis_port=cost_basis_port,
-            dry_run=dry_run,
-        )
-        results.append(res)
-        # A skipped result can still be a fail-closed live error (REQ-123).
-        # Do not let the no-write marker hide a failed sync from automation.
-        if res.get("error"):
-            exit_code = 1
+    try:
+        for sid in slot_ids:
+            res = sync_slot_to_broker(
+                slot_id=sid,
+                slot_repo=slot_repo,
+                slot_service=slot_service,
+                broker=broker_port,
+                instance_key=instance,
+                warn_store=warn_store,
+                cost_basis_port=cost_basis_port,
+                dry_run=dry_run,
+                alarm_emitter=alarm_emitter_sync,
+                notification_store=notification_store,
+            )
+            results.append(res)
+            # A skipped result can still be a fail-closed live error (REQ-123).
+            # Do not let the no-write marker hide a failed sync from automation.
+            if res.get("error"):
+                exit_code = 1
+    finally:
+        _alarm_transport_sync.close()
 
     if args.json:
         print(_json.dumps({"instance": instance, "dry_run": dry_run, "results": results},
@@ -850,6 +861,9 @@ def _cmd_verify(args, cfg) -> int:
     from trex_core.adapters.portfolio.reconcile_warning_store import (
         PostgreSQLReconcileWarningStore,
     )
+    from trex_core.adapters.portfolio.reconcile_notification_store import (
+        PostgreSQLReconcileNotificationStore,
+    )
     from trex_core.application.portfolio import SnapshotStore
     from trex_core.application.reconciliation import verify_slot_reconcile
 
@@ -867,6 +881,9 @@ def _cmd_verify(args, cfg) -> int:
         return _broker_read_failure(exc, args.json)
 
     warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)
+    notification_store = PostgreSQLReconcileNotificationStore(cfg.dsn)
+    # req-0007 / ac-0024: alarm emitter for resolved notifications.
+    alarm_emitter_verify, _alarm_transport_verify = _default_alarm_emitter(cfg)
     slot_repo = PostgreSQLSlotRepository(cfg.dsn)
 
     # Build a minimal snapshot store (we only need asset_quantity for C-case check).
@@ -909,18 +926,23 @@ def _cmd_verify(args, cfg) -> int:
 
     results = []
     exit_code = 0
-    for pid in target_pids:
-        res = verify_slot_reconcile(
-            portfolio_id=pid,
-            store=store,
-            broker=broker_port,
-            slot_repo=slot_repo,
-            instance_key=instance,
-            warn_store=warn_store,
-        )
-        results.append(res)
-        if res["result"] == "rejected":
-            exit_code = 1
+    try:
+        for pid in target_pids:
+            res = verify_slot_reconcile(
+                portfolio_id=pid,
+                store=store,
+                broker=broker_port,
+                slot_repo=slot_repo,
+                instance_key=instance,
+                warn_store=warn_store,
+                alarm_emitter=alarm_emitter_verify,
+                notification_store=notification_store,
+            )
+            results.append(res)
+            if res["result"] == "rejected":
+                exit_code = 1
+    finally:
+        _alarm_transport_verify.close()
 
     if args.json:
         print(_json.dumps({"instance": instance, "results": results},

trex_core/runtime/daemon.py

@@ -49,6 +49,9 @@ from trex_core.adapters.portfolio import (
     StubPortfolioRead,
 )
 from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore
+from trex_core.adapters.portfolio.reconcile_notification_store import (
+    PostgreSQLReconcileNotificationStore,
+)
 from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.account_snapshot import AccountSnapshotService
 from trex_core.application.broker_port import BrokerPort
@@ -198,6 +201,8 @@ class Daemon:
         self.pause_store = PostgreSQLPauseStore(config.dsn)
         # BAT-050 / REQ-102: durable reconcile-warning store (slot_id PK).
         self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
+        # req-0007 / ac-0024: durable reconcile notification deduplication store.
+        self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)
         # BAT-012: day-scoped kill-switch exclusion store + service.
         self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn)
         self.portfolio_read = _build_portfolio_read(config)
@@ -273,6 +278,7 @@ class Daemon:
             instance_key=config.instance_key,
             alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
             warn_store=self.warn_store,  # BAT-050 / REQ-103: durable reconcile-warning
+            notification_store=self.notification_store,  # req-0007 / ac-0024
             credential_fetch_enabled=config.credential_fetch_enabled,  # dcy-036 / REQ-113
             credential_refresh=self.broker_provider.refresh_day_start_credential,
             # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.