"""The RACE HAND-OVER: what the viewer sees when the broadcast moves to the next race.

Moving on was already not a restart — no page reload, no black screen (`live_handover.test.mjs`). But
what replaced the black screen was a ~30 s FROZEN PICTURE, and to a viewer that is worse: the boats,
the race clock and the drone all stopped dead, under a feed that was perfectly healthy.

The cause is a mismatch between what the producer publishes and what the renderer presents. The
renderer presents `frontier − liveLagMs` (a real broadcast delay: it is where the commentary has
settled, and it is the boat-cam's sync clock). A race joined in progress was scored from its whole
elapsed track but PUBLISHED only from the join, so the cursor sat below `feed.startSec`, `stateAt`
clamped, and the same frame was returned for a full delay's worth of wall time.

So the coverage window goes on the wire before the first live tick, and the production layer's
opening rides it — which is the other half of a hand-over a viewer can follow: being told, in words
and on the card, which race this now is and where its clock stands.
"""
import re
import sys
from pathlib import Path

_LIVE = Path(__file__).resolve().parents[2] / "live"
_INGEST = Path(__file__).resolve().parents[3] / "ingest"
_WEB = Path(__file__).resolve().parents[1] / "web"
sys.path.insert(0, str(_LIVE))
sys.path.insert(0, str(_INGEST))

import serve_v3_live as S  # noqa: E402
from live_feed_server import LiveFeedServer  # noqa: E402
from race_queue import (  # noqa: E402
    FINISH_TAIL_S,
    PRESENTATION_LAG_S,
    RESULT_HOLD_S,
    RESULT_VISIBLE_S,
    coverage_window,
    finish_handover_sec,
    programme_meta,
    programme_return_commentary,
    race_number_of,
)
from streaming_score import FINALISE_LAG_S  # noqa: E402


class _Recorder(LiveFeedServer):
    """A feed server that records what was published instead of fanning it out."""

    def __init__(self):
        self.inits: list[dict] = []
        self.deltas: list[dict] = []
        self.results: list[dict] = []

    def publish_init(self, meta, course, teams, params, media=None, video=None):  # noqa: ANN001
        self.inits.append({"meta": meta, "teams": teams})

    def publish_delta(self, sec, standings, events, positions, ais=None, program=None,  # noqa: ANN001
                      marks=None, wind=None, onboard=None):
        self.deltas.append({"sec": sec, "standings": standings, "events": events,
                            "positions": positions, "program": program})

    def publish_result(self, result):  # noqa: ANN001
        self.results.append(result)

    def publish_stale(self, stale, *, reason="", since_s=0.0):  # noqa: ANN001
        pass


def _joined_race(elapsed_s: int, *, race_name="Race 4"):
    """An `_OnAirRace` for a race joined `elapsed_s` seconds after its track began, with a synthetic
    back-fill of that whole elapsed track. Returns (recorder, race)."""
    d = S.seed_demo_store()
    t0 = d["meta"]["trackStartUtcMs"]
    meta = {**d["meta"], "raceName": race_name}
    lat0, lon0 = 59.32, 18.04

    def backfill(_instance, boats, _t0, _t1):
        # A straight drift north, one frame a second — enough that a moving cursor is distinguishable
        # from a clamped one, which is the whole assertion.
        return {str(s): {b: {"lat": lat0 + s * 1e-5, "lon": lon0, "speed": 3.0, "course": 0.0}
                         for b in boats}
                for s in range(elapsed_s + 1)}

    rec = _Recorder()
    race = S._OnAirRace(
        d["store"], rec, instance_id=d["instance_id"], group_id=d["group_id"], boats=d["boats"],
        meta=meta, course=d["course"], team_config=d["team_config"], course_id=d["course_id"],
        join_ms=t0 + elapsed_s * 1000, backfill=backfill, on_log=lambda _m: None, first_race=False)
    return rec, race


# --- the freeze ----------------------------------------------------------------------------------

def test_a_joined_race_publishes_its_coverage_window_before_the_first_live_tick():
    """THE FIX. The renderer's cursor lands `PRESENTATION_LAG_S` behind the join, so those seconds
    have to exist on the wire — otherwise it clamps and the picture holds one frame for the delay."""
    elapsed = 120
    rec, _race = _joined_race(elapsed)
    secs = [d["sec"] for d in rec.deltas]
    assert secs == list(range(elapsed - PRESENTATION_LAG_S, elapsed + 1)), \
        "exactly the coverage window, oldest first — the seconds the viewer is about to watch"


def test_the_published_history_carries_what_the_picture_reads():
    """Positions alone would leave the tiles, the running order and the margin graphic blank for the
    whole window, and the clock without a phase."""
    rec, _race = _joined_race(120)
    assert all(d["positions"] for d in rec.deltas)
    assert all(d["standings"] is not None for d in rec.deltas)
    assert all(d["program"] and d["program"]["state"] in ("countdown", "racing", "result")
               for d in rec.deltas)


