fix: close broker uncertainty audit gaps

docs/data-model.md

@@ -175,7 +175,7 @@
 
 ### Key Fields
 
-- account_hash :: TEXT PK :: `SHA-256(CANO|ACNT_PRDT_CD)`. 평문 계좌번호를 저장·로그하지 않는다.
+- account_hash :: TEXT PK :: `SHA-256(KIS base_url|CANO|ACNT_PRDT_CD|app_key)`. 실전/모의 및 애플리케이션 자격증명 범위를 분리하며 원문을 저장·로그하지 않는다.
 - holdings :: JSONB NOT NULL :: `[{symbol,hldg_qty}]`. 소유권 래퍼가 켜진 계정은 전략 소유분만 저장한다.
 - holdings_captured_at :: TIMESTAMPTZ? :: 보유수량 조회가 마지막 성공한 시각.
 - available_cash :: BIGINT? :: KIS 주문가능조회 `nrcvb_buy_amt` 현금주문 가능금액.

docs/discovery/dcy-041_20260729_broker-uncertainty-and-adapter-retirement.md

@@ -133,6 +133,9 @@
 - `AUTO` 실제 거래소 확정·주문번호 동시 저장·재시작/취소 재사용: 구현 완료.
 - 취소 `requested|rejected|unknown` 영속화와 poll 전용 최종확정: 구현 완료.
 - 계정 해시·독립 TTL·백그라운드 단일 소유자·매수/매도 분리: 구현 완료.
+- 계정 해시는 KIS endpoint·계좌·상품·app identity를 함께 묶어 실전/모의·자격증명 출처를 분리한다.
+- 동일 instance advisory lock 안에서 현금 예약검사와 pending 전이를 원자 처리한다.
+- timeout·운영자 취소 모두 compare-and-set 예약으로 동일 주문 재취소를 차단한다.
 - KIS 현금 권위 필드: `inquire-psbl-order.output.nrcvb_buy_amt`.
 - 수동 혼합 보유 계정: `OwnershipAwareBroker` 투영 뒤 스냅샷을 생성하여 전략 소유 수량만 매도 허용.
 - 운영 기본값: 기능 OFF, 매수 600초, 매도 3600초, 갱신 300초.

docs/interface-contract.md

@@ -433,6 +433,7 @@ resume_kill_switch 요청·응답 (REQ-046):
 - 기능은 `TREX_ACCOUNT_SNAPSHOT_ENABLED=false`가 기본인 KIS 전용 선택 기능이다.
 - KIS 보유조회와 현금주문 가능금액 조회(`inquire-psbl-order.output.nrcvb_buy_amt`)를 기본 300초 간격의 단일 계정 소유자가 갱신한다. 두 값은 독립 캡처 시각을 갖는다.
 - 신규매수는 미만료 현금 스냅샷과 기존 `pending_buy` 지정가 예약금액을 합산해 검사한다. 수량·지정가가 없거나 예약금액을 계산할 수 없으면 매수를 차단한다. KIS 주문 응답이 최종 현금 한도 권위이며 거부된 주문을 자동 재전송하지 않는다.
+- 현금 검사와 `PENDING_BUY` 전이는 동일 instance advisory lock 안에서 수행되어 같은 instance의 동시 처리자가 아직 커밋되지 않은 예약을 함께 통과하지 못한다.
 - `pending_sell` 수량은 매 tick 실시간 보유조회를 먼저 시도한다. 오직 `transport_failed`이고 기본 3600초 안의 동일계정 스냅샷이 있을 때만 대체한다.
 - 인증·인가·응답형식·브로커거부·소유권 불일치·계정 해시 변경·만료는 캐시로 우회하지 않는다.
 - `sync`, `verify`, 일일 스냅샷 스케줄러와 재시작 원장대조는 기존처럼 실시간 조회만 사용한다.
