cctop 0.28.0

//! `cctop serve` — the same table, in a browser.
//!
//! The dashboard's headline is which session is waiting on you, and that only
//! pays off while someone is looking at the terminal. Agents sit amber for
//! twenty minutes because the person who could answer them is in a meeting.
//! This is the same data on a surface they have with them: a page that streams
//! the rows over SSE, a per-session report that answers the question the live
//! table cannot — where an afternoon's money went — the conversation itself,
//! what the session has edited, what it is allowed to reach, a prompt box
//! for answering the agent that was waiting, and an analytics page for the
//! questions no single row answers: when the fleet works, and on what.
//!
//! # It can act, and that is a decision with a cost
//!
//! This started read-only, and the argument for that is still true: a bug in a
//! route that only reads is a disclosure, and the same bug in one that writes is
//! someone else's agent taking an action in your repository. What changed is
//! what the page is for. A dashboard that says an agent has been waiting twenty
//! minutes, on a phone, from a meeting, and cannot answer it, has shown you a
//! problem and withheld the fix. So [`actions`] can type a prompt at a live
//! session, resume a dead one, and hand one's work to a different harness —
//! nothing that destroys work, and nothing the terminal UI could not already do.
//!
//! The whole authorisation is the token in the URL, and that is worth stating
//! plainly rather than burying: **whoever holds the link can drive the agents on
//! this machine.** Hence the defaults — loopback, a token, and both of
//! `--no-token` and `--no-actions` able to take the acting half away. On top of
//! the token, an action must be a `POST` carrying a JSON body, which is what
//! stops a page in another tab from firing one with a token it cannot read; see
//! [`http`] for why that pairing and not a header of our own invention.
//!
//! # What guards the socket
//!
//! - **Loopback by default.** `--bind` is how it reaches the network and
//!   `--tunnel` is how it reaches the internet, and both say so on stderr when
//!   they do. Nobody exposes this by not reading a flag.
//! - **A token on every request**, generated per run, carried in the URL. The
//!   URL is therefore the credential, which is what makes the link shareable
//!   over whatever channel the user already trusts — and what makes `--no-token`
//!   a thing to justify rather than a convenience. It also gates the actions:
//!   no token, no acting, because there would be nothing left to authorise with.
//!   Beside it a second token is minted — the read-only link. It opens every
//!   page and every GET, and `/api/act/*` answers it 403: read-only is a
//!   property of the credential, not a claim a request can make about itself.
//!   `--token-file` keeps both across restarts; see [`tokens`]. The one path
//!   outside the gate is `GET /metrics`, which carries aggregates and no
//!   transcript — see [`metrics`] for why a scrape needs no secret.
//! - **Nothing destructive.** No route stops an agent, kills a process or
//!   deletes a transcript. Those stay in the terminal, where the confirmation
//!   prompt is.
//! - **Bounded everything.** [`MAX_CONNECTIONS`] threads, one snapshot shared
//!   between them, and reads deadlined in [`http`]. The expensive work — a full
//!   transcript parse — happens on the routes that need one, for one session, on
//!   request: the report, the conversation, and a handoff's brief.
//!
//! # Why a thread per connection
//!
//! Because there are at most a handful. This serves one person's browsers, and
//! an SSE stream is idle between snapshots, so the async runtime that would
//! make this scale is dependency weight bought against a load that does not
//! exist. The connection cap is what makes the arithmetic safe rather than
//! optimistic.
//!
//! ponytail: no TLS of its own. A loopback socket does not want it, and
//! terminating TLS for the LAN case means certificates cctop has no business
//! managing. `--tunnel` is the answer for reaching this from elsewhere — see
//! [`tunnel`], which borrows Cloudflare's certificate and Cloudflare's edge, and
//! is therefore encryption in transit rather than a private channel. For a LAN,
//! the tunnel the user already has (ssh, Tailscale) is still better than
//! anything here, because it authenticates rather than merely encrypting.

pub(crate) mod actions;
/// Cross-session aggregation behind `/api/analytics` — the data the
/// analytics page charts, built from the snapshot plus cached extractions.
mod analytics;
/// The conversation reader. Public to the crate because a handoff brief
/// carries what was said as well as what was done — see [`crate::handoff`].
pub mod chat;
/// Routes that only exist in a build that asked for them. Not a default
/// feature, so a released cctop contains none of this — see the module docs.
#[cfg(feature = "debug")]
mod debug;
/// A whole conversation as markdown, behind `/api/chat/<id>/markdown`.
/// Crate-visible because `cctop --export` prints the same document, and a
/// serve answers for a remote row by running that on the row's machine.
pub(crate) mod export;
mod http;
/// `/metrics`, the snapshot in Prometheus's text format.
mod metrics;
/// The `--notify` webhook. Crate-visible because the TUI POSTs crossings to
/// `$CCTOP_NOTIFY_URL` through the same send — one transport, two triggers.
pub(crate) mod notify;
mod quota;
/// The per-session postmortem behind `/api/report`. Crate-visible because
/// `cctop --report` prints the same document — the flag exists so a serve can
/// answer for a remote row by running it on the machine that has the file.
pub(crate) mod report;
mod search;
/// The TUI's tabs, read from rmux, for `/api/tabs`.
mod tabs;
/// rmux's terminal app, served here so a session page can frame it.
mod term;
/// The run's credentials, and `--token-file`, which keeps them across runs.
mod tokens;
pub mod tunnel;

use crate::cli;
use crate::fleet;
use crate::loader::Loader;
use crate::pricing::Plan;
use crate::session::Session;
use crate::watch::Watch;
use http::{EventStream, Request};
use std::collections::HashMap;
use std::io::Write;
use std::net::{TcpListener, TcpStream, ToSocketAddrs};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

/// The port `cctop serve` prefers.
///
/// Above the range a privileged process would want and clear of the numbers
/// development servers habitually take (3000, 5173, 8000, 8080), so the default
/// works without a flag on a machine already running the things a person who
/// wants this is likely to be running.
const DEFAULT_PORT: u16 = 7777;

/// How many ports past the default to try before giving up.
///
/// Only when the port was not asked for. Someone who wrote `--port 8080` meant
/// that port, and quietly serving on 8081 gives them a link that works and a
/// reverse proxy that does not.
const PORT_SEARCH: u16 = 16;

/// Concurrent connections served before further ones are refused.
///
/// Each is a thread. A browser opens perhaps six — the page, its fetches, and
/// one SSE stream it holds — so this is several tabs' worth and still a bound
/// small enough that a peer opening sockets in a loop achieves nothing.
const MAX_CONNECTIONS: usize = 32;

/// How often a full walk of every provider directory happens.
///
/// The light refresh re-reads the rows that can still change, which is nearly
/// everything that matters; only a session that did not exist before needs the
/// walk. [`Watch`] usually reports those as they appear, and this is the safety
/// net for when it cannot — a provider directory that was not watchable, or an
/// inotify budget the machine had already spent.
const FULL_WALK: Duration = Duration::from_secs(30);

/// The longest an SSE stream goes without writing anything before it sends a
/// `ping` event.
///
/// Idle connections have to produce traffic or the hops in between drop them,
/// and a browser that has gone away only surfaces as a write error once
/// something is written at it. The page holds the other end of this bargain:
/// a stream it has heard nothing on for a little over twice this is one it
/// closes and reopens (`STALE_AFTER` in `web/src/hooks/use-live.ts`), so the
/// two numbers move together.
const SSE_KEEPALIVE: Duration = Duration::from_secs(15);

/// What a request naming no session, or an ambiguous prefix of one, is told.
///
/// One string because four routes say it, and four spellings of one refusal is
/// how a page comes to show three different messages for the same mistake.
const NO_SUCH_SESSION: &str = "no session with that id, or the prefix matches more than one";

/// The web UI: the React app in `web/`, built by `npm run build` into one HTML
/// file with every script, stylesheet and font inlined, and committed.
///
/// `include_str!` rather than a directory served off disk: an installed cctop
/// is one binary, and a page that loads its own files is a page that breaks the
/// moment the binary is moved. It is also what lets the response promise a
/// content policy that forbids loading anything at all. What the server knows
/// and the page cannot reaches it as JSON in place of `__CCTOP_CONFIG__`; see
/// [`app_config`]. Every page — the table, a session, the workspace, analytics
/// — is a route inside it.
const APP_HTML: &str = include_str!("assets/app/index.html");

/// A favicon small enough to keep inline: the table's dark tile with the amber
/// dot it draws on a session that is waiting.
///
/// Served rather than embedded into the HTML — a `<link rel="icon">` is one
/// request more, and embedding SVG in the pages would repeat it into both.
/// `img-src 'self'` in the content policy is what lets the browser fetch it.
const FAVICON: &str = "<svg xmlns=\"http://www.w3.org/2000/svg\" viewBox=\"0 0 64 64\">\
    <rect width=\"64\" height=\"64\" rx=\"14\" fill=\"#16151a\"/>\
    <circle cx=\"32\" cy=\"32\" r=\"10\" fill=\"#e0a52c\"/></svg>";

/// The web manifest, for a browser that wants to install the page as an app.
///
/// Its own route rather than a data: URL in a `<link>` because manifest fetches
/// are governed by `manifest-src`, which `data:` is not a part of here.
static FAVICON_P: std::sync::LazyLock<http::Packed> =
    std::sync::LazyLock::new(|| http::Packed::new(FAVICON));

static MANIFEST_P: std::sync::LazyLock<http::Packed> =
    std::sync::LazyLock::new(|| http::Packed::new(MANIFEST));

const MANIFEST: &str = concat!(
    r#"{"name":"cctop","short_name":"cctop","display":"standalone","#,
    r#""start_url":"/","icons":[{"src":"/favicon.svg","sizes":"any","#,
    // `r##` rather than `r#` here: the colour hex begins `"#`, which is the
    // short delimiter's own terminator.
    r##""type":"image/svg+xml"}],"theme_color":"#16151a","background_color":"#16151a"}"##
);

/// Baseline gap between usage checks for a standalone `serve`.
///
/// A minute rather than the session refresh's seconds: quota moves slowly and
/// the endpoints throttle aggressively — a 30s poll was once enough to earn a
/// sustained 429 with a ~15 minute `retry-after`. When a provider asks for
/// longer, `retry_delay_secs` honours that instead.
const QUOTA_INTERVAL_SECS: u64 = 60;

/// How often the quota poller wakes to see whether either provider is due.
const QUOTA_TICK: Duration = Duration::from_secs(10);

/// Everything a connection thread needs, shared behind one `Arc`.
struct Shared {
    /// The per-run access token, or empty under `--no-token`.
    token: String,
    /// The second token minted each run, and the link it makes possible.
    ///
    /// A token is the whole authorisation a request carries, so "read-only"
    /// cannot be a flag a request sends — it has to be a different credential.
    /// This one opens every page and every GET, and `/api/act/*` answers it
    /// with 403. Empty under `--no-token`, which has nothing to withhold.
    readonly: String,
    /// Whether the routes that act on a session answer at all.
    ///
    /// The token is what makes them safe to have, so `--no-token` turns them
    /// off and `--no-actions` turns them off for someone who wants the page and
    /// not the buttons. Read on every action request rather than only when the
    /// page is built: a page cached from a run that had them on must not be able
    /// to act on a run that has them off.
    actions: bool,
    /// The port being served, because it is part of the access cookie's name —
    /// cookies are scoped to the host and ignore the port, so two serves on
    /// one machine would otherwise hand each other's credential back.
    port: u16,
    plan: Plan,
    /// The latest snapshot, and a version that only ever increases.
    ///
    /// SSE threads wait on the condvar for a version past the one they last
    /// sent, which is what makes an idle stream cost nothing: no polling, and
    /// one wakeup per refresh rather than one per connection per tick.
    latest: Mutex<Arc<Snapshot>>,
    updated: Condvar,
    /// Transcript parses for the report route.
    ///
    /// Its own store rather than the refresh thread's, because that one is
    /// behind a `&mut Loader` on a thread that is usually mid-walk, and a
    /// report should not queue behind a directory sweep. It is never saved:
    /// two owners writing one cache file is how a cache file gets corrupted,
    /// and the refresh thread is the owner that has something worth keeping.
    store: crate::cache::Store,
    /// The `/api/quota` document, already serialised.
    ///
    /// Rendered by whoever produced the numbers — the standalone serve's own
    /// poller, or the dashboard feeding [`Serving::publish_with_quota`] — so
    /// the route itself is a memory read and never touches the rate-limited
    /// usage endpoints on a browser's schedule.
    quota: Mutex<String>,
    /// The topical tier behind `/api/search`, built lazily on the first query
    /// that wants it.
    topics: Mutex<search::Topics>,
    /// The webhook `--notify` points at, when one was asked for.
    notify: Option<notify::Webhook>,
    /// The `--host` machines by target name, so a route for a remote row can
    /// ask the cctop that actually holds the transcript — see [`remote_json`].
    /// Present even when `scan` is off: a dashboard-hosted serve still owes
    /// its remote rows an answer.
    hosts: HashMap<String, fleet::Host>,
    /// The app page as each credential is served it, compressed once — see
    /// [`app_page`]. Indexed by [`Access`]: the page differs only in which
    /// token and actions flag it carries, both fixed for the run.
    pages: [std::sync::OnceLock<http::Packed>; 2],
}

/// One publish of the whole table.
struct Snapshot {
    version: u64,
    /// The `--json` document, already serialised — every SSE client sends the
    /// same bytes, so they are rendered once per refresh rather than once per
    /// client.
    json: String,
    /// Each row of `json` on its own, by session id, for a stream that follows
    /// one session — see [`events`]. `json` is these joined, so the two cannot
    /// disagree.
    rows: Vec<(String, String)>,
    /// The rows behind it, kept so the report route can find its session
    /// without re-walking the disk.
    sessions: Vec<Session>,
    /// Hosts named with `--host` that could not be read, and why.
    host_errors: Vec<(String, String)>,
}

/// Remote rows, written by the host pollers and read by the refresh thread.
#[derive(Default)]
struct Remotes {
    rows: HashMap<String, Vec<Session>>,
    errors: HashMap<String, String>,
}

pub const HELP: &str = "\
cctop serve — the session table, and the sessions themselves, in a browser

USAGE:
  cctop serve [OPTIONS]

OPTIONS:
  --bind <ADDR>    Address to listen on [default: 127.0.0.1]. Anything other
                   than a loopback address puts the page on your network, which
                   is announced on stderr when it happens
  --port <PORT>    Port to listen on [default: 7777]. Without this flag a busy
                   port is stepped past; with it, a busy port is an error
  --no-token       Serve without an access token. Every process and user on the
                   machine can then read your sessions — and since the token is
                   what authorises an action, this also turns actions off
  --no-actions     Serve the pages without the buttons: no prompts, no resuming,
                   no handing a session to another agent
  --tunnel         Also reach the page from anywhere, over a trycloudflare quick
                   tunnel. Needs nothing installed, lasts as long as this
                   process, and puts the link on the public internet — so the
                   token is what stands between it and your agents
  --plan <PLAN>    Billing plan for cost figures: retail, max, or included
                   [default: retail]
  --delay <SECS>   Seconds between refreshes [default: 2]
  --host <HOST>    Also serve the sessions on another machine, over ssh.
                   Repeatable; same syntax as `cctop --host`
  --notify <URL>   POST a JSON event to this URL when a session crosses into
                   waiting or asking — once per crossing, on a short deadline.
                   [env: CCTOP_NOTIFY_URL]
  --token-file <PATH>
                   Keep the tokens across restarts: read them from PATH, or
                   mint them and write PATH (mode 600) when it does not exist.
                   A file other users can read, or do not own, is refused
  --rotate-token   With --token-file: replace the tokens in it, revoking every
                   link built on the old ones
  -h, --help       Print this help

The page shows each session's conversation, what it edited, and what it can
reach, and it can send a prompt to a live session, resume a dead one, or hand
one to a different agent. Whoever holds the link can do all of that, which is
why the link carries a token and the default is loopback only.
";

