fix: reconcile KIS correction order chains

tests/adapters/broker/test_kis.py

@@ -7,6 +7,7 @@ Unit tests run always; the sandbox integration test runs only when KIS
 from __future__ import annotations
 
 import os
+from datetime import UTC, datetime, timedelta
 
 import httpx
 import pytest
@@ -564,7 +565,9 @@ def test_submit_unknown_confirmed_absent_becomes_retryable_reject(monkeypatch):
     status = b.poll_fill(ack.broker_order_id, side="buy", symbol="005930")
 
     assert status.state == "rejected"
-    assert status.raw == {"reason": "submission_confirmed_absent"}
+    assert status.raw["reason"] == "submission_confirmed_absent"
+    assert status.raw["poll_performed"] is True
+    assert status.raw["poll_count"] == 1
 
 
 def test_fill_poll_transport_failure_uses_backoff(monkeypatch):
@@ -588,8 +591,12 @@ def test_fill_poll_transport_failure_uses_backoff(monkeypatch):
     )
 
     assert first.state == "unknown"
+    assert first.raw["reason"] == "fill_inquiry_transport"
+    assert "timed out" not in str(first.raw)
     assert immediate.state == "accepted"
-    assert immediate.raw == {"reason": "fill_poll_backoff"}
+    assert immediate.raw["reason"] == "fill_poll_backoff"
+    assert immediate.raw["poll_performed"] is False
+    assert immediate.raw["poll_count"] == 1
     assert len(calls) == 1
 
     clock[0] += 31
@@ -598,3 +605,68 @@ def test_fill_poll_transport_failure_uses_backoff(monkeypatch):
     )
     assert again.state == "unknown"
     assert len(calls) == 2
+
+
+def test_unresolved_fill_poll_uses_elapsed_bounded_backoff_and_converges(monkeypatch):
+    wall_now = datetime(2026, 9, 10, 12, 0, tzinfo=UTC)
+    b = KISBroker(
+        "12345678-01",
+        app_key="k",
+        app_secret="s",
+        base_url=_MOCK_URL,
+        clock=lambda: wall_now,
+    )
+    monkeypatch.setattr(b, "_headers", lambda _tr: {})
+    mono = [100.0]
+    monkeypatch.setattr("trex_core.adapters.broker.kis.monotonic", lambda: mono[0])
+    calls = []
+    terminal = [False]
+
+    class Response:
+        status_code = 200
+
+        def json(self):
+            calls.append(1)
+            parent = {
+                "odno": "1234", "orgn_odno": "", "ord_qty": "10",
+                "tot_ccld_qty": "0", "avg_prvs": "0", "cncl_yn": "",
+                "rjct_qty": "0", "rmn_qty": "0", "ord_dt": "20260910",
+                "ord_gno_brno": "91258", "pdno": "005930",
+            }
+            child = {
+                **parent, "odno": "1235", "orgn_odno": "1234",
+                "cnc_cfrm_qty": "10" if terminal[0] else "0",
+            }
+            return {"rt_cd": "0", "output1": [parent, child]}
+
+    monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response())
+    submitted_at = wall_now - timedelta(hours=7)
+
+    first = b.poll_fill(
+        "91258:1234", side="buy", symbol="005930",
+        order_date="20260910", submitted_at=submitted_at,
+    )
+    assert first.state == "unknown"
+    assert first.raw["poll_performed"] is True
+    assert first.raw["poll_count"] == 1
+    assert first.raw["backoff_sec"] == 300.0
+    assert first.raw["stuck_duration_sec"] == 7 * 3600
+
+    terminal[0] = True
+    mono[0] += 299
+    skipped = b.poll_fill(
+        "91258:1234", side="buy", symbol="005930",
+        order_date="20260910", submitted_at=submitted_at,
+    )
+    assert skipped.state == "accepted"
+    assert skipped.raw["last_result_state"] == "unknown"
+    assert len(calls) == 1
+
+    mono[0] += 1
+    resolved = b.poll_fill(
+        "91258:1234", side="buy", symbol="005930",
+        order_date="20260910", submitted_at=submitted_at,
+    )
+    assert resolved.state == "cancelled"
+    assert resolved.raw["poll_count"] == 2
+    assert len(calls) == 2

tests/adapters/broker/test_kis_cancel.py

@@ -222,6 +222,209 @@ def test_correction_child_row_not_misclassified_as_cancel(monkeypatch):
     assert not raw.get("order_open_confirmed", False)
 
 
