"""ask-gpt routes through solwebd when it is up, and never fights it for the browser."""

import io
import json
import threading
import urllib.error
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

import pytest

import ask_gpt
import browser
import chat
import registry
import solwebd
from fake_browser import seat_for

TOKEN = "routing-token"
CONV_ID = "11111111-1111-4111-8111-111111111111"


def _sse(events: list[dict]) -> bytes:
    return b"".join(f"data: {json.dumps(e)}\n\n".encode() for e in events)


class Stub(BaseHTTPRequestHandler):
    ok = True
    authorised = True
    truncate = False
    die = False
    old = False  # a daemon that predates streaming /v1/ask: blocking JSON, ignores `stream`
    asks: list[dict] = []
    reply = {"text": "meow", "thinking": "", "images": ["/tmp/cat.png"],
            "files": ["/tmp/theme.zip"], "url": f"https://chatgpt.com/c/{CONV_ID}",
            "conversation_id": CONV_ID}

    def log_message(self, *a):
        pass

    def _json(self, status, payload):
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if not Stub.authorised:
            self._json(401, {"error": {"message": "nope"}})
            return
        self._json(200, {"ok": Stub.ok, "conversations": 0, "queue": 0, "turn_p50_s": 0.0})

    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
        Stub.asks.append(body)
        if Stub.truncate:
            Stub.ok = not Stub.die
            # headers promising a body, then the socket goes away: what a handler
            # exception in the real daemon looks like from the client side
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", "500")
            self.end_headers()
            self.close_connection = True
            return
        if Stub.old or not body.get("stream"):
            self._json(200, {"text": Stub.reply["text"], "images": Stub.reply["images"],
                             "files": Stub.reply["files"], "conversation": Stub.reply["url"]})
            return
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Connection", "close")
        self.end_headers()
        self.close_connection = True
        self.wfile.write(_sse([
            {"type": "meta", "phase": "named", "conversation_id": Stub.reply["conversation_id"],
             "url": Stub.reply["url"]},
            {"type": "answer", "phase": "streaming", "delta": Stub.reply["text"]},
            {"type": "response_done", "phase": "response_done",
             "thinking": Stub.reply["thinking"], "answer": Stub.reply["text"]},
            {"type": "done", "phase": "done", "reply": Stub.reply},
        ]))
        self.wfile.flush()


@pytest.fixture
def daemon(monkeypatch, tmp_path):
    Stub.ok, Stub.authorised, Stub.truncate, Stub.die, Stub.old, Stub.asks = \
        True, True, False, False, False, []
    server = HTTPServer(("127.0.0.1", 0), Stub)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    token_file = tmp_path / "token"
    token_file.write_text(TOKEN + "\n")
    monkeypatch.setattr(ask_gpt, "DAEMON_PORT", server.server_port)
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token_file)
    yield server
    server.shutdown()


@pytest.fixture
def no_browser(monkeypatch):
    """Taking `profile.lock` is the exact thing the daemon path must not do."""
    def refuse(*_a, **_k):
        raise AssertionError("ask-gpt opened a browser session while the daemon was up")
    monkeypatch.setattr(browser.Session, "__enter__", refuse)


@pytest.fixture
def no_log(monkeypatch, tmp_path):
    """Route the conversation log under tmp_path so tests never touch the real
    ~/.overdeck tree, and never block on a lock a previous test left behind."""
    import convlog
    monkeypatch.setattr(convlog, "STATE", tmp_path / "gptbridge")
    monkeypatch.setattr(convlog, "LOGS", tmp_path / "gptbridge" / "logs")
    monkeypatch.chdir(tmp_path)


def test_the_daemon_answers_and_no_browser_session_is_opened(daemon, no_browser, no_log, capsys):
    assert ask_gpt.main(["draw me a cat", "--effort", "high"]) == 0
    output = capsys.readouterr().out
    assert "meow" in output
    assert "saved: /tmp/cat.png" in output
    assert "saved: /tmp/theme.zip" in output
    sent = Stub.asks[0]
    assert sent["prompt"] == "draw me a cat" and sent["effort"] == "high"
    assert sent["stream"] is True


