diff --git a/tests/adapters/portfolio/test_reconcile_warning_store.py b/tests/adapters/portfolio/test_reconcile_warning_store.py new file mode 100644 index 0000000..05cc113 --- /dev/null +++ b/tests/adapters/portfolio/test_reconcile_warning_store.py @@ -0,0 +1,42 @@ +"""PostgreSQLReconcileWarningStore integration tests.""" +from __future__ import annotations + +from trex_core.adapters.portfolio.reconcile_warning_store import ( + PostgreSQLReconcileWarningStore, +) + + +def test_warning_store_preserves_multiple_cases_for_same_slot(pg_clean): + store = PostgreSQLReconcileWarningStore(pg_clean) + store.set_warned("slot-1", portfolio_id="pf-A", case_code="B", reason="b") + store.set_warned("slot-1", portfolio_id="pf-A", case_code="C", reason="c") + + assert store.is_warned("slot-1") is True + info = store.get_warning_info("slot-1") + assert info is not None + assert info["case_codes"] == ["B", "C"] + + +def test_warning_store_case_specific_clear_and_full_clear(pg_clean): + store = PostgreSQLReconcileWarningStore(pg_clean) + store.set_warned("slot-2", portfolio_id="pf-A", case_code="B") + store.set_warned("slot-2", portfolio_id="pf-A", case_code="C") + + store.clear_warned("slot-2", case_code="C") + info = store.get_warning_info("slot-2") + assert info is not None + assert info["case_codes"] == ["B"] + + store.clear_warned("slot-2") + assert store.is_warned("slot-2") is False + + +def test_warning_store_is_durable_across_reinstantiation(pg_clean): + a = PostgreSQLReconcileWarningStore(pg_clean) + a.set_warned("slot-restart", portfolio_id="pf-A", case_code="B") + a.set_warned("slot-restart", portfolio_id="pf-A", case_code="C") + + b = PostgreSQLReconcileWarningStore(pg_clean) + info = b.get_warning_info("slot-restart") + assert info is not None + assert info["case_codes"] == ["B", "C"] diff --git a/tests/application/test_reconcile_notification.py b/tests/application/test_reconcile_notification.py index 1c1de0d..b95044d 100644 --- a/tests/application/test_reconcile_notification.py +++ b/tests/application/test_reconcile_notification.py @@ -896,7 +896,7 @@ def test_b_sync_does_not_publish_resolved_when_clear_warned_fails(): from trex_core.application.reconciliation import sync_slot_to_broker class FailingWarnStore(InMemoryReconcileWarningStore): - def clear_warned(self, slot_id): + def clear_warned(self, slot_id, *, case_code=None): raise RuntimeError("DB unavailable") ns = InMemoryReconcileNotificationStore() diff --git a/tests/application/test_reconcile_warning_blocking.py b/tests/application/test_reconcile_warning_blocking.py index 73487b7..8acff52 100644 --- a/tests/application/test_reconcile_warning_blocking.py +++ b/tests/application/test_reconcile_warning_blocking.py @@ -892,6 +892,8 @@ def test_slot_to_dict_reconcile_warned_false_when_no_warn_store(): row = slot_to_dict(slot, warn_store=None) assert "reconcile_warned" in row assert row["reconcile_warned"] is False + assert row["trade_eligible"] is True + assert row["reconcile_warning_cases"] == [] def test_slot_to_dict_reconcile_warned_true_when_warned(): @@ -902,6 +904,8 @@ def test_slot_to_dict_reconcile_warned_true_when_warned(): warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B") row = slot_to_dict(slot, warn_store=warn_store) assert row["reconcile_warned"] is True + assert row["trade_eligible"] is False + assert row["reconcile_warning_cases"] == ["B"] def test_slot_to_dict_paused_and_warned_independent(): diff --git a/tests/application/test_reconcile_warning_store.py b/tests/application/test_reconcile_warning_store.py index 97f329e..7224aa9 100644 --- a/tests/application/test_reconcile_warning_store.py +++ b/tests/application/test_reconcile_warning_store.py @@ -33,14 +33,27 @@ def test_clear_warned_returns_false(): assert store.is_warned("slot-1") is False -# AC-3 (REQ-102): duplicate set_warned on same slot_id => upsert (single entry). -def test_set_warned_upsert_same_slot(): +# AC-3 (REQ-123): same slot can carry multiple active warning causes. +def test_set_warned_preserves_multiple_cases_on_same_slot(): store = _store() store.set_warned("slot-1", portfolio_id="pf-A", case_code="B", reason="first") store.set_warned("slot-1", portfolio_id="pf-A", case_code="C", reason="second") - # After second set, is_warned still True — single entry (no duplicate row). assert store.is_warned("slot-1") is True assert store.list_warned() == {"slot-1"} + info = store.get_warning_info("slot-1") + assert info is not None + assert info["case_codes"] == ["B", "C"] + + +def test_clear_warned_single_case_preserves_other_case(): + store = _store() + store.set_warned("slot-1", portfolio_id="pf-A", case_code="B") + store.set_warned("slot-1", portfolio_id="pf-A", case_code="C") + store.clear_warned("slot-1", case_code="C") + info = store.get_warning_info("slot-1") + assert info is not None + assert info["case_codes"] == ["B"] + assert store.is_warned("slot-1") is True # AC-5 (REQ-102): list_warned returns set of warned slot_ids. diff --git a/tests/application/test_slots_handler_quantity_mismatch.py b/tests/application/test_slots_handler_quantity_mismatch.py index 124a495..d181765 100644 --- a/tests/application/test_slots_handler_quantity_mismatch.py +++ b/tests/application/test_slots_handler_quantity_mismatch.py @@ -128,6 +128,44 @@ def test_ac1_single_portfolio_c_case_excluded(): assert resp.payload["slots"] == [] assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1 + assert resp.payload["excluded_slots"]["reconcile_warning_cases"] == {"C": 1} + + +def test_holding_slot_with_b_case_is_excluded_for_strategy(): + repo = InMemorySlotRepository() + svc = SlotService(repo) + warn_store = InMemoryReconcileWarningStore() + + slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10) + warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B") + + handler = _make_handler(repo, warn_store) + resp = handler(_req(filter="holding")) + + assert resp.payload["slots"] == [] + assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1 + assert resp.payload["excluded_slots"]["reconcile_warning_cases"] == {"B": 1} + + +def test_holding_slot_with_b_and_c_cases_keeps_both_and_reports_both(): + repo = InMemorySlotRepository() + svc = SlotService(repo) + warn_store = InMemoryReconcileWarningStore() + + slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10) + slot_id = str(slot.slot_id) + warn_store.set_warned(slot_id, portfolio_id="pf-A", case_code="B") + warn_store.set_warned(slot_id, portfolio_id="pf-A", case_code="C") + + handler = _make_handler(repo, warn_store) + resp = handler(_req(filter="holding")) + + assert resp.payload["slots"] == [] + assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1 + assert resp.payload["excluded_slots"]["reconcile_warning_cases"] == {"B": 1, "C": 1} + info = warn_store.get_warning_info(slot_id) + assert info is not None + assert info["case_codes"] == ["B", "C"] # --------------------------------------------------------------------------- @@ -286,7 +324,8 @@ def test_ac5_excluded_slots_count_zero_when_no_warnings(): resp = handler(_req(filter="holding")) assert "excluded_slots" in resp.payload - assert resp.payload["excluded_slots"] == {"reconcile_quantity_mismatch": 0} + assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 0 + assert resp.payload["excluded_slots"]["reconcile_warning_cases"] == {} # --------------------------------------------------------------------------- @@ -312,8 +351,8 @@ def test_ac6_no_warn_store_no_filtering_no_excluded_slots_key(): # --------------------------------------------------------------------------- class BrokenWarnStore(InMemoryReconcileWarningStore): - """Simulates a warn_store that fails on list_c_case_warned.""" - def list_c_case_warned(self): + """Simulates a warn_store that fails on warning reads.""" + def list_warned(self): raise RuntimeError("db connection lost") @@ -356,7 +395,7 @@ def test_ac23_exclusion_precedes_broker_and_strategy_state_readers(): ) response = handler(_req(filter="holding")).payload assert response["slots"] == [] - assert response["excluded_slots"] == {"reconcile_quantity_mismatch": 1} + assert response["excluded_slots"]["reconcile_quantity_mismatch"] == 1 # --------------------------------------------------------------------------- diff --git a/tests/application/test_sync.py b/tests/application/test_sync.py index c09d08f..4079789 100644 --- a/tests/application/test_sync.py +++ b/tests/application/test_sync.py @@ -199,6 +199,22 @@ def test_sync_holding_qty_mismatch_updated(): assert not ws_out.is_warned(str(slot.slot_id)) +def test_sync_clears_only_b_case_when_slot_has_multiple_warning_causes(): + slot = _make_slot(SlotState.HOLDING, qty=5.0) + ws = InMemoryReconcileWarningStore() + slot_id = str(slot.slot_id) + ws.set_warned(slot_id, portfolio_id="pf-005930", case_code="B") + ws.set_warned(slot_id, portfolio_id="pf-005930", case_code="C") + + result, _repo, ws_out = _setup(slot, {"005930": 10}, warn_store=ws) + + assert result["warning_cleared"] is True + assert ws_out.is_warned(slot_id) is True + info = ws_out.get_warning_info(slot_id) + assert info is not None + assert info["case_codes"] == ["C"] + + # --------------------------------------------------------------------------- # AC-4: WAC seed from cost_basis_port when available # --------------------------------------------------------------------------- diff --git a/tests/application/test_verify.py b/tests/application/test_verify.py index 5adc88d..61f94cf 100644 --- a/tests/application/test_verify.py +++ b/tests/application/test_verify.py @@ -313,6 +313,40 @@ def test_verify_c_case_clear_preserves_b_case_warning(): assert b_info is not None and b_info["case_code"] == "B" +def test_verify_c_case_clear_on_slot_with_b_and_c_keeps_b(): + sym = "005930" + slot = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-c") + + repo = InMemorySlotRepository() + repo.add_slot(slot) + + ws = InMemoryReconcileWarningStore() + slot_id = str(slot.slot_id) + ws.set_warned(slot_id, portfolio_id="pf-c", case_code="B") + ws.set_warned(slot_id, portfolio_id="pf-c", case_code="C") + + store = SnapshotStore() + store.swap([ + PortfolioItem(portfolio_id="pf-c", symbol=sym, cash_balance=0.0, asset_quantity=10.0), + ]) + broker = FakeBroker({sym: 10}) + + result = verify_slot_reconcile( + portfolio_id="pf-c", + store=store, + broker=broker, + slot_repo=repo, + instance_key=INSTANCE, + warn_store=ws, + ) + + assert result["result"] == "cleared" + assert ws.is_warned(slot_id) is True + info = ws.get_warning_info(slot_id) + assert info is not None + assert info["case_codes"] == ["B"] + + def test_verify_c_case_clear_non_holding_slot(): """A non-HOLDING slot with C-case warning for the same symbol must also be cleared.""" sym = "005930" @@ -360,11 +394,11 @@ class _FailAfterNStore(InMemoryReconcileWarningStore): self._calls = 0 self._fail_after = fail_after - def clear_warned(self, slot_id: str) -> None: + def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None: self._calls += 1 if self._calls > self._fail_after: raise RuntimeError("simulated DB error") - super().clear_warned(slot_id) + super().clear_warned(slot_id, case_code=case_code) def test_verify_partial_clear_failure_returns_clear_failed(): diff --git a/tests/runtime/test_cli_operator.py b/tests/runtime/test_cli_operator.py index f110464..91f6d1e 100644 --- a/tests/runtime/test_cli_operator.py +++ b/tests/runtime/test_cli_operator.py @@ -418,7 +418,7 @@ def test_sync_fail_closed_skipped_error_exits_nonzero_without_writes( return SimpleNamespace(get_balance=lambda: []), None class _WarningStore: - def clear_warned(self, *_args): + def clear_warned(self, *_args, **_kwargs): raise AssertionError("warning clear must not be called") class _Connection: diff --git a/trex_core/adapters/persistence/sql/016_create_slot_reconcile_warning.sql b/trex_core/adapters/persistence/sql/016_create_slot_reconcile_warning.sql index e663962..98664c3 100644 --- a/trex_core/adapters/persistence/sql/016_create_slot_reconcile_warning.sql +++ b/trex_core/adapters/persistence/sql/016_create_slot_reconcile_warning.sql @@ -1,6 +1,6 @@ -- BAT-050 / REQ-102: reconcile-warning durable store (dcy-034). --- PK: slot_id (TEXT, UUID string). pause_store 동형 설계. --- ON CONFLICT DO UPDATE: 동일 slot_id 재삽입 시 최신 값으로 갱신. +-- Initial schema; migration 028 upgrades PK to (slot_id, case_code) +-- so B/C causes can coexist for the same slot. CREATE TABLE IF NOT EXISTS slot_reconcile_warning ( slot_id TEXT PRIMARY KEY, portfolio_id TEXT NOT NULL, diff --git a/trex_core/adapters/persistence/sql/028_preserve_reconcile_warning_cases.sql b/trex_core/adapters/persistence/sql/028_preserve_reconcile_warning_cases.sql new file mode 100644 index 0000000..11e967d --- /dev/null +++ b/trex_core/adapters/persistence/sql/028_preserve_reconcile_warning_cases.sql @@ -0,0 +1,7 @@ +-- REQ-123 / issue #83: +-- Preserve simultaneous warning causes by keying warnings per (slot_id, case_code). +ALTER TABLE slot_reconcile_warning + DROP CONSTRAINT IF EXISTS slot_reconcile_warning_pkey; + +ALTER TABLE slot_reconcile_warning + ADD CONSTRAINT slot_reconcile_warning_pkey PRIMARY KEY (slot_id, case_code); diff --git a/trex_core/adapters/portfolio/reconcile_warning_store.py b/trex_core/adapters/portfolio/reconcile_warning_store.py index 731b565..91f8c7a 100644 --- a/trex_core/adapters/portfolio/reconcile_warning_store.py +++ b/trex_core/adapters/portfolio/reconcile_warning_store.py @@ -1,8 +1,4 @@ -"""PostgreSQL-backed durable reconcile-warning store (REQ-102, BAT-050). - -pause_store.py와 동형 설계. PK: slot_id. case_code: 'B' | 'C'. -Survives daemon restart (dcy-034 NFR-1). -""" +"""PostgreSQL-backed durable reconcile-warning store (REQ-102, BAT-050).""" from __future__ import annotations from datetime import datetime, timezone @@ -12,7 +8,7 @@ from trex_core.adapters.persistence.postgresql import _require_psycopg class PostgreSQLReconcileWarningStore: - """`slot_reconcile_warning` table CRUD. slot_id PK is the gate (REQ-102).""" + """`slot_reconcile_warning` table CRUD keyed by (slot_id, case_code).""" def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: self._psycopg = _require_psycopg(psycopg_module) @@ -34,9 +30,8 @@ class PostgreSQLReconcileWarningStore: "INSERT INTO slot_reconcile_warning " "(slot_id, portfolio_id, case_code, actor, reason, created_at, updated_at) " "VALUES (%s, %s, %s, %s, %s, %s, %s) " - "ON CONFLICT (slot_id) DO UPDATE " + "ON CONFLICT (slot_id, case_code) DO UPDATE " "SET portfolio_id = EXCLUDED.portfolio_id, " - " case_code = EXCLUDED.case_code, " " actor = EXCLUDED.actor, " " reason = EXCLUDED.reason, " " updated_at = EXCLUDED.updated_at", @@ -44,13 +39,19 @@ class PostgreSQLReconcileWarningStore: ) conn.commit() - def clear_warned(self, slot_id: str) -> None: + def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None: with self._psycopg.connect(self._dsn, autocommit=False) as conn: with conn.cursor() as cur: - cur.execute( - "DELETE FROM slot_reconcile_warning WHERE slot_id = %s", - (slot_id,), - ) + if case_code is None: + cur.execute( + "DELETE FROM slot_reconcile_warning WHERE slot_id = %s", + (slot_id,), + ) + else: + cur.execute( + "DELETE FROM slot_reconcile_warning WHERE slot_id = %s AND case_code = %s", + (slot_id, case_code), + ) conn.commit() def is_warned(self, slot_id: str) -> bool: @@ -65,7 +66,7 @@ class PostgreSQLReconcileWarningStore: def list_warned(self) -> set[str]: with self._psycopg.connect(self._dsn) as conn: with conn.cursor() as cur: - cur.execute("SELECT slot_id FROM slot_reconcile_warning") + cur.execute("SELECT DISTINCT slot_id FROM slot_reconcile_warning") return {row[0] for row in cur.fetchall()} def list_c_case_warned(self) -> set[str]: @@ -73,26 +74,30 @@ class PostgreSQLReconcileWarningStore: with self._psycopg.connect(self._dsn) as conn: with conn.cursor() as cur: cur.execute( - "SELECT slot_id FROM slot_reconcile_warning WHERE case_code = %s", + "SELECT DISTINCT slot_id FROM slot_reconcile_warning WHERE case_code = %s", ("C",), ) return {row[0] for row in cur.fetchall()} def get_warning_info(self, slot_id: str) -> dict | None: - """Return {"portfolio_id", "case_code", "actor", "reason"} or None (REQ-105/106).""" + """Return warning info with full active case set, or None.""" with self._psycopg.connect(self._dsn) as conn: with conn.cursor() as cur: cur.execute( "SELECT portfolio_id, case_code, actor, reason " - "FROM slot_reconcile_warning WHERE slot_id = %s LIMIT 1", + "FROM slot_reconcile_warning WHERE slot_id = %s " + "ORDER BY CASE WHEN case_code = 'C' THEN 0 ELSE 1 END, updated_at DESC", (slot_id,), ) - row = cur.fetchone() - if row is None: + rows = cur.fetchall() + if not rows: return None + case_codes = sorted({str(row[1]) for row in rows if row[1]}) + preferred = rows[0] return { - "portfolio_id": row[0], - "case_code": row[1], - "actor": row[2], - "reason": row[3], + "portfolio_id": preferred[0], + "case_code": preferred[1], + "case_codes": case_codes, + "actor": preferred[2], + "reason": preferred[3], } diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index 9e5f291..f50c04a 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -263,28 +263,24 @@ def make_slots_rpc_handler( for s in holding_slots: watchdog_store.update(str(s.slot_id)) - # Build merged slot rows (REQ-074) - # BAT-050 / REQ-104: warn_store passed to slot_to_dict for reconcile_warned field. - slot_rows = [ - slot_to_dict( - s, - snapshot_store=snapshot_store, - watchdog_store=watchdog_store, - pause_store=pause_store, - warn_store=warn_store, - watchdog_timeout_sec=watchdog_timeout_sec, - ) - for s in slots - ] - - # AC-0023 exclusion must precede all optional data readers so excluded - # slots cannot expose state or make an otherwise valid response fail. - excluded_count = 0 + warning_cases_by_slot: dict[str, set[str]] = {} if warn_store is not None: try: - c_warned_ids = warn_store.list_c_case_warned() + warned_slot_ids = warn_store.list_warned() + for sid in warned_slot_ids: + info = warn_store.get_warning_info(sid) + if info is None: + continue + raw_codes = info.get("case_codes") + if isinstance(raw_codes, list): + codes = {str(code) for code in raw_codes if code} + else: + case_code = str(info.get("case_code", "")) + codes = {case_code} if case_code else set() + if codes: + warning_cases_by_slot[str(sid)] = codes except Exception: # noqa: BLE001 - _log.exception("warn_store.list_c_case_warned() failed — fail-closed") + _log.exception("warn_store warning query failed — fail-closed") return RpcResponse(payload={ "slots": [], "available_slots": None, @@ -296,18 +292,42 @@ def make_slots_rpc_handler( ), "excluded_slots": {"reconcile_quantity_mismatch": 0}, }) - if c_warned_ids: - included = [ - (slot, row) - for slot, row in zip(slots, slot_rows, strict=True) - if not ( - slot.status.value == "holding" - and str(slot.slot_id) in c_warned_ids - ) - ] - excluded_count = len(slots) - len(included) + + # Build merged slot rows (REQ-074) + slot_rows = [ + slot_to_dict( + s, + snapshot_store=snapshot_store, + watchdog_store=watchdog_store, + pause_store=pause_store, + warning_cases_by_slot=warning_cases_by_slot, + watchdog_timeout_sec=watchdog_timeout_sec, + ) + for s in slots + ] + + # Exclusion must precede all optional data readers so blocked slots cannot + # expose state or make an otherwise valid response fail. + excluded_count = 0 + excluded_case_counts: dict[str, int] = {} + if warn_store is not None: + included: list[tuple[object, dict]] = [] + for slot, row in zip(slots, slot_rows, strict=True): + blocked_holding = ( + slot.status.value == "holding" and bool(row.get("reconcile_warned")) + ) + if blocked_holding: + excluded_count += 1 + for code in row.get("reconcile_warning_cases", []): + excluded_case_counts[code] = excluded_case_counts.get(code, 0) + 1 + continue + included.append((slot, row)) + if included: slots = [slot for slot, _ in included] slot_rows = [row for _, row in included] + else: + slots = [] + slot_rows = [] broker_qty_by_symbol: dict[str, int] = {} broker_qty_available = False @@ -410,7 +430,10 @@ def make_slots_rpc_handler( "message": message, } if warn_store is not None: - base_payload["excluded_slots"] = {"reconcile_quantity_mismatch": excluded_count} + base_payload["excluded_slots"] = { + "reconcile_quantity_mismatch": excluded_count, + "reconcile_warning_cases": excluded_case_counts, + } return RpcResponse(payload=base_payload) return handler diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index abe519b..ad2b7c0 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -359,7 +359,7 @@ class InMemoryKillSwitchStore: class ReconcileWarningStore(Protocol): - """Durable reconcile-warning overlay — slot_id PK (REQ-102, dcy-034).""" + """Durable reconcile-warning overlay keyed by (slot_id, case_code).""" def set_warned( self, @@ -372,7 +372,7 @@ class ReconcileWarningStore(Protocol): ) -> None: ... - def clear_warned(self, slot_id: str) -> None: + def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None: ... def is_warned(self, slot_id: str) -> bool: @@ -382,10 +382,14 @@ class ReconcileWarningStore(Protocol): ... def get_warning_info(self, slot_id: str) -> dict | None: - """Return {"portfolio_id", "case_code", "actor", "reason"} or None. + """Return warning info dict or None. - Used by verify to determine whether the warning is case_code='C' - (verify's responsibility) vs 'B' (sync's responsibility, REQ-106 OQ-C). + Returned dict includes: + {"portfolio_id", "case_code", "case_codes", "actor", "reason"} + where case_code is a backward-compatible preferred case and case_codes is + the full active case set for the slot. + + Used by verify to determine whether C-case is active (REQ-106 OQ-C). Returns None when the slot has no active warning. """ ... @@ -403,8 +407,8 @@ class InMemoryReconcileWarningStore: def __init__(self) -> None: self._lock = threading.Lock() - # slot_id -> {"portfolio_id", "case_code", "actor", "reason"} - self._warned: dict[str, dict] = {} + # slot_id -> case_code -> {"portfolio_id", "actor", "reason"} + self._warned: dict[str, dict[str, dict]] = {} def set_warned( self, @@ -416,16 +420,24 @@ class InMemoryReconcileWarningStore: reason: str = "", ) -> None: with self._lock: - self._warned[slot_id] = { + per_case = self._warned.setdefault(slot_id, {}) + per_case[case_code] = { "portfolio_id": portfolio_id, - "case_code": case_code, "actor": actor, "reason": reason, } - def clear_warned(self, slot_id: str) -> None: + def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None: with self._lock: - self._warned.pop(slot_id, None) + if case_code is None: + self._warned.pop(slot_id, None) + return + per_case = self._warned.get(slot_id) + if not per_case: + return + per_case.pop(case_code, None) + if not per_case: + self._warned.pop(slot_id, None) def is_warned(self, slot_id: str) -> bool: with self._lock: @@ -437,12 +449,23 @@ class InMemoryReconcileWarningStore: 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 + per_case = self._warned.get(slot_id) + if not per_case: + return None + case_codes = sorted(per_case.keys()) + preferred_case = "C" if "C" in per_case else case_codes[0] + preferred = per_case[preferred_case] + return { + "portfolio_id": preferred.get("portfolio_id", ""), + "case_code": preferred_case, + "case_codes": case_codes, + "actor": preferred.get("actor", ""), + "reason": preferred.get("reason", ""), + } 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"} + return {sid for sid, per_case in self._warned.items() if "C" in per_case} class ReconcileNotificationStore(Protocol): @@ -544,6 +567,7 @@ def slot_to_dict( watchdog_store=None, pause_store: PauseStore | None = None, warn_store: ReconcileWarningStore | None = None, + warning_cases_by_slot: dict[str, set[str]] | None = None, watchdog_timeout_sec: int = 120, ) -> dict: """Merged slot row for trex.slots.req (REQ-073/074). @@ -618,10 +642,23 @@ def slot_to_dict( # BAT-050 / REQ-104: reconcile_warned overlay (ReconcileWarningStore) — additive field. # Bulk list_warned() called once per RPC (caller pre-fetches; see make_slots_rpc_handler). # warn_store=None → reconcile_warned=False (backward compatible, NFR-4). - reconcile_warned = False - if warn_store is not None: - reconcile_warned = str(s.slot_id) in warn_store.list_warned() + warning_cases: set[str] = set() + if warning_cases_by_slot is not None: + warning_cases = set(warning_cases_by_slot.get(str(s.slot_id), set())) + elif warn_store is not None: + info = warn_store.get_warning_info(str(s.slot_id)) + if info is not None: + raw_cases = info.get("case_codes") + if isinstance(raw_cases, list): + warning_cases = {str(c) for c in raw_cases if c} + else: + case_code = str(info.get("case_code", "")) + if case_code: + warning_cases = {case_code} + reconcile_warned = bool(warning_cases) d["reconcile_warned"] = reconcile_warned + d["reconcile_warning_cases"] = sorted(warning_cases) + d["trade_eligible"] = not reconcile_warned return d @@ -637,4 +674,3 @@ def _sanitize(message: str) -> str: out = out[:idx] + "" break return out[:200] - diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index aca2c5d..bf326c1 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -629,7 +629,7 @@ def sync_slot_to_broker( Broker qty == 0 for slot → skipped with WARN (slot-side A-case guard). Broker-only (no slot) → NOT handled here; caller filters (FR-3). - After update, if slot.qty == broker_qty: clear_warned(slot_id). + After update, if slot.qty == broker_qty: clear_warned(slot_id, case_code="B"). """ from uuid import UUID @@ -870,14 +870,15 @@ def sync_slot_to_broker( after_avg = float(refreshed.avg_price) if (refreshed and refreshed.avg_price is not None) else None after = {"qty": after_qty, "status": after_status, "avg_price": after_avg} - # Clear warning when qty now matches broker + # Clear B-case warning when qty now matches broker warning_cleared = False if after_qty is not None and int(after_qty) == broker_qty: try: - warn_store.clear_warned(slot_id) + warn_store.clear_warned(slot_id, case_code="B") warning_cleared = True _log.info( - "sync: reconcile-warning cleared slot=%s symbol=%s broker_qty=%d actor=sync", + "sync: reconcile-warning cleared slot=%s symbol=%s case_code=B " + "broker_qty=%d actor=sync", slot_id, sym, broker_qty, ) except Exception: # noqa: BLE001 @@ -962,7 +963,7 @@ def verify_slot_reconcile( "reason": "no active reconcile-warning (already cleared or never set)", } - # Check case_code — verify is C-only (REQ-106 OQ-C). + # Check case set — verify is C-only (REQ-106 OQ-C). info = warn_store.get_warning_info(slot_id_str) if info is None: return { @@ -974,8 +975,13 @@ def verify_slot_reconcile( "reason": "warning exists but get_warning_info returned None (unexpected)", } - case_code = info.get("case_code", "") - if case_code != "C": + case_codes_raw = info.get("case_codes") + if isinstance(case_codes_raw, list): + case_codes = {str(code) for code in case_codes_raw if code} + else: + case_code = str(info.get("case_code", "")) + case_codes = {case_code} if case_code else set() + if "C" not in case_codes: return { "portfolio_id": portfolio_id, "slot_id": slot_id_str, @@ -983,7 +989,7 @@ def verify_slot_reconcile( "hub_qty": None, "broker_qty": None, "reason": ( - f"warning case_code='{case_code}' is not 'C' — " + f"warning case_codes={sorted(case_codes)} do not include 'C' — " "use 'trex-core sync' to resolve B-case (slot↔broker qty mismatch)" ), } @@ -1014,9 +1020,17 @@ def verify_slot_reconcile( continue sid_str = str(s.slot_id) warn_info = warn_store.get_warning_info(sid_str) - if warn_info is None or warn_info.get("case_code") != "C": + if warn_info is None: continue - warn_store.clear_warned(sid_str) + warned_codes_raw = warn_info.get("case_codes") + if isinstance(warned_codes_raw, list): + warned_codes = {str(code) for code in warned_codes_raw if code} + else: + case_code = str(warn_info.get("case_code", "")) + warned_codes = {case_code} if case_code else set() + if "C" not in warned_codes: + continue + warn_store.clear_warned(sid_str, case_code="C") store.set_warning(s.portfolio_id, False) cleared_slots.append(sid_str) _log.info( diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 128ef21..d985ad0 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -912,7 +912,15 @@ def _cmd_verify(args, cfg) -> int: c_slots: list[str] = [] for sid in warn_store.list_warned(): info = warn_store.get_warning_info(sid) - if info and info.get("case_code") == "C": + if not info: + continue + raw_codes = info.get("case_codes") + if isinstance(raw_codes, list): + case_codes = {str(code) for code in raw_codes if code} + else: + case_code = str(info.get("case_code", "")) + case_codes = {case_code} if case_code else set() + if "C" in case_codes: c_slots.append(info.get("portfolio_id", "")) target_pids = [p for p in c_slots if p] diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index d232973..8af08bb 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -198,7 +198,7 @@ class Daemon: # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader). self.watchdog = WatchdogStore() self.pause_store = PostgreSQLPauseStore(config.dsn) - # BAT-050 / REQ-102: durable reconcile-warning store (slot_id PK). + # BAT-050 / REQ-102: durable reconcile-warning store ((slot_id, case_code) PK). self.warn_store = PostgreSQLReconcileWarningStore(config.dsn) # req-0007 / ac-0024: durable reconcile notification deduplication store. self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)