feat(alerts): publish reconciliation quantity mismatches to trex-notifier (req-0007/ac-0024)

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

trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql

@@ -0,0 +1,16 @@
+-- req-0007 / ac-0024: durable mismatch notification lifecycle state.
+-- PK: identity_key (slot_id TEXT for B-case; "{instance_key}\x00C\x00{symbol}" for C-case).
+-- lifecycle_state: 'notified' | 'pending_resolved'.
+-- ON CONFLICT DO UPDATE: replay-safe; re-running migration is a no-op.
+CREATE TABLE IF NOT EXISTS reconcile_notification_state (
+    identity_key    TEXT        PRIMARY KEY,
+    case_code       TEXT        NOT NULL CHECK (case_code IN ('B', 'C')),
+    instance_key    TEXT        NOT NULL,
+    symbol          TEXT        NOT NULL,
+    quantity_a      INTEGER     NOT NULL,
+    quantity_b      INTEGER     NOT NULL,
+    lifecycle_state TEXT        NOT NULL
+        CHECK (lifecycle_state IN ('notified', 'pending_resolved')),
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
+);

trex_core/adapters/portfolio/reconcile_notification_store.py

@@ -0,0 +1,122 @@
+"""PostgreSQL-backed durable reconcile notification lifecycle store (req-0007, ac-0024).
+
+Tracks per-identity B/C-case mismatch notification fingerprints so that
+duplicate alarms are suppressed across scheduler ticks and process restarts,
+and pending resolved transitions can be retried.
+
+PK: identity_key (slot_id for B-case; "{instance_key}\\x00C\\x00{symbol}" for C-case).
+lifecycle_state: 'notified' | 'pending_resolved'.
+"""
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from trex_core.adapters.persistence.postgresql import _require_psycopg
+
+
+class PostgreSQLReconcileNotificationStore:
+    """`reconcile_notification_state` table CRUD (req-0007, ac-0024)."""
+
+    def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
+        self._psycopg = _require_psycopg(psycopg_module)
+        self._dsn = dsn
+
+    def get_state(self, identity_key: str) -> dict | None:
+        """Return state dict or None."""
+        with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT identity_key, case_code, instance_key, symbol, "
+                "       quantity_a, quantity_b, lifecycle_state "
+                "FROM reconcile_notification_state "
+                "WHERE identity_key = %s LIMIT 1",
+                (identity_key,),
+            )
+            row = cur.fetchone()
+            if row is None:
+                return None
+            return {
+                "identity_key": row[0],
+                "case_code": row[1],
+                "instance_key": row[2],
+                "symbol": row[3],
+                "quantity_a": row[4],
+                "quantity_b": row[5],
+                "lifecycle_state": row[6],
+            }
+
+    def set_state(
+        self,
+        identity_key: str,
+        *,
+        case_code: str,
+        instance_key: str,
+        symbol: str,
+        quantity_a: int,
+        quantity_b: int,
+        lifecycle_state: str,
+    ) -> None:
+        """Upsert notification state."""
+        now = datetime.now(tz=UTC)
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "INSERT INTO reconcile_notification_state "
+                    "(identity_key, case_code, instance_key, symbol, "
+                    " quantity_a, quantity_b, lifecycle_state, created_at, updated_at) "
+                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) "
+                    "ON CONFLICT (identity_key) DO UPDATE "
+                    "SET case_code       = EXCLUDED.case_code, "
+                    "    instance_key    = EXCLUDED.instance_key, "
+                    "    symbol          = EXCLUDED.symbol, "
+                    "    quantity_a      = EXCLUDED.quantity_a, "
+                    "    quantity_b      = EXCLUDED.quantity_b, "
+                    "    lifecycle_state = EXCLUDED.lifecycle_state, "
+                    "    updated_at      = EXCLUDED.updated_at",
+                    (
+                        identity_key,
+                        case_code,
+                        instance_key,
+                        symbol,
+                        quantity_a,
+                        quantity_b,
+                        lifecycle_state,
+                        now,
+                        now,
+                    ),
+                )
+                conn.commit()
+
+    def clear_state(self, identity_key: str) -> None:
+        """Delete state row (called after successful resolved publish)."""
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "DELETE FROM reconcile_notification_state WHERE identity_key = %s",
+                    (identity_key,),
+                )
+                conn.commit()
+
+    def list_states_by_lifecycle(self, lifecycle_state: str) -> list[dict]:
+        """Return all state dicts with the given lifecycle_state."""
+        with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT identity_key, case_code, instance_key, symbol, "
+                "       quantity_a, quantity_b, lifecycle_state "
+                "FROM reconcile_notification_state "
+                "WHERE lifecycle_state = %s",
+                (lifecycle_state,),
+            )
+            rows = cur.fetchall()
+            return [
+                {
+                    "identity_key": row[0],
+                    "case_code": row[1],
+                    "instance_key": row[2],
+                    "symbol": row[3],
+                    "quantity_a": row[4],
+                    "quantity_b": row[5],
+                    "lifecycle_state": row[6],
+                }
+                for row in rows
+            ]

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,16 @@ durable store is additive and Optional (None => warn-log-only, backward compat).
 
 BAT-051 / REQ-105: sync_slot_to_broker — operator CLI sync path.
 BAT-051 / REQ-106: verify_slot_reconcile — operator CLI verify path (C-only, read-centric).
