"""Serve the owner's ChatGPT web session as an OpenAI-compatible endpoint.

One or more browser seats for the daemon's lifetime, so each seat's ~27s cold start
is paid once and its profile lock is held by exactly one process. `/v1/ask` takes
any free seat; an agentic `/v1/chat` conversation is pinned to the seat it opened on,
since the composer's typed-but-unsent state lives in that one browser. Default is
one seat, serialising every request exactly as before `--seats` existed.

Filesystem confinement is the calling agent's own permission system; nothing here
executes a tool. `sandbox.py`/`workspace.py` guard the MCP tunnel path, not this one.
"""

from __future__ import annotations

import argparse
from collections import deque
import json
import os
import queue
import secrets
import statistics
import sys
import subprocess
import threading
import time
import uuid
import re
import hashlib
import stat
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import browser
import chat
import protocol
import registry
import translate
from browser import MarionetteError, Session
from protocol import Final, Malformed, ToolCall

STATE = Path(os.environ.get("OVERDECK_GPTBRIDGE_STATE", str(Path.home() / ".overdeck" / "gptbridge")))
TOKEN_FILE = STATE / "solwebd.token"
def _build_identity() -> str:
    override = os.environ.get("OVERDECK_GPTBRIDGE_VERSION") or os.environ.get("OVERDECK_BUILD_ID")
    if override:
        return override
    try:
        return subprocess.check_output(["git", "-C", str(Path(__file__).resolve().parents[2]), "rev-parse", "HEAD"], text=True, stderr=subprocess.DEVNULL).strip()
    except Exception:
        return "unavailable"

BUILD_ID = _build_identity()
DEFAULT_PORT = 8791
MODEL_PREFIX = "sol-web-"
MODELS = [MODEL_PREFIX + effort for effort in chat.EFFORTS]
# Factory models coexist with legacy instant even when chat.EFFORTS changes.
for _effort in ("medium", "high", "xhigh", "pro", "instant"):
    if MODEL_PREFIX + _effort not in MODELS:
        MODELS.append(MODEL_PREFIX + _effort)
CLIENT_SESSION_RE = re.compile(r"^v1\.[A-Za-z0-9_-]{43}$")
MAX_REQUEST_BODY = 10 * 1024 * 1024
KEEPALIVE_SECONDS = 5.0
# A client that walked away cannot cancel a Marionette call in flight, so the turn
# timeout is also how long one abandoned request can hold the single browser seat.
TURN_TIMEOUT = float(os.environ.get("SOLWEBD_TURN_TIMEOUT", "300"))

# ChatGPT renders fenced code as DOM nodes, so fence markers never reach innerText.
# Tool calls are read out of the reply turn's `pre code` elements instead.
REPLY_BLOCKS = """
let last = null;
for (const t of document.querySelectorAll(arguments[0])) {
  if (t.querySelector(arguments[1])) continue;
  last = t;
}
if (!last) return null;
return {blocks: Array.from(last.querySelectorAll('pre code')).map(c => c.textContent),
        text: (last.innerText || '').replace(/^ChatGPT said:\\s*/, '').trim(),
        url: location.href};
"""


# One seat hitting the cap means the account is capped, not just that seat; other
# seats fail fast on the same state for this long instead of each burning a turn
# to discover it themselves.
CAP_COOLDOWN_SECONDS = 300.0


def _process_start_ticks(pid: int) -> int | None:
    try:
        return int(Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[19])
    except (OSError, IndexError, ValueError):
        return None


class Capped(Exception):
    def __init__(self, state: dict):
        super().__init__(state.get("text", "usage cap"))
        self.state = state


class LoggedOut(Exception):
    pass


class ProtocolBroken(Exception):
    pass


class ConversationNotFound(Exception):
    pass


def read_token() -> str:
    """One bearer token per machine; any local process could otherwise drive the
    owner's real ChatGPT account."""
    STATE.mkdir(parents=True, exist_ok=True)
    if not TOKEN_FILE.exists():
        TOKEN_FILE.touch(mode=0o600)
        TOKEN_FILE.write_text(secrets.token_urlsafe(32) + "\n")
    TOKEN_FILE.chmod(0o600)
    return TOKEN_FILE.read_text().strip()


def effort_of(model: str) -> str:
    effort = model.split(",")[-1].strip()
    if not effort.startswith(MODEL_PREFIX):
        raise ValueError(f"model must be one of {MODELS}")
    effort = effort[len(MODEL_PREFIX):]
    if effort not in chat.EFFORTS:
        raise ValueError(f"model must be one of {MODELS}")
    return effort


def split_messages(messages: list[dict]) -> tuple[str, list[dict]]:
    system = "\n\n".join(translate._text(m) for m in messages if m.get("role") == "system")
    return system, [m for m in messages if m.get("role") != "system"]


