fix(ac-0024): injective C identity via SHA-256 length-prefix digest + PostgreSQL migration/round-trip tests

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

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

@@ -845,16 +845,25 @@ def test_publish_alarm_returns_bool():
 # 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."""
+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}"
-    # 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
+    # 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():

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

@@ -1,5 +1,5 @@
 -- 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).
+-- 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 (

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}\\x1fC\\x1f{symbol}" for C-case).
+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

trex_core/application/reconciliation.py

@@ -25,6 +25,7 @@ published via AlarmEmitter with durable deduplication through ReconcileNotificat
 """
 from __future__ import annotations
 
+import hashlib
 import logging
 import math
 import re
@@ -105,12 +106,23 @@ 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 (PostgreSQL TEXT-safe, collision-resistant).
+    """Return an injective, PostgreSQL TEXT-safe key for a (instance_key, symbol) C-case pair.
 
-    Uses ASCII Unit Separator (\\x1f) which PostgreSQL TEXT accepts (only \\x00 is
-    forbidden) and which cannot appear in well-formed instance_key / symbol values.
+    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")
     """
-    return f"{instance_key}\x1fC\x1f{symbol}"
+    raw = f"{len(instance_key)}:{instance_key}:{symbol}"
+    digest = hashlib.sha256(raw.encode()).hexdigest()
+    return f"C:{digest}"
 
 
 def _build_alarm_fields(