+
+req-0007 / ac-0024: B/C-case mismatch alarms (detected, changed, resolved) are
+published via AlarmEmitter with durable deduplication through ReconcileNotificationStore.
 """
 from __future__ import annotations
 
 import logging
 import math
 import re
-from typing import Optional, cast
+from typing import cast
 
 from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.broker_port import BrokerPort
@@ -47,7 +50,7 @@ _COST_BASIS_REASON_CODES = frozenset({
 _COST_BASIS_LEGACY_SOURCE_ALIASES = frozenset({"avg_cost", "wac", "cost_basis"})
 
 
-def _sanitize_cost_basis_reason(reason: object) -> Optional[str]:
+def _sanitize_cost_basis_reason(reason: object) -> str | None:
     """Keep public diagnostics in the bounded REQ-123 reason taxonomy."""
     if not isinstance(reason, str):
         return None
@@ -62,7 +65,7 @@ def _normalize_cost_basis_diagnostics(
     value: object,
     source: object,
     reason: object,
-) -> tuple[Optional[float], str, Optional[str]]:
+) -> tuple[float | None, str, str | None]:
     """Normalize diagnostic-port output before it reaches reconciliation results."""
     safe_reason = _sanitize_cost_basis_reason(reason)
     if source == "canonical":
@@ -89,7 +92,7 @@ def _normalize_cost_basis_diagnostics(
     return normalized_value, normalized_source, safe_reason
 
 
-def _normalize_legacy_cost_basis(value: object) -> tuple[Optional[float], str, Optional[str]]:
+def _normalize_legacy_cost_basis(value: object) -> tuple[float | None, str, str | None]:
     """Make the diagnostic-free compatibility seam obey safe WAC semantics."""
     normalized_value, _, _ = _normalize_cost_basis_diagnostics(value, "legacy", "invalid")
     if normalized_value is None:
@@ -97,14 +100,233 @@ def _normalize_legacy_cost_basis(value: object) -> tuple[Optional[float], str, O
     return normalized_value, "legacy", None
 
 
+# ---------------------------------------------------------------------------
+# req-0007 / ac-0024: Notification lifecycle helpers
+# ---------------------------------------------------------------------------
+
+def _c_identity_key(instance_key: str, symbol: str) -> str:
+    """Stable C-case notification identity key (null-byte separated to avoid collisions)."""
+    return f"{instance_key}\x00C\x00{symbol}"
+
+
+def _build_alarm_fields(
+    *,
+    case_code: str,
+    instance_key: str,
+    symbol: str,
+    quantity_a: int,
+    quantity_b: int,
+    slot_id: str | None = None,
+) -> dict:
+    """Build sanitized alarm context fields (ac-0024 allowlist).
+
+    Excluded: account number/hash, credential, token, DSN, raw payload, exception.
+    """
+    fields: dict = {
+        "instance_key": instance_key,
+        "symbol": symbol,
+        "case_code": case_code,
+        "broker_quantity": quantity_b,
+    }
+    if case_code == "B":
+        fields["slot_quantity"] = quantity_a
+        if slot_id is not None:
+            fields["slot_id"] = slot_id
+    else:
+        fields["hub_quantity"] = quantity_a
+    return fields
+
+
+def _notify_mismatch(
+    alarm_emitter: AlarmEmitter,
+    notification_store,
+    *,
+    identity_key: str,
+    case_code: str,
+    instance_key: str,
+    symbol: str,
+    quantity_a: int,
+    quantity_b: int,
+    slot_id: str | None = None,
+) -> None:
+    """Publish detected or changed alarm; advance fingerprint only on publish success.
+
+    Suppresses if the identity fingerprint is unchanged from the last notified state.
+    Retries on every detection if publish previously failed (fingerprint not advanced).
+    """
+    try:
+        state = notification_store.get_state(identity_key)
+    except Exception:  # noqa: BLE001
+        _log.exception(
+            "reconcile_notification: get_state failed for %s — skipping notify", identity_key,
+        )
+        return
+
+    if state is not None and state.get("lifecycle_state") == "notified":
+        if state.get("quantity_a") == quantity_a and state.get("quantity_b") == quantity_b:
+            return  # unchanged — suppress
+        event_type = "changed"
+    else:
+        # No prior state, OR pending_resolved (new mismatch after failed resolved) → re-detect
+        event_type = "detected"
+
+    fields = _build_alarm_fields(
+        case_code=case_code,
+        instance_key=instance_key,
+        symbol=symbol,
+        quantity_a=quantity_a,
+        quantity_b=quantity_b,
+        slot_id=slot_id,
+    )
+    published = alarm_emitter.publish_alarm(
+        f"reconcile_mismatch_{event_type}",
+        f"reconciliation mismatch {event_type}: {case_code}-case symbol={symbol}",
+        fields=fields,
+    )
+
+    if published:
+        try:
+            notification_store.set_state(
+                identity_key,
+                case_code=case_code,
+                instance_key=instance_key,
+                symbol=symbol,
+                quantity_a=quantity_a,
+                quantity_b=quantity_b,
+                lifecycle_state="notified",
+            )
+        except Exception:  # noqa: BLE001
+            # Fingerprint write failure: at-least-once duplicate on next detection.
+            _log.exception(
+                "reconcile_notification: fingerprint persist failed for %s — "
+                "at-least-once duplicate possible on next detection",
+                identity_key,
+            )
+
+
+def _notify_resolved(
+    alarm_emitter: AlarmEmitter,
+    notification_store,
+    *,
+    identity_key: str,
+    case_code: str,
+    instance_key: str,
+    symbol: str,
+    slot_id: str | None = None,
+) -> None:
+    """Publish resolved alarm; clear or mark pending_resolved on publish failure.
+
+    Warning clear is always preserved by the caller regardless of this function's outcome.
+    - Publish success: clear notification state.
+    - Publish failure + state writable: mark pending_resolved for retry.
+    - State persistence failure: log loss risk (business path unaffected).
+    """
+    try:
+        state = notification_store.get_state(identity_key)
+    except Exception:  # noqa: BLE001
+        _log.exception(
+            "reconcile_notification: get_state failed for %s — skipping resolved notify",
+            identity_key,
+        )
+        return
+
+    if state is None:
+        return  # idempotent: already resolved or never detected
+
+    quantity_a = state["quantity_a"]
+    quantity_b = state["quantity_b"]
+
+    fields = _build_alarm_fields(
+        case_code=case_code,
+        instance_key=instance_key,
+        symbol=symbol,
+        quantity_a=quantity_a,
+        quantity_b=quantity_b,
+        slot_id=slot_id,
+    )
+    published = alarm_emitter.publish_alarm(
+        "reconcile_mismatch_resolved",
+        f"reconciliation mismatch resolved: {case_code}-case symbol={symbol}",
+        fields=fields,
+    )
+
+    if published:
+        try:
+            notification_store.clear_state(identity_key)
+        except Exception:  # noqa: BLE001
+            # At-least-once: retry may produce duplicate resolved event.
+            _log.exception(
+                "reconcile_notification: clear_state failed for %s — "
+                "at-least-once duplicate resolved possible",
+                identity_key,
+            )
+    else:
+        # Publish failed — persist pending_resolved so the next opportunity retries.
+        try:
+            notification_store.set_state(
+                identity_key,
+                case_code=case_code,
+                instance_key=instance_key,
+                symbol=symbol,
+                quantity_a=quantity_a,
+                quantity_b=quantity_b,
+                lifecycle_state="pending_resolved",
+            )
+        except Exception:  # noqa: BLE001
+            # State persistence failed: warning clear is kept; resolved event may be lost.
+            _log.error(
+                "reconcile_notification: resolved publish failed AND state persist failed "
+                "for %s — resolved event may be lost",
+                identity_key,
+            )
+
+
+def _retry_pending_resolved(
+    alarm_emitter: AlarmEmitter,
+    notification_store,
+    *,
+    case_code: str,
+    active_identity_keys: set[str],
+) -> None:
+    """Retry pending_resolved publishes for identities no longer actively mismatched.
+
+    Called at the end of each reconcile pass to drain the pending_resolved queue
+    for identities that have since been cleared by operator sync/verify.
+    """
+    try:
+        pending = notification_store.list_states_by_lifecycle("pending_resolved")
+    except Exception:  # noqa: BLE001
+        _log.exception(
+            "reconcile_notification: list_states_by_lifecycle failed — skipping pending retry",
+        )
+        return
+
+    for state in pending:
+        if state.get("case_code") != case_code:
+            continue
+        identity_key = state["identity_key"]
+        if identity_key in active_identity_keys:
+            continue  # still mismatched — do not retry resolved yet
+        _notify_resolved(
+            alarm_emitter,
+            notification_store,
+            identity_key=identity_key,
+            case_code=state["case_code"],
+            instance_key=state["instance_key"],
+            symbol=state["symbol"],
+            slot_id=identity_key if case_code == "B" else None,
+        )
+
+
 def reconcile_core_holdings(
     *,
     slot_repo,
     slot_service,
     broker: BrokerPort,
     instance_key: str,
-    alarm_emitter: Optional[AlarmEmitter] = None,
-    warn_store: Optional[ReconcileWarningStore] = None,
+    alarm_emitter: AlarmEmitter | None = None,
+    warn_store: ReconcileWarningStore | None = None,
+    notification_store=None,
 ) -> dict[str, str]:
     """dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding
     vs broker divergence and set durable reconcile-warning for B case.
