From a717cdfea3d82a0c6278fa08ed66aeccadbbaeea Mon Sep 17 00:00:00 2001
From: Codex <codex@local>
Date: Thu, 7 May 2026 00:22:20 +0800
Subject: [PATCH] Harden market truth fallback failures

---
 .../market_state/market_truth_supervisor.py   | 58 +++++++++++++------
 tests/unit/test_market_truth_runtime.py       | 22 +++++++
 2 files changed, 62 insertions(+), 18 deletions(-)

diff --git a/src/services/market_state/market_truth_supervisor.py b/src/services/market_state/market_truth_supervisor.py
index 1c0e6c8..db10daa 100644
--- a/src/services/market_state/market_truth_supervisor.py
+++ b/src/services/market_state/market_truth_supervisor.py
@@ -143,11 +143,22 @@ class MarketTruthSupervisor:
         while not self._stop_event.is_set():
             now = datetime.now(timezone.utc)
             for subscription in self.stream_service.iter_subscriptions():
-                self.current_state(
-                    instrument_id=subscription.instrument_id,
-                    timeframe=subscription.timeframe,
-                    now=now,
-                )
+                try:
+                    self.current_state(
+                        instrument_id=subscription.instrument_id,
+                        timeframe=subscription.timeframe,
+                        now=now,
+                    )
+                except Exception as exc:
+                    self._record_incident(
+                        instrument_id=subscription.instrument_id,
+                        timeframe=subscription.timeframe,
+                        incident_type=INCIDENT_TYPE_STREAM,
+                        incident_code="market_truth_supervisor_error",
+                        status=INCIDENT_STATUS_OPEN,
+                        observed_ts=now,
+                        details={"error": str(exc)},
+                    )
             self._get_news_snapshot(now=now, force_refresh=False)
             for _ in range(max(self.interval_seconds, 1)):
                 if self._stop_event.is_set():
@@ -188,19 +199,30 @@ class MarketTruthSupervisor:
         if not symbol:
             return
         inst_id = build_okx_inst_id(symbol)
-        rows = self.rest_client.get_candles(inst_id=inst_id, bar=timeframe, limit=self.fallback_limit)
-        normalized = normalize_okx_candles(symbol=symbol, venue="OKX", timeframe=timeframe, candles=rows)
-        confirmed = [item for item in normalized if item.confirm]
-        written = self.candle_repository.upsert_many(confirmed, instrument_id=instrument_id)
-        self._record_incident(
-            instrument_id=instrument_id,
-            timeframe=timeframe,
-            incident_type=INCIDENT_TYPE_FALLBACK,
-            incident_code="rest_backfill_applied",
-            status=INCIDENT_STATUS_RESOLVED if written else INCIDENT_STATUS_OPEN,
-            observed_ts=now,
-            details={"symbol": symbol, "inst_id": inst_id, "timeframe": timeframe, "fetched": len(rows), "written": written},
-        )
+        try:
+            rows = self.rest_client.get_candles(inst_id=inst_id, bar=timeframe, limit=self.fallback_limit)
+            normalized = normalize_okx_candles(symbol=symbol, venue="OKX", timeframe=timeframe, candles=rows)
+            confirmed = [item for item in normalized if item.confirm]
+            written = self.candle_repository.upsert_many(confirmed, instrument_id=instrument_id)
+            self._record_incident(
+                instrument_id=instrument_id,
+                timeframe=timeframe,
+                incident_type=INCIDENT_TYPE_FALLBACK,
+                incident_code="rest_backfill_applied",
+                status=INCIDENT_STATUS_RESOLVED if written else INCIDENT_STATUS_OPEN,
+                observed_ts=now,
+                details={"symbol": symbol, "inst_id": inst_id, "timeframe": timeframe, "fetched": len(rows), "written": written},
+            )
+        except Exception as exc:
+            self._record_incident(
+                instrument_id=instrument_id,
+                timeframe=timeframe,
+                incident_type=INCIDENT_TYPE_FALLBACK,
+                incident_code="rest_backfill_failed",
+                status=INCIDENT_STATUS_OPEN,
+                observed_ts=now,
+                details={"symbol": symbol, "inst_id": inst_id, "timeframe": timeframe, "error": str(exc)},
+            )
 
     def _get_news_snapshot(self, *, now: datetime, force_refresh: bool = False) -> NewsProviderSnapshot:
         if not settings.news_provider_enabled:
diff --git a/tests/unit/test_market_truth_runtime.py b/tests/unit/test_market_truth_runtime.py
index 5f575df..a5d7c70 100644
--- a/tests/unit/test_market_truth_runtime.py
+++ b/tests/unit/test_market_truth_runtime.py
@@ -122,6 +122,23 @@ class MarketTruthSupervisorTests(unittest.TestCase):
         self.assertTrue(any(item["incident_code"] == "rest_backfill_applied" for item in repository.incidents))
         self.assertTrue(any(item["incident_code"] == STREAM_CODE_DISCONNECTED for item in repository.incidents))
 
+    def test_supervisor_keeps_running_when_rest_fallback_fails(self) -> None:
+        now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc)
+        repository = _SupervisorRepository(now=now)
+        supervisor = MarketTruthSupervisor(
+            repository=repository,
+            candle_repository=_FallbackCandleRepository(repository),
+            stream_service=_DisconnectedStreamService(now=now),
+            rest_client=_FailingOKXClient(),
+            news_provider=_InactiveNewsProvider(now=now),
+            enabled=False,
+        )
+
+        state = supervisor.current_state(instrument_id=1, timeframe="1m", now=now)
+
+        self.assertEqual(state.stream_state["code"], STREAM_CODE_DISCONNECTED)
+        self.assertTrue(any(item["incident_code"] == "rest_backfill_failed" for item in repository.incidents))
+
     def test_market_state_service_uses_supervisor_runtime_state(self) -> None:
         now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc)
         repository = _SupervisorRepository(now=now)
@@ -254,6 +271,11 @@ class _FallbackOKXClient:
         ]
 
 
+class _FailingOKXClient:
+    def get_candles(self, inst_id: str, bar: str, limit: int):
+        raise TimeoutError("rest_timeout")
+
+
 class _InactiveNewsProvider:
     def __init__(self, now: datetime) -> None:
         self.now = now