""" Phase C2b orchestrator tests: degraded mode + reconnect + post-reconnect reconcile + safe-mode behavior. 13 tests: Detection (3): 1. classify_broker_error: ConnectionError -> "transient" 2. classify_broker_error: 401 unauthorized -> "auth" 3. broker exception in scan/manage marks orchestrator degraded Reconnect (2): 4. mid-loop reconnect succeeds -> degraded cleared, reconcile runs 5. mid-loop reconnect fails -> stays degraded Safe mode (3): 6. degraded -> _scan_entries skipped (no opener calls) 7. degraded -> manage EXIT logged "exit_deferred_disconnected" 8. degraded -> watchdog skipped Hard timeout (1): 9. degraded > broker_degraded_max_minutes -> exit code 4 Post-reconnect reconcile (2): 10. reconcile_startup called with post_reconnect=True after recovery 11. reconcile failure post-reconnect doesn't break loop E2E disconnect-defer-resume (1): 12. trade open, brain says EXIT during disconnect (deferred), reconnect, brain re-says EXIT, executes successfully Run: cd ~/apex_v16 python tests/test_orchestrator_phase_c2b.py """ from __future__ import annotations import asyncio import sys from datetime import datetime, timedelta, timezone from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from broker.broker_base import ( BrokerBase, CancelResult, ClosedTrade, Order, OrderResult, Position, ) from core.config import RuntimeConfig, RunMode, AccountKind from core.contracts import ( BrainContext, BrainDecision, BrainName, Direction, EntryDecision, EntryEvalResult, MarketStructure, Regime, TradeAction, TradeEntry, TradeRuntime, ) from orchestrator import ( EXIT_BROKER_UNRECOVERABLE, Orchestrator, classify_broker_error, ) from persistence.state_store import ActiveTrade, SessionState def _ok(label): print(f" ok {label}") # ============================================================ # FAKES # ============================================================ class FakeAI: async def ask(self, prompt, temperature=0.2, max_tokens=None): from brain.ai_client import AIResponse return AIResponse(text=None, error_kind="unknown") class JsonlSink: def __init__(self): self.events = [] def write(self, event, **fields): self.events.append({"event": event, **fields}) class FakeLogger: def __init__(self): import logging self.brain_log = JsonlSink() self.session_log = JsonlSink() self.error_log = JsonlSink() self.system = logging.getLogger("test.c2b") def log_session_event(self, event, **fields): self.session_log.write(event, **fields) def log_error(self, where, error, **extra): self.error_log.write("error", where=where, error=error, **extra) def log_trade_opened(self, **fields): self.brain_log.write("trade_opened", **fields) def log_trade_closed(self, **fields): self.brain_log.write("trade_closed", **fields) class FakeProvider: async def get_bars(self, symbol, timeframe, n): return pd.DataFrame(columns=["open", "high", "low", "close", "volume"]) class _MemoryStore: def save(self, state): pass class FakeBroker(BrokerBase): """Programmable BrokerBase for C2b scenarios.""" name = "FakeBroker" def __init__(self): self.calls: list[str] = [] self.connected = True self.connect_attempts = 0 self.fail_connect = False # Make the named methods raise on next call (one-shot, then auto-clear). self.raise_on_balance: BaseException | None = None self.raise_on_close: BaseException | None = None async def connect(self): self.connect_attempts += 1 if self.fail_connect: return False self.connected = True return True async def disconnect(self): self.calls.append("disconnect") self.connected = False async def is_connected(self): return self.connected async def get_last_price(self, symbol): return 5800.0 async def positions_get(self, symbol=None): self.calls.append("positions_get") return [] async def pending_orders(self, symbol=None): return [] async def recent_trades(self, symbol=None, since=None, limit=50): return [] async def place_market_bracket(self, **kw): self.calls.append("place_market_bracket") return OrderResult(success=True, entry_price=5800.0, sl_price=5790.0, tp_price=5820.0, entry_id="E", stop_id="S", target_id="T") async def place_stop_order(self, symbol, side, contracts, stop_price): return OrderResult(success=True, sl_price=stop_price, stop_id="S", entry_id="S") async def place_limit_order(self, symbol, side, contracts, limit_price): return OrderResult(success=True, tp_price=limit_price, target_id="T", entry_id="T") async def cancel_order(self, symbol, order_id): return CancelResult(success=True) async def cancel_all_for_symbol(self, symbol): return 0 async def close_position(self, symbol, contracts=None): self.calls.append("close_position") if self.raise_on_close is not None: exc = self.raise_on_close self.raise_on_close = None raise exc return OrderResult(success=True) async def partial_close_via_opposite_order( self, symbol, direction, contracts_to_close, residual_contracts, new_sl_price, new_tp_price, old_stop_order_id, old_target_order_id, ): return OrderResult( success=True, entry_id="C-FAKE", stop_id="S2-FAKE", target_id="T2-FAKE", sl_price=new_sl_price, tp_price=new_tp_price, ) async def modify_stop(self, symbol, order_id, new_sl_price): return OrderResult(success=True, sl_price=new_sl_price) async def get_account_balance(self): self.calls.append("get_account_balance") if self.raise_on_balance is not None: exc = self.raise_on_balance self.raise_on_balance = None raise exc return 50000.0 async def fetch_bars(self, symbol, timeframe, n): return pd.DataFrame(columns=["open", "high", "low", "close", "volume"]) class FakeReconciler: def __init__(self, raise_in_reconcile: bool = False): self.startup_calls: list[bool] = [] # records post_reconnect flag self.watchdog_calls = 0 self._raise = raise_in_reconcile async def reconcile_startup(self, *, post_reconnect: bool = False): if self._raise: raise RuntimeError("reconcile boom") self.startup_calls.append(post_reconnect) return None async def watchdog_naked_positions(self): self.watchdog_calls += 1 return None class FakeBrain: """Driver that returns canned BrainDecision per call.""" def __init__(self, action=TradeAction.HOLD.value, reason="canned"): self.action = action self.reason = reason self.calls = 0 async def evaluate_entry(self, symbol, tech, *, bias_data=None, last_entry_eval_time=0.0): return EntryEvalResult(decision=None) async def manage_exit(self, ctx): self.calls += 1 return BrainDecision( action=self.action, reason=self.reason, move_sl_to=None, metadata={}, ) # ============================================================ # FIXTURES # ============================================================ FIXED_DAY_UTC: datetime = datetime(2026, 4, 28, 14, 0, 0, tzinfo=timezone.utc) def make_orch(*, broker=None, reconciler=None, max_iterations=2, brain_dispatch=None, opener=None, closer=None, risk_manager=None, state=None, degraded_max_minutes: int = 15) -> tuple: cfg = RuntimeConfig(mode=RunMode.LIVE, account=AccountKind.INELIGIBLE) cfg.asset_filter = ["MES"] cfg.loop_sleep_seconds = 0 cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 0 cfg.maintenance_loop_interval_seconds = 0 cfg.reconcile_interval_iterations = 1 cfg.broker_degraded_max_minutes = degraded_max_minutes state = state or SessionState() logger = FakeLogger() orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=_MemoryStore(), logger=logger, brain_dispatch=brain_dispatch or {}, opener=opener, closer=closer, risk_manager=risk_manager, broker=broker, reconciler=reconciler, now_utc_provider=lambda: FIXED_DAY_UTC, max_iterations=max_iterations, ) return orch, logger def _minimal_tech() -> "TechSnapshot": """Build a TechSnapshot with safe defaults — used when manage_exit is the system under test, not the indicator computation. We only need build_tech to NOT return None so the action dispatch is reached.""" from analysis.tech_snapshot import TechSnapshot return TechSnapshot( symbol="MES", price=5800.0, open=5798.0, candle_time=0, is_candle_closed=False, candle_age_seconds=0.0, rsi=55.0, rsi_prev=54.0, rsi_h1=52.0, rsi_h4=50.0, atr_m5_points=2.0, atr_ratio=1.0, vol_regime="NORMAL", vol_spike=False, market_structure=MarketStructure.BULLISH_EXPANSION.value, h1_struct_bull=True, h1_struct_bear=False, trend_maturity=2, regime=Regime.TRENDING.value, regime_reason="test", regime_near_trending=[], deviation_pct=0.0, divergence="NONE", macd_decelerating=False, macd_hist_last=0.0, candle_strength=0.5, hammer=False, shooting_star=False, bull_engulfing=False, bear_engulfing=False, doji=False, doji_type=None, piercing=False, dark_cloud=False, morning_star=False, evening_star=False, volume_weak=False, buy_absorption=False, sell_absorption=False, vwap=5800.0, vwap_deviation_pct=0.0, bias="NEUTRO", allowed_direction="BOTH", h1_compatibility=1.0, h1_reason="test", ) def stub_tech(orch): snap = _minimal_tech() async def _fake_tech(_symbol): return snap orch._build_tech = _fake_tech def make_active_trade(symbol="MES", contracts=2) -> ActiveTrade: entry = TradeEntry( symbol=symbol, brain_name=BrainName.TF.value, direction=Direction.BUY.value, contracts=contracts, entry_price=5800.0, sl_price=5790.0, tp_price=5820.0, opened_at=datetime.now(timezone.utc) - timedelta(minutes=5), rsi_m5_at_entry=55.0, rsi_h1_at_entry=52.0, rsi_h4_at_entry=50.0, atr_ratio_at_entry=1.0, market_structure_at_entry=MarketStructure.BULLISH_EXPANSION.value, regime_at_entry=Regime.TRENDING.value, h1_compat_at_entry=1.0, confidence_at_entry=70, entry_order_id="E1", stop_order_id="S1", target_order_id="T1", ) return ActiveTrade(entry=entry, runtime=TradeRuntime()) # ============================================================ # 0. CLOCK INJECTION DEFAULT (regression — DRY surfaced this bug) # ============================================================ def test_now_utc_default_is_real_clock_not_self_reference(): """ Regression: a global replace_all of datetime.now(timezone.utc) in orchestrator.py once turned the default fallback for now_utc_provider into a self-referential lambda (`lambda: self._now_utc()`), causing RecursionError on first call in production. Verify the default returns a real tz-aware datetime, not infinite recursion. """ cfg = RuntimeConfig(mode=RunMode.PAPER, account=AccountKind.INELIGIBLE) cfg.asset_filter = ["MES"] cfg.loop_sleep_seconds = 0 cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 0 cfg.maintenance_loop_interval_seconds = 0 orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=SessionState(), store=_MemoryStore(), logger=FakeLogger(), brain_dispatch={}, broker=None, # explicitly NOT passing now_utc_provider — exercise the default max_iterations=1, ) now = orch._now_utc() assert isinstance(now, datetime), f"_now_utc must return datetime, got {type(now)}" assert now.tzinfo is not None, "_now_utc default must be tz-aware" # Must be close to wall-clock now (within 5s) — proves it's the real clock delta = abs((datetime.now(timezone.utc) - now).total_seconds()) assert delta < 5.0, f"_now_utc default drifted from wall clock: {delta}s" _ok("clock injection default: real UTC tz-aware datetime, no recursion") # ============================================================ # 1-3. DETECTION # ============================================================ def test_classify_broker_error_connection(): e = ConnectionError("WebSocket closed: connection reset") assert classify_broker_error(e) == "transient" _ok("classify: ConnectionError + 'connection reset' -> transient") def test_classify_broker_error_auth(): class Auth401(Exception): pass assert classify_broker_error(Auth401("HTTP 401 unauthorized")) == "auth" _ok("classify: '401 unauthorized' -> auth") def test_broker_exception_marks_degraded(): """A broker raise during _refresh_account_balance routes through _maybe_handle_broker_exception, but balance fetch swallows exception in the existing C1 path. Verify direct exception path: scan with a raising opener marks degraded.""" broker = FakeBroker() broker.raise_on_balance = ConnectionError("ws disconnect") orch, _ = make_orch(broker=broker, max_iterations=1) # _refresh_account_balance catches broker exceptions silently (C1 design). # We exercise the explicit handler instead: orch._maybe_handle_broker_exception("test", ConnectionError("ws drop")) assert orch._broker_degraded is True assert orch._degraded_since is not None _ok("_maybe_handle_broker_exception: ConnectionError -> degraded mode entered") # ============================================================ # 4-5. RECONNECT # ============================================================ def test_mid_loop_reconnect_succeeds(): """Reconnect succeeds -> degraded cleared, reconciler called with post_reconnect=True.""" broker = FakeBroker() rec = FakeReconciler() orch, logger = make_orch(broker=broker, reconciler=rec, max_iterations=2) async def runner(): # Manually mark degraded so iter 1 attempts reconnect. orch._mark_degraded("test", ConnectionError("ws drop")) await orch.run() asyncio.run(runner()) assert orch._broker_degraded is False, "must recover after successful reconnect" assert rec.startup_calls == [True], \ f"reconciler called once with post_reconnect=True, got {rec.startup_calls}" assert any(e["event"] == "broker_degraded_exited" for e in logger.brain_log.events) _ok("mid-loop reconnect succeeds: degraded cleared, post-reconnect reconcile run") def test_mid_loop_reconnect_fails_stays_degraded(): """Reconnect fails -> stays degraded, no reconcile call.""" broker = FakeBroker() broker.fail_connect = True rec = FakeReconciler() orch, _ = make_orch(broker=broker, reconciler=rec, max_iterations=2) async def runner(): orch._mark_degraded("test", ConnectionError("ws drop")) await orch.run() asyncio.run(runner()) assert orch._broker_degraded is True, "must remain degraded after failed reconnect" assert rec.startup_calls == [], "no reconcile call when reconnect failed" _ok("mid-loop reconnect fails: stays degraded, no reconcile attempt") # ============================================================ # 6-8. SAFE MODE # ============================================================ def test_degraded_skips_scan(): """_scan_entries not called while degraded.""" broker = FakeBroker() orch, _ = make_orch(broker=broker, max_iterations=1) orch._mark_degraded("test", ConnectionError("ws drop")) broker.fail_connect = True # don't recover this iter scan_called = {"n": 0} async def _patched_scan(): scan_called["n"] += 1 orch._scan_entries = _patched_scan # type: ignore asyncio.run(orch.run()) assert scan_called["n"] == 0, "_scan_entries must NOT run while degraded" _ok("safe mode: _scan_entries skipped while degraded") def test_degraded_defers_exit_action(): """When manage_exit returns EXIT during degraded, broker.close_position is NOT called.""" broker = FakeBroker() state = SessionState() state.active_trades["MES"] = make_active_trade() brain = FakeBrain(action=TradeAction.EXIT.value, reason="exit-while-down") broker.fail_connect = True # stay degraded orch, logger = make_orch( broker=broker, brain_dispatch={BrainName.TF.value: brain}, state=state, max_iterations=1, ) stub_tech(orch) orch._mark_degraded("test", ConnectionError("ws drop")) asyncio.run(orch.run()) assert "close_position" not in broker.calls, \ "EXIT must be deferred — close_position MUST NOT be called" assert "MES" in state.active_trades, "trade preserved through deferred EXIT" assert any(e["event"] == "exit_deferred_disconnected" for e in logger.brain_log.events) _ok("safe mode: EXIT deferred (close_position NOT called, log emitted)") def test_degraded_skips_watchdog(): """Watchdog not called while degraded.""" broker = FakeBroker() broker.fail_connect = True rec = FakeReconciler() orch, _ = make_orch(broker=broker, reconciler=rec, max_iterations=1) orch._mark_degraded("test", ConnectionError("ws drop")) asyncio.run(orch.run()) assert rec.watchdog_calls == 0, "watchdog must NOT run while degraded" _ok("safe mode: watchdog skipped while degraded") # ============================================================ # 9. HARD TIMEOUT # ============================================================ def test_hard_timeout_exits_with_code_4(): """degraded > broker_degraded_max_minutes -> exit code EXIT_BROKER_UNRECOVERABLE.""" broker = FakeBroker() broker.fail_connect = True orch, _ = make_orch(broker=broker, max_iterations=10, degraded_max_minutes=15) orch._mark_degraded("test", ConnectionError("ws drop")) # Backdate the degraded_since timestamp past the threshold (relative # to the orchestrator's injected clock, not real wall-clock). orch._degraded_since = FIXED_DAY_UTC - timedelta(minutes=20) rc = asyncio.run(orch.run()) assert rc == EXIT_BROKER_UNRECOVERABLE, \ f"expected exit code {EXIT_BROKER_UNRECOVERABLE}, got {rc}" _ok(f"hard timeout: degraded > 15min -> exit code {EXIT_BROKER_UNRECOVERABLE}") # ============================================================ # 10-11. POST-RECONNECT RECONCILE # ============================================================ def test_reconcile_startup_called_with_post_reconnect_kwarg(): """After recovery, reconcile_startup gets post_reconnect=True.""" broker = FakeBroker() rec = FakeReconciler() orch, _ = make_orch(broker=broker, reconciler=rec, max_iterations=1) orch._mark_degraded("startup", ConnectionError("test")) asyncio.run(orch.run()) assert rec.startup_calls == [True] _ok("post-reconnect: reconcile_startup(post_reconnect=True) invoked") def test_post_reconnect_reconcile_failure_doesnt_break_loop(): """If reconciler.reconcile_startup raises after reconnect, loop continues.""" broker = FakeBroker() rec = FakeReconciler(raise_in_reconcile=True) orch, _ = make_orch(broker=broker, reconciler=rec, max_iterations=1) orch._mark_degraded("test", ConnectionError("ws drop")) rc = asyncio.run(orch.run()) assert rc == 0, f"loop must continue after reconcile error, got rc={rc}" assert orch._broker_degraded is False, "still recovered (reconcile is non-fatal)" _ok("post-reconnect reconcile error: loop continues (rc=0)") # ============================================================ # 12. modify_stop FAILURE (atomicity — D-C2b-6) # ============================================================ def test_modify_stop_failure_runtime_unchanged(): """ broker.modify_stop raises -> runtime.current_sl_price NOT updated (broker is authoritative; state lying is worse than missed update). 3-loop arch: drive _manage_open_trades() directly instead of orch.run() so we don't depend on scan/manage/maintenance interleaving. The custom broker overrides positions_get to return a non-empty list so _check_external_close (now run before manage_exit) doesn't short- circuit the flow with state cleanup. """ state = SessionState() at = make_active_trade() at.runtime.current_sl_price = 5790.0 state.active_trades["MES"] = at class _BrokerModifyFail(FakeBroker): async def positions_get(self, symbol=None): self.calls.append("positions_get") return [{"symbol": "MES", "size": 2}] async def modify_stop(self, symbol, order_id, new_sl_price): raise ConnectionError("ws drop during modify_stop") broker = _BrokerModifyFail() brain = FakeBrain(action=TradeAction.MOVE_SL.value, reason="trail") async def _manage_with_move(ctx): return BrainDecision( action=TradeAction.MOVE_SL.value, reason="trail", move_sl_to=5795.0, metadata={}, ) brain.manage_exit = _manage_with_move orch, _logger = make_orch( broker=broker, brain_dispatch={BrainName.TF.value: brain}, state=state, max_iterations=1, ) stub_tech(orch) asyncio.run(orch._manage_open_trades()) assert state.active_trades["MES"].runtime.current_sl_price == 5790.0, \ "modify_stop failure: runtime.current_sl_price MUST stay at original" _ok("modify_stop fails: runtime.current_sl_price unchanged (broker authoritative)") # ============================================================ # 13. E2E DISCONNECT-DEFER-RESUME # ============================================================ def test_e2e_disconnect_defer_resume(): """ Trade open + brain says EXIT during disconnect (deferred), reconnect, brain re-says EXIT, executes successfully -> trade closed. 3-loop arch: drive each phase directly. orch.run() races manage_loop and maintenance_loop; here we sequence the same effects by calling _manage_open_trades() (phase 1), then _maybe_reconnect() + _manage_open_trades() (phase 2). Same invariants, no interleaving. """ from trading.trade_closer import TradeCloser from trading.risk_manager import RiskManager class _BrokerWithPosition(FakeBroker): # Models realistic broker: position is open until close_position is # processed, then positions_get reflects the closed state. The # closer's V17 naked-position probe relies on positions_get # truthfully reporting post-close state. async def positions_get(self, symbol=None): self.calls.append("positions_get") if "close_position" in self.calls: return [] return [{"symbol": "MES", "size": 2}] broker = _BrokerWithPosition() state = SessionState() state.active_trades["MES"] = make_active_trade(contracts=2) rec = FakeReconciler() brain = FakeBrain(action=TradeAction.EXIT.value, reason="exit-now") cfg = RuntimeConfig(mode=RunMode.LIVE, account=AccountKind.INELIGIBLE) cfg.asset_filter = ["MES"] cfg.loop_sleep_seconds = 0 cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 0 cfg.maintenance_loop_interval_seconds = 0 cfg.reconcile_interval_iterations = 0 closer = TradeCloser(broker=broker, is_paper=False, logger=FakeLogger()) risk_manager = RiskManager(cfg, state, logger=FakeLogger()) logger = FakeLogger() orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=_MemoryStore(), logger=logger, brain_dispatch={BrainName.TF.value: brain}, opener=None, closer=closer, risk_manager=risk_manager, broker=broker, reconciler=rec, now_utc_provider=lambda: FIXED_DAY_UTC, max_iterations=1, ) stub_tech(orch) # PHASE 1: degraded; manage tick sees EXIT but defers it. # _check_external_close short-circuits to False on degraded, so the # flow reaches brain.manage_exit and the EXIT branch. orch._mark_degraded("test", ConnectionError("ws drop")) broker.fail_connect = True asyncio.run(orch._manage_open_trades()) assert "MES" in state.active_trades, "phase 1: trade preserved" assert "close_position" not in broker.calls, "phase 1: close NOT attempted" assert any(e["event"] == "exit_deferred_disconnected" for e in logger.brain_log.events), \ "phase 1: exit_deferred_disconnected event must be logged" # PHASE 2: maintenance reconnects, manage executes the close. broker.fail_connect = False asyncio.run(orch._maybe_reconnect()) assert orch._broker_degraded is False, "phase 2: reconnect must recover" asyncio.run(orch._manage_open_trades()) assert "close_position" in broker.calls, \ f"phase 2: close_position must be called, calls={broker.calls}" assert "MES" not in state.active_trades, "phase 2: trade removed after close" _ok("E2E: trade preserved through disconnect, closed cleanly post-reconnect") # ============================================================ # RUN # ============================================================ if __name__ == "__main__": print("test_orchestrator_phase_c2b.py") test_now_utc_default_is_real_clock_not_self_reference() test_classify_broker_error_connection() test_classify_broker_error_auth() test_broker_exception_marks_degraded() test_mid_loop_reconnect_succeeds() test_mid_loop_reconnect_fails_stays_degraded() test_degraded_skips_scan() test_degraded_defers_exit_action() test_degraded_skips_watchdog() test_hard_timeout_exits_with_code_4() test_reconcile_startup_called_with_post_reconnect_kwarg() test_post_reconnect_reconcile_failure_doesnt_break_loop() test_modify_stop_failure_runtime_unchanged() test_e2e_disconnect_defer_resume() print("ALL 14 TESTS PASSED")