/// Parse `cctop serve`'s own flags and run the server until interrupted.
///
/// Flags are parsed here rather than in [`crate::cli::Args`] for the reason
/// `doctor` and `attach` are: cctop takes no positionals, so clap answers a
/// bare `serve` with a usage error before it can reach any of this.
/// What a run of the server needs to know, however it was asked for.
///
/// The CLI fills this from flags and the dashboard fills it from a keypress,
/// which is the point: one server, reached two ways, rather than a second
/// implementation behind a key.
pub struct Options {
    pub bind: String,
    pub port: u16,
    /// Whether `port` was asked for. A port nobody chose steps past a busy one;
    /// a port somebody chose is an error when it is taken, because silently
    /// serving somewhere else is worse than saying so.
    pub port_given: bool,
    pub no_token: bool,
    pub no_actions: bool,
    pub tunnel: bool,
    pub plan: Plan,
    pub delay: Duration,
    /// The machines whose sessions this serve shows, already parsed.
    ///
    /// `Host`s rather than spec strings because the dashboard's copy was
    /// collected once at startup — passing specs back through `collect` here
    /// would merge `$CCTOP_HOSTS` a second time for it.
    pub hosts: Vec<fleet::Host>,
    /// Where to POST when a session crosses into waiting or asking.
    ///
    /// `None` still falls back to `CCTOP_NOTIFY_URL` inside [`start`], which is
    /// how a dashboard-hosted serve — no flag to have asked with — gets one.
    pub notify: Option<String>,
    /// Credentials to serve with instead of minting fresh ones — `--token-file`
    /// after [`tokens::load_or_create`] has read or written it.
    ///
    /// Loaded by the caller rather than here so `run` can say which of the two
    /// happened, and so the dashboard, which never passes one, keeps meaning
    /// "stopping it revokes every link".
    pub tokens: Option<tokens::Tokens>,
    /// Whether the server scans for sessions itself.
    ///
    /// True for `cctop serve`, which is the only thing running. False for the
    /// dashboard, which already walks every session five times a second and
    /// would otherwise do it twice in one process — and worse, show a page that
    /// disagrees with the table beside it. The dashboard feeds
    /// [`Serving::publish`] instead.
    pub scan: bool,
}

impl Default for Options {
    fn default() -> Self {
        Options {
            bind: "127.0.0.1".to_string(),
            port: DEFAULT_PORT,
            port_given: false,
            no_token: false,
            no_actions: false,
            tunnel: false,
            plan: Plan::Retail,
            delay: Duration::from_secs(2),
            hosts: Vec::new(),
            notify: None,
            tokens: None,
            scan: true,
        }
    }
}

/// A server that is up, and the links that reach it.
///
/// Dropping it revokes the tunnel and stops the accept loop, so a dashboard
/// that stops serving stops being reachable — including from a link somebody
/// has already opened.
pub struct Serving {
    /// The loopback link, always present.
    pub local: String,
    /// The public one, when a tunnel was asked for and registered.
    pub public: Option<String>,
    /// The same page, read-only: every route answers but the ones that act.
    ///
    /// Carries the run's second token, on whichever origin is the one to hand
    /// out — the tunnel's when there is one. Empty when the run has no token,
    /// since a tokenless serve has nothing for a second credential to withhold.
    pub readonly: String,
    pub actions: bool,
    shared: Arc<Shared>,
    remotes: Arc<Mutex<Remotes>>,
    plan: Plan,
    version: Mutex<u64>,
    /// Held so dropping this unregisters from Cloudflare's edge.
    _tunnel: Option<tunnel::Tunnel>,
    /// Cleared on drop; the accept loop reads it after every connection and
    /// stops when it is false.
    running: Arc<AtomicBool>,
    port: u16,
}

impl Serving {
    /// The link to hand somebody: the public one when there is one.
    pub fn best(&self) -> &str {
        self.public.as_deref().unwrap_or(&self.local)
    }

    /// Show the page these rows and this usage reading, replacing whatever it
    /// was showing.
    ///
    /// For a dashboard-hosted server, which has both already. Cheap enough to
    /// call on every refresh: it costs one JSON encode of what the table is
    /// already holding, and it is what wakes the event stream. The quota comes
    /// along because the dashboard polls the usage endpoints for its own
    /// panes — the page shares that reading rather than standing up a second
    /// poller against the same rate-limited endpoints, and the two can never
    /// disagree.
    pub fn publish_with_quota(&self, sessions: &[Session], quota: &crate::quota::Quota) {
        let Ok(mut version) = self.version.lock() else {
            return;
        };
        publish(
            &self.shared,
            &self.remotes,
            sessions,
            self.plan,
            &self.shared.store,
            &mut version,
            Some(quota),
        );
    }
}

/// Hand `url` to whatever this desktop opens links with.
///
/// Best effort by design: there is no answer worth waiting for and plenty of
/// machines with no browser to give it to — a headless box, an ssh session, a
/// container. `false` means nothing was launched, which the caller says on the
/// status line so the link can be copied instead.
///
/// Spawned and forgotten rather than waited on: `xdg-open` on some desktops
/// does not return until the browser it started exits, and a dashboard frozen
/// behind somebody's Firefox is not a trade worth making.
pub fn open_in_browser(url: &str) -> bool {
    // `xdg-open` is what every freedesktop environment answers to.
    std::process::Command::new("xdg-open")
        .arg(url)
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
        .is_ok()
}

impl Drop for Serving {
    fn drop(&mut self) {
        self.running.store(false, Ordering::Relaxed);
        // The accept loop is parked inside `accept`, so clearing the flag is not
        // enough to end it — one connection of our own wakes it, it sees the
        // flag, and returns. Loopback whatever the server is bound to: it is
        // listening there too for any bind cctop allows, and a refused connect
        // here costs a thread that ends when the process does.
        let _ = std::net::TcpStream::connect(("127.0.0.1", self.port));
    }
}

/// Bring the server up and hand back the links, without blocking.
///
/// Everything `cctop serve` does apart from printing and parking. The accept
/// loop moves to a thread of its own so a dashboard can go on drawing, which is
/// the whole reason this is split out of [`run`].
pub fn start(options: Options) -> anyhow::Result<Serving> {
    let listener = listen(&options.bind, options.port, options.port_given)?;
    let addr = listener.local_addr()?;
    if let Some(why) = tunnel_objection(options.tunnel, addr.ip().is_loopback(), options.no_token) {
        anyhow::bail!("{why}");
    }
    // A second credential for the same pages minus the actions — see
    // `Shared::readonly` for why read-only is a token rather than a flag.
    let tokens = match (options.no_token, options.tokens) {
        (true, Some(_)) => anyhow::bail!(
            "--token-file with --no-token names tokens and then serves without them. \
             Drop one of the two."
        ),
        (true, None) => tokens::Tokens::none(),
        (false, Some(given)) => given,
        (false, None) => tokens::Tokens::fresh(),
    };
    let token = tokens.full;
    // The token is the whole authorisation story for an action, so there are no
    // actions without one. `--no-token` is already documented as "every process
    // and user on this machine can read your sessions"; letting that also mean
    // "and type at your agents" is a different sentence, and not one anybody
    // reads a flag expecting.
    let actions = !options.no_actions && !options.no_token;
    // Started before anything is announced, so the link works when it is read,
    // and before the accept loop because there is nothing to reach yet.
    let tunnel = match options.tunnel {
        true => Some(tunnel::start(addr.port())?),
        false => None,
    };

    // The origin a notification's link is built on: the tunnel's when there is
    // one, since a webhook that names loopback is a link that works only on the
    // machine that sent it.
    let origin = tunnel
        .as_ref()
        .map(|t| t.url.clone())
        .unwrap_or_else(|| format!("http://127.0.0.1:{}", addr.port()));

    let shared = Arc::new(Shared {
        token: token.clone(),
        readonly: tokens.readonly,
        actions,
        port: addr.port(),
        plan: options.plan,
        latest: Mutex::new(Arc::new(Snapshot {
            version: 0,
            json: "[]".to_string(),
            rows: Vec::new(),
            sessions: Vec::new(),
            host_errors: Vec::new(),
        })),
        updated: Condvar::new(),
        store: crate::cache::Store::new(),
        quota: Mutex::new(quota::EMPTY.to_string()),
        topics: Mutex::new(search::Topics::default()),
        notify: options
            .notify
            .or_else(|| {
                // The dashboard builds its Options without a flag, so the
                // environment is answered here — where every way of starting
                // a serve passes through — rather than only in `run`'s parser.
                std::env::var("CCTOP_NOTIFY_URL")
                    .ok()
                    .filter(|url| !url.is_empty())
            })
            .map(|target| notify::Webhook::new(target, origin.clone(), token.clone())),
        pages: Default::default(),
        hosts: options
            .hosts
            .iter()
            .map(|host| (host.target.clone(), host.clone()))
            .collect(),
    });

    let remotes = Arc::new(Mutex::new(Remotes::default()));
    // A dashboard-hosted serve leaves the polling to the dashboard, which is
    // already running one loop per host — spawning these anyway would poll
    // every remote twice.
    if options.scan {
        for host in &options.hosts {
            spawn_host_poller(host.clone(), Arc::clone(&remotes));
        }
    }
    if options.scan {
        spawn_refresher(
            Arc::clone(&shared),
            Arc::clone(&remotes),
            options.plan,
            options.delay,
        );
        // The dashboard's own poller feeds the page through
        // `publish_with_quota`; a standalone serve has to ask the endpoints
        // itself, on the slow cadence they demand.
        spawn_quota_poller(Arc::clone(&shared));
    }

    let query = match token.is_empty() {
        true => String::new(),
        false => format!("?t={token}"),
    };
    crate::elog::event(
        "serve",
        "listen",
        serde_json::json!({
            "addr": addr.to_string(),
            "token": !token.is_empty(),
            "actions": actions,
            "tunnel": options.tunnel,
            "notify": shared.notify.is_some(),
        }),
    );
    let running = Arc::new(AtomicBool::new(true));
    {
        let (shared, running) = (Arc::clone(&shared), Arc::clone(&running));
        std::thread::Builder::new()
            .name("cctop-serve-accept".into())
            .spawn(move || accept_loop(listener, shared, running))?;
    }

    // The read-only link rides the same origin as whichever link is the one to
    // hand out — public when a tunnel registered, loopback otherwise.
    let readonly_link = match shared.readonly.is_empty() {
        true => String::new(),
        false => format!("{origin}/?t={}", shared.readonly),
    };

    Ok(Serving {
        local: format!("http://127.0.0.1:{}/{query}", addr.port()),
        public: tunnel.as_ref().map(|t| format!("{}/{query}", t.url)),
        readonly: readonly_link,
        actions,
        shared,
        remotes,
        plan: options.plan,
        version: Mutex::new(0),
        _tunnel: tunnel,
        running,
        port: addr.port(),
    })
}

/// Answer connections until `running` goes false.
fn accept_loop(listener: TcpListener, shared: Arc<Shared>, running: Arc<AtomicBool>) {
    let live = Arc::new(AtomicUsize::new(0));
    for stream in listener.incoming() {
        if !running.load(Ordering::Relaxed) {
            return;
        }
        let Ok(mut stream) = stream else { continue };

        // Checked before the thread is spawned, so refusing a connection costs
        // a response rather than a thread — which is the point of a cap.
        if live.load(Ordering::Relaxed) >= MAX_CONNECTIONS {
            http::respond_error(&mut stream, None, 503, "too many open connections");
            continue;
        }
        let slot = Connection::take(&live);
        let shared = Arc::clone(&shared);
        // A spawn failure must not take the accept loop down with it: the
        // machine is out of threads, which the next connection may not be. The
        // slot is released either way — dropping the closure unspawned drops
        // the guard inside it, and so does a handler that panics.
        let _ = std::thread::Builder::new()
            .name("cctop-serve".into())
            .spawn(move || {
                let _slot = slot;
                serve_connection(&shared, &mut stream);
            });
    }
}

pub fn run(argv: &[String]) -> anyhow::Result<i32> {
    let mut bind = "127.0.0.1".to_string();
    let mut port = DEFAULT_PORT;
    let mut port_given = false;
    let mut no_token = false;
    let mut no_actions = false;
    let mut want_tunnel = false;
    let mut plan = Plan::Retail;
    let mut delay = Duration::from_secs(2);
    let mut hosts: Vec<String> = Vec::new();
    // `None` here still honours CCTOP_NOTIFY_URL — the fallback lives in
    // `start`, so it also covers a dashboard, which has no flag to give.
    let mut notify = None;
    let mut token_file: Option<std::path::PathBuf> = None;
    let mut rotate_token = false;

    let mut it = argv.iter();
    while let Some(flag) = it.next() {
        let mut value = || {
            it.next()
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("{flag} needs a value"))
        };
        match flag.as_str() {
            "-h" | "--help" => {
                print!("{HELP}");
                return Ok(0);
            }
            "--bind" => bind = value()?,
            "--port" => {
                port = value()?
                    .parse()
                    .map_err(|_| anyhow::anyhow!("--port takes a number from 1 to 65535"))?;
                port_given = true;
            }
            "--no-token" => no_token = true,
            "--no-actions" => no_actions = true,
            "--tunnel" => want_tunnel = true,
            "--plan" => {
                let given = value()?;
                plan = Plan::parse(&given).ok_or_else(|| {
                    anyhow::anyhow!("unsupported plan '{given}'; use retail, max or included")
                })?;
            }
            "--delay" => {
                let secs: f64 = value()?
                    .parse()
                    .map_err(|_| anyhow::anyhow!("--delay takes a number of seconds"))?;
                if !secs.is_finite() || !(0.5..3600.0).contains(&secs) {
                    anyhow::bail!("--delay must be between 0.5 and 3600 seconds");
                }
                delay = Duration::from_secs_f64(secs);
            }
            "--host" => hosts.push(value()?),
            "--notify" => notify = Some(value()?),
            "--token-file" => token_file = Some(value()?.into()),
            "--rotate-token" => rotate_token = true,
            other => anyhow::bail!("unknown option '{other}'\n\n{HELP}"),
        }
    }

    // Said before the call and not inside it: registering with the edge takes a
    // second or more, and `serve` has nothing else on screen to show for it.
    // The dashboard, which calls the same code, has a spinner instead — see
    // [`tunnel::start`] for why that one must not print.
    // Before the listener, so a refused file costs no socket and the reason is
    // the only thing on screen.
    let tokens = match &token_file {
        // `start` refuses this too; asked here first so the answer is the
        // contradiction, not whatever was wrong with the file.
        Some(_) if no_token => anyhow::bail!(
            "--token-file with --no-token names tokens and then serves without them. \
             Drop one of the two."
        ),
        Some(path) => {
            let (tokens, how) = tokens::load_or_create(path, rotate_token)?;
            match how {
                tokens::Loaded::Reused => eprintln!(
                    "cctop: tokens from {} — links from earlier runs still work",
                    path.display()
                ),
                tokens::Loaded::Created => eprintln!(
                    "cctop: new tokens written to {} — they outlive this run; \
                     --rotate-token replaces them",
                    path.display()
                ),
            }
            Some(tokens)
        }
        None if rotate_token => anyhow::bail!(
            "--rotate-token rotates the tokens in a --token-file; without one, \
             every run already mints new tokens"
        ),
        None => None,
    };
    if want_tunnel {
        eprintln!("cctop: opening a trycloudflare tunnel…");
        let _ = std::io::stderr().flush();
    }
    let serving = start(Options {
        bind: bind.clone(),
        port,
        port_given,
        no_token,
        no_actions,
        tunnel: want_tunnel,
        plan,
        delay,
        hosts: fleet::Host::collect(&hosts),
        notify,
        tokens,
        // The only thing in this process, so it does its own walking.
        scan: true,
    })?;
    announce(&serving, &bind, no_token);

    // Nothing left to do on this thread: the accept loop has its own. Parking
    // rather than returning is what keeps the process — and with it the tunnel
    // and every thread above — alive, since `serve` is the whole command.
    loop {
        std::thread::park();
    }
}

/// One of [`MAX_CONNECTIONS`], released when the handler ends however it ends.
///
/// A guard rather than a decrement at the bottom of the thread: a handler that
/// panics would otherwise leak its slot, and enough of those turn the cap into
/// a server that refuses everything and cannot be talked out of it.
struct Connection(Arc<AtomicUsize>);

impl Connection {
    fn take(live: &Arc<AtomicUsize>) -> Connection {
        live.fetch_add(1, Ordering::Relaxed);
        Connection(Arc::clone(live))
    }
}

impl Drop for Connection {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::Relaxed);
    }
}

