"""Lock order and mutual exclusion (docs/specs/2026-08-10-ask-gpt-resume-live-design.md
§3): conversation before seat, everywhere. The daemon's per-conversation mutex is the
correctness floor; the CLI flock additionally covers daemon-down direct runs."""

from __future__ import annotations

import json
import threading
import time

import pytest

import chat
import convlog
import solwebd
import translate
from fake_browser import block, pool_for, seat_for

CID = "55555555-5555-4555-8555-555555555555"
MUTEX_SYSTEM = {"role": "system", "content": "be terse"}
MUTEX_ASK = {"role": "user", "content": "hi"}
MUTEX_DONE = block(json.dumps({"tool": "final", "text": "ok"}))


def mutex_request(messages, model="sol-web-medium"):
    return {"model": model, "messages": messages}


def mutex_key(messages):
    return translate.session_key(messages, [], solwebd.effort_of("sol-web-medium"))


@pytest.fixture(autouse=True)
def no_sleep(monkeypatch):
    import registry
    registry.append("owner@example.com", CID, "existing")
    monkeypatch.setattr(chat.time, "sleep", lambda *_: None)


def test_conversation_lock_is_acquired_before_the_seat_is_ever_touched(monkeypatch):
    seat, marionette = seat_for([{"blocks": [], "text": "hi"}])
    marionette.turns = 1  # open_resume requires >=1 existing turn
    engine = solwebd.Engine(seat)

    opened = []
    real_open = seat.open

    def tracking_open():
        opened.append(True)
        return real_open()
    monkeypatch.setattr(seat, "open", tracking_open)

    conv_lock = engine._key_lock(f"ask-conv:{CID}")
    conv_lock.acquire()
    done = threading.Event()
    thread = threading.Thread(
        target=lambda: (engine.ask({"prompt": "hi", "stamp": "s", "conversation": CID}),
                        done.set()),
        daemon=True)
    thread.start()
    assert not done.wait(0.3)
    assert opened == [], "the seat was touched before the conversation lock was free"
    conv_lock.release()
    assert done.wait(2)
    assert opened, "the seat was never touched once the conversation lock was free"


def test_two_same_id_ask_requests_are_serialised_not_interleaved(monkeypatch):
    pool, marionettes = pool_for([[], []])
    for marionette in marionettes:
        marionette.turns = 1
    engine = solwebd.Engine(pool)

    order = []
    gate = threading.Event()
    real_ask_on = chat.ask_on

    def scripted_ask_on(m, prompt, **kw):
        order.append(("start", prompt))
        if prompt == "first":
            gate.wait(2)
        order.append(("end", prompt))
        return chat.Reply(text="ok", url=f"https://chatgpt.com/c/{CID}", conversation_id=CID)
    monkeypatch.setattr(solwebd.chat, "ask_on", scripted_ask_on)

    first = threading.Thread(
        target=lambda: engine.ask({"prompt": "first", "stamp": "s", "conversation": CID}),
        daemon=True)
    first.start()
    threading.Event().wait(0.2)
    assert order == [("start", "first")], "the first turn had not even started yet"

    second_done = threading.Event()
    second = threading.Thread(
        target=lambda: (engine.ask({"prompt": "second", "stamp": "s", "conversation": CID}),
                        second_done.set()),
        daemon=True)
    second.start()
    assert not second_done.wait(0.3), "the second same-id turn ran while the first was in flight"

    gate.set()
    first.join(2)
    assert second_done.wait(2)
    assert order == [("start", "first"), ("end", "first"), ("start", "second"),
                     ("end", "second")]
    monkeypatch.setattr(solwebd.chat, "ask_on", real_ask_on)


