fix(execution): KIS child cancel identity binding, sanitized logging, PG cancel SQL tests

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

tests/adapters/broker/test_kis_cancel.py

@@ -134,7 +134,7 @@ def test_full_fill_unaffected_by_cancel_flag(monkeypatch):
 
 # ── Regression: KIS child cancel row (orgn_odno) ─────────────────────────────
 
-def _payload_with_child_cancel(parent_odno: str, child_odno: str, ord_qty: str = "89"):
+def _payload_with_child_cancel(parent_odno: str, child_odno: str, ord_qty: str = "89", symbol: str = "225190"):
     """Simulate KIS returning original row (cncl_yn='') + separate cancel child."""
     parent = {
         "odno": parent_odno,
@@ -147,6 +147,7 @@ def _payload_with_child_cancel(parent_odno: str, child_odno: str, ord_qty: str =
         "rmn_qty": "0",
         "ord_dt": "20260901",
         "ord_gno_brno": "06010",
+        "pdno": symbol,
     }
     child = {
         "odno": child_odno,
@@ -159,6 +160,7 @@ def _payload_with_child_cancel(parent_odno: str, child_odno: str, ord_qty: str =
         "rmn_qty": "0",
         "ord_dt": "20260901",
         "ord_gno_brno": "06010",
+        "pdno": symbol,
     }
     return {"rt_cd": "0", "output1": [parent, child]}
 
@@ -175,7 +177,12 @@ def test_child_cancel_row_orgn_odno_maps_to_cancelled(monkeypatch):
 
 
 def test_correction_child_row_not_misclassified_as_cancel(monkeypatch):
-    """Correction child (cncl_yn='N') must not be treated as a cancel."""
+    """Correction child (cncl_yn!='Y') must not be treated as a cancel.
+
+    Issue #81 AC: a correction/replacement child means the order is superseded
+    but we cannot safely determine terminal status without the replacement
+    identity → fail-closed as 'unknown' (not 'accepted/open_confirmed').
+    """
     parent = {
         "odno": ODNO,
         "orgn_odno": "",
@@ -186,6 +193,7 @@ def test_correction_child_row_not_misclassified_as_cancel(monkeypatch):
         "rjct_qty": "0",
         "ord_dt": "20260901",
         "ord_gno_brno": "06010",
+        "pdno": "225190",
     }
     correction_child = {
         "odno": "0000123458",
@@ -197,6 +205,7 @@ def test_correction_child_row_not_misclassified_as_cancel(monkeypatch):
         "rjct_qty": "0",
         "ord_dt": "20260901",
         "ord_gno_brno": "06010",
+        "pdno": "225190",
     }
     monkeypatch.setattr(
         kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, correction_child]})
@@ -204,8 +213,12 @@ def test_correction_child_row_not_misclassified_as_cancel(monkeypatch):
     fs = _broker().poll_fill(
         f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901"
     )
-    # Correction child must not infer cancellation; original has no fill evidence
+    # Correction child means replacement identity is unresolved → fail-closed
-    assert fs.state == "accepted", f"expected accepted, got {fs.state}"
+    assert fs.state == "unknown", f"expected unknown, got {fs.state}"
+    raw = fs.raw if isinstance(fs.raw, dict) else {}
+    assert raw.get("reason") == "correction_child_present_unresolved"
+    # Must NOT be open_confirmed — that would allow new orders on the same slot
+    assert not raw.get("order_open_confirmed", False)
 
 
 def test_partial_fill_with_child_cancel_row(monkeypatch):
@@ -220,6 +233,7 @@ def test_partial_fill_with_child_cancel_row(monkeypatch):
         "rjct_qty": "0",
         "ord_dt": "20260901",
         "ord_gno_brno": "06010",
+        "pdno": "225190",
     }
     child = {
         "odno": "0000123459",
@@ -231,6 +245,7 @@ def test_partial_fill_with_child_cancel_row(monkeypatch):
         "rjct_qty": "0",
         "ord_dt": "20260901",
         "ord_gno_brno": "06010",
+        "pdno": "225190",
     }
     monkeypatch.setattr(
         kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, child]})