def test_the_history_carries_no_commentary():
    """The lines for these same seconds arrive settled, through the normal finalisation path, on the
    first live tick. Emitting them here would dump a backlog into the broadcast all at once."""
    rec, _race = _joined_race(120)
    assert all(d["events"] == [] for d in rec.deltas)


def test_the_history_burst_is_bounded():
    """A race joined twenty minutes in must not dump 1200 deltas onto every connected client."""
    rec, _race = _joined_race(1200)
    assert len(rec.deltas) == PRESENTATION_LAG_S + 1


def test_the_whole_elapsed_track_is_still_scored():
    """Published is not the same as scored: `enrich` is a single-pass whole-feed scorer, so the boats
    would otherwise start at mark 0 mid-course and the standings, legs and finish would all be wrong."""
    elapsed = 120
    _rec, race = _joined_race(elapsed)
    assert len(race.scorer._frames) == elapsed + 1


def test_a_race_joined_at_its_arm_publishes_nothing_and_asks_for_nothing():
    """Zero elapsed: there is no history to publish and none is invented."""
    rec, _race = _joined_race(0)
    assert rec.deltas == []


# --- the announcement ----------------------------------------------------------------------------

def test_the_producer_tells_the_production_layer_where_the_coverage_starts():
    rec, _race = _joined_race(120)
    prog = rec.inits[-1]["meta"]["programme"]
    assert prog["coverageStartSec"] == 120 - PRESENTATION_LAG_S
    assert prog["raceLabel"] == "Race 4"                 # preserves the Race Office's visible title
    assert prog["raceNumber"] == 4                      # off the race office's own label
    assert prog["firstRace"] is False


def test_the_race_number_comes_from_the_race_offices_label():
    assert race_number_of("Race 4") == 4
    assert race_number_of("15317") == 15317            # the paperwork's number, whatever its size
    assert race_number_of("Heat A") is None            # …and never invented
    assert race_number_of(None) is None


def test_the_hand_over_names_the_new_race_and_says_where_its_clock_is():
    """What a viewer is owed at a hand-over: which race this now is, and how long to its gun. Both
    beats are `structure`, so the claim gate never withholds them."""
    rec, race = _joined_race(120)
    positions = {b: {"lat": 59.3213, "lon": 18.04, "speed": 3.0, "course": 0.0}
                 for b in race.boats}
    for sec in range(121, 160):
        race.scorer.push_frame(sec, positions)
    opens = [e for e in race.scorer._events if e["type"] == "race_open"]
    assert opens, "a race going on air must say which race it is"
    assert opens[0]["tSec"] == race.meta["programme"]["coverageStartSec"], \
        "…at the second the viewer's coverage of it begins, not at the race's own zero"
    assert opens[0]["claim"] == "structure"
    assert "Red" in opens[0]["text"] and "Blue" in opens[0]["text"]
    joined = " ".join(e["text"] for e in opens)
    # Where the race clock stands — counting down to a gun that has not gone, or, for a race already
    # sailing, which leg they are on AND how far into it. (The mid-race wording moved from "We join
    # them on the run" to "They are away and on the run, N minutes and M seconds into the race": the
    # leg alone could not distinguish a start just missed from one twelve minutes old.)
    assert re.search(r"to the start|into the race", joined), \
        "…and where the race clock stands: counting down, or how far into a race already sailing"


def test_a_handover_preserves_a_named_race_phase_for_the_card_and_speaker():
    """An official title such as a semi-final is identity, not decoration to be discarded."""
    rec, race = _joined_race(120, race_name="Semi-final 2")
    assert rec.inits[-1]["meta"]["programme"]["raceLabel"] == "Semi-final 2"
    positions = {b: {"lat": 59.3213, "lon": 18.04, "speed": 3.0, "course": 0.0}
                 for b in race.boats}
    for sec in range(121, 160):
        race.scorer.push_frame(sec, positions)
    opens = [e for e in race.scorer._events if e["type"] == "race_open"]
    assert opens and opens[0]["cardTitle"] == "Semi-final 2"
    assert "Semi-final 2" in opens[0]["text"]


def test_a_batch_feed_is_never_framed():
    """The gate that keeps every golden byte-identical: no `meta.programme`, no framing pass."""
    from streaming_score import StreamingScorer
    assert StreamingScorer({}, {}, [])._frame_programme is False
    assert StreamingScorer({"programme": {"coverageStartSec": 3}}, {}, [])._frame_programme is True


# --- the constants that have to agree ------------------------------------------------------------