def test_json_reports_generated_files(daemon, no_browser, no_log, capsys):
    assert ask_gpt.main(["build a theme", "--json"]) == 0
    result = json.loads(capsys.readouterr().out)
    assert result["files"] == ["/tmp/theme.zip"]
    assert result["images"] == ["/tmp/cat.png"]
    assert result["conversation"] == Stub.reply["url"]
    assert result["conversation_id"] == CONV_ID
    assert result["resume"] == f"ask-gpt --resume {CONV_ID}"


def test_attachment_and_out_paths_are_absolute(
        daemon, no_browser, no_log, tmp_path, monkeypatch):
    picture = tmp_path / "in.png"
    picture.write_bytes(b"x")
    monkeypatch.chdir(tmp_path)
    ask_gpt.main(["look", "-a", "in.png", "--out", "shots"])
    sent = Stub.asks[0]
    assert sent["attach"] == [str(picture)]
    assert sent["out"] == str(tmp_path / "shots")


def test_a_rejected_token_is_an_error_not_a_fallthrough(daemon, no_browser, no_log, capsys):
    Stub.authorised = False
    assert ask_gpt.main(["hi"]) == 1
    assert "rejected our token" in capsys.readouterr().err


def test_nothing_listening_takes_the_direct_path(monkeypatch, tmp_path):
    token_file = tmp_path / "token"
    token_file.write_text(TOKEN + "\n")
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token_file)
    monkeypatch.setattr(ask_gpt, "DAEMON_PORT", 1)  # nothing binds port 1
    assert ask_gpt.daemon_endpoint() is None

    seen = {}

    def direct(prompt, **kw):
        seen["prompt"] = prompt
        return chat.Reply(text="direct", url="u")

    monkeypatch.setattr(ask_gpt, "ask", direct)
    assert ask_gpt.main(["hi"]) == 0
    assert seen["prompt"] == "hi"


@pytest.mark.parametrize(
    ("selector", "expected_mode"),
    [
        ([], "virtual"),
        (["--mode", "show"], "virtual"),
        (["--mode", "anything"], "virtual"),
        (["--visible"], "show"),
        (["--mode", "anything", "--visible"], "show"),
    ],
)
def test_cli_display_selection_is_agent_safe(selector, expected_mode, monkeypatch, no_log):
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)
    seen = {}

    def direct(_prompt, **kwargs):
        seen["mode"] = kwargs["mode"]
        return chat.Reply(text="ok", url="u")

    monkeypatch.setattr(ask_gpt, "ask", direct)
    assert ask_gpt.main(["hi", *selector]) == 0
    assert seen["mode"] == expected_mode


def test_visible_help_is_human_only(capsys):
    with pytest.raises(SystemExit) as exc:
        ask_gpt.main(["--help"])
    assert exc.value.code == 0
    help_text = capsys.readouterr().out
    assert "--visible" in help_text
    assert "human invocation only" in help_text.lower()
    assert "agents must not choose" in help_text.lower()
    assert "--mode" not in help_text


