trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve multi-cause reconcile warnings and align slot blocking

Browse Source 
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-09-12 13:22:40 +00:00
committed by GitHub
co-authored by genichin
parent e47bebacaf
commit 45a5e4157d
16 changed files with 337 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/adapters/persistence/sql/016_create_slot_reconcile_warning.sql
+2 -2
View File
@@ -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,
trex_core/adapters/persistence/sql/028_preserve_reconcile_warning_cases.sql
+7
View File
@@ -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);
trex_core/adapters/portfolio/reconcile_warning_store.py
+28 -23
View File
@@ -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],
}
trex_core/application/messaging_handlers.py
+53 -30
View File
@@ -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
trex_core/application/portfolio.py
+54 -18
View File
@@ -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] + "<redacted>"
break
return out[:200]
trex_core/application/reconciliation.py
+24 -10
View File
@@ -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(
trex_core/runtime/cli.py
+9 -1
View File
@@ -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]
trex_core/runtime/daemon.py
+1 -1
View File
@@ -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)