monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x138ba2210>

    @pytest.mark.unit
    def test_periodic_router_persists_decision_so_bus_handler_converges(monkeypatch):
        """Bug 175b5b8e49db5cc51935827e355ad010ebc4cb24 — the periodic router and the LIVE bus handler must
        converge on ONE decision per level onset. The seam is ``TideReasoningDecisions``: the periodic
        router (this code path) writes via ``store_reasoning``, and ``_load_level_event_decision`` reads
        it back when the bus event fires. With a stubbed reasoner answering LONG, the bus handler's read
        of the SAME (trading_day, event_et, kind) MUST return the same LONG."""
        from rtrader.services import market_state_etf as mse
        from rtrader.services.signal_sim_engine import LEVEL_SUPPORT_BREAK, LEVEL_SUPPORT_RECLAIM
    
        monkeypatch.setattr(mse, "_reasoning_execute_enabled", lambda: True)
        monkeypatch.setattr(mse, "_decision_cutoff_time", lambda: _time(11, 0))
        monkeypatch.setattr(mse, "confluence_score", lambda *a, **k: 3)
        monkeypatch.setattr(mse, "account_short_enabled", lambda nick: True)
    
        # Stub the LIVE reasoner — same seam the bus handler's reason_live_level_events calls.
        from rtrader.services import tide_etf_reasoning as TER
        monkeypatch.setattr(TER, "reason_for_level_event",
                            lambda day, ev, **k: {
                                "decision": "SHORT", "confidence": 0.7, "scale": 1.2,
                                "rationale": "(stub short)", "story": "stub",
                                "pattern": "SUPPORT-BREAK", "parse_failed": False})
    
        # Stub the Mongo path so store_reasoning writes to the in-memory FakeDB the bus handler reads.
        db = _FakeDB()
        monkeypatch.setattr(CEL, "_open_db", lambda: db)
        monkeypatch.setattr(CEL, "_close_db", lambda d: None)
    
        # (1) PERIODIC ROUTER consult — origin='backtest' (this is the SIM path).
        out = CEL._sleeve_consolidated_level_event(
            DAY, datetime(2026, 7, 1, 10, 16, 0, tzinfo=ET), "down", "yuchao",
            signal_source=LEVEL_SUPPORT_BREAK, origin="backtest", dbf=lambda: db)
        assert out["source"] == "llm" and out["decision"] == "SHORT" and out["leg"] == "bear"
    
        # (2) BUS HANDLER read — _load_level_event_decision reads the same TideReasoningDecisions row.
        # live_only=True → returns ONLY origin='live' rows. The backtest row is structurally excluded;
        # the bus handler's own consult (reason_live_level_events) writes the LIVE row independently.
        live_rows = db["TideReasoningDecisions"].find(
            {"trading_day": DAY.isoformat(), "kind": "support_break", "et": "10:16",
             "$or": [{"origin": "live"}, {"origin": {"$exists": False}}]})
>       assert live_rows == [], "backtest row must NOT leak through the live-only filter"
E       AssertionError: backtest row must NOT leak through the live-only filter
E       assert <test_combine...t 0x13c791550> == []
E         
E         Full diff:
E         - []
E         + <test_combined_engine_live._FakeCursor object at 0x13c791550>

tests/services/test_combined_engine_live.py:2763: AssertionError