def test_default_invocation_stdout_is_byte_identical_to_before_the_feature(
        monkeypatch, tmp_path, capsys, no_log):
    """The mandatory golden test: no new flags, not --json — stdout must be exactly
    what it was before events/resume/log/--live existed, no matter what the turn
    engine now does internally or on stderr."""
    token_file = tmp_path / "token"
    token_file.write_text(TOKEN + "\n")
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token_file)
    monkeypatch.setattr(ask_gpt, "DAEMON_PORT", 1)  # nothing binds port 1

    def direct(prompt, **kw):
        on_event = kw.get("on_event")
        if on_event is not None:
            on_event({"type": "meta", "phase": "named",
                      "conversation_id": "88888888-8888-4888-8888-888888888888",
                      "url": "https://chatgpt.com/c/88888888-8888-4888-8888-888888888888"})
            on_event({"type": "done", "phase": "done", "reply": {
                "text": "the answer", "thinking": "", "images": [], "files": [],
                "url": "https://chatgpt.com/c/88888888-8888-4888-8888-888888888888",
                "conversation_id": "88888888-8888-4888-8888-888888888888"}})
        return chat.Reply(text="the answer", images=[Path("/tmp/pic.png")],
                          files=[Path("/tmp/doc.zip")], url="u",
                          conversation_id="88888888-8888-4888-8888-888888888888")

    monkeypatch.setattr(ask_gpt, "ask", direct)
    monkeypatch.chdir(tmp_path)
    assert ask_gpt.main(["hello"]) == 0
    out = capsys.readouterr().out
    assert out == "the answer\n\nsaved: /tmp/pic.png\n\nsaved: /tmp/doc.zip\n"


def test_a_dead_daemon_falls_through_to_the_browser(daemon, monkeypatch, capsys):
    """Healthz passed, the connection then dropped, and the daemon is gone."""
    def drop(*_a, **_k):
        Stub.ok = False  # the re-probe must find it down before the browser is opened
        raise ConnectionResetError("Remote end closed connection without response")

    monkeypatch.setattr(ask_gpt, "_ask_stream", drop)
    monkeypatch.setattr(ask_gpt, "ask",
                        lambda prompt, **kw: chat.Reply(text="direct", url="u"))
    assert ask_gpt.main(["hi"]) == 0
    captured = capsys.readouterr()
    assert "direct" in captured.out
    assert "died mid-request" in captured.err


def test_a_surviving_daemon_is_an_error_not_a_fallthrough(daemon, no_browser, no_log, capsys):
    """It still holds profile.lock, so the direct path would block for 600s."""
    Stub.truncate = True
    assert ask_gpt.main(["hi"]) == 1
    err = capsys.readouterr().err
    assert "still listening" in err
    assert "IncompleteRead" in err, "the truncated-body branch was not the one exercised"


def test_a_body_truncated_after_the_headers_is_caught(daemon, monkeypatch, capsys):
    """IncompleteRead is not an OSError; a real socket is the only way to raise it."""
    Stub.truncate, Stub.die = True, True
    monkeypatch.setattr(ask_gpt, "ask",
                        lambda prompt, **kw: chat.Reply(text="direct", url="u"))
    assert ask_gpt.main(["hi"]) == 0
    assert "direct" in capsys.readouterr().out


def test_a_daemon_error_response_is_reported_not_retried(
        daemon, no_browser, no_log, monkeypatch, capsys):
    """An HTTPError carries the daemon's own message; falling through would hide it."""
    def refuse(*_a, **_k):
        raise urllib.error.HTTPError(
            "http://x/v1/ask", 500, "Server Error", {},
            io.BytesIO(json.dumps({"error": {"message": "seat is wedged"}}).encode()))

    monkeypatch.setattr(ask_gpt, "_ask_stream", refuse)
    assert ask_gpt.main(["hi"]) == 1
    assert "seat is wedged" in capsys.readouterr().err


def test_daemon_registry_failure_returns_service_error_before_browser(tmp_path):
    registry.PATH.parent.mkdir(parents=True)
    registry.PATH.write_text("not-json\n")
    seat, marionette = seat_for([])
    engine = solwebd.Engine(seat)
    handler = object.__new__(solwebd.Handler)

    status, payload, headers = handler._run(
        {"conversation": CONV_ID, "download": True, "stamp": "s"}, engine.ask)

    assert status == 503
    assert payload == {"error": {"message":
        "conversation registry unavailable: conversation registry line 1 is invalid JSON"}}
    assert headers == {}
    assert marionette.navigations == []
    assert marionette.typed == []


