fix(ac-0024): address 4 blockers: NUL byte in C identity, instance_key isolation, clear_warned gate, production wiring

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

tests/application/test_reconcile_notification.py

@@ -839,3 +839,85 @@ def test_publish_alarm_returns_bool():
     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_contains_no_nul_byte():
+    """_c_identity_key must not contain NUL (\\x00) — PostgreSQL TEXT bind would fail."""
+    key = _c_identity_key("my-instance", "005930")
+    assert "\x00" not in key, f"identity key contains NUL: {key!r}"
+    # Must still be collision-resistant (different symbols must produce different keys)
+    key2 = _c_identity_key("my-instance", "005931")
+    assert key != key2
+    # Must still be collision-resistant (different instances must produce different keys)
+    key3 = _c_identity_key("other-instance", "005930")
+    assert key != key3
+
+
+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"

trex_core/adapters/portfolio/reconcile_notification_store.py

@@ -4,7 +4,7 @@ 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).
+PK: identity_key (slot_id for B-case; "{instance_key}\\x1fC\\x1f{symbol}" for C-case).
 lifecycle_state: 'notified' | 'pending_resolved'.
 """
 from __future__ import annotations

trex_core/application/reconciliation.py

@@ -105,8 +105,12 @@ def _normalize_legacy_cost_basis(value: object) -> tuple[float | None, str, str
 # ---------------------------------------------------------------------------
 
 def _c_identity_key(instance_key: str, symbol: str) -> str:
-    """Stable C-case notification identity key (null-byte separated to avoid collisions)."""
+    """Stable C-case notification identity key (PostgreSQL TEXT-safe, collision-resistant).
-    return f"{instance_key}\x00C\x00{symbol}"
+
+    Uses ASCII Unit Separator (\\x1f) which PostgreSQL TEXT accepts (only \\x00 is
+    forbidden) and which cannot appear in well-formed instance_key / symbol values.
+    """
+    return f"{instance_key}\x1fC\x1f{symbol}"
 
 
 def _build_alarm_fields(
@@ -286,12 +290,17 @@ def _retry_pending_resolved(
     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")
@@ -304,6 +313,8 @@ def _retry_pending_resolved(
     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
@@ -446,6 +457,7 @@ def reconcile_core_holdings(
                 alarm_emitter,
                 notification_store,
                 case_code="B",
+                instance_key=instance_key,
                 active_identity_keys=active_b_identity_keys,
             )
         except Exception:  # noqa: BLE001
@@ -568,6 +580,7 @@ def reconcile_positions(
                 alarm_emitter,
                 notification_store,
                 case_code="C",
+                instance_key=instance_key,
                 active_identity_keys=active_c_identity_keys,
             )
         except Exception:  # noqa: BLE001
@@ -857,8 +870,8 @@ 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).
+        # req-0007 / ac-0024: publish B-case resolved alarm only after warning is cleared.
-        if alarm_emitter is not None and notification_store is not None:
+        if warning_cleared and alarm_emitter is not None and notification_store is not None:
             try:
                 _notify_resolved(
                     alarm_emitter,

trex_core/application/snapshot_scheduler.py

@@ -49,6 +49,7 @@ class SnapshotScheduler:
         instance_key: str = "default",
         alarm_emitter: Optional[AlarmEmitter] = None,
         warn_store: Optional[ReconcileWarningStore] = None,
+        notification_store=None,
         credential_fetch_enabled: bool = False,
         kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None,
         credential_refresh: Optional[Callable[[], object]] = None,
@@ -71,6 +72,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 +315,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 +334,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,6 +795,7 @@ def _cmd_sync(args, cfg, confirm) -> int:
 
     results = []
     exit_code = 0
+    try:
         for sid in slot_ids:
             res = sync_slot_to_broker(
                 slot_id=sid,
@@ -799,12 +806,16 @@ def _cmd_sync(args, cfg, confirm) -> int:
                 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,6 +926,7 @@ def _cmd_verify(args, cfg) -> int:
 
     results = []
     exit_code = 0
+    try:
         for pid in target_pids:
             res = verify_slot_reconcile(
                 portfolio_id=pid,
@@ -917,10 +935,14 @@ def _cmd_verify(args, cfg) -> int:
                 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.