class BrowserSeat:
    """The single live conversation surface. Owns the browser; knows nothing about HTTP."""

    def __init__(self, mode: str = "virtual", session_factory=None):
        self.mode = mode
        self._factory = session_factory or (lambda: Session(mode=mode))
        self._session = None
        self._m = None

    def open(self):
        if self._m is None:
            self._session = self._factory()
            self._m = self._session.__enter__()
            self._m.navigate("https://chatgpt.com/")
            chat.wait_ready(self._m)
            chat.use_chat_surface(self._m)
        return self._m

    def close(self) -> None:
        if self._session is not None:
            self._session.__exit__(None, None, None)
        self._session, self._m = None, None

    def rebuild(self):
        """A dead browser is rebuilt once; a second failure is reported, not retried."""
        self.close()
        return self.open()

    def new_conversation(self) -> str:
        m = self.open()
        m.navigate("https://chatgpt.com/")
        chat.wait_ready(m)
        chat.use_chat_surface(m)
        return m.url()

    def turn(self, text: str, effort: str, timeout: float = 900.0) -> dict:
        """Type one delta, send it, and read the reply's code blocks back."""
        m = self.open()
        cap = chat.detect_cap(m)
        if cap:
            raise Capped(cap)
        # Re-asserted every send: whether the effort control is per-conversation or
        # composer-global is a UI detail this daemon must not bet on.
        chat.set_effort(m, effort)
        chat.type_prompt(m, text)
        try:
            chat.send(m, timeout=timeout)
        except chat.ChatError:
            cap = chat.detect_cap(m)
            if cap:
                raise Capped(cap) from None
            raise
        data = m.sync_script(REPLY_BLOCKS, [chat.TURN, chat.USER])
        if not data:
            raise chat.ChatError("no reply turn appeared in the conversation")
        return data


class SeatPool:
    """N browser seats, each bound to its own profile slot.

    Ownership of a seat is exactly its `threading.Lock`: there is no separate
    free-list to fall out of sync with it. "Any free seat" scans the locks for one
    not held; a pinned request waits on its own seat's lock specifically. With one
    seat both reduce to the single lock every request serialised on before pools
    existed.
    """

    def __init__(self, max_seats: int = 1, mode: str = "virtual", session_factory=None):
        if max_seats < 1:
            raise ValueError("max_seats must be >= 1")
        factory = session_factory or (lambda slot: Session(mode=mode, slot=slot))
        self.seats: list[BrowserSeat] = []
        for i in range(max_seats):
            slot = i + 1
            self.seats.append(
                BrowserSeat(mode=mode, session_factory=lambda slot=slot: factory(slot)))
        self.locks = [threading.Lock() for _ in self.seats]
        self._cv = threading.Condition()
        self._capped_state: dict | None = None
        self._capped_until = 0.0
        self._active = 0

    @classmethod
    def wrapping(cls, seat: BrowserSeat) -> "SeatPool":
        """Adopt one already-built seat as a 1-seat pool — the default `--seats 1`
        path, and every caller still constructing a bare `BrowserSeat` directly."""
        pool = cls.__new__(cls)
        pool.seats = [seat]
        pool.locks = [threading.Lock()]
        pool._cv = threading.Condition()
        pool._capped_state = None
        pool._capped_until = 0.0
        pool._active = 0
        return pool

    def seat(self, index: int) -> BrowserSeat:
        return self.seats[index]

    def acquire_any(self, timeout: float | None = None, deadline: float | None = None,
                    on_wait=None) -> int:
        """Wait within one caller deadline for any free seat, lock it, and return its index."""
        if deadline is None and timeout is not None:
            deadline = time.monotonic() + timeout
        announced = False
        with self._cv:
            while True:
                for index, lock in enumerate(self.locks):
                    if lock.acquire(blocking=False):
                        self._active += 1
                        return index
                remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
                if not announced and on_wait is not None:
                    on_wait({"type": "queued", "wait_s": remaining})
                    announced = True
                if remaining is not None and remaining <= 0:
                    raise chat.ChatError("prompt submission deadline expired waiting for a browser seat")
                self._cv.wait(0.5 if remaining is None else min(0.5, remaining))

    def acquire_seat(self, index: int) -> int:
        """Block until this specific (pinned) seat is free, then lock and return it."""
        self.locks[index].acquire()
        with self._cv:
            self._active += 1
        return index

    def release(self, index: int) -> None:
        self.locks[index].release()
        with self._cv:
            self._active -= 1
            self._cv.notify_all()

    def mark_capped(self, state: dict) -> None:
        self._capped_state, self._capped_until = state, time.monotonic() + CAP_COOLDOWN_SECONDS

    def capped(self) -> dict | None:
        if self._capped_state and time.monotonic() < self._capped_until:
            return self._capped_state
        self._capped_state = None
        return None

    def close(self) -> None:
        for seat in self.seats:
            seat.close()


