"""The prompt-dispatch gate is global, but browser reads never acquire it."""

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor

import pytest

import chat
import solwebd
from fake_browser import seat_for


def test_first_dispatch_writes_stamp_without_wait(monkeypatch):
    now = [100.0]
    sleeps = []
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "5")
    monkeypatch.setattr(chat.time, "time", lambda: now[0])
    monkeypatch.setattr(chat.time, "sleep", sleeps.append)

    chat._rate_limit_dispatch()

    assert sleeps == []
    assert float((chat.STATE / "dispatch.stamp").read_text()) == 100.0


def test_second_dispatch_waits_remaining_gap_and_emits_event(monkeypatch, capsys):
    now = [100.0]
    sleeps = []
    events = []
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "5")
    monkeypatch.setattr(chat.time, "time", lambda: now[0])

    def sleep(seconds):
        sleeps.append(seconds)
        now[0] += seconds
    monkeypatch.setattr(chat.time, "sleep", sleep)

    chat._rate_limit_dispatch()
    now[0] = 101.0
    chat._rate_limit_dispatch(on_event=events.append)

    assert sleeps == [4.0]
    assert events == [{"type": "ratelimit", "wait_s": 4.0}]
    assert "rate-limit: waiting 4.0s before dispatch" in capsys.readouterr().err
    assert float((chat.STATE / "dispatch.stamp").read_text()) == 105.0


def test_stamp_remains_valid_when_threads_race(monkeypatch):
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "0")
    monkeypatch.setattr(chat.time, "time", lambda: 100.0)

    with ThreadPoolExecutor(max_workers=2) as pool:
        list(pool.map(lambda _: chat._rate_limit_dispatch(), range(2)))

    assert float((chat.STATE / "dispatch.stamp").read_text()) == 100.0


def test_submit_threads_rate_limit_event_through_its_callback(monkeypatch):
    now = [100.0]
    events = []
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "2")
    monkeypatch.setattr(chat.time, "time", lambda: now[0])

    def sleep(seconds):
        now[0] += seconds
    monkeypatch.setattr(chat.time, "sleep", sleep)
    chat._rate_limit_dispatch()
    now[0] = 101.0

    fake = seat_for([])[1]
    chat.submit(fake, on_event=events.append)

    assert events == [{"type": "ratelimit", "wait_s": 1.0}]
    assert fake.clicks == 1


def test_dispatch_lock_wait_cannot_exceed_shared_deadline(monkeypatch):
    lock = chat.STATE / "dispatch.lock"
    chat.STATE.mkdir(parents=True)
    holder = lock.open("w")
    chat.fcntl.flock(holder, chat.fcntl.LOCK_EX)
    try:
        with pytest.raises(chat.ChatError, match="prompt submission deadline expired"):
            chat._rate_limit_dispatch(deadline=chat.time.monotonic())
    finally:
        chat.fcntl.flock(holder, chat.fcntl.LOCK_UN)
        holder.close()


def test_dispatch_wait_cannot_exceed_submission_budget(monkeypatch):
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "75")
    monkeypatch.setattr(chat.time, "time", lambda: 101.0)
    chat.STATE.mkdir(parents=True)
    (chat.STATE / "dispatch.stamp").write_text("100\n")

    with pytest.raises(chat.ChatError, match="prompt submission deadline expired"):
        chat._rate_limit_dispatch(deadline=chat.time.monotonic() + 5)


def test_long_reply_timeout_does_not_extend_pre_submission_budget(monkeypatch):
    captured = []
    monkeypatch.setattr(chat, "_rate_limit_dispatch", lambda deadline, **_: captured.append(deadline - chat.time.monotonic()))
    fake = seat_for([])[1]

    chat.submit(fake, timeout=10_800)

    assert captured[0] == pytest.approx(chat.SUBMISSION_TIMEOUT, abs=0.1)


def test_busy_seat_wait_uses_bounded_submission_budget(monkeypatch):
    captured = []
    pool = solwebd.SeatPool(max_seats=1, session_factory=lambda _slot: None)
    engine = solwebd.Engine(pool)
    monkeypatch.setattr(pool, "acquire_any", lambda **kwargs: captured.append(kwargs) or (_ for _ in ()).throw(chat.ChatError("stop")))

    with pytest.raises(chat.ChatError, match="stop"):
        engine.ask({"prompt": "hello", "timeout": 10_800}, on_event=lambda _event: None)

    assert captured[0]["deadline"] - chat.time.monotonic() == pytest.approx(
        chat.SUBMISSION_TIMEOUT, abs=0.1)


def test_busy_seat_wait_emits_queued_event_and_times_out():
    pool = solwebd.SeatPool(max_seats=1, session_factory=lambda _slot: None)
    assert pool.acquire_any() == 0
    events = []

    with pytest.raises(chat.ChatError, match="prompt submission deadline expired waiting for a browser seat"):
        pool.acquire_any(timeout=0, on_wait=events.append)

    assert events == [{"type": "queued", "wait_s": 0.0}]
    pool.release(0)


def test_timed_out_seat_wait_does_not_leave_phantom_queue_depth(monkeypatch):
    pool = solwebd.SeatPool(max_seats=1, session_factory=lambda _slot: None)
    engine = solwebd.Engine(pool)
    assert pool.acquire_any() == 0
    monkeypatch.setattr(engine.pool, "capped", lambda: None)

    with pytest.raises(chat.ChatError, match="prompt submission deadline expired waiting for a browser seat"):
        engine.ask({"prompt": "hello", "timeout": 0.01}, on_event=lambda _event: None)

    assert engine.waiting == 0
    pool.release(0)


def test_submit_emits_truthful_submission_lifecycle(monkeypatch):
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "0")
    events = []
    fake = seat_for([])[1]
    turn = chat.TurnState()

    chat.submit(fake, timeout=5, turn=turn, on_event=events.append)

    assert [event["type"] for event in events] == ["submitting", "submitted"]
    assert [event["phase"] for event in events] == ["submitting", "submitted"]


def test_submit_emits_submitted_when_response_growth_proves_acceptance(monkeypatch):
    monkeypatch.setenv("ASKGPT_DISPATCH_GAP", "0")
    events = []
    fake = seat_for([{"text": "reply"}])[1]
    fake._accept_reads = 1
    turn = chat.TurnState()

    chat.submit(fake, timeout=5, turn=turn, on_event=events.append)

    assert [event["type"] for event in events].count("submitted") == 1


def test_run_log_prints_pre_submission_progress_for_non_live_calls(capsys):
    from ask_gpt import RunLog

    log = RunLog("prompt", None, chat.STATE, "stamp", None, live=False)
    log.on_event({"type": "queued", "phase": "queued", "wait_s": 10.0})
    log.on_event({"type": "submitting", "phase": "submitting"})
    log.on_event({"type": "submitted", "phase": "submitted"})

    assert capsys.readouterr().err.splitlines() == [
        "ask-gpt: queued for a browser seat (up to 10s)",
        "ask-gpt: submitting prompt",
        "ask-gpt: prompt submitted; waiting for response",
    ]


def test_read_only_download_path_does_not_reference_submit():
    # The gate lives at the sole send seam, so browsing an existing conversation
    # cannot consume a dispatch slot.
    assert "submit" not in chat.download_attachments.__code__.co_names