@@ -241,3 +256,138 @@ def test_partial_fill_with_child_cancel_row(monkeypatch):
     assert fs.state == "partially_filled"
     assert fs.cancelled and fs.is_partial_cancel
     assert fs.filled_qty == 4
+
+
+# ── Regression: identity mismatch / ambiguous child rows ─────────────────────
+
+def test_child_cancel_mismatched_branch_is_ignored(monkeypatch):
+    """Cancel child from a different branch (ord_gno_brno) must not affect parent."""
+    parent = {
+        "odno": ODNO,
+        "orgn_odno": "",
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "",
+        "rjct_qty": "0",
+        "ord_dt": "20260901",
+        "ord_gno_brno": "06010",
+        "pdno": "225190",
+    }
+    child_wrong_branch = {
+        "odno": "0000199999",
+        "orgn_odno": ODNO,
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "Y",
+        "rjct_qty": "0",
+        "ord_dt": "20260901",
+        "ord_gno_brno": "99999",  # different branch
+        "pdno": "225190",
+    }
+    monkeypatch.setattr(
+        kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, child_wrong_branch]})
+    )
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901"
+    )
+    # Child has wrong branch → not a matching cancel → parent stays accepted
+    assert fs.state == "accepted", f"expected accepted (mismatch branch), got {fs.state}"
+
+
+def test_child_cancel_mismatched_symbol_is_ignored(monkeypatch):
+    """Cancel child with different symbol (pdno) must not affect parent."""
+    parent = {
+        "odno": ODNO,
+        "orgn_odno": "",
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "",
+        "rjct_qty": "0",
+        "ord_dt": "20260901",
+        "ord_gno_brno": "06010",
+        "pdno": "225190",
+    }
+    child_wrong_symbol = {
+        "odno": "0000199998",
+        "orgn_odno": ODNO,
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "Y",
+        "rjct_qty": "0",
+        "ord_dt": "20260901",
+        "ord_gno_brno": "06010",
+        "pdno": "000660",  # different symbol
+    }
+    monkeypatch.setattr(
+        kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, child_wrong_symbol]})
+    )
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901"
+    )
+    assert fs.state == "accepted", f"expected accepted (mismatch symbol), got {fs.state}"
+
+
+def test_child_cancel_mismatched_date_is_ignored(monkeypatch):
+    """Cancel child with different ord_dt must not affect parent."""
+    parent = {
+        "odno": ODNO,
+        "orgn_odno": "",
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "",
+        "rjct_qty": "0",
+        "ord_dt": "20260901",
+        "ord_gno_brno": "06010",
+        "pdno": "225190",
+    }
+    child_wrong_date = {
+        "odno": "0000199997",
+        "orgn_odno": ODNO,
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "Y",
+        "rjct_qty": "0",
+        "ord_dt": "20260831",  # different date
+        "ord_gno_brno": "06010",
+        "pdno": "225190",
+    }
+    monkeypatch.setattr(
+        kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, child_wrong_date]})
+    )
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901"
+    )
+    assert fs.state == "accepted", f"expected accepted (mismatch date), got {fs.state}"
+
+
+def test_ambiguous_multiple_child_cancel_rows_return_unknown(monkeypatch):
+    """Two matching cancel children → ambiguous identity → fail-closed as unknown."""
+    parent = {
+        "odno": ODNO,
+        "orgn_odno": "",
+        "ord_qty": "10",
+        "tot_ccld_qty": "0",
+        "avg_prvs": "0",
+        "cncl_yn": "",
+        "rjct_qty": "0",
+        "ord_dt": "20260901",
+        "ord_gno_brno": "06010",
+        "pdno": "225190",
+    }
+    child1 = {**parent, "odno": "0000123460", "orgn_odno": ODNO, "cncl_yn": "Y", "ord_qty": "10"}
+    child2 = {**parent, "odno": "0000123461", "orgn_odno": ODNO, "cncl_yn": "Y", "ord_qty": "10"}
+    monkeypatch.setattr(
+        kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, child1, child2]})
+    )
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901"
+    )
+    assert fs.state == "unknown", f"expected unknown (ambiguous children), got {fs.state}"
+    raw = fs.raw if isinstance(fs.raw, dict) else {}
+    assert raw.get("reason") == "child_cancel_ambiguous"