class Engine:
    """Request service over a seat pool: `/v1/ask` takes any free seat; an agentic
    conversation stays pinned to the seat it opened on."""

    def __init__(self, pool: "SeatPool | BrowserSeat"):
        self.pool = pool if isinstance(pool, SeatPool) else SeatPool.wrapping(pool)
        self.registry = translate.Registry()
        self.waiting = 0
        self.durations: list[float] = []
        self._key_guard = threading.Lock()
        self._key_locks: dict[str, threading.Lock] = {}
        self.runtime_state = "ready"
        self.runtime_detail = "browser authentication has not been probed"
        self.account: str | None = None
        self.authenticated = True

    def probe_health(self) -> None:
        """Probe the live seats once; health never invents an account."""
        try:
            accounts = []
            for seat in self.pool.seats:
                accounts.append(chat.authenticated_account(seat.open()))
            if not all(accounts):
                self.runtime_state, self.runtime_detail = "logged_out", "browser profile is logged out"
            elif len(set(str(a).lower() for a in accounts)) != 1:
                self.runtime_state, self.runtime_detail = "degraded", "browser seats use different accounts"
            else:
                self.account = str(accounts[0]).lower()
                self.authenticated, self.runtime_state, self.runtime_detail = True, "ready", ""
        except Exception as exc:
            self.runtime_state, self.runtime_detail = "degraded", f"browser probe failed: {exc}"

    def _key_lock(self, key: str) -> threading.Lock:
        with self._key_guard:
            return self._key_locks.setdefault(key, threading.Lock())

    def _conv_uuid_lock(self, url: str) -> threading.Lock | None:
        """Same `ask-conv:<uuid>` key `/v1/ask` locks on, so a resumed `/v1/ask`
        and an agentic `/v1/chat` turn on the same real conversation serialize.
        No uuid in `url` yet (never opened, or a conversation kind that never
        carries one) fails open to session-key-only locking."""
        conv_id = chat.conversation_id_from_url(url) if url else None
        return self._key_lock(f"ask-conv:{conv_id}") if conv_id else None

    @property
    def queue_depth(self) -> int:
        return self.waiting

    def health(self) -> dict:
        cap = self.pool.capped()
        state = "capped" if cap else getattr(self, "runtime_state", "starting")
        authenticated = bool(getattr(self, "authenticated", False))
        account = getattr(self, "account", None)
        # Bare Engine instances are a deterministic in-process test seam; the
        # shipped daemon always has a generation and has been probe_health()'d.
        if state == "ready" and (not authenticated or (not account and getattr(self, "instance_generation", None))):
            state = "degraded"
        result = {"ok": state == "ready", "state": state, "endpoint": f"http://127.0.0.1:{DEFAULT_PORT}/v1",
                  "host": "127.0.0.1", "account": account, "seats": len(self.pool.seats),
                  "queue": self.queue_depth, "active": self.pool._active, "models": MODELS,
                  "authenticated": authenticated, "pid": os.getpid(),
                  "process_start_ticks": _process_start_ticks(os.getpid()),
                  "instance_generation": getattr(self, "instance_generation", None), "version": BUILD_ID,
                  "conversations": len(self.registry),
                  "turn_p50_s": round(statistics.median(self.durations), 1) if self.durations else 0.0}
        if cap and cap.get("resume_at"):
            result["cap_resume_at"] = cap["resume_at"]
        if state != "ready":
            result["detail"] = cap.get("text", "usage capped") if cap else getattr(self, "runtime_detail", "runtime not authenticated")
        return result

    def complete(self, body: dict, client_session_id: str | None = None) -> dict:
        effort = effort_of(str(body.get("model", "")))
        tools = body.get("tools") or []
        messages = body.get("messages") or []
        system, history = split_messages(messages)
        key = translate.session_key(messages, tools, effort, client_session_id)

        cap = self.pool.capped()
        if cap:
            raise Capped(cap)

        self.waiting += 1
        started = time.monotonic()
        try:
            # The key lock orders same-conversation turns and makes the pin lookup
            # atomic with the seat acquisition: without it, two same-key requests
            # can both find no conversation, take two different seats, and the
            # loser sends its turn on a browser showing another conversation.
            with self._key_lock(key):
                existing = self.registry.get(key)
                # Order: session-key lock (held) -> conversation-uuid lock -> seat,
                # same order /v1/ask uses (uuid before seat) so the two can never
                # deadlock on each other. A brand-new conversation has no uuid yet;
                # its lock is picked up inside _open once one exists.
                conv_lock = self._conv_uuid_lock(existing.url) if existing is not None else None
                if conv_lock is not None:
                    conv_lock.acquire()
                lock_holder = [conv_lock]
                try:
                    pin = existing.seat if existing is not None and existing.seat >= 0 else None
                    index = self.pool.acquire_seat(pin) if pin is not None else self.pool.acquire_any()
                    try:
                        return self._serve(key, system, tools, history, effort, index, lock_holder)
                    finally:
                        self.pool.release(index)
                finally:
                    if lock_holder[0] is not None:
                        lock_holder[0].release()
        finally:
            self.waiting -= 1
            self.durations.append(time.monotonic() - started)
            del self.durations[:-50]

    def ask(self, body: dict, on_event=None) -> dict:
        """One-shot ask, on any free seat — never pinned, since it never returns to
        make a second turn. `conversation` resumes an existing one instead of
        opening a fresh chat; §3's lock order (conversation before seat) makes
        `/v1/ask` with `conversation` concurrency-safe for any authenticated caller,
        not only the flock-taking CLI. `download` (§4b) is read-only recovery on an
        existing conversation — no prompt, same lock. `list`/`search`/`list_all`
        (§4c) need neither a conversation id nor its lock — nothing is typed."""
        if body.get("bootstrap"):
            return self._bootstrap_registry()
        if body.get("list") or body.get("search") is not None or body.get("list_all"):
            raise ValueError("conversation listing is local-only; use ask-gpt CLI")

        download = bool(body.get("download"))
        prompt = str(body.get("prompt") or "").strip()
        if download and prompt:
            raise ValueError("download is a read-only mode; prompt must be omitted")
        if not download and not prompt:
            raise ValueError("prompt is required")
        effort = body.get("effort") or None
        if effort and effort not in chat.EFFORTS:
            raise ValueError(f"effort must be one of {list(chat.EFFORTS)}")
        attachments = [Path(p) for p in body.get("attach") or []]
        missing = [str(p) for p in attachments if not p.is_file()]
        if missing:
            raise ValueError(f"no such file: {', '.join(missing)}")
        out_dir = Path(body.get("out") or chat.OUT_DIR)
        timeout_value = body.get("timeout")
        timeout = 900.0 if timeout_value is None else float(timeout_value)
        if timeout <= 0:
            raise ValueError("timeout must be greater than zero")
        # Lowercased so a raw API caller passing an uppercase uuid still lands on
        # the same lock/navigation target as one lowercase — chatgpt.com/c/ urls
        # and _conv_uuid_lock's `conversation_id_from_url` are always lowercase.
        conversation = str(body.get("conversation") or "").strip().lower() or None
        if download and not conversation:
            raise ValueError("download requires conversation")
        expected_account = None
        if conversation:
            owned = registry.lookup(conversation)
            if owned is None:
                raise ConversationNotFound(f"conversation {conversation} is not in the local registry")
            expected_account = owned.account

        if not download:
            cap = self.pool.capped()
            if cap:
                raise Capped(cap)

        conv_lock = self._key_lock(f"ask-conv:{conversation}") if conversation else None
        download_attempt = str(body.get("attempt") or "daemon-1")
        download_turn = chat.TurnState(
            phase="queued", conversation_id=conversation or "") if download else None
        if download_turn is not None and on_event is not None:
            on_event({
                "type": "download_phase", "state": "queued", "attempt": download_attempt,
                "conversation_id": conversation, "phase": download_turn.phase,
            })
        if conv_lock is not None:
            conv_lock.acquire()
        try:
            # Counted before the seat wait so /healthz's queue depth includes
            # requests still blocked on a busy pool, as it did when this was one
            # global lock.
            self.waiting += 1
            started = time.monotonic()
            submission_deadline = started + min(timeout, chat.SUBMISSION_TIMEOUT)
            index = None
            try:
                index = self.pool.acquire_any(
                    deadline=submission_deadline, on_wait=on_event)
                if download_turn is not None and on_event is not None:
                    download_turn.phase = "opening"
                    on_event({
                        "type": "download_phase", "state": "opening",
                        "attempt": download_attempt, "conversation_id": conversation,
                        "phase": download_turn.phase,
                    })
                seat = self.pool.seat(index)
                if conversation:
                    m = seat.open()
                    try:
                        chat.open_resume(m, conversation, expected_account)
                    except chat.ChatError as exc:
                        raise ConversationNotFound(str(exc)) from None
                else:
                    seat.new_conversation()
                    m = seat.open()
                if download:
                    result = chat.download_attachments_on(
                        m, conversation, expected_account=expected_account, out_dir=out_dir,
                        attempt=download_attempt, on_event=on_event, turn=download_turn,
                        conversation_opened=True)
                else:
                    cap = chat.detect_cap(m)
                    if cap:
                        self.pool.mark_capped(cap)
                        raise Capped(cap)
                    reply = chat.ask_on(
                        m, prompt, effort=effort, attachments=attachments,
                                    out_dir=out_dir, stamp=str(body.get("stamp") or ""),
                                    timeout=timeout, submission_deadline=submission_deadline,
                                    on_event=on_event)
                    result = {"text": reply.text, "thinking": reply.thinking,
                             "images": [str(p) for p in reply.images],
                             "files": [str(p) for p in reply.files],
                             "conversation": reply.url,
                             "conversation_id": reply.conversation_id,
                             "account": reply.account, "title": reply.title}
            finally:
                if index is not None:
                    self.pool.release(index)
                self.waiting -= 1
                self.durations.append(time.monotonic() - started)
                del self.durations[:-50]
        finally:
            if conv_lock is not None:
                conv_lock.release()
        return result

    def _bootstrap_registry(self) -> dict:
        """The daemon's explicit sidebar migration read; registry writes stay in CLI."""
        self.waiting += 1
        index = self.pool.acquire_any()
        try:
            m = self.pool.seat(index).open()
            m.navigate("https://chatgpt.com/")
            chat.wait_ready(m)
            return {"account": chat.authenticated_account(m), "threads": chat.list_all_threads(m)}
        finally:
            self.pool.release(index)
            self.waiting -= 1

    def _list(self, body: dict) -> dict:
        """§4c: read-only sidebar listing — any free seat, no conversation lock."""
        self.waiting += 1
        index = self.pool.acquire_any()
        try:
            seat = self.pool.seat(index)
            m = seat.open()
            m.navigate("https://chatgpt.com/")
            chat.wait_ready(m)
            if body.get("list_all"):
                items = chat.list_all_threads(m)
            elif body.get("search") is not None:
                items = chat.search_threads(m, str(body["search"]))
            else:
                items = chat.list_threads(m, limit=int(body.get("limit") or 10))
        finally:
            self.pool.release(index)
            self.waiting -= 1
        return {"threads": items}

    def _serve(self, key, system, tools, history, effort, index: int, lock_holder: list) -> dict:
        conv = self.registry.get(key)
        preamble = protocol.render_preamble(system, tools)
        fresh = conv is None
        if fresh:
            conv = self._open(key, preamble, effort, index, lock_holder)
            text = translate.deliver_all(conv, history)
        else:
            try:
                text = translate.deliver(conv, history)
            except translate.ReplayNeeded:
                conv = self._open(key, preamble, effort, index, lock_holder)
                text = translate.deliver_all(conv, history)
                fresh = True
        return self._exchange(conv, text, tools, effort, index, tools if fresh else [])

    def _open(self, key, preamble, effort, index: int, lock_holder: list):
        seat = self.pool.seat(index)
        url = seat.new_conversation()
        conv = self.registry.open(key, preamble, effort, opened_at=time.time(), url=url,
                                  seat=index)
        conv.pending_preamble = True  # type: ignore[attr-defined]
        # The old conv (if any) is abandoned by this reopen — its uuid lock no
        # longer guards anything real; swap to the new uuid's lock instead. A
        # brand-new uuid is uncontended by construction (nobody else can name
        # it yet), so acquiring it here, after the seat, is safe.
        if lock_holder[0] is not None:
            lock_holder[0].release()
            lock_holder[0] = None
        new_lock = self._conv_uuid_lock(url)
        if new_lock is not None:
            new_lock.acquire()
            lock_holder[0] = new_lock
        return conv

    def _exchange(self, conv, text: str, tools: list[dict], effort: str, index: int, preamble_tools: list[dict] | None = None) -> dict:
        if getattr(conv, "pending_preamble", False):
            preamble = f"{conv.preamble}\n\n"
            text = f"{preamble}{text}"
            preamble_input = translate.estimate_input([], preamble_tools or [], rendered=preamble)
            translate.check_context(conv, preamble_input)
            conv.estimated_input += preamble_input
            conv.pending_preamble = False  # type: ignore[attr-defined]
        reply = self._send(conv, text, effort, index)
        parsed = protocol.parse_reply(reply["blocks"], tools, reply["text"])
        if isinstance(parsed, Malformed):
            correction = protocol.correction_prompt(parsed.reason)
            translate.check_context(conv, translate.estimate_input([], rendered=correction))
            conv.estimated_input += translate.estimate_input([], rendered=correction)
            reply = self._send(conv, correction, effort, index)
            parsed = protocol.parse_reply(reply["blocks"], tools, reply["text"])
            if isinstance(parsed, Malformed):
                raise ProtocolBroken(parsed.reason)
        conv.turns += 1
        visible_output = reply.get("text", "")
        for block in reply.get("blocks", []):
            if block not in visible_output:
                visible_output += f"\n{block}"
        conv.estimated_output += translate.estimate_tokens(visible_output)
        return self._render(parsed, conv)

    def _send(self, conv, text: str, effort: str, index: int) -> dict:
        seat = self.pool.seat(index)
        try:
            return seat.turn(text, effort, timeout=TURN_TIMEOUT)
        except Capped as exc:
            self.pool.mark_capped(exc.state)
            raise
        except MarionetteError:
            seat.rebuild()
            try:
                return seat.turn(text, effort, timeout=TURN_TIMEOUT)
            except Capped as exc:
                self.pool.mark_capped(exc.state)
                raise

    def _render(self, parsed, conv) -> dict:
        base = {"id": f"chatcmpl-{uuid.uuid4().hex[:16]}", "object": "chat.completion",
                "created": int(time.time()), "model": MODEL_PREFIX + conv.effort,
                "usage": translate.usage(conv)}
        if isinstance(parsed, ToolCall):
            call_id = f"call_{conv.turns}_{uuid.uuid4().hex[:8]}"
            message = {"role": "assistant", "content": None, "tool_calls": [
                {"id": call_id, "type": "function",
                 "function": {"name": parsed.name,
                              "arguments": json.dumps(parsed.arguments)}}]}
            finish = "tool_calls"
        else:
            message = {"role": "assistant", "content": parsed.text}
            finish = "stop"
        base["choices"] = [{"index": 0, "message": message, "finish_reason": finish}]
        return base


