//! The browser a thread shares with its agent: a headless Chrome run by
//! Vercel's `agent-browser` CLI, one session per thread. The agent drives it
//! with CLI commands. The app runs the same commands for the panel's buttons,
//! watches the daemon's viewport stream (JPEG frames over a WebSocket on
//! localhost) and sends the user's mouse and keyboard input back over it.

use std::{
    collections::HashSet,
    io::ErrorKind,
    net::{SocketAddr, TcpStream},
    path::{Path, PathBuf},
    process::{Command, Stdio},
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
        mpsc,
    },
    thread,
    time::{Duration, Instant},
};

use anyhow::{Context as _, Result, anyhow};
use base64::Engine as _;
use gpui_kit::RenderImage;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tungstenite::{Message, WebSocket};

use crate::claude::is_executable;

pub const INSTALL_COMMAND: &str = "npm install -g agent-browser && agent-browser install";
const BINARY: &str = "agent-browser";
/// The stream sends at most this many frames a second.
const MAX_FPS: u32 = 30;
/// How long a socket read waits before the stream thread sends queued input.
const POLL: Duration = Duration::from_millis(10);
/// How often the stream thread looks for a daemon while none runs.
const WAIT: Duration = Duration::from_millis(500);
/// A command the panel ran shows up in the stream within this long, if the
/// stream was connected when it started; a later one is the agent's.
const OWN_COMMAND: Duration = Duration::from_secs(3);

/// The agent-browser session of a thread.
pub fn session_name(thread_id: &str) -> String {
    format!("vectrum-{thread_id}")
}

/// Where `agent-browser` is: PATH, then the usual npm, bun and Homebrew bin folders.
pub fn find_binary() -> Option<PathBuf> {
    find_program(BINARY)
}