tests/adapters/persistence/test_postgresql_repository.py

@@ -2,6 +2,8 @@
 Skipped without TREX_POSTGRESQL_TEST_DSN."""
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
+
 from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore  # noqa: F401
 from trex_core.adapters.persistence.postgresql import (
     PostgreSQLIntentStore,
@@ -244,3 +246,80 @@ def test_strategy_state_write_cannot_follow_close(pg_clean):
     result = store.set(slot.slot_id, "instance", expected_version=1, data="after", slot_active=True)
     assert not result.ok and result.reason == "slot_inactive"
     assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance"].data == "before"
+
+
+# ── Regression: reserve_cancel_attempt() SQL paths (issue #81 Fix A) ─────────
+# These tests exercise the real PostgreSQL code path that previously failed with
+# IndeterminateDatatype when retry_before=None was bound to %s IS NOT NULL.
+
+def test_reserve_cancel_first_attempt_retry_before_none_pg(pg_clean):
+    """First reservation with retry_before=None must succeed exactly once."""
+    repo, _, intents = _wire(pg_clean)
+    repo.reconcile_universe(INST, {"PF-1": "005930"})
+    d = intents.submit_buy(BuyIntent(intent_id="pg-can-first", qty=1, portfolio_id="PF-1"))
+    repo.mark_dispatched(d.slot_id, "ord-first")
+
+    # First reservation: retry_before=None — must not raise IndeterminateDatatype
+    assert repo.reserve_cancel_attempt(d.slot_id, "ord-first") is True
+
+    slot = repo.fetch_slot(d.slot_id)
+    assert slot.cancel_state == "unknown"
+    assert slot.cancel_attempts == 1
+    assert slot.cancel_requested_at is not None
+
+    # Duplicate CAS must be rejected (cancel_state is no longer NULL)
+    assert repo.reserve_cancel_attempt(d.slot_id, "ord-first") is False
+    assert repo.fetch_slot(d.slot_id).cancel_attempts == 1  # not incremented
+
+
+def test_reserve_cancel_retry_backoff_gated_pg(pg_clean):
+    """Retry path with retry_before is accepted when backoff window passes."""
+    import psycopg
+
+    repo, _, intents = _wire(pg_clean)
+    repo.reconcile_universe(INST, {"PF-1": "005930"})
+    d = intents.submit_buy(BuyIntent(intent_id="pg-can-retry", qty=1, portfolio_id="PF-1"))
+    repo.mark_dispatched(d.slot_id, "ord-retry")
+
+    # Simulate a previous cancel attempt that left state=rejected a while ago
+    past = datetime.now(tz=UTC) - timedelta(seconds=60)
+    with psycopg.connect(pg_clean, autocommit=False) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "UPDATE daily_slot SET cancel_state='rejected', cancel_requested_at=%s, "
+                "cancel_attempts=1 WHERE slot_id=%s",
+                (past, d.slot_id),
+            )
+        conn.commit()
+
+    # retry_before = now means backoff window has elapsed
+    retry_before = datetime.now(tz=UTC)
+    assert repo.reserve_cancel_attempt(d.slot_id, "ord-retry", retry_before=retry_before) is True
+
+    slot = repo.fetch_slot(d.slot_id)
+    assert slot.cancel_state == "unknown"
+    assert slot.cancel_attempts == 2
+
+
+def test_reserve_cancel_retry_max_attempts_blocked_pg(pg_clean):
+    """Retry is rejected when cancel_attempts >= 4."""
+    import psycopg
+
+    repo, _, intents = _wire(pg_clean)
+    repo.reconcile_universe(INST, {"PF-1": "005930"})
+    d = intents.submit_buy(BuyIntent(intent_id="pg-can-max", qty=1, portfolio_id="PF-1"))
+    repo.mark_dispatched(d.slot_id, "ord-max")
+
+    past = datetime.now(tz=UTC) - timedelta(seconds=120)
+    with psycopg.connect(pg_clean, autocommit=False) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "UPDATE daily_slot SET cancel_state='rejected', cancel_requested_at=%s, "
+                "cancel_attempts=4 WHERE slot_id=%s",
+                (past, d.slot_id),
+            )
+        conn.commit()
+
+    retry_before = datetime.now(tz=UTC)
+    assert repo.reserve_cancel_attempt(d.slot_id, "ord-max", retry_before=retry_before) is False
+    assert repo.fetch_slot(d.slot_id).cancel_attempts == 4  # unchanged

tests/runtime/test_daemon_execution_loop_exception_logging.py

@@ -1,19 +1,22 @@
-"""Regression: execution loop exceptions must be logged, not silently swallowed.
+"""Regression: execution loop exceptions must be logged without leaking secrets.
 
-dcy-031 / issue #81: exceptions from execution.run_once() were silently ignored
+dcy-031 / issue #81 (Fix C):
-via bare `continue`, making it impossible to observe repository or adapter errors.
+- Exceptions from execution.run_once() were silently ignored via bare `continue`.
+- Sanitized logging requirement: only type(exc).__name__ is emitted; the
+  exception message, traceback, and exc_info must NOT appear in the log record
+  so adapter exceptions containing credentials/DSN/raw payload are never exposed.
+- Worker must continue after the error (loop survives).
 """
 from __future__ import annotations
 
 import logging
 import threading
