_parity_runtime = None
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x11a073fe0>
visibility = 'gap_driven'

    @pytest.mark.unit
    @pytest.mark.parametrize("visibility", ["gap_driven", "pre_built"])
    def test_gap_toggle_on_the_0724_onset_fixture(_parity_runtime, monkeypatch, visibility):
        """THE ACCEPTANCE PIN (bug bf9febcbf266): SAME onset fixture, ONE toggle, ZERO consult latency —
        so the arrival GAP is the ONLY variable (this is the experiment that isolates the TIDE-arrival
        cause from throttle params / stale thresholds / routing differences, which are PINNED constant:
        the freshness horizon is pinned at 5 min here so a settings retune can never silently relabel
        this fixture's 7/6-min-old onsets as fresh — the 2026-07-24 5→7 retune did exactly that).
    
        gap_driven: both onsets sit behind a 7.5-min arrival gap; the first tick that can SEE them
        (10:27, data floor = the 10:26:30 confirming reading) already finds them 7 and 6 min old →
        by_status stale-ignored == 6 (2 onsets x 3 accounts, the live 7/24 outcome). COUNTER-CASE
        (what would refute the gap theory): pre_built on the byte-identical fixture must route them
        FRESH — stale-ignored ABSENT, first onset dry-run on all 3 accounts. If pre_built ALSO staled
        them, the divergence would not be arrival-driven and this test would fail loudly."""
        monkeypatch.setattr(CEL, "periodic_max_onset_age_min", lambda: 5.0)
        loader_calls = []
        seen_as_of = []
        inner = _causal_ctx_builder(_GAP_ONSET_EVENTS)
    
        def spy_builder(d, c, a):
            seen_as_of.append(a)
            return inner(d, c, a)
    
        def spy_loader(d):
            loader_calls.append(d)
            return _gap_arrivals(d)
    
        out = RP.run_router_parity_day(
            DAY, accounts=ACCOUNTS, signals=[TNA_TIDE_SPREAD],
            consult_latency_fn=None, latency_meta={"mode": "zero", "seconds": 0.0},
            sequential=True, cfg=SimConfig(), ctx_builder=spy_builder,
            tide_visibility=visibility, tide_ts_loader=spy_loader, store=False)
        assert out["summary"]["tide_visibility"] == visibility
        tide_rows = [r for r in out["routes"] if r["signal_source"] == TNA_TIDE_SPREAD]
        assert len(tide_rows) == 6                       # 2 onsets x 3 accounts, both arms
        if visibility == "gap_driven":
            assert out["summary"]["by_status"].get("stale-ignored") == 6
            assert all(r["status"] == "stale-ignored" for r in tide_rows)
            assert all(r.get("engine_reason") == "onset-stale" for r in tide_rows)
            assert {float(r["stale_age_min"]) for r in tide_rows} == {7.0, 6.0}
            # The replay READ the MarketTide ts sequence…
            assert loader_calls == [DAY]
            # …and every context the router saw was truncated to an ARRIVED reading's ts — visibility
            # keys off DATA time (the 10:26:30 confirming reading, seconds precision), never the
            # nominal tick minute inside the gap.
            arrival_strs = {ts.strftime("%H:%M:%S") for ts in _gap_arrivals(DAY)}
            assert set(seen_as_of) <= arrival_strs | {"00:00:00"}
>           assert "10:26:30" in seen_as_of
E           AssertionError: assert '10:26:30' in ['10:14:00', '10:15:00', '10:16:00', '10:17:00', '10:18:00', '10:19:00', ...]

../robinhood-pm-bf9febcbf266/tests/services/test_signal_sim_router_parity.py:341: AssertionError