def test_two_ask_requests_naming_the_same_id_in_different_case_still_serialise(monkeypatch):
    """A raw API caller cannot dodge the mutex by sending the uuid uppercase —
    `conversation` is normalised before it becomes a lock key."""
    pool, marionettes = pool_for([[], []])
    for marionette in marionettes:
        marionette.turns = 1
    engine = solwebd.Engine(pool)

    order = []
    gate = threading.Event()
    real_ask_on = chat.ask_on

    def scripted_ask_on(m, prompt, **kw):
        order.append(("start", prompt))
        if prompt == "first":
            gate.wait(2)
        order.append(("end", prompt))
        return chat.Reply(text="ok", url=f"https://chatgpt.com/c/{CID}", conversation_id=CID)
    monkeypatch.setattr(solwebd.chat, "ask_on", scripted_ask_on)

    first = threading.Thread(
        target=lambda: engine.ask({"prompt": "first", "stamp": "s", "conversation": CID}),
        daemon=True)
    first.start()
    threading.Event().wait(0.2)
    assert order == [("start", "first")], "the first turn had not even started yet"

    second_done = threading.Event()
    second = threading.Thread(
        target=lambda: (engine.ask({"prompt": "second", "stamp": "s",
                                    "conversation": CID.upper()}), second_done.set()),
        daemon=True)
    second.start()
    assert not second_done.wait(0.3), \
        "the uppercase-id second turn ran while the lowercase-id first was still in flight"

    gate.set()
    first.join(2)
    assert second_done.wait(2)
    assert order == [("start", "first"), ("end", "first"), ("start", "second"),
                     ("end", "second")]
    monkeypatch.setattr(solwebd.chat, "ask_on", real_ask_on)


def test_a_disconnected_turn_holds_the_mutex_until_its_own_terminal(monkeypatch):
    """The daemon finishes the turn it started rather than cancelling it — a
    reconnecting or competing client cannot start a second turn mid-generation."""
    seat, marionette = seat_for([])
    marionette.turns = 1
    engine = solwebd.Engine(seat)
    gate = threading.Event()

    def slow_ask_on(m, prompt, **kw):
        gate.wait(2)
        return chat.Reply(text="ok", url=f"https://chatgpt.com/c/{CID}", conversation_id=CID)
    monkeypatch.setattr(solwebd.chat, "ask_on", slow_ask_on)

    first_done = threading.Event()
    threading.Thread(
        target=lambda: (engine.ask({"prompt": "hi", "stamp": "s", "conversation": CID}),
                        first_done.set()),
        daemon=True).start()
    time.sleep(0.2)  # the "client" walks away here; the turn keeps running regardless

    competing_done = threading.Event()
    threading.Thread(
        target=lambda: (engine.ask({"prompt": "hi", "stamp": "s", "conversation": CID}),
                        competing_done.set()),
        daemon=True).start()
    assert not competing_done.wait(0.3)
    gate.set()
    assert first_done.wait(2)
    assert competing_done.wait(2)


def test_a_competing_resume_loses_the_cli_flock_while_meta_is_held(tmp_path, monkeypatch):
    monkeypatch.setattr(convlog, "STATE", tmp_path / "gptbridge")
    monkeypatch.setattr(convlog, "LOGS", tmp_path / "gptbridge" / "logs")
    with convlog.held(CID, timeout=0.2):
        try:
            with convlog.held(CID, timeout=0.2):
                raise AssertionError("a second holder should never have gotten the lock")
        except convlog.ConvLogError as exc:
            assert CID in str(exc)