def test_the_presentation_lag_matches_the_renderers():
    """`PRESENTATION_LAG_S` exists to describe what the RENDERER does. If the two drift, the producer
    publishes the wrong amount of history and the freeze comes back silently."""
    cfg = (_WEB / "config.js").read_text(encoding="utf-8")
    m = re.search(r"liveLagMs\s*:\s*(\d+)", cfg)
    assert m, "config.js no longer declares liveLagMs — find where the renderer's delay moved to"
    assert int(m.group(1)) == PRESENTATION_LAG_S * 1000


def test_the_backlog_covers_the_presentation_lag():
    """The same failure, reached by the other road: a RECONNECTING source (Cloud Run replaces an
    instance every hour) gets only the backlog, so a backlog shorter than the lag freezes it too."""
    assert LiveFeedServer.LIVE_BACKLOG >= PRESENTATION_LAG_S


def test_the_coverage_window_leaves_room_for_commentary_to_settle():
    """Why publishing history is safe at all: by the time the cursor reaches an aired history event,
    that event has had at least `FINALISE_LAG_S` to settle and can no longer be rewritten."""
    assert PRESENTATION_LAG_S > FINALISE_LAG_S


def test_the_result_hold_outlasts_the_presentation_delay():
    """The wrap-up is presented `PRESENTATION_LAG_S` after the scorer calls the winner, so a hold
    shorter than that hands over before the viewer has seen the finish at all."""
    assert RESULT_HOLD_S > PRESENTATION_LAG_S


def test_the_client_holds_the_finish_longer_than_the_producer_waits():
    """`race_queue`'s standing coupling: publish the next race before the client is ready and it lands
    on top of the old race's wrap-up."""
    app = (_WEB / "app.js").read_text(encoding="utf-8")
    m = re.search(r"winnerGraceS\s*:\s*(\d+)", app)
    assert m, "app.js no longer declares winnerGraceS"
    assert int(m.group(1)) > RESULT_HOLD_S


def test_coverage_window_is_the_join_minus_the_lag():
    assert coverage_window(120) == (98, 98)
    assert coverage_window(5) == (0, 0)                 # never before the race's own start
    assert coverage_window(0) == (0, 0)


def test_programme_meta_is_what_both_producers_stamp():
    """One helper, so the simulator and the production producer cannot open a race differently."""
    assert programme_meta(elapsed_s=120, race_name="Race 4", first_race=True) == {
        "coverageStartSec": 120 - PRESENTATION_LAG_S, "raceLabel": "Race 4", "raceNumber": 4,
        "leavingRaceLabel": None, "leavingRaceNumber": None, "firstRace": True,
        "returnCommentary": programme_return_commentary()}
    # A hand-over also carries the race it is LEAVING, so the opening can announce a departure and
    # not only an arrival. Both numbers come from the race office's own names, parsed the same way.
    assert programme_meta(elapsed_s=300, race_name="Race 1007", first_race=False,
                          leaving_race_name="Race 1006") == {
        "coverageStartSec": 300 - PRESENTATION_LAG_S, "raceLabel": "Race 1007", "raceNumber": 1007,
        "leavingRaceLabel": "Race 1006", "leavingRaceNumber": 1006, "firstRace": False,
        "returnCommentary": programme_return_commentary()}


def test_external_return_commentary_has_ten_distinct_localised_variants():
    lines = programme_return_commentary()
    assert len(lines) == 10
    assert len({line["i18n"]["en"] for line in lines}) == 10
    assert all(line["i18n"].get("sv") and line["commentaryRole"] == "lead" for line in lines)


def test_programme_never_attaches_a_course_tour_to_a_race_handover():
    # Course orientation belongs to the paced waiting programme. A race-number cadence made every
    # matching hand-over perform a rapid route sweep while the next fleet was already preparing.
    assert "courseTour" not in programme_meta(
        elapsed_s=120, race_name="Race 6", first_race=False)


def test_a_result_reaches_the_client_so_the_finish_can_be_shown():
    """The winner gallery, the frozen winning margin and the wrap-up all read `feed.result`, and the
    live producer must publish one once the delayed picture reaches the deciding finish."""
    rec, race = _joined_race(10)
    race.scorer._result = {"winner": "RED", "winnerSec": 12}
    race.adapter.poll_once = lambda _seen: {}       # preserve the injected presentation result
    race._last_published_sec = 12 + PRESENTATION_LAG_S - 1
    race.tick()
    assert not rec.results, "a frontier result must not overlay the earlier delayed picture"
    race._last_published_sec += 1
    race.tick()
    assert rec.results and rec.results[0]["winner"] == "RED"
    n = len(rec.results)
    race.tick()
    assert len(rec.results) == n, "published once, not every tick"


def test_finish_handover_is_last_boat_or_penultimate_plus_tail():
    assert finish_handover_sec([101, 108], 4) is None
    assert finish_handover_sec([101, 108, 115], 4) == 115 + FINISH_TAIL_S
    assert finish_handover_sec([101, 108, 115, 122], 4) == 122