def keepalive_chunk(model: str, first: bool = False) -> dict:
    """An empty-but-well-formed delta.

    SSE comments (`: ping`) are legal but a translating proxy in the middle parses
    every data event and cancels the stream when nothing arrives for a browser turn
    (10-120s). A real chunk survives the translation.
    """
    delta = {"role": "assistant", "content": ""} if first else {"content": ""}
    return {"id": "chatcmpl-keepalive", "object": "chat.completion.chunk",
            "created": int(time.time()), "model": model,
            "choices": [{"index": 0, "delta": delta, "finish_reason": None}]}


def as_chunk(completion: dict) -> dict:
    """The whole answer as one streamed delta — the browser has no token stream."""
    choice = completion["choices"][0]
    return {"id": completion["id"], "object": "chat.completion.chunk",
            "created": completion["created"], "model": completion["model"],
            "choices": [{"index": 0, "finish_reason": choice["finish_reason"],
                         "delta": choice["message"]}]}


# A pre-ask_on failure (cap already spent, unknown conversation, bad args) never
# reaches chat.py's own event emitter, so its terminal `error` event is synthesised
# here from the HTTP status `_run` already classified it into.
_ASK_STATUS_CLASS = {400: "chat", 429: "capped", 404: "not_found", 503: "marionette",
                     502: "protocol"}