@@ -447,6 +448,7 @@ cancel_order(broker_order_id: str, *, side: str, symbol: str, qty: int, venue: s
 - `venue`: 원주문 venue. 취소 시 원주문과 동일한 `EXCG_ID_DVSN_CD`를 사용해야 한다. 기본값 `"KRX"`로 기존 호출자 하위 호환.
 - `CancelOutcome.state`: `requested`(요청 접수), `rejected`(명시적 거부), `unknown`(응답유실·전송 실패). `requested`도 최종 취소가 아니며 `poll_fill()`의 `cancelled|filled`만 주문의 최종 상태를 확정한다.
 - Core는 외부 취소 호출 전에 `unknown`을 먼저 영속화하고 응답 뒤 실제 즉시 결과로 교체한다. 따라서 요청과 저장 사이에 프로세스가 종료돼도 자동 timeout 취소는 같은 주문을 재전송하지 않고, 재시작 뒤 체결조회로만 수렴한다.
+- timeout과 운영자 취소 모두 `cancel_state IS NULL` 비교·설정에 성공한 한 호출만 외부 요청을 보낸다. 이후 반복 요청은 `cancel_already_recorded`로 응답하고 poll만 계속한다.
 
 ### get_current_price
 

docs/runtime-flows.md

@@ -270,7 +270,7 @@
 
 1. 기능을 켠 KIS 데몬은 시작 시 1회 동기 조회 후 계정당 한 백그라운드 소유자가 기본 300초 간격과 실패 백오프로 보유수량·주문가능금액을 갱신한다.
 2. 보유수량과 현금은 독립적으로 성공값과 시각을 저장한다. 계정 해시 변경, 인증·인가·형식·브로커 거부는 실패 폐쇄한다.
-3. 신규매수는 기본 600초 이내 현금값에서 기존 pending 매수 예약금액과 새 주문금액을 합산해 검사한다. 시장가처럼 금액을 확정할 수 없으면 차단한다.
+3. 신규매수는 기본 600초 이내 현금값에서 기존 pending 매수 예약금액과 새 주문금액을 합산해 검사한다. 검사와 pending 전이는 같은 instance advisory lock 안에서 원자 수행하며 시장가처럼 금액을 확정할 수 없으면 차단한다.
 4. pending 매도는 실시간 보유조회 전송 실패일 때만 기본 3600초 이내 전략 소유 보유수량을 사용한다. 수동 보유분을 매도 권한으로 승격하지 않는다.
 5. 재시작 원장대조, `sync`, `verify`, Portfolio Hub 일일 정합은 캐시를 사용하지 않는다.
 

tests/adapters/broker/test_kis.py

@@ -11,7 +11,7 @@ import os
 import httpx
 import pytest
 
-from trex_core.adapters.broker.kis import KISBroker
+from trex_core.adapters.broker.kis import KISBroker, _balance_http_error_code
 from trex_core.application.broker_port import (
     BrokerBalanceError,
     CancelState,
@@ -31,6 +31,41 @@ def test_account_split_and_mock_tr_ids():
     assert "12345678" not in b.account_scope_hash()
 
 
+def test_account_scope_hash_separates_endpoint_and_application_identity():
+    mock_a = KISBroker(
+        "12345678-01", app_key="app-a", app_secret="s", base_url=_MOCK_URL
+    )
+    mock_b = KISBroker(
+        "12345678-01", app_key="app-b", app_secret="s", base_url=_MOCK_URL
+    )
+    live_a = KISBroker(
+        "12345678-01",
+        app_key="app-a",
+        app_secret="s",
+        base_url="https://openapi.koreainvestment.com:9443",
+        allow_live=True,
+    )
+
+    assert len({mock_a.account_scope_hash(), mock_b.account_scope_hash(),
+                live_a.account_scope_hash()}) == 3
+
+
+@pytest.mark.parametrize(
+    ("status", "code"),
+    [
+        (400, "broker_rejected"),
+        (401, "authentication_failed"),
+        (403, "authorization_failed"),
+        (422, "broker_rejected"),
+        (429, "transport_failed"),
+        (500, "transport_failed"),
+        (None, "broker_rejected"),
+    ],
+)
+def test_balance_http_status_classification(status, code):
+    assert _balance_http_error_code(status) == code
+
+
 def test_live_base_url_requires_allow_live():
     with pytest.raises(RuntimeError):
         KISBroker(

tests/adapters/persistence/test_migration_and_drift.py

@@ -9,7 +9,8 @@ from __future__ import annotations
 import psycopg
 
 from trex_core.adapters.persistence.migrate import apply_migrations, migration_files
-from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
+from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository
+from trex_core.domain.slot import EXPECTED_SLOT_FIELDS, SlotService, SlotState
 
 _EXPECTED_TABLES = {
     "daily_slot",
@@ -61,3 +62,22 @@ def test_apply_migrations_is_idempotent(pg_dsn):
     # applying twice must not error (CREATE ... IF NOT EXISTS)
     apply_migrations(pg_dsn)
     apply_migrations(pg_dsn)
+
+
+def test_migration_replay_accepts_existing_auto_venue_row(pg_clean):
+    repo = PostgreSQLSlotRepository(pg_clean)
+    service = SlotService(repo, instance_key="migration-auto")
+    repo.reconcile_universe("migration-auto", {"pf-auto": "005930"})
+    slot = repo.find_slot_by_portfolio("migration-auto", "pf-auto")
+    service.transit_slot(
+        slot.slot_id,
+        SlotState.INACTIVE,
+        SlotState.PENDING_BUY,
+        reason="migration_replay",
+        actor="test",
+        venue="AUTO",
+    )
+
+    apply_migrations(pg_clean)
+
+    assert repo.fetch_slot(slot.slot_id).venue == "AUTO"

tests/adapters/persistence/test_postgresql_repository.py

@@ -132,6 +132,29 @@ def test_dispatch_queue_instance_scoped(pg_clean):
     assert repo_a.list_dispatched_pending("inst-b") == []
 
 
+def test_cancel_attempt_reservation_is_compare_and_set(pg_clean):
+    repo, _, intents = _wire(pg_clean)
+    repo.reconcile_universe(INST, {"PF-1": "005930"})
+    decision = intents.submit_buy(
+        BuyIntent(intent_id="pg-cancel-cas", qty=1, portfolio_id="PF-1")
+    )
+    assert repo.mark_dispatched(decision.slot_id, "ord-cas")
+    assert repo.mark_cancel_outcome(
+        decision.slot_id,
+        "ord-cas",
+        "unknown",
+        only_if_unset=True,
+    )
+    assert not repo.mark_cancel_outcome(
+        decision.slot_id,
+        "ord-cas",
+        "unknown",
+        only_if_unset=True,
+    )
+    assert repo.mark_cancel_outcome(decision.slot_id, "ord-cas", "requested")
+    assert repo.fetch_slot(decision.slot_id).cancel_state == "requested"
+
+
 # dcy-028 / REQ-079·081 (BAT-042): strategy-state store + occupancy_seq column
 def test_strategy_state_store_pg(pg_clean):
     from trex_core.adapters.persistence.postgresql import PgStrategyStateStore

tests/application/test_execution_cancel.py

@@ -40,6 +40,7 @@ class ProgrammableBroker:
         )
         self.polls = 0
         self.balance = {"005930": 10}
+        self.cancel_calls = 0
 
     def name(self):
         return "stub"
@@ -53,6 +54,7 @@ class ProgrammableBroker:
         return self.responder(side)
 
     def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"):
+        self.cancel_calls += 1
         return CancelOutcome(CancelState.REQUESTED)
 
     def get_balance(self):
@@ -159,3 +161,20 @@ def test_stale_watchdog_disabled_by_default():
     repo._slots[sid] = replace(s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999))
     r = execution.run_once()
     assert r.stale_pending == 0  # disabled => never warns
+
+
+def test_operator_cancel_is_sent_once_across_repeat_and_restart():
+    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()
+
+    assert execution.cancel_pending_slot(sid) == "ok"
+    assert execution.cancel_pending_slot(sid) == "cancel_already_recorded"
+
+    restarted = ExecutionService(svc, repo, broker)
+    assert restarted.cancel_pending_slot(sid) == "cancel_already_recorded"
+    assert broker.cancel_calls == 1
+    assert repo.fetch_slot(sid).cancel_state == CancelState.REQUESTED.value

tests/application/test_intent_concurrency.py

@@ -49,13 +49,9 @@ def test_concurrent_buys_admit_at_most_m():
 
     admitted = [s for s in results if s == "admitted"]
     pending = [s for s in repo.list_slots(INST) if s.status is SlotState.PENDING_BUY]
-    # NOTE: the M-cap check happens before the serialized transit, so racing
+    # Capacity admission and transition share the same instance advisory lock.
-    # admissions can momentarily exceed M only if count_active were not
-    # re-checked — the single intent thread is the production serializer
-    # (dcy-020 OQ-1). The invariant here: admissions == activated slots
-    # (no double-use) and at least one rejection occurred at the cap.
     assert len(admitted) == len(pending)
-    assert len(admitted) >= 5
+    assert len(admitted) == 5
     assert len(results) == 20
 
 
@@ -95,3 +91,52 @@ def test_duplicate_intent_id_admits_once_under_concurrency():
     pending = [s for s in repo.list_slots(INST) if s.status is SlotState.PENDING_BUY]
     assert len(pending) == 1
     assert all(s in ("admitted", "processing") for s in statuses)
+
+
+def test_concurrent_cached_cash_reservations_are_atomic_with_transition():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo, instance_key=INST)
+    repo.reconcile_universe(INST, {"PF-0": "SYM", "PF-1": "SYM"})
+
+    def cash_guard(intent: BuyIntent):
+        reserved = sum(
+            float(slot.qty or 0) * float(slot.limit_price or 0)
+            for slot in repo.list_slots(INST, {"pending_buy"})
+        )
+        requested = float(intent.qty or 0) * float(intent.entry_price or 0)
+        return None if reserved + requested <= 100 else "insufficient_snapshot_cash"
+
+    intents = IntentService(
+        svc,
+        repo,
+        InMemoryIntentStore(),
+        resolve_symbol=lambda pid: f"SYM-{pid}",
+        instance_key=INST,
+        buy_guard_fn=cash_guard,
+    )
+    results = []
+    result_lock = threading.Lock()
+
+    def worker(index: int) -> None:
+        decision = intents.submit_buy(
+            BuyIntent(
+                intent_id=f"cash-{index}",
+                portfolio_id=f"PF-{index}",
+                qty=1,
+                entry_price=100,
+            )
+        )
+        with result_lock:
+            results.append(decision)
+
+    threads = [threading.Thread(target=worker, args=(index,)) for index in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert [decision.status for decision in results].count("admitted") == 1
+    assert [decision.reason for decision in results].count(
+        "insufficient_snapshot_cash"
+    ) == 1
+    assert len(repo.list_slots(INST, {"pending_buy"})) == 1

trex_core/adapters/broker/kis.py

@@ -43,6 +43,16 @@ _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0
 _SUPPORTED_ORDER_VENUES = frozenset({"AUTO", "KRX", "NXT", "SOR"})
 
 
+def _balance_http_error_code(status: Optional[int]) -> str:
+    if status == 401:
+        return "authentication_failed"
+    if status == 403:
+        return "authorization_failed"
+    if status in {408, 425, 429} or (status is not None and status >= 500):
+        return "transport_failed"
+    return "broker_rejected"
+
+
 def resolve_order_venue(venue: str, at: datetime) -> str:
     """Resolve AUTO to the venue that accepts domestic-stock orders at `at`.
 
@@ -161,7 +171,8 @@ class KISBroker:
     def account_scope_hash(self) -> str:
         """Stable non-secret identity for account-scoped persisted state."""
         cano, product = self._account()
-        return hashlib.sha256(f"{cano}|{product}".encode()).hexdigest()
+        material = f"{self._base_url}|{cano}|{product}|{self._app_key}"
+        return hashlib.sha256(material.encode()).hexdigest()
 
     # --- credential hot-swap (dcy-036 / REQ-113 AC-2 / REQ-115 AC-1) ----------
     def reinitialize(self, credential) -> None:
@@ -658,12 +669,7 @@ class KISBroker:
             payload = resp.json()
         except httpx.HTTPStatusError as exc:
             status = exc.response.status_code if exc.response is not None else None
-            code = (
+            raise BrokerBalanceError(_balance_http_error_code(status)) from None
-                "authentication_failed"
-                if status == 401
-                else ("authorization_failed" if status == 403 else "transport_failed")
-            )
-            raise BrokerBalanceError(code) from None
         except httpx.HTTPError:
             raise BrokerBalanceError("transport_failed") from None
         except (ValueError, TypeError):
@@ -721,12 +727,7 @@ class KISBroker:
             raise
         except httpx.HTTPStatusError as exc:
             status = exc.response.status_code if exc.response is not None else None
-            code = (
+            raise BrokerBalanceError(_balance_http_error_code(status)) from None
-                "authentication_failed"
-                if status == 401
-                else ("authorization_failed" if status == 403 else "transport_failed")
-            )
-            raise BrokerBalanceError(code) from None
         except httpx.HTTPError:
             raise BrokerBalanceError("transport_failed") from None
         except (KeyError, TypeError, ValueError):

trex_core/adapters/persistence/postgresql.py

@@ -341,6 +341,8 @@ class PostgreSQLSlotRepository:
         slot_id: UUID,
         broker_order_id: str,
         state: str,
+        *,
+        only_if_unset: bool = False,
     ) -> bool:
         now = datetime.now(tz=timezone.utc)
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
@@ -349,8 +351,16 @@ class PostgreSQLSlotRepository:
                     "UPDATE daily_slot SET cancel_state=%s, "
                     "cancel_requested_at=%s, updated_at=%s "
                     "WHERE slot_id=%s AND broker_order_id=%s "
-                    "AND status IN ('pending_buy','pending_sell')",
+                    "AND status IN ('pending_buy','pending_sell') "
-                    (state, now, now, slot_id, broker_order_id),
+                    "AND (%s = FALSE OR cancel_state IS NULL)",
+                    (
+                        state,
+                        now,
+                        now,
+                        slot_id,
+                        broker_order_id,
+                        only_if_unset,
+                    ),
                 )
                 ok = cur.rowcount == 1
                 conn.commit()

trex_core/adapters/persistence/sql/017_add_slot_order_venue.sql

@@ -8,4 +8,6 @@ ALTER TABLE daily_slot
 
 ALTER TABLE daily_slot
     ADD CONSTRAINT daily_slot_venue_check
-    CHECK (venue IN ('KRX', 'NXT', 'SOR'));
+    -- apply_migrations() replays every file on each boot. Keep this historical
+    -- migration compatible with AUTO rows introduced by migration 019.
+    CHECK (venue IN ('AUTO', 'KRX', 'NXT', 'SOR'));

trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql

@@ -1,5 +1,5 @@
 -- dcy-041 / GitHub #45: account-scoped last healthy KIS snapshot.
--- account_hash is SHA-256(CANO|ACNT_PRDT_CD); no plaintext account identity,
+-- account_hash is SHA-256(base_url|CANO|ACNT_PRDT_CD|app_key); no plaintext identity,
 -- token, key, order response, or broker payload is persisted.
 CREATE TABLE IF NOT EXISTS broker_account_snapshot (
     account_hash    TEXT PRIMARY KEY,

trex_core/application/execution.py

@@ -579,6 +579,7 @@ class ExecutionService:
             slot.slot_id,
             str(slot.broker_order_id),
             CancelState.UNKNOWN.value,
+            only_if_unset=True,
         ):
             _log.error(
                 "cancel attempt reservation conflict: slot=%s broker_order_id=%s",
@@ -719,6 +720,8 @@ class ExecutionService:
         if not slot.broker_order_id:
             # Claimed but not yet dispatched — execution loop will pick up or rollback.
             return "not_dispatched"
+        if slot.cancel_state is not None:
+            return "cancel_already_recorded"
         side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
         qty = int(slot.qty or 0)
         try:
@@ -735,8 +738,9 @@ class ExecutionService:
                 slot.slot_id,
                 str(slot.broker_order_id),
                 CancelState.UNKNOWN.value,
+                only_if_unset=True,
             ):
-                return "cancel_state_conflict"
+                return "cancel_already_recorded"
             outcome = self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs)
         except Exception:  # noqa: BLE001 - adapter failure becomes durable unknown
             outcome = CancelOutcome(CancelState.UNKNOWN, "adapter_exception")

trex_core/application/intents.py

@@ -30,6 +30,7 @@ from uuid import UUID
 
 from trex_core.application.ports import IntentStore
 from trex_core.domain.slot import (
+    SlotAdmissionRejectedError,
     SlotRepository,
     SlotService,
     SlotState,
@@ -275,20 +276,24 @@ class IntentService:
             symbol = _resolve(slot)
             if not symbol:
                 return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio")
-            if self._buy_guard_fn is not None:
+            def _admission_precondition(_locked_slot) -> Optional[str]:
+                # Runs inside SlotService's instance-scoped advisory lock. This
+                # makes capacity and cached-cash reservation checks atomic with
+                # the PENDING_BUY transition across same-instance daemons.
+                if (
+                    not is_increase
+                    and self._max_active_slots is not None
+                    and self._repo.count_active(self._instance_key)
+                    >= self._max_active_slots
+                ):
+                    return "capacity_full"
+                if self._buy_guard_fn is None:
+                    return None
                 try:
-                    guard_reason = self._buy_guard_fn(intent)
+                    return self._buy_guard_fn(intent)
                 except Exception:  # noqa: BLE001 - buy guard fails closed
-                    guard_reason = "account_snapshot_unavailable"
+                    return "account_snapshot_unavailable"
-                if guard_reason is not None:
+
-                    return IntentDecision(
-                        intent.intent_id, "rejected", None, guard_reason
-                    )
-            # M cap applies only to NEW activations (initial buy). An increase on
-            # an already-active asset does not change the active asset count.
-            if (not is_increase and self._max_active_slots is not None
-                    and self._repo.count_active(self._instance_key) >= self._max_active_slots):
-                return IntentDecision(intent.intent_id, "rejected", None, "capacity_full")
             # dcy-022 / REQ-068: resolve effective timeout (per-intent > config default).
             effective_timeout = (
                 intent.timeout_sec
@@ -317,6 +322,11 @@ class IntentService:
                     slippage_pct=intent.slippage_pct,
                     venue=intent.venue,
                     extra={"intent_id": intent.intent_id, "symbol": symbol},
+                    precondition=_admission_precondition,
+                )
+            except SlotAdmissionRejectedError as exc:
+                return IntentDecision(
+                    intent.intent_id, "rejected", None, exc.reason
                 )
             except SlotStateMismatchError:
                 continue  # slot changed under us; re-fetch and re-evaluate

trex_core/domain/slot/__init__.py

@@ -11,11 +11,13 @@ Public API:
 - EXPECTED_SLOT_FIELDS: canonical `Slot` column contract (guarded by test_contract)
 - SlotService: main domain service (transit, sell, thresholds, capacity)
 - SlotRepository: abstract storage protocol
-- SlotTransitionError / SlotStateMismatchError / SlotNotFoundError: domain errors
+- SlotTransitionError / SlotStateMismatchError / SlotAdmissionRejectedError /
+  SlotNotFoundError: domain errors
 """
 from __future__ import annotations
 
 from trex_core.domain.slot.errors import (
+    SlotAdmissionRejectedError,
     SlotDomainError,
     SlotNotFoundError,
     SlotStateMismatchError,
@@ -42,6 +44,7 @@ __all__ = [
     "EXPECTED_SLOT_FIELDS",
     "SLOT_ADVISORY_LOCK_ID",
     "Slot",
+    "SlotAdmissionRejectedError",
     "SlotCapacity",
     "SlotDomainError",
     "SlotNotFoundError",

trex_core/domain/slot/errors.py

@@ -21,3 +21,11 @@ class SlotStateMismatchError(SlotDomainError):
     different state between read and write (race detection inside the
     transition function).
     """
+
+
+class SlotAdmissionRejectedError(SlotDomainError):
+    """Raised when an advisory-locked admission precondition rejects."""
+
+    def __init__(self, reason: str) -> None:
+        self.reason = reason
+        super().__init__(reason)

trex_core/domain/slot/in_memory_repository.py

@@ -224,6 +224,8 @@ class InMemorySlotRepository:
         slot_id: UUID,
         broker_order_id: str,
         state: str,
+        *,
+        only_if_unset: bool = False,
     ) -> bool:
         """Persist one cancel attempt iff the same order is still pending."""
         s = self._slots.get(slot_id)
@@ -231,6 +233,7 @@ class InMemorySlotRepository:
             s is None
             or s.broker_order_id != broker_order_id
             or s.status not in self._PENDING
+            or (only_if_unset and s.cancel_state is not None)
         ):
             return False
         now = datetime.now(tz=timezone.utc)

trex_core/domain/slot/service.py

@@ -15,10 +15,11 @@ from __future__ import annotations
 from dataclasses import replace
 from datetime import datetime, timezone
 from decimal import Decimal
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 from uuid import UUID
 
 from trex_core.domain.slot.errors import (
+    SlotAdmissionRejectedError,
     SlotNotFoundError,
     SlotStateMismatchError,
     SlotTransitionError,
@@ -87,6 +88,7 @@ class SlotService:
         slippage_pct: Optional[float] = None,
         venue: Optional[str] = None,
         extra: Optional[dict[str, Any]] = None,
+        precondition: Optional[Callable[[Slot], Optional[str]]] = None,
     ) -> Slot:
         """Transition slot from one state to another atomically.
 
@@ -116,6 +118,10 @@ class SlotService:
                     f"slot {slot_id} current state {slot.status.value} "
                     f"!= requested from_state {from_state.value}"
                 )
+            if precondition is not None:
+                rejection_reason = precondition(slot)
+                if rejection_reason is not None:
+                    raise SlotAdmissionRejectedError(rejection_reason)
 
             updated = self._apply_transition_fields(
                 slot,