/// Bind the listener, stepping past a busy port only when none was asked for.
fn listen(bind: &str, port: u16, port_given: bool) -> anyhow::Result<TcpListener> {
    let last = match port_given {
        true => port,
        false => port.saturating_add(PORT_SEARCH),
    };
    let mut last_error = None;
    for candidate in port..=last {
        let addr = (bind, candidate)
            .to_socket_addrs()
            .map_err(|e| anyhow::anyhow!("could not resolve --bind {bind}: {e}"))?
            .next()
            .ok_or_else(|| anyhow::anyhow!("--bind {bind} resolved to no address"))?;
        match TcpListener::bind(addr) {
            Ok(listener) => return Ok(listener),
            Err(e) => last_error = Some(e),
        }
    }
    let why = last_error.map_or_else(|| "no port to try".to_string(), |e| e.to_string());
    match port_given {
        true => anyhow::bail!("could not listen on {bind}:{port}: {why}"),
        false => anyhow::bail!(
            "could not listen on {bind}, ports {port} to {last}: {why}\n\
             Use --port to name a free one."
        ),
    }
}

/// Why this `--tunnel` is refused, if it is.
///
/// Both refusals are combinations that read as one wish and mean another, and
/// each is cheaper to say than to explain afterwards:
///
/// - **`--tunnel` with a token turned off.** The tunnel's whole job is to put
///   the page somewhere anyone can reach; the token is then the only thing
///   between the internet and a prompt box wired to a live agent. `--no-token`
///   is defensible on loopback and indefensible here, so it is an error rather
///   than a warning nobody scrolls back to.
/// - **`--tunnel` with a non-loopback `--bind`.** The tunnel dials out from this
///   machine, so it reaches a loopback listener perfectly well. Binding wider
///   also publishes the page on the local network, which is a second decision
///   and not the one that was asked for.
///
/// Split out from [`run`] because it is the whole of the policy and wants a test
/// rather than a careful reading.
fn tunnel_objection(want_tunnel: bool, loopback: bool, no_token: bool) -> Option<&'static str> {
    if !want_tunnel {
        return None;
    }
    if no_token {
        return Some(
            "--tunnel with --no-token would publish your sessions, and a prompt \
             box for your agents, to anyone who finds the URL.\n\
             Drop one of the two: the token is what makes the link a credential.",
        );
    }
    if !loopback {
        return Some(
            "--tunnel does not need --bind: the tunnel connects out from this \
             machine, so it reaches a loopback listener.\n\
             Binding wider would put the page on your local network as well, \
             which --tunnel is not asking for.",
        );
    }
    None
}

/// Print where to point a browser, and say plainly when the page is reachable
/// beyond this machine.
///
/// On stderr, so `cctop serve > /dev/null` still tells someone what happened,
/// and so the URL is not mistaken for output a script should parse.
///
/// With a tunnel the public URL is printed first and the loopback one after it:
/// the public one is what someone asked for and what goes to the phone, and the
/// loopback one is still the right link from this machine. What follows both is
/// the sentence that matters — that the first link is a way in, not a view.
fn announce(serving: &Serving, bind: &str, no_token: bool) {
    if let Some(public) = &serving.public {
        eprintln!("cctop: serving on {public}");
        eprintln!("cctop: also on {}", serving.local);
        // The second link is a different credential, and labelled as such so
        // it is never handed out as the full one: a link that can only read is
        // only useful if the person holding it knows that is what they have.
        if !serving.readonly.is_empty() {
            eprintln!("cctop: read-only link — no actions: {}", serving.readonly);
        }
        eprintln!(
            "cctop: that first link is on the public internet. Anyone who has it \
             can read every session on this machine — and, unless --no-actions, \
             type at your agents, which runs commands as you. Cloudflare carries \
             the traffic and can read it. The tunnel ends when this process does."
        );
        let _ = std::io::stderr().flush();
    } else {
        eprintln!("cctop: serving on {}", serving.local);
        if !serving.readonly.is_empty() {
            eprintln!("cctop: read-only link — no actions: {}", serving.readonly);
        }
    }
    if bind != "127.0.0.1" && serving.public.is_none() {
        eprintln!(
            "cctop: bound to {bind}, so this is on your network and not just this \
             machine"
        );
    }
    if no_token {
        eprintln!(
            "cctop: no token — every process and user on this machine can read \
             your sessions"
        );
    }
    let _ = std::io::stderr().flush();
}

/// A token for this run.
///
/// See [`crate::util::random_bytes`] for where the entropy comes from and what
/// it is worth: enough that a token cannot be guessed from across a network,
/// which is why the default bind is loopback and this is defence in depth
/// rather than the defence.
fn new_token() -> String {
    crate::util::random_hex(TOKEN_BYTES)
}

/// Bytes of entropy behind a token, which is twice as many hex characters.
const TOKEN_BYTES: usize = 16;

/// Compare a presented token against the real one without leaking where they
/// diverged.
///
/// The timing channel is narrow over a network and completely open over
/// loopback, which is exactly where an unprivileged local process would be
/// standing. Lengths are compared separately and do leak, which tells an
/// attacker something they can already read out of `--help`.
fn token_matches(expected: &str, given: &str) -> bool {
    if expected.len() != given.len() {
        return false;
    }
    let mut diff = 0u8;
    for (a, b) in expected.bytes().zip(given.bytes()) {
        diff |= a ^ b;
    }
    diff == 0
}

/// Poll one host forever, publishing whatever it last said.
///
/// A failed poll keeps the previous rows and records why: the ssh connection
/// dropping has not stopped those agents, and blanking the machine would make
/// the totals look complete when they are not — the same call [`crate::ui`]
/// makes for the same reason.
fn spawn_host_poller(host: fleet::Host, remotes: Arc<Mutex<Remotes>>) {
    std::thread::spawn(move || {
        loop {
            let snapshot = host.poll();
            if let Ok(mut remotes) = remotes.lock() {
                match snapshot {
                    fleet::Snapshot::Rows(rows) => {
                        remotes.errors.remove(&host.target);
                        remotes.rows.insert(host.target.clone(), rows);
                    }
                    fleet::Snapshot::Failed(why) => {
                        remotes.errors.insert(host.target.clone(), why);
                    }
                }
            }
            std::thread::sleep(fleet::POLL);
        }
    });
}

/// Poll each provider's usage endpoint on the slow cadence they demand,
/// storing the rendered `/api/quota` document each time one answers.
///
/// Each provider is paced by its own last outcome — a throttled one backs off
/// without stalling the other — which is the same arithmetic the dashboard's
/// poller runs (`spawn_quota_poller` in `src/ui/runloop.rs`), minus the burn
/// log: a serve records nothing, it only reports.
fn spawn_quota_poller(shared: Arc<Shared>) {
    std::thread::spawn(move || {
        let mut claude: Vec<crate::quota::ProfileQuota> = Vec::new();
        let mut codex: Vec<crate::quota::ProfileQuota> = Vec::new();
        let (mut claude_due, mut codex_due) = (Instant::now(), Instant::now());
        loop {
            let now = Instant::now();
            let mut changed = false;
            // Each profile is its own account with its own limits, so each is
            // asked separately. They share one due time: the interval exists
            // to be polite to the provider, and a machine with two logins is
            // not entitled to twice the requests.
            if now >= claude_due {
                claude = crate::config::accounts_for(crate::pricing::Provider::Claude)
                    .iter()
                    .map(|profile| crate::quota::ProfileQuota {
                        profile: profile.name.clone(),
                        status: crate::quota::fetch_claude(profile),
                        source: profile.source,
                    })
                    .collect();
                // Paced by whichever account is most throttled, so backing off
                // for one does not keep asking on behalf of another.
                let delay = claude
                    .iter()
                    .map(|q| q.status.retry_delay_secs(QUOTA_INTERVAL_SECS))
                    .max()
                    .unwrap_or(QUOTA_INTERVAL_SECS);
                claude_due = now + Duration::from_secs(delay);
                changed = true;
            }
            if now >= codex_due {
                codex = crate::config::accounts_for(crate::pricing::Provider::Codex)
                    .iter()
                    .map(|profile| crate::quota::ProfileQuota {
                        profile: profile.name.clone(),
                        status: crate::quota::fetch_codex(profile),
                        source: profile.source,
                    })
                    .collect();
                let delay = codex
                    .iter()
                    .map(|q| q.status.retry_delay_secs(QUOTA_INTERVAL_SECS))
                    .max()
                    .unwrap_or(QUOTA_INTERVAL_SECS);
                codex_due = now + Duration::from_secs(delay);
                changed = true;
            }
            if changed && let Ok(mut slot) = shared.quota.lock() {
                *slot = serde_json::to_string(&quota::document(&crate::quota::Quota {
                    fetched: true,
                    claude: claude.clone(),
                    codex: codex.clone(),
                }))
                .unwrap_or_else(|_| quota::EMPTY.to_string());
            }
            std::thread::sleep(QUOTA_TICK);
        }
    });
}

/// Walk, refresh, and publish, forever.
///
/// The cadence mirrors the UI's: a light refresh every `delay` that re-reads
/// only rows that can still change, and a full walk when [`Watch`] says a
/// session file appeared or [`FULL_WALK`] passes without it saying anything.
/// A walk on every tick would re-read every provider directory on disk to
/// discover, almost always, nothing.
fn spawn_refresher(shared: Arc<Shared>, remotes: Arc<Mutex<Remotes>>, plan: Plan, delay: Duration) {
    std::thread::spawn(move || {
        // Blocking, once, before the first snapshot: a dashboard whose first
        // frame shows every cost as zero is worse than one that appears a
        // second later with the truth.
        crate::pricing::refresh_pricing_blocking();

        let mut loader = Loader::new();
        let watch = Watch::start();
        // A hook listener of our own: agents report to every socket in the
        // directory, so a standalone serve sees a permission prompt the moment
        // it goes up whether or not a dashboard process is running.
        let listener = crate::hook::Listener::start();
        let mut reports = crate::hook::Reports::new();
        // Reading an agent's screen needs somewhere to read it from — a shim
        // socket or an rmux pane — which [`Peek`] finds once and reuses. The
        // same opt-in as the TUI's: the phrases are each agent's own UI, not a
        // contract, so they are only looked for when the file says so.
        let mut peek = crate::peek::Peek::new();
        let read_screen = crate::settings::Settings::load().read_screen == Some(true);
        loader.set_hook_claims(reports.claims.clone());
        // The YOLO switch, read on every pass. Only a serve that may act
        // competes to answer for it — see [`crate::yolo`] for why only one
        // cctop on the machine ever presses.
        let mut yolo = crate::yolo::Auto::new(shared.actions);
        let mut rows = loader.load(plan);
        stamp(&mut rows, &reports, &mut peek, read_screen);
        yolo.tick(&mut rows);
        let mut walked = Instant::now();
        let mut version = 0u64;
        publish(
            &shared,
            &remotes,
            &rows,
            plan,
            loader.store(),
            &mut version,
            None,
        );

        loop {
            std::thread::sleep(delay);

            // Hook events first: a `SessionStart` is a session to walk for now
            // rather than at the next poll, and a moved pid claim belongs in
            // the file the loader resolves pids by.
            let mut appeared = false;
            for event in listener.as_ref().map(|l| l.drain()).unwrap_or_default() {
                let (lifecycle, moved) = reports.observe(&event);
                appeared |= lifecycle;
                if moved {
                    crate::hook::save_claims(&reports.claims);
                    loader.set_hook_claims(reports.claims.clone());
                }
            }
            // A prompt auto mode never answered promotes itself, no event
            // needed — the grace running out is the news.
            reports.promote_matured();
            appeared |= watch.as_ref().is_some_and(|w| {
                w.took_structural_change()
                    || w.awaiting_discovery(|path| {
                        rows.iter().any(|s| s.data_file.as_deref() == Some(path))
                    })
            });
            // With no watcher there is nothing to tell us a session appeared,
            // so the timer is the only trigger and has to carry the whole job.
            if appeared || walked.elapsed() >= FULL_WALK {
                rows = loader.load(plan);
                walked = Instant::now();
            } else {
                loader.refresh_live(plan, &mut rows);
            }
            stamp(&mut rows, &reports, &mut peek, read_screen);
            // After the stamp, which is what says a prompt is up: an answer
            // pressed here is a refresh tick behind the prompt at most.
            yolo.tick(&mut rows);
            publish(
                &shared,
                &remotes,
                &rows,
                plan,
                loader.store(),
                &mut version,
                None,
            );
        }
    });
}

/// Stamp each row with what reports and screens say about it.
///
/// The same [`Session::apply_reports`] the TUI's `App::apply_reports` calls,
/// for the same reason: whatever the rows end up believing is decided once.
/// Rows are rebuilt by the walk and re-derived by the refresh, so the stamp
/// is re-applied every pass — a permission prompt survives the rebuild
/// precisely because the report that raised it lives in `reports`, not the
/// row.
fn stamp(
    rows: &mut [Session],
    reports: &crate::hook::Reports,
    peek: &mut crate::peek::Peek,
    read_screen: bool,
) {
    for session in rows.iter_mut() {
        // A screen read is worth its cost only for a live local agent — a
        // stopped one has no screen, and the transcript already said the rest.
        let screened = match (read_screen, session.is_running()) {
            // `provider`, not `harness`: the column is the display name
            // ("ClaudeCode"), and the recognizer's ids are the lowercase ones.
            (true, true) => session
                .root_pid()
                .and_then(|pid| peek.read(session.provider.as_str(), pid)),
            _ => None,
        };
        session.apply_reports(reports.report(&session.session_id), screened.as_ref());
        reports.stamp_sandbox(session);
    }
}

/// Render one snapshot and wake everyone waiting on it.
/// `loader` is the refresh thread's own, and has to be: the document is built
/// by reading each row's cached extraction, and a fresh loader would have an
/// empty cache and re-parse every transcript on disk once per refresh.
fn publish(
    shared: &Shared,
    remotes: &Mutex<Remotes>,
    local: &[Session],
    plan: Plan,
    store: &crate::cache::Store,
    version: &mut u64,
    usage: Option<&crate::quota::Quota>,
) {
    let (mut sessions, host_errors) = match remotes.lock() {
        Ok(remotes) => {
            let mut merged = local.to_vec();
            for rows in remotes.rows.values() {
                merged.extend(rows.iter().cloned());
            }
            let mut errors: Vec<(String, String)> = remotes
                .errors
                .iter()
                .map(|(h, why)| (h.clone(), why.clone()))
                .collect();
            errors.sort();
            (merged, errors)
        }
        Err(_) => (local.to_vec(), Vec::new()),
    };
    // Newest first, so the page never has to decide what order means. Remote
    // rows arrive on their own schedule and would otherwise land wherever the
    // merge happened to put them.
    sessions.sort_by(|a, b| b.last_active.cmp(&a.last_active));

    // The same document `--json` prints and `--host` parses, from the same
    // builder: a browser being shown different figures than the terminal is a
    // bug nobody would think to look for. See [`cli::json_sessions`].
    let document = cli::json_sessions(&sessions, plan, store);
    // Serialised a row at a time and joined, which is byte for byte what the
    // whole array would serialise to — so the dashboard's document is
    // unchanged, and a session page's stream can send its one row of it.
    let rows: Vec<(String, String)> = document
        .iter()
        .filter_map(|row| {
            let value = serde_json::to_value(row).ok()?;
            let id = value.get("session_id")?.as_str()?.to_string();
            Some((id, value.to_string()))
        })
        .collect();
    let json = format!(
        "[{}]",
        rows.iter()
            .map(|(_, r)| r.as_str())
            .collect::<Vec<_>>()
            .join(",")
    );

    // Written through the same lock as the snapshot, so the page's quota panel
    // can never be newer than the rows it sits beside.
    if let Some(usage) = usage
        && let Ok(mut slot) = shared.quota.lock()
    {
        *slot = serde_json::to_string(&quota::document(usage))
            .unwrap_or_else(|_| quota::EMPTY.to_string());
    }

    *version += 1;
    let snapshot = Arc::new(Snapshot {
        version: *version,
        json,
        rows,
        sessions,
        host_errors,
    });
    crate::elog::event(
        "scan",
        "snapshot",
        serde_json::json!({
            "version": snapshot.version,
            "sessions": snapshot.sessions.len(),
            "running": snapshot.sessions.iter().filter(|s| s.is_running()).count(),
            "remote_hosts": snapshot.host_errors.len(),
        }),
    );
    // The crossing is read off the pair of snapshots as they meet, which is the
    // one place both feeds — this server's own refresher and the dashboard's —
    // pass through. A webhook asked for here therefore covers both.
    let mut crossed = Vec::new();
    if let Ok(mut latest) = shared.latest.lock() {
        if let Some(hook) = &shared.notify {
            crossed = hook
                .crossed(&latest.sessions, &snapshot.sessions)
                .into_iter()
                .map(|(event, s)| (event, s.clone()))
                .collect();
        }
        *latest = snapshot;
    }
    shared.updated.notify_all();
    if let Some(hook) = &shared.notify {
        for (event, session) in crossed {
            hook.post(event, &session);
        }
    }
}