def test_direct_metadata_registry_failure_is_reported_once_without_traceback(
        monkeypatch, tmp_path, capsys, no_log):
    registry.PATH.parent.mkdir(parents=True)
    registry.PATH.write_text("not-json\n")
    _, marionette = seat_for([{"blocks": [], "text": "reply"}])
    events = []
    monkeypatch.setattr(chat.time, "sleep", lambda *_: None)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)

    def direct(prompt, **kwargs):
        def on_event(event):
            events.append(event)
            kwargs["on_event"](event)
        return chat.ask_on(marionette, prompt, out_dir=kwargs["out_dir"],
                           stamp=kwargs["stamp"], on_event=on_event)

    monkeypatch.setattr(ask_gpt, "ask", direct)

    assert ask_gpt.main(["hello", "--out", str(tmp_path / "out")]) == 1
    captured = capsys.readouterr()
    assert "ask-gpt: conversation registry line 1 is invalid JSON" in captured.err
    assert "Traceback" not in captured.err
    assert [event["type"] for event in events] == ["submitting", "submitted", "error"]
    assert events[-1]["phase"] == "named"
    assert marionette.typed == ["hello"]


def test_no_token_file_means_no_daemon(monkeypatch, tmp_path):
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", tmp_path / "absent")
    assert ask_gpt.daemon_endpoint() is None


def test_an_old_daemon_is_a_hard_error_never_a_silent_downgrade(
        daemon, no_browser, no_log, capsys):
    """No SSE support means no conversation_id, no thinking, no reliable log — the
    owner is told to restart solwebd rather than getting a silently degraded run."""
    Stub.old = True
    assert ask_gpt.main(["hi"]) == 1
    assert "restart solwebd" in capsys.readouterr().err


def test_the_ask_route_holds_the_same_lock_as_an_agentic_turn(monkeypatch):
    """Two writers on one composer would interleave keystrokes."""
    monkeypatch.setattr(chat.time, "sleep", lambda *_: None)
    seat, marionette = seat_for([{"blocks": [], "text": "hi"}])
    engine = solwebd.Engine(seat)
    engine.pool.locks[0].acquire()
    done = threading.Event()
    threading.Thread(target=lambda: (engine.ask({"prompt": "hi", "stamp": "s"}), done.set()),
                     daemon=True).start()
    assert not done.wait(0.3), "/v1/ask ran while the only seat's lock was held"
    engine.pool.locks[0].release()


def test_an_unknown_effort_is_rejected_before_the_browser_is_touched():
    engine = solwebd.Engine(seat_for([])[0])
    with pytest.raises(ValueError):
        engine.ask({"prompt": "hi", "effort": "turbo", "stamp": "s"})


def test_a_missing_attachment_is_rejected(tmp_path):
    engine = solwebd.Engine(seat_for([])[0])
    with pytest.raises(ValueError):
        engine.ask({"prompt": "hi", "attach": [str(tmp_path / "gone.png")], "stamp": "s"})


def test_reset_accumulation_clears_a_dead_daemons_partial_deltas(tmp_path, no_log):
    """A daemon attempt that streamed part of an answer before dying must not have
    its text glued onto the direct-browser retry's own answer for the same turn."""
    rl = ask_gpt.RunLog("hi", None, tmp_path, "s", None, live=False)
    rl.on_event({"type": "meta", "phase": "named",
                "conversation_id": "77777777-7777-4777-8777-777777777777",
                "url": "https://chatgpt.com/c/77777777-7777-4777-8777-777777777777"})
    rl.on_event({"type": "answer", "phase": "streaming", "delta": "from the dead daemon"})
    assert rl.answer == "from the dead daemon"
    rl.reset_accumulation()
    assert rl.answer == "" and rl.thinking == "" and rl.artifacts == []
    rl.on_event({"type": "answer", "phase": "streaming", "delta": "from the retry"})
    assert rl.answer == "from the retry"
    rl.close(ok=True)