/// Where a command-line program is: PATH, then the usual npm, bun, nvm and
/// Homebrew bin folders, since desktop launchers often start apps without
/// the user's shell PATH.
pub fn find_program(name: &str) -> Option<PathBuf> {
    let on_path = std::env::var_os("PATH")
        .into_iter()
        .flat_map(|paths| std::env::split_paths(&paths).collect::<Vec<_>>())
        .map(|dir| dir.join(name));
    let home = std::env::var_os("HOME").map(PathBuf::from).unwrap_or_default();
    let mut fallbacks: Vec<PathBuf> = [".npm-global/bin", ".local/bin", ".bun/bin", ".volta/bin", ".local/share/pnpm"]
        .iter()
        .map(|dir| home.join(dir).join(name))
        .collect();
    fallbacks.extend(
        ["/home/linuxbrew/.linuxbrew/bin", "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]
            .iter()
            .map(|dir| Path::new(dir).join(name)),
    );
    // nvm keeps one bin directory per Node version.
    if let Ok(versions) = std::fs::read_dir(home.join(".nvm/versions/node")) {
        fallbacks.extend(versions.flatten().map(|version| version.path().join("bin").join(name)));
    }
    on_path.chain(fallbacks).find(|path| is_executable(path))
}

/// The folder agent-browser keeps its daemons' sockets, pid and port files in.
fn run_dir() -> PathBuf {
    let var = |name: &str| std::env::var_os(name).filter(|value| !value.is_empty());
    let base = var("AGENT_BROWSER_SOCKET_DIR").map(PathBuf::from).unwrap_or_else(|| match var("XDG_RUNTIME_DIR") {
        Some(dir) => PathBuf::from(dir).join("agent-browser"),
        None => var("HOME").map(PathBuf::from).unwrap_or_else(std::env::temp_dir).join(".agent-browser"),
    });
    match std::env::var("AGENT_BROWSER_NAMESPACE") {
        Ok(namespace) if !namespace.is_empty() => base.join("namespaces").join(namespace).join("run"),
        _ => base,
    }
}

fn pid_alive(pid: i32) -> bool {
    pid > 0 && unsafe { libc::kill(pid, 0) } == 0
}

/// Whether a session's daemon runs, from its pid file.
pub fn is_running(session: &str) -> bool {
    std::fs::read_to_string(run_dir().join(format!("{session}.pid")))
        .ok()
        .and_then(|pid| pid.trim().parse().ok())
        .is_some_and(pid_alive)
}

/// The sessions whose daemons run.
pub fn running_sessions() -> HashSet<String> {
    let Ok(entries) = std::fs::read_dir(run_dir()) else {
        return HashSet::new();
    };
    entries
        .flatten()
        .filter_map(|entry| {
            let name = entry.file_name().into_string().ok()?;
            let session = name.strip_suffix(".pid")?;
            let pid = std::fs::read_to_string(entry.path()).ok()?.trim().parse().ok()?;
            pid_alive(pid).then(|| session.to_string())
        })
        .collect()
}

/// The localhost port a running daemon streams its viewport on. Every daemon
/// opens one; the daemon writes it next to its socket.
fn stream_port(session: &str) -> Option<u16> {
    if !is_running(session) {
        return None;
    }
    std::fs::read_to_string(run_dir().join(format!("{session}.stream"))).ok()?.trim().parse().ok()
}

/// What the agent is told about the panel.
pub fn agent_prompt(session: &str) -> String {
    format!(
        "The user's app has a browser panel showing a shared headless browser that you drive with the \
         `agent-browser` CLI. Its session is `{session}`, already set in AGENT_BROWSER_SESSION, so plain \
         `agent-browser <command>` uses it; don't pass another --session. The user watches the browser \
         live and may click and type in it too. Core commands: `agent-browser open <url>`, \
         `agent-browser snapshot -i` (interactive elements with @refs), `agent-browser click @e2`, \
         `agent-browser fill @e3 \"text\"`, `agent-browser screenshot [path]`, `agent-browser get url`, \
         `agent-browser back|forward|reload`. `agent-browser skills get core` has the full guide. Use it \
         to check web pages you build or need to see, and leave the page open when you're done so the \
         user can look at it."
    )
}

/// The browser instructions for jcode, whose tools run in its daemon rather
/// than under the thread's process, so the session and binary are spelled out.
pub fn jcode_prompt(session: &str, binary: &Path) -> String {
    let binary = binary.display();
    format!(
        "The user's app has a browser panel showing a shared headless browser that you drive with the \
         `agent-browser` CLI at `{binary}`. Its session is `{session}`; pass `--session {session}` to every \
         command. The user watches the browser live and may click and type in it too. Core commands: \
         `open <url>`, `snapshot -i` (interactive elements with @refs), `click @e2`, `fill @e3 \"text\"`, \
         `screenshot [path]`, `get url`, `back|forward|reload`; `skills get core` has the full guide. Use it \
         to check web pages you build or need to see, and leave the page open when you're done so the user \
         can look at it. Prefer it over your own browser tool for pages the user should see."
    )
}

/// Sets a thread's browser session in an agent's environment, with
/// `agent-browser`'s folder on PATH, and returns the prompt that tells the
/// agent about it. `None` when agent-browser isn't installed.
pub fn prepare_agent(command: &mut Command, session: &str) -> Option<String> {
    let binary = find_binary()?;
    command.env("AGENT_BROWSER_SESSION", session);
    if let Some(dir) = binary.parent() {
        let current = command
            .get_envs()
            .find(|(key, _)| *key == "PATH")
            .and_then(|(_, value)| value.map(ToOwned::to_owned))
            .or_else(|| std::env::var_os("PATH"))
            .unwrap_or_default();
        let mut paths: Vec<PathBuf> = std::env::split_paths(&current).collect();
        if !paths.iter().any(|path| path == dir) {
            paths.push(dir.to_path_buf());
            if let Ok(joined) = std::env::join_paths(paths) {
                command.env("PATH", joined);
            }
        }
    }
    Some(agent_prompt(session))
}

/// Closes a session's browser and daemon, if it runs.
pub fn stop(session: &str) {
    if is_running(session)
        && let Some(binary) = find_binary()
    {
        let _ = Command::new(binary)
            .args(["--session", session, "close"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status();
    }
}

/// What the user typed into the address field, as a URL: `https://` is added
/// when there's no scheme, `http://` for local servers.
pub fn normalize_url(input: &str) -> Option<String> {
    let input = input.trim();
    if input.is_empty() {
        return None;
    }
    if input.contains("://") || ["about:", "data:", "file:", "chrome:"].iter().any(|scheme| input.starts_with(scheme)) {
        return Some(input.to_string());
    }
    let host = input.split(['/', '?', '#']).next().unwrap_or_default();
    let host = host
        .rsplit_once(':')
        .filter(|(_, port)| port.chars().all(|c| c.is_ascii_digit()))
        .map_or(host, |(host, _)| host);
    let local =
        matches!(host, "localhost" | "0.0.0.0" | "[::1]") || host.starts_with("127.") || host.ends_with(".localhost");
    Some(format!("{}://{input}", if local { "http" } else { "https" }))
}

// ---- the stream protocol --------------------------------------------------

/// A message the daemon's stream sends.
#[derive(Debug, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum StreamMessage {
    /// A screencast frame: a base64 JPEG of the viewport.
    Frame { data: String, metadata: FrameMetadata },
    /// The active tab navigated.
    Url { url: String },
    /// Sent on connect and when the browser or screencast starts or stops.
    Status {
        /// Whether a browser runs; the daemon can run without one.
        #[serde(default)]
        connected: bool,
    },
    /// A command the daemon started running, from any client.
    Command { action: String },
    #[serde(other)]
    Other,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FrameMetadata {
    /// The viewport in CSS pixels, the space input coordinates are in.
    pub device_width: f32,
    pub device_height: f32,
}

/// Input sent to the page over the stream; the daemon passes it on to
/// Chrome's `Input.dispatchMouseEvent` and `Input.dispatchKeyEvent`.
#[derive(Clone, Debug, Serialize)]
#[serde(tag = "type")]
pub enum Input {
    #[serde(rename = "input_mouse")]
    Mouse(MouseInput),
    #[serde(rename = "input_keyboard")]
    Key(KeyInput),
}

/// CDP's modifier bits.
pub mod modifiers {
    pub const ALT: u8 = 1;
    pub const CTRL: u8 = 2;
    pub const META: u8 = 4;
    pub const SHIFT: u8 = 8;
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum MouseEventType {
    #[serde(rename = "mouseMoved")]
    Move,
    #[serde(rename = "mousePressed")]
    Down,
    #[serde(rename = "mouseReleased")]
    Up,
    #[serde(rename = "mouseWheel")]
    Wheel,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum PointerButton {
    None,
    Left,
    Middle,
    Right,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MouseInput {
    pub event_type: MouseEventType,
    pub x: f64,
    pub y: f64,
    pub button: PointerButton,
    pub click_count: u32,
    pub delta_x: f64,
    pub delta_y: f64,
    pub modifiers: u8,
}

impl MouseInput {
    pub fn new(event_type: MouseEventType, (x, y): (f64, f64), modifiers: u8) -> Self {
        Self { event_type, x, y, button: PointerButton::None, click_count: 0, delta_x: 0., delta_y: 0., modifiers }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum KeyEventType {
    /// A key press that types `text`.
    #[serde(rename = "keyDown")]
    Down,
    /// A key press that types nothing (arrows, shortcuts).
    #[serde(rename = "rawKeyDown")]
    RawDown,
    #[serde(rename = "keyUp")]
    Up,
    /// Text typed without a key, like a paste.
    #[serde(rename = "char")]
    Char,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct KeyInput {
    pub event_type: KeyEventType,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub key: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub code: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub text: Option<String>,
    pub windows_virtual_key_code: u32,
    pub modifiers: u8,
}

/// A GPUI key press or release as a DOM key event. `key` is GPUI's key name
/// (`a`, `enter`, `left`), `typed` the character it types, if any.
pub fn key_input(key: &str, typed: Option<&str>, modifiers: u8, down: bool) -> KeyInput {
    let named = |dom: &str, code: &str, key_code: u32| (dom.to_string(), Some(code.to_string()), key_code);
    let (dom_key, code, key_code) = match key {
        "enter" => named("Enter", "Enter", 13),
        "backspace" => named("Backspace", "Backspace", 8),
        "tab" => named("Tab", "Tab", 9),
        "escape" => named("Escape", "Escape", 27),
        "space" => named(" ", "Space", 32),
        "left" => named("ArrowLeft", "ArrowLeft", 37),
        "up" => named("ArrowUp", "ArrowUp", 38),
        "right" => named("ArrowRight", "ArrowRight", 39),
        "down" => named("ArrowDown", "ArrowDown", 40),
        "home" => named("Home", "Home", 36),
        "end" => named("End", "End", 35),
        "pageup" => named("PageUp", "PageUp", 33),
        "pagedown" => named("PageDown", "PageDown", 34),
        "delete" => named("Delete", "Delete", 46),
        "insert" => named("Insert", "Insert", 45),
        _ if key.len() > 1 && key.starts_with('f') && key[1..].parse::<u32>().is_ok_and(|n| (1..=12).contains(&n)) => {
            let n: u32 = key[1..].parse().unwrap_or(1);
            (format!("F{n}"), Some(format!("F{n}")), 111 + n)
        }
        _ => {
            let mut chars = key.chars();
            match (chars.next(), chars.next()) {
                (Some(c), None) if c.is_ascii_alphabetic() => {
                    let upper = c.to_ascii_uppercase();
                    (typed.unwrap_or(key).to_string(), Some(format!("Key{upper}")), upper as u32)
                }
                (Some(c), None) if c.is_ascii_digit() => {
                    (typed.unwrap_or(key).to_string(), Some(format!("Digit{c}")), c as u32)
                }
                _ => (typed.unwrap_or(key).to_string(), None, 0),
            }
        }
    };
    let shortcut = modifiers & (modifiers::CTRL | modifiers::META | modifiers::ALT) != 0;
    // Enter types a carriage return, which is what submits forms.
    let text = match key {
        "enter" => Some("\r".to_string()),
        "space" => Some(" ".to_string()),
        _ => typed.filter(|text| !text.is_empty() && !text.chars().any(char::is_control)).map(str::to_string),
    }
    .filter(|_| !shortcut);
    let event_type = match (down, &text) {
        (false, _) => KeyEventType::Up,
        (true, Some(_)) => KeyEventType::Down,
        (true, None) => KeyEventType::RawDown,
    };
    KeyInput {
        event_type,
        key: Some(dom_key),
        code,
        text: if down { text } else { None },
        windows_virtual_key_code: key_code,
        modifiers,
    }
}

/// Text typed at the caret without keys, as a paste does: one `char` event
/// per character, which is all Chrome takes from one.
pub fn text_input(text: &str) -> Vec<KeyInput> {
    text.chars()
        .filter(|c| *c != '\r')
        .map(|c| KeyInput {
            event_type: KeyEventType::Char,
            key: None,
            code: None,
            text: Some(if c == '\n' { "\r".to_string() } else { c.to_string() }),
            windows_virtual_key_code: 0,
            modifiers: 0,
        })
        .collect()
}

/// Where a frame of `image` pixels is drawn when fit inside `area` keeping its
/// aspect ratio and centered, as GPUI's `ObjectFit::Contain` draws it:
/// `(x, y, width, height)` relative to the area.
pub fn contain(area: (f32, f32), image: (f32, f32)) -> (f32, f32, f32, f32) {
    let (area_w, area_h) = area;
    let (image_w, image_h) = image;
    if image_w <= 0. || image_h <= 0. || area_w <= 0. || area_h <= 0. {
        return (0., 0., 0., 0.);
    }
    let (w, h) = if area_w / area_h > image_w / image_h {
        (image_w * area_h / image_h, area_h)
    } else {
        (area_w, image_h * area_w / image_w)
    };
    ((area_w - w) / 2., (area_h - h) / 2., w, h)
}

/// Maps a point in the viewport area (relative to its top left) to page
/// coordinates in CSS pixels. `None` outside the drawn frame.
pub fn page_point(point: (f32, f32), area: (f32, f32), image: (f32, f32), viewport: (f32, f32)) -> Option<(f64, f64)> {
    let (x, y, w, h) = contain(area, image);
    if w <= 0. || h <= 0. {
        return None;
    }
    let (rel_x, rel_y) = ((point.0 - x) / w, (point.1 - y) / h);
    if !(0. ..=1.).contains(&rel_x) || !(0. ..=1.).contains(&rel_y) {
        return None;
    }
    let page_x = (rel_x * viewport.0).min(viewport.0 - 1.).max(0.);
    let page_y = (rel_y * viewport.1).min(viewport.1 - 1.).max(0.);
    Some((page_x as f64, page_y as f64))
}

// ---- a session ------------------------------------------------------------

/// A decoded viewport frame.
pub struct Frame {
    pub image: Arc<RenderImage>,
    /// The frame's size in pixels.
    pub size: (f32, f32),
    /// The page's viewport in CSS pixels.
    pub viewport: (f32, f32),
}

/// What a session reports to the panel.
#[derive(Debug, PartialEq)]
pub enum BrowserEvent {
    /// `agent-browser` isn't installed.
    Missing,
    /// No daemon runs for the session.
    Waiting,
    /// Watching the daemon's stream; `browser` tells whether Chrome runs.
    Live {
        browser: bool,
    },
    /// A new frame waits in `BrowserSession::take_frame`.
    Frame,
    Url(String),
    /// The agent (or anyone but the panel) ran a command.
    Agent(String),
    /// One of the panel's commands failed.
    Failed(String),
}

struct Shared {
    stop: AtomicBool,
    /// Whether the panel shows, so the stream is worth watching.
    watching: AtomicBool,
    frame: Mutex<Option<Frame>>,
    /// Commands the panel ran that the stream hasn't echoed yet.
    own: Mutex<Vec<(&'static str, Instant)>>,
}

/// One thread's browser, as the panel drives and watches it.
pub struct BrowserSession {
    name: String,
    binary: Option<PathBuf>,
    shared: Arc<Shared>,
    events: smol::channel::Sender<BrowserEvent>,
    input: mpsc::Sender<String>,
}

impl BrowserSession {
    /// Starts watching a session. Nothing runs until `set_watching(true)`.
    pub fn new(name: String, events: smol::channel::Sender<BrowserEvent>) -> Self {
        let binary = find_binary();
        let shared = Arc::new(Shared {
            stop: AtomicBool::new(false),
            watching: AtomicBool::new(false),
            frame: Mutex::new(None),
            own: Mutex::new(Vec::new()),
        });
        let (input, input_rx) = mpsc::channel();
        if binary.is_some() {
            let (name, shared, events) = (name.clone(), shared.clone(), events.clone());
            let _ = thread::Builder::new()
                .name("browser-stream".into())
                .spawn(move || watch(&name, &shared, &events, &input_rx));
        } else {
            let _ = events.try_send(BrowserEvent::Missing);
        }
        Self { name, binary, shared, events, input }
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub fn installed(&self) -> bool {
        self.binary.is_some()
    }

    /// Connects to the stream while the panel shows, and lets go of it otherwise.
    pub fn set_watching(&self, watching: bool) {
        self.shared.watching.store(watching, Ordering::Relaxed);
    }

    pub fn open(&self, url: &str) {
        self.run("navigate", &["open", url]);
    }

    pub fn back(&self) {
        self.run("back", &["back"]);
    }

    pub fn forward(&self) {
        self.run("forward", &["forward"]);
    }

    pub fn reload(&self) {
        self.run("reload", &["reload"]);
    }

    /// Sizes the page's viewport, in CSS pixels at `scale` device pixels each.
    pub fn set_viewport(&self, width: u32, height: u32, scale: f32) {
        let (width, height, scale) = (width.to_string(), height.to_string(), format!("{scale:.2}"));
        self.run("viewport", &["set", "viewport", &width, &height, &scale]);
    }

    pub fn send(&self, input: Input) {
        if let Ok(json) = serde_json::to_string(&input) {
            let _ = self.input.send(json);
        }
    }

    /// The newest frame not taken yet.
    pub fn take_frame(&self) -> Option<Frame> {
        self.shared.frame.lock().ok()?.take()
    }

    /// Runs `agent-browser --session <name> <args> --json` in the background.
    fn run(&self, action: &'static str, args: &[&str]) {
        let Some(binary) = self.binary.clone() else {
            let _ = self.events.try_send(BrowserEvent::Missing);
            return;
        };
        if let Ok(mut own) = self.shared.own.lock() {
            own.retain(|(_, at)| at.elapsed() < OWN_COMMAND);
            own.push((action, Instant::now()));
        }
        let mut command = Command::new(binary);
        command.args(["--session", &self.name]).args(args).arg("--json");
        let events = self.events.clone();
        let _ = thread::Builder::new().name("browser-command".into()).spawn(move || {
            if let Err(err) = run_command(command) {
                let _ = events.send_blocking(BrowserEvent::Failed(format!("{err:#}")));
            }
        });
    }
}

impl Drop for BrowserSession {
    fn drop(&mut self) {
        self.shared.stop.store(true, Ordering::Relaxed);
    }
}

fn run_command(mut command: Command) -> Result<()> {
    let output = command
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .output()
        .context("failed to run agent-browser")?;
    let reply: Option<Value> = serde_json::from_slice(&output.stdout).ok();
    let success = reply.as_ref().and_then(|reply| reply["success"].as_bool()).unwrap_or(output.status.success());
    if success {
        return Ok(());
    }
    let message = reply
        .as_ref()
        .and_then(|reply| reply["error"].as_str().map(str::to_string))
        .or_else(|| {
            let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
            (!stderr.is_empty()).then_some(stderr)
        })
        .unwrap_or_else(|| format!("agent-browser exited with {}", output.status));
    Err(anyhow!(message))
}

/// The stream thread: waits for the session's daemon, watches its stream
/// while the panel shows, and reconnects when the daemon restarts.
fn watch(name: &str, shared: &Shared, events: &smol::channel::Sender<BrowserEvent>, input: &mpsc::Receiver<String>) {
    let mut waiting = false;
    while !shared.stop.load(Ordering::Relaxed) && !events.is_closed() {
        if !shared.watching.load(Ordering::Relaxed) {
            while input.try_recv().is_ok() {}
            thread::sleep(WAIT / 2);
            continue;
        }
        let socket = stream_port(name).map(connect);
        match socket {
            Some(Ok(mut socket)) => {
                waiting = false;
                stream(&mut socket, shared, events, input);
                let _ = socket.close(None);
                let _ = socket.flush();
            }
            Some(Err(_)) => thread::sleep(WAIT),
            None => {
                if !waiting {
                    waiting = true;
                    let _ = events.send_blocking(BrowserEvent::Waiting);
                }
                while input.try_recv().is_ok() {}
                thread::sleep(WAIT);
            }
        }
    }
}

fn connect(port: u16) -> Result<WebSocket<TcpStream>> {
    let address = SocketAddr::from(([127, 0, 0, 1], port));
    let tcp = TcpStream::connect_timeout(&address, Duration::from_secs(2))?;
    tcp.set_nodelay(true)?;
    // A daemon that just started or is shutting down may never answer.
    tcp.set_read_timeout(Some(Duration::from_secs(3)))?;
    let url = format!("ws://127.0.0.1:{port}/?maxFps={MAX_FPS}");
    let (socket, _) = tungstenite::client(url, tcp).map_err(|err| anyhow!("stream handshake failed: {err}"))?;
    socket.get_ref().set_read_timeout(Some(POLL))?;
    Ok(socket)
}

/// Relays one connection: frames and page changes to the panel, the user's
/// input to the page. Returns when the connection ends or isn't wanted.
fn stream(
    socket: &mut WebSocket<TcpStream>,
    shared: &Shared,
    events: &smol::channel::Sender<BrowserEvent>,
    input: &mpsc::Receiver<String>,
) {
    loop {
        if shared.stop.load(Ordering::Relaxed) || !shared.watching.load(Ordering::Relaxed) {
            return;
        }
        while let Ok(message) = input.try_recv() {
            if socket.send(Message::text(message)).is_err() {
                return;
            }
        }
        let text = match socket.read() {
            Ok(Message::Text(text)) => text,
            Ok(Message::Close(_)) => return,
            Ok(_) => continue,
            Err(tungstenite::Error::Io(err)) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
                continue;
            }
            Err(_) => return,
        };
        let Ok(message) = serde_json::from_str::<StreamMessage>(text.as_str()) else {
            continue;
        };
        let event = match message {
            StreamMessage::Frame { data, metadata, .. } => {
                let Ok(frame) = decode_frame(&data, &metadata) else {
                    continue;
                };
                let Ok(mut slot) = shared.frame.lock() else {
                    return;
                };
                // The panel hasn't drawn the last one yet; this one replaces it.
                if slot.replace(frame).is_some() {
                    continue;
                }
                BrowserEvent::Frame
            }
            StreamMessage::Url { url } => BrowserEvent::Url(url),
            StreamMessage::Status { connected } => BrowserEvent::Live { browser: connected },
            StreamMessage::Command { action } => {
                let own = shared.own.lock().ok().and_then(|mut own| {
                    let ix = own.iter().position(|(own, at)| *own == action && at.elapsed() < OWN_COMMAND)?;
                    Some(own.remove(ix))
                });
                if own.is_some() {
                    continue;
                }
                BrowserEvent::Agent(action)
            }
            StreamMessage::Other => continue,
        };
        if events.send_blocking(event).is_err() {
            return;
        }
    }
}

/// Decodes a frame's JPEG into the BGRA image GPUI draws.
fn decode_frame(data: &str, metadata: &FrameMetadata) -> Result<Frame> {
    let jpeg = base64::engine::general_purpose::STANDARD.decode(data)?;
    let mut pixels = image::load_from_memory_with_format(&jpeg, image::ImageFormat::Jpeg)?.into_rgba8();
    let size = (pixels.width() as f32, pixels.height() as f32);
    for pixel in pixels.chunks_exact_mut(4) {
        pixel.swap(0, 2);
    }
    Ok(Frame {
        image: Arc::new(RenderImage::new(vec![image::Frame::new(pixels)])),
        size,
        viewport: (metadata.device_width, metadata.device_height),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[::core::prelude::v1::test]
    fn parses_stream_messages() {
        let frame = json!({
            "type": "frame", "seq": 41, "data": "abc",
            "metadata": {"deviceWidth": 1280, "deviceHeight": 720, "pageScaleFactor": 1, "offsetTop": 0,
                "scrollOffsetX": 0, "scrollOffsetY": 0, "timestamp": 1785038682238u64},
        });
        match serde_json::from_value::<StreamMessage>(frame).unwrap() {
            StreamMessage::Frame { data, metadata } => {
                assert_eq!(data, "abc");
                assert_eq!((metadata.device_width, metadata.device_height), (1280., 720.));
            }
            other => panic!("unexpected {other:?}"),
        }
        let url = json!({"type": "url", "url": "https://example.com/", "timestamp": 1});
        assert!(
            matches!(serde_json::from_value(url).unwrap(), StreamMessage::Url { url } if url == "https://example.com/")
        );
        let status = json!({"type": "status", "connected": true, "screencasting": false, "viewportWidth": 1280,
            "viewportHeight": 720, "engine": "chrome", "recording": false});
        assert!(matches!(serde_json::from_value(status).unwrap(), StreamMessage::Status { connected: true }));
        let tabs = json!({"type": "tabs", "timestamp": 1, "tabs": [
            {"active": true, "label": null, "tabId": "t1", "title": "One", "type": "page", "url": "a"},
        ]});
        assert!(matches!(serde_json::from_value(tabs).unwrap(), StreamMessage::Other));
        let command = json!({"type": "command", "action": "navigate", "id": "r1", "params": {}, "timestamp": 1});
        assert!(
            matches!(serde_json::from_value(command).unwrap(), StreamMessage::Command { action } if action == "navigate")
        );
        let result = json!({"type": "result", "id": "r1", "action": "navigate", "success": true, "data": {}});
        assert!(matches!(serde_json::from_value(result).unwrap(), StreamMessage::Other));
    }

    #[::core::prelude::v1::test]
    fn serializes_input() {
        let mut click = MouseInput::new(MouseEventType::Down, (100., 200.), modifiers::SHIFT);
        click.button = PointerButton::Left;
        click.click_count = 1;
        assert_eq!(
            serde_json::to_value(Input::Mouse(click)).unwrap(),
            json!({"type": "input_mouse", "eventType": "mousePressed", "x": 100.0, "y": 200.0, "button": "left",
                "clickCount": 1, "deltaX": 0.0, "deltaY": 0.0, "modifiers": 8}),
        );
        let mut wheel = MouseInput::new(MouseEventType::Wheel, (1., 2.), 0);
        wheel.delta_y = 120.;
        let wheel = serde_json::to_value(Input::Mouse(wheel)).unwrap();
        assert_eq!((wheel["eventType"].as_str(), wheel["deltaY"].as_f64()), (Some("mouseWheel"), Some(120.)));
        assert_eq!(
            serde_json::to_value(Input::Key(key_input("a", Some("a"), 0, true))).unwrap(),
            json!({"type": "input_keyboard", "eventType": "keyDown", "key": "a", "code": "KeyA", "text": "a",
                "windowsVirtualKeyCode": 65, "modifiers": 0}),
        );
        assert_eq!(
            serde_json::to_value(Input::Key(text_input("p").remove(0))).unwrap(),
            json!({"type": "input_keyboard", "eventType": "char", "text": "p", "windowsVirtualKeyCode": 0,
                "modifiers": 0}),
        );
    }

    #[::core::prelude::v1::test]
    fn maps_keys() {
        let enter = key_input("enter", None, 0, true);
        assert_eq!(
            (enter.event_type, enter.key.as_deref(), enter.text.as_deref()),
            (KeyEventType::Down, Some("Enter"), Some("\r"))
        );
        assert_eq!(enter.windows_virtual_key_code, 13);
        let left = key_input("left", None, 0, true);
        assert_eq!(
            (left.event_type, left.key.as_deref(), left.windows_virtual_key_code),
            (KeyEventType::RawDown, Some("ArrowLeft"), 37)
        );
        let shifted = key_input("a", Some("A"), modifiers::SHIFT, true);
        assert_eq!(
            (shifted.key.as_deref(), shifted.code.as_deref(), shifted.text.as_deref()),
            (Some("A"), Some("KeyA"), Some("A"))
        );
        let shortcut = key_input("a", None, modifiers::CTRL, true);
        assert_eq!((shortcut.event_type, shortcut.text), (KeyEventType::RawDown, None));
        let up = key_input("a", Some("a"), 0, false);
        assert_eq!((up.event_type, up.text), (KeyEventType::Up, None));
        assert_eq!(key_input("f5", None, 0, true).windows_virtual_key_code, 116);
        assert_eq!(key_input("7", Some("7"), 0, true).code.as_deref(), Some("Digit7"));
        assert_eq!(key_input(",", Some(","), 0, true).code, None);
    }

    #[::core::prelude::v1::test]
    fn maps_points_onto_the_page() {
        // A 1280×720 frame in a 640×600 area: drawn 640×360, centered vertically.
        assert_eq!(contain((640., 600.), (1280., 720.)), (0., 120., 640., 360.));
        assert_eq!(page_point((320., 300.), (640., 600.), (1280., 720.), (1280., 720.)), Some((640., 360.)));
        assert_eq!(page_point((10., 50.), (640., 600.), (1280., 720.), (1280., 720.)), None);
        // A wide area pillarboxes; the frame may be scaled from the CSS viewport.
        assert_eq!(contain((1000., 360.), (640., 360.)), (180., 0., 640., 360.));
        assert_eq!(page_point((180., 0.), (1000., 360.), (640., 360.), (1280., 720.)), Some((0., 0.)));
        assert_eq!(page_point((820., 360.), (1000., 360.), (640., 360.), (1280., 720.)), Some((1279., 719.)));
        assert_eq!(page_point((5., 5.), (0., 0.), (1280., 720.), (1280., 720.)), None);
    }

    #[::core::prelude::v1::test]
    fn normalizes_urls() {
        assert_eq!(normalize_url("example.com").as_deref(), Some("https://example.com"));
        assert_eq!(normalize_url("  example.com/a?b ").as_deref(), Some("https://example.com/a?b"));
        assert_eq!(normalize_url("http://example.com").as_deref(), Some("http://example.com"));
        assert_eq!(normalize_url("localhost:3000/x").as_deref(), Some("http://localhost:3000/x"));
        assert_eq!(normalize_url("127.0.0.1:8080").as_deref(), Some("http://127.0.0.1:8080"));
        assert_eq!(normalize_url("app.localhost").as_deref(), Some("http://app.localhost"));
        assert_eq!(normalize_url("about:blank").as_deref(), Some("about:blank"));
        assert_eq!(normalize_url("file:///tmp/a.html").as_deref(), Some("file:///tmp/a.html"));
        assert_eq!(normalize_url("   "), None);
    }

    /// Drives a real session: `cargo test live_session -- --ignored`. Needs
    /// agent-browser and its Chrome, and the network for example.com.
    #[::core::prelude::v1::test]
    #[ignore]
    fn live_session() {
        let binary = find_binary().expect("agent-browser is not installed");
        let name = session_name(&format!("test-{}", std::process::id()));
        let cli = |args: &[&str]| -> Value {
            let output = Command::new(&binary).args(["--session", &name]).args(args).arg("--json").output().unwrap();
            serde_json::from_slice(&output.stdout).unwrap()
        };
        let (tx, rx) = smol::channel::unbounded();
        let session = BrowserSession::new(name.clone(), tx);
        session.set_watching(true);
        let deadline = Instant::now() + Duration::from_secs(60);
        let agent_actions = std::cell::RefCell::new(Vec::new());
        let next = || {
            let left = deadline.saturating_duration_since(Instant::now());
            let event = smol::block_on(smol::future::or(async { rx.recv().await.ok() }, async {
                smol::Timer::after(left).await;
                None
            }))
            .expect("timed out waiting for the browser");
            match &event {
                BrowserEvent::Failed(err) => panic!("{err}"),
                BrowserEvent::Agent(action) => agent_actions.borrow_mut().push(action.clone()),
                _ => {}
            }
            event
        };

        // No daemon yet; the panel's `open` starts one, and frames follow.
        assert_eq!(next(), BrowserEvent::Waiting);
        session.open("https://example.com");
        let frame = loop {
            if next() == BrowserEvent::Frame {
                break session.take_frame().expect("a frame");
            }
        };
        assert!(frame.size.0 > 0. && frame.size.1 > 0.);
        assert_eq!(frame.viewport, (1280., 720.));

        // A page with a link and a field at known places.
        let page = std::env::temp_dir().join(format!("vectrum-browser-test-{}.html", std::process::id()));
        std::fs::write(
            &page,
            r##"<!doctype html><title>test</title><body style="margin:0">
            <a href="#clicked" style="position:absolute;left:100px;top:100px">link</a>
            <input id="field" style="position:absolute;left:100px;top:200px;width:300px"></body>"##,
        )
        .unwrap();
        session.open(&format!("file://{}", page.display()));
        while !matches!(next(), BrowserEvent::Url(url) if url.ends_with(".html")) {}
        let click = |(x, y): (f64, f64)| {
            session.send(Input::Mouse(MouseInput::new(MouseEventType::Move, (x, y), 0)));
            for event_type in [MouseEventType::Down, MouseEventType::Up] {
                let mut click = MouseInput::new(event_type, (x, y), 0);
                click.button = PointerButton::Left;
                click.click_count = 1;
                session.send(Input::Mouse(click));
            }
        };
        click((110., 108.));
        while !matches!(next(), BrowserEvent::Url(url) if url.ends_with("#clicked")) {}

        // Typing and pasting into the field.
        click((150., 210.));
        for (key, typed) in [("h", "h"), ("i", "i")] {
            session.send(Input::Key(key_input(key, Some(typed), 0, true)));
            session.send(Input::Key(key_input(key, Some(typed), 0, false)));
        }
        for input in text_input(" there") {
            session.send(Input::Key(input));
        }
        thread::sleep(Duration::from_millis(500));
        assert_eq!(cli(&["get", "value", "#field"])["data"]["value"], "hi there");

        // Commands from anyone but the panel show as the agent's.
        while !agent_actions.borrow().iter().any(|action| action == "inputvalue") {
            next();
        }
        assert!(!agent_actions.borrow().iter().any(|action| action == "navigate"), "{agent_actions:?}");

        drop(session);
        let _ = std::fs::remove_file(page);
        stop(&name);
        let closing = Instant::now();
        while is_running(&name) && closing.elapsed() < Duration::from_secs(10) {
            thread::sleep(Duration::from_millis(100));
        }
        assert!(!is_running(&name));
    }

    #[::core::prelude::v1::test]
    fn names_sessions() {
        assert_eq!(session_name("18f-2a-0"), "vectrum-18f-2a-0");
        assert!(agent_prompt("vectrum-x").contains("`vectrum-x`"));
    }
}