-import types
 
-import pytest
+_SENTINEL = "SECRET-DSN-CREDENTIAL-RAWPAYLOAD-DO-NOT-LOG"
 
 
-def test_execution_loop_logs_exception(caplog):
+def test_execution_loop_logs_exception_without_leaking_secrets(caplog):
-    """When execution.run_once() raises, the error is logged and the loop continues."""
+    """Exception type is logged; secret sentinel in exc message is NOT logged."""
     import trex_core.runtime.daemon as daemon_mod
 
     call_count = 0
@@ -24,7 +27,8 @@ def test_execution_loop_logs_exception(caplog):
             nonlocal call_count
             call_count += 1
             if call_count == 1:
-                raise RuntimeError("simulated repo error: IndeterminateDatatype")
+                raise RuntimeError(f"dsn=postgresql://user:{_SENTINEL}@host/db")
+            # Stop after second successful tick
             stop_event.set()
 
     class _FakePublisher:
@@ -34,8 +38,6 @@ def test_execution_loop_logs_exception(caplog):
     class _FakeConfig:
         execution_interval_sec = 0.001
 
-    # Build a minimal stub Daemon without invoking its real __init__, so we can
-    # inject only the attributes _execution_loop reads.
     daemon = object.__new__(daemon_mod.Daemon)
     daemon.stop = stop_event
     daemon.config = _FakeConfig()
@@ -48,6 +50,21 @@ def test_execution_loop_logs_exception(caplog):
         t.join(timeout=2.0)
 
     assert not t.is_alive(), "execution loop did not exit cleanly"
-    assert any(
+
-        "execution loop error" in r.message for r in caplog.records
+    error_records = [r for r in caplog.records if "execution loop error" in r.message]
-    ), f"expected 'execution loop error' log, got: {[r.message for r in caplog.records]}"
+    assert error_records, (
+        f"expected 'execution loop error' log record, got: {[r.message for r in caplog.records]}"
+    )
+
+    # Secret sentinel must NOT appear in any log record (message or exc_info output)
+    for rec in caplog.records:
+        assert _SENTINEL not in rec.getMessage(), (
+            f"secret sentinel leaked into log: {rec.getMessage()}"
+        )
+        # exc_info must not be attached
+        assert rec.exc_info is None or rec.exc_info == (None, None, None), (
+            "exc_info must not be set on execution loop error record"
+        )
+
+    # The loop continued: second call_count tick was reached
+    assert call_count >= 2, "loop did not continue after exception"

trex_core/adapters/broker/kis.py

@@ -443,22 +443,70 @@ class KISBroker:
         # dcy-013 ext: KIS sometimes records an external cancel as a separate
         # child row with orgn_odno=<original> and cncl_yn="Y" rather than
         # setting cncl_yn on the original row.  Scan sibling rows to detect