SSE_QUEUE_MAXSIZE = 64


class SSEQueue:
    """Lossless ordered controls plus bounded contiguous delta ranges."""

    _DELTA_TYPES = {"thinking", "answer", "replace"}

    def __init__(self, maxsize: int = SSE_QUEUE_MAXSIZE):
        if maxsize < 1:
            raise ValueError("maxsize must be positive")
        self._maxsize = maxsize
        self._controls: deque[tuple[int, dict]] = deque()
        self._deltas: deque[tuple[int, dict, str, str]] = deque()
        self._condition = threading.Condition()
        self._sequence = 0
        self._thinking = ""
        self._answer = ""

    def put(self, event: dict) -> None:
        with self._condition:
            self._sequence += 1
            stored = {**event, "seq": self._sequence}
            self._apply(stored)
            if stored.get("type") in self._DELTA_TYPES:
                self._deltas.append(
                    (self._sequence, stored, self._thinking, self._answer))
                self._compact_deltas()
            else:
                self._controls.append((self._sequence, stored))
            self._condition.notify()

    def _apply(self, event: dict) -> None:
        if event.get("type") == "thinking":
            self._thinking += event.get("delta", "")
        elif event.get("type") == "answer":
            self._answer += event.get("delta", "")
        elif event.get("type") == "replace":
            self._thinking = event.get("thinking", self._thinking)
            self._answer = event.get("answer", self._answer)

    def _compact_deltas(self) -> None:
        while len(self._deltas) > self._maxsize:
            entries = list(self._deltas)
            compacted = False
            control_sequences = {sequence for sequence, _ in self._controls}
            for index in range(len(entries) - 1):
                left, right = entries[index], entries[index + 1]
                if any(left[0] < sequence < right[0] for sequence in control_sequences):
                    continue
                sequence, _, thinking, answer = right
                replacement = {
                    "type": "replace", "phase": right[1].get("phase", "streaming"),
                    "thinking": thinking, "answer": answer, "seq": sequence,
                }
                entries[index:index + 2] = [(sequence, replacement, thinking, answer)]
                self._deltas = deque(entries)
                compacted = True
                break
            if compacted:
                continue
            sequence, _, thinking, answer = self._deltas.popleft()
            replacement = {
                "type": "replace", "phase": "streaming", "thinking": thinking,
                "answer": answer, "seq": sequence,
            }
            controls = list(self._controls)
            controls.append((sequence, replacement))
            controls.sort(key=lambda entry: entry[0])
            self._controls = deque(controls)

    def get(self, timeout: float):
        deadline = time.monotonic() + timeout
        with self._condition:
            while not self._controls and not self._deltas:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise queue.Empty
                self._condition.wait(remaining)
            if not self._controls:
                return self._deltas.popleft()[1]
            if not self._deltas:
                return self._controls.popleft()[1]
            if self._controls[0][0] < self._deltas[0][0]:
                return self._controls.popleft()[1]
            return self._deltas.popleft()[1]