// ---------------------------------------------------------------------------
// Routing
// ---------------------------------------------------------------------------

/// Which of this run's two credentials a request presented.
///
/// The distinction lives at the router rather than inside each route: the
/// check is the same for everything the read-only link may not do, and a route
/// added later that forgets it is the bug the single gate prevents.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Access {
    /// The full link: every page, every GET, and the actions when they are on.
    Full,
    /// The read-only link: everything but `/api/act/*`.
    ReadOnly,
}

/// The cookie a page hands back for the run's token. Named with the port
/// because a cookie is scoped to the host alone — `127.0.0.1:7777` and
/// `127.0.0.1:9999` share one jar — and a second serve would otherwise spend
/// its visits overwriting the first one's credential.
fn cookie_name(port: u16) -> String {
    format!("cctop_access_{port}")
}

/// What the presented token buys, or `None` when it buys nothing.
fn access_for(shared: &Shared, presented: &str) -> Option<Access> {
    if shared.token.is_empty() || token_matches(&shared.token, presented) {
        return Some(Access::Full);
    }
    // The emptiness check is load-bearing: `token_matches` answers true for
    // two empty strings, and an absent read-only credential must never be one
    // a request can present.
    if !shared.readonly.is_empty() && token_matches(&shared.readonly, presented) {
        return Some(Access::ReadOnly);
    }
    None
}

/// Read one request, answer it, and let the connection close.
fn serve_connection(shared: &Shared, stream: &mut TcpStream) {
    let request = match Request::parse(stream) {
        Ok(request) => request,
        Err((status, why)) => return http::respond_error(stream, None, status, why),
    };

    // The one route in front of the gate. It is aggregate counts, costs and
    // short session ids — no transcript, title or prompt — and what it says is
    // meant to be copied into a metrics store, so a scrape config should not
    // have to hold a secret to say it. GET and HEAD only: it is a read, and a
    // POST here is not a scrape. Answering it here rather than in the router
    // keeps the gate below unconditional for every other path.
    if request.path == "/metrics" {
        if request.method == "POST" {
            return http::respond_error(stream, Some(&request), 405, "/metrics answers GET");
        }
        let snapshot = current(shared);
        let body = metrics::render(
            &snapshot.sessions,
            shared.plan,
            &shared.store,
            snapshot.host_errors.len(),
        );
        return http::respond(
            stream,
            Some(&request),
            200,
            metrics::CONTENT_TYPE,
            body.as_bytes(),
        );
    }

    // Before the route, so a wrong token cannot be used to find out which
    // routes exist. Every path is behind it, including the ones that only
    // return HTML. Either minted token opens the door; which one it was
    // decides what lies behind it. The cookie is the same credential on its
    // second visit: `?t=` gets a page in once, the page's `Set-Cookie` is what
    // a reload — which has no query left — presents instead. It adds no new
    // way in: the cookie only ever repeats a token that was already minted.
    // `Authorization: Bearer` is the same credential again, in the form a
    // script or HTTP client sends rather than a browser.
    let Some(access) = access_for(shared, request.token())
        .or_else(|| access_for(shared, request.bearer()))
        .or_else(|| {
            access_for(
                shared,
                request.cookie(&cookie_name(shared.port)).unwrap_or(""),
            )
        })
    else {
        crate::elog::event(
            "http",
            "request",
            serde_json::json!({"method": request.method, "path": request.path, "access": "denied"}),
        );
        return http::respond_error(
            stream,
            Some(&request),
            403,
            "missing or wrong access token — open the link cctop printed",
        );
    };
    crate::elog::event(
        "http",
        "request",
        serde_json::json!({
            "method": request.method,
            "path": request.path,
            "access": match access {
                Access::Full => "full",
                Access::ReadOnly => "readonly",
            },
        }),
    );

    // Before the router, so an armed fault covers every API route rather than
    // the handful somebody remembered to touch.
    #[cfg(feature = "debug")]
    if debug::intercept(stream, &request) {
        return;
    }

    let path = request.path.clone();
    #[cfg(feature = "debug")]
    if let Some(rest) = path.strip_prefix("/api/debug/")
        && debug::route(shared, stream, &request, rest)
    {
        return;
    }
    // A terminal's socket, for a page that reached cctop from elsewhere; see
    // [`relay_terminal`]. Before the app's files because it is not one.
    if let Some(rest) = path.strip_prefix("/rmux-ws/") {
        return relay_terminal(shared, stream, &request, rest, access);
    }
    // rmux's terminal app, for a session page to frame — see [`term`]. Static
    // and behind the same token as everything else, so only someone already
    // holding the link can load it.
    if let Some(file) = term::file(&path) {
        return http::respond_packed(
            stream,
            &request,
            file.content_type,
            file.body,
            file.cache,
            "",
            Some(http::TERM_POLICY),
        );
    }
    match path.as_str() {
        "/" => app_page(shared, stream, &request, access),
        // Fixed for a build but not named by one, so revalidated rather than
        // kept: a `304` is as cheap as a cache hit and survives an upgrade.
        "/favicon.svg" => http::respond_packed(
            stream,
            &request,
            "image/svg+xml",
            &FAVICON_P,
            http::REVALIDATE,
            "",
            None,
        ),
        "/manifest.webmanifest" => http::respond_packed(
            stream,
            &request,
            "application/manifest+json",
            &MANIFEST_P,
            http::REVALIDATE,
            "",
            None,
        ),
        "/api/sessions" => {
            let snapshot = current(shared);
            http::respond(
                stream,
                Some(&request),
                200,
                "application/json; charset=utf-8",
                snapshot.json.as_bytes(),
            );
        }
        "/api/hosts" => {
            let snapshot = current(shared);
            let body = serde_json::to_string(&snapshot.host_errors).unwrap_or_default();
            http::respond(
                stream,
                Some(&request),
                200,
                "application/json; charset=utf-8",
                body.as_bytes(),
            );
        }
        "/api/quota" => {
            // Served from memory: the document is rendered by whoever polled
            // the endpoints last, on their schedule rather than the browser's.
            let body = shared
                .quota
                .lock()
                .map(|q| q.clone())
                .unwrap_or_else(|_| quota::EMPTY.to_string());
            http::respond(
                stream,
                Some(&request),
                200,
                "application/json; charset=utf-8",
                body.as_bytes(),
            );
        }
        "/api/search" => api_search(shared, stream, &request),
        "/api/events" => events(shared, stream, &request),
        "/insight/optimize" => api_insight(shared, stream, &request, "optimize"),
        "/insight/compare" => api_insight(shared, stream, &request, "compare"),
        "/analytics" => app_page(shared, stream, &request, access),
        // Every tab's terminal, tiled. The page itself is harmless to a
        // read-only link — opening a terminal is the action, and that route
        // checks the credential as every action does.
        "/workspace" => app_page(shared, stream, &request, access),
        // The whole fleet's history in one document — the analytics page
        // filters and charts it client-side, so this one read-only route is
        // all the server owes it. Untrimmed buckets are affordable here
        // because the page fetches on its own cadence, not every refresh.
        "/api/analytics" => {
            let snapshot = current(shared);
            let built = analytics::build(&snapshot.sessions, shared.plan, &shared.store);
            json(stream, &request, &built);
        }
        // What can be handed a session's work, and whether this run will act at
        // all. The page asks once and hides the controls it cannot use, rather
        // than offering buttons that answer 404.
        "/api/agents" => {
            let body = serde_json::json!({
                "actions": shared.actions && access == Access::Full,
                "agents": actions::agents(),
            });
            http::respond(
                stream,
                Some(&request),
                200,
                "application/json; charset=utf-8",
                body.to_string().as_bytes(),
            );
        }
        // Not under /api/act/<id>/ — a launch names no session because there is
        // none yet. The guards are the shared ones: same link, same flag, same
        // insistence on a JSON POST.
        "/api/launch" => {
            let Some(body) = may_act(shared, stream, &request, access) else {
                return;
            };
            let agent = body
                .get("agent")
                .and_then(serde_json::Value::as_str)
                .unwrap_or_default();
            let cwd = body.get("cwd").and_then(serde_json::Value::as_str);
            match actions::launch_agent(agent, cwd) {
                Ok(done) => json(stream, &request, &done),
                Err((status, why)) => http::respond_error(stream, Some(&request), status, &why),
            }
        }
        _ if path.starts_with("/session/") => app_page(shared, stream, &request, access),
        // A popped-out terminal: the same app, drawing only the terminal.
        _ if path.starts_with("/window/") => app_page(shared, stream, &request, access),
        _ if path.starts_with("/api/report/") => {
            api_report(shared, stream, &request, &path["/api/report/".len()..]);
        }
        _ if path.starts_with("/api/chat/") && path.ends_with("/markdown") => {
            let id = &path["/api/chat/".len()..path.len() - "/markdown".len()];
            api_chat_markdown(shared, stream, &request, id);
        }
        _ if path.starts_with("/api/chat/") => {
            api_chat(shared, stream, &request, &path["/api/chat/".len()..]);
        }
        _ if path.starts_with("/api/access/") => {
            api_access(shared, stream, &request, &path["/api/access/".len()..]);
        }
        // Where this session can be handed: every (agent, account) pair but
        // the one it is on. Per session rather than folded into /api/agents,
        // because the pair that is left out is this session's — and empty for
        // a link that could not act on the answer, as /api/agents says
        // `actions: false` to the same link.
        _ if path.starts_with("/api/handoff/") => {
            let snapshot = current(shared);
            let id = &path["/api/handoff/".len()..];
            let Some(session) = find(&snapshot.sessions, id) else {
                return http::respond_error(stream, Some(&request), 404, NO_SUCH_SESSION);
            };
            let targets = match shared.actions && access == Access::Full && session.remote.is_none()
            {
                true => crate::handoff::targets(session),
                false => Vec::new(),
            };
            let targets: Vec<serde_json::Value> = targets
                .iter()
                .map(|t| {
                    serde_json::json!({
                        "agent": t.agent,
                        "account": t.account,
                        "label": t.label(),
                    })
                })
                .collect();
            json(stream, &request, &serde_json::json!({ "targets": targets }))
        }
        // The size rmux has fitted a tab's window to. The page's terminal asks
        // when you come back to it, compares it with its own grid, and if the
        // window is some other client's — the TUI's, typically — nudges its
        // own size so rmux hands the window over; rmux's `latest` does not
        // follow input on its own. Read-only, and only for cctop's tabs.
        _ if path.starts_with("/api/window/") => {
            let name = &path["/api/window/".len()..];
            if !tabs::is_tab(&crate::rmux::running(), name) {
                return http::respond_error(
                    stream,
                    Some(&request),
                    404,
                    "no open tab by that name",
                );
            }
            match crate::rmux::window_size(name) {
                // The browsers' own sizes ride along, so the page can tell the
                // window is already a browser's without measuring its drawing.
                Some((cols, rows)) => {
                    let web: Vec<[u16; 2]> = crate::rmux::clients(name)
                        .into_iter()
                        .filter(|(browser, _, _)| *browser)
                        .map(|(_, c, r)| [c, r])
                        .collect();
                    json(
                        stream,
                        &request,
                        &serde_json::json!({ "cols": cols, "rows": rows, "web": web }),
                    )
                }
                None => http::respond_error(stream, Some(&request), 404, "that tab has no window"),
            }
        }
        "/api/tabs" => {
            let snapshot = current(shared);
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(0, |d| d.as_secs());
            let list = tabs::build(crate::rmux::running_in_tab_order(), &snapshot.sessions, now);
            // Polled every few seconds by every open page and nearly always
            // the same answer, so it goes out tagged and comes back a `304`.
            // Tab names, titles and states — what the page shows in its bar,
            // nothing from a transcript — so it may sit in the browser cache.
            let body = serde_json::json!({ "tabs": list }).to_string();
            http::respond_revalidated(
                stream,
                &request,
                "application/json; charset=utf-8",
                body.as_bytes(),
            );
        }
        _ if path.starts_with("/api/tab/") => {
            api_tab(shared, stream, &request, &path["/api/tab/".len()..], access);
        }
        _ if path.starts_with("/api/act/") => {
            api_act(shared, stream, &request, &path["/api/act/".len()..], access);
        }
        _ => http::respond_error(stream, Some(&request), 404, "no such page"),
    }
}

/// Both transcript-search tiers over the current snapshot.
fn api_search(shared: &Shared, stream: &mut TcpStream, request: &Request) {
    let needle = request
        .query
        .get("q")
        .map(|q| q.trim().to_lowercase())
        .unwrap_or_default();
    // Under a few characters a query is a keystroke rather than a question;
    // answering one costs a scan of every transcript for a result that is
    // nearly all of them.
    if needle.chars().count() < search::MIN_CHARS {
        return json(stream, request, &serde_json::json!({ "hits": [] }));
    }
    let snapshot = current(shared);
    let hits = search::run(&shared.topics, &snapshot.sessions, &needle);
    json(stream, request, &serde_json::json!({ "hits": hits }))
}

/// The text `cctop optimize` or `cctop compare` would print, as plain text.
///
/// Priced the way the report route is: the analysis re-parses every transcript
/// because the tool calls it reasons about are never cached, so it happens on
/// request rather than on the refresh clock — the same cost the CLI pays when
/// asked the same question.
fn api_insight(shared: &Shared, stream: &mut TcpStream, request: &Request, which: &str) {
    let analyses = crate::insight::scan(shared.plan);
    let selected = crate::insight::only(&analyses, None);
    let text = match which {
        "optimize" => crate::insight::optimize::report(&selected),
        _ => crate::insight::compare::report(&selected),
    };
    http::respond(
        stream,
        Some(request),
        200,
        "text/plain; charset=utf-8",
        text.as_bytes(),
    );
}

/// Serve the conversation for one session.
///
/// `?before=<n>` pages backwards: the window returned ends just before turn
/// `n` rather than at the newest, which is how the page reaches turns the
/// first response counted but did not send.
fn api_chat(shared: &Shared, stream: &mut TcpStream, request: &Request, id: &str) {
    let snapshot = current(shared);
    let Some(session) = find(&snapshot.sessions, id) else {
        return http::respond_error(stream, Some(request), 404, NO_SUCH_SESSION);
    };
    let before = request
        .query
        .get("before")
        .and_then(|v| v.parse::<usize>().ok());
    let since = chat::Since {
        stamp: request
            .query
            .get("since")
            .cloned()
            .filter(|s| !s.is_empty()),
        after: request.query.get("after").and_then(|v| v.parse().ok()),
    };
    if session.remote.is_some() {
        let mut args = vec!["--chat".to_string(), session.session_id.clone()];
        if let Some(before) = before {
            args.extend(["--before".to_string(), before.to_string()]);
        }
        // The peer is asked the old question, because a cctop older than
        // `since` would reject the flag and the chat would not open at all.
        // The narrowing happens here instead: the ssh read is still whole,
        // but the browser gets — and redraws — only what changed.
        return remote_chat(shared, stream, request, session, &args, &since);
    }
    // Before the read, because skipping the read is the point: an unchanged
    // file is answered from its metadata alone.
    if let Some(path) = session.data_file.as_deref()
        && since.stamp.is_some()
        && since.stamp == chat::stamp_of(path)
    {
        return json(
            stream,
            request,
            &chat::Conversation {
                supported: true,
                stamp: since.stamp,
                unchanged: true,
                ..Default::default()
            },
        );
    }
    json(
        stream,
        request,
        &chat::build(session, before).narrowed(&since),
    );
}