def test_structured_finish_progress_waits_for_the_presented_timeline():
    from race_queue import OnAirQueue

    now = [100.0]
    q = OnAirQueue(presentation_lag_s=22, finish_tail_s=15,
                   now_fn=lambda: now[0])
    q.offer("race-1", 90_000)
    q.take_next()
    q.offer("race-2", 200_000)
    result = {"expectedFinishers": 4,
              "finishers": [{"finishSec": 100}, {"finishSec": 105}, {"finishSec": 110}]}
    q.note_finish_progress(result, frontier_sec=146)     # viewer is at 124, before target 125
    assert q.take_next() is None
    q.note_finish_progress(result, frontier_sec=147)     # viewer reaches penultimate + 15
    assert q.take_next() is None, "the winner must remain readable when the cursor first reaches it"
    now[0] += RESULT_VISIBLE_S
    assert q.take_next()[0] == "race-2"


def test_last_finisher_releases_at_her_presented_crossing():
    from race_queue import OnAirQueue

    now = [100.0]
    q = OnAirQueue(presentation_lag_s=22, now_fn=lambda: now[0])
    q.offer("race-1", 90_000)
    q.take_next()
    q.offer("race-2", 200_000)
    result = {"expectedFinishers": 4,
              "finishers": [{"finishSec": 100}, {"finishSec": 105},
                            {"finishSec": 110}, {"finishSec": 116}]}
    q.note_finish_progress(result, frontier_sec=137)
    assert q.take_next() is None
    q.note_finish_progress(result, frontier_sec=138)
    assert q.take_next() is None
    now[0] += RESULT_VISIBLE_S
    assert q.take_next()[0] == "race-2"


def test_an_isolated_last_boat_finish_stays_on_air_for_one_minute():
    from race_queue import ISOLATED_FINISH_HOLD_S, OnAirQueue

    now = [100.0]
    q = OnAirQueue(presentation_lag_s=22, result_visible_s=6, now_fn=lambda: now[0])
    q.offer("race-1", 90_000)
    q.take_next()
    result = {"expectedFinishers": 4,
              "finishers": [{"finishSec": 100}, {"finishSec": 105},
                            {"finishSec": 110}, {"finishSec": 116}]}
    q.note_finish_progress(result, frontier_sec=138)     # viewer reaches the last crossing

    now[0] += ISOLATED_FINISH_HOLD_S - 1
    assert not q.done(), "the completed race remains visible for the full isolated-finish minute"
    assert not q.waiting_view_due()
    now[0] += 1
    assert q.done()
    assert q.waiting_view_due(), "the minute ends on the waiting presentation, never a frozen result"


def test_a_new_race_inside_t_minus_three_replaces_the_isolated_finish():
    from race_queue import OnAirQueue

    now = [100.0]
    q = OnAirQueue(presentation_lag_s=22, result_visible_s=6, now_fn=lambda: now[0])
    q.offer("race-1", 90_000)
    q.take_next()
    q.note_finish_progress(
        {"expectedFinishers": 2,
         "finishers": [{"finishSec": 100}, {"finishSec": 105}]}, frontier_sec=127)

    now[0] += 10
    q.offer("race-2", int((now[0] + 179) * 1000))       # newly armed, just inside T−3
    assert q.take_next()[0] == "race-2", \
        "a close new start takes over directly without showing waiting in between"


def test_an_early_new_arm_yields_to_waiting_then_remains_gated_until_t_minus_three():
    from race_queue import ISOLATED_FINISH_HOLD_S, OnAirQueue

    now = [100.0]
    q = OnAirQueue(presentation_lag_s=22, result_visible_s=6, now_fn=lambda: now[0])
    q.offer("race-1", 90_000)
    q.take_next()
    q.note_finish_progress(
        {"expectedFinishers": 2,
         "finishers": [{"finishSec": 100}, {"finishSec": 105}]}, frontier_sec=127)
    q.offer("race-2", 400_000)                           # T−5 at the presented finish

    now[0] += ISOLATED_FINISH_HOLD_S
    assert q.take_next() is None, "an early arm must not erase the waiting transition"
    assert q.waiting_view_due()
    q.release_to_waiting()
    assert q.take_next() is None, "the waiting view remains until the new start reaches T−3"
    now[0] = 220.0                                       # exactly T−3
    assert q.take_next()[0] == "race-2"


def test_a_race_already_armed_at_the_finish_keeps_the_ordinary_handover():
    from race_queue import RESULT_VISIBLE_S, OnAirQueue

    now = [100.0]
    q = OnAirQueue(presentation_lag_s=22, now_fn=lambda: now[0])
    q.offer("race-1", 90_000)
    q.take_next()
    q.offer("race-2", 400_000)                           # already armed when the last boat finishes
    q.note_finish_progress(
        {"expectedFinishers": 2,
         "finishers": [{"finishSec": 100}, {"finishSec": 105}]}, frontier_sec=127)

    now[0] += RESULT_VISIBLE_S
    assert q.take_next()[0] == "race-2", \
        "the extended minute applies only when no competing race was armed at the finish"


