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

@@ -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/예외 문자열은 노출되지 않음을 검증한다.

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"])

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."""

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

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,