@@ -122,6 +344,7 @@ def reconcile_core_holdings(
                      → set by reconcile_positions
 
     warn_store: Optional — when None, WARN-log-only (backward compat, REQ-103 AC-5).
+    notification_store: Optional — when None, alarm deduplication is skipped (backward compat).
     Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'|
     'managed_non_holding_divergence'} for telemetry.
     """
@@ -130,6 +353,7 @@ def reconcile_core_holdings(
     }
     decisions: dict[str, str] = {}
     matched: set[str] = set()
+    active_b_identity_keys: set[str] = set()  # req-0007: track active B mismatches for retry
     for slot in slot_repo.list_holding_slots(instance_key):
         sym = slot.symbol
         matched.add(sym)
@@ -157,6 +381,7 @@ def reconcile_core_holdings(
         elif slot.qty is not None and int(slot.qty) != bq:  # B case (REQ-103) — qty mismatch
             slot_id_str = str(slot.slot_id)
             decisions[slot_id_str] = "qty_warning"
+            active_b_identity_keys.add(slot_id_str)
             # REQ-107: structured WARNING log for B case.
             _log.warning(
                 "reconcile_warning set: slot_id=%s portfolio_id=%s symbol=%s "
@@ -177,6 +402,24 @@ def reconcile_core_holdings(
                     _log.exception(
                         "failed to set reconcile_warning for slot %s", slot.slot_id
                     )
+            # req-0007 / ac-0024: publish B-case detected/changed alarm (deduped).
+            if alarm_emitter is not None and notification_store is not None:
+                try:
+                    _notify_mismatch(
+                        alarm_emitter,
+                        notification_store,
+                        identity_key=slot_id_str,
+                        case_code="B",
+                        instance_key=instance_key,
+                        symbol=sym,
+                        quantity_a=int(slot.qty),
+                        quantity_b=bq,
+                        slot_id=slot_id_str,
+                    )
+                except Exception:  # noqa: BLE001
+                    _log.exception(
+                        "reconcile_core_holdings: notify_mismatch failed for slot=%s", slot_id_str,
+                    )
         else:
             decisions[str(slot.slot_id)] = "ok"
     for sym, bq in broker_qty.items():  # unmatched broker position
@@ -196,6 +439,17 @@ def reconcile_core_holdings(
                     "unmanaged position: symbol=%s qty=%d instance_key=%s",
                     sym, bq, instance_key,
                 )
+    # req-0007 / ac-0024: retry pending_resolved for B-case identities no longer mismatched.
+    if alarm_emitter is not None and notification_store is not None:
+        try:
+            _retry_pending_resolved(
+                alarm_emitter,
+                notification_store,
+                case_code="B",
+                active_identity_keys=active_b_identity_keys,
+            )
+        except Exception:  # noqa: BLE001
+            _log.exception("reconcile_core_holdings: pending_resolved retry failed")
     return decisions
 
 
@@ -205,7 +459,9 @@ def reconcile_positions(
     broker: BrokerPort,
     slot_repo=None,
     instance_key: str = "default",
-    warn_store: Optional[ReconcileWarningStore] = None,
+    warn_store: ReconcileWarningStore | None = None,
+    alarm_emitter: AlarmEmitter | None = None,
+    notification_store=None,
 ) -> dict[str, str]:
     """Run one reconciliation pass (REQ-031 + REQ-103 BAT-050).
 