/// A remote session's conversation, narrowed to what the page lacks.
///
/// A remote row has no file here to stamp, so its stamp is a hash of the
/// answer itself: the same conversation hashes the same, which is all a stamp
/// has to do.
fn remote_chat(
    shared: &Shared,
    stream: &mut TcpStream,
    request: &Request,
    session: &Session,
    args: &[String],
    since: &chat::Since,
) {
    if since.stamp.is_none() && since.after.is_none() {
        return remote_json(shared, stream, request, session, args);
    }
    let Some(host) = session
        .remote
        .as_ref()
        .and_then(|r| shared.hosts.get(&r.host))
    else {
        return remote_json(shared, stream, request, session, args);
    };
    let argv: Vec<&str> = args.iter().map(String::as_str).collect();
    let body = match host.run(&argv) {
        Ok(body) => body,
        Err(_) => return remote_json(shared, stream, request, session, args),
    };
    let Ok(mut conversation) = serde_json::from_str::<chat::Conversation>(&body) else {
        return http::respond(
            stream,
            Some(request),
            200,
            "application/json; charset=utf-8",
            body.as_bytes(),
        );
    };
    use std::hash::{Hash, Hasher};
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    body.hash(&mut hasher);
    conversation.stamp = Some(format!("r{:016x}", hasher.finish()));
    json(stream, request, &conversation.narrowed(since));
}

/// Serve one session's conversation as a markdown document to paste elsewhere.
///
/// `?variant=` picks the document: `conversation` (the default) is every turn
/// with its calls listed, `tools` adds each call's result, and `brief` is the
/// handoff brief — the summary [`crate::handoff`] writes for exactly the
/// paste-into-another-agent case, offered beside the transcript because on a
/// long session it is the one that fits.
///
/// Text rather than JSON, because the body *is* the answer: the page copies or
/// downloads it as it arrives, and `curl` gets a file it can use.
fn api_chat_markdown(shared: &Shared, stream: &mut TcpStream, request: &Request, id: &str) {
    let snapshot = current(shared);
    let Some(session) = find(&snapshot.sessions, id) else {
        return http::respond_error(stream, Some(request), 404, NO_SUCH_SESSION);
    };
    let variant = request.query.get("variant").map(String::as_str);
    let options = export::Options {
        tool_output: variant == Some("tools"),
    };
    let markdown = match (variant, session.remote.is_some()) {
        (Some("brief"), false) => {
            let data = shared.store.session_data_fresh(session);
            crate::handoff::build(session, Some(&data)).to_markdown()
        }
        (Some("brief"), true) => {
            match remote_text(shared, session, &["--handoff", &session.session_id]) {
                Ok(text) => text,
                Err(why) => return http::respond_error(stream, Some(request), 502, &why),
            }
        }
        (_, false) => export::render(session, &chat::whole(session), options),
        (_, true) => match remote_markdown(shared, session, options) {
            Ok(text) => text,
            Err(why) => return http::respond_error(stream, Some(request), 502, &why),
        },
    };
    http::respond(
        stream,
        Some(request),
        200,
        "text/markdown; charset=utf-8",
        markdown.as_bytes(),
    );
}

/// A remote row's export, rendered by the cctop that holds the transcript.
///
/// A peer older than `--export` rejects the flag, and an export that refused
/// to work against it would be the only session route that did. So it falls
/// back to the question every peer answers — `--chat`, the page's window — and
/// renders that here, which the document then says is partial.
fn remote_markdown(
    shared: &Shared,
    session: &Session,
    options: export::Options,
) -> Result<String, String> {
    let id = session.session_id.as_str();
    let mut args = vec!["--export", id];
    if options.tool_output {
        args.push("--tool-output");
    }
    let first = match remote_text(shared, session, &args) {
        Ok(text) => return Ok(text),
        Err(why) => why,
    };
    let body = remote_text(shared, session, &["--chat", id]).map_err(|_| first.clone())?;
    let conversation = serde_json::from_str::<chat::Conversation>(&body).map_err(|_| first)?;
    Ok(export::render(session, &conversation, options))
}

/// Run a cctop command on the machine a remote row came from, and return what
/// it printed — [`remote_json`] for a caller that wants the text in hand.
fn remote_text(shared: &Shared, session: &Session, args: &[&str]) -> Result<String, String> {
    let remote = session
        .remote
        .as_ref()
        .expect("only a remote row reaches remote_text");
    let Some(host) = shared.hosts.get(&remote.host) else {
        return Err(format!(
            "this session is on {}, which this run is not connected to",
            remote.host
        ));
    };
    host.run(args)
        .map_err(|why| format!("{} could not answer: {why}", remote.host))
}

/// Serve what one session can reach.
fn api_access(shared: &Shared, stream: &mut TcpStream, request: &Request, id: &str) {
    let snapshot = current(shared);
    let Some(session) = find(&snapshot.sessions, id) else {
        return http::respond_error(stream, Some(request), 404, NO_SUCH_SESSION);
    };
    if session.remote.is_some() {
        return remote_json(
            shared,
            stream,
            request,
            session,
            &["--access".to_string(), session.session_id.clone()],
        );
    }
    // The tool counts are the one part of this that needs the transcript, and
    // the rest of the answer is worth having without it — so a session whose
    // extraction fails still reports its instructions, skills and servers.
    let data = shared.store.session_data_fresh(session);
    json(stream, request, &crate::access::build(session, Some(&data)));
}

/// The guards every acting route shares, answered here once: a read-only link,
/// a `--no-actions` serve, and a request that is not a JSON POST each get their
/// own answer rather than a generic refusal. Returns the parsed body when the
/// request may act — a route that forgets to go through this is the bug the
/// shared shape exists to prevent.
fn may_act(
    shared: &Shared,
    stream: &mut TcpStream,
    request: &Request,
    access: Access,
) -> Option<serde_json::Value> {
    // Before every other guard, because it is not one: this is the link doing
    // what it was minted to do, and the answer names that rather than leaning
    // on a flag the read-only page never had.
    if access == Access::ReadOnly {
        http::respond_error(stream, Some(request), 403, "this link is read-only");
        return None;
    }
    if !shared.actions {
        http::respond_error(
            stream,
            Some(request),
            403,
            "this cctop serve is read-only — restart it without --no-actions, \
             and with a token, to act on a session",
        );
        return None;
    }
    // A `POST` with a JSON body, both of which are load-bearing: see the
    // module docs in `http` for why a form on another origin cannot be one.
    if !request.wants_json() {
        http::respond_error(
            stream,
            Some(request),
            405,
            "an action is a POST with a JSON body",
        );
        return None;
    }
    match request.json() {
        Ok(body) => Some(body),
        Err((status, why)) => {
            http::respond_error(stream, Some(request), status, why);
            None
        }
    }
}

/// Do something to a session: `/api/act/<verb>/<id>`.
///
/// One route for the three verbs rather than three, because the guards in front
/// of them are the whole security surface and they are identical — a route added
/// later that forgets one of them is the bug this shape prevents.
/// `POST /api/tab/<name>/terminal`: a tab's terminal, behind the same guards
/// as every action — the token, a JSON POST, a serve that allows actions.
fn api_tab(shared: &Shared, stream: &mut TcpStream, request: &Request, rest: &str, access: Access) {
    let Some(body) = may_act(shared, stream, request, access) else {
        return;
    };
    let Some(name) = rest.strip_suffix("/terminal") else {
        return http::respond_error(stream, Some(request), 404, "no such action");
    };
    let frontend = body
        .get("origin")
        .and_then(serde_json::Value::as_str)
        .and_then(actions::frontend_for);
    let fresh = body.get("fresh").and_then(serde_json::Value::as_bool) == Some(true);
    match actions::tab_terminal(name, frontend.as_deref(), fresh) {
        Ok(terminal) => json(stream, request, &terminal),
        Err((status, why)) => http::respond_error(stream, Some(request), status, &why),
    }
}

fn api_act(shared: &Shared, stream: &mut TcpStream, request: &Request, rest: &str, access: Access) {
    let Some(body) = may_act(shared, stream, request, access) else {
        return;
    };
    let Some((verb, id)) = rest.split_once('/') else {
        return http::respond_error(stream, Some(request), 404, "no such action");
    };

    let snapshot = current(shared);
    let Some(session) = find(&snapshot.sessions, id) else {
        return http::respond_error(stream, Some(request), 404, NO_SUCH_SESSION);
    };

    let field = |name: &str| {
        body.get(name)
            .and_then(serde_json::Value::as_str)
            .unwrap_or_default()
            .to_string()
    };
    // Also out of shape, and also not about the session it names: an image is
    // filed on this machine and the answer is where. The session id rides along
    // so the route is the same shape as its neighbours — one place that checks
    // the token, insists on a JSON POST, and refuses a read-only serve.
    if verb == "image" {
        return match actions::image(&field("data")) {
            Ok(filed) => json(stream, request, &filed),
            Err((status, why)) => http::respond_error(stream, Some(request), status, &why),
        };
    }
    // Answered before the others because it does not answer in their shape: a
    // terminal is a link and a reach, not a sentence about what was done.
    if verb == "terminal" {
        let frontend = actions::frontend_for(&field("origin"));
        let fresh = body.get("fresh").and_then(serde_json::Value::as_bool) == Some(true);
        return match actions::terminal(session, frontend.as_deref(), fresh) {
            Ok(terminal) => json(stream, request, &terminal),
            Err((status, why)) => http::respond_error(stream, Some(request), status, &why),
        };
    }
    let outcome = match verb {
        "send" => actions::send(session, &field("text")),
        "answer" => actions::answer(session, &field("choice")),
        "yolo" => match body.get("on").and_then(serde_json::Value::as_bool) {
            Some(on) => actions::yolo(session, on),
            None => Err((400, "YOLO is `{\"on\": true}` or `{\"on\": false}`".into())),
        },
        "resume" => actions::resume(session),
        "handoff" => {
            // The brief is built from the extraction, so this one pays for a
            // transcript read — the same read the report route makes, and for
            // the same reason: a brief assembled from the row alone carries the
            // header and none of the work.
            let data = shared.store.session_data_fresh(session);
            actions::handoff(
                session,
                Some(&data),
                &field("agent"),
                Some(&field("account")),
            )
        }
        _ => return http::respond_error(stream, Some(request), 404, "no such action"),
    };
    match outcome {
        Ok(done) => json(stream, request, &done),
        Err((status, why)) => http::respond_error(stream, Some(request), status, &why),
    }
}

/// Send a value as JSON, or say why it could not be rendered.
fn json<T: serde::Serialize>(stream: &mut TcpStream, request: &Request, value: &T) {
    match serde_json::to_string(value) {
        Ok(body) => http::respond(
            stream,
            Some(request),
            200,
            "application/json; charset=utf-8",
            body.as_bytes(),
        ),
        Err(e) => http::respond_error(
            stream,
            Some(request),
            503,
            &format!("could not render that: {e}"),
        ),
    }
}

/// The current snapshot, cloned out of the lock rather than held under it.
fn current(shared: &Shared) -> Arc<Snapshot> {
    match shared.latest.lock() {
        Ok(latest) => Arc::clone(&latest),
        // A panicked refresh thread would poison this. The stale snapshot in
        // there is still true of some moment, and is a better answer than 500.
        Err(poisoned) => Arc::clone(&poisoned.into_inner()),
    }
}

/// Serve the app page, with the token stitched in.
///
/// The page needs the token to make its own requests, and it cannot read the
/// one in its URL without either parsing `location` in script — which is fine —
/// or being handed it. It is handed it, because the same page is fetched with
/// no token at all under `--no-token` and a single substitution keeps both
/// cases on one code path.
///
/// What it costs to send is decided by the fact that it changes only with the
/// credential. There are two of those per run, so the page is built and
/// compressed once for each — most of a megabyte, a third of that compressed,
/// and it used to be gzipped again on every request — and sent from then on as
/// it was packed.
///
/// Cached as `private, no-cache` with an ETag, which is the most a page holding
/// the token can safely be:
///
/// - **Not `no-store`.** That made every reload download the whole UI again —
///   300 KB over a tunnel to a phone — for a page that had not changed. With
///   an ETag a reload is a `304` and no body.
/// - **Not `max-age`.** The page is the one thing that knows which cctop it
///   was built by; kept without asking, it would go on running the old UI
///   against an upgraded server, or a token a restart had already replaced.
///   `no-cache` asks every time, and the answer is free when nothing moved.
/// - **`private`.** The body carries a credential, so no shared cache between
///   here and the browser — the tunnel's CDN, a proxy — may keep it. The
///   browser's own copy adds little to what it already holds: the token is in
///   its history, in the access cookie, and in the page it is showing.
///
/// The tag is a hash of the body, so it differs per credential: the read-only
/// link's page never answers a revalidation of the full one's.
fn app_page(shared: &Shared, stream: &mut TcpStream, request: &Request, access: Access) {
    // Which credential the page carries — and whether it may act — is decided
    // by the one the request arrived with, not by the run: the read-only link
    // opens the same page wired to the narrower token, so a page it hands out
    // can neither act nor leak the token that could.
    let (credential, actions) = match access {
        Access::Full => (shared.token.as_str(), shared.actions),
        Access::ReadOnly => (shared.readonly.as_str(), false),
    };
    let page = shared.pages[access as usize].get_or_init(|| {
        http::Packed::new(APP_HTML.replace("__CCTOP_CONFIG__", &app_config(credential, actions)))
    });

    // Hand the credential back as a cookie so a reload — which has no `?t=`
    // left, the page having stripped it — still gets in. Only a request that
    // presented a token mints one: a page that got in on the cookie already
    // has it, and a read-only link does not push a full cookie out of a jar
    // that holds one.
    let mut headers = String::new();
    if !request.token().is_empty() && !credential.is_empty() {
        let held = request
            .cookie(&cookie_name(shared.port))
            .and_then(|c| access_for(shared, c));
        if !(access == Access::ReadOnly && held == Some(Access::Full)) {
            headers = format!(
                "Set-Cookie: {}={credential}; Path=/; HttpOnly; SameSite=Strict\r\n",
                cookie_name(shared.port)
            );
        }
    }
    http::respond_packed(
        stream,
        request,
        "text/html; charset=utf-8",
        page,
        http::REVALIDATE,
        &headers,
        None,
    );
}

/// Carry a terminal's WebSocket from the page to rmux's loopback endpoint.
///
/// A page opened from another machine — the LAN, or `--tunnel`'s
/// trycloudflare host — cannot reach `ws://127.0.0.1:<port>`, which is where
/// rmux serves a share. It used to get a second tunnel for that, through
/// localhost.run, which dropped often enough to be most of the refused frames.
/// Instead the share's endpoint is rewritten to this route at the page's own
/// origin (see `rmux::relay_link`), and the socket rides whatever road the page
/// came in on, the quick tunnel included: it carries WebSocket upgrades.
///
/// Only to a port a share of cctop's named — anything else would make this a
/// way into every local service that trusts localhost — and only for a link
/// that may act, since the far end is a shell. The handshake is passed on
/// whole, Host rewritten, and then the bytes are pumped both ways until either
/// side closes. The connection holds one of [`MAX_CONNECTIONS`] while it lives.
fn relay_terminal(
    shared: &Shared,
    stream: &mut TcpStream,
    request: &Request,
    rest: &str,
    access: Access,
) {
    use std::io::{Read, Write};
    if access != Access::Full || !shared.actions {
        return http::respond_error(
            stream,
            Some(request),
            403,
            "this link cannot open terminals",
        );
    }
    if !request.is_websocket() {
        return http::respond_error(
            stream,
            Some(request),
            400,
            "a terminal socket is a WebSocket",
        );
    }
    let Some((port, tail)) = rest.split_once('/') else {
        return http::respond_error(stream, Some(request), 404, "no such terminal");
    };
    let Some(port) = port
        .parse::<u16>()
        .ok()
        .filter(|p| crate::rmux::relayable(*p))
    else {
        return http::respond_error(stream, Some(request), 404, "no such terminal");
    };
    let Ok(mut upstream) = TcpStream::connect(("127.0.0.1", port)) else {
        return http::respond_error(
            stream,
            Some(request),
            502,
            "the terminal's share has closed",
        );
    };
    let query = match request.raw_query() {
        "" => String::new(),
        q => format!("?{q}"),
    };
    let mut head = format!("GET /{tail}{query} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n");
    for (name, value) in request.headers() {
        if !name.eq_ignore_ascii_case("host") {
            head.push_str(&format!("{name}: {value}\r\n"));
        }
    }
    head.push_str("\r\n");
    if upstream.write_all(head.as_bytes()).is_err() {
        return http::respond_error(
            stream,
            Some(request),
            502,
            "the terminal's share has closed",
        );
    }
    // A terminal sits quiet for minutes at a time; the request deadlines that
    // guard every other route would cut it off mid-session.
    let _ = stream.set_read_timeout(None);
    let _ = stream.set_write_timeout(None);
    let (Ok(mut down_read), Ok(mut up_write)) = (stream.try_clone(), upstream.try_clone()) else {
        return;
    };
    let pump = std::thread::spawn(move || {
        let mut buf = [0u8; 16 * 1024];
        while let Ok(n) = upstream.read(&mut buf) {
            if n == 0 || down_read.write_all(&buf[..n]).is_err() {
                break;
            }
        }
        let _ = down_read.shutdown(std::net::Shutdown::Both);
    });
    let mut buf = [0u8; 16 * 1024];
    while let Ok(n) = stream.read(&mut buf) {
        if n == 0 || up_write.write_all(&buf[..n]).is_err() {
            break;
        }
    }
    let _ = up_write.shutdown(std::net::Shutdown::Both);
    let _ = pump.join();
}