-        # this case.  Correction (정정) children will NOT have cncl_yn="Y" so
+        # this case.
-        # they are not misclassified as cancels here.
+        #
-        child_cancel_row = next(
+        # Safety requirements:
-            (
+        #   - Child must share the parent's canonical identity:
-                r
+        #     ord_gno_brno (branch), ord_dt (date), pdno (symbol).
-                for r in rows
+        #     Mismatched or ambiguous children → fail-closed as "unknown".
-                if (str(r.get("orgn_odno") or "").strip().lstrip("0") or "0") == normalized_odno
+        #   - Correction (정정) children have cncl_yn != "Y"; they are not
-                and str(r.get("cncl_yn") or "").strip().upper() == "Y"
+        #     treated as cancels here.  If a correction child is present with
-            ),
+        #     a new odno the evidence is ambiguous → "unknown" so the slot is
-            None,
+        #     not incorrectly unlocked.
-        )
+        parent_brno = str(row.get("ord_gno_brno") or "").strip()
-        if child_cancel_row is not None and str(row.get("cncl_yn") or "").strip().upper() != "Y":
+        parent_dt = str(row.get("ord_dt") or "").strip()
+        parent_pdno = str(row.get("pdno") or "").strip()
+
+        child_cancel_candidates = [
+            r
+            for r in rows
+            if (str(r.get("orgn_odno") or "").strip().lstrip("0") or "0") == normalized_odno
+            and str(r.get("cncl_yn") or "").strip().upper() == "Y"
+            and str(r.get("ord_gno_brno") or "").strip() == parent_brno
+            and str(r.get("ord_dt") or "").strip() == parent_dt
+            and str(r.get("pdno") or "").strip() == parent_pdno
+        ]
+
+        # Correction children: same orgn_odno but cncl_yn != "Y".  If present,
+        # the situation is ambiguous — a replacement order may be live.
+        correction_candidates = [
+            r
+            for r in rows
+            if (str(r.get("orgn_odno") or "").strip().lstrip("0") or "0") == normalized_odno
+            and str(r.get("cncl_yn") or "").strip().upper() != "Y"
+        ]
+
+        if len(child_cancel_candidates) > 1:
+            # More than one matching cancel child — identity is ambiguous.
+            return FillStatus(
+                state="unknown",
+                raw={
+                    "reason": "child_cancel_ambiguous",
+                    "order_visible": True,
+                    "inquiry_date": inquiry_date,
+                },
+            )
+
+        if correction_candidates:
+            # A correction/replacement child is present; the original order is
+            # superseded but we cannot safely infer terminal status without
+            # explicit broker confirmation of the replacement identity.
+            return FillStatus(
+                state="unknown",
+                raw={
+                    "reason": "correction_child_present_unresolved",
+                    "order_visible": True,
+                    "inquiry_date": inquiry_date,
+                },
+            )
+
+        if child_cancel_candidates and str(row.get("cncl_yn") or "").strip().upper() != "Y":
+            child_cancel_row = child_cancel_candidates[0]
             # Inject the cancel signal from the child row into the parent row so
             # _fill_status_from_row can treat it as cancelled, while keeping
             # the original row's fill quantities authoritative.
             row = {**row, "cncl_yn": "Y", "_child_cancel_odno": str(child_cancel_row.get("odno") or "")}
+
         status = self._fill_status_from_row(
             row, side=side, today=inquiry_date, fallback_odno=odno
         )

trex_core/runtime/daemon.py

@@ -408,8 +408,14 @@ class Daemon:
             self.event_publisher.keepalive_tick()
             try:
                 self.execution.run_once()
-            except Exception:  # noqa: BLE001 - loop must survive transient errors
+            except Exception as exc:  # noqa: BLE001 - loop must survive transient errors
-                _log.exception("execution loop error — worker continues")
+                # Sanitized: emit only the exception type, not the message or
+                # traceback, so credentials / DSN / raw payloads in adapter
+                # exceptions are never written to the log stream.
+                _log.error(
+                    "execution loop error — worker continues: exc_type=%s",
+                    type(exc).__name__,
+                )
 
     def _ledger_loop(self) -> None:
         while not self.stop.wait(self.config.ledger_interval_sec):