class Handler(BaseHTTPRequestHandler):
    engine: Engine
    token: str
    protocol_version = "HTTP/1.1"

    def log_message(self, fmt, *args):
        sys.stderr.write("solwebd %s\n" % (fmt % args))

    def _send_json(self, status: int, payload: dict, headers: dict | None = None) -> None:
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        for name, value in (headers or {}).items():
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(body)

    def _authed(self) -> bool:
        if self.headers.get("Authorization") == f"Bearer {self.token}":
            return True
        self.close_connection = True
        self._send_json(401, {"error": {"message": "missing or wrong bearer token"}})
        return False

    def _content_length(self) -> int:
        if self.headers.get_all("Transfer-Encoding"):
            raise ValueError("Transfer-Encoding is not supported")
        values = self.headers.get_all("Content-Length") or []
        if len(values) != 1 or not re.fullmatch(r"(?:0|[1-9][0-9]*)", values[0].strip()):
            raise ValueError("one canonical Content-Length is required")
        length = int(values[0])
        if length > MAX_REQUEST_BODY:
            raise ValueError("request body is too large")
        return length

    def _read_body(self, length: int) -> bytes:
        body = self.rfile.read(length)
        if len(body) != length:
            self.close_connection = True
            raise ValueError("request body ended early")
        return body

    def do_GET(self):
        if not self._authed():
            return
        path = self.path.split("?")[0].rstrip("/")
        if path.endswith("/v1/models"):
            self._send_json(200, {"object": "list",
                                  "data": [{"id": m, "object": "model"} for m in MODELS]})
        elif path.endswith("/healthz"):
            self._send_json(200, self.engine.health())
        else:
            self._send_json(404, {"error": {"message": "not found"}})

    def do_POST(self):
        try:
            length = self._content_length()
        except ValueError as exc:
            self.close_connection = True
            self._send_json(400, {"error": {"message": str(exc)}})
            return
        if not self._authed():
            return
        try:
            raw_body = self._read_body(length)
        except ValueError as exc:
            self._send_json(400, {"error": {"message": str(exc)}})
            return
        path = self.path.split("?")[0].rstrip("/")
        if path.endswith("/shutdown"):
            self._send_json(200, {"ok": True})
            threading.Thread(target=self.server.shutdown, daemon=True).start()
            return
        if not (path.endswith("/chat/completions") or path.endswith("/v1/ask")):
            self._send_json(404, {"error": {"message": "not found"}})
            return
        client_session = self.headers.get("X-Overdeck-Client-Session")
        if client_session is not None and not CLIENT_SESSION_RE.fullmatch(client_session):
            self._send_json(400, {"error": {"message": "invalid X-Overdeck-Client-Session"}})
            return
        try:
            body = json.loads(raw_body or b"{}")
        except ValueError as exc:
            self._send_json(400, {"error": {"message": f"invalid json: {exc}"}})
            return
        if path.endswith("/v1/ask"):
            if body.get("stream"):
                self._stream_ask(body)
            else:
                status, payload, headers = self._run(body, self.engine.ask)
                self._send_json(status, payload, headers)
            return
        if body.get("stream"):
            self._stream(body, client_session)
        else:
            self._blocking(body, client_session)

    def _run(self, body: dict, call=None, client_session_id: str | None = None) -> tuple[int, dict, dict]:
        try:
            fn = call or (lambda value: self.engine.complete(value, client_session_id))
            return 200, fn(body), {}
        except ValueError as exc:
            return 400, {"error": {"message": str(exc)}}, {}
        except Capped as exc:
            failure = {"kind": "capped", "detail": str(exc), "retry_after_seconds": 900}
            if exc.state.get("resume_at"): failure["resume_at"] = exc.state["resume_at"]
            return 429, {"error": failure}, {"Retry-After": "900"}
        except ConversationNotFound as exc:
            return 404, {"error": {"message": str(exc)}}, {}
        except registry.RegistryError as exc:
            return 503, {"error": {"message": f"conversation registry unavailable: {exc}"}}, {}
        except (LoggedOut, MarionetteError, chat.ChatError) as exc:
            return 503, {"error": {"kind":"unavailable", "detail": f"chatgpt session unavailable: {exc}"}}, {}
        except ProtocolBroken as exc:
            return 502, {"error": {"kind":"protocol", "detail": f"model would not hold the protocol: {exc}"}}, {}
        except translate.ContextOverflow as exc:
            return 400, {"error": {"kind":"context_overflow", "code": "context_length_exceeded", "detail": str(exc)}}, {}

    def _blocking(self, body: dict, client_session_id: str | None = None) -> None:
        status, payload, headers = self._run(body, client_session_id=client_session_id)
        self._send_json(status, payload, headers)

    def _stream(self, body: dict, client_session_id: str | None = None) -> None:
        result: dict = {}
        worker = threading.Thread(target=lambda: result.update(
            zip(("status", "payload", "headers"), self._run(body, client_session_id=client_session_id))), daemon=True)
        worker.start()
        # A turn takes 10-20s; a silent socket for that long reads as a hung provider.
        worker.join(timeout=KEEPALIVE_SECONDS)
        if worker.is_alive():
            model = str(body.get("model", "")).split(",")[-1].strip() or MODEL_PREFIX + "medium"
            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()
            first = True
            while worker.is_alive():
                self._write_event(keepalive_chunk(model, first))
                first = False
                worker.join(timeout=KEEPALIVE_SECONDS)
            self._write_stream_tail(result)
            return
        if result.get("status") != 200:
            self._send_json(result["status"], result["payload"], result.get("headers"))
            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._write_stream_tail(result)

    def _write_event(self, payload: dict) -> None:
        self.wfile.write(f"data: {json.dumps(payload)}\n\n".encode())
        self.wfile.flush()

    def _stream_ask(self, body: dict) -> None:
        """The `/v1/ask` wire schema (frozen, §6): one `data: <TurnEvent JSON>` line
        per event, `: ka` comments for keepalive, no `event:`/`id:`/`retry:`, no
        `[DONE]`. The daemon never writes conversation logs — that stays CLI-only."""
        sink = SSEQueue()
        result: dict = {}
        sent_terminal = False

        def run() -> None:
            result["status"], result["payload"], result["headers"] = self._run(
                body, lambda b: self.engine.ask(b, on_event=sink.put))

        worker = threading.Thread(target=run, daemon=True)
        worker.start()

        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()

        while True:
            try:
                event = sink.get(timeout=KEEPALIVE_SECONDS)
            except queue.Empty:
                if not worker.is_alive():
                    break
                self.wfile.write(b": ka\n\n")
                self.wfile.flush()
                continue
            self._write_event(event)
            if event.get("type") in ("done", "error"):
                sent_terminal = True
                break
        worker.join()
        if not sent_terminal and result.get("status") != 200:
            payload = result.get("payload", {})
            failure = payload.get("error", {})
            self._write_event({
                "type": "error", "phase": "attaching",
                "class": _ASK_STATUS_CLASS.get(result.get("status"), "chat"),
                "detail": failure.get("detail", failure.get("message", "failed")),
                "provider_failure": failure,
                "partial": {"phase": "attaching", "thinking": "", "answer": "",
                            "saved_artifacts": []},
            })

    def _write_stream_tail(self, result: dict) -> None:
        if result.get("status") != 200:
            # HTTP may already be committed by keepalives; retain the original
            # status and headers in the terminal OpenAI error event.
            self._write_event({"error": result.get("payload", {}).get("error", {"message": "failed"}),
                               "provider_status": result.get("status"),
                               "provider_headers": result.get("headers", {})})
        else:
            self._write_event(as_chunk(result["payload"]))
        self.wfile.write(b"data: [DONE]\n\n")
        self.wfile.flush()