@@ -216,6 +472,10 @@ def reconcile_positions(
     (ledger/hub qty vs broker) also set durable reconcile-warning on the corresponding
     slot (portfolio_id -> slot lookup via slot_repo). When either is None, falls back
     to in-memory warning only (backward compat, REQ-103 AC-5).
+
+    req-0007 / ac-0024: when alarm_emitter and notification_store are provided,
+    C-case detected/changed alarms are published per-symbol (one aggregate event
+    per symbol regardless of portfolio count).
     """
     broker_qty: dict[str, int] = {
         item.symbol: int(item.hldg_qty) for item in broker.get_balance()
@@ -230,6 +490,7 @@ def reconcile_positions(
         by_symbol.setdefault(sym, []).append((a.item.portfolio_id, a.item.asset_quantity))
 
     decisions: dict[str, str] = {}
+    active_c_identity_keys: set[str] = set()  # req-0007: track active C mismatches for retry
     for symbol, entries in by_symbol.items():
         hub_total = int(sum(qty for _, qty in entries))
         broker_total = broker_qty.get(symbol, 0)
@@ -277,6 +538,41 @@ def reconcile_positions(
                     decisions[pid] = "cleared"
                 else:
                     decisions[pid] = "ok"
+
+        # req-0007 / ac-0024: publish one C-case detected/changed alarm per symbol
+        # (aggregate mismatch lifecycle — not per portfolio).
+        if alarm_emitter is not None and notification_store is not None:
+            c_key = _c_identity_key(instance_key, symbol)
+            if not match:
+                active_c_identity_keys.add(c_key)
+                try:
+                    _notify_mismatch(
+                        alarm_emitter,
+                        notification_store,
+                        identity_key=c_key,
+                        case_code="C",
+                        instance_key=instance_key,
+                        symbol=symbol,
+                        quantity_a=hub_total,
+                        quantity_b=broker_total,
+                    )
+                except Exception:  # noqa: BLE001
+                    _log.exception(
+                        "reconcile_positions: notify_mismatch failed for symbol=%s", symbol,
+                    )
+
+    # req-0007 / ac-0024: retry pending_resolved for C-case identities no longer mismatched.
+    if alarm_emitter is not None and notification_store is not None:
+        try:
+            _retry_pending_resolved(
+                alarm_emitter,
+                notification_store,
+                case_code="C",
+                active_identity_keys=active_c_identity_keys,
+            )
+        except Exception:  # noqa: BLE001
+            _log.exception("reconcile_positions: pending_resolved retry failed")
+
     return decisions
 
 
@@ -291,9 +587,11 @@ def sync_slot_to_broker(
     slot_service,
     broker: BrokerPort,
     instance_key: str,
-    warn_store: "ReconcileWarningStore",
+    warn_store: ReconcileWarningStore,
     cost_basis_port=None,  # Optional[PortfolioHubCostBasisPort]
     dry_run: bool = False,
+    alarm_emitter: AlarmEmitter | None = None,
+    notification_store=None,
 ) -> dict:
     """Sync one slot's qty to the broker ground truth (REQ-105, BAT-051).
 
@@ -373,9 +671,9 @@ def sync_slot_to_broker(
     avg_price = slot.avg_price
     seeded_avg_price = False
     cost_basis_source = "none"
-    cost_basis_reason: Optional[str] = None
+    cost_basis_reason: str | None = None
 
-    def _set_cost_basis_unavailable_message() -> tuple[str, Optional[str]]:
+    def _set_cost_basis_unavailable_message() -> tuple[str, str | None]:
         if pid:
             return "cost_basis_unavailable", f"inactive slot {slot_id} portfolio_id={pid}"
         return "cost_basis_unavailable", "inactive slot without portfolio_id"
@@ -559,6 +857,22 @@ def sync_slot_to_broker(
             )
         except Exception:  # noqa: BLE001
             _log.exception("sync: failed to clear_warned for slot=%s", slot_id)
+        # req-0007 / ac-0024: publish B-case resolved alarm (best-effort, warning clear kept).
+        if alarm_emitter is not None and notification_store is not None:
+            try:
+                _notify_resolved(
+                    alarm_emitter,
+                    notification_store,
+                    identity_key=slot_id,
+                    case_code="B",
+                    instance_key=instance_key,
+                    symbol=sym,
+                    slot_id=slot_id,
+                )
+            except Exception:  # noqa: BLE001
+                _log.exception(
+                    "sync: resolved notify failed for slot=%s — warning clear kept", slot_id,
+                )
 
     return {
         "slot_id": slot_id, "symbol": sym, "portfolio_id": pid,
@@ -583,7 +897,9 @@ def verify_slot_reconcile(
     broker: BrokerPort,
     slot_repo,
     instance_key: str,
-    warn_store: "ReconcileWarningStore",
+    warn_store: ReconcileWarningStore,
+    alarm_emitter: AlarmEmitter | None = None,
+    notification_store=None,
 ) -> dict:
     """Re-check C-case reconcile-warning for a portfolio (REQ-106, BAT-051).
 
@@ -693,6 +1009,21 @@ def verify_slot_reconcile(
                 "broker_qty": broker_qty,
                 "reason": "warn_store error during C-case clear — warnings NOT fully cleared",
             }
+        # req-0007 / ac-0024: publish C-case resolved alarm (best-effort, warning clear kept).
+        if alarm_emitter is not None and notification_store is not None:
+            try:
+                _notify_resolved(
+                    alarm_emitter,
+                    notification_store,
+                    identity_key=_c_identity_key(instance_key, sym),
+                    case_code="C",
+                    instance_key=instance_key,
+                    symbol=sym,
+                )
+            except Exception:  # noqa: BLE001
+                _log.exception(
+                    "verify: resolved notify failed for symbol=%s — warning clear kept", sym,
+                )
         return {
             "portfolio_id": portfolio_id,
             "slot_id": slot_id_str,