feat(execution): reconcile timed-out orders safely

tests/adapters/broker/test_ownership.py

@@ -25,6 +25,7 @@ class FakeBroker:
     def __init__(self, balance, fill=None):
         self.balance = balance
         self.fill = fill or FillStatus(state="accepted")
+        self.poll_metadata = None
 
     def name(self):
         return "fake"
@@ -32,7 +33,16 @@ class FakeBroker:
     def submit_order(self, req):
         return OrderAck(accepted=True, broker_order_id="1")
 
-    def poll_fill(self, broker_order_id, *, side, symbol):
+    def poll_fill(
+        self,
+        broker_order_id,
+        *,
+        side,
+        symbol,
+        order_date=None,
+        submitted_at=None,
+    ):
+        self.poll_metadata = (order_date, submitted_at)
         return self.fill
 
     def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"):
@@ -77,6 +87,23 @@ def test_confirmed_fill_is_persisted_once_and_manual_remainder_stays_hidden(tmp_
     assert StrategyOwnershipStore(store.path).snapshot()[0] == {"005930": 2}
 
 
+def test_poll_preserves_original_order_metadata_through_ownership_decorator(tmp_path):
+    store = StrategyOwnershipStore(tmp_path / "ownership.json")
+    delegate = FakeBroker([])
+    broker = OwnershipAwareBroker(delegate, store)
+    submitted_at = datetime(2026, 8, 14, 6, 20, tzinfo=timezone.utc)
+
+    broker.poll_fill(
+        "1",
+        side="buy",
+        symbol="005930",
+        order_date="20260814",
+        submitted_at=submitted_at,
+    )
+
+    assert delegate.poll_metadata == ("20260814", submitted_at)
+
+
 def test_strategy_quantity_above_raw_broker_total_fails_closed(tmp_path):
     store = StrategyOwnershipStore(tmp_path / "ownership.json")
     store.apply_fill(

tests/adapters/persistence/test_postgresql_slot_repository_parity.py

@@ -36,7 +36,7 @@ _SLOT_COLUMNS = (
     "sell_strategy_id, forced_close_at, stop_loss_price, take_profit_price, "
     "created_at, updated_at, instance_key, retired_at, "
     "timeout_sec, slippage_pct, occupancy_seq, venue, "
-    "cancel_state, cancel_requested_at"
+    "cancel_state, cancel_requested_at, order_submitted_at, order_date, cancel_attempts"
 )
 _MANAGED_SLOT_SQL = (
     f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
@@ -149,6 +149,9 @@ def _row(slot: Slot) -> tuple:
         slot.venue,
         slot.cancel_state,
         slot.cancel_requested_at,
+        slot.order_submitted_at,
+        slot.order_date,
+        slot.cancel_attempts,
     )
 
 

tests/application/test_execution_policy.py

@@ -290,7 +290,8 @@ class TestTimeoutCancel:
         # age the slot past timeout
         s = repo.fetch_slot(sid)
         repo._slots[sid] = replace(
-            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
+            s,
+            order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
         )
         r = execution.run_once()
         assert r.timeout_cancelled == 1
@@ -310,13 +311,13 @@ class TestTimeoutCancel:
         assert r.timeout_cancelled == 0
         assert len(broker.cancel_calls) == 0
 
-    def test_no_timeout_sec_no_cancel(self):
+    def test_updated_at_does_not_move_hard_fallback_deadline(self):
         broker = _RecordingBroker()
         repo, svc, intents, execution = _wire(broker)
         sid = _buy_to_pending(intents, timeout_sec=None)
         broker.responder = lambda side: FillStatus(state="accepted")
         execution.run_once()  # dispatch
-        # age the slot
+        # Mutable row bookkeeping is not the dispatch/deadline clock.
         s = repo.fetch_slot(sid)
         repo._slots[sid] = replace(
             s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999),
@@ -335,7 +336,8 @@ class TestTimeoutCancel:
         execution.run_once()  # dispatch
         s = repo.fetch_slot(sid)
         repo._slots[sid] = replace(
-            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
+            s,
+            order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
         )
         with caplog.at_level(logging.WARNING):
             r = execution.run_once()
@@ -357,7 +359,8 @@ class TestTimeoutCancel:
         execution.run_once()
         s = repo.fetch_slot(sid)
         repo._slots[sid] = replace(
-            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
+            s,
+            order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
         )
         execution.run_once()
         # slot still PENDING_BUY — cancel is best-effort; rollback on next poll
@@ -376,7 +379,7 @@ class TestTimeoutCancel:
         slot = repo.fetch_slot(sid)
         repo._slots[sid] = replace(
             slot,
-            updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
+            order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
         )
 
         result = execution.run_once()
@@ -395,7 +398,8 @@ class TestTimeoutCancel:
         # age the slot → cancel sent
         s = repo.fetch_slot(sid)
         repo._slots[sid] = replace(
-            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
+            s,
+            order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
         )
         r1 = execution.run_once()
         assert r1.timeout_cancelled == 1

tests/application/test_execution_policy_handler.py

@@ -75,14 +75,14 @@ class TestIntentPayloadParsing:
         assert slot.timeout_sec == 60
         assert slot.slippage_pct == 1.0
 
-    def test_buy_policy_fields_absent_means_none(self):
-        """Backward-compat: no policy fields in payload → None on slot."""
+    def test_buy_policy_fields_absent_uses_hard_fallback(self):
+        """No policy field still receives the hard 3,600-second fallback."""
         repo, _, intents = _wire()
         handler = make_intent_rpc_handler(intents)
         resp = handler(_buy_req())
         sid = UUID(resp.payload["slot_id"])
         slot = repo.fetch_slot(sid)
-        assert slot.timeout_sec is None
+        assert slot.timeout_sec == 3600
         assert slot.slippage_pct is None
 
     def test_sell_timeout_sec_parsed(self):
@@ -166,13 +166,13 @@ class TestDefaultOrderTimeoutSec:
         slot = repo.fetch_slot(d.slot_id)
         assert slot.timeout_sec == 30
 
-    def test_no_config_default_no_intent_timeout_means_none(self):
+    def test_no_config_default_no_intent_timeout_uses_hard_fallback(self):
         repo, _, intents = self._wire_with_default(default_timeout=None)
         d = intents.submit_buy(BuyIntent(
             intent_id="b1", portfolio_id="PF-1", qty=10,
         ))
         slot = repo.fetch_slot(d.slot_id)
-        assert slot.timeout_sec is None
+        assert slot.timeout_sec == 3600
 
 
 # ============================================================================

tests/application/test_order_timeout_reconciliation.py

@@ -0,0 +1,342 @@
+"""Synthetic coverage for GitHub #58 timeout/cancel reconciliation."""
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from uuid import UUID
+
+import pytest
+
+from trex_core.adapters.broker import kis as kis_mod
+from trex_core.adapters.broker.kis import KISBroker
+from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
+from trex_core.application.broker_port import CancelOutcome, CancelState, FillStatus
+from trex_core.application.execution import ExecutionService
+from trex_core.application.intents import BuyIntent, IntentService, SellIntent
+from trex_core.application.messaging import RpcRequest
+from trex_core.application.messaging_handlers import make_intent_rpc_handler
+from trex_core.domain.slot import SlotService, SlotState
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+from trex_core.runtime.config import DaemonInstanceConfig
+from tests.application.test_execution_policy import _RecordingBroker
+
+_KST = timezone(timedelta(hours=9))
+
+
+def _wire(*, default_timeout=None):
+    repo = InMemorySlotRepository()
+    service = SlotService(repo)
+    intents = IntentService(
+        service,
+        repo,
+        InMemoryIntentStore(),
+        resolve_symbol=lambda _pid: "005930",
+        default_order_timeout_sec=default_timeout,
+    )
+    repo.reconcile_universe("default", {"PF-1": "005930"})
+    return repo, service, intents
+
+
+@pytest.mark.parametrize("value", [0, -1, 3601, True, 1.5, "30"])
+def test_direct_strategy_timeout_outside_integer_bounds_is_rejected(value):
+    repo, _, intents = _wire()
+    decision = intents.submit_buy(
+        BuyIntent(intent_id=f"b-{value!r}", portfolio_id="PF-1", qty=1, timeout_sec=value)
+    )
+    assert decision.status == "rejected"
+    assert decision.reason.startswith("bad_request:timeout_sec_")
+    assert repo.list_undispatched_pending() == []
+
+
+@pytest.mark.parametrize("value", [0, -1, 3601, True, 1.5, "30"])
+def test_direct_sell_timeout_outside_integer_bounds_is_rejected(value):
+    repo, _, intents = _wire()
+    decision = intents.submit_sell(
+        SellIntent(intent_id=f"s-{value!r}", slot_id=UUID(int=1), timeout_sec=value)
+    )
+    assert decision.status == "rejected"
+    assert decision.reason.startswith("bad_request:timeout_sec_")
+    assert repo.list_undispatched_pending() == []
+
+
+@pytest.mark.parametrize("value", [0, -1, 3601, 1.5, True, "30", "not-an-int"])
+@pytest.mark.parametrize("intent_type", ["buy", "sell"])
+def test_wire_strategy_timeout_bad_request_before_admission(value, intent_type):
+    repo, _, intents = _wire()
+    payload = {
+        "intent_type": intent_type,
+        "intent_id": "bad-timeout",
+        "timeout_sec": value,
+    }
+    if intent_type == "buy":
+        payload.update({"portfolio_id": "PF-1", "qty": 1})
+    else:
+        payload["slot_id"] = str(UUID(int=1))
+    response = make_intent_rpc_handler(intents)(
+        RpcRequest(
+            correlation_id="bad-timeout",
+            payload=payload,
+        )
+    )
+    assert response.payload["status"] == "rejected"
+    assert response.payload["reason"].startswith("bad_request:")
+    assert repo.list_undispatched_pending() == []
+
+
+@pytest.mark.parametrize("value", [1, 3600])
+def test_strategy_timeout_inclusive_bounds_are_admitted(value):
+    repo, _, intents = _wire()
+    decision = intents.submit_buy(
+        BuyIntent(intent_id=f"b-{value}", portfolio_id="PF-1", qty=1, timeout_sec=value)
+    )
+    assert decision.admitted
+    assert repo.fetch_slot(decision.slot_id).timeout_sec == value
+
+
+def test_buy_and_sell_omission_use_instance_override_or_hard_fallback():
+    repo, service, intents = _wire(default_timeout=120)
+    buy = intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="PF-1", qty=1))
+    assert repo.fetch_slot(buy.slot_id).timeout_sec == 120
+    service.transit_slot(
+        buy.slot_id,
+        SlotState.PENDING_BUY,
+        SlotState.HOLDING,
+        reason="fixture",
+        actor="test",
+        qty=1,
+        avg_price=1000,
+    )
+    sell = intents.submit_sell(SellIntent(intent_id="s1", slot_id=buy.slot_id))
+    assert sell.admitted
+    assert repo.fetch_slot(buy.slot_id).timeout_sec == 120
+
+    fallback_repo, _, fallback = _wire(default_timeout=None)
+    fallback_buy = fallback.submit_buy(
+        BuyIntent(intent_id="fallback", portfolio_id="PF-1", qty=1)
+    )
+    assert fallback_repo.fetch_slot(fallback_buy.slot_id).timeout_sec == 3600
+
+
+@pytest.mark.parametrize("value", [0, -1, 3601, 1.5, True, "30"])
+def test_instance_timeout_outside_toml_integer_contract_fails_startup(tmp_path, value):
+    path = tmp_path / "instance.toml"
+    literal = str(value).lower() if isinstance(value, bool) else repr(value)
+    path.write_text(f'instance_key = "x"\ndefault_order_timeout_sec = {literal}\n')
+    with pytest.raises(RuntimeError, match="default_order_timeout_sec"):
+        DaemonInstanceConfig.from_toml(str(path))
+
+
+def test_instance_timeout_omission_and_bounds(tmp_path):
+    path = tmp_path / "instance.toml"
+    path.write_text('instance_key = "x"\n')
+    assert DaemonInstanceConfig.from_toml(str(path)).default_order_timeout_sec == 3600
+    for value in (1, 3600):
+        path.write_text(f'instance_key = "x"\ndefault_order_timeout_sec = {value}\n')
+        assert DaemonInstanceConfig.from_toml(str(path)).default_order_timeout_sec == value
+
+
+def test_migration_does_not_invent_legacy_submission_metadata():
+    migration = (
+        Path(__file__).parents[2]
+        / "trex_core/adapters/persistence/sql/022_persist_order_timeout_anchor.sql"
+    ).read_text(encoding="utf-8")
+
+    assert "ADD COLUMN IF NOT EXISTS order_submitted_at" in migration
+    assert "ADD COLUMN IF NOT EXISTS order_date" in migration
+    assert "UPDATE daily_slot" not in migration
+    assert "COALESCE(updated_at, created_at)" not in migration
+
+
+def test_dispatch_anchor_and_original_date_survive_bookkeeping_and_restart():
+    broker = _RecordingBroker()
+    repo, service, intents = _wire()
+    execution = ExecutionService(service, repo, broker)
+    decision = intents.submit_buy(
+        BuyIntent(intent_id="b1", portfolio_id="PF-1", qty=1, timeout_sec=60)
+    )
+    broker.responder = lambda _side: FillStatus(state="accepted")
+    execution.run_once()
+    dispatched = repo.fetch_slot(decision.slot_id)
+    submitted_at = dispatched.order_submitted_at
+    order_date = dispatched.order_date
+    assert submitted_at is not None and order_date == submitted_at.astimezone(_KST).strftime("%Y%m%d")
+
+    assert repo.replace_dispatch(dispatched.slot_id, dispatched.broker_order_id, "resolved-1")
+    resolved = repo.fetch_slot(dispatched.slot_id)
+    assert resolved.order_submitted_at == submitted_at
+    assert resolved.order_date == order_date
+
+    repo.mark_cancel_outcome(dispatched.slot_id, "resolved-1", "unknown")
+    persisted = repo.fetch_slot(dispatched.slot_id)
+    assert persisted.updated_at != dispatched.updated_at
+    assert persisted.order_submitted_at == submitted_at
+    assert persisted.order_date == order_date
+    assert ExecutionService(service, repo, broker).reconcile().filled == 0
+    restarted = repo.fetch_slot(dispatched.slot_id)
+    assert restarted.order_submitted_at == submitted_at
+    assert restarted.order_date == order_date
+
+    service.transit_slot(
+        dispatched.slot_id,
+        SlotState.PENDING_BUY,
+        SlotState.HOLDING,
+        reason="terminal-fixture",
+        actor="test",
+        qty=1,
+        avg_price=1000,
+    )
+    repo.clear_dispatch(dispatched.slot_id)
+    terminal = repo.fetch_slot(dispatched.slot_id)
+    assert terminal.order_submitted_at is None
+    assert terminal.order_date is None
+    assert terminal.cancel_attempts == 0
+
+
+class _AlarmRecorder:
+    def __init__(self):
+        self.calls = []
+
+    def publish_alarm(self, code, title, *, fields=None, severity="critical"):
+        self.calls.append((code, severity, fields))
+
+
+def test_legacy_pending_without_authoritative_metadata_is_fail_closed_and_alarm_bounded():
+    broker = _RecordingBroker()
+    repo, service, intents = _wire()
+    alarms = _AlarmRecorder()
+    execution = ExecutionService(service, repo, broker, alarm_emitter=alarms)
+    decision = intents.submit_buy(
+        BuyIntent(intent_id="legacy", portfolio_id="PF-1", qty=1, timeout_sec=1)
+    )
+    broker.responder = lambda _side: FillStatus(state="accepted")
+    execution.run_once()
+    slot = repo.fetch_slot(decision.slot_id)
+    repo._slots[slot.slot_id] = replace(
+        slot,
+        order_submitted_at=None,
+        order_date=None,
+        updated_at=datetime.now(tz=timezone.utc) - timedelta(days=2),
+    )
+
+    execution.run_once()
+    execution.run_once()
+
+    persisted = repo.fetch_slot(slot.slot_id)
+    assert persisted.status is SlotState.PENDING_BUY
+    assert broker.cancel_calls == []
+    metadata_alarms = [
+        call for call in alarms.calls if call[0] == "order_reconciliation_metadata_unknown"
+    ]
+    assert len(metadata_alarms) == 1
+    assert set(metadata_alarms[0][2]) == {"slot_id", "symbol"}
+
+
+def test_cancel_rejected_is_fail_closed_alarm_and_authoritative_bounded_retry():
+    broker = _RecordingBroker()
+    broker.cancel_outcome = CancelOutcome(CancelState.REJECTED, "closed_session")
+    repo, service, intents = _wire()
+    alarms = _AlarmRecorder()
+    execution = ExecutionService(service, repo, broker, alarm_emitter=alarms)
+    decision = intents.submit_buy(
+        BuyIntent(intent_id="b1", portfolio_id="PF-1", qty=1, timeout_sec=1)
+    )
+    broker.responder = lambda _side: FillStatus(state="accepted")
+    execution.run_once()
+    slot = repo.fetch_slot(decision.slot_id)
+    repo._slots[slot.slot_id] = replace(
+        slot,
+        order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(minutes=1),
+    )
+    execution.run_once()
+    rejected = repo.fetch_slot(slot.slot_id)
+    assert rejected.status is SlotState.PENDING_BUY
+    assert rejected.cancel_state == "rejected"
+    assert len(broker.cancel_calls) == 1
+    assert any(code == "order_cancel_unresolved" and severity == "critical" for code, severity, _ in alarms.calls)
+
+    # No visible broker row => no blind retry, even after restart/backoff.
+    repo._slots[slot.slot_id] = replace(
+        rejected,
+        cancel_requested_at=datetime.now(tz=timezone.utc) - timedelta(minutes=10),
+    )
+    ExecutionService(service, repo, broker, alarm_emitter=alarms).run_once()
+    assert len(broker.cancel_calls) == 1
+
+    # An authoritative open row permits one backoff-gated retry.
+    broker.responder = lambda _side: FillStatus(
+        state="accepted", raw={"order_open_confirmed": True}
+    )
+    broker.cancel_outcome = CancelOutcome(CancelState.REQUESTED)
+    ExecutionService(service, repo, broker, alarm_emitter=alarms).run_once()
+    assert len(broker.cancel_calls) == 2
+    retried = repo.fetch_slot(slot.slot_id)
+    assert retried.cancel_state == "requested"
+    assert retried.cancel_attempts == 2
+    ExecutionService(service, repo, broker).run_once()
+    assert len(broker.cancel_calls) == 2
+
+
+class _Resp:
+    status_code = 200
+
+    def __init__(self, payload):
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+
+def _kis(clock):
+    broker = KISBroker(
+        "12345678-01",
+        app_key="k",
+        app_secret="s",
+        base_url="https://openapivts.koreainvestment.com:29443",
+        clock=clock,
+    )
+    broker._token = "token"
+    broker._token_exp = datetime.now(tz=timezone.utc) + timedelta(hours=1)
+    return broker
+
+
+def test_kis_rollover_uses_original_date_and_row_visibility_is_nonterminal(monkeypatch):
+    now = datetime(2026, 8, 15, 9, 0, tzinfo=_KST)
+    submitted = datetime(2026, 8, 14, 15, 20, tzinfo=_KST)
+    seen = {}
+
+    def fake_get(*_args, **kwargs):
+        seen.update(kwargs["params"])
+        return _Resp({"rt_cd": "0", "output1": []})
+
+    monkeypatch.setattr(kis_mod.httpx, "get", fake_get)
+    status = _kis(lambda: now).poll_fill(
+        "06010:0000123456",
+        side="buy",
+        symbol="005930",
+        order_date="20260814",
+        submitted_at=submitted,
+    )
+    assert seen["INQR_STRT_DT"] == seen["INQR_END_DT"] == "20260814"
+    assert status.state == "unknown"
+    assert status.raw["reason"] == "order_not_visible_unresolved"
+    assert status.raw["order_visible"] is False
+
+
+def test_kis_recent_missing_row_stays_in_bounded_visibility_grace(monkeypatch):
+    now = datetime(2026, 8, 14, 9, 0, tzinfo=_KST)
+    monkeypatch.setattr(
+        kis_mod.httpx,
+        "get",
+        lambda *_args, **_kwargs: _Resp({"rt_cd": "0", "output1": []}),
+    )
+    status = _kis(lambda: now).poll_fill(
+        "06010:0000123457",
+        side="sell",
+        symbol="005930",
+        order_date="20260814",
+        submitted_at=now - timedelta(seconds=30),
+    )
+    assert status.state == "accepted"
+    assert status.raw["reason"] == "order_visibility_grace"
+    assert status.raw["order_visible"] is False

tests/integration/test_kis_submission_recovery_virtual.py

@@ -144,7 +144,13 @@ def _wire(broker: KISBroker):
         instance_key=INSTANCE,
     )
     repo.reconcile_universe(INSTANCE, {PORTFOLIO: SYMBOL})
-    execution = ExecutionService(slots, repo, broker, instance_key=INSTANCE)
+    execution = ExecutionService(
+        slots,
+        repo,
+        broker,
+        instance_key=INSTANCE,
+        clock=lambda: VIRTUAL_AT,
+    )
     decision = intents.submit_buy(
         BuyIntent(
             intent_id="virtual-buy-1",
@@ -170,7 +176,13 @@ def test_timeout_after_exchange_accept_resolves_without_duplicate_and_survives_r
         assert len(state.orders) == 1
 
         state.fill_orders = True
-        restarted = ExecutionService(slots, repo, _broker(server), instance_key=INSTANCE)
+        restarted = ExecutionService(
+            slots,
+            repo,
+            _broker(server),
+            instance_key=INSTANCE,
+            clock=lambda: VIRTUAL_AT,
+        )
         restarted.reconcile()
 
         holding = repo.fetch_slot(slot_id)