def test_server_preliminary_result_survives_missing_tracking_finish():
    """The persisted result reaches air, at picture time, when the local scorer has no finish."""
    rec, race = _joined_race(10)
    winner = race.teams[0]["teamId"]
    race.scorer._result = None
    race.note_server_closed(winner, race.meta["trackStartUtcMs"] + 12_000)
    race.adapter.poll_once = lambda _seen: {}
    race._last_published_sec = 12 + PRESENTATION_LAG_S - 1
    race.tick()
    assert not rec.results, "the server fallback obeys the same delayed-picture gate"
    race._last_published_sec += 1
    race.tick()
    assert rec.results[-1]["winner"] == winner
    assert rec.results[-1]["source"] == "finish_detector"
    assert rec.results[-1]["presentationEndSec"] == 12


# --- the broadcast follows the regatta in the order it is sailed ---------------------------------

def test_the_next_race_is_the_one_that_started_next():
    """The queue is ordered by GUN, not by the order races happened to be reported in."""
    from race_queue import OnAirQueue
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=lambda: 0)
    for inst, gun in (("r3", 3000), ("r1", 1000), ("r2", 2000)):    # reported out of order
        q.offer(inst, gun)
    assert q.waiting() == ["r1", "r2", "r3"]
    aired = []
    for _ in range(3):
        aired.append(q.take_next()[0])
        q.release()                        # that race finished; the broadcast is free to move on
    assert aired == ["r1", "r2", "r3"]


def test_a_race_armed_out_of_order_still_airs_in_start_order():
    """The Race Deck can arm a race out of order, and a re-arm can move a gun earlier. Either way the
    broadcast follows the water, not the paperwork's arrival order."""
    from race_queue import OnAirQueue
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=lambda: 0)
    q.offer("later", 5000)
    q.offer("earlier", 1000)              # armed afterwards, but starts first
    assert q.take_next()[0] == "earlier"


def test_a_race_handed_back_keeps_its_place_and_is_never_dropped():
    """THE SKIP. A race whose fleet has not resolved yet is handed back — and used to go to the BACK
    of the queue (production) or vanish entirely (the sim), so the broadcast moved on to a race that
    started after it. It must return to its own place and be retried."""
    from race_queue import OnAirQueue
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=lambda: 0)
    q.offer("r1", 1000)
    q.offer("r2", 2000)
    nxt = q.take_next()
    assert nxt[0] == "r1"
    q.release()                            # …caller finds it unpresentable
    q.offer(*nxt)
    assert q.waiting() == ["r1", "r2"], "it goes back to its own place, not behind r2"
    assert q.take_next()[0] == "r1", "…and is retried before the race that started after it"


def test_a_re_arm_supersedes_and_re_files_the_race():
    """A race re-armed to a new gun is one race, filed under its NEW start."""
    from race_queue import OnAirQueue
    q = OnAirQueue(winner_hold_s=0, min_air_s=0)
    q.offer("r1", 1000)
    q.offer("r2", 2000)
    q.offer("r1", 9000)                    # general recall: r1 restarts after r2
    assert q.waiting() == ["r2", "r1"], "one entry per race, at its current gun"


def test_neither_driver_drops_a_race_it_cannot_present_yet():
    """Both producers must hand an unpresentable race BACK to the queue. The sim only released it,
    which popped it and never returned it — a permanent skip.

    ONE deliberate exception since 2026-08-10: a race the plane no longer LISTS (an operator's
    prune under a running producer) is dropped, not re-offered — re-offering it parked a deleted
    race at the head of the start-ordered queue for ever and starved every race behind it. The
    hand-back must therefore be gated on `races.lists(...)`, with `q.drop(...)` on the other arm.

    The same rule reaches the race ON AIR: `closed()` answers False for a row that no longer
    exists, so a race deleted while showing was held for ever (rev 00029's zombie on deleted
    racemgmt-15318) — the follow loop must ask `lists` about the on-air race too and let it go."""
    live = Path(S.__file__).read_text(encoding="utf-8")
    sim = (Path(__file__).resolve().parents[3] / "sim" / "bridge_3d.py").read_text(encoding="utf-8")
    assert "q.release()" in live and "q.offer(instance, gun)" in live
    assert 'getattr(races, "lists", lambda _i: True)(instance)' in live, \
        "the hand-back must ask the race plane whether the race still exists"
    assert "q.drop(instance)" in live, "a vanished race is dropped, never re-offered"
    assert 'getattr(races, "lists", lambda _i: True)(on_air.instance_id)' in live, \
        "the ON-AIR race must be asked about too — closed() cannot see a deletion"
    assert "q.drop(on_air.instance_id)" in live, "a deleted on-air race is let go, never held"
    assert "onair.release()\n                    onair.offer(*nxt)" in sim