/// What the app page is told about this run, as the JSON that replaces
/// `__CCTOP_CONFIG__` inside a `<script type="application/json">`.
///
/// `<` is escaped because the JSON sits inside a script element, and a home
/// directory is free to contain `</script>`: the browser would end the element
/// there, whatever the JSON parser would have made of it.
fn app_config(credential: &str, actions: bool) -> String {
    let home = dirs::home_dir()
        .map(|p| p.to_string_lossy().into_owned())
        .unwrap_or_default();
    serde_json::json!({
        "token": credential,
        "actions": actions,
        "home": home,
        "version": env!("CARGO_PKG_VERSION"),
    })
    .to_string()
    .replace('<', "\\u003c")
}

/// Hold an SSE stream open, sending each new snapshot as it lands.
fn events(shared: &Shared, stream: &mut TcpStream, request: &Request) {
    events_every(shared, stream, request, SSE_KEEPALIVE);
}

/// [`events`], with the ping interval as an argument so a test can wait for
/// one without waiting fifteen seconds.
///
/// Two shapes of stream. A session page asks for its own row and is sent it,
/// whole, only when it moved. The table is sent whole once, as `sessions`, and
/// after that as `rows`: only the rows that changed since this stream last
/// wrote, and the order of ids when that moved — see [`table_delta`].
fn events_every(shared: &Shared, stream: &mut TcpStream, request: &Request, keepalive: Duration) {
    // A session page asks for its own row only. It used to be sent the whole
    // table on every refresh — every session on the machine and on every
    // `--host`, re-parsed in the browser several times a minute — to read one
    // row's state out of it.
    let only = request
        .query
        .get("session")
        .cloned()
        .filter(|s| !s.is_empty());
    // A page that knows the field-level shape says so; one loaded before this
    // server was upgraded does not, and keeps the shape it can apply rather
    // than silently dropping every patch it was sent.
    let fine = request.query.get("rows").is_some_and(|v| v == "patch");
    let mut last_row: Option<String> = None;
    // The table this stream last wrote, which is what the next delta is taken
    // against. Per stream rather than per snapshot: a stream that fell behind
    // skips snapshots, and its delta has to cover everything it skipped.
    let mut last_table: Option<Arc<Snapshot>> = None;
    let Ok(mut sse) = EventStream::open(stream, request) else {
        return;
    };
    crate::elog::event("sse", "open", serde_json::json!({}));

    let mut sent = 0u64;
    let mut wrote = Instant::now();
    // Which write lost the client is the difference between "browser closed"
    // and "network went" — the reason is kept rather than collapsed.
    let by = loop {
        // The wait is what makes an idle stream free: no polling, and one
        // wakeup per refresh rather than one per connection per tick. It is
        // bounded by the ping that is next due, not by a fixed window: a
        // refresh lands every couple of seconds, and a stream whose own rows
        // never move would otherwise wake on every one, write nothing, and
        // never come round to pinging at all.
        let snapshot = {
            let Ok(latest) = shared.latest.lock() else {
                break "lock";
            };
            let due = keepalive.saturating_sub(wrote.elapsed());
            let (latest, _) = match shared
                .updated
                .wait_timeout_while(latest, due, |s| s.version <= sent)
            {
                Ok(pair) => pair,
                Err(_) => break "wait",
            };
            Arc::clone(&latest)
        };

        if snapshot.version > sent {
            let frame = match &only {
                None => match &last_table {
                    None => Some(("sessions", snapshot.json.clone())),
                    Some(before) => match fine {
                        true => table_patch(&before.rows, &snapshot.rows),
                        false => table_delta(&before.rows, &snapshot.rows),
                    }
                    .map(|delta| ("rows", delta)),
                },
                Some(id) => {
                    let row = snapshot
                        .rows
                        .iter()
                        .find(|(sid, _)| sid == id)
                        .or_else(|| {
                            snapshot
                                .rows
                                .iter()
                                .find(|(sid, _)| sid.starts_with(id.as_str()))
                        })
                        .map(|(_, r)| r.clone());
                    // Only when it moved: a refresh that changed some other
                    // row is nothing to this page.
                    match row {
                        Some(row) if last_row.as_ref() != Some(&row) => {
                            last_row = Some(row.clone());
                            Some(("sessions", format!("[{row}]")))
                        }
                        _ => None,
                    }
                }
            };
            if let Some((event, body)) = frame {
                if sse.send(event, &body).is_err() {
                    break "send";
                }
                wrote = Instant::now();
            }
            if only.is_none() {
                last_table = Some(Arc::clone(&snapshot));
            }
            sent = snapshot.version;
        }

        // Nothing written for a whole interval — no new snapshot, or none that
        // moved anything this stream shows — so say the stream is alive, which
        // is also what surfaces a client that quietly went away.
        if wrote.elapsed() >= keepalive {
            if sse.ping().is_err() {
                break "ping";
            }
            wrote = Instant::now();
        }
    };
    crate::elog::event(
        "sse",
        "close",
        serde_json::json!({ "by": by, "sent": sent }),
    );
}

/// What a stream that last sent the table `before` needs to be told to be
/// showing `after`, as the body of a `rows` event — `None` when nothing moved.
///
/// The body is `{"set": [row, …]}`, the rows that are new or changed, plus
/// `"order": [id, …]` when the sequence of ids is not what it was — which is
/// also how a row that went away is said, by its id no longer being in it.
///
/// This exists because the whole table on every refresh is more than some
/// links carry. On a shared box with eight hundred sessions it is 3.6 MB every
/// two seconds; through an ssh forward or a trycloudflare tunnel that queued
/// behind itself, and the page fell tens of seconds behind the server — still
/// offering Allow on a prompt that had been answered, which the server then
/// refused, because *it* was current. A refresh moves a handful of rows, so
/// the stream now carries a handful of rows.
///
/// Rows are compared as the strings already rendered for the snapshot, so a
/// row counts as changed exactly when its JSON did — no second notion of
/// equality to drift from what the page would have seen.
///
/// This is the shape a page from before [`table_patch`] understands, and what
/// a stream that did not ask for the finer one is still sent.
///
/// ponytail: rows are told apart by session id, as the page already does for
/// its keys and its routes. Two rows sharing one — the same session reported
/// by this machine and by a `--host` — are one row to the delta, and the page
/// shows whichever it was sent last.
fn table_delta(before: &[(String, String)], after: &[(String, String)]) -> Option<String> {
    let had: HashMap<&str, &str> = before
        .iter()
        .map(|(id, row)| (id.as_str(), row.as_str()))
        .collect();
    let set: Vec<&str> = after
        .iter()
        .filter(|(id, row)| had.get(id.as_str()) != Some(&row.as_str()))
        .map(|(_, row)| row.as_str())
        .collect();
    let reordered =
        before.len() != after.len() || before.iter().zip(after).any(|((a, _), (b, _))| a != b);
    if set.is_empty() && !reordered {
        return None;
    }
    let mut body = format!("{{\"set\":[{}]", set.join(","));
    if reordered {
        let ids: Vec<&str> = after.iter().map(|(id, _)| id.as_str()).collect();
        body.push_str(",\"order\":");
        body.push_str(&serde_json::to_string(&ids).unwrap_or_else(|_| "[]".into()));
    }
    body.push('}');
    Some(body)
}

/// [`table_delta`], down to the field: what a page that asked for it is sent.
///
/// [`table_delta`] still sent too much. A session row is four kilobytes —
/// costs per model, subagents, an activity histogram — and what moves between
/// two refreshes is usually one field of it: a running agent's CPU figure,
/// which changes on every tick for every live row. And the order of eight
/// hundred ids was sent whole whenever one session became the newest, which is
/// most refreshes on a busy machine. Measured on one, both together were most
/// of 5 MB a minute. So:
///
/// - A row that existed before goes as `"patch": [{"id", "to": {field: value},
///   "drop": [field]}]` — the fields whose JSON changed, and the ones that are
///   gone — unless that would be no shorter than the row, which then goes whole
///   in `"set"` as before. New rows always go whole.
/// - A changed order goes as `"head": [id, …]` and `"gone": [id, …]`: the new
///   order is `head`, then every row the page had that is in neither list, in
///   the order the page had them. A refresh moves a few rows to the top and
///   leaves the rest as they were, which is a head of a few ids. Any other
///   reshuffle that would make `head` as long as the table goes as `"order"`.
///
/// Fields are compared as parsed JSON, and only for rows whose string already
/// differed, so a quiet refresh parses nothing.
fn table_patch(before: &[(String, String)], after: &[(String, String)]) -> Option<String> {
    use serde_json::{Map, Value};
    let had: HashMap<&str, &str> = before
        .iter()
        .map(|(id, row)| (id.as_str(), row.as_str()))
        .collect();
    let mut set: Vec<&str> = Vec::new();
    let mut patch: Vec<Value> = Vec::new();
    for (id, row) in after {
        let Some(old) = had.get(id.as_str()) else {
            set.push(row);
            continue;
        };
        if old == row {
            continue;
        }
        let parsed = (
            serde_json::from_str::<Map<String, Value>>(old),
            serde_json::from_str::<Map<String, Value>>(row),
        );
        let (Ok(old), Ok(new)) = parsed else {
            set.push(row);
            continue;
        };
        let to: Map<String, Value> = new
            .iter()
            .filter(|(k, v)| old.get(*k) != Some(*v))
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        let drop: Vec<&String> = old.keys().filter(|k| !new.contains_key(*k)).collect();
        let mut one = serde_json::json!({ "id": id, "to": to });
        if !drop.is_empty() {
            one["drop"] = serde_json::json!(drop);
        }
        match one.to_string().len() < row.len() {
            true => patch.push(one),
            false => set.push(row),
        }
    }

    let reordered =
        before.len() != after.len() || before.iter().zip(after).any(|((a, _), (b, _))| a != b);
    if set.is_empty() && patch.is_empty() && !reordered {
        return None;
    }
    let mut body = format!("{{\"set\":[{}]", set.join(","));
    if !patch.is_empty() {
        body.push_str(",\"patch\":");
        body.push_str(&Value::from(patch).to_string());
    }
    if reordered {
        let ids: Vec<&str> = after.iter().map(|(id, _)| id.as_str()).collect();
        let list = |ids: &[&str]| serde_json::to_string(ids).unwrap_or_else(|_| "[]".into());
        match order_edit(before, &ids) {
            Some((head, gone)) => {
                body.push_str(&format!(
                    ",\"head\":{},\"gone\":{}",
                    list(&head),
                    list(&gone)
                ));
            }
            None => body.push_str(&format!(",\"order\":{}", list(&ids))),
        }
    }
    body.push('}');
    Some(body)
}

/// The order `after` as an edit of `before`'s: the ids to put first, and the
/// ids that are gone — with every other row keeping its place relative to the
/// others. `None` when that is no shorter than the whole order, or when an id
/// appears twice (see [`table_delta`]'s ponytail), where "its place" means
/// nothing.
///
/// `head` is the shortest prefix of `after` whose removal leaves the rest of
/// `after` in the order `before` had it. Found from the back: the longest tail
/// of `after` that `before` contains in sequence, matched greedily from the
/// end, which is what makes the head left over the shortest. A row new to the
/// table cannot be in that tail — `before` does not have it — so it is always
/// in `head`, where the page finds it next to its row in `set`.
fn order_edit<'a>(
    before: &'a [(String, String)],
    after: &[&'a str],
) -> Option<(Vec<&'a str>, Vec<&'a str>)> {
    use std::collections::HashSet;
    let now: HashSet<&str> = after.iter().copied().collect();
    let was: HashSet<&str> = before.iter().map(|(id, _)| id.as_str()).collect();
    if now.len() != after.len() || was.len() != before.len() {
        return None;
    }
    let gone: Vec<&str> = before
        .iter()
        .map(|(id, _)| id.as_str())
        .filter(|id| !now.contains(id))
        .collect();
    let kept: Vec<&str> = before
        .iter()
        .map(|(id, _)| id.as_str())
        .filter(|id| now.contains(id))
        .collect();
    let mut j = kept.len();
    let mut tail = 0;
    for id in after.iter().rev() {
        match kept[..j].iter().rposition(|k| k == id) {
            Some(at) => {
                j = at;
                tail += 1;
            }
            None => break,
        }
    }
    let head: Vec<&str> = after[..after.len() - tail].to_vec();
    (head.len() + gone.len() < after.len()).then_some((head, gone))
}

/// Build and send one session's report.
fn api_report(shared: &Shared, stream: &mut TcpStream, request: &Request, id: &str) {
    let snapshot = current(shared);
    let Some(session) = find(&snapshot.sessions, id) else {
        return http::respond_error(stream, Some(request), 404, NO_SUCH_SESSION);
    };

    // A remote row names a transcript on the machine it came from, so the
    // report is built there and the body relayed — never parsed here.
    if session.remote.is_some() {
        return remote_json(
            shared,
            stream,
            request,
            session,
            &["--report".to_string(), session.session_id.clone()],
        );
    }

    // The one expensive call in the whole server, and the reason it is on this
    // route alone: the report is built from tool arguments and per-request
    // context readings, neither of which the cache carries.
    let data = shared.store.session_data_fresh(session);
    let built = report::build(session, &data, shared.plan);
    match serde_json::to_string(&built) {
        Ok(body) => http::respond(
            stream,
            Some(request),
            200,
            "application/json; charset=utf-8",
            body.as_bytes(),
        ),
        Err(e) => http::respond_error(
            stream,
            Some(request),
            503,
            &format!("could not render the report: {e}"),
        ),
    }
}

/// Answer a session route for a remote row by asking the cctop on the machine
/// it came from, and relay the body verbatim.
///
/// The far side runs `cctop --report`, `--chat` or `--access` over the same ssh
/// channel `--host` polls with, so a remote report is as reachable as the row
/// was. Building the answer here instead would read whatever happens to live
/// at the same path on *this* filesystem — the failure `Session::remote`
/// exists to prevent.
fn remote_json(
    shared: &Shared,
    stream: &mut TcpStream,
    request: &Request,
    session: &Session,
    args: &[String],
) {
    let remote = session
        .remote
        .as_ref()
        .expect("only a remote row reaches remote_json");
    let Some(host) = shared.hosts.get(&remote.host) else {
        // The row is real — it arrived over ssh — so a host this map does not
        // know is a spec that was dropped between polls, not a session that
        // does not exist. 502 rather than 404 says which of those it is.
        return http::respond_error(
            stream,
            Some(request),
            502,
            &format!(
                "this session is on {}, which this run is not connected to",
                remote.host
            ),
        );
    };
    let argv: Vec<&str> = args.iter().map(String::as_str).collect();
    match host.run(&argv) {
        Ok(body) => http::respond(
            stream,
            Some(request),
            200,
            "application/json; charset=utf-8",
            body.as_bytes(),
        ),
        // The far side's own words are usually the answer: "command not
        // found", "no session id starts with …", or — for a cctop older than
        // these flags — clap's "unexpected argument".
        Err(why) => http::respond_error(
            stream,
            Some(request),
            502,
            &format!("{} could not answer: {why}", remote.host),
        ),
    }
}

