diff --git a/tests/adapters/broker/test_kis_cancel.py b/tests/adapters/broker/test_kis_cancel.py index b18d074..8494f0a 100644 --- a/tests/adapters/broker/test_kis_cancel.py +++ b/tests/adapters/broker/test_kis_cancel.py @@ -7,7 +7,7 @@ pre-seeded, httpx.get faked) and assert the FillStatus mapping. from __future__ import annotations -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from trex_core.adapters.broker import kis as kis_mod from trex_core.adapters.broker.kis import KISBroker @@ -23,7 +23,7 @@ def _broker(): base_url="https://openapivts.koreainvestment.com:29443", # mock endpoint ) b._token = "tok" - b._token_exp = datetime.now(tz=timezone.utc) + timedelta(hours=1) + b._token_exp = datetime.now(tz=UTC) + timedelta(hours=1) return b @@ -52,6 +52,7 @@ def _row(**kw): "rjct_qty": "0", "ord_dt": "20260608", "ord_gno_brno": "06010", + "pdno": "005930", } base.update(kw) return {"rt_cd": "0", "output1": [base]} @@ -130,3 +131,519 @@ def test_full_fill_unaffected_by_cancel_flag(monkeypatch): f"06010:{ODNO}", side="sell", symbol="005930", order_date="20260608" ) assert fs.is_filled and fs.filled_qty == 10 and not fs.is_partial_cancel + + +# ── Regression: KIS child cancel row (orgn_odno) ───────────────────────────── + +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, + "orgn_odno": "", + "ord_qty": ord_qty, + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "", + "rjct_qty": "0", + "rmn_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + "pdno": symbol, + } + child = { + "odno": child_odno, + "orgn_odno": parent_odno, + "ord_qty": ord_qty, + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "Y", + "rjct_qty": "0", + "rmn_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + "pdno": symbol, + } + return {"rt_cd": "0", "output1": [parent, child]} + + +def test_child_cancel_row_orgn_odno_maps_to_cancelled(monkeypatch): + """KIS child row with orgn_odno= + cncl_yn=Y → full cancel.""" + payload = _payload_with_child_cancel(ODNO, "0000123457") + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + assert fs.state == "cancelled", f"expected cancelled, got {fs.state}" + assert fs.is_terminal_reject + + +def test_correction_child_row_not_misclassified_as_cancel(monkeypatch): + """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": "", + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "", + "rjct_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + "pdno": "225190", + } + correction_child = { + "odno": "0000123458", + "orgn_odno": ODNO, + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "N", # correction, not cancel + "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]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + # Correction child means replacement identity is unresolved → fail-closed + 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): + """Partial fill on parent + cancel child → partial_cancel.""" + parent = { + "odno": ODNO, + "orgn_odno": "", + "ord_qty": "10", + "tot_ccld_qty": "4", + "avg_prvs": "70000", + "cncl_yn": "", + "rjct_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + "pdno": "225190", + } + child = { + "odno": "0000123459", + "orgn_odno": ODNO, + "ord_qty": "6", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "Y", + "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]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + 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" + + +# ── Regression: parent row identity mismatch with request ──────────────────── + +def test_parent_row_wrong_branch_returns_unknown(monkeypatch): + """Parent row with wrong ord_gno_brno (branch) relative to broker_order_id → 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": "99999", # wrong branch — request has 06010 + "pdno": "225190", + } + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent]}) + ) + # broker_order_id encodes branch 06010, but row has 99999 + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + assert fs.state == "unknown", f"expected unknown (wrong parent branch), got {fs.state}" + raw = fs.raw if isinstance(fs.raw, dict) else {} + assert "mismatch" in raw.get("reason", "") + + +def test_parent_row_wrong_date_returns_unknown(monkeypatch): + """Parent row with wrong ord_dt relative to order_date arg → unknown.""" + parent = { + "odno": ODNO, + "orgn_odno": "", + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "", + "rjct_qty": "0", + "ord_dt": "20260831", # wrong date — request has 20260901 + "ord_gno_brno": "06010", + "pdno": "225190", + } + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + assert fs.state == "unknown", f"expected unknown (wrong parent date), got {fs.state}" + raw = fs.raw if isinstance(fs.raw, dict) else {} + assert "mismatch" in raw.get("reason", "") + + +def test_parent_row_wrong_symbol_returns_unknown(monkeypatch): + """Parent row with wrong pdno (symbol) relative to symbol arg → 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": "000660", # wrong symbol — request has 225190 + } + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + assert fs.state == "unknown", f"expected unknown (wrong parent symbol), got {fs.state}" + raw = fs.raw if isinstance(fs.raw, dict) else {} + assert "mismatch" in raw.get("reason", "") + + +def test_correction_child_wrong_identity_is_not_treated_as_correction(monkeypatch): + """Correction child with wrong branch/date/symbol must not trigger correction_child_present_unresolved.""" + 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", + } + unrelated_correction = { + "odno": "0000199990", + "orgn_odno": ODNO, + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "N", + "rjct_qty": "0", + "ord_dt": "20260831", # different date → not related to our request + "ord_gno_brno": "99999", # different branch + "pdno": "000660", # different symbol + } + monkeypatch.setattr( + kis_mod.httpx, + "get", + _fake_get({"rt_cd": "0", "output1": [parent, unrelated_correction]}), + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + # Unrelated correction must not cause correction_child_present_unresolved + assert fs.state == "accepted", f"expected accepted (unrelated correction), got {fs.state}" + raw = fs.raw if isinstance(fs.raw, dict) else {} + assert raw.get("reason") != "correction_child_present_unresolved" + + +# ── Regression: parent identity boundary cases (missing / wrong-first / ambiguous) ─ + +def test_parent_missing_identity_fields_returns_unknown(monkeypatch): + """Parent row with missing pdno and ord_dt → treated as identity mismatch (fail-closed).""" + parent_no_identity = { + "odno": ODNO, + "orgn_odno": "", + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "Y", # would be cancelled if accepted + "rjct_qty": "0", + # ord_dt, ord_gno_brno, pdno intentionally absent + } + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent_no_identity]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + # Missing identity → fail-closed, not terminal + assert fs.state == "unknown", f"expected unknown (missing identity), got {fs.state}" + assert not fs.is_terminal_reject + + +def test_child_cancel_missing_identity_fields_not_accepted_as_cancel(monkeypatch): + """Cancel child with missing identity fields must NOT be accepted as matching child.""" + 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_no_identity = { + "odno": "0000123461", + "orgn_odno": ODNO, + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "Y", + "rjct_qty": "0", + # ord_dt, ord_gno_brno, pdno intentionally absent + } + monkeypatch.setattr( + kis_mod.httpx, + "get", + _fake_get({"rt_cd": "0", "output1": [parent, child_no_identity]}), + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + # Child has missing identity → not a matching cancel; parent has no fill → accepted + assert fs.state == "accepted", f"expected accepted (child missing identity), got {fs.state}" + + +def test_wrong_parent_first_exact_parent_later_finds_cancel(monkeypatch): + """When the wrong-branch parent comes first, the exact parent+cancel child later must win.""" + wrong_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": "99999", # wrong branch — comes first in list + "pdno": "225190", + } + exact_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", # correct branch — comes second + "pdno": "225190", + } + child = { + "odno": "0000123462", + "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": "225190", + } + monkeypatch.setattr( + kis_mod.httpx, + "get", + _fake_get({"rt_cd": "0", "output1": [wrong_parent, exact_parent, child]}), + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + # Must find the exact parent + cancel child even though wrong parent comes first + assert fs.state == "cancelled", f"expected cancelled (exact parent later), got {fs.state}" + + +def test_ambiguous_exact_parents_returns_unknown(monkeypatch): + """Two rows with identical exact identity → ambiguous parent → unknown.""" + parent1 = { + "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", + } + parent2 = {**parent1} # exact duplicate + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent1, parent2]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + assert fs.state == "unknown", f"expected unknown (ambiguous parents), got {fs.state}" + raw = fs.raw if isinstance(fs.raw, dict) else {} + assert raw.get("reason") == "parent_identity_ambiguous" diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index c422c00..01d13d0 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/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 diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index 1fae52f..e606db3 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -179,3 +179,115 @@ def test_operator_cancel_is_sent_once_across_repeat_and_restart(): assert restarted.cancel_pending_slot(sid) == "cancel_already_recorded" assert broker.cancel_calls == 1 assert repo.fetch_slot(sid).cancel_state == CancelState.REQUESTED.value + + +# ── Regression: first-cancel reservation (retry_before=None) ───────────────── + +def test_first_cancel_reservation_with_no_retry_before_succeeds(): + """First cancel reservation with retry_before=None must reserve exactly once (fix for + PostgreSQL IndeterminateDatatype; the in-memory repo mirrors the corrected SQL path).""" + broker = ProgrammableBroker() + repo, svc, intents, execution, _, _ = _wire(broker) + sid = _buy_to_holding(intents, execution) + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + broker.responder = lambda side: FillStatus(state="accepted") + execution.run_once() # dispatch + + # First operator cancel — retry_before=None path + result = execution.cancel_pending_slot(sid) + assert result == "ok", f"expected ok, got {result}" + assert broker.cancel_calls == 1 + # Second call must be idempotent (cancel_state already set) + result2 = execution.cancel_pending_slot(sid) + assert result2 == "cancel_already_recorded" + assert broker.cancel_calls == 1 # no second broker call + + +def test_timeout_cancel_first_attempt_with_retry_before_none(): + """Timeout cancel's first attempt passes retry_before=None; must call broker once.""" + broker = ProgrammableBroker() + repo, svc, intents, execution, _, _ = _wire(broker) + sid = _buy_to_holding(intents, execution) + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + broker.responder = lambda side: FillStatus( + state="accepted", + raw={"order_open_confirmed": True}, + ) + execution.run_once() # dispatch + + # Simulate timeout: set order_submitted_at well in the past + s = repo.fetch_slot(sid) + repo._slots[sid] = replace( + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120), + order_date="20260901", + timeout_sec=30, + ) + execution.run_once() # should trigger timeout cancel + assert broker.cancel_calls == 1 + + +# ── Regression: concurrent timeout/operator cancel race — broker call count 1 ─ + +def test_timeout_operator_race_broker_called_exactly_once(): + """Timeout cancel loop and operator cancel_pending_slot race to the same first + reservation. The repository CAS guarantees exactly one winner; broker must + receive cancel_order exactly once regardless of scheduling order. + + Uses a threading barrier so both paths enter reserve_cancel_attempt + simultaneously, making the race deterministic. + """ + import threading + + broker = ProgrammableBroker() + repo, _, intents, execution, _, _ = _wire(broker) + sid = _buy_to_holding(intents, execution) + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + broker.responder = lambda side: FillStatus( + state="accepted", + raw={"order_open_confirmed": True}, + ) + execution.run_once() # dispatch sell + + # Backdate so timeout_cancel fires in the next run_once() + s = repo.fetch_slot(sid) + repo._slots[sid] = replace( + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120), + order_date="20260901", + timeout_sec=30, + ) + + barrier = threading.Barrier(2) + original_reserve = repo.reserve_cancel_attempt + + def _barrier_reserve(slot_id, broker_order_id, *, retry_before=None): + barrier.wait(timeout=5) + return original_reserve(slot_id, broker_order_id, retry_before=retry_before) + + repo.reserve_cancel_attempt = _barrier_reserve + + errors = [] + + def _operator(): + try: + execution.cancel_pending_slot(sid) + except Exception as exc: # noqa: BLE001 + errors.append(exc) + + t = threading.Thread(target=_operator) + t.start() + + # Execution loop's timeout cancel is the second concurrent caller. + try: + execution.run_once() + except Exception as exc: # noqa: BLE001 + errors.append(exc) + + t.join(timeout=5) + assert not errors, f"thread errors: {errors}" + + # CAS ensures exactly one winner — broker cancel must be called once. + assert broker.cancel_calls == 1, ( + f"expected exactly 1 broker cancel call, got {broker.cancel_calls}" + ) diff --git a/tests/runtime/test_daemon_execution_loop_exception_logging.py b/tests/runtime/test_daemon_execution_loop_exception_logging.py new file mode 100644 index 0000000..5657fb2 --- /dev/null +++ b/tests/runtime/test_daemon_execution_loop_exception_logging.py @@ -0,0 +1,70 @@ +"""Regression: execution loop exceptions must be logged without leaking secrets. + +dcy-031 / issue #81 (Fix C): +- 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 + +_SENTINEL = "SECRET-DSN-CREDENTIAL-RAWPAYLOAD-DO-NOT-LOG" + + +def test_execution_loop_logs_exception_without_leaking_secrets(caplog): + """Exception type is logged; secret sentinel in exc message is NOT logged.""" + import trex_core.runtime.daemon as daemon_mod + + call_count = 0 + stop_event = threading.Event() + + class _FakeExecution: + def run_once(self): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError(f"dsn=postgresql://user:{_SENTINEL}@host/db") + # Stop after second successful tick + stop_event.set() + + class _FakePublisher: + def keepalive_tick(self): + pass + + class _FakeConfig: + execution_interval_sec = 0.001 + + daemon = object.__new__(daemon_mod.Daemon) + daemon.stop = stop_event + daemon.config = _FakeConfig() + daemon.event_publisher = _FakePublisher() + daemon.execution = _FakeExecution() + + with caplog.at_level(logging.ERROR, logger="trex_core.runtime.daemon"): + t = threading.Thread(target=daemon._execution_loop, daemon=True) + t.start() + t.join(timeout=2.0) + + assert not t.is_alive(), "execution loop did not exit cleanly" + + error_records = [r for r in caplog.records if "execution loop error" in r.message] + 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" diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index abb6982..32dadcd 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -413,15 +413,56 @@ class KISBroker: self._fill_poll_not_before[broker_order_id] = monotonic() + _FILL_POLL_MIN_INTERVAL_SECONDS rows = payload.get("output1") or [] normalized_odno = odno.lstrip("0") or "0" - row = next( - ( - r - for r in rows - if (str(r.get("odno") or "").strip().lstrip("0") or "0") == normalized_odno - ), - None, - ) - if row is None: + # Canonical request identity from the caller — these are authoritative + # and must not be overridden by whatever the response row contains. + request_brno = broker_order_id.split(":")[0] if ":" in broker_order_id else "" + request_date = inquiry_date # == order_date + request_symbol = symbol + + def _row_odno_normalized(r: dict) -> str: + return str(r.get("odno") or "").strip().lstrip("0") or "0" + + def _row_matches_request_identity(r: dict) -> bool: + """True when a row's identity fields are present and agree with the + authoritative request identity. An absent field in the row is + treated as a mismatch (fail-closed), not as a wildcard.""" + row_brno = str(r.get("ord_gno_brno") or "").strip() + row_dt = str(r.get("ord_dt") or "").strip() + row_pdno = str(r.get("pdno") or "").strip() + # brno: required only when request carries one; if request has no + # brno (legacy odno-only format) skip the check to preserve + # backwards compatibility. + if request_brno and (not row_brno or row_brno != request_brno): + return False + # date and symbol: always required — these never legitimately vary + # for the same order on the same inquiry date. + if not row_dt or row_dt != request_date: + return False + return bool(row_pdno) and row_pdno == request_symbol + + # Collect ALL rows whose odno matches. We must then require exactly + # one that also passes the full identity check. Using next() here + # would accept the first odno-match regardless of identity and stop + # the search, missing authoritative rows that come later and ignoring + # the ambiguous-parent edge case. + odno_matches = [ + r for r in rows if _row_odno_normalized(r) == normalized_odno + ] + exact_parent_matches = [r for r in odno_matches if _row_matches_request_identity(r)] + + if len(exact_parent_matches) == 0: + # No row with the exact request identity. If there are odno + # matches they all failed identity validation → unknown (not grace). + if odno_matches: + return FillStatus( + state="unknown", + raw={ + "reason": "parent_identity_mismatch", + "order_visible": True, + "inquiry_date": inquiry_date, + }, + ) + # Truly invisible — apply the submission visibility grace window. visible_grace = ( submitted_at is None or (self._clock().astimezone(UTC) - submitted_at.astimezone(UTC)) @@ -440,6 +481,94 @@ class KISBroker: "inquiry_date": inquiry_date, }, ) + + if len(exact_parent_matches) > 1: + # More than one row claims the same exact identity — ambiguous. + return FillStatus( + state="unknown", + raw={ + "reason": "parent_identity_ambiguous", + "order_visible": True, + "inquiry_date": inquiry_date, + }, + ) + + row = exact_parent_matches[0] + + # dcy-013 ext: KIS sometimes records an external cancel as a separate + # child row with orgn_odno= and cncl_yn="Y" rather than + # setting cncl_yn on the original row. Scan sibling rows to detect + # this case. + # + # Safety requirements: + # - Child must share the AUTHORITATIVE request identity (branch, date, + # symbol from the caller). All identity fields must be present and + # equal — absent fields are NOT treated as wildcards (fail-closed). + # - Ambiguous (> 1 matching cancel) → fail-closed "unknown". + # - Correction (정정) children (cncl_yn != "Y") that match the same + # request identity are also fail-closed → "unknown". + # - Unrelated rows whose identity fields differ from the request are + # excluded from both candidate lists. + def _child_matches_request(r: dict) -> bool: + """True when a child row's identity fields are all present and + equal to the authoritative request identity. Missing = mismatch.""" + child_brno = str(r.get("ord_gno_brno") or "").strip() + child_dt = str(r.get("ord_dt") or "").strip() + child_pdno = str(r.get("pdno") or "").strip() + if request_brno and (not child_brno or child_brno != request_brno): + return False + if not child_dt or child_dt != request_date: + return False + return bool(child_pdno) and child_pdno == request_symbol + + 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 _child_matches_request(r) + ] + + # Correction children: same orgn_odno + matching identity but cncl_yn != "Y". + 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" + and _child_matches_request(r) + ] + + 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 ) diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index cbe5154..6a52b3d 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -385,20 +385,37 @@ class PostgreSQLSlotRepository: *, retry_before: Optional[datetime] = None, ) -> bool: - """Atomically reserve first cancel or an authoritative backoff-gated retry.""" + """Atomically reserve first cancel or an authoritative backoff-gated retry. + + Two separate SQL paths avoid nullable-placeholder type-inference errors + (PostgreSQL IndeterminateDatatype when binding None to %s IS NOT NULL). + """ now = datetime.now(tz=timezone.utc) with self._psycopg.connect(self._dsn, autocommit=False) as conn: with conn.cursor() as cur: - cur.execute( - "UPDATE daily_slot SET cancel_state='unknown', cancel_requested_at=%s, " - "cancel_attempts=cancel_attempts+1, updated_at=%s " - "WHERE slot_id=%s AND broker_order_id=%s " - "AND status IN ('pending_buy','pending_sell') " - "AND (cancel_state IS NULL OR " - "(%s IS NOT NULL AND cancel_state IN ('rejected','unknown') " - "AND cancel_attempts < 4 AND cancel_requested_at <= %s))", - (now, now, slot_id, broker_order_id, retry_before, retry_before), - ) + if retry_before is None: + # First reservation: only match when no cancel has been + # attempted yet (cancel_state IS NULL). + cur.execute( + "UPDATE daily_slot SET cancel_state='unknown', cancel_requested_at=%s, " + "cancel_attempts=cancel_attempts+1, updated_at=%s " + "WHERE slot_id=%s AND broker_order_id=%s " + "AND status IN ('pending_buy','pending_sell') " + "AND cancel_state IS NULL", + (now, now, slot_id, broker_order_id), + ) + else: + # Retry: backoff-gated, only match retryable states. + cur.execute( + "UPDATE daily_slot SET cancel_state='unknown', cancel_requested_at=%s, " + "cancel_attempts=cancel_attempts+1, updated_at=%s " + "WHERE slot_id=%s AND broker_order_id=%s " + "AND status IN ('pending_buy','pending_sell') " + "AND cancel_state IN ('rejected','unknown') " + "AND cancel_attempts < 4 " + "AND cancel_requested_at <= %s", + (now, now, slot_id, broker_order_id, retry_before), + ) ok = cur.rowcount == 1 conn.commit() return ok diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 9e8e61e..d232973 100644 --- a/trex_core/runtime/daemon.py +++ b/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 - continue + except Exception as exc: # noqa: BLE001 - loop must survive transient errors + # 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):