From 59bb4928839354844928fd3b4ea999547d1e43f2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:27:27 +0000 Subject: [PATCH 01/10] Initial plan From 084e05a97db0a29d7453c1737ac4cffe5562a1e0 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:33:12 +0000 Subject: [PATCH 02/10] feat(ops): track broker balance worker failure lifecycle and backoff Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/adapters/broker/test_kis.py | 30 +++- tests/application/test_execution.py | 58 ++++++++ tests/application/test_snapshot_scheduler.py | 57 ++++++++ trex_core/adapters/broker/kis.py | 10 +- trex_core/application/account_snapshot.py | 7 +- trex_core/application/balance_health.py | 139 +++++++++++++++++++ trex_core/application/broker_port.py | 2 +- trex_core/application/execution.py | 80 ++++++++++- trex_core/application/snapshot_scheduler.py | 97 +++++++++++-- 9 files changed, 453 insertions(+), 27 deletions(-) create mode 100644 trex_core/application/balance_health.py diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index f2251bf..ff36ab6 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -59,7 +59,8 @@ def test_account_scope_hash_separates_endpoint_and_application_identity(): (403, "authorization_failed"), (422, "broker_rejected"), (429, "transport_failed"), - (500, "transport_failed"), + (500, "http_5xx"), + (504, "timeout"), (None, "broker_rejected"), ], ) @@ -350,8 +351,8 @@ def test_get_balance_fails_closed_on_later_transport_and_http_errors(monkeypatch request = httpx.Request("GET", "https://example.test/balance") response = httpx.Response(503, request=request) for failure, expected_code in [ - (httpx.ConnectTimeout("unavailable"), "transport_failed"), - (httpx.HTTPStatusError("bad gateway", request=request, response=response), "transport_failed"), + (httpx.ConnectTimeout("unavailable"), "timeout"), + (httpx.HTTPStatusError("bad gateway", request=request, response=response), "http_5xx"), ]: broker, calls = _balance_broker(monkeypatch, [ _BalanceResponse(_balance_payload([{"pdno": "005930", "hldg_qty": "2"}], fk="A", nk="1"), "F"), @@ -407,6 +408,29 @@ def test_get_available_cash_uses_cash_only_amount(monkeypatch): assert b.get_available_cash() == 695_591 +@pytest.mark.parametrize( + ("failure", "expected_code"), + [ + (httpx.ReadTimeout("timed out"), "timeout"), + ( + httpx.HTTPStatusError( + "bad gateway", + request=httpx.Request("GET", "https://example.test"), + response=httpx.Response(502, request=httpx.Request("GET", "https://example.test")), + ), + "http_5xx", + ), + ], +) +def test_get_available_cash_classifies_timeout_and_http_5xx(monkeypatch, failure, expected_code): + b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) + monkeypatch.setattr(b, "_headers", lambda _tr: {}) + monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: (_ for _ in ()).throw(failure)) + with pytest.raises(BrokerBalanceError) as raised: + b.get_available_cash() + assert raised.value.code == expected_code + + def test_submit_transport_failure_blocks_automatic_duplicate(monkeypatch): b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) monkeypatch.setattr(b, "_headers", lambda _tr: {}) diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 6b6ba80..e5df5c4 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -5,8 +5,11 @@ In-memory adapters; deterministic paper double drives the full async path. from __future__ import annotations +from datetime import UTC, datetime, timedelta + from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore +from trex_core.application.account_snapshot import AccountSnapshotUnavailable from trex_core.application.broker_port import ( BalanceItem, CancelOutcome, @@ -214,6 +217,61 @@ def test_pending_sell_uses_injected_bounded_balance_reader(): assert repo.fetch_slot(sid).status is SlotState.PENDING_SELL +def test_pending_sell_balance_failure_is_fail_closed_with_backoff_and_recovery(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + broker = PaperBroker(fill_after_polls=0) + repo, svc, intents, _ = _wire(broker) + sid = _buy(intents, qty=3).slot_id + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + clock=lambda: now["t"], + ) + execution.run_once() + svc.sell(sid, actor="t", reason="full") + broker._fill_after = 5 + + calls = {"n": 0} + reader_state = {"ok": False} + + def failing_reader(): + calls["n"] += 1 + if not reader_state["ok"]: + raise AccountSnapshotUnavailable("expired") + return [BalanceItem("005931", 3)] + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=failing_reader, + clock=lambda: now["t"], + ) + first = execution.run_once() + assert first.sell_skipped_no_balance == 1 + assert calls["n"] == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "unavailable" + assert health["failure_code"] == "stale_snapshot" + assert health["consecutive_failures"] == 1 + + second = execution.run_once() + assert second.sell_skipped_no_balance == 1 + assert calls["n"] == 1 # bounded backoff prevented another live read + + now["t"] += timedelta(seconds=16) + reader_state["ok"] = True + third = execution.run_once() + assert third.dispatched == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "ok" + assert health["consecutive_failures"] == 0 + + def test_partial_sell_qty_is_min_intent_broker(): """dcy-021/REQ-062 (D4): partial sell dispatches min(intent qty, broker); remainder stays holding (broker-authoritative qty), cost basis preserved.""" diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 6b0c7c2..e6aa00e 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -665,3 +665,60 @@ def test_build_kis_reinit_callback_returns_callable_for_kis(): result(new_cred) assert broker._account_ref == "new-acct" assert broker._app_key == "new-k" + + +def test_balance_worker_health_isolated_backoff_and_recovery(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class SequencedBroker(PaperBroker): + def __init__(self): + super().__init__() + self.steps = [] + + def get_balance(self): + if not self.steps: + return [] + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=timezone.utc)} + read = _make_stub() + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + broker = SequencedBroker() + sched = SnapshotScheduler( + read_port=read, + store=store, + broker=broker, + slot_repo=slot_repo, + slot_service=slot_service, + clock=lambda: now["t"], + ) + + broker.steps = [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + assert sched.run_once() is True + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "unavailable" + assert state["position_reconciliation"]["failure_code"] == "http_5xx" + assert state["position_reconciliation"]["consecutive_failures"] == 1 + assert state["core_holding_reconciliation"]["state"] == "ok" + + broker.steps = [[BalanceItem("X", 0)]] + assert sched.run_once() is True + assert broker.steps == [] # position worker stayed in backoff, core worker still ran + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["consecutive_failures"] == 1 + assert state["core_holding_reconciliation"]["state"] == "ok" + + now["t"] += timedelta(seconds=16) + broker.steps = [[BalanceItem("X", 0)], [BalanceItem("X", 0)]] + assert sched.run_once() is True + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "ok" + assert state["position_reconciliation"]["consecutive_failures"] == 0 diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 4c2e711..d936632 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -54,7 +54,11 @@ def _balance_http_error_code(status: int | None) -> str: return "authentication_failed" if status == 403: return "authorization_failed" - if status in {408, 425, 429} or (status is not None and status >= 500): + if status in {408, 504}: + return "timeout" + if status is not None and status >= 500: + return "http_5xx" + if status in {425, 429}: return "transport_failed" return "broker_rejected" @@ -1235,6 +1239,8 @@ class KISBroker: except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None raise BrokerBalanceError(_balance_http_error_code(status)) from None + except httpx.TimeoutException: + raise BrokerBalanceError("timeout") from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (ValueError, TypeError): @@ -1318,6 +1324,8 @@ class KISBroker: except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None raise BrokerBalanceError(_balance_http_error_code(status)) from None + except httpx.TimeoutException: + raise BrokerBalanceError("timeout") from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (KeyError, TypeError, ValueError): diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index c61c08c..c42439b 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -52,10 +52,11 @@ class AccountSnapshotUnavailable(RuntimeError): "read_failed", }) - def __init__(self, code: str) -> None: + def __init__(self, code: str, *, detail_code: str | None = None) -> None: if code not in self._CODES: raise ValueError(f"unsupported account snapshot error code: {code}") self.code = code + self.detail_code = detail_code super().__init__(f"account snapshot unavailable ({code})") @@ -217,7 +218,9 @@ class AccountSnapshotService: return self._broker.get_balance() except BrokerBalanceError as exc: if exc.code != "transport_failed": - raise AccountSnapshotUnavailable("live_read_failed") from None + raise AccountSnapshotUnavailable( + "live_read_failed", detail_code=exc.code + ) from None snapshot = self._load_current() age = self._age(snapshot.holdings_captured_at, self._sell_max_age) _log.warning( diff --git a/trex_core/application/balance_health.py b/trex_core/application/balance_health.py new file mode 100644 index 0000000..37194d5 --- /dev/null +++ b/trex_core/application/balance_health.py @@ -0,0 +1,139 @@ +"""Broker-balance worker health tracking with bounded backoff.""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from threading import Lock +from typing import Callable + + +@dataclass(frozen=True) +class BalanceWorkerHealth: + worker: str + last_success_at: datetime | None + consecutive_failures: int + failure_code: str | None + failure_at: datetime | None + next_retry_at: datetime | None + + def as_dict(self, *, now: datetime, stale_after_sec: float) -> dict[str, object]: + stale = False + if self.last_success_at is not None and stale_after_sec >= 0: + stale = (now - self.last_success_at).total_seconds() > stale_after_sec + state = "ok" + if self.last_success_at is None: + state = "unavailable" + elif self.consecutive_failures > 0: + state = "unavailable" + elif stale: + state = "stale" + return { + "worker": self.worker, + "state": state, + "last_success_at": ( + self.last_success_at.astimezone(UTC).isoformat() + if self.last_success_at is not None + else None + ), + "consecutive_failures": self.consecutive_failures, + "failure_code": self.failure_code, + "failure_at": ( + self.failure_at.astimezone(UTC).isoformat() + if self.failure_at is not None + else None + ), + "next_retry_at": ( + self.next_retry_at.astimezone(UTC).isoformat() + if self.next_retry_at is not None + else None + ), + "stale": stale, + } + + +class BalanceHealthTracker: + def __init__( + self, + *, + clock: Callable[[], datetime] | None = None, + base_backoff_sec: float = 15.0, + max_backoff_sec: float = 300.0, + ) -> None: + self._clock = clock or (lambda: datetime.now(tz=UTC)) + self._base_backoff_sec = max(1.0, float(base_backoff_sec)) + self._max_backoff_sec = max(self._base_backoff_sec, float(max_backoff_sec)) + self._lock = Lock() + self._states: dict[str, BalanceWorkerHealth] = {} + + def _now(self) -> datetime: + return self._clock() + + def _get(self, worker: str) -> BalanceWorkerHealth: + return self._states.get( + worker, + BalanceWorkerHealth( + worker=worker, + last_success_at=None, + consecutive_failures=0, + failure_code=None, + failure_at=None, + next_retry_at=None, + ), + ) + + def record_success(self, worker: str, *, captured_at: datetime | None = None) -> str | None: + with self._lock: + prev = self._get(worker) + recovered = prev.consecutive_failures > 0 + now = self._now() + self._states[worker] = BalanceWorkerHealth( + worker=worker, + last_success_at=captured_at or now, + consecutive_failures=0, + failure_code=None, + failure_at=None, + next_retry_at=None, + ) + return "recovered" if recovered else None + + def record_failure(self, worker: str, *, code: str) -> tuple[str | None, float]: + with self._lock: + prev = self._get(worker) + now = self._now() + failures = prev.consecutive_failures + 1 + backoff = min( + self._base_backoff_sec * (2 ** max(0, failures - 1)), + self._max_backoff_sec, + ) + self._states[worker] = BalanceWorkerHealth( + worker=worker, + last_success_at=prev.last_success_at, + consecutive_failures=failures, + failure_code=code, + failure_at=now, + next_retry_at=now + timedelta(seconds=backoff), + ) + lifecycle: str | None = None + if prev.consecutive_failures == 0 or prev.failure_code != code: + lifecycle = "detected" + elif failures in {3, 6, 12}: + lifecycle = "ongoing" + return lifecycle, backoff + + def ready(self, worker: str) -> tuple[bool, float]: + with self._lock: + state = self._get(worker) + now = self._now() + retry_at = state.next_retry_at + if retry_at is None: + return True, 0.0 + remain = (retry_at - now).total_seconds() + if remain <= 0: + return True, 0.0 + return False, remain + + def status(self, worker: str, *, stale_after_sec: float) -> dict[str, object]: + with self._lock: + state = self._get(worker) + now = self._now() + return state.as_dict(now=now, stale_after_sec=stale_after_sec) diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index 297a217..d72a110 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -20,7 +20,7 @@ class BrokerBalanceError(RuntimeError): _CODES = frozenset({ "token_issuance_failed", "authentication_failed", "authorization_failed", - "transport_failed", "malformed_response", "broker_rejected", + "transport_failed", "timeout", "http_5xx", "malformed_response", "broker_rejected", "ownership_mismatch", }) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index dc74257..3e2eab3 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -30,13 +30,16 @@ from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.balance_health import BalanceHealthTracker from trex_core.application.broker_port import ( BalanceItem, + BrokerBalanceError, BrokerPort, CancelOutcome, CancelState, OrderRequest, ) +from trex_core.application.account_snapshot import AccountSnapshotUnavailable from trex_core.application.ledger_outbox import LedgerOutbox from trex_core.application.ledger_port import LedgerTrade from trex_core.application.ledger_skip import publish_ledger_skip_alarm @@ -139,6 +142,8 @@ class ExecutionService: # account snapshot only through this injected reader. Reconciliation # remains wired directly to the live broker. self._sell_dispatch_balance_reader = balance_reader or broker.get_balance + self._balance_health = BalanceHealthTracker(clock=self._clock) + self._balance_worker = "execution" # ------------------------------------------------------------ def run_once(self) -> ExecutionResult: @@ -171,7 +176,22 @@ class ExecutionService: # surfaces via reconcile_core_holdings (REQ-062 A). No mirror # fallback by design. if sell_balance is None: - sell_balance = self._sell_dispatch_balance_map() + ready, backoff_left = self._balance_health.ready(self._balance_worker) + if not ready: + r.sell_skipped_no_balance += 1 + _log.info( + "sell dispatch deferred: broker balance backoff active " + "instance_key=%s worker=%s backoff_left_sec=%.1f", + self._instance_key, + self._balance_worker, + backoff_left, + ) + return + try: + sell_balance = self._sell_dispatch_balance_map() + except (AccountSnapshotUnavailable, BrokerBalanceError): + r.sell_skipped_no_balance += 1 + return bal = sell_balance.get(slot.symbol or "", 0) # dcy-021 / REQ-062 (D4): full sell (slot.qty is None) => broker # balance; partial sell => min(order qty, broker balance). Broker @@ -420,10 +440,60 @@ class ExecutionService: def _sell_dispatch_balance_map(self) -> dict: """Balance used only to size a not-yet-submitted pending sell.""" - return { - b.symbol: int(b.hldg_qty) - for b in self._sell_dispatch_balance_reader() - } + try: + holdings = self._sell_dispatch_balance_reader() + except AccountSnapshotUnavailable as exc: + code = "stale_snapshot" if exc.code == "expired" else ( + getattr(exc, "detail_code", None) or exc.code + ) + lifecycle, backoff = self._balance_health.record_failure( + self._balance_worker, code=code + ) + if lifecycle is not None: + _log.warning( + "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s " + "failure_code=%s consecutive_failures=%d backoff_sec=%.1f", + lifecycle, + self._balance_worker, + self._instance_key, + code, + self._balance_health.status(self._balance_worker, stale_after_sec=600)[ + "consecutive_failures" + ], + backoff, + ) + raise + except BrokerBalanceError as exc: + lifecycle, backoff = self._balance_health.record_failure( + self._balance_worker, code=exc.code + ) + if lifecycle is not None: + _log.warning( + "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s " + "failure_code=%s consecutive_failures=%d backoff_sec=%.1f", + lifecycle, + self._balance_worker, + self._instance_key, + exc.code, + self._balance_health.status(self._balance_worker, stale_after_sec=600)[ + "consecutive_failures" + ], + backoff, + ) + raise + lifecycle = self._balance_health.record_success(self._balance_worker) + if lifecycle == "recovered": + state = self._balance_health.status(self._balance_worker, stale_after_sec=600) + _log.info( + "broker balance recovered: worker=%s instance_key=%s last_success_at=%s", + self._balance_worker, + self._instance_key, + state["last_success_at"], + ) + return {b.symbol: int(b.hldg_qty) for b in holdings} + + def balance_worker_health(self, *, stale_after_sec: float = 600.0) -> dict[str, object]: + return self._balance_health.status(self._balance_worker, stale_after_sec=stale_after_sec) def _resolve_fill_price(self, fs, slot) -> tuple[float | None, bool]: """Resolve the fill price from FillStatus, with limit_price backfill. diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index b6ef0bc..ca50da9 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -16,7 +16,8 @@ from datetime import date, datetime, time, timedelta, timezone from typing import Callable, Optional from trex_core.application.alarm_emitter import AlarmEmitter -from trex_core.application.broker_port import BrokerPort +from trex_core.application.balance_health import BalanceHealthTracker +from trex_core.application.broker_port import BrokerBalanceError, BrokerPort from trex_core.application.portfolio import ( BrokerCredentialPayload, DegradedReason, @@ -95,6 +96,7 @@ class SnapshotScheduler: # REQ-114: entry gate. True = new entries allowed (default). # credential_fetch_enabled=False keeps this always True (no gate). self._entry_allowed: bool = True + self._balance_health = BalanceHealthTracker(clock=self._clock) # ---- callable from a daemon thread loop -------------------------------- def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None: @@ -310,14 +312,18 @@ class SnapshotScheduler: self._store.swap(items) try: # BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning. - reconcile_positions( - store=self._store, - broker=self._broker, - slot_repo=self._slot_repo, - instance_key=self._instance_key, - warn_store=self._warn_store, - alarm_emitter=self._alarm_emitter, - notification_store=self._notification_store, + self._run_balance_worker( + worker="position_reconciliation", + stale_after_sec=24 * 60 * 60, + run=lambda: reconcile_positions( + store=self._store, + broker=self._broker, + slot_repo=self._slot_repo, + instance_key=self._instance_key, + warn_store=self._warn_store, + alarm_emitter=self._alarm_emitter, + notification_store=self._notification_store, + ), ) except Exception: # noqa: BLE001 _log.exception("reconciliation pass failed") @@ -330,18 +336,79 @@ class SnapshotScheduler: try: from trex_core.application.reconciliation import reconcile_core_holdings # BAT-050 / REQ-103: pass warn_store for B-case durable warning. - reconcile_core_holdings( - slot_repo=self._slot_repo, slot_service=self._slot_service, - broker=self._broker, instance_key=self._instance_key, - alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 - warn_store=self._warn_store, # BAT-050 / REQ-103 - notification_store=self._notification_store, # req-0007 / ac-0024 + self._run_balance_worker( + worker="core_holding_reconciliation", + stale_after_sec=24 * 60 * 60, + run=lambda: reconcile_core_holdings( + slot_repo=self._slot_repo, slot_service=self._slot_service, + broker=self._broker, instance_key=self._instance_key, + alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 + warn_store=self._warn_store, # BAT-050 / REQ-103 + notification_store=self._notification_store, # req-0007 / ac-0024 + ), ) except Exception: # noqa: BLE001 _log.exception("core-holding divergence reconcile failed") self._last_fired_date = self._clock() return True + def _run_balance_worker( + self, + *, + worker: str, + stale_after_sec: float, + run: Callable[[], object], + ) -> None: + ready, backoff_left = self._balance_health.ready(worker) + if not ready: + _log.info( + "broker balance worker deferred: worker=%s instance_key=%s backoff_left_sec=%.1f", + worker, + self._instance_key, + backoff_left, + ) + return + try: + run() + except BrokerBalanceError as exc: + lifecycle, backoff = self._balance_health.record_failure(worker, code=exc.code) + if lifecycle is not None: + state = self._balance_health.status(worker, stale_after_sec=stale_after_sec) + _log.warning( + "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s " + "failure_code=%s consecutive_failures=%s backoff_sec=%.1f", + lifecycle, + worker, + self._instance_key, + exc.code, + state["consecutive_failures"], + backoff, + ) + return + lifecycle = self._balance_health.record_success(worker) + if lifecycle == "recovered": + state = self._balance_health.status(worker, stale_after_sec=stale_after_sec) + _log.info( + "broker balance recovered: worker=%s instance_key=%s last_success_at=%s", + worker, + self._instance_key, + state["last_success_at"], + ) + + def balance_worker_health( + self, + *, + stale_after_sec: float = 24 * 60 * 60, + ) -> dict[str, dict[str, object]]: + return { + "position_reconciliation": self._balance_health.status( + "position_reconciliation", stale_after_sec=stale_after_sec + ), + "core_holding_reconciliation": self._balance_health.status( + "core_holding_reconciliation", stale_after_sec=stale_after_sec + ), + } + # ---- asset-centric slot reconcile (dcy-020/REQ-058, BAT-028) ---------- def _reconcile_slots(self, items: list) -> None: universe = { From e3142017c9f1ce0c8970a68d8ba7ece6361b5328 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:34:51 +0000 Subject: [PATCH 03/10] fix(account-snapshot): treat timeout/http5xx as transient fallback failures Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/application/test_account_snapshot.py | 15 +++++++++++++-- trex_core/application/account_snapshot.py | 4 ++-- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/tests/application/test_account_snapshot.py b/tests/application/test_account_snapshot.py index 7a171e5..45332ec 100644 --- a/tests/application/test_account_snapshot.py +++ b/tests/application/test_account_snapshot.py @@ -75,10 +75,11 @@ def test_refresh_persists_independent_holdings_and_cash_capture_times(): assert second.cash_captured_at == first.cash_captured_at -def test_sell_falls_back_only_for_transport_failure_and_within_sell_ttl(): +@pytest.mark.parametrize("error_code", ["transport_failed", "timeout", "http_5xx"]) +def test_sell_falls_back_for_transient_live_read_failures_within_sell_ttl(error_code): broker, _, service, now = _wire() assert service.refresh() - broker.balance_error = "transport_failed" + broker.balance_error = error_code now[0] += timedelta(seconds=601) assert service.holdings_for_sell() == [BalanceItem("005930", 2)] @@ -126,6 +127,16 @@ def test_buy_guard_accounts_for_existing_pending_cash_and_never_renews_age(): ) +@pytest.mark.parametrize("error_code", ["transport_failed", "timeout", "http_5xx"]) +def test_buy_guard_keeps_existing_cash_snapshot_on_transient_cash_refresh_failure(error_code): + broker, _, service, _ = _wire() + assert service.refresh() + broker.cash_error = error_code + + assert service.refresh() is False + assert service.buy_guard(qty=2, limit_price=100_000) is None + + def test_current_account_hash_is_required_on_every_consume(): broker, _, service, _ = _wire() assert service.refresh() diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index c42439b..a31a82b 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -217,7 +217,7 @@ class AccountSnapshotService: try: return self._broker.get_balance() except BrokerBalanceError as exc: - if exc.code != "transport_failed": + if exc.code not in {"transport_failed", "timeout", "http_5xx"}: raise AccountSnapshotUnavailable( "live_read_failed", detail_code=exc.code ) from None @@ -253,7 +253,7 @@ class AccountSnapshotService: reserved_cash: float = 0.0, ) -> Optional[str]: """Return a stable rejection reason, or None when bounded admission passes.""" - if self._last_cash_error not in {None, "transport_failed"}: + if self._last_cash_error not in {None, "transport_failed", "timeout", "http_5xx"}: return "account_snapshot_unavailable" try: snapshot = self._load_current() From 4a2e7dd893322caffcc9f2f33691d86a8c83c4d5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:36:14 +0000 Subject: [PATCH 04/10] fix(snapshot): preserve reconcile failure signaling after health tracking Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- trex_core/application/snapshot_scheduler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index ca50da9..287570a 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -384,7 +384,7 @@ class SnapshotScheduler: state["consecutive_failures"], backoff, ) - return + raise lifecycle = self._balance_health.record_success(worker) if lifecycle == "recovered": state = self._balance_health.status(worker, stale_after_sec=stale_after_sec) From 9c5a01ed676f4cb939e0a5623676bf696b3b74a1 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:37:45 +0000 Subject: [PATCH 05/10] fix(health): preserve transient failure codes through stale snapshot fallback Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/application/test_account_snapshot.py | 12 ++++++++++++ trex_core/application/account_snapshot.py | 9 ++++++++- trex_core/application/execution.py | 8 +++++--- 3 files changed, 25 insertions(+), 4 deletions(-) diff --git a/tests/application/test_account_snapshot.py b/tests/application/test_account_snapshot.py index 45332ec..7152495 100644 --- a/tests/application/test_account_snapshot.py +++ b/tests/application/test_account_snapshot.py @@ -209,3 +209,15 @@ def test_slots_rpc_normalizes_snapshot_store_read_failure(): with pytest.raises(AccountSnapshotUnavailable) as raised: service.holdings_for_slots_rpc() assert raised.value.code == "read_failed" + + +def test_sell_stale_snapshot_preserves_transient_live_failure_detail_code(): + broker, _, service, now = _wire() + assert service.refresh() + broker.balance_error = "timeout" + now[0] += timedelta(seconds=3601) + + with pytest.raises(AccountSnapshotUnavailable) as raised: + service.holdings_for_sell() + assert raised.value.code == "expired" + assert raised.value.detail_code == "timeout" diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index a31a82b..ae90af6 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -214,6 +214,7 @@ class AccountSnapshotService: def holdings_for_sell(self) -> list[BalanceItem]: """Prefer a live balance; fallback only on transient transport failure.""" + transient_code: str | None = None try: return self._broker.get_balance() except BrokerBalanceError as exc: @@ -221,8 +222,14 @@ class AccountSnapshotService: raise AccountSnapshotUnavailable( "live_read_failed", detail_code=exc.code ) from None + transient_code = exc.code snapshot = self._load_current() - age = self._age(snapshot.holdings_captured_at, self._sell_max_age) + try: + age = self._age(snapshot.holdings_captured_at, self._sell_max_age) + except AccountSnapshotUnavailable as exc: + raise AccountSnapshotUnavailable( + exc.code, detail_code=transient_code + ) from None _log.warning( "sell balance using account snapshot: age_sec=%.0f", age, diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 3e2eab3..777037b 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -443,9 +443,11 @@ class ExecutionService: try: holdings = self._sell_dispatch_balance_reader() except AccountSnapshotUnavailable as exc: - code = "stale_snapshot" if exc.code == "expired" else ( - getattr(exc, "detail_code", None) or exc.code - ) + detail_code = getattr(exc, "detail_code", None) + if exc.code == "expired" and detail_code is None: + code = "stale_snapshot" + else: + code = detail_code or exc.code lifecycle, backoff = self._balance_health.record_failure( self._balance_worker, code=code ) From 1f728f4d7dbbd56ac4172d303a6bd264456a6500 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 06:25:28 +0000 Subject: [PATCH 06/10] fix(broker): restore main wiring and add production retry cadence + status surface Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0031.md | 27 ++++ .proofline/identities.json | 2 +- tests/application/test_balance_health.py | 38 +++++ tests/application/test_execution.py | 31 ++++ .../test_slots_handler_broker_quantity.py | 84 +++++++++++ tests/application/test_snapshot_scheduler.py | 141 ++++++++++++++++++ trex_core/application/messaging_handlers.py | 75 ++++++++++ trex_core/application/snapshot_scheduler.py | 91 ++++++----- trex_core/runtime/daemon.py | 11 ++ 9 files changed, 464 insertions(+), 36 deletions(-) create mode 100644 .proofline/criteria/ac-0031.md create mode 100644 tests/application/test_balance_health.py diff --git a/.proofline/criteria/ac-0031.md b/.proofline/criteria/ac-0031.md new file mode 100644 index 0000000..77919e0 --- /dev/null +++ b/.proofline/criteria/ac-0031.md @@ -0,0 +1,27 @@ +--- +id: "ac-0031" +status: active +--- + +# Broker balance failure lifecycle observability and worker-isolated retry + +## Criterion + +`trex-core`는 broker balance snapshot을 사용하는 worker(`execution`, `position_reconciliation`, `core_holding_reconciliation`)별로 독립 상태를 유지해야 한다. 상태는 최소 `state(ok|unavailable|stale)`, `last_success_at`, `consecutive_failures`, sanitized `failure_code`, `next_retry_at`를 포함해야 하며, 한 worker의 성공/실패가 다른 worker의 상태를 덮어쓰면 안 된다. + +balance read failure는 `timeout`, `http_5xx`, `transport_failed`, `authentication_failed`, `authorization_failed`, `malformed_response`, `broker_rejected`, `ownership_mismatch`를 구분해야 하며, validated empty success(`[]`)는 failure로 취급하면 안 된다. live read transient failure 뒤 snapshot fallback이 stale/expired여도 downstream 상태에는 transient failure 분류가 보존되어야 한다. + +daily snapshot cadence와 balance-worker retry cadence는 분리되어야 한다. daily refresh가 이미 수행된 날짜에도 failure worker는 `next_retry_at` 이후에만 재시도되어야 하며(backoff 중 broker read 금지), 성공 시 lifecycle recovery와 정상 cadence로 자동 복귀해야 한다. fail-closed 경계는 유지되어 unavailable/stale snapshot에서 신규 주문/경고 해제 경로가 정상 snapshot으로 해석되어서는 안 된다. + +health tracker의 상태는 process-local로 유지되며 daemon restart 시 초기화된다. restart 이후 첫 failure는 새 lifecycle `detected`로 다시 기록되어야 한다. + +operator-visible status surface는 instance+worker별 상태를 노출해야 하며 account/token/credential/raw broker payload/예외 원문을 포함해서는 안 된다. + +## Verification + +- worker isolation: execution/position/core-holding 중 하나만 실패시켜도 다른 worker의 상태가 변경되지 않음을 검증한다. +- retry cadence: production scheduler path(`run_forever`/`_maybe_tick`)에서 daily refresh 여부와 무관하게 due worker만 backoff deadline 이후 재시도되는지 검증한다. +- backoff/lifecycle: 장기 outage에서 backoff가 cap에 고정되고, detected/ongoing/recovered lifecycle이 중복 경고 없이 전이되는지 검증한다. +- classification/fail-closed: empty success, timeout, HTTP 5xx, auth/credential, malformed, pagination-middle failure를 구분하고 각 worker의 fail-closed 결과와 health state를 검증한다. +- restart semantics: daemon restart(새 tracker 인스턴스) 후 이전 health/backoff가 초기화되고 첫 failure가 detected로 재발행되는지 검증한다. +- sanitization/visibility: operator status 응답에 instance+worker 상태 필드는 노출되지만 민감정보/원시 payload/예외 문자열은 노출되지 않음을 검증한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index eb60750..7ce36f9 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, "next_line_number": 10, - "next_ac_number": 31 + "next_ac_number": 32 } diff --git a/tests/application/test_balance_health.py b/tests/application/test_balance_health.py new file mode 100644 index 0000000..29be582 --- /dev/null +++ b/tests/application/test_balance_health.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +from trex_core.application.balance_health import BalanceHealthTracker + + +def test_long_outage_backoff_is_capped_and_status_stays_valid(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + tracker = BalanceHealthTracker(clock=lambda: now["t"], base_backoff_sec=15, max_backoff_sec=300) + + backoffs: list[float] = [] + for _ in range(20): + lifecycle, backoff = tracker.record_failure("execution", code="http_5xx") + backoffs.append(backoff) + assert lifecycle in {None, "detected", "ongoing"} + now["t"] += timedelta(seconds=301) + + assert max(backoffs) == 300.0 + state = tracker.status("execution", stale_after_sec=600) + assert state["state"] == "unavailable" + assert state["consecutive_failures"] == 20 + assert state["failure_code"] == "http_5xx" + + +def test_restart_resets_process_local_health_and_re_emits_detected(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + before_restart = BalanceHealthTracker(clock=lambda: now["t"]) + before_restart.record_success("execution") + lifecycle, _ = before_restart.record_failure("execution", code="timeout") + assert lifecycle == "detected" + + after_restart = BalanceHealthTracker(clock=lambda: now["t"]) + lifecycle, _ = after_restart.record_failure("execution", code="timeout") + assert lifecycle == "detected" + state = after_restart.status("execution", stale_after_sec=600) + assert state["last_success_at"] is None + assert state["consecutive_failures"] == 1 diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index e5df5c4..9981d51 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -272,6 +272,37 @@ def test_pending_sell_balance_failure_is_fail_closed_with_backoff_and_recovery() assert health["consecutive_failures"] == 0 +def test_pending_sell_health_preserves_non_transient_snapshot_failure_code(): + broker = PaperBroker(fill_after_polls=0) + repo, svc, intents, _ = _wire(broker) + sid = _buy(intents, qty=3).slot_id + ExecutionService( + svc, + repo, + broker, + instance_key=INST, + ).run_once() + svc.sell(sid, actor="t", reason="full") + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=lambda: (_ for _ in ()).throw( + AccountSnapshotUnavailable( + "live_read_failed", + detail_code="authentication_failed", + ) + ), + ) + result = execution.run_once() + assert result.sell_skipped_no_balance == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "unavailable" + assert health["failure_code"] == "authentication_failed" + + def test_partial_sell_qty_is_min_intent_broker(): """dcy-021/REQ-062 (D4): partial sell dispatches min(intent qty, broker); remainder stays holding (broker-authoritative qty), cost basis preserved.""" diff --git a/tests/application/test_slots_handler_broker_quantity.py b/tests/application/test_slots_handler_broker_quantity.py index 7a8d571..1471f67 100644 --- a/tests/application/test_slots_handler_broker_quantity.py +++ b/tests/application/test_slots_handler_broker_quantity.py @@ -38,6 +38,14 @@ class _Reader: return list(self._holdings), self._captured_at +class _HealthReader: + def __init__(self, payload: dict[str, dict[str, object]]) -> None: + self.payload = payload + + def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]: + return self.payload + + class _SlotsReader(Protocol): def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... @@ -61,6 +69,7 @@ class _Broker: def _holding_handler( *, broker_reader: _SlotsReader, + health_reader: _HealthReader | None = None, instance_key: str = "default", portfolio_id: str = "pf-A", symbol: str = "104620", @@ -96,6 +105,7 @@ def _holding_handler( instance_key=instance_key, snapshot_store=store, broker_quantity_reader=broker_reader, + broker_balance_health_reader=health_reader, ) return repo, handler @@ -229,3 +239,77 @@ def test_broker_quantity_reader_is_instance_bound_per_handler(): row_b = handler_b(_req_holding()).payload["slots"][0] assert row_a["qty"] == 440.0 assert row_b["qty"] == 111.0 + + +def test_slots_response_includes_instance_worker_balance_health(): + _, handler = _holding_handler( + broker_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=440)], + captured_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + ), + health_reader=_HealthReader( + { + "execution": { + "state": "unavailable", + "last_success_at": "2026-08-24T11:59:00+00:00", + "consecutive_failures": 2, + "failure_code": "http_5xx", + "next_retry_at": "2026-08-24T12:00:15+00:00", + }, + "position_reconciliation": { + "state": "ok", + "last_success_at": "2026-08-24T12:00:00+00:00", + "consecutive_failures": 0, + "failure_code": None, + "next_retry_at": None, + }, + "core_holding_reconciliation": { + "state": "stale", + "last_success_at": "2026-08-24T08:00:00+00:00", + "consecutive_failures": 0, + "failure_code": None, + "next_retry_at": None, + }, + } + ), + ) + payload = handler(_req_holding()).payload + health = payload["broker_balance_health"] + assert health["execution"]["instance_key"] == "default" + assert health["execution"]["worker"] == "execution" + assert health["execution"]["failure_code"] == "http_5xx" + assert health["execution"]["consecutive_failures"] == 2 + assert health["position_reconciliation"]["state"] == "ok" + assert health["core_holding_reconciliation"]["state"] == "stale" + + +def test_slots_balance_health_surface_sanitizes_unexpected_fields_and_codes(): + _, handler = _holding_handler( + broker_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=440)], + captured_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + ), + health_reader=_HealthReader( + { + "execution": { + "state": "unknown-state", + "last_success_at": "2026-08-24T11:59:00+00:00", + "consecutive_failures": "bad", + "failure_code": "account=12345678 token=secret", + "next_retry_at": "2026-08-24T12:00:15+00:00", + "raw_payload": {"secret": "leak"}, + "error": "Traceback: token", + } + } + ), + ) + execution = handler(_req_holding()).payload["broker_balance_health"]["execution"] + assert execution == { + "instance_key": "default", + "worker": "execution", + "state": "unavailable", + "last_success_at": "2026-08-24T11:59:00+00:00", + "consecutive_failures": 0, + "failure_code": None, + "next_retry_at": "2026-08-24T12:00:15+00:00", + } diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 381e5a2..1ec45fd 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -57,10 +57,12 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch): store = SnapshotStore() projected_broker = object() authoritative_broker = object() + unmanaged_store = object() captured = {} def _capture(**kwargs): captured["broker"] = kwargs["broker"] + captured["unmanaged_position_store"] = kwargs.get("unmanaged_position_store") return {} monkeypatch.setattr( @@ -74,11 +76,13 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch): core_holdings_broker=authoritative_broker, slot_repo=object(), slot_service=SimpleNamespace(instance_key="inst-1"), + unmanaged_position_store=unmanaged_store, ) sched._reconcile_slots = lambda _items: None assert sched.run_once() is True assert captured["broker"] is authoritative_broker + assert captured["unmanaged_position_store"] is unmanaged_store def test_swap_preserves_warning_for_carried_over_portfolio(): @@ -755,3 +759,140 @@ def test_balance_worker_health_isolated_backoff_and_recovery(): state = sched.balance_worker_health(stale_after_sec=86_400) assert state["position_reconciliation"]["state"] == "ok" assert state["position_reconciliation"]["consecutive_failures"] == 0 + + +def test_maybe_tick_retries_due_worker_without_daily_refresh(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class CountingRead: + def __init__(self): + self.calls = 0 + + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + self.calls += 1 + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class SequenceBroker: + def __init__(self, steps): + self.steps = list(steps) + self.calls = 0 + + def get_balance(self): + self.calls += 1 + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=UTC)} + read = CountingRead() + position_broker = SequenceBroker( + [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + ) + core_broker = SequenceBroker([[BalanceItem("X", 0)]]) + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + + sched = SnapshotScheduler( + read_port=read, + store=store, + broker=position_broker, + core_holdings_broker=core_broker, + slot_repo=slot_repo, + slot_service=slot_service, + schedule_kst=datetime(2026, 9, 12, 23, 59).time(), + clock=lambda: now["t"], + ) + + assert sched.run_once() is True + assert read.calls == 1 + assert position_broker.calls == 1 + assert core_broker.calls == 1 + + now["t"] += timedelta(seconds=10) + sched._maybe_tick() + assert read.calls == 1 + assert position_broker.calls == 1 # backoff window still active + assert core_broker.calls == 1 # no pending core failure, so no retry call + + now["t"] += timedelta(seconds=6) + sched._maybe_tick() + assert read.calls == 1 + assert position_broker.calls == 2 # due retry fired without daily snapshot refresh + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "ok" + assert health["position_reconciliation"]["consecutive_failures"] == 0 + assert health["core_holding_reconciliation"]["state"] == "ok" + + +def test_run_forever_retries_due_worker_on_production_loop_path(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + + class LoopRead: + def __init__(self): + self.calls = 0 + + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + self.calls += 1 + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class SequenceBroker: + def __init__(self, steps): + self.steps = list(steps) + self.calls = 0 + + def get_balance(self): + self.calls += 1 + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + class StopAfter: + def __init__(self, waits: int, tick): + self._waits = waits + self._tick = tick + self._count = 0 + + def wait(self, _sec: float): + if self._count >= self._waits: + return True + self._tick(self._count) + self._count += 1 + return False + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=UTC)} + read = LoopRead() + position_broker = SequenceBroker( + [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + ) + sched = SnapshotScheduler( + read_port=read, + store=SnapshotStore(), + broker=position_broker, + schedule_kst=datetime(2026, 9, 12, 23, 59).time(), + clock=lambda: now["t"], + ) + stop = StopAfter( + waits=3, + tick=lambda i: now.__setitem__("t", now["t"] + timedelta(seconds=(10 if i == 0 else 6))), + ) + + sched.run_forever(stop, tick_sec=1.0) + + # Boot refresh once + no extra daily refresh, but one due retry after backoff deadline. + assert read.calls == 1 + assert position_broker.calls == 2 + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "ok" + assert state["position_reconciliation"]["consecutive_failures"] == 0 diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index f525d95..dc8f62f 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -192,6 +192,63 @@ class SlotsBrokerQuantityReader(Protocol): def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... +class SlotsBrokerBalanceHealthReader(Protocol): + def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]: ... + + +_BALANCE_HEALTH_STATES = frozenset({"ok", "unavailable", "stale"}) +_BALANCE_HEALTH_FAILURE_CODES = frozenset({ + "token_issuance_failed", + "authentication_failed", + "authorization_failed", + "transport_failed", + "timeout", + "http_5xx", + "malformed_response", + "broker_rejected", + "ownership_mismatch", + "stale_snapshot", + "account_mismatch", + "not_ready", + "expired", + "live_read_failed", + "read_failed", +}) + + +def _sanitize_balance_worker_status( + *, + instance_key: str, + worker: str, + raw: dict[str, object], +) -> dict[str, object]: + state = str(raw.get("state") or "unavailable") + if state not in _BALANCE_HEALTH_STATES: + state = "unavailable" + try: + consecutive_failures = max(0, int(raw.get("consecutive_failures", 0))) + except (TypeError, ValueError): + consecutive_failures = 0 + failure_code_raw = raw.get("failure_code") + failure_code = ( + str(failure_code_raw) + if isinstance(failure_code_raw, str) + and failure_code_raw in _BALANCE_HEALTH_FAILURE_CODES + else None + ) + last_success_at = raw.get("last_success_at") + next_retry_at = raw.get("next_retry_at") + return { + "instance_key": instance_key, + "worker": worker, + "state": state, + "last_success_at": last_success_at if isinstance(last_success_at, str) else None, + "consecutive_failures": consecutive_failures, + "failure_code": failure_code, + "next_retry_at": next_retry_at if isinstance(next_retry_at, str) else None, + } + + def make_slots_rpc_handler( slot_repo: SlotRepository, *, @@ -205,6 +262,7 @@ def make_slots_rpc_handler( max_active_slots: Optional[int] = None, strategy_state_store=None, broker_quantity_reader: Optional[SlotsBrokerQuantityReader] = None, + broker_balance_health_reader: Optional[SlotsBrokerBalanceHealthReader] = None, ): """Build the RPC handler for `trex.slots.req` (REQ-073/074/116). @@ -437,6 +495,23 @@ def make_slots_rpc_handler( "reason": reason, "message": message, } + if broker_balance_health_reader is not None: + try: + raw_health = broker_balance_health_reader.balance_health_for_slots_rpc() + except Exception: # noqa: BLE001 - status surface must stay available + raw_health = {} + health_payload: dict[str, dict[str, object]] = {} + if isinstance(raw_health, dict): + for worker, raw in raw_health.items(): + if not isinstance(worker, str) or not worker or not isinstance(raw, dict): + continue + health_payload[worker] = _sanitize_balance_worker_status( + instance_key=instance_key, + worker=worker, + raw=raw, + ) + if health_payload: + base_payload["broker_balance_health"] = health_payload if warn_store is not None: base_payload["excluded_slots"] = { "reconcile_quantity_mismatch": excluded_count, diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index 39da01e..c7a5da3 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -117,6 +117,7 @@ class SnapshotScheduler: _log.exception("snapshot scheduler tick failed") def _maybe_tick(self) -> None: + self._maybe_retry_balance_workers() now_kst = self._clock().astimezone(_KST) target = datetime.combine(now_kst.date(), self._schedule_kst, tzinfo=_KST) if now_kst < target: @@ -319,19 +320,7 @@ class SnapshotScheduler: self._store.swap(items) try: # BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning. - self._run_balance_worker( - worker="position_reconciliation", - stale_after_sec=24 * 60 * 60, - run=lambda: reconcile_positions( - store=self._store, - broker=self._broker, - slot_repo=self._slot_repo, - instance_key=self._instance_key, - warn_store=self._warn_store, - alarm_emitter=self._alarm_emitter, - notification_store=self._notification_store, - ), - ) + self._run_position_reconciliation_worker() except Exception: # noqa: BLE001 _log.exception("reconciliation pass failed") if self._slot_repo is not None: @@ -341,34 +330,66 @@ class SnapshotScheduler: _log.exception("slot reconcile pass failed") if self._slot_service is not None: try: - from trex_core.application.reconciliation import reconcile_core_holdings # BAT-050 / REQ-103: pass warn_store for B-case durable warning. -<<<<<<< HEAD - self._run_balance_worker( - worker="core_holding_reconciliation", - stale_after_sec=24 * 60 * 60, - run=lambda: reconcile_core_holdings( - slot_repo=self._slot_repo, slot_service=self._slot_service, - broker=self._broker, instance_key=self._instance_key, - alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 - warn_store=self._warn_store, # BAT-050 / REQ-103 - notification_store=self._notification_store, # req-0007 / ac-0024 - ), -======= - reconcile_core_holdings( - slot_repo=self._slot_repo, slot_service=self._slot_service, - broker=self._core_holdings_broker, instance_key=self._instance_key, - alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 - warn_store=self._warn_store, # BAT-050 / REQ-103 - notification_store=self._notification_store, # req-0007 / ac-0024 - unmanaged_position_store=self._unmanaged_position_store, ->>>>>>> origin/main - ) + self._run_core_holding_reconciliation_worker() except Exception: # noqa: BLE001 _log.exception("core-holding divergence reconcile failed") self._last_fired_date = self._clock() return True + def _run_position_reconciliation_worker(self) -> None: + self._run_balance_worker( + worker="position_reconciliation", + stale_after_sec=24 * 60 * 60, + run=lambda: reconcile_positions( + store=self._store, + broker=self._broker, + slot_repo=self._slot_repo, + instance_key=self._instance_key, + warn_store=self._warn_store, + alarm_emitter=self._alarm_emitter, + notification_store=self._notification_store, + ), + ) + + def _run_core_holding_reconciliation_worker(self) -> None: + from trex_core.application.reconciliation import reconcile_core_holdings + + self._run_balance_worker( + worker="core_holding_reconciliation", + stale_after_sec=24 * 60 * 60, + run=lambda: reconcile_core_holdings( + slot_repo=self._slot_repo, + slot_service=self._slot_service, + broker=self._core_holdings_broker, + instance_key=self._instance_key, + alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 + warn_store=self._warn_store, # BAT-050 / REQ-103 + notification_store=self._notification_store, # req-0007 / ac-0024 + unmanaged_position_store=self._unmanaged_position_store, + ), + ) + + def _has_pending_balance_failure(self, worker: str) -> bool: + status = self._balance_health.status(worker, stale_after_sec=24 * 60 * 60) + return int(status.get("consecutive_failures", 0)) > 0 + + def _maybe_retry_balance_workers(self) -> None: + if self._has_pending_balance_failure("position_reconciliation"): + try: + self._run_position_reconciliation_worker() + except Exception: # noqa: BLE001 + _log.exception("reconciliation retry pass failed") + if ( + self._slot_repo is not None + and self._slot_service is not None + and self._has_pending_balance_failure("core_holding_reconciliation") + ): + try: + self._run_core_holding_reconciliation_worker() + except Exception: # noqa: BLE001 + _log.exception("core-holding divergence retry failed") + def _run_balance_worker( self, *, diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 5091f86..78c5051 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -466,10 +466,21 @@ class Daemon: max_active_slots=self.config.max_active_slots, strategy_state_store=self.strategy_state_store, # dcy-028 C-1 broker_quantity_reader=self.account_snapshot_service, + broker_balance_health_reader=self, ) transport.consume_rpc_forever( topology.slots_queue(self.config.instance_key), handler, self.stop) + def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]: + reconcile = self.snapshot_scheduler.balance_worker_health( + stale_after_sec=24 * 60 * 60, + ) + return { + "execution": self.execution.balance_worker_health(stale_after_sec=600), + "position_reconciliation": reconcile["position_reconciliation"], + "core_holding_reconciliation": reconcile["core_holding_reconciliation"], + } + def _strategy_state_set_loop(self) -> None: # dcy-028 / REQ-080: opaque strategy-state write RPC (entry-version CAS). transport = _build_transport(self.config) From 5eef76d59e83bd4cd3331f3bbea8b526f052c0f6 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 06:28:33 +0000 Subject: [PATCH 07/10] chore(proofline): keep AC-0031 as draft pending user exact approval Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0031.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.proofline/criteria/ac-0031.md b/.proofline/criteria/ac-0031.md index 77919e0..2b5c28b 100644 --- a/.proofline/criteria/ac-0031.md +++ b/.proofline/criteria/ac-0031.md @@ -1,6 +1,6 @@ --- id: "ac-0031" -status: active +status: draft --- # Broker balance failure lifecycle observability and worker-isolated retry From 8b3e7ac7530b563461faf33f27414ac5a688c5da Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 06:40:06 +0000 Subject: [PATCH 08/10] fix(broker): cap outage backoff overflow and expand worker matrix coverage Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0031.md | 4 +- tests/application/test_balance_health.py | 22 +++ tests/application/test_execution.py | 66 +++++++ tests/application/test_snapshot_scheduler.py | 171 +++++++++++++++++++ trex_core/application/balance_health.py | 18 +- 5 files changed, 275 insertions(+), 6 deletions(-) diff --git a/.proofline/criteria/ac-0031.md b/.proofline/criteria/ac-0031.md index 2b5c28b..4d3c0f0 100644 --- a/.proofline/criteria/ac-0031.md +++ b/.proofline/criteria/ac-0031.md @@ -9,7 +9,7 @@ status: draft `trex-core`는 broker balance snapshot을 사용하는 worker(`execution`, `position_reconciliation`, `core_holding_reconciliation`)별로 독립 상태를 유지해야 한다. 상태는 최소 `state(ok|unavailable|stale)`, `last_success_at`, `consecutive_failures`, sanitized `failure_code`, `next_retry_at`를 포함해야 하며, 한 worker의 성공/실패가 다른 worker의 상태를 덮어쓰면 안 된다. -balance read failure는 `timeout`, `http_5xx`, `transport_failed`, `authentication_failed`, `authorization_failed`, `malformed_response`, `broker_rejected`, `ownership_mismatch`를 구분해야 하며, validated empty success(`[]`)는 failure로 취급하면 안 된다. live read transient failure 뒤 snapshot fallback이 stale/expired여도 downstream 상태에는 transient failure 분류가 보존되어야 한다. +balance read failure는 `timeout`, `http_5xx`, `transport_failed`, `authentication_failed`, `token_issuance_failed`(credential failure), `authorization_failed`, `malformed_response`, `broker_rejected`, `ownership_mismatch`를 구분해야 하며, validated empty success(`[]`)는 failure로 취급하면 안 된다. live read transient failure 뒤 snapshot fallback이 stale/expired여도 downstream 상태에는 transient failure 분류가 보존되어야 한다. daily snapshot cadence와 balance-worker retry cadence는 분리되어야 한다. daily refresh가 이미 수행된 날짜에도 failure worker는 `next_retry_at` 이후에만 재시도되어야 하며(backoff 중 broker read 금지), 성공 시 lifecycle recovery와 정상 cadence로 자동 복귀해야 한다. fail-closed 경계는 유지되어 unavailable/stale snapshot에서 신규 주문/경고 해제 경로가 정상 snapshot으로 해석되어서는 안 된다. @@ -22,6 +22,6 @@ operator-visible status surface는 instance+worker별 상태를 노출해야 하 - worker isolation: execution/position/core-holding 중 하나만 실패시켜도 다른 worker의 상태가 변경되지 않음을 검증한다. - retry cadence: production scheduler path(`run_forever`/`_maybe_tick`)에서 daily refresh 여부와 무관하게 due worker만 backoff deadline 이후 재시도되는지 검증한다. - backoff/lifecycle: 장기 outage에서 backoff가 cap에 고정되고, detected/ongoing/recovered lifecycle이 중복 경고 없이 전이되는지 검증한다. -- classification/fail-closed: empty success, timeout, HTTP 5xx, auth/credential, malformed, pagination-middle failure를 구분하고 각 worker의 fail-closed 결과와 health state를 검증한다. +- classification/fail-closed: empty success, timeout, HTTP 5xx, auth/credential(`authentication_failed`, `token_issuance_failed`), malformed, pagination-middle failure를 구분하고 각 worker의 fail-closed 결과와 health state를 검증한다. - restart semantics: daemon restart(새 tracker 인스턴스) 후 이전 health/backoff가 초기화되고 첫 failure가 detected로 재발행되는지 검증한다. - sanitization/visibility: operator status 응답에 instance+worker 상태 필드는 노출되지만 민감정보/원시 payload/예외 문자열은 노출되지 않음을 검증한다. diff --git a/tests/application/test_balance_health.py b/tests/application/test_balance_health.py index 29be582..e7ed80a 100644 --- a/tests/application/test_balance_health.py +++ b/tests/application/test_balance_health.py @@ -23,6 +23,28 @@ def test_long_outage_backoff_is_capped_and_status_stays_valid(): assert state["failure_code"] == "http_5xx" +def test_long_outage_over_1100_failures_never_overflows_and_stays_capped(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + tracker = BalanceHealthTracker(clock=lambda: now["t"], base_backoff_sec=15, max_backoff_sec=300) + + last_lifecycle = None + for i in range(1_100): + lifecycle, backoff = tracker.record_failure("execution", code="http_5xx") + last_lifecycle = lifecycle + if i >= 5: + assert backoff == 300.0 + else: + assert backoff <= 300.0 + now["t"] += timedelta(seconds=301) + + state = tracker.status("execution", stale_after_sec=600) + assert state["state"] == "unavailable" + assert state["consecutive_failures"] == 1_100 + assert state["failure_code"] == "http_5xx" + assert state["next_retry_at"] is not None + assert last_lifecycle in {None, "ongoing"} + + def test_restart_resets_process_local_health_and_re_emits_detected(): now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} before_restart = BalanceHealthTracker(clock=lambda: now["t"]) diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 9981d51..6037f9a 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -7,6 +7,8 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta +import pytest + from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.account_snapshot import AccountSnapshotUnavailable @@ -303,6 +305,70 @@ def test_pending_sell_health_preserves_non_transient_snapshot_failure_code(): assert health["failure_code"] == "authentication_failed" +@pytest.mark.parametrize( + "error_code", + [ + "timeout", + "http_5xx", + "authentication_failed", + "token_issuance_failed", + "malformed_response", + ], + ids=["timeout", "http_5xx", "auth", "token", "pagination_middle_malformed"], +) +def test_execution_wrapper_classification_fail_closed_matrix(error_code): + broker = PaperBroker(fill_after_polls=0) + repo, svc, intents, _ = _wire(broker) + sid = _buy(intents, qty=3).slot_id + ExecutionService(svc, repo, broker, instance_key=INST).run_once() + svc.sell(sid, actor="t", reason="full") + + calls = {"n": 0} + + def fail_reader(): + calls["n"] += 1 + raise AccountSnapshotUnavailable("live_read_failed", detail_code=error_code) + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=fail_reader, + ) + result = execution.run_once() + assert result.dispatched == 0 + assert result.sell_skipped_no_balance == 1 + assert calls["n"] == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "unavailable" + assert health["consecutive_failures"] == 1 + assert health["failure_code"] == error_code + + +def test_execution_wrapper_treats_validated_empty_success_as_success_not_failure(): + broker = PaperBroker(fill_after_polls=0) + repo, svc, intents, _ = _wire(broker) + sid = _buy(intents, qty=3).slot_id + ExecutionService(svc, repo, broker, instance_key=INST).run_once() + svc.sell(sid, actor="t", reason="full") + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=lambda: [], + ) + result = execution.run_once() + assert result.dispatched == 0 + assert result.sell_skipped_no_balance == 1 # broker flat is valid success + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "ok" + assert health["consecutive_failures"] == 0 + assert health["failure_code"] is None + + def test_partial_sell_qty_is_min_intent_broker(): """dcy-021/REQ-062 (D4): partial sell dispatches min(intent qty, broker); remainder stays holding (broker-authoritative qty), cost basis preserved.""" diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 1ec45fd..067956f 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -5,6 +5,8 @@ from datetime import UTC, datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import MagicMock +import pytest + from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.portfolio.stub_read import StubPortfolioRead from trex_core.application.alarm_emitter import AlarmEmitter @@ -896,3 +898,172 @@ def test_run_forever_retries_due_worker_on_production_loop_path(): state = sched.balance_worker_health(stale_after_sec=86_400) assert state["position_reconciliation"]["state"] == "ok" assert state["position_reconciliation"]["consecutive_failures"] == 0 + + +@pytest.mark.parametrize( + "error_code", + [ + "timeout", + "http_5xx", + "authentication_failed", + "token_issuance_failed", + "malformed_response", + ], + ids=["timeout", "http_5xx", "auth", "token", "pagination_middle_malformed"], +) +def test_position_worker_classification_matrix_is_fail_closed(error_code): + from trex_core.application.broker_port import BrokerBalanceError + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class ReadOnce: + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=1)] + + class RaiseOrListBroker: + def __init__(self, failure: str | None): + self.failure = failure + + def get_balance(self): + if self.failure is not None: + raise BrokerBalanceError(self.failure) + return [] + + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + store.swap([PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=1)]) + store.set_warning("pf-X", True) + sched = SnapshotScheduler( + read_port=ReadOnce(), + store=store, + broker=RaiseOrListBroker(error_code), + core_holdings_broker=RaiseOrListBroker(None), + slot_repo=slot_repo, + slot_service=slot_service, + ) + + assert sched.run_once() is True + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "unavailable" + assert health["position_reconciliation"]["failure_code"] == error_code + assert health["position_reconciliation"]["consecutive_failures"] == 1 + assert health["core_holding_reconciliation"]["state"] == "ok" + # fail-closed on failure: warning clear path did not run. + assert store.get("pf-X").warning is True + + +def test_position_worker_validated_empty_success_does_not_create_failure_state(): + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class ReadOnce: + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class EmptyBalanceBroker: + def get_balance(self): + return [] + + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + sched = SnapshotScheduler( + read_port=ReadOnce(), + store=store, + broker=EmptyBalanceBroker(), + core_holdings_broker=EmptyBalanceBroker(), + slot_repo=slot_repo, + slot_service=slot_service, + ) + + assert sched.run_once() is True + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "ok" + assert health["position_reconciliation"]["consecutive_failures"] == 0 + assert health["position_reconciliation"]["failure_code"] is None + + +@pytest.mark.parametrize( + "error_code", + [ + "timeout", + "http_5xx", + "authentication_failed", + "token_issuance_failed", + "malformed_response", + ], + ids=["timeout", "http_5xx", "auth", "token", "pagination_middle_malformed"], +) +def test_core_worker_classification_matrix_is_fail_closed(error_code): + from trex_core.application.broker_port import BrokerBalanceError + from trex_core.domain.slot import SlotService, SlotState + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class ReadOnce: + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class EmptyBalanceBroker: + def get_balance(self): + return [] + + class ErrorBroker: + def __init__(self, failure: str): + self.failure = failure + + def get_balance(self): + raise BrokerBalanceError(self.failure) + + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + slot = slot_repo.find_slot_by_portfolio("default", "pf-X") + pending = slot_service.transit_slot( + slot.slot_id, + SlotState.INACTIVE, + SlotState.PENDING_BUY, + reason="buy", + actor="test", + symbol="X", + ) + slot_service.transit_slot( + pending.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="filled", + actor="test", + avg_price=100.0, + qty=1.0, + ) + before = slot_repo.fetch_slot(slot.slot_id) + sched = SnapshotScheduler( + read_port=ReadOnce(), + store=SnapshotStore(), + broker=EmptyBalanceBroker(), + core_holdings_broker=ErrorBroker(error_code), + slot_repo=slot_repo, + slot_service=slot_service, + ) + + assert sched.run_once() is True + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "ok" + assert health["core_holding_reconciliation"]["state"] == "unavailable" + assert health["core_holding_reconciliation"]["failure_code"] == error_code + assert health["core_holding_reconciliation"]["consecutive_failures"] == 1 + # fail-closed on failure: no slot mutation on this failed core reconcile. + after = slot_repo.fetch_slot(slot.slot_id) + assert after.status is before.status + assert after.qty == before.qty diff --git a/trex_core/application/balance_health.py b/trex_core/application/balance_health.py index 37194d5..adbbaea 100644 --- a/trex_core/application/balance_health.py +++ b/trex_core/application/balance_health.py @@ -1,6 +1,7 @@ """Broker-balance worker health tracking with bounded backoff.""" from __future__ import annotations +import math from dataclasses import dataclass from datetime import UTC, datetime, timedelta from threading import Lock @@ -64,6 +65,11 @@ class BalanceHealthTracker: self._max_backoff_sec = max(self._base_backoff_sec, float(max_backoff_sec)) self._lock = Lock() self._states: dict[str, BalanceWorkerHealth] = {} + if self._base_backoff_sec >= self._max_backoff_sec: + self._max_backoff_exponent = 0 + else: + ratio = self._max_backoff_sec / self._base_backoff_sec + self._max_backoff_exponent = int(math.floor(math.log2(ratio))) def _now(self) -> datetime: return self._clock() @@ -101,10 +107,14 @@ class BalanceHealthTracker: prev = self._get(worker) now = self._now() failures = prev.consecutive_failures + 1 - backoff = min( - self._base_backoff_sec * (2 ** max(0, failures - 1)), - self._max_backoff_sec, - ) + exponent = max(0, failures - 1) + if exponent > self._max_backoff_exponent: + backoff = self._max_backoff_sec + else: + backoff = min( + self._base_backoff_sec * (2.0 ** exponent), + self._max_backoff_sec, + ) self._states[worker] = BalanceWorkerHealth( worker=worker, last_success_at=prev.last_success_at, From 635a9a9f9fbedbcb760bb86e93723cbe3d4186b4 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 13 Sep 2026 15:49:06 +0900 Subject: [PATCH 09/10] fix(snapshot): avoid duplicate scheduled balance retry --- tests/application/test_snapshot_scheduler.py | 40 ++++++++++++++++++++ trex_core/application/snapshot_scheduler.py | 13 ++++--- 2 files changed, 47 insertions(+), 6 deletions(-) diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 067956f..3411d6f 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -834,6 +834,46 @@ def test_maybe_tick_retries_due_worker_without_daily_refresh(): assert health["core_holding_reconciliation"]["state"] == "ok" +def test_maybe_tick_daily_pass_does_not_duplicate_due_worker_retry(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + + class SequenceBroker(PaperBroker): + def __init__(self): + super().__init__() + self.steps = [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + self.calls = 0 + + def get_balance(self): + self.calls += 1 + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=UTC)} + broker = SequenceBroker() + sched = SnapshotScheduler( + read_port=_make_stub(), + store=SnapshotStore(), + broker=broker, + schedule_kst=datetime(2026, 9, 12, 0, 0).time(), + clock=lambda: now["t"], + ) + + assert sched.run_once() is True + assert broker.calls == 1 + + # On the next day's scheduled tick, both the daily pass and the failed + # worker retry are due. The daily pass must satisfy both with one read. + now["t"] += timedelta(days=1) + sched._maybe_tick() + + assert broker.calls == 2 + assert broker.steps == [] + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "ok" + + def test_run_forever_retries_due_worker_on_production_loop_path(): from trex_core.application.broker_port import BalanceItem, BrokerBalanceError diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index c7a5da3..547f70a 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -117,15 +117,16 @@ class SnapshotScheduler: _log.exception("snapshot scheduler tick failed") def _maybe_tick(self) -> None: - self._maybe_retry_balance_workers() now_kst = self._clock().astimezone(_KST) target = datetime.combine(now_kst.date(), self._schedule_kst, tzinfo=_KST) - if now_kst < target: - return last = self._last_fired_date - if last is not None and last.date() == now_kst.date(): - return # already fired today - self.run_once() + daily_due = now_kst >= target and (last is None or last.date() != now_kst.date()) + if daily_due: + # The daily pass already invokes every eligible balance worker. + # Do not run a due retry first and duplicate the same broker read. + self.run_once() + return + self._maybe_retry_balance_workers() # ---- entry-allowed gate (REQ-114) ------------------------------------ def get_entry_allowed(self) -> bool: From 8577f54ae66d3a1b125894eb78e0e273d8089daa Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 13 Sep 2026 16:07:19 +0900 Subject: [PATCH 10/10] chore(proofline): activate approved AC-0031 --- .proofline/criteria/ac-0031.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.proofline/criteria/ac-0031.md b/.proofline/criteria/ac-0031.md index 4d3c0f0..fbb4349 100644 --- a/.proofline/criteria/ac-0031.md +++ b/.proofline/criteria/ac-0031.md @@ -1,6 +1,6 @@ --- id: "ac-0031" -status: draft +status: active --- # Broker balance failure lifecycle observability and worker-isolated retry