/// Find a session by id, or by a prefix that matches exactly one.
///
/// Prefixes because session ids are uuids and the report link is a thing people
/// paste into chat. An ambiguous prefix resolves to nothing rather than to the
/// first match — the wrong session's costs is a worse answer than none.
fn find<'a>(sessions: &'a [Session], id: &str) -> Option<&'a Session> {
    if id.is_empty() {
        return None;
    }
    if let Some(exact) = sessions.iter().find(|s| s.session_id == id) {
        return Some(exact);
    }
    let mut matches = sessions.iter().filter(|s| s.session_id.starts_with(id));
    let first = matches.next()?;
    matches.next().is_none().then_some(first)
}

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

    fn session(id: &str) -> Session {
        Session::new(crate::pricing::Provider::Claude, id.into())
    }

    #[test]
    fn a_token_matches_only_itself() {
        let token = new_token();
        assert!(token_matches(&token, &token));
        assert!(!token_matches(&token, ""));
        assert!(!token_matches(&token, &token[..token.len() - 1]));

        let mut wrong = token.clone();
        // Last byte, which is where a comparison that returned early would have
        // stopped leaking and started being wrong.
        wrong.pop();
        wrong.push(if token.ends_with('a') { 'b' } else { 'a' });
        assert!(!token_matches(&token, &wrong));
    }

    #[test]
    fn tokens_differ_between_runs() {
        // Not a randomness test — that belongs to the OS — but it does catch a
        // fallback that has stopped varying at all.
        assert_ne!(new_token(), new_token());
        assert_eq!(new_token().len(), TOKEN_BYTES * 2);
    }

    #[test]
    fn sessions_resolve_by_id_and_by_unambiguous_prefix() {
        let rows = vec![session("abc123"), session("abd999"), session("zz")];
        assert_eq!(find(&rows, "abc123").unwrap().session_id, "abc123");
        assert_eq!(find(&rows, "abc").unwrap().session_id, "abc123");
        assert_eq!(find(&rows, "zz").unwrap().session_id, "zz");
    }

    #[test]
    fn an_ambiguous_prefix_resolves_to_nothing() {
        let rows = vec![session("abc123"), session("abd999")];
        // `ab` matches both. Answering with either one would put another
        // session's costs under this one's heading.
        assert!(find(&rows, "ab").is_none());
        assert!(find(&rows, "").is_none());
        assert!(find(&rows, "nope").is_none());
    }

    #[test]
    fn a_tunnel_refuses_the_two_combinations_that_undo_it() {
        // A tunnel with no token is a public prompt box wired to a live agent.
        assert!(tunnel_objection(true, true, true).is_some());
        // A tunnel plus a wider bind is two exposures for one request.
        assert!(tunnel_objection(true, false, false).is_some());
        // And what it is actually for.
        assert!(tunnel_objection(true, true, false).is_none());
        // Without --tunnel neither of those is this function's business:
        // --bind and --no-token are each documented on their own terms.
        assert!(tunnel_objection(false, false, true).is_none());
    }

    #[test]
    fn the_app_page_is_built_and_takes_its_config() {
        // The built app is committed (web/ is its source); a checkout where it
        // is missing or stale-shaped would serve a page that cannot start.
        assert!(APP_HTML.contains(
            r#"<script id="cctop-config" type="application/json">__CCTOP_CONFIG__</script>"#
        ));
        assert!(APP_HTML.contains("<div id=\"root\"></div>"));
        // Everything inlined: a page that loads a file of its own breaks under
        // the content policy, which allows no URL at all.
        assert!(!APP_HTML.contains(" src=\"/assets/"));
        assert!(!APP_HTML.contains("<link rel=\"stylesheet\""));
    }

    #[test]
    fn the_app_config_cannot_end_its_script_element() {
        let json = app_config("abc", true);
        assert!(!json.contains('<'));
        let v: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(v["token"], "abc");
        assert_eq!(v["actions"], true);
    }

    #[test]
    fn a_token_is_safe_to_paste_into_a_url_unescaped() {
        // The links cctop prints, and the `Set-Cookie` value `app_page` writes,
        // carry the token by interpolation rather than by escaping. That is
        // only correct while the generator stays hex.
        let token = new_token();
        assert!(
            token.chars().all(|c| c.is_ascii_hexdigit()),
            "token {token} would need escaping in a URL"
        );
    }

    #[test]
    fn a_named_port_is_not_stepped_past() {
        // Bind one, then ask for the same one: with `--port` given that is an
        // error, and without it the search moves on.
        let held = TcpListener::bind("127.0.0.1:0").expect("a loopback port");
        let taken = held.local_addr().unwrap().port();

        assert!(listen("127.0.0.1", taken, true).is_err());
        let stepped = listen("127.0.0.1", taken, false).expect("the search finds a free port");
        assert_ne!(stepped.local_addr().unwrap().port(), taken);
    }

    fn shared(token: &str, readonly: &str) -> Shared {
        Shared {
            token: token.to_string(),
            readonly: readonly.to_string(),
            actions: true,
            port: 7777,
            plan: Plan::Retail,
            latest: Mutex::new(Arc::new(Snapshot {
                version: 0,
                json: "[]".to_string(),
                rows: Vec::new(),
                sessions: Vec::new(),
                host_errors: Vec::new(),
            })),
            updated: Condvar::new(),
            store: crate::cache::Store::new(),
            quota: Mutex::new(quota::EMPTY.to_string()),
            topics: Mutex::new(search::Topics::default()),
            notify: None,
            hosts: HashMap::new(),
            pages: Default::default(),
        }
    }

    #[test]
    fn the_readonly_token_opens_the_page_but_not_the_actions() {
        // `access_for` is the single gate the whole distinction hangs on, so
        // its table is what is worth pinning: both minted tokens get in, and
        // which one it was is what `/api/act/*` later refuses on.
        let guarded = shared("full", "view");
        assert_eq!(access_for(&guarded, "full"), Some(Access::Full));
        assert_eq!(access_for(&guarded, "view"), Some(Access::ReadOnly));
        assert_eq!(access_for(&guarded, "wrong"), None);
        assert_eq!(access_for(&guarded, ""), None);

        // A tokenless run has nothing to withhold, so it grants full access
        // to everything — the same answer `--no-token` has always given.
        let open = shared("", "");
        assert_eq!(access_for(&open, ""), Some(Access::Full));
        assert_eq!(access_for(&open, "anything"), Some(Access::Full));
    }

    /// Ask `api_search` the way the router does — a request parsed off a real
    /// socket — and hand back whatever it wrote.
    fn api_search_body(shared: &Shared, target: &str) -> String {
        use std::io::Read;
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let mut client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
        let (mut server, _) = listener.accept().unwrap();
        client
            .write_all(format!("GET {target} HTTP/1.1\r\n\r\n").as_bytes())
            .unwrap();
        let request = Request::parse(&server).unwrap();
        api_search(shared, &mut server, &request);
        // Closing the answering half is what lets the client read the response
        // to its end rather than waiting out the connection.
        drop(server);
        let mut raw = String::new();
        client.read_to_string(&mut raw).unwrap();
        raw
    }

    /// A request with no `q` — or one shorter than a question — is an empty
    /// answer, not an error and not a scan of every transcript.
    #[test]
    fn a_search_without_a_real_query_finds_nothing() {
        let shared = shared("", "");
        for target in ["/api/search", "/api/search?q=", "/api/search?q=ab"] {
            let body = api_search_body(&shared, target);
            assert!(body.contains("\"hits\":[]"), "{target}: {body}");
        }
    }

    /// The route end to end, minus the listener: a session whose transcript
    /// holds the word comes back under `session_id`, the field the page joins
    /// its rows on.
    ///
    /// Only the hit direction is pinned here. A miss asks the topical tier,
    /// which on a machine with the model fetched would load it and rewrite the
    /// real embedding cache — a side effect no unit test should have; the
    /// literal miss itself is covered in `session::search`'s tests.
    #[test]
    fn a_search_names_the_session_whose_transcript_matched() {
        let path =
            std::env::temp_dir().join(format!("cctop-serve-search-{}.jsonl", std::process::id()));
        std::fs::write(&path, "{\"text\":\"please fix the flywheel\"}\n").unwrap();
        let mut s = Session::new(crate::pricing::Provider::Claude, "sess-1".into());
        s.data_file = Some(path.clone());

        let shared = shared("", "");
        *shared.latest.lock().unwrap() = Arc::new(Snapshot {
            version: 1,
            json: "[]".to_string(),
            rows: Vec::new(),
            sessions: vec![s],
            host_errors: Vec::new(),
        });

        let body = api_search_body(&shared, "/api/search?q=flywheel");
        let _ = std::fs::remove_file(path);
        assert!(body.contains("\"session_id\":\"sess-1\""), "{body}");
        assert!(body.contains("flywheel"), "{body}");
    }

    /// A whole request through the router — gate, scope check and route — and
    /// the status line it was answered with.
    fn status_of(shared: &Shared, method: &str, target: &str, headers: &str) -> String {
        let raw = response_of(shared, method, target, headers);
        raw.lines().next().unwrap_or_default().to_string()
    }

    /// A whole request through the router, and the whole raw response.
    fn response_of(shared: &Shared, method: &str, target: &str, headers: &str) -> String {
        use std::io::Read;
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let mut client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
        let (mut server, _) = listener.accept().unwrap();
        client
            .write_all(format!("{method} {target} HTTP/1.1\r\n{headers}\r\n").as_bytes())
            .unwrap();
        serve_connection(shared, &mut server);
        drop(server);
        let mut raw = String::new();
        client.read_to_string(&mut raw).unwrap();
        raw
    }

    /// The export route end to end: behind the same token as its neighbours,
    /// served as markdown, the whole conversation, and tool results only when
    /// the variant asks for them.
    #[test]
    fn the_markdown_route_serves_the_conversation_behind_the_token() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("sess-md.jsonl");
        let lines = [
            r###"{"type":"user","timestamp":"2026-10-07T10:00:00Z","message":{"content":"## run the tests"}}"###,
            r#"{"type":"assistant","timestamp":"2026-10-07T10:00:05Z","message":{"content":[{"type":"text","text":"Running them."},{"type":"tool_use","id":"t1","name":"Bash","input":{"command":"cargo test"}}]}}"#,
            r#"{"type":"user","timestamp":"2026-10-07T10:00:09Z","message":{"content":[{"type":"tool_result","tool_use_id":"t1","content":"test result: ok"}]}}"#,
        ];
        std::fs::write(&path, lines.join("\n")).unwrap();
        let mut s = Session::new(crate::pricing::Provider::Claude, "sess-md".into());
        s.data_file = Some(path);

        let guarded = shared("full", "view");
        *guarded.latest.lock().unwrap() = Arc::new(Snapshot {
            version: 1,
            json: "[]".to_string(),
            rows: Vec::new(),
            sessions: vec![s],
            host_errors: Vec::new(),
        });
        let bearer = "Authorization: Bearer view\r\n";

        let plain = response_of(&guarded, "GET", "/api/chat/sess-md/markdown", bearer);
        assert!(plain.starts_with("HTTP/1.1 200 "), "{plain}");
        assert!(plain.contains("text/markdown"), "{plain}");
        assert!(plain.contains("> ## run the tests"), "{plain}");
        assert!(plain.contains("- **Bash** `cargo test`"), "{plain}");
        assert!(!plain.contains("test result: ok"), "{plain}");

        let tools = response_of(
            &guarded,
            "GET",
            "/api/chat/sess-md/markdown?variant=tools",
            bearer,
        );
        assert!(tools.contains("test result: ok"), "{tools}");

        let missing = status_of(&guarded, "GET", "/api/chat/nope/markdown", bearer);
        assert!(missing.contains(" 404 "), "{missing}");
        let anonymous = status_of(&guarded, "GET", "/api/chat/sess-md/markdown", "");
        assert!(anonymous.contains(" 403 "), "{anonymous}");
    }

    #[test]
    fn a_bearer_header_is_the_same_check_as_the_query() {
        let guarded = shared("full", "view");
        let bearer = |token: &str| format!("Authorization: Bearer {token}\r\n");
        assert!(status_of(&guarded, "GET", "/api/hosts", &bearer("full")).contains(" 200 "));
        assert!(status_of(&guarded, "GET", "/api/hosts", &bearer("view")).contains(" 200 "));
        assert!(status_of(&guarded, "GET", "/api/hosts", &bearer("wrong")).contains(" 403 "));
        assert!(status_of(&guarded, "GET", "/api/hosts", "").contains(" 403 "));
        // Same scopes whichever way the token arrives: read-only cannot act.
        let act = status_of(
            &guarded,
            "POST",
            "/api/launch",
            &format!(
                "{}Content-Type: application/json\r\nContent-Length: 2\r\n\r\n{{}}",
                bearer("view")
            ),
        );
        assert!(act.contains(" 403 "), "{act}");
    }

    #[test]
    fn metrics_answers_without_a_token_and_nothing_else_does() {
        let guarded = shared("full", "view");
        for target in ["/metrics", "/metrics?t=wrong"] {
            let status = status_of(&guarded, "GET", target, "");
            assert!(status.contains(" 200 "), "{target}: {status}");
        }
        let wrong = status_of(
            &guarded,
            "GET",
            "/metrics",
            "Authorization: Bearer wrong\r\n",
        );
        assert!(wrong.contains(" 200 "), "{wrong}");
        // The bypass is that one path: its neighbours keep the gate.
        for target in ["/api/sessions", "/metrics/", "/metricsx", "/"] {
            let status = status_of(&guarded, "GET", target, "");
            assert!(status.contains(" 403 "), "{target}: {status}");
        }
    }

    fn rows(pairs: &[(&str, &str)]) -> Vec<(String, String)> {
        pairs
            .iter()
            .map(|(id, row)| (id.to_string(), row.to_string()))
            .collect()
    }

    #[test]
    fn a_delta_carries_only_the_rows_that_moved() {
        let before = rows(&[("a", r#"{"id":"a","n":1}"#), ("b", r#"{"id":"b","n":1}"#)]);

        // The same table again is nothing to send.
        assert_eq!(table_delta(&before, &before.clone()), None);

        // One row changed in place: that row, and no order, since none moved.
        let changed = rows(&[("a", r#"{"id":"a","n":2}"#), ("b", r#"{"id":"b","n":1}"#)]);
        assert_eq!(
            table_delta(&before, &changed).as_deref(),
            Some(r#"{"set":[{"id":"a","n":2}]}"#)
        );

        // A row that appeared is sent, and the order says where it goes.
        let grown = rows(&[
            ("c", r#"{"id":"c"}"#),
            ("a", r#"{"id":"a","n":1}"#),
            ("b", r#"{"id":"b","n":1}"#),
        ]);
        assert_eq!(
            table_delta(&before, &grown).as_deref(),
            Some(r#"{"set":[{"id":"c"}],"order":["c","a","b"]}"#)
        );

        // A row that went away is said by the order alone.
        let shrunk = rows(&[("b", r#"{"id":"b","n":1}"#)]);
        assert_eq!(
            table_delta(&before, &shrunk).as_deref(),
            Some(r#"{"set":[],"order":["b"]}"#)
        );

        // And rows that only swapped places are an order and nothing else.
        let swapped = rows(&[("b", r#"{"id":"b","n":1}"#), ("a", r#"{"id":"a","n":1}"#)]);
        assert_eq!(
            table_delta(&before, &swapped).as_deref(),
            Some(r#"{"set":[],"order":["b","a"]}"#)
        );
    }

    /// Put a table in front of every stream, the way `publish` does.
    fn put(shared: &Shared, version: u64, table: &[(&str, &str)]) {
        let rows = rows(table);
        let json = format!(
            "[{}]",
            rows.iter()
                .map(|(_, r)| r.as_str())
                .collect::<Vec<_>>()
                .join(",")
        );
        *shared.latest.lock().unwrap() = Arc::new(Snapshot {
            version,
            json,
            rows,
            sessions: Vec::new(),
            host_errors: Vec::new(),
        });
        shared.updated.notify_all();
    }

    /// Open `/api/events<query>` against `shared` on a thread of its own, and
    /// hand back a reader of its frames as `(event, data)`.
    fn open_stream(
        shared: &Arc<Shared>,
        query: &str,
        keepalive: Duration,
    ) -> impl FnMut() -> (String, String) {
        open_stream_with(shared, query, "", keepalive).1
    }

    /// [`open_stream`] with request headers, handing back the response head
    /// too. A gzip stream is decoded as it arrives, the way a browser does.
    fn open_stream_with(
        shared: &Arc<Shared>,
        query: &str,
        headers: &str,
        keepalive: Duration,
    ) -> (String, impl FnMut() -> (String, String)) {
        use std::io::{BufRead, BufReader};
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let mut client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
        let (mut server, _) = listener.accept().unwrap();
        client
            .write_all(format!("GET /api/events{query} HTTP/1.1\r\n{headers}\r\n").as_bytes())
            .unwrap();
        let request = Request::parse(&server).unwrap();
        let shared = Arc::clone(shared);
        std::thread::spawn(move || events_every(&shared, &mut server, &request, keepalive));
        client
            .set_read_timeout(Some(Duration::from_secs(10)))
            .unwrap();
        let mut reader = BufReader::new(client);
        // The response head, which ends at its blank line.
        let mut head = String::new();
        loop {
            let mut line = String::new();
            reader.read_line(&mut line).expect("the response head");
            if line == "\r\n" {
                break;
            }
            head.push_str(&line);
        }
        let mut reader: Box<dyn BufRead> = match head.contains("Content-Encoding: gzip") {
            true => Box::new(BufReader::new(flate2::read::GzDecoder::new(reader))),
            false => Box::new(reader),
        };
        let next = move || {
            let (mut event, mut data) = (String::new(), String::new());
            loop {
                let mut line = String::new();
                reader
                    .read_line(&mut line)
                    .expect("a frame before the timeout");
                let line = line.trim_end_matches('\n');
                if line.is_empty() {
                    return (event, data);
                }
                if let Some(e) = line.strip_prefix("event: ") {
                    event = e.to_string();
                } else if let Some(d) = line.strip_prefix("data: ") {
                    data = d.to_string();
                }
            }
        };
        (head, next)
    }

    /// The table goes out whole once, then as the rows that moved — and a
    /// refresh that moved nothing writes nothing.
    #[test]
    fn the_table_stream_sends_the_table_once_and_then_what_changed() {
        let shared = Arc::new(shared("", ""));
        put(&shared, 1, &[("a", r#"{"a":1}"#), ("b", r#"{"b":1}"#)]);
        let mut next = open_stream(&shared, "", Duration::from_secs(30));

        assert_eq!(next(), ("sessions".into(), r#"[{"a":1},{"b":1}]"#.into()));
        // Unchanged, then changed: the first is never written, so the next
        // frame is the second's one row.
        put(&shared, 2, &[("a", r#"{"a":1}"#), ("b", r#"{"b":1}"#)]);
        put(&shared, 3, &[("a", r#"{"a":1}"#), ("b", r#"{"b":2}"#)]);
        assert_eq!(next(), ("rows".into(), r#"{"set":[{"b":2}]}"#.into()));
    }

    /// The bug a comment keepalive hid: refreshes land every two seconds, and
    /// a stream whose own row never moved woke on each one, wrote nothing,
    /// and so never reached its keepalive at all — which the page could not
    /// tell from a dead connection. A ping is due by the clock of the last
    /// write, however many snapshots arrive in between.
    #[test]
    fn a_quiet_stream_pings_even_while_refreshes_keep_landing() {
        let shared = Arc::new(shared("", ""));
        put(&shared, 1, &[("abc", r#"{"s":"asking"}"#)]);
        let mut next = open_stream(&shared, "?session=abc", Duration::from_millis(300));
        assert_eq!(next(), ("sessions".into(), r#"[{"s":"asking"}]"#.into()));

        let churn = {
            let shared = Arc::clone(&shared);
            std::thread::spawn(move || {
                for version in 2..60 {
                    put(&shared, version, &[("abc", r#"{"s":"asking"}"#)]);
                    std::thread::sleep(Duration::from_millis(50));
                }
            })
        };
        assert_eq!(next(), ("ping".into(), "1".into()));
        churn.join().unwrap();
    }

    #[test]
    fn a_patch_carries_only_the_fields_that_moved() {
        let row = |id: &str, cpu: u32, extra: &str| {
            format!(
                r#"{{"session_id":"{id}","title":"a long title that is the bulk of the row","cpu":{cpu}{extra}}}"#
            )
        };
        let before = vec![
            ("a".to_string(), row("a", 1, r#","permission":"edits""#)),
            ("b".to_string(), row("b", 1, "")),
        ];
        assert_eq!(table_patch(&before, &before.clone()), None);

        // One field moved and one went: the row is not resent, only those.
        let after = vec![
            ("a".to_string(), row("a", 2, "")),
            ("b".to_string(), row("b", 1, "")),
        ];
        assert_eq!(
            table_patch(&before, &after).as_deref(),
            Some(r#"{"set":[],"patch":[{"id":"a","to":{"cpu":2},"drop":["permission"]}]}"#)
        );

        // A new row goes whole, and the order as the ids to put first.
        let grown = vec![
            ("c".to_string(), row("c", 1, "")),
            ("a".to_string(), before[0].1.clone()),
            ("b".to_string(), before[1].1.clone()),
        ];
        let delta = table_patch(&before, &grown).unwrap();
        assert!(
            delta.starts_with(&format!(r#"{{"set":[{}]"#, row("c", 1, ""))),
            "{delta}"
        );
        assert!(delta.ends_with(r#""head":["c"],"gone":[]}"#), "{delta}");
    }

    /// The order edit is checked by doing what the page does with it — `head`,
    /// then every row it had that is neither in `head` nor gone — and
    /// comparing with the order wanted.
    #[test]
    fn an_order_edit_rebuilds_the_order_it_describes() {
        let ids = |s: &str| -> Vec<(String, String)> {
            s.chars().map(|c| (c.to_string(), String::new())).collect()
        };
        let apply = |before: &str, head: &[&str], gone: &[&str]| -> String {
            let mut out: String = head.concat();
            out.extend(
                before
                    .chars()
                    .filter(|c| !head.contains(&c.to_string().as_str()))
                    .filter(|c| !gone.contains(&c.to_string().as_str())),
            );
            out
        };
        for (before, after) in [
            ("abcdefgh", "dabcefgh"),  // one row became the newest
            ("abcdefgh", "xabcdefgh"), // a new row arrived on top
            ("abcdefgh", "abcdefg"),   // the oldest went away
            ("abcdefgh", "abdefgh"),   // one in the middle went away
            ("abcdefgh", "hgabcdef"),  // two moved up together
            ("abcdefgh", "fabcdegh"),  // one moved up, one went
        ] {
            let after_ids: Vec<String> = after.chars().map(String::from).collect();
            let after_refs: Vec<&str> = after_ids.iter().map(String::as_str).collect();
            let before_ids = ids(before);
            let (head, gone) = order_edit(&before_ids, &after_refs)
                .unwrap_or_else(|| panic!("{before} -> {after} should be an edit"));
            assert_eq!(apply(before, &head, &gone), after, "{before} -> {after}");
            assert!(head.len() <= 2, "{before} -> {after}: head {head:?}");
        }
        // Turned around entirely, nearly every id has to be named, and the
        // whole order is no longer than that plus the departures.
        let after_ids: Vec<String> = "hgf".chars().map(String::from).collect();
        let after_refs: Vec<&str> = after_ids.iter().map(String::as_str).collect();
        assert_eq!(order_edit(&ids("abcdefgh"), &after_refs), None);
        // And a repeated id has no place to keep.
        assert_eq!(order_edit(&ids("aab"), &["b", "a", "a"]), None);
    }

    /// A page that did not ask for patches is sent whole rows, which is all a
    /// page loaded before the patch shape existed knows how to apply.
    #[test]
    fn only_a_stream_that_asks_is_sent_patches() {
        let shared = Arc::new(shared("", ""));
        let row = |n: u32| {
            format!(
                r#"{{"session_id":"a","title":"{}","n":{n}}}"#,
                "x".repeat(200)
            )
        };
        let first = row(1);
        put(&shared, 1, &[("a", &first)]);
        let mut old = open_stream(&shared, "", Duration::from_secs(30));
        let mut new = open_stream(&shared, "?rows=patch", Duration::from_secs(30));
        assert_eq!(old().0, "sessions");
        assert_eq!(new().0, "sessions");
        let second = row(2);
        put(&shared, 2, &[("a", &second)]);
        assert_eq!(old(), ("rows".into(), format!(r#"{{"set":[{second}]}}"#)));
        assert_eq!(
            new(),
            (
                "rows".into(),
                r#"{"set":[],"patch":[{"id":"a","to":{"n":2}}]}"#.into()
            )
        );
    }

    /// Gzip to a client that offers it, and each event decodable the moment
    /// it is written: a stream that only came out of the compressor when the
    /// next event pushed it would hold every update back by one refresh.
    #[test]
    fn the_event_stream_is_gzip_and_flushed_per_event() {
        let shared = Arc::new(shared("", ""));
        put(&shared, 1, &[("a", r#"{"a":1}"#)]);
        let (head, mut next) = open_stream_with(
            &shared,
            "",
            "Accept-Encoding: gzip, deflate, br\r\n",
            Duration::from_millis(300),
        );
        assert!(head.contains("Content-Encoding: gzip"), "{head}");
        assert_eq!(next(), ("sessions".into(), r#"[{"a":1}]"#.into()));
        put(&shared, 2, &[("a", r#"{"a":2}"#)]);
        assert_eq!(next(), ("rows".into(), r#"{"set":[{"a":2}]}"#.into()));
        // The ping goes through the same encoder.
        assert_eq!(next(), ("ping".into(), "1".into()));

        let (plain, _) = open_stream_with(&shared, "", "", Duration::from_secs(30));
        assert!(!plain.contains("Content-Encoding"), "{plain}");
    }

    /// One request through the router, as the head and the undecoded body.
    fn exchange(shared: &Shared, target: &str, headers: &str) -> (String, Vec<u8>) {
        use std::io::Read;
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let mut client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
        let (mut server, _) = listener.accept().unwrap();
        client
            .write_all(format!("GET {target} HTTP/1.1\r\n{headers}\r\n").as_bytes())
            .unwrap();
        // Answered on a thread of its own: the app page is most of a megabyte,
        // more than a socket buffers, and its write would wait on this read.
        let mut raw = Vec::new();
        std::thread::scope(|scope| {
            scope.spawn(move || serve_connection(shared, &mut server));
            client.read_to_end(&mut raw).unwrap();
        });
        let split = raw
            .windows(4)
            .position(|w| w == b"\r\n\r\n")
            .expect("a response head");
        let head = String::from_utf8(raw[..split].to_vec()).unwrap();
        (head, raw[split + 4..].to_vec())
    }

    /// The value of header `name` in a response head.
    fn header<'a>(head: &'a str, name: &str) -> Option<&'a str> {
        head.lines().find_map(|line| {
            let (k, v) = line.split_once(':')?;
            k.eq_ignore_ascii_case(name).then_some(v.trim())
        })
    }

    const JS_PATH: &str = "/_astro/index.astro_astro_type_script_index_0_lang.CdYCpIGm.js";

    #[test]
    fn hashed_files_are_kept_for_a_year_and_pages_are_revalidated() {
        let guarded = shared("full", "view");
        let auth = "Authorization: Bearer full\r\n";
        for (path, cache) in [
            (JS_PATH, http::IMMUTABLE),
            ("/_astro/index.D2bSaP4U.css", http::IMMUTABLE),
            (
                "/_astro/rmux_web_crypto_wasm_bg.C8R0tHIf.wasm",
                http::IMMUTABLE,
            ),
            ("/term/crabs/lime-dark.svg", http::IMMUTABLE),
            ("/term/", http::REVALIDATE),
            ("/", http::REVALIDATE),
            ("/favicon.svg", http::REVALIDATE),
        ] {
            let (head, _) = exchange(&guarded, path, auth);
            assert!(head.starts_with("HTTP/1.1 200 "), "{path}: {head}");
            assert_eq!(header(&head, "Cache-Control"), Some(cache), "{path}");
            if cache == http::REVALIDATE {
                assert!(header(&head, "ETag").is_some(), "{path} has no ETag");
            }
        }
        // Behind the gate like everything else: caching is not a way round it.
        let (head, _) = exchange(&guarded, JS_PATH, "");
        assert!(head.starts_with("HTTP/1.1 403 "), "{head}");
        // And what holds a transcript is still never kept.
        let (head, _) = exchange(&guarded, "/api/sessions", auth);
        assert_eq!(header(&head, "Cache-Control"), Some("no-store"));
    }

    #[test]
    fn a_page_the_browser_holds_is_answered_304_and_no_body() {
        let guarded = shared("full", "view");
        for path in ["/?t=full", "/term/", JS_PATH] {
            let (head, body) = exchange(&guarded, path, "Authorization: Bearer full\r\n");
            let etag = header(&head, "ETag").expect("an ETag").to_string();
            assert!(!body.is_empty());
            let (head, body) = exchange(
                &guarded,
                path,
                &format!("Authorization: Bearer full\r\nIf-None-Match: {etag}\r\n"),
            );
            assert!(head.starts_with("HTTP/1.1 304 "), "{path}: {head}");
            assert!(body.is_empty(), "{path}: a 304 carries no body");
            assert_eq!(header(&head, "ETag"), Some(etag.as_str()));
            // A stale tag is answered in full.
            let (head, _) = exchange(
                &guarded,
                path,
                "Authorization: Bearer full\r\nIf-None-Match: W/\"stale\"\r\n",
            );
            assert!(head.starts_with("HTTP/1.1 200 "), "{path}: {head}");
        }
        // The page carries the credential, so each credential's page has its
        // own tag: the read-only link's never validates the full one's.
        let tag = |token: &str| {
            let (head, _) = exchange(&guarded, &format!("/?t={token}"), "");
            header(&head, "ETag").unwrap().to_string()
        };
        assert_ne!(tag("full"), tag("view"));
        // A 304 on a link's first visit still hands its cookie back.
        let full = tag("full");
        let (head, _) = exchange(&guarded, "/?t=full", &format!("If-None-Match: {full}\r\n"));
        assert!(head.starts_with("HTTP/1.1 304 "), "{head}");
        assert!(header(&head, "Set-Cookie").is_some(), "{head}");
    }

    #[test]
    fn bodies_are_compressed_for_a_client_that_offers_it() {
        use std::io::Read;
        let guarded = shared("full", "view");
        let auth = "Authorization: Bearer full\r\n";
        let js = term::file(JS_PATH).unwrap();
        let raw = exchange(&guarded, JS_PATH, auth);
        assert_eq!(header(&raw.0, "Content-Encoding"), None);
        assert_eq!(header(&raw.0, "Vary"), Some("Accept-Encoding"));

        let (head, body) = exchange(
            &guarded,
            JS_PATH,
            &format!("{auth}Accept-Encoding: gzip\r\n"),
        );
        assert_eq!(header(&head, "Content-Encoding"), Some("gzip"));
        assert_eq!(
            header(&head, "Content-Length"),
            Some(body.len().to_string().as_str())
        );
        let mut plain = Vec::new();
        flate2::read::GzDecoder::new(&body[..])
            .read_to_end(&mut plain)
            .unwrap();
        assert_eq!(plain, raw.1);
        assert!(
            body.len() < raw.1.len() / 3,
            "{} of {}",
            body.len(),
            raw.1.len()
        );

        let (head, body) = exchange(
            &guarded,
            JS_PATH,
            &format!("{auth}Accept-Encoding: gzip, deflate, br\r\n"),
        );
        assert_eq!(header(&head, "Content-Encoding"), Some("br"));
        let mut plain = Vec::new();
        brotli::Decompressor::new(&body[..], 4096)
            .read_to_end(&mut plain)
            .unwrap();
        assert_eq!(plain, raw.1);
        assert_eq!(js.content_type, "application/javascript; charset=utf-8");

        // A route's JSON, built per request, is compressed too.
        let table: Vec<(String, String)> = (0..40)
            .map(|i| {
                (
                    format!("s{i}"),
                    format!(r#"{{"session_id":"s{i}","title":"a session"}}"#),
                )
            })
            .collect();
        let pairs: Vec<(&str, &str)> = table
            .iter()
            .map(|(a, b)| (a.as_str(), b.as_str()))
            .collect();
        put(&guarded, 1, &pairs);
        let (head, _) = exchange(
            &guarded,
            "/api/sessions",
            &format!("{auth}Accept-Encoding: gzip\r\n"),
        );
        assert_eq!(header(&head, "Content-Encoding"), Some("gzip"), "{head}");
    }
}