def serve(port: int = DEFAULT_PORT, mode: str = "virtual",
         seat: BrowserSeat | None = None, seats: int = 1):
    """`seat` is the single-seat back-compat/test path; it names one exact seat, so
    it is incompatible with asking for more than one — fail closed rather than
    silently discard it and build real Sessions in its place."""
    if seat is not None and seats != 1:
        raise ValueError("serve(seat=...) is a single-seat path; pass seats=1 or drop seat")
    pool = SeatPool.wrapping(seat) if seat is not None else SeatPool(max_seats=seats, mode=mode)
    engine = Engine(pool)
    handler = type("BoundHandler", (Handler,), {"engine": engine, "token": read_token()})
    # 127.0.0.1 only: this endpoint drives the owner's real account.
    server = ThreadingHTTPServer(("127.0.0.1", port), handler)
    return server, engine


def _publish_runtime() -> str:
    """Publish a private generation receipt for solwebctl ownership checks."""
    STATE.mkdir(parents=True, exist_ok=True)
    generation = str(uuid.uuid4())
    payload = {"pid": os.getpid(), "process_start_ticks": _process_start_ticks(os.getpid()),
               "instance_generation": generation, "version": BUILD_ID}
    tmp = STATE / f"runtime.json.{uuid.uuid4().hex}"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(payload, f); f.flush(); os.fsync(f.fileno())
    os.replace(tmp, STATE / "runtime.json")
    return generation


def main() -> int:
    parser = argparse.ArgumentParser(prog="solwebd")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT)
    parser.add_argument("--mode", default="virtual", choices=Session.MODES)
    parser.add_argument("--seats", type=int, default=1,
                        help="parallel browser seats (default 1, today's behavior)")
    args = parser.parse_args()
    if not 1 <= args.seats <= browser.MAX_SLOTS:
        parser.error(f"--seats must be between 1 and {browser.MAX_SLOTS}")
    server, engine = serve(port=args.port, mode=args.mode, seats=args.seats)
    engine.instance_generation = _publish_runtime()
    engine.probe_health()
    print(json.dumps({"listening": f"127.0.0.1:{args.port}", "models": MODELS,
                      "seats": args.seats, "token_file": str(TOKEN_FILE)}), flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        engine.pool.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())