def test_a_deleted_race_is_forgotten_not_requeued():
    """`OnAirQueue.drop` — the queue-side half of the vanished-race rule: the entry leaves the
    pending list entirely, and the next race in start order becomes the head."""
    from race_queue import OnAirQueue
    q = OnAirQueue(winner_hold_s=0, min_air_s=0)
    q.offer("deleted", 1000)
    q.offer("real", 2000)
    q.drop("deleted")
    assert q.waiting() == ["real"]
    nxt = q.take_next()
    assert nxt is not None and nxt[0] == "real", "the queue moves on to the race that exists"


def test_the_simulator_runs_the_same_hand_over_policy():
    """The sim is a faithful client: a second copy of these numbers here would mean the tool we verify
    with exercising a policy production no longer has (which `RESULT_HOLD_S` literally was)."""
    src = (Path(__file__).resolve().parents[3] / "sim" / "bridge_3d.py").read_text(encoding="utf-8")
    assert not re.search(r"^RESULT_HOLD_S\s*=", src, re.M), \
        "bridge_3d must IMPORT the hand-over policy from race_queue, never restate it"
    assert "seed_history" in src, "the sim must publish a joined race's coverage window too"


# --- a PRIORITISED race is followed whole -------------------------------------------------------
#
# The race office marks a final, a decider or a medal race, and that race is followed from T−2 to the
# finish. It is the one exception to "a new start never cuts away from the race on air" — and it is
# what turns a complete per-race recording from an accident of nothing else overlapping into an
# intention. The tie-break is the operator's: among prioritised races the LATER gun wins.

def _clock(t: list):
    """A fake `now_fn` reading seconds off a one-element list the test moves."""
    return lambda: t[0]


def test_a_prioritised_race_cuts_in_at_t_minus_two():
    """The whole point. `r1` is on air and nowhere near done; `r2` is flagged and its gun is two
    minutes away, so it takes the air anyway."""
    from race_queue import PRIORITY_LEAD_MS, OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=999, min_air_s=999, now_fn=_clock(t))
    q.offer("r1", 900_000)
    assert q.take_next()[0] == "r1"

    gun2 = 1_300_000                                       # r2's gun, 300 s after "now"
    q.set_priority("r2", True)
    q.offer("r2", gun2)
    assert q.take_next() is None, "still outside T−2 — the race on air keeps the air"

    t[0] = (gun2 - PRIORITY_LEAD_MS) / 1000.0              # exactly T−2
    assert q.take_next()[0] == "r2"
    assert q.on_air == "r2"
    assert q.waiting() == ["r1"], "the interrupted race remains accepted and cannot be lost"


def test_an_unprioritised_overlap_never_cuts_away_from_the_current_finish():
    """The current finish wins; the overlapping start waits while the race remains on air."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=999, min_air_s=999, now_fn=_clock(t))
    q.offer("r1", 900_000)
    q.take_next()
    q.offer("r2", 1_000_000)
    assert q.take_next() is None
    assert q.on_air == "r1"


def test_finished_coverage_joins_the_oldest_queued_overlap_without_skipping_it():
    """A later overlapping gun must not erase a race already accepted into the programme."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=_clock(t))
    q.offer("r1", 700_000)
    q.take_next()
    q.offer("r2", 800_000)
    q.offer("r3", 900_000)
    q.note_result(True)
    assert q.take_next()[0] == "r2"
    assert q.waiting() == ["r3"]
    assert q.on_air == "r2", "the first queued live race must retain its place"


def test_an_ongoing_overlap_is_not_skipped_for_a_future_start():
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=_clock(t))
    q.offer("r1", 700_000)
    q.take_next()
    q.offer("missed", 900_000)
    q.offer("next", 1_100_000)
    q.note_result(True)
    assert q.take_next()[0] == "missed"
    assert q.waiting() == ["next"]


def test_a_fresh_producer_rejoins_the_oldest_open_race_before_a_future_start():
    """Restarting the producer must not turn an armed later race into a skip instruction."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=_clock(t))
    q.offer("old", 700_000)
    q.offer("current", 900_000)
    q.offer("future", 1_100_000)
    assert q.take_next()[0] == "old"
    assert q.waiting() == ["current", "future"]


def test_a_fresh_producer_rejoins_the_oldest_open_race_without_a_future_start():
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=_clock(t))
    q.offer("old", 700_000)
    q.offer("newer", 900_000)
    assert q.take_next()[0] == "old"
    assert q.waiting() == ["newer"]


def test_an_ordinary_start_cannot_cut_a_prioritised_race():
    """A final or decider explicitly selected by the race office remains protected to its finish."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=999, min_air_s=999, now_fn=_clock(t))
    q.set_priority("final", True)
    q.offer("final", 900_000)
    q.take_next()
    q.offer("ordinary", 1_000_000)
    assert q.take_next() is None
    assert q.on_air == "final"