def test_chat_turn_and_ask_resume_on_same_conversation_serialize(monkeypatch):
    """The known limitation this closes: a /v1/chat turn in flight on conv X must
    block a concurrent /v1/ask resume naming the same X, even though a second seat
    sits free for it to take."""
    pool, marionettes = pool_for([[], []])
    for m in marionettes:
        m.turns = 1  # open_resume requires >=1 existing turn
    engine = solwebd.Engine(pool)

    gate = threading.Event()
    started = threading.Event()
    order: list[str] = []

    def blocking_turn(text, effort, timeout=900.0):
        order.append("chat-start")
        started.set()
        gate.wait(2)
        order.append("chat-end")
        return {"blocks": MUTEX_DONE["blocks"], "text": MUTEX_DONE["text"],
                "url": marionettes[0].url()}
    monkeypatch.setattr(pool.seat(0), "turn", blocking_turn)

    messages = [MUTEX_SYSTEM, MUTEX_ASK]
    chat_done = threading.Event()
    chat_thread = threading.Thread(
        target=lambda: (engine.complete(mutex_request(messages)), chat_done.set()), daemon=True)
    chat_thread.start()
    assert started.wait(2), "the chat turn never started"

    conv = engine.registry.get(mutex_key(messages))
    conv_id = chat.conversation_id_from_url(conv.url)
    assert conv_id, "a fresh conversation must have a resolvable uuid after _open"
    import registry
    registry.append("owner@example.com", conv_id, "existing")

    real_ask_on = chat.ask_on

    def scripted_ask_on(m, prompt, **kw):
        order.append("ask-start")
        order.append("ask-end")
        return chat.Reply(text="hi back", url=m.url(), conversation_id=conv_id)
    monkeypatch.setattr(solwebd.chat, "ask_on", scripted_ask_on)

    ask_done = threading.Event()
    ask_thread = threading.Thread(
        target=lambda: (engine.ask({"prompt": "resume", "stamp": "s", "conversation": conv_id}),
                        ask_done.set()),
        daemon=True)
    ask_thread.start()
    assert not ask_done.wait(0.3), \
        "/v1/ask resumed the same conversation while the /v1/chat turn was still in flight"

    gate.set()
    assert chat_done.wait(2)
    assert ask_done.wait(2)
    assert order == ["chat-start", "chat-end", "ask-start", "ask-end"]
    monkeypatch.setattr(solwebd.chat, "ask_on", real_ask_on)


def test_different_conversation_uuids_do_not_contend():
    seat, _ = seat_for([])
    engine = solwebd.Engine(seat)
    url_a = "https://chatgpt.com/c/11111111-1111-4111-8111-111111111111"
    url_b = "https://chatgpt.com/c/22222222-2222-4222-8222-222222222222"
    lock_a = engine._conv_uuid_lock(url_a)
    lock_b = engine._conv_uuid_lock(url_b)
    assert lock_a is not lock_b
    assert lock_a.acquire(blocking=False)
    try:
        assert lock_b.acquire(blocking=False), \
            "two different conversation uuids must not share a lock"
        lock_b.release()
    finally:
        lock_a.release()


def test_conversation_without_a_uuid_in_its_url_is_unaffected(monkeypatch):
    seat, marionette = seat_for([MUTEX_DONE])
    marionette.no_meta = True  # url never becomes /c/<uuid>, as in the real protocol case
    engine = solwebd.Engine(seat)

    result = engine.complete(mutex_request([MUTEX_SYSTEM, MUTEX_ASK]))

    assert result["choices"][0]["finish_reason"] == "stop"
    assert engine._conv_uuid_lock(marionette.url()) is None
    assert not any(k.startswith("ask-conv:") for k in engine._key_locks), \
        "a url with no uuid must never mint an ask-conv lock"


def test_new_conversation_acquires_uuid_lock_after_open_and_releases_at_turn_end(monkeypatch):
    seat, marionette = seat_for([])
    engine = solwebd.Engine(seat)
    held_mid_turn = {}

    def blocking_turn(text, effort, timeout=900.0):
        conv = engine.registry.get(mutex_key([MUTEX_SYSTEM, MUTEX_ASK]))
        conv_id = chat.conversation_id_from_url(conv.url)
        assert conv_id, "the new conversation must have gotten a uuid from _open"
        lock = engine._key_lock(f"ask-conv:{conv_id}")
        held_mid_turn["acquirable"] = lock.acquire(blocking=False)
        if held_mid_turn["acquirable"]:
            lock.release()  # undo the probe; the real holder still owns it
        held_mid_turn["conv_id"] = conv_id
        return {"blocks": MUTEX_DONE["blocks"], "text": MUTEX_DONE["text"],
                "url": marionette.url()}
    monkeypatch.setattr(seat, "turn", blocking_turn)

    engine.complete(mutex_request([MUTEX_SYSTEM, MUTEX_ASK]))

    assert held_mid_turn["acquirable"] is False, \
        "the uuid lock acquired in _open must already be held mid-turn"

    lock = engine._key_lock(f"ask-conv:{held_mid_turn['conv_id']}")
    assert lock.acquire(blocking=False), \
        "the uuid lock acquired mid-turn in _open must be released once the turn ends"
    lock.release()