fix(broker): restore main wiring and add production retry cadence + status surface

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

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

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

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",
+    }

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