+def test_correction_child_cancel_confirmed_quantity_maps_to_cancelled(monkeypatch):
+    """KIS cancel-confirm quantity is terminal evidence even when cncl_yn is blank."""
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="31", symbol="053050")
+    parent, child = payload["output1"]
+    child.update(cncl_yn="", cnc_cfrm_qty="31")
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+
+    assert fs.state == "cancelled"
+    assert fs.filled_qty == 0
+    assert fs.raw["reason"] == "replacement_chain_terminal_cancel"
+    assert "resolved_broker_order_id" not in fs.raw
+
+
+def test_live_replacement_remains_fail_closed(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458")
+    _, child = payload["output1"]
+    child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="10", ord_qty="10")
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "unknown"
+    assert fs.raw["reason"] == "replacement_live_unresolved"
+    assert not fs.raw.get("order_open_confirmed", False)
+
+
+def test_replacement_partial_fill_then_confirmed_cancel(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10")
+    _, child = payload["output1"]
+    child.update(
+        cncl_yn="",
+        cnc_cfrm_qty="6",
+        tot_ccld_qty="4",
+        rmn_qty="0",
+        avg_prvs="70100",
+    )
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "partially_filled"
+    assert fs.filled_qty == 4
+    assert fs.avg_price == 70100.0
+    assert fs.is_partial_cancel
+    assert fs.raw["reason"] == "replacement_chain_terminal_cancel"
+
+
+def test_repeated_cumulative_fill_across_generations_is_not_ambiguous(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10")
+    parent, child = payload["output1"]
+    parent.update(tot_ccld_qty="4", rmn_qty="0", avg_prvs="70100")
+    child.update(
+        cncl_yn="",
+        cnc_cfrm_qty="6",
+        tot_ccld_qty="4",
+        rmn_qty="0",
+        avg_prvs="70100",
+    )
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "partially_filled"
+    assert fs.filled_qty == 4
+    assert fs.avg_price == 70100.0
+    assert fs.is_partial_cancel
+
+
+def test_multi_generation_replacement_chain_terminal_cancel(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10")
+    parent, replacement = payload["output1"]
+    replacement.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0")
+    terminal = {
+        **replacement,
+        "odno": "0000123459",
+        "orgn_odno": replacement["odno"],
+        "cnc_cfrm_qty": "10",
+    }
+    payload["output1"] = [terminal, parent, replacement]
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "cancelled"
+    assert fs.raw["replacement_depth"] == 2
+
+
+def test_replacement_chain_branching_is_ambiguous(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10")
+    parent, child = payload["output1"]
+    child.update(cncl_yn="", cnc_cfrm_qty="10")
+    sibling = {**child, "odno": "0000123459"}
+    payload["output1"] = [parent, child, sibling]
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "unknown"
+    assert fs.raw["reason"] == "replacement_chain_ambiguous"
+
+
+def test_zero_remaining_without_cancel_confirmation_remains_unknown(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10")
+    _, child = payload["output1"]
+    child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0")
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "unknown"
+    assert fs.raw["reason"] == "correction_child_present_unresolved"
+
+
+def test_stable_zero_fill_replacement_chain_converges_after_grace(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="31", symbol="053050")
+    parent, child = payload["output1"]
+    parent.update(rmn_qty="0")
+    child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0")
+    monotonic_now = [100.0]
+    monkeypatch.setattr(kis_mod, "monotonic", lambda: monotonic_now[0])
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+    broker = _broker()
+
+    first = broker.poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+    assert first.state == "unknown"
+    assert first.raw["reason"] == "correction_child_present_unresolved"
+
+    monotonic_now[0] += 90
+    confirmed = broker.poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+    assert confirmed.state == "cancelled"
+    assert confirmed.raw["reason"] == "replacement_chain_stable_zero_terminal"
+
+
+def test_zero_fill_terminal_grace_resets_when_chain_changes(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="31", symbol="053050")
+    parent, child = payload["output1"]
+    parent.update(rmn_qty="0")
+    child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0")
+    monotonic_now = [100.0]
+    monkeypatch.setattr(kis_mod, "monotonic", lambda: monotonic_now[0])
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+    broker = _broker()
+
+    first = broker.poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+    assert first.state == "unknown"
+
+    monotonic_now[0] += 90
+    child["rmn_qty"] = "31"
+    changed = broker.poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+    assert changed.raw["reason"] == "replacement_live_unresolved"
+
+    monotonic_now[0] += 90
+    child["rmn_qty"] = "0"
+    restarted_grace = broker.poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+    assert restarted_grace.state == "unknown"
+
+    monotonic_now[0] += 90
+    confirmed = broker.poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901"
+    )
+    assert confirmed.state == "cancelled"
+
+
+def test_incomplete_cancel_confirmation_remains_unknown(monkeypatch):
+    payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10")
+    _, child = payload["output1"]
+    child.update(cncl_yn="", cnc_cfrm_qty="3", rmn_qty="0")
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload))
+
+    fs = _broker().poll_fill(
+        f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901"
+    )
+
+    assert fs.state == "unknown"
+    assert fs.raw["reason"] == "correction_child_present_unresolved"
+
+
 def test_partial_fill_with_child_cancel_row(monkeypatch):
     """Partial fill on parent + cancel child → partial_cancel."""
     parent = {

tests/application/test_execution_cancel.py

@@ -104,6 +104,58 @@ def test_full_external_cancel_returns_slot_to_holding():
     assert r.rolled_back == 1
 
 
+def test_unresolved_poll_observability_is_sanitized_and_deduplicated(caplog):
+    broker = ProgrammableBroker()
+    repo, _svc, intents, execution, _ledger, _writer = _wire(broker)
+    sid = _buy_to_holding(intents, execution)
+    assert sid is not None
+    intents.submit_sell(SellIntent(intent_id="s-poll-observe", slot_id=sid))
+    samples = iter([
+        ("correction_child_present_unresolved", 10),
+        ("correction_child_present_unresolved", 10),
+        ("correction_child_present_unresolved", 61),
+        ("correction_child_present_unresolved", 61),
+        ("parent_identity_mismatch", 61),
+    ])
+
+    def unresolved(side):
+        del side
+        reason, age = next(samples)
+        return FillStatus(
+            state="unknown",
+            raw={
+                "poll_performed": True,
+                "poll_count": broker.polls,
+                "stuck_duration_sec": age,
+                "last_result_reason": reason,
+                "backoff_sec": 15.0,
+                "raw_payload": {"authorization": "secret-token"},
+                "broker_order_id": "sensitive-order-id",
+            },
+        )
+
+    broker.responder = unresolved
+    with caplog.at_level(logging.INFO):
+        for _ in range(5):
+            execution.run_once()
+
+    observations = [
+        record for record in caplog.records
+        if getattr(record, "event", None) == "order_poll"
+    ]
+    warnings = [
+        record for record in caplog.records
+        if getattr(record, "event", None) == "order_poll_unresolved"
+    ]
+    assert len(observations) == 5
+    assert len(warnings) == 3  # first, long-duration threshold, reason change
+    assert observations[-1].poll_count == 6
+    assert observations[-1].stuck_duration_sec == 61
+    assert observations[-1].result_reason == "parent_identity_mismatch"
+    assert "secret-token" not in caplog.text
+    assert "sensitive-order-id" not in caplog.text
+
+
 def test_partial_fill_then_cancel_records_and_keeps_remainder():
     broker = ProgrammableBroker()
     repo, svc, intents, execution, ledger, writer = _wire(broker)

tests/integration/test_kis_submission_recovery_virtual.py

@@ -12,7 +12,7 @@ from zoneinfo import ZoneInfo
 from trex_core.adapters.broker.kis import KISBroker
 from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
 from trex_core.application.execution import ExecutionService
-from trex_core.application.intents import BuyIntent, IntentService
+from trex_core.application.intents import BuyIntent, IntentService, SellIntent
 from trex_core.domain.slot import SlotService, SlotState
 from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
 
@@ -30,12 +30,15 @@ class _ExchangeState:
         self.orders: list[dict[str, str]] = []
         self.fill_orders = False
         self.reject_orders = False
+        self.next_side = "02"
+        self.correction_cancel_sell = False
+        self.balance_qty: int | None = None
 
     def submit(self, body: dict[str, str]) -> tuple[bool, dict[str, object]]:
         self.submit_attempts += 1
         order = {
             "pdno": body["PDNO"],
-            "sll_buy_dvsn_cd": "02",
+            "sll_buy_dvsn_cd": self.next_side,
             "ord_qty": body["ORD_QTY"],
             "ord_unpr": body["ORD_UNPR"],
             "ord_dt": VIRTUAL_AT.strftime("%Y%m%d"),
@@ -65,6 +68,20 @@ class _ExchangeState:
                 order["rjct_qty"] = order["ord_qty"]
                 order["rmn_qty"] = "0"
             row = dict(order)
+            if self.correction_cancel_sell and order["sll_buy_dvsn_cd"] == "01":
+                row["rmn_qty"] = "0"
+                rows.append(row)
+                rows.append(
+                    {
+                        **row,
+                        "odno": f"{int(order['odno']) + 1:010d}",
+                        "orgn_odno": order["odno"],
+                        "cncl_yn": "",
+                        "cnc_cfrm_qty": "0",
+                        "rmn_qty": "0",
+                    }
+                )
+                continue
             if self.fill_orders and order["rjct_qty"] == "0":
                 row["tot_ccld_qty"] = row["ord_qty"]
                 row["avg_prvs"] = row["ord_unpr"]
@@ -72,6 +89,8 @@ class _ExchangeState:
         return {"rt_cd": "0", "output1": rows}
 
     def balance(self) -> dict[str, object]:
+        quantity = self.balance_qty
+        if quantity is None:
             quantity = (
                 sum(int(order["ord_qty"]) for order in self.orders if order["rjct_qty"] == "0")
                 if self.fill_orders
@@ -265,3 +284,54 @@ def test_exchange_reject_after_broker_ack_rolls_back_and_allows_one_retry():
     finally:
         server.shutdown()
         server.server_close()
+
+
+def test_stable_correction_chain_returns_pending_sell_to_holding(monkeypatch):
+    state = _ExchangeState(first_attempt_reaches_exchange=True)
+    server = _serve(state)
+    mono = [100.0]
+    monkeypatch.setattr("trex_core.adapters.broker.kis.monotonic", lambda: mono[0])
+    try:
+        repo, slots, intents, execution, slot_id = _wire(_broker(server))
+        assert slot_id is not None
+
+        execution.run_once()
+        state.fill_orders = True
+        execution = ExecutionService(
+            slots,
+            repo,
+            _broker(server),
+            instance_key=INSTANCE,
+            clock=lambda: VIRTUAL_AT,
+        )
+        execution.reconcile()
+        bought = repo.fetch_slot(slot_id)
+        assert bought is not None
+        assert bought.status is SlotState.HOLDING
+
+        state.fill_orders = False
+        state.balance_qty = 2
+        state.next_side = "01"
+        state.correction_cancel_sell = True
+        admitted = intents.submit_sell(
+            SellIntent(intent_id="virtual-sell-corrected", slot_id=slot_id)
+        )
+        assert admitted.status == "admitted"
+
+        execution.run_once()
+        unresolved = repo.fetch_slot(slot_id)
+        assert unresolved is not None
+        assert unresolved.status is SlotState.PENDING_SELL
+        assert state.submit_attempts == 2
+
+        mono[0] += 90
+        execution.run_once()
+        holding = repo.fetch_slot(slot_id)
+        assert holding is not None
+        assert holding.status is SlotState.HOLDING
+        assert holding.qty == 2
+        assert holding.broker_order_id is None
+        assert state.submit_attempts == 2
+    finally:
+        server.shutdown()
+        server.server_close()

trex_core/adapters/broker/kis.py

@@ -19,6 +19,7 @@ from dataclasses import replace
 from datetime import UTC, datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from time import monotonic
+from typing import Any
 
 import httpx
 
@@ -39,6 +40,7 @@ _SUBMISSION_UNKNOWN_PREFIX = "submission-unknown:"
 _MOCK_HINTS = ("openapivts", "virtual", "mock")
 _FILL_POLL_MIN_INTERVAL_SECONDS = 5.0
 _FILL_POLL_TRANSPORT_BACKOFF_SECONDS = 30.0
+_FILL_POLL_MAX_BACKOFF_SECONDS = 300.0
 _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0
 _SUPPORTED_ORDER_VENUES = frozenset({"KRX", "NXT", "SOR"})
 
@@ -150,6 +152,7 @@ class KISBroker:
         self._token_cache = token_cache  # optional TokenCache (L2, REQ-016)
         self._clock = clock or (lambda: datetime.now(tz=_KST))
         self._fill_poll_not_before: dict[str, float] = {}
+        self._fill_poll_stats: dict[str, dict[str, Any]] = {}
         is_mock = any(h in self._base_url for h in _MOCK_HINTS)
         if not is_mock and not allow_live:
             raise RuntimeError(
@@ -358,6 +361,118 @@ class KISBroker:
             "EXCG_ID_DVSN_CD": "ALL",
         }
 
+    def _fill_poll_age_seconds(self, submitted_at: datetime | None) -> int:
+        if submitted_at is None:
+            return 0
+        return max(
+            0,
+            int(
+                (
+                    self._clock().astimezone(UTC)
+                    - submitted_at.astimezone(UTC)
+                ).total_seconds()
+            ),
+        )
+
+    @staticmethod
+    def _fill_poll_backoff_seconds(
+        status: FillStatus,
+        *,
+        age_seconds: int,
+        transport_failure: bool,
+    ) -> float:
+        """Return bounded state/age-based delay for the next authoritative poll."""
+        if status.state != "unknown":
+            delay = _FILL_POLL_MIN_INTERVAL_SECONDS
+        elif age_seconds < 60:
+            delay = _FILL_POLL_MIN_INTERVAL_SECONDS
+        elif age_seconds < 5 * 60:
+            delay = 15.0
+        elif age_seconds < 60 * 60:
+            delay = 30.0
+        elif age_seconds < 6 * 60 * 60:
+            delay = 120.0
+        else:
+            delay = _FILL_POLL_MAX_BACKOFF_SECONDS
+        if transport_failure:
+            delay = max(delay, _FILL_POLL_TRANSPORT_BACKOFF_SECONDS)
+        return min(delay, _FILL_POLL_MAX_BACKOFF_SECONDS)
+
+    def _record_fill_poll(
+        self,
+        broker_order_id: str,
+        status: FillStatus,
+        *,
+        submitted_at: datetime | None,
+        transport_failure: bool = False,
+    ) -> FillStatus:
+        """Attach allowlisted poll telemetry and schedule the next broker read."""
+        age_seconds = self._fill_poll_age_seconds(submitted_at)
+        backoff_seconds = self._fill_poll_backoff_seconds(
+            status,
+            age_seconds=age_seconds,
+            transport_failure=transport_failure,
+        )
+        raw = status.raw if isinstance(status.raw, dict) else {}
+        reason = raw.get("reason") if isinstance(raw.get("reason"), str) else status.state
+        previous = self._fill_poll_stats.get(broker_order_id, {})
+        if reason != "correction_child_present_unresolved":
+            previous.pop("zero_fill_chain_fingerprint", None)
+            previous.pop("zero_fill_chain_first_seen", None)
+        poll_count = int(previous.get("poll_count", 0)) + 1
+        self._fill_poll_not_before[broker_order_id] = monotonic() + backoff_seconds
+        self._fill_poll_stats[broker_order_id] = {
+            **previous,
+            "poll_count": poll_count,
+            "last_result_state": status.state,
+            "last_result_reason": reason,
+            "backoff_sec": backoff_seconds,
+        }
+        return replace(
+            status,
+            raw={
+                **raw,
+                "poll_performed": True,
+                "poll_count": poll_count,
+                "stuck_duration_sec": age_seconds,
+                "last_result_state": status.state,
+                "last_result_reason": reason,
+                "backoff_sec": backoff_seconds,
+            },
+        )
+
+    def _replacement_zero_fill_terminal_confirmed(
+        self,
+        broker_order_id: str,
+        chain: list[dict],
+    ) -> bool:
+        """Require a stable grace-period snapshot before terminal inference."""
+        fingerprint = tuple(
+            (
+                str(row.get("odno") or "").strip(),
+                str(row.get("orgn_odno") or "").strip(),
+                _safe_int(row.get("ord_qty")),
+                _safe_int(row.get("tot_ccld_qty")),
+                _safe_int(row.get("rmn_qty")),
+            )
+            for row in chain
+        )
+        stats = self._fill_poll_stats.setdefault(broker_order_id, {})
+        now = monotonic()
+        if stats.get("zero_fill_chain_fingerprint") != fingerprint:
+            stats["zero_fill_chain_fingerprint"] = fingerprint
+            stats["zero_fill_chain_first_seen"] = now
+            return False
+        first_seen = float(stats.get("zero_fill_chain_first_seen", now))
+        return now - first_seen >= _SUBMISSION_LOOKUP_GRACE_SECONDS
+
+    def _clear_replacement_zero_fill_candidate(self, broker_order_id: str) -> None:
+        stats = self._fill_poll_stats.get(broker_order_id)
+        if stats is None:
+            return
+        stats.pop("zero_fill_chain_fingerprint", None)
+        stats.pop("zero_fill_chain_first_seen", None)
+
     def poll_fill(
         self,
         broker_order_id: str,
@@ -379,9 +494,18 @@ class KISBroker:
             )
         now = monotonic()
         if now < self._fill_poll_not_before.get(broker_order_id, 0.0):
+            stats = self._fill_poll_stats.get(broker_order_id, {})
             return FillStatus(
                 state="accepted",
-                raw={"reason": "fill_poll_backoff"},
+                raw={
+                    "reason": "fill_poll_backoff",
+                    "poll_performed": False,
+                    "poll_count": int(stats.get("poll_count", 0)),
+                    "stuck_duration_sec": self._fill_poll_age_seconds(submitted_at),
+                    "last_result_state": stats.get("last_result_state"),
+                    "last_result_reason": stats.get("last_result_reason"),
+                    "backoff_sec": float(stats.get("backoff_sec", 0.0)),
+                },
             )
         # Set the normal gate before I/O so concurrent execution ticks cannot
         # open duplicate KIS inquiries for the same order.
@@ -397,20 +521,20 @@ class KISBroker:
                 timeout=self._timeout,
             )
             payload = resp.json()
-        except (httpx.HTTPError, ValueError) as exc:
-            self._fill_poll_not_before[broker_order_id] = (
-                monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
+        except (httpx.HTTPError, ValueError):
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(state="unknown", raw={"reason": "fill_inquiry_transport"}),
+                submitted_at=submitted_at,
+                transport_failure=True,
             )
-            return FillStatus(state="unknown", raw={"error": str(exc)})
         if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0":
-            self._fill_poll_not_before[broker_order_id] = (
-                monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(state="unknown", raw={"reason": "fill_inquiry_rejected"}),
+                submitted_at=submitted_at,
+                transport_failure=True,
             )
-            return FillStatus(
-                state="unknown",
-                raw={"reason": "fill_inquiry_rejected"},
-            )
-        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"
         # Canonical request identity from the caller — these are authoritative
@@ -454,13 +578,17 @@ class KISBroker:
             # 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(
+                return self._record_fill_poll(
+                    broker_order_id,
+                    FillStatus(
                         state="unknown",
                         raw={
                             "reason": "parent_identity_mismatch",
                             "order_visible": True,
                             "inquiry_date": inquiry_date,
                         },
+                    ),
+                    submitted_at=submitted_at,
                 )
             # Truly invisible — apply the submission visibility grace window.
             visible_grace = (
@@ -469,7 +597,9 @@ class KISBroker:
                 .total_seconds()
                 < _SUBMISSION_LOOKUP_GRACE_SECONDS
             )
-            return FillStatus(
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(
                     state="accepted" if visible_grace else "unknown",
                     raw={
                         "reason": (
@@ -480,107 +610,197 @@ class KISBroker:
                         "order_visible": False,
                         "inquiry_date": inquiry_date,
                     },
+                ),
+                submitted_at=submitted_at,
             )
 
         if len(exact_parent_matches) > 1:
             # More than one row claims the same exact identity — ambiguous.
-            return FillStatus(
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(
                     state="unknown",
                     raw={
                         "reason": "parent_identity_ambiguous",
                         "order_visible": True,
                         "inquiry_date": inquiry_date,
                     },
+                ),
+                submitted_at=submitted_at,
             )
 
         row = exact_parent_matches[0]
 
-        # 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.
-        #
-        # 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
+        # Follow only response rows bound to the immutable request identity.
+        # KIS may represent cancel/modify as a chain where each child points to
+        # its immediate predecessor through orgn_odno. Row order and numerical
+        # proximity are never evidence of parentage.
+        identity_rows = [r for r in rows if _row_matches_request_identity(r)]
 
-        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)
+        def _row_parent_odno_normalized(r: dict) -> str:
+            return str(r.get("orgn_odno") or "").strip().lstrip("0") or "0"
+
+        chain = [row]
+        visited = {normalized_odno}
+        current_odno = normalized_odno
+        ambiguous_reason: str | None = None
+        while True:
+            children = [
+                candidate
+                for candidate in identity_rows
+                if _row_parent_odno_normalized(candidate) == current_odno
             ]
+            if not children:
+                break
+            if len(children) > 1:
+                ambiguous_reason = (
+                    "child_cancel_ambiguous"
+                    if all(
+                        str(child.get("cncl_yn") or "").strip().upper() == "Y"
+                        for child in children
+                    )
+                    else "replacement_chain_ambiguous"
+                )
+                break
+            child = children[0]
+            child_odno = _row_odno_normalized(child)
+            if child_odno == "0" or child_odno in visited:
+                ambiguous_reason = "replacement_chain_ambiguous"
+                break
+            chain.append(child)
+            visited.add(child_odno)
+            current_odno = child_odno
 
-        # 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(
+        if ambiguous_reason is not None:
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(
                     state="unknown",
                     raw={
-                    "reason": "child_cancel_ambiguous",
+                        "reason": ambiguous_reason,
                         "order_visible": True,
                         "inquiry_date": inquiry_date,
                     },
+                ),
+                submitted_at=submitted_at,
             )
 
-        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(
+        terminal_row = chain[-1]
+        terminal_order_qty = _safe_int(terminal_row.get("ord_qty")) or 0
+        terminal_filled_qty = _safe_int(terminal_row.get("tot_ccld_qty")) or 0
+        terminal_cancelled_qty = _safe_int(terminal_row.get("cnc_cfrm_qty")) or 0
+        terminal_cancel = (
+            str(terminal_row.get("cncl_yn") or "").strip().upper() == "Y"
+            or (
+                terminal_order_qty > 0
+                and terminal_cancelled_qty > 0
+                and terminal_cancelled_qty + terminal_filled_qty >= terminal_order_qty
+            )
+        )
+        terminal_remaining = _safe_int(terminal_row.get("rmn_qty"))
+        positive_fill_rows = [
+            candidate
+            for candidate in chain
+            if (_safe_int(candidate.get("tot_ccld_qty")) or 0) > 0
+        ]
+        positive_fill_signatures = {
+            (
+                _safe_int(candidate.get("tot_ccld_qty")),
+                str(candidate.get("avg_prvs") or "").strip(),
+            )
+            for candidate in positive_fill_rows
+        }
+        repeated_cumulative_fill = (
+            len(positive_fill_rows) > 1 and len(positive_fill_signatures) == 1
+        )
+        stable_zero_fill_candidate = (
+            len(chain) > 1
+            and not terminal_cancel
+            and not positive_fill_rows
+            and all((_safe_int(candidate.get("ord_qty")) or 0) > 0 for candidate in chain)
+            and all(_safe_int(candidate.get("rmn_qty")) == 0 for candidate in chain)
+            and all((_safe_int(candidate.get("rjct_qty")) or 0) == 0 for candidate in chain)
+        )
+        stable_zero_fill_terminal = False
+        if stable_zero_fill_candidate:
+            stable_zero_fill_terminal = self._replacement_zero_fill_terminal_confirmed(
+                broker_order_id,
+                chain,
+            )
+        else:
+            self._clear_replacement_zero_fill_candidate(broker_order_id)
+
+        if len(positive_fill_rows) > 1 and not repeated_cumulative_fill:
+            status = FillStatus(
                 state="unknown",
+                raw={"reason": "replacement_chain_fill_ambiguous"},
+            )
+        elif (terminal_cancel or stable_zero_fill_terminal) and terminal_remaining in {None, 0}:
+            # A child cancel row often carries no fills of its own. Use the one
+            # unambiguous filled generation when present; otherwise preserve the
+            # original quantity context and inject only the terminal cancel bit.
+            evidence_row = positive_fill_rows[-1] if positive_fill_rows else row
+            status = self._fill_status_from_row(
+                {**evidence_row, "cncl_yn": "Y"},
+                side=side,
+                today=inquiry_date,
+                fallback_odno=odno,
+            )
+            status = replace(
+                status,
                 raw={
-                    "reason": "correction_child_present_unresolved",
-                    "order_visible": True,
-                    "inquiry_date": inquiry_date,
+                    **(status.raw if isinstance(status.raw, dict) else {}),
+                    "reason": (
+                        "replacement_chain_terminal_cancel"
+                        if terminal_cancel
+                        else "replacement_chain_stable_zero_terminal"
+                    ),
+                    "replacement_depth": len(chain) - 1,
                 },
             )
-
-        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 "")}
-
+        elif len(chain) > 1:
+            leaf_status = self._fill_status_from_row(
+                terminal_row,
+                side=side,
+                today=inquiry_date,
+                fallback_odno=odno,
+            )
+            if leaf_status.is_filled:
+                status = leaf_status
+            else:
+                status = FillStatus(
+                    state="unknown",
+                    raw={
+                        "reason": (
+                            "replacement_live_unresolved"
+                            if (terminal_remaining or 0) > 0
+                            else "correction_child_present_unresolved"
+                        ),
+                        "replacement_depth": len(chain) - 1,
+                    },
+                )
+        else:
             status = self._fill_status_from_row(
                 row, side=side, today=inquiry_date, fallback_odno=odno
             )
-        return replace(
+
+        status = replace(
             status,
             raw={
                 **(status.raw if isinstance(status.raw, dict) else {}),
                 "order_visible": True,
-                "order_open_confirmed": status.state in {"accepted", "partially_filled"},
+                "order_open_confirmed": (
+                    len(chain) == 1
+                    and status.state in {"accepted", "partially_filled"}
+                ),
                 "inquiry_date": inquiry_date,
             },
         )
+        return self._record_fill_poll(
+            broker_order_id,
+            status,
+            submitted_at=submitted_at,
+        )
 
     def _poll_unknown_submission(
         self,
@@ -595,16 +815,33 @@ class KISBroker:
                 state="unknown",
                 raw={"reason": "submission_unknown_metadata_invalid"},
             )
+        try:
+            submitted_at = datetime.fromisoformat(str(metadata["submitted_at"])).astimezone(
+                _KST
+            )
+        except (KeyError, TypeError, ValueError):
+            return FillStatus(
+                state="unknown",
+                raw={"reason": "submission_unknown_metadata_invalid"},
+            )
         now_monotonic = monotonic()
         if now_monotonic < self._fill_poll_not_before.get(broker_order_id, 0.0):
+            stats = self._fill_poll_stats.get(broker_order_id, {})
             return FillStatus(
                 state="accepted",
-                raw={"reason": "submission_lookup_backoff"},
+                raw={
+                    "reason": "submission_lookup_backoff",
+                    "poll_performed": False,
+                    "poll_count": int(stats.get("poll_count", 0)),
+                    "stuck_duration_sec": self._fill_poll_age_seconds(submitted_at),
+                    "last_result_state": stats.get("last_result_state"),
+                    "last_result_reason": stats.get("last_result_reason"),
+                    "backoff_sec": float(stats.get("backoff_sec", 0.0)),
+                },
             )
         self._fill_poll_not_before[broker_order_id] = (
             now_monotonic + _FILL_POLL_MIN_INTERVAL_SECONDS
         )
-        submitted_at = datetime.fromisoformat(str(metadata["submitted_at"])).astimezone(_KST)
         today = submitted_at.strftime("%Y%m%d")
         tr_id, params = self._daily_fill_request(
             today,
@@ -619,20 +856,18 @@ class KISBroker:
             )
             payload = resp.json()
         except (httpx.HTTPError, ValueError):
-            self._fill_poll_not_before[broker_order_id] = (
-                monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
-            )
-            return FillStatus(
-                state="unknown",
-                raw={"reason": "submission_lookup_transport"},
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(state="unknown", raw={"reason": "submission_lookup_transport"}),
+                submitted_at=submitted_at,
+                transport_failure=True,
             )
         if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0":
-            self._fill_poll_not_before[broker_order_id] = (
-                monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
-            )
-            return FillStatus(
-                state="unknown",
-                raw={"reason": "submission_lookup_rejected"},
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(state="unknown", raw={"reason": "submission_lookup_rejected"}),
+                submitted_at=submitted_at,
+                transport_failure=True,
             )
         matches = [
             row
@@ -640,19 +875,29 @@ class KISBroker:
             if self._matches_unknown_submission(row, metadata, submitted_at)
         ]
         if len(matches) != 1:
-            age = (datetime.now(tz=_KST) - submitted_at).total_seconds()
+            age = (self._clock().astimezone(_KST) - submitted_at).total_seconds()
             if not matches and age >= _SUBMISSION_LOOKUP_GRACE_SECONDS:
-                return FillStatus(
+                return self._record_fill_poll(
+                    broker_order_id,
+                    FillStatus(
                         state="rejected",
                         raw={"reason": "submission_confirmed_absent"},
+                    ),
+                    submitted_at=submitted_at,
                 )
-            return FillStatus(
+            return self._record_fill_poll(
+                broker_order_id,
+                FillStatus(
                     state="unknown",
                     raw={
                         "reason": (
-                        "submission_lookup_ambiguous" if matches else "submission_lookup_pending"
+                            "submission_lookup_ambiguous"
+                            if matches
+                            else "submission_lookup_pending"
                         )
                     },
+                ),
+                submitted_at=submitted_at,
             )
         row = matches[0]
         resolved_order_id = self._broker_order_id_from_row(row)
@@ -662,12 +907,16 @@ class KISBroker:
             today=today,
             fallback_odno=resolved_order_id.split(":")[-1],
         )
-        return replace(
+        return self._record_fill_poll(
+            broker_order_id,
+            replace(
                 status,
                 raw={
                     **(status.raw if isinstance(status.raw, dict) else {}),
                     "resolved_broker_order_id": resolved_order_id,
                 },
+            ),
+            submitted_at=submitted_at,
         )
 
     @staticmethod

trex_core/application/execution.py

@@ -62,6 +62,24 @@ SlotEventEmitter = Callable[
 ]  # (event_type, slot, routing_strategy_id) -> None
 
 _MAX_DISPATCH_PER_TICK = 64
+_ORDER_POLL_WARNING_THRESHOLDS_SEC = (60, 5 * 60, 15 * 60, 60 * 60, 6 * 60 * 60)
+_ORDER_POLL_REASON_ALLOWLIST = frozenset({
+    "correction_child_present_unresolved",
+    "fill_inquiry_rejected",
+    "fill_inquiry_transport",
+    "order_not_visible_unresolved",
+    "parent_identity_ambiguous",
+    "parent_identity_mismatch",
+    "replacement_chain_ambiguous",
+    "replacement_chain_fill_ambiguous",
+    "replacement_chain_stable_zero_terminal",
+    "replacement_chain_terminal_cancel",
+    "replacement_live_unresolved",
+    "submission_lookup_ambiguous",
+    "submission_lookup_pending",
+    "submission_lookup_rejected",
+    "submission_lookup_transport",
+})
 
 
 @dataclass
@@ -115,6 +133,7 @@ class ExecutionService:
         # A restart intentionally emits once again, covering a crash between a
         # durable cancel reservation and best-effort alarm publication.
         self._order_alarm_keys: set[tuple[str, str, int]] = set()
+        self._order_poll_warning_state: dict[str, tuple[str, str, int]] = {}
         self._clock = clock or (lambda: datetime.now(tz=UTC))
         # dcy-041 / GitHub #45: pending-sell quantity may use a bounded
         # account snapshot only through this injected reader. Reconciliation
@@ -282,6 +301,7 @@ class ExecutionService:
                 self._emit_order_metadata_unknown_alarm(slot)
                 continue
             fs = self._poll_fill(slot, side)
+            self._observe_order_poll(slot, side, fs)
             resolved_order_id = (
                 fs.raw.get("resolved_broker_order_id") if isinstance(fs.raw, dict) else None
             )
@@ -344,6 +364,55 @@ class ExecutionService:
             kwargs["submitted_at"] = slot.order_submitted_at
         return poll(slot.broker_order_id, **kwargs)
 
+    def _observe_order_poll(self, slot, side: str, fs) -> None:
+        """Emit sanitized poll telemetry and deduplicated unresolved warnings."""
+        raw = fs.raw if isinstance(fs.raw, dict) else {}
+        if raw.get("poll_performed") is not True:
+            return
+
+        def _metric_int(name: str) -> int:
+            try:
+                return max(0, int(raw.get(name, 0)))
+            except (TypeError, ValueError):
+                return 0
+
+        reason_value = raw.get("last_result_reason")
+        reason = (
+            str(reason_value)
+            if reason_value in _ORDER_POLL_REASON_ALLOWLIST
+            else "none"
+        )
+        poll_count = _metric_int("poll_count")
+        stuck_duration_sec = _metric_int("stuck_duration_sec")
+        try:
+            backoff_sec = max(0.0, float(raw.get("backoff_sec", 0.0)))
+        except (TypeError, ValueError):
+            backoff_sec = 0.0
+        fields = {
+            "event": "order_poll",
+            "side": side,
+            "result_state": fs.state,
+            "result_reason": reason,
+            "poll_count": poll_count,
+            "stuck_duration_sec": stuck_duration_sec,
+            "backoff_sec": backoff_sec,
+        }
+        _log.info("order poll observed", extra=fields)
+
+        slot_key = str(slot.slot_id)
+        if fs.state != "unknown":
+            self._order_poll_warning_state.pop(slot_key, None)
+            return
+        threshold_index = sum(
+            stuck_duration_sec >= threshold
+            for threshold in _ORDER_POLL_WARNING_THRESHOLDS_SEC
+        )
+        warning_key = (fs.state, reason, threshold_index)
+        if self._order_poll_warning_state.get(slot_key) == warning_key:
+            return
+        self._order_poll_warning_state[slot_key] = warning_key
+        _log.warning("order poll unresolved", extra={**fields, "event": "order_poll_unresolved"})
+
     # ------------------------------------------------------------
     def _balance_map(self) -> dict:
         """Current broker balance as {symbol: hldg_qty} (REQ-062 ground truth)."""