"""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, *, case_code=None):
            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"