def test_later_gun_wins_the_immediate_cut_between_two_prioritised_races():
    """The later priority wins the immediate picture; every other accepted race remains queued."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=999, min_air_s=999, now_fn=_clock(t))
    for inst, gun in (("early", 1_000_000), ("late", 1_060_000)):
        q.set_priority(inst, True)
        q.offer(inst, gun)
    assert q.take_next()[0] == "late", "both are past T−2; the later gun takes the air"

    # …and the earlier one cannot take it back, even though it is still pending and still flagged.
    assert q.take_next() is None
    assert q.on_air == "late"
    assert q.waiting() == ["early"]


def test_a_prioritised_race_on_air_is_not_cut_by_an_earlier_prioritised_one():
    """The mirror of the rule above, and the case that would otherwise flap: with two flagged races
    both eligible, the queue must not hand the air back and forth between them."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=999, min_air_s=999, now_fn=_clock(t))
    q.set_priority("late", True)
    q.offer("late", 1_060_000)
    assert q.take_next()[0] == "late"

    q.set_priority("early", True)
    q.offer("early", 1_000_000)                            # earlier gun, flagged, past its T−2
    assert q.take_next() is None
    assert q.on_air == "late"


def test_a_free_air_still_gives_the_ordinary_lead_in():
    """T−2 is when priority may INTERRUPT, not a shorter countdown. With nothing on air the flagged
    race goes up the moment it is offered — which the watcher does at the ordinary `PRE_START_MS`."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=0, min_air_s=0, now_fn=_clock(t))
    q.set_priority("r1", True)
    q.offer("r1", 1_180_000)                               # gun is 180 s away: outside T−2
    assert q.take_next()[0] == "r1", "nothing was on air, so nothing had to be cut"


def test_clearing_the_flag_stops_the_protection_at_once():
    """The race office changed its mind. A flag read only at `offer` would honour a decision that had
    already been reversed — so the producer refreshes it every tick."""
    from race_queue import OnAirQueue
    t = [1_000.0]
    q = OnAirQueue(winner_hold_s=999, min_air_s=999, now_fn=_clock(t))
    q.set_priority("r1", True)
    q.offer("r1", 900_000)
    q.take_next()

    q.set_priority("r2", True)
    q.offer("r2", 1_060_000)
    q.set_priority("r2", False)                            # unmarked before its T−2 came round
    assert q.take_next() is None, "no longer flagged, so it waits its turn like any other race"


def test_a_race_goes_on_air_on_the_start_countdown_trigger():
    """The event's full start sequence is programme time: the waiting state yields on ARM, then the
    opening, course guide and line-up start at race-second zero. Camera capture opens on the same
    trigger, but camera availability must not hold the entire programme back."""
    from race_queue import announce_lead_ms

    for pre_start_s in (120, 180, 240, 300, 600):
        assert announce_lead_ms(pre_start_s) == pre_start_s * 1000
    assert announce_lead_ms(240) == 240_000, "a four-minute sequence goes on air at T−4"
    assert announce_lead_ms(180) == 180_000, "a three-minute sequence goes on air at T−3"


def test_no_published_sequence_falls_back_rather_than_refusing():
    """Unlike an ARM, a lead-in has a safe default: joining three minutes out is at worst early or
    late coverage, never a wrong gun. `pre_start_seconds` refuses; this one does not have to."""
    from race_queue import PRE_START_MS, announce_lead_ms

    assert announce_lead_ms(None) == PRE_START_MS
    assert announce_lead_ms(0) == PRE_START_MS


def test_the_watcher_joins_each_race_on_its_own_events_sequence():
    """A broadcast following an operator can see races from more than one event, and their sequences
    need not agree — so the lead is per race, not per producer."""
    from race_queue import StartedRaces

    now = 10_000_000
    rows = [{"instanceId": "four", "officialStartUtcMs": now + 181_000},
            {"instanceId": "three", "officialStartUtcMs": now + 181_000}]
    w = StartedRaces(lambda: rows, now_ms=lambda: now,
                     pre_start_for=lambda inst: 240 if inst == "four" else 180)
    assert [r for r, _g in w.poll()] == ["four"], (
        "the four-minute race has reached its own countdown trigger; the three-minute race has not")

    rows[1]["officialStartUtcMs"] = now + 179_000
    assert [r for r, _g in w.poll()] == ["three"]


def test_the_lead_falls_back_when_the_event_lookup_fails():
    """A read-plane hiccup must not decide when a race airs — it retries on the next poll rather than
    holding the race off the air or throwing inside the producer loop."""
    from race_queue import PRE_START_MS, StartedRaces

    def _boom(_inst):
        raise ConnectionError("read plane is having a moment")

    now = 10_000_000
    rows = [{"instanceId": "r1", "officialStartUtcMs": now + PRE_START_MS - 1_000}]
    w = StartedRaces(lambda: rows, now_ms=lambda: now, pre_start_for=_boom)
    assert [r for r, _g in w.poll()] == ["r1"]


def test_closed_history_never_resolves_event_start_sequences():
    """The live watcher may list a whole race day, but terminal history is not live work.

    REST sequence resolution expands one row into several authenticated reads.  Asking for it before
    checking the stop/state fields blocked the one-second picture loop behind every old heat whenever
    a producer restarted or one of those reads had failed on an earlier pass.
    """
    from race_queue import StartedRaces

    now = 10_000_000
    calls = []
    rows = [
        {"instanceId": f"finished-{i}", "officialStartUtcMs": now - 100_000 - i,
         "officialStopUtcMs": now - 10_000}
        for i in range(160)
    ]
    rows.extend([
        {"instanceId": "abandoned", "officialStartUtcMs": now - 20_000,
         "raceState": "abandoned"},
        {"instanceId": "live", "officialStartUtcMs": now - 1_000},
    ])
    watcher = StartedRaces(lambda: rows, now_ms=lambda: now,
                           pre_start_for=lambda instance: calls.append(instance) or 240)

    assert watcher.poll() == [("live", now - 1_000)]
    assert calls == ["live"], "terminal history must not perform per-race REST resolution"


def test_the_watcher_reports_the_flag_for_every_race_it_can_see():
    """`priorities()` feeds `set_priority` each tick, and it must cover races that did not just start
    — a flag set mid-race is the case it exists for."""
    from race_queue import StartedRaces
    rows = [{"instanceId": "r1", "officialStartUtcMs": 1_000, "priority": True},
            {"instanceId": "r2", "officialStartUtcMs": 2_000}]
    w = StartedRaces(lambda: rows, now_ms=lambda: 10_000)
    w.poll()
    assert w.priorities() == {"r1": True, "r2": False}, "absent reads as false, never as unknown"


def test_the_watcher_preserves_the_race_planes_preliminary_result():
    from race_queue import StartedRaces

    rows = [{"instanceId": "r1", "officialStartUtcMs": 1_000,
             "officialStopUtcMs": 9_000, "preliminaryWinnerTeamId": "team-a"}]
    w = StartedRaces(lambda: rows, now_ms=lambda: 10_000)
    w.poll()
    assert w.preliminary_result("r1") == ("team-a", 9_000)


def test_an_empty_scope_is_no_scope_not_no_races():
    """`--course ""` must mean no course filter, and `--owner ""` no owner filter.

    The Cloud Run producer expands `--course "$VR_COURSE"` in its container CMD, and the documented
    race-day deploy sets the variable EMPTY — argparse then handed `""` to `StartedRaces`, whose
    scope filter tests `is not None`, so every race was dropped before the eligibility test ever
    ran. Every OTHER consumer of the flag uses truthiness, which is why the symptom was so quiet:
    the waiting picture happily showed a nearby unstarted race while `poll()` returned nothing, for
    ever ("Waiting for a race", armed trwc-sim races sailing past unaired). An empty string is no
    scope, never an impossible one."""
    from race_queue import StartedRaces

    now = 1_000_000
    rows = [{"instanceId": "r1", "officialStartUtcMs": now - 60_000,
             "courseId": "trwc-sim-course", "ownerUserId": "trwc-sim"}]
    w = StartedRaces(lambda: rows, course_id="", owner="", lead_ms=0, now_ms=lambda: now)
    assert [r for r, _g in w.poll()] == ["r1"], "an empty scope must not filter anything"
    assert w.course_id is None and w.owner is None, \
        "normalized at the door, so every caller — present and future — is covered"


def test_next_gun_and_programme_transition_are_the_same_arm():
    """The earliest future gun both recalls a guided waiting excursion and releases the programme.
    There is no second, hidden preparatory delay between the operator's trigger and the opening."""
    from race_queue import StartedRaces

    now = 10_000_000
    rows = [{"instanceId": "armed", "officialStartUtcMs": now + 240_000},     # armed at T−4
            {"instanceId": "sailing", "officialStartUtcMs": now - 60_000},    # already racing
            {"instanceId": "stopped", "officialStartUtcMs": now + 300_000,
             "officialStopUtcMs": now - 1_000},
            {"instanceId": "off", "officialStartUtcMs": now + 120_000, "raceState": "abandoned"}]
    w = StartedRaces(lambda: rows, now_ms=lambda: now,
                     pre_start_for=lambda inst: 240 if inst == "armed" else None)
    assert [r for r, _g in w.poll()] == ["sailing", "armed"]
    assert w.next_gun_ms() == now + 240_000, "the same arm is visible to the waiting camera"

    w2 = StartedRaces(lambda: [], now_ms=lambda: now)
    w2.poll()
    assert w2.next_gun_ms() is None, "nothing armed — the tour may fly"