semlith 0.44.0

//! A synchronous HTTP/1.1 server, just large enough to serve the portal.
//!
//! Nothing here is a general-purpose web server and nothing here should become
//! one. It answers requests from one browser on one machine over the loopback
//! interface, which is why it is a few hundred lines of `std::net` rather than
//! a dependency: every HTTP crate that does chunked responses and a thread pool
//! brings a tree semlith would then be shipping to every user of a local search
//! tool, and the parts of HTTP this needs are the parts that fit on a page.
//!
//! Three rules are enforced here, before any handler sees a request, because
//! they are the product's privacy claim and a claim enforced in one place is
//! one that can be read in one place:
//!
//! - **Loopback only.** [`Server::bind`] binds `127.0.0.1` and there is no flag
//!   to change it.
//! - **A per-run token, carried in a request header.** Generated at start,
//!   printed once in the URL, handed to the page from that URL, and sent back
//!   as `Semlith-Token` on every API request. A request without it gets 401 and
//!   an empty body — not a login page, not an error object, nothing that tells a
//!   prober what is here.
//!
//!   It was a cookie until 0.14.0, and a cookie is the wrong container for it:
//!   `localhost` is one site however many ports are listening on it, so
//!   `SameSite=Strict` never separated this server from a page served by
//!   anything else on 127.0.0.1. A header the browser will not attach
//!   cross-origin does separate them. The cost is that a browser attaches no
//!   header to a stylesheet, a font, a favicon or an `<img src>`, so the page
//!   and its own static assets are served without a credential — they are the
//!   same bytes in every copy of the binary — and everything that answers about
//!   this machine is not.
//!
//! - **A write has to come from this origin.** Every request that is not a GET
//!   or a HEAD must carry `Sec-Fetch-Site: same-origin`, or — for a client that
//!   sends no fetch metadata at all, which is every client that is not a
//!   browser — an `Origin` matching this server or no `Origin`. It must also
//!   carry a JSON content type, because a body a browser can send with no
//!   preflight is a body this server will not read. Both are checked before the
//!   token, so a cross-origin page learns nothing from the difference between a
//!   right and a wrong guess.
//! - **A persisted agent key, and only for `/mcp`.** From 0.13.0 the daemon
//!   also answers MCP over HTTP, and that endpoint needs a credential a client
//!   can keep in a configuration file — which the session token can never be,
//!   because it changes every run. The two have deliberately separate
//!   lifetimes and separate reach: the agent key opens `/mcp` and nothing else,
//!   so a key in a config file can never reach the rotate, adopt or upgrade
//!   routes, and rotating the session token does not disconnect an agent.
//! - **A `Host` header that is localhost.** A page on another origin cannot
//!   read a cross-origin response, but it can *send* the request, and a DNS
//!   name that resolves to 127.0.0.1 would otherwise reach this server with the
//!   browser's cookies attached. Only `localhost` and `127.0.0.1`, with an
//!   optional port, are answered.
//!
//! Every response also carries a Content-Security-Policy that allows only
//! `'self'`, and no CORS header is ever emitted — there is no origin this
//! server wants to be readable from.

use anyhow::{Context, Result};
use std::collections::BTreeMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;

/// The port `semlith start` listens on unless told otherwise.
///
/// 7365 spells SEML on a phone keypad, is registered to nothing in the IANA
/// list, sits below every OS ephemeral range, and is nowhere near the ports
/// local development already uses — 3000, 4200, 5000, 5173, 8000, 8080, 8888,
/// 9000, and the usual database and model-server ports. A stable port is what
/// lets the URL be a bookmark.
pub const DEFAULT_PORT: u16 = 7365;

/// Overrides [`DEFAULT_PORT`].
pub const PORT_ENV: &str = "SEMLITH_PORT";

/// The header the session token travels in.
///
/// A custom header is the point rather than a detail: a browser will not attach
/// one to a request a page on another origin made, and it cannot set one at all
/// on an `<img>`, a form or a stylesheet. That is the whole separation a cookie
/// could not give, because every port on `localhost` is the same site.
pub const TOKEN_HEADER: &str = "Semlith-Token";

/// [`TOKEN_HEADER`] as the map holds it. Header names are lowercased on the way
/// in so a lookup never depends on a client's casing.
const TOKEN_HEADER_KEY: &str = "semlith-token";

/// The one path the agent key opens.
pub const MCP_PATH: &str = "/mcp";

/// Threads answering requests.
///
/// A browser opens up to six connections to one origin, and this server closes
/// each connection after its response, so eight is enough for one portal with
/// a long-running index stream on one of them and nothing queueing behind it.
const WORKERS: usize = 8;

/// The largest request line and header block accepted. A browser's is under
/// 2 KiB; anything near this is not a browser.
///
/// Counted as the bytes arrive rather than once a line is complete, so it is
/// also the cap on any one line: a header that never sends its newline is
/// refused at 16 KiB instead of growing a `String` at loopback speed.
const MAX_HEADERS: usize = 16 * 1024;

/// How many idle connections — open, and silent for [`IDLE_GRACE`] — one
/// address may keep. A ninth closes the oldest.
///
/// A browser opens up to six spare sockets per origin ahead of the requests it
/// puts on them, and re-opens one that was closed, so eight is room for a
/// portal tab and a client beside it. Every local process connects from the
/// same loopback address, so on this server it is in effect the cap on idle
/// connections overall — which is the point: before 0.40.0 an idle socket
/// counted against [`MAX_CONNECTIONS`] for 30 s, and 32 of them were a 503 for
/// everyone (S11).
const IDLE_PER_PEER: usize = 8;

/// How long a connection that has sent nothing is safe from
/// [`IDLE_PER_PEER`]. A request's bytes land with its connect on loopback, but
/// the thread that reads them can wait for a core, and closing a connection
/// before its thread has looked refuses a real request in a burst of them --
/// twenty at once in `tests/daemon.rs` did exactly that with no grace.
const IDLE_GRACE: Duration = Duration::from_millis(250);

/// The most connections waiting for their head at once, from anyone. Past it
/// the oldest that has sent nothing is closed, or failing that the oldest: a
/// flood inside [`IDLE_GRACE`] or of dribbling senders is still bounded. Each
/// costs a thread and two descriptors, and a daemon run as a login service may
/// have 256 descriptors in all, so this stays well under that.
const MAX_WAITING: usize = 48;

/// How often a connection still waiting for its head looks up to see whether
/// it was closed to make room. Shutting a socket wakes a blocked read on
/// macOS and Linux at once; this bounds how long an evicted connection's
/// thread can outlive its slot anywhere that does not.
const HEAD_POLL: Duration = Duration::from_secs(1);

/// The largest request body accepted. The only bodies this server takes are
/// small JSON objects naming paths.
const MAX_BODY: usize = 1024 * 1024;

/// How long a connection may sit idle mid-request before it is dropped, so a
/// half-open socket cannot hold a worker forever.
const IO_TIMEOUT: Duration = Duration::from_secs(30);

/// How long a whole request has to arrive once its connection is accepted.
///
/// The read timeout above catches a socket that goes quiet; this catches one
/// that never does — a byte a second is not a stalled connection and would
/// hold a worker for as long as the sender felt like. Ten seconds is far more
/// than any request over loopback needs. It is checked before every read of
/// the head as well as after the body: until 0.40.0 the head was read with no
/// deadline at all, before the token was looked at.
const REQUEST_DEADLINE: Duration = Duration::from_secs(10);

/// How many connections may be in hand at once.
///
/// Eight workers, a waiting room that can be holding refusals for up to four
/// seconds, and streams that have left the pool: the cap is on all of them
/// together, because what it is protecting against is memory and file
/// descriptors rather than worker time. Past it the answer is 503 and the
/// connection closes, which is what a local client should see from a daemon
/// that is already at its limit. A connection counts once its request head has
/// arrived; before that it is bounded by [`MAX_WAITING`] instead, so the two
/// together are the most this server holds.
const MAX_CONNECTIONS: usize = 32;

/// Set when the accept loop should be woken so it can notice `stop`.
static WAKE: AtomicBool = AtomicBool::new(false);

/// Why a request was refused, for the one-line-per-class count the daemon logs.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Refusal {
    /// No token, or the wrong one.
    Unauthorized,
    /// A `Host` header that is not loopback.
    ForeignHost,
    /// Malformed, oversized, or a method this server does not answer.
    Malformed,
    /// A write that did not come from this origin.
    CrossOrigin,
    /// A write whose body was not offered as JSON.
    BadContentType,
    /// An index or a fetch for a path outside the store's boundary.
    OutsideBoundary,
    /// An index for a path the deny-list names.
    DeniedPath,
    /// A fetch for an address that is not on the public internet.
    PrivateAddress,
}

impl Refusal {
    pub fn as_str(self) -> &'static str {
        match self {
            Refusal::Unauthorized => "unauthorized",
            Refusal::ForeignHost => "foreign host",
            Refusal::Malformed => "malformed",
            Refusal::CrossOrigin => "cross-origin",
            Refusal::BadContentType => "bad-content-type",
            Refusal::OutsideBoundary => "outside-boundary",
            Refusal::DeniedPath => "denied-path",
            Refusal::PrivateAddress => "private-address",
        }
    }
}

/// One parsed request.
#[derive(Clone)]
pub struct Request {
    pub method: String,
    /// Path with the query string removed, percent-decoded.
    pub path: String,
    pub query: BTreeMap<String, String>,
    /// Header names lowercased, so a lookup never depends on a client's casing.
    pub headers: BTreeMap<String, String>,
    pub body: Vec<u8>,
}

impl Request {
    pub fn query(&self, key: &str) -> Option<&str> {
        self.query.get(key).map(String::as_str)
    }

    /// Every value a repeated query key has, for `?lang=rust&lang=go`.
    ///
    /// The map holds them joined by `\u{1}`, which cannot appear in a
    /// percent-decoded query value that a browser produced.
    pub fn query_all(&self, key: &str) -> Vec<String> {
        match self.query.get(key) {
            Some(v) => v.split('\u{1}').map(str::to_string).collect(),
            None => Vec::new(),
        }
    }

    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(name).map(String::as_str)
    }

    /// The body as JSON, and the only place a body is read.
    ///
    /// The content type is checked here as well as in [`answer`], so a route
    /// that grew a second body reader would still be refusing the one thing a
    /// cross-origin form can send without a preflight.
    pub fn json(&self) -> Result<serde_json::Value> {
        anyhow::ensure!(
            json_body(self.header("content-type")),
            "this route reads JSON; send Content-Type: application/json"
        );
        Ok(serde_json::from_slice(&self.body)?)
    }
}

/// Whether a `Content-Type` offers a JSON body.
///
/// `application/json` and nothing else. The three types a browser will post
/// from a form without asking anybody's permission first — `text/plain`,
/// `application/x-www-form-urlencoded`, `multipart/form-data` — are exactly the
/// ones this refuses.
fn json_body(content_type: Option<&str>) -> bool {
    content_type
        .map(|value| value.split(';').next().unwrap_or("").trim())
        .is_some_and(|media| media.eq_ignore_ascii_case("application/json"))
}

/// What a handler writes a streaming body through.
///
/// Each [`Chunks::send`] is one HTTP chunk, flushed immediately, which is what
/// makes an index run's progress visible while it is still running rather than
/// all at once when it ends.
pub struct Chunks<'a> {
    out: &'a mut dyn Write,
}

impl Chunks<'_> {
    pub fn send(&mut self, line: &str) -> std::io::Result<()> {
        write!(self.out, "{:x}\r\n", line.len())?;
        self.out.write_all(line.as_bytes())?;
        self.out.write_all(b"\r\n")?;
        self.out.flush()
    }
}

type Streamer = Box<dyn FnOnce(&mut Chunks) -> std::io::Result<()> + Send>;

enum Body {
    Bytes(Vec<u8>),
    /// Sent with `Transfer-Encoding: chunked`, for a response whose length is
    /// not known when it starts.
    Stream(Streamer),
}

pub struct Response {
    status: u16,
    content_type: &'static str,
    extra: Vec<(String, String)>,
    /// A Content-Security-Policy in place of [`PAGE_CSP`]. Only ever a
    /// narrower one: the report preview, which allows no script at all.
    csp: Option<&'static str>,
    body: Body,
}

/// The policy every response carries unless it names a narrower one.
///
/// Everything the page loads is served by this server, nothing may frame it,
/// and no form may post anywhere else. There is no origin this server wants to
/// be readable from, so there is no CORS header here or anywhere.
const PAGE_CSP: &str = "default-src 'self'; \
     img-src 'self' data:; \
     style-src 'self'; \
     script-src 'self'; \
     font-src 'self'; \
     connect-src 'self'; \
     form-action 'none'; \
     frame-ancestors 'none'; \
     base-uri 'none'";

impl Response {
    pub fn new(status: u16, content_type: &'static str, body: Vec<u8>) -> Self {
        Self {
            status,
            content_type,
            extra: Vec::new(),
            csp: None,
            body: Body::Bytes(body),
        }
    }

    pub fn json(value: &serde_json::Value) -> Self {
        Self::new(
            200,
            "application/json; charset=utf-8",
            serde_json::to_vec(value).unwrap_or_else(|_| b"{}".to_vec()),
        )
    }

    /// The status this response will be sent with.
    pub fn status(&self) -> u16 {
        self.status
    }

    /// The body as JSON, when it is a whole body that is JSON. A streamed
    /// response has no body to read here.
    pub fn json_value(&self) -> Option<serde_json::Value> {
        match &self.body {
            Body::Bytes(bytes) => serde_json::from_slice(bytes).ok(),
            Body::Stream(_) => None,
        }
    }

    /// A refusal a handler decided on: still JSON, because the portal shows it.
    pub fn error(status: u16, message: &str) -> Self {
        Self::new(
            status,
            "application/json; charset=utf-8",
            serde_json::to_vec(&serde_json::json!({ "error": message }))
                .unwrap_or_else(|_| b"{}".to_vec()),
        )
    }

    pub fn asset(content_type: &'static str, bytes: &'static [u8]) -> Self {
        Self::new(200, content_type, bytes.to_vec())
    }

    /// A response whose body is produced as it goes.
    pub fn stream(f: impl FnOnce(&mut Chunks) -> std::io::Result<()> + Send + 'static) -> Self {
        Self {
            status: 200,
            content_type: "application/x-ndjson; charset=utf-8",
            extra: Vec::new(),
            csp: None,
            body: Body::Stream(Box::new(f)),
        }
    }

    pub fn header(mut self, name: &str, value: impl Into<String>) -> Self {
        self.extra.push((name.to_string(), value.into()));
        self
    }

    /// Serve under `csp` instead of the page's policy. A second
    /// Content-Security-Policy header could only narrow the first, so a
    /// document with a policy of its own replaces it rather than adding one.
    pub fn with_csp(mut self, csp: &'static str) -> Self {
        self.csp = Some(csp);
        self
    }

    /// The policy this response will be sent under.
    pub fn csp(&self) -> &'static str {
        self.csp.unwrap_or(PAGE_CSP)
    }
}

/// A handler: everything the daemon serves, behind one function.
pub type Handler = Arc<dyn Fn(&Request) -> Response + Send + Sync>;

/// A refused connection and the moment it may be told so.
struct Late {
    stream: TcpStream,
    due: std::time::Instant,
    /// Held for as long as the refusal is, so a connection waiting in the
    /// waiting room still counts against [`MAX_CONNECTIONS`].
    _in_flight: InFlight,
}

/// One connection's place in the count, given up when it is dropped.
///
/// A guard rather than a pair of increments, because this connection can leave
/// by several doors — answered on a worker, held in the waiting room, or
/// streaming on a thread of its own — and a count that is only right on the
/// paths somebody remembered is a leak waiting for the one they did not.
/// A connection the server is holding, and the priority it lifts while held.
///
/// The daemon sits in background state while it is idle (see
/// `crate::priority`), and on a busy machine a background thread can wait
/// seconds for a core. Measured on the reference M1 with three builds running:
/// a request made just after the daemon dropped back waited past a 30-second
/// client timeout. A request is work somebody is waiting on, so the process
/// is lifted from the moment the connection is accepted until it closes.
struct InFlight(
    Arc<std::sync::atomic::AtomicUsize>,
    #[allow(dead_code)] crate::priority::Embedding,
);

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

impl InFlight {
    /// A place in `count`, lifting the process's priority while it is held.
    fn taken(count: &Arc<std::sync::atomic::AtomicUsize>) -> Self {
        count.fetch_add(1, Ordering::Relaxed);
        Self(Arc::clone(count), crate::priority::request())
    }

    /// Move this place to `count`; the priority lift comes along.
    fn move_to(&mut self, count: &Arc<std::sync::atomic::AtomicUsize>) {
        count.fetch_add(1, Ordering::Relaxed);
        std::mem::replace(&mut self.0, Arc::clone(count)).fetch_sub(1, Ordering::Relaxed);
    }
}

/// A connection whose request head has arrived whole, on its way to a worker.
struct Ready {
    stream: TcpStream,
    /// Holds whatever of the body arrived with the head.
    reader: BufReader<TcpStream>,
    request: Request,
    in_flight: InFlight,
    accepted: std::time::Instant,
}

/// What a waiting connection's thread and [`Pending`] both see: whether any
/// byte has arrived, and whether the connection was closed to make room.
#[derive(Default)]
struct HeadState {
    heard: AtomicBool,
    evicted: AtomicBool,
}

/// A connection still waiting for its head.
struct Waiting {
    state: Arc<HeadState>,
    peer: std::net::IpAddr,
    since: std::time::Instant,
    /// A handle to close it by while its thread is blocked reading.
    handle: TcpStream,
    /// Its place in the count. Held here rather than on its thread, so closing
    /// it frees the place at once whether or not the platform wakes the read
    /// its thread is blocked in.
    slot: InFlight,
}

/// The connections that have not sent a whole request head yet, oldest first.
///
/// They hold no worker — the head is read on the connection's own thread —
/// and no place under [`MAX_CONNECTIONS`], which idle sockets used to fill. What
/// they cost is a thread and two descriptors each, bounded by
/// [`IDLE_PER_PEER`] and [`MAX_WAITING`].
#[derive(Default)]
struct Pending(std::collections::VecDeque<Waiting>);

impl Pending {
    /// Hold a connection as waiting, then close whatever that leaves over the
    /// limits: idle connections past [`IDLE_PER_PEER`] for its address, then
    /// anything past [`MAX_WAITING`]. Always the oldest, because the newest is
    /// the one most likely to be a request about to arrive.
    fn admit(
        &mut self,
        peer: std::net::IpAddr,
        handle: TcpStream,
        slot: InFlight,
    ) -> Arc<HeadState> {
        let state = Arc::new(HeadState::default());
        self.0.push_back(Waiting {
            state: Arc::clone(&state),
            peer,
            since: std::time::Instant::now(),
            handle,
            slot,
        });
        let idle = |w: &Waiting| {
            w.peer == peer
                && !w.state.heard.load(Ordering::Relaxed)
                && w.since.elapsed() >= IDLE_GRACE
        };
        while self.0.iter().filter(|w| idle(w)).count() > IDLE_PER_PEER {
            let Some(at) = self.0.iter().position(idle) else {
                break;
            };
            self.evict(at);
        }
        while self.0.len() > MAX_WAITING {
            // Never the one just admitted, unless everything else is talking.
            let older = self.0.len() - 1;
            let at = self
                .0
                .iter()
                .take(older)
                .position(|w| !w.state.heard.load(Ordering::Relaxed))
                .unwrap_or(0);
            self.evict(at);
        }
        state
    }

    fn evict(&mut self, at: usize) {
        if let Some(gone) = self.0.remove(at) {
            gone.state.evicted.store(true, Ordering::Relaxed);
            trace(&gone.handle, || {
                "closed while waiting, to make room".to_string()
            });
            let _ = gone.handle.shutdown(std::net::Shutdown::Both);
        }
    }

    /// The place of a connection whose head arrived, or nothing if it was
    /// closed to make room in the meantime.
    fn release(&mut self, state: &Arc<HeadState>) -> Option<InFlight> {
        let at = self.0.iter().position(|w| Arc::ptr_eq(&w.state, state))?;
        self.0.remove(at).map(|w| w.slot)
    }
}

/// Why a request head was refused, as the status it is answered with.
#[derive(Debug)]
struct HeadRefused(u16);

impl std::fmt::Display for HeadRefused {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "request head refused with {}", self.0)
    }
}

impl std::error::Error for HeadRefused {}

/// The agent key, and the one it replaced.
///
/// `previous` is kept valid until this process exits so that a session in
/// flight when somebody rotated finishes rather than failing mid-answer. It is
/// not persisted: a restart is where the old key stops being accepted, which
/// is also the moment every client had to be told about the new one anyway.
#[derive(Default)]
struct Agent {
    current: String,
    /// The key this one replaced, and when it stops being accepted.
    previous: Option<(String, std::time::Instant)>,
}

/// How long a rotated agent key keeps working.
///
/// It used to be "until this process exits", which on a daemon somebody leaves
/// running is not a grace period but a second live credential — a key rotated
/// because it leaked stayed valid for as long as the machine was up. Fifteen
/// minutes is long enough for a tool call in flight and a client that has to be
/// restarted, and short enough to be a window rather than a state.
const KEY_GRACE: Duration = Duration::from_secs(15 * 60);

pub struct Server {
    listener: TcpListener,
    port: u16,
    /// Behind a lock so the Privacy page's Rotate button can replace it while
    /// the server is running.
    token: Arc<Mutex<String>>,
    agent: Arc<Mutex<Agent>>,
    /// Whether `/mcp` answers. Closing it drops the route, not the daemon.
    mcp_open: Arc<AtomicBool>,
    /// What a refused request costs right now.
    penalty: Arc<Mutex<Penalty>>,
}

impl Server {
    /// Bind the loopback interface, or explain what is already on the port.
    ///
    /// Never falls back to another port. A URL that moves is not a bookmark,
    /// and from 0.11.0 it is also a client's MCP endpoint — so a taken port is
    /// an error naming the flag, not a silent reassignment.
    pub fn bind(port: u16) -> Result<Self> {
        let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
        let listener = TcpListener::bind(addr).with_context(|| {
            format!(
                "port {port} is already in use — stop whatever is listening on it, \
                 or choose another with --port or {PORT_ENV}"
            )
        })?;
        let port = listener.local_addr().map(|a| a.port()).unwrap_or(port);
        Ok(Self {
            listener,
            port,
            token: Arc::new(Mutex::new(new_token())),
            agent: Arc::new(Mutex::new(Agent::default())),
            mcp_open: Arc::new(AtomicBool::new(true)),
            penalty: Arc::new(Mutex::new(Penalty::default())),
        })
    }

    /// Install the persisted agent key. Until this is called `/mcp` accepts
    /// the session token alone, because an empty key matches nothing.
    pub fn set_agent_key(&self, key: &str) {
        self.agent.lock().unwrap_or_else(|e| e.into_inner()).current = key.to_string();
    }

    pub fn agent_key(&self) -> String {
        self.agent
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .current
            .clone()
    }

    /// Replace the agent key.
    ///
    /// `now` drops the previous key immediately; otherwise it stays valid for
    /// [`KEY_GRACE`], and never past this process's exit, so a client
    /// mid-session finishes its work and only then needs the new stanza.
    pub fn rotate_agent(&self, key: &str, now: bool) {
        let mut agent = self.agent.lock().unwrap_or_else(|e| e.into_inner());
        let was = std::mem::replace(&mut agent.current, key.to_string());
        agent.previous = if now || was.is_empty() {
            None
        } else {
            Some((was, std::time::Instant::now() + KEY_GRACE))
        };
    }

    /// How long the previous key keeps working, if one still does.
    ///
    /// `None` once it has expired or was dropped immediately, which is what the
    /// Privacy page shows as a countdown rather than as "until this exits".
    pub fn key_grace(&self) -> Option<Duration> {
        let agent = self.agent.lock().unwrap_or_else(|e| e.into_inner());
        agent
            .previous
            .as_ref()
            .map(|(_, until)| until.saturating_duration_since(std::time::Instant::now()))
            .filter(|left| !left.is_zero())
    }

    pub fn mcp_open(&self) -> bool {
        self.mcp_open.load(Ordering::Relaxed)
    }

    pub fn set_mcp_open(&self, open: bool) {
        self.mcp_open.store(open, Ordering::Relaxed);
    }

    pub fn port(&self) -> u16 {
        self.port
    }

    pub fn token(&self) -> String {
        self.token.lock().unwrap_or_else(|e| e.into_inner()).clone()
    }

    /// Invalidate the current token and return the new one.
    pub fn rotate(&self) -> String {
        let fresh = new_token();
        *self.token.lock().unwrap_or_else(|e| e.into_inner()) = fresh.clone();
        fresh
    }

    /// The URL to print. The token appears here and nowhere else.
    pub fn url(&self) -> String {
        format!("http://127.0.0.1:{}/?token={}", self.port, self.token())
    }

    /// Serve until `stop` is set, calling `refused` once per refused request.
    ///
    /// `stop` is `'static` because a thread has to watch it: `accept` blocks,
    /// and something has to notice the flag and wake it. Polling `accept`
    /// instead would put the poll interval on the front of every request —
    /// this server closes each connection after answering, so that is every
    /// request, not every session.
    pub fn serve(
        &self,
        handler: Handler,
        stop: &'static AtomicBool,
        refused: impl Fn(Refusal) + Send + Sync + 'static,
    ) -> Result<()> {
        let (tx, rx) = mpsc::channel::<Ready>();
        // Behind a slot, so shutdown can close the pool while a socket is
        // still waiting for its first byte: a waiting thread holding a sender
        // of its own would keep every worker's `recv` open until its read
        // timed out.
        let pool = Arc::new(Mutex::new(Some(tx)));
        let rx = Arc::new(Mutex::new(rx));
        let refused = Arc::new(refused);
        let live = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let pending = Arc::new(Mutex::new(Pending::default()));
        // Places held by connections still waiting for their head; nothing
        // reads it, but every `InFlight` counts somewhere.
        let waiting = Arc::new(std::sync::atomic::AtomicUsize::new(0));

        // Refused requests are answered by a thread of their own. Holding a
        // worker for the delay would mean eight wrong tokens at once could stop
        // the portal answering anything, which would make the throttle a denial
        // of service with extra steps.
        let (late_tx, late_rx) = mpsc::channel::<Late>();
        let waiting_room = std::thread::spawn(move || {
            while let Ok(late) = late_rx.recv() {
                let now = std::time::Instant::now();
                if late.due > now {
                    std::thread::sleep(late.due - now);
                }
                let mut stream = late.stream;
                let _ = write_response(&mut stream, Response::new(401, "text/plain", Vec::new()));
            }
        });

        let port = self.port;
        // One worker, built the same way whether it is one of the first eight
        // or the replacement for one that died. Every lock it takes is
        // recovered rather than expected: a thread that panicked while holding
        // one left the data behind it intact — `Response` and `Auth` are values,
        // not half-written state — and refusing to look at it afterwards turns
        // one panic into a dead server.
        let spawn_worker = {
            let rx = Arc::clone(&rx);
            let late_tx = late_tx.clone();
            let penalty = Arc::clone(&self.penalty);
            let handler = Arc::clone(&handler);
            let token = Arc::clone(&self.token);
            let agent = Arc::clone(&self.agent);
            let mcp_open = Arc::clone(&self.mcp_open);
            let refused = Arc::clone(&refused);
            move || {
                let rx = Arc::clone(&rx);
                let late_tx = late_tx.clone();
                let penalty = Arc::clone(&penalty);
                let handler = Arc::clone(&handler);
                let token = Arc::clone(&token);
                let agent = Arc::clone(&agent);
                let mcp_open = Arc::clone(&mcp_open);
                let refused = Arc::clone(&refused);
                std::thread::spawn(move || {
                    loop {
                        // The receiver is behind a mutex only for the recv; a
                        // worker that took a connection does not hold up the
                        // next.
                        let next = {
                            let guard = rx.lock().unwrap_or_else(|e| e.into_inner());
                            guard.recv()
                        };
                        let Ok(ready) = next else {
                            return;
                        };
                        let hold = |stream, in_flight| {
                            let delay = penalty.lock().unwrap_or_else(|e| e.into_inner()).charge();
                            let _ = late_tx.send(Late {
                                stream,
                                due: std::time::Instant::now() + delay,
                                _in_flight: in_flight,
                            });
                        };
                        let auth = {
                            let agent = agent.lock().unwrap_or_else(|e| e.into_inner());
                            Auth {
                                token: token.lock().unwrap_or_else(|e| e.into_inner()).clone(),
                                agent: agent.current.clone(),
                                previous: agent.previous.clone(),
                                mcp_open: mcp_open.load(Ordering::Relaxed),
                                port,
                            }
                        };
                        if let Some(class) = answer(ready, &handler, &auth, &hold) {
                            refused(class);
                        }
                    }
                })
            }
        };

        let mut workers: Vec<_> = (0..WORKERS).map(|_| spawn_worker()).collect();

        // The waker: four wakeups a second while idle, nothing per request.
        // `WAKE` covers the case where the loop leaves for its own reasons, so
        // this thread never outlives the server that started it.
        let waker = {
            let port = self.port;
            std::thread::spawn(move || {
                while !stop.load(Ordering::Relaxed) && !WAKE.load(Ordering::Relaxed) {
                    std::thread::sleep(Duration::from_millis(250));
                }
                // One connection that goes nowhere, purely to return `accept`.
                let _ = TcpStream::connect(("127.0.0.1", port));
            })
        };

        while !stop.load(Ordering::Relaxed) {
            match self.listener.accept() {
                Ok((stream, _)) => {
                    // The waker's own connection arrives here too; it carries
                    // no request, so `answer` reads nothing and closes it.
                    if stop.load(Ordering::Relaxed) {
                        break;
                    }
                    let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
                    let _ = stream.set_write_timeout(Some(IO_TIMEOUT));

                    // Counted from the moment a head has arrived, queued or
                    // not: a connection waiting on a worker is still one this
                    // process is holding open. A connection that has not sent
                    // its head yet is bounded by `Pending` instead, so idle
                    // sockets cannot fill this count (S11).
                    let busy = live.load(Ordering::Relaxed);
                    trace(&stream, || format!("accepted, {busy} live"));
                    if busy >= MAX_CONNECTIONS {
                        let in_flight = InFlight::taken(&live);
                        // Off the accept loop: it may wait for the request.
                        std::thread::spawn(move || {
                            refuse_busy(stream);
                            drop(in_flight);
                        });
                        continue;
                    }
                    // To the pool only once its whole head has arrived. A
                    // browser opens sockets ahead of the requests it will put
                    // on them, and a worker that took one sat in `read_head`
                    // until the 30 s read timeout: eight of them held the
                    // whole pool, and the requests behind them queued past the
                    // limit into 503s and, on Windows, resets (#181). Until
                    // 0.40.0 the first byte was enough, so eight sockets that
                    // each sent one byte and then dribbled held the pool just
                    // the same (S11). The head is read here, on the socket's
                    // own thread, against the request deadline, and
                    // `Pending` bounds how many such threads there can be.
                    let accepted = std::time::Instant::now();
                    let (Ok(peer), Ok(handle)) = (stream.peer_addr(), stream.try_clone()) else {
                        continue;
                    };
                    let state = pending.lock().unwrap_or_else(|e| e.into_inner()).admit(
                        peer.ip(),
                        handle,
                        InFlight::taken(&waiting),
                    );
                    let pool = Arc::clone(&pool);
                    let pending = Arc::clone(&pending);
                    let refused = Arc::clone(&refused);
                    let live = Arc::clone(&live);
                    std::thread::spawn(move || {
                        let mut reader = BufReader::new(stream);
                        let head = read_head(&mut reader, accepted, &state);
                        let Some(mut in_flight) = pending
                            .lock()
                            .unwrap_or_else(|e| e.into_inner())
                            .release(&state)
                        else {
                            return;
                        };
                        match head {
                            Ok(Some(request)) => {
                                in_flight.move_to(&live);
                                let _ = reader.get_ref().set_read_timeout(Some(IO_TIMEOUT));
                                if let (Ok(stream), Some(tx)) = (
                                    reader.get_ref().try_clone(),
                                    pool.lock().unwrap_or_else(|e| e.into_inner()).as_ref(),
                                ) {
                                    let _ = tx.send(Ready {
                                        stream,
                                        reader,
                                        request,
                                        in_flight,
                                        accepted,
                                    });
                                }
                            }
                            Ok(None) => {
                                trace(reader.get_ref(), || "closed before a request".to_string())
                            }
                            Err(e) => {
                                trace(reader.get_ref(), || format!("bad request head: {e:#}"));
                                let status = e.downcast_ref::<HeadRefused>().map_or(400, |r| r.0);
                                let _ = write_response(
                                    reader.get_mut(),
                                    Response::new(status, "text/plain", Vec::new()),
                                );
                                drop(in_flight);
                                refused(Refusal::Malformed);
                            }
                        }
                    });
                    // A worker that ended for any reason is replaced here, so
                    // the pool is eight threads for as long as the daemon runs
                    // rather than eight minus however many requests have gone
                    // wrong since it started. Checked on accept because that is
                    // the only moment something is known to be happening.
                    for worker in &mut workers {
                        if worker.is_finished() {
                            *worker = spawn_worker();
                        }
                    }
                }
                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                Err(e) => {
                    WAKE.store(true, Ordering::Relaxed);
                    let _ = waker.join();
                    return Err(e).context("accepting a connection");
                }
            }
        }

        WAKE.store(true, Ordering::Relaxed);
        let _ = waker.join();

        // Dropping the sender ends every worker's `recv`, so shutdown finishes
        // the requests in flight and starts no more.
        pool.lock().unwrap_or_else(|e| e.into_inner()).take();
        for worker in workers {
            let _ = worker.join();
        }
        // `spawn_worker` captured a sender of its own so it could hand one to
        // each worker it makes, and it outlives the workers — so dropping the
        // one below is not enough on its own. Both have to go before the
        // waiting room's `recv` can end, and a shutdown that waits on a channel
        // nothing will ever close is a daemon that does not exit when it is
        // asked to.
        drop(spawn_worker);
        // Dropped after the workers have been joined, so the waiting room
        // answers whatever it still owes before it ends. A shutdown that closed
        // it first would look like a dropped connection to whoever was being
        // held.
        drop(late_tx);
        let _ = waiting_room.join();
        Ok(())
    }
}

/// The credentials in force for one request.
struct Auth {
    token: String,
    agent: String,
    previous: Option<(String, std::time::Instant)>,
    mcp_open: bool,
    /// The port this server answers on, so an `Origin` header can be compared
    /// against the origin it actually names.
    port: u16,
}

impl Auth {
    /// Whether a bearer credential opens the MCP endpoint.
    ///
    /// An empty configured key matches nothing: `same` compares lengths first,
    /// and an empty bearer against an empty key would otherwise be a match.
    fn is_agent(&self, bearer: &str) -> bool {
        if bearer.is_empty() {
            return false;
        }
        if !self.agent.is_empty() && same(bearer, &self.agent) {
            return true;
        }
        self.previous.as_ref().is_some_and(|(previous, until)| {
            !previous.is_empty() && *until > std::time::Instant::now() && same(bearer, previous)
        })
    }
}

/// Whether `SEMLITH_HTTP_TRACE=1`: one stderr line per connection event --
/// accepted, closed before a request, answered (route, status, time, whether
/// the write reached the client). A diagnostic for transport faults that only
/// one platform shows, such as #181 on Windows; off by default.
fn tracing() -> bool {
    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *ON.get_or_init(|| std::env::var("SEMLITH_HTTP_TRACE").is_ok_and(|v| v == "1"))
}

/// One trace line, when tracing is on.
fn trace(stream: &TcpStream, what: impl FnOnce() -> String) {
    if tracing() {
        let port = stream.peer_addr().map(|a| a.port()).unwrap_or(0);
        eprintln!("semlith http: {port} {}", what());
    }
}

/// How long a connection past the limit may take to finish sending its
/// request before it is answered anyway.
const BUSY_READ: Duration = Duration::from_millis(50);

/// The most of a refused request read before answering it.
const BUSY_DRAIN: usize = 64 * 1024;

/// Answer a connection past [`MAX_CONNECTIONS`] with 503, having read its
/// request first.
///
/// Closing a socket whose request is still unread is a reset, not a close, on
/// Windows and Linux alike, and a browser shows a reset as
/// ERR_CONNECTION_RESET rather than the 503 -- the Windows drive's intermittent
/// failure (#181). So the head and the body it declares are read off first,
/// up to [`BUSY_DRAIN`], and the write side is shut so the client reads the
/// 503 and then an end of stream. On loopback the request lands with the
/// connect, so this is microseconds; [`BUSY_READ`] bounds what a slow sender
/// can cost the accept loop.
fn refuse_busy(mut stream: TcpStream) {
    let started = std::time::Instant::now();
    let _ = stream.set_read_timeout(Some(BUSY_READ));
    let mut seen: Vec<u8> = Vec::new();
    let mut want: Option<usize> = None;
    let mut buf = [0u8; 8 * 1024];
    // A socket that has sent nothing is closed rather than answered: it is a
    // browser's spare, and a 503 sent before any request is bytes the browser
    // reads as the answer to the request it later puts there. Closed, it opens
    // another.
    let mut first = [0u8; 1];
    if !matches!(stream.peek(&mut first), Ok(n) if n > 0) {
        trace(&stream, || "over the limit, closed: no request".to_string());
        return;
    }
    trace(&stream, || "over the limit, answered 503".to_string());
    while seen.len() < BUSY_DRAIN && started.elapsed() < BUSY_READ {
        if want.is_some_and(|total| seen.len() >= total) {
            break;
        }
        match stream.read(&mut buf) {
            Ok(0) | Err(_) => break,
            Ok(n) => seen.extend_from_slice(&buf[..n]),
        }
        if want.is_none()
            && let Some(end) = seen.windows(4).position(|w| w == b"\r\n\r\n")
        {
            let head = String::from_utf8_lossy(&seen[..end]).to_ascii_lowercase();
            let body = head
                .lines()
                .find_map(|l| l.strip_prefix("content-length:"))
                .and_then(|v| v.trim().parse::<usize>().ok())
                .unwrap_or(0);
            want = Some(end + 4 + body);
        }
    }
    let _ = write_response(&mut stream, Response::new(503, "text/plain", Vec::new()));
}

/// Answer one connection. Returns why it was refused, if it was.
///
/// `hold` takes a connection whose credential was wrong and answers it later,
/// off this thread. Everything else is answered here and now: a malformed
/// request, a foreign `Host` and a cross-origin write are rule violations rather
/// than guesses at a secret, and delaying them would only slow down the person
/// who made a mistake.
fn answer(
    ready: Ready,
    handler: &Handler,
    auth: &Auth,
    hold: &impl Fn(TcpStream, InFlight),
) -> Option<Refusal> {
    // The head was read before this request reached a worker, and only the
    // head. Everything that can refuse this request is decided from it, so a
    // request that is going to be refused never has its body read into
    // memory — which is what keeps a thousand cross-origin attempts from
    // costing a thousand bodies' worth of it.
    let Ready {
        mut stream,
        mut reader,
        mut request,
        in_flight,
        accepted: started,
    } = ready;

    // Host first: a request from a foreign origin is refused before the token
    // is even compared, so a wrong guess and a right one cost the same.
    if !loopback_host(request.header("host")) {
        drain_body(&mut reader, &request, started);
        let _ = write_response(&mut stream, Response::new(400, "text/plain", Vec::new()));
        return Some(Refusal::ForeignHost);
    }

    let want = auth.token.as_str();
    let for_mcp = request.path == MCP_PATH;

    // A closed endpoint is not a route. Answered before the credential is
    // looked at, so a client that was told to stop learns the same thing
    // whether or not it still holds a key.
    if for_mcp && !auth.mcp_open {
        drain_body(&mut reader, &request, started);
        let _ = write_response(&mut stream, Response::error(404, "no such route"));
        return None;
    }

    // HEAD is GET without the body. Answered by running the handler and
    // dropping what it produced, so a probe sees the real status and the real
    // headers rather than the 404 an unhandled method would give it.
    let head_only = request.method == "HEAD";
    let reading = head_only || request.method == "GET";

    // A write proves where it came from before anything else looks at it, so a
    // page on another local port gets the same answer whether its token guess
    // was right or wrong — and gets it without a handler having run.
    if !reading {
        if !same_origin(&request, auth.port) {
            drain_body(&mut reader, &request, started);
            let _ = write_response(&mut stream, Response::new(403, "text/plain", Vec::new()));
            return Some(Refusal::CrossOrigin);
        }
        if !json_body(request.header("content-type")) {
            drain_body(&mut reader, &request, started);
            let _ = write_response(&mut stream, Response::new(403, "text/plain", Vec::new()));
            return Some(Refusal::BadContentType);
        }
    }

    // The page and the bytes it loads carry no credential, because a browser
    // attaches no header to a stylesheet, a font or a favicon. They are the
    // same bytes in every copy of this binary; everything that answers about
    // this machine is below and needs the token.
    if !public(&request, reading) {
        // The agent key opens `/mcp` and nothing else. This is the whole reason
        // there are two credentials: a key that a client keeps in a file on disk
        // must not be able to rotate a token, adopt a store or start an upgrade.
        let by_agent = for_mcp
            && request
                .header("authorization")
                .and_then(|value| value.strip_prefix("Bearer "))
                .is_some_and(|bearer| auth.is_agent(bearer.trim()));
        let by_header = request
            .header(TOKEN_HEADER_KEY)
            .is_some_and(|t| same(t.trim(), want));

        // The agent key where it does not reach: known, and forbidden. A 403
        // rather than the prober's held 401, because the caller already holds
        // a real key and the answer it needs is "not with that" — the routes
        // that accept a refused file among them (2.5).
        let agent_elsewhere = !for_mcp
            && !by_header
            && request
                .header("authorization")
                .and_then(|value| value.strip_prefix("Bearer "))
                .is_some_and(|bearer| auth.is_agent(bearer.trim()));
        if agent_elsewhere {
            drain_body(&mut reader, &request, started);
            let _ = write_response(&mut stream, Response::new(403, "text/plain", Vec::new()));
            return Some(Refusal::Unauthorized);
        }
        if !by_header && !by_agent {
            // Empty body, and not yet: a prober learns that something refused
            // it, nothing else, and not quickly.
            drain_body(&mut reader, &request, started);
            hold(stream, in_flight);
            return Some(Refusal::Unauthorized);
        }
    }

    // Allowed, so now the body — and now the clock matters, because from here
    // the connection is one this server intends to do work for.
    if read_body(&mut reader, &mut request).is_err() {
        let _ = write_response(&mut stream, Response::new(400, "text/plain", Vec::new()));
        return Some(Refusal::Malformed);
    }
    if started.elapsed() > REQUEST_DEADLINE {
        // A request dribbled out one byte at a time holds a worker for as long
        // as the sender likes. Ten seconds is far more than a loopback request
        // needs and far less than a client can hold.
        let _ = write_response(&mut stream, Response::new(408, "text/plain", Vec::new()));
        return Some(Refusal::Malformed);
    }

    let request = if head_only {
        Request {
            method: "GET".to_string(),
            ..request
        }
    } else {
        request
    };

    // A panicking route answers 500 and the daemon keeps serving. Before
    // 0.14.0 it took the worker with it permanently and poisoned whatever locks
    // it held, so one malformed request cost an eighth of the server until a
    // restart — and eight of them cost all of it.
    let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| handler(&request)));
    let mut response = match caught {
        Ok(response) => response,
        Err(_) => {
            eprintln!(
                "semlith: {} {} panicked; answered 500 and kept serving",
                request.method, request.path
            );
            Response::new(500, "text/plain", Vec::new())
        }
    };
    if head_only {
        response.body = Body::Bytes(Vec::new());
    }

    // A streamed response leaves the pool. An index run holds its connection
    // for as long as the run takes — minutes on a large corpus — and doing that
    // on a worker means the portal is answering on seven threads while it runs,
    // or on none at all once somebody starts eight.
    if matches!(response.body, Body::Stream(_)) {
        std::thread::spawn(move || {
            let _in_flight = in_flight;
            let _ = write_response(&mut stream, response);
        });
        return None;
    }

    let status = response.status;
    let wrote = write_response(&mut stream, response);
    trace(&stream, || {
        format!(
            "{} {} -> {status} in {} ms, write {:?}",
            request.method,
            request.path,
            started.elapsed().as_millis(),
            wrote.as_ref().map(|_| "ok")
        )
    });
    None
}

/// Whether this request may be answered without a credential.
///
/// The page itself and the files it loads, and the 404 for a path that is
/// neither. Nothing that says anything about this machine: `/api` and `/mcp`
/// are below and need the token.
///
/// The token in the printed URL's query is never read here: it is handed to
/// the page, which sends it back in a header, so a reload with no token in the
/// address bar still gets the page and a request for data still does not.
///
/// The last clause is why a missing asset is a 404 rather than a 401. A
/// browser attaches no header to an icon named by a manifest, so a wrong path
/// in that manifest read as an authentication failure and sent whoever
/// debugged it to look at the token. "This file is not here" discloses
/// nothing: the asset list is the same in every copy of this binary.
fn public(request: &Request, reading: bool) -> bool {
    if !reading {
        return false;
    }
    if request.path == "/" || crate::portal::asset(&request.path).is_some() {
        return true;
    }
    !request.path.starts_with("/api") && request.path != MCP_PATH
}

/// Whether a write came from this server's own origin.
///
/// `Sec-Fetch-Site` is the browser's own statement and cannot be set by script,
/// so when it is present it decides — and `same-site` is a refusal, because
/// every other port on `localhost` is same-site and none of them is this
/// server. A client that sends no fetch metadata is not a browser; it is judged
/// on `Origin` if it sent one and on the token alone if it did not.
fn same_origin(request: &Request, port: u16) -> bool {
    if let Some(site) = request.header("sec-fetch-site") {
        return site.trim().eq_ignore_ascii_case("same-origin");
    }
    match request.header("origin") {
        Some(origin) => own_origin(origin.trim(), port),
        None => true,
    }
}

/// Whether an `Origin` names this server: loopback, over http, on this port.
fn own_origin(origin: &str, port: u16) -> bool {
    let Some(authority) = origin.strip_prefix("http://") else {
        return false;
    };
    let Some((host, stated)) = authority.rsplit_once(':') else {
        return false;
    };
    let host = host.trim_start_matches('[').trim_end_matches(']');
    let loopback = host.eq_ignore_ascii_case("localhost") || host == "127.0.0.1" || host == "::1";
    loopback && stated.parse::<u16>() == Ok(port)
}

/// Whether `Host` is this machine talking to itself.
///
/// A missing header is refused: HTTP/1.1 requires one, and the clients that
/// omit it are not browsers.
fn loopback_host(host: Option<&str>) -> bool {
    let Some(host) = host else { return false };
    let name = host.rsplit_once(':').map_or(host, |(n, _)| n);
    let name = name.trim_start_matches('[').trim_end_matches(']');
    name.eq_ignore_ascii_case("localhost") || name == "127.0.0.1" || name == "::1"
}

/// Compare in time that does not depend on where the first wrong byte is.
///
/// The token is a secret and this is the check that guards it. The difference
/// is unmeasurable across a loopback socket, but a constant-time compare costs
/// nothing and removes the question.
fn same(a: &str, b: &str) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.bytes().zip(b.bytes()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// The request line and the headers, with no body read yet.
///
/// Split from the body on purpose. Every rule that can refuse a request is
/// decided from what is here, so a refusal costs the head and nothing more.
///
/// Read before the token is looked at, so it is bounded three ways: by
/// [`MAX_HEADERS`] as the bytes arrive (431), by [`REQUEST_DEADLINE`] from the
/// moment the connection was accepted however steadily they arrive (408), and
/// by `state.evicted`, which [`Pending`] sets when it closes this connection to
/// make room. `state.heard` tells [`Pending`] a byte has arrived, so the
/// connection is not idle. A connection that sent nothing before any of those is closed with no
/// answer: it is a browser's spare, and bytes sent before a request are bytes
/// the browser reads as the answer to the request it later puts there.
/// The head parser over `bytes`, as a peer that sent them and closed: the
/// fuzz target's way in (`fuzz/fuzz_targets/http_head.rs`). Through a real
/// loopback connection, because [`read_head`] sets its deadlines on the
/// socket; one listener for the whole run.
#[doc(hidden)]
pub fn fuzz_head(bytes: &[u8]) {
    use std::io::Write;
    static LISTENER: std::sync::OnceLock<std::net::TcpListener> = std::sync::OnceLock::new();
    let listener = LISTENER
        .get_or_init(|| std::net::TcpListener::bind("127.0.0.1:0").expect("a loopback port"));
    let Ok(mut peer) = TcpStream::connect(listener.local_addr().expect("a bound port")) else {
        return;
    };
    let _ = peer.write_all(bytes);
    let _ = peer.shutdown(std::net::Shutdown::Write);
    let Ok((accepted, _)) = listener.accept() else {
        return;
    };
    let state = HeadState {
        heard: AtomicBool::new(false),
        evicted: AtomicBool::new(false),
    };
    let _ = read_head(
        &mut BufReader::new(accepted),
        std::time::Instant::now(),
        &state,
    );
}

fn read_head(
    reader: &mut BufReader<TcpStream>,
    accepted: std::time::Instant,
    state: &HeadState,
) -> Result<Option<Request>> {
    let mut head = String::new();
    let mut line: Vec<u8> = Vec::new();
    let mut read = 0;
    loop {
        let nothing_yet = read == 0;
        let left = REQUEST_DEADLINE.saturating_sub(accepted.elapsed());
        if left.is_zero() || state.evicted.load(Ordering::Relaxed) {
            if nothing_yet {
                return Ok(None);
            }
            return Err(HeadRefused(408).into());
        }
        reader
            .get_ref()
            .set_read_timeout(Some(left.min(HEAD_POLL)))?;
        let buf = match reader.fill_buf() {
            Ok(buf) => buf,
            // A slice of the deadline passed; the checks above decide.
            Err(e)
                if matches!(
                    e.kind(),
                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                ) =>
            {
                continue;
            }
            // Reset or shut before a byte: as good as closed.
            Err(_) if nothing_yet => return Ok(None),
            Err(e) => return Err(e.into()),
        };
        if buf.is_empty() {
            // The peer closed before sending anything: a browser probing a
            // connection it then did not use. Not a refusal, not an error.
            anyhow::ensure!(nothing_yet, "the connection closed mid-head");
            return Ok(None);
        }
        state.heard.store(true, Ordering::Relaxed);
        let (taken, complete) = match buf.iter().position(|b| *b == b'\n') {
            Some(at) => (at + 1, true),
            None => (buf.len(), false),
        };
        read += taken;
        if read > MAX_HEADERS {
            return Err(HeadRefused(431).into());
        }
        line.extend_from_slice(&buf[..taken]);
        reader.consume(taken);
        if !complete {
            continue;
        }
        let text = String::from_utf8(std::mem::take(&mut line)).context("head is not UTF-8")?;
        if text == "\r\n" || text == "\n" {
            break;
        }
        head.push_str(&text);
    }

    let mut lines = head.lines();
    let start = lines.next().context("empty request")?;
    let mut parts = start.split_whitespace();
    let method = parts.next().context("no method")?.to_string();
    let target = parts.next().context("no target")?.to_string();

    let mut headers = BTreeMap::new();
    for line in lines {
        if let Some((name, value)) = line.split_once(':') {
            let name = name.trim().to_ascii_lowercase();
            // A header every check reads once, sent twice, is two requests
            // in one: the checks saw the last copy and a proxy in front might
            // honour the first (audit S23). Refused, not resolved.
            if SINGLE.contains(&name.as_str()) && headers.contains_key(&name) {
                return Err(HeadRefused(400).into());
            }
            headers.insert(name, value.trim().to_string());
        }
    }

    let (raw_path, raw_query) = match target.split_once('?') {
        Some((p, q)) => (p, q),
        None => (target.as_str(), ""),
    };

    Ok(Some(Request {
        method,
        path: percent_decode(raw_path),
        query: parse_query(raw_query),
        headers,
        body: Vec::new(),
    }))
}

/// Read and throw away a declared body, so a refused peer can read its refusal.
///
/// A server that answers and closes while the client is still writing gives
/// that client a reset connection rather than the status it was sent: the
/// response is in the socket, and the reset takes it with it. So a refusal
/// drains what the client said it was sending — through a small buffer, never
/// into one the size of the body, which is the whole point of refusing before
/// the body is read. Bounded by the same cap a real body has and by the
/// request deadline, so draining cannot be the hold it prevents.
fn drain_body(reader: &mut BufReader<TcpStream>, request: &Request, started: std::time::Instant) {
    let Some(len) = request
        .headers
        .get("content-length")
        .and_then(|v| v.parse::<usize>().ok())
    else {
        return;
    };
    let mut left = len.min(MAX_BODY);
    let mut sink = [0u8; 8 * 1024];
    while left > 0 && started.elapsed() <= REQUEST_DEADLINE {
        let want = left.min(sink.len());
        match reader.read(&mut sink[..want]) {
            Ok(0) | Err(_) => return,
            Ok(n) => left -= n,
        }
    }
}

/// The body, once the request has earned one.
fn read_body(reader: &mut BufReader<TcpStream>, request: &mut Request) -> Result<()> {
    let Some(len) = request
        .headers
        .get("content-length")
        .and_then(|v| v.parse::<usize>().ok())
    else {
        return Ok(());
    };
    anyhow::ensure!(len <= MAX_BODY, "request body too large");
    request.body.resize(len, 0);
    reader.read_exact(&mut request.body)?;
    Ok(())
}

/// `a=1&b=2` to a map, with a repeated key's values joined by `\u{1}`.
fn parse_query(raw: &str) -> BTreeMap<String, String> {
    let mut out: BTreeMap<String, String> = BTreeMap::new();
    for pair in raw.split('&').filter(|p| !p.is_empty()) {
        let (key, value) = pair.split_once('=').unwrap_or((pair, ""));
        let key = percent_decode(&key.replace('+', " "));
        let value = percent_decode(&value.replace('+', " "));
        out.entry(key)
            .and_modify(|existing| {
                existing.push('\u{1}');
                existing.push_str(&value);
            })
            .or_insert(value);
    }
    out
}

fn percent_decode(raw: &str) -> String {
    let bytes = raw.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'%' && i + 2 < bytes.len() {
            let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).unwrap_or("");
            if let Ok(byte) = u8::from_str_radix(hex, 16) {
                out.push(byte);
                i += 3;
                continue;
            }
        }
        out.push(bytes[i]);
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

fn write_response(stream: &mut TcpStream, response: Response) -> std::io::Result<()> {
    let written = write_whole(stream, response);
    linger(stream);
    written
}

/// How long a closing connection waits for the client to close its side.
const LINGER: Duration = Duration::from_millis(500);

/// Close the way a web server must when it closes after every response: shut
/// the write side, so the client reads the response and then an end of
/// stream, and read whatever it still sends until it closes or [`LINGER`]
/// passes. Closing a socket with bytes unread in it sends a reset instead of
/// a close on Windows, macOS and Linux, and a browser that receives one
/// discards the response it has already read -- the Windows drive's
/// ERR_CONNECTION_RESET on routes the daemon had answered 200 and written in
/// full (#181).
fn linger(stream: &mut TcpStream) {
    if stream.shutdown(std::net::Shutdown::Write).is_err() {
        return;
    }
    let started = std::time::Instant::now();
    let _ = stream.set_read_timeout(Some(LINGER));
    let mut sink = [0u8; 4096];
    while started.elapsed() < LINGER {
        match stream.read(&mut sink) {
            Ok(0) | Err(_) => return,
            Ok(_) => {}
        }
    }
}

fn write_whole(stream: &mut TcpStream, response: Response) -> std::io::Result<()> {
    let Response {
        status,
        content_type,
        extra,
        csp,
        body,
    } = response;

    let mut head = format!("HTTP/1.1 {status} {}\r\n", reason(status));
    head.push_str(&format!("Content-Type: {content_type}\r\n"));
    head.push_str(&format!(
        "Content-Security-Policy: {}\r\n",
        csp.unwrap_or(PAGE_CSP)
    ));
    head.push_str("X-Content-Type-Options: nosniff\r\n");
    head.push_str("Referrer-Policy: no-referrer\r\n");
    // The token lives in the first URL. Nothing here may be stored by a proxy
    // that does not exist, but a browser's back-forward cache holding a
    // rotated token is a real thing to avoid.
    head.push_str("Cache-Control: no-store\r\n");
    head.push_str("Connection: close\r\n");
    for (name, value) in &extra {
        head.push_str(&format!("{name}: {value}\r\n"));
    }

    match body {
        Body::Bytes(bytes) => {
            head.push_str(&format!("Content-Length: {}\r\n\r\n", bytes.len()));
            stream.write_all(head.as_bytes())?;
            stream.write_all(&bytes)?;
            stream.flush()
        }
        Body::Stream(f) => {
            head.push_str("Transfer-Encoding: chunked\r\n\r\n");
            stream.write_all(head.as_bytes())?;
            stream.flush()?;
            let result = {
                let mut chunks = Chunks { out: stream };
                f(&mut chunks)
            };
            // The terminating chunk goes out whether or not the producer
            // failed: a client reading a stream that simply stops has no way
            // to tell a finished run from a dropped connection.
            stream.write_all(b"0\r\n\r\n")?;
            stream.flush()?;
            result
        }
    }
}

/// The headers a request may carry once.
const SINGLE: &[&str] = &[
    "host",
    "origin",
    "sec-fetch-site",
    "content-type",
    "content-length",
    "semlith-token",
    "authorization",
];

fn reason(status: u16) -> &'static str {
    match status {
        200 => "OK",
        400 => "Bad Request",
        401 => "Unauthorized",
        403 => "Forbidden",
        404 => "Not Found",
        405 => "Method Not Allowed",
        408 => "Request Timeout",
        409 => "Conflict",
        431 => "Request Header Fields Too Large",
        413 => "Content Too Large",
        500 => "Internal Server Error",
        502 => "Bad Gateway",
        503 => "Service Unavailable",
        // A status this table does not know is still not "OK" (audit S23).
        _ => "Unknown",
    }
}

/// A 256-bit token, hex encoded.
///
/// From the OS random source, the same construction `home::new_agent_key` uses.
/// It was a blake3 hash of the clock, this process's id and the address of a
/// fresh allocation until 0.14.0 — none of which is a secret. A clock is
/// readable by anything on the machine, a pid is in `/proc`, and an allocator
/// address under a defeated ASLR is a small space; a hash of three guessable
/// things is a guessable thing, however long its output is.
fn new_token() -> String {
    let mut bytes = [0u8; 32];
    let filled = getrandom::fill(&mut bytes).map(|()| bytes);
    token_from(filled)
}

/// The hex encoding, and the one place the random source failing is decided.
///
/// Split out so a test can hand it a failure: there is no way to make the OS
/// refuse, and a token minted from a fallback nobody noticed is exactly the
/// thing this function exists to prevent.
fn token_from(filled: Result<[u8; 32], getrandom::Error>) -> String {
    let bytes = filled
        .expect("the OS random source refused; semlith will not mint a session token without it");
    let mut out = String::with_capacity(64);
    for byte in bytes {
        use std::fmt::Write;
        let _ = write!(out, "{byte:02x}");
    }
    out
}

/// What a refused request costs, and what a run of them costs.
///
/// A wrong token is answered late. The delay is what makes guessing a 256-bit
/// token over a loopback socket pointless rather than merely impractical, and
/// it grows with how many refusals this daemon has just answered, so a run of
/// guesses gets slower while one mistyped URL costs a quarter of a second.
///
/// The window lives here and resets with the daemon, which is the honest
/// lifetime: the token it guards resets then too.
#[derive(Default)]
struct Penalty {
    /// When each refusal in the last minute was answered.
    recent: std::collections::VecDeque<std::time::Instant>,
}

/// What one refusal costs before any escalation.
const REFUSAL_DELAY: Duration = Duration::from_millis(250);

/// The longest a refusal is ever held.
const MAX_REFUSAL_DELAY: Duration = Duration::from_secs(4);

/// How many refusals a minute may hold before each further block doubles it.
const REFUSALS_PER_STEP: usize = 20;

impl Penalty {
    /// Record a refusal and say how long to hold it.
    fn charge(&mut self) -> Duration {
        let now = std::time::Instant::now();
        while self
            .recent
            .front()
            .is_some_and(|t| now.duration_since(*t) > Duration::from_secs(60))
        {
            self.recent.pop_front();
        }
        self.recent.push_back(now);
        // Every further twenty in the window doubles it: 250 ms, then 500, then
        // one second, two, and four.
        let steps = (self.recent.len().saturating_sub(1)) / REFUSALS_PER_STEP;
        let delay = REFUSAL_DELAY * 2u32.saturating_pow(steps.min(8) as u32);
        delay.min(MAX_REFUSAL_DELAY)
    }
}

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

    /// The Host check is what stops a page on another origin reaching this
    /// server with the browser's cookies attached, so the names it accepts are
    /// worth stating exactly.
    #[test]
    fn only_loopback_hosts_are_answered() {
        assert!(loopback_host(Some("localhost")));
        assert!(loopback_host(Some("localhost:7365")));
        assert!(loopback_host(Some("127.0.0.1:7365")));
        assert!(loopback_host(Some("LOCALHOST:7365")));
        assert!(loopback_host(Some("[::1]:7365")));

        assert!(!loopback_host(Some("evil.example")));
        assert!(!loopback_host(Some("evil.example:7365")));
        // A name that merely ends in localhost is a different host.
        assert!(!loopback_host(Some("notlocalhost")));
        assert!(!loopback_host(Some("localhost.evil.example")));
        // 127.0.0.1 by another spelling is still a foreign Host header: the
        // check is on the name the browser sent, not on where it resolves.
        assert!(!loopback_host(Some("127.1")));
        assert!(!loopback_host(None));
    }

    fn write(headers: &[(&str, &str)]) -> Request {
        Request {
            method: "POST".into(),
            path: "/api/index".into(),
            query: BTreeMap::new(),
            headers: headers
                .iter()
                .map(|(k, v)| (k.to_ascii_lowercase(), v.to_string()))
                .collect(),
            body: Vec::new(),
        }
    }

    /// The rule that replaces the cookie. A page on another port of the same
    /// host is `same-site`, which is why `same-site` has to be a refusal: it is
    /// exactly the case the cookie could not tell apart from this server.
    #[test]
    fn only_a_same_origin_write_is_accepted() {
        assert!(same_origin(
            &write(&[("Sec-Fetch-Site", "same-origin")]),
            7365
        ));
        assert!(!same_origin(
            &write(&[("Sec-Fetch-Site", "same-site")]),
            7365
        ));
        assert!(!same_origin(
            &write(&[("Sec-Fetch-Site", "cross-site")]),
            7365
        ));
        assert!(!same_origin(&write(&[("Sec-Fetch-Site", "none")]), 7365));

        // A browser sends both; the fetch metadata decides, so an Origin a page
        // could have lied about never rescues a cross-site write.
        assert!(!same_origin(
            &write(&[
                ("Sec-Fetch-Site", "same-site"),
                ("Origin", "http://127.0.0.1:7365"),
            ]),
            7365
        ));

        // A client that is not a browser sends neither, and is judged on the
        // token alone — `curl` has to keep working.
        assert!(same_origin(&write(&[]), 7365));
        assert!(same_origin(
            &write(&[("Origin", "http://localhost:7365")]),
            7365
        ));
        assert!(!same_origin(
            &write(&[("Origin", "http://127.0.0.1:9999")]),
            7365
        ));
        assert!(!same_origin(&write(&[("Origin", "null")]), 7365));
        assert!(!same_origin(
            &write(&[("Origin", "https://127.0.0.1:7365")]),
            7365
        ));
    }

    /// The three types a form can post with no preflight are the three this
    /// refuses, which is what makes a cross-origin form useless against a route.
    #[test]
    fn only_a_json_body_is_read() {
        assert!(json_body(Some("application/json")));
        assert!(json_body(Some("application/json; charset=utf-8")));
        assert!(json_body(Some("Application/JSON")));

        assert!(!json_body(Some("text/plain")));
        assert!(!json_body(Some("application/x-www-form-urlencoded")));
        assert!(!json_body(Some("multipart/form-data; boundary=x")));
        assert!(!json_body(Some("application/json-patch+json")));
        assert!(!json_body(None));
    }

    /// The page and its own bytes are public; anything that answers about this
    /// machine is not, however it is asked for.
    #[test]
    fn only_the_page_and_its_assets_are_public() {
        let get = |path: &str| Request {
            method: "GET".into(),
            path: path.into(),
            query: BTreeMap::new(),
            headers: BTreeMap::new(),
            body: Vec::new(),
        };
        assert!(public(&get("/"), true));
        assert!(public(&get("/app.js"), true));
        assert!(public(&get("/style.css"), true));

        // A path that is not an asset is public so that it can be answered
        // with a 404 rather than a 401 — see `public`.
        assert!(public(&get("/nope.png"), true));
        assert!(public(&get("/icons/nope.png"), true));

        assert!(!public(&get("/api/stores"), true));
        assert!(!public(&get("/api/nope"), true));
        assert!(!public(&get("/api/image"), true));
        assert!(!public(&get(MCP_PATH), true));
        // A write is never public, whatever it names.
        assert!(!public(&get("/"), false));
    }

    #[test]
    fn tokens_compare_by_value_and_by_length() {
        assert!(same("abc", "abc"));
        assert!(!same("abc", "abd"));
        assert!(!same("abc", "abcd"));
        assert!(!same("", "a"));
    }

    /// `?lang=rust&lang=go` is how the portal sends a repeated filter, and a
    /// map that kept only the last one would silently narrow the search.
    #[test]
    fn a_repeated_query_key_keeps_every_value() {
        let q = parse_query("lang=rust&lang=go&k=5");
        let request = Request {
            method: "GET".into(),
            path: "/".into(),
            query: q,
            headers: BTreeMap::new(),
            body: Vec::new(),
        };
        assert_eq!(request.query_all("lang"), vec!["rust", "go"]);
        assert_eq!(request.query("k"), Some("5"));
        assert!(request.query_all("ext").is_empty());
    }

    #[test]
    fn percent_and_plus_both_decode() {
        assert_eq!(percent_decode("a%20b"), "a b");
        assert_eq!(percent_decode("%2Fsrc%2Flib.rs"), "/src/lib.rs");
        // A stray percent is left alone rather than eating the next bytes.
        assert_eq!(percent_decode("100%"), "100%");
        let q = parse_query("query=Fleet%3A%3Awritable+opens");
        assert_eq!(q.get("query").unwrap(), "Fleet::writable opens");
    }

    /// Two tokens from one process must differ, or a rotate would hand back
    /// the token it was asked to invalidate.
    #[test]
    fn every_token_is_new() {
        let a = new_token();
        let b = new_token();
        assert_ne!(a, b);
        assert_eq!(a.len(), 64, "256 bits, hex encoded");
        assert!(
            a.bytes()
                .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
            "not lowercase hex: {a}"
        );
    }

    /// The OS random source is the only source. A token minted from a fallback
    /// nobody noticed is the finding this replaced, so the failure is a panic
    /// naming what refused rather than a weaker token.
    #[test]
    fn a_token_is_never_minted_without_the_os_random_source() {
        let refused = std::panic::catch_unwind(|| token_from(Err(getrandom::Error::UNSUPPORTED)))
            .expect_err("a failing random source minted a token anyway");
        let message = refused
            .downcast_ref::<String>()
            .map(String::as_str)
            .or_else(|| refused.downcast_ref::<&str>().copied())
            .unwrap_or("");
        assert!(
            message.contains("the OS random source refused"),
            "the panic does not say what refused: {message}"
        );

        // The same bytes always encode the same way, so the hex half is not
        // where the entropy is.
        assert_eq!(token_from(Ok([0u8; 32])), "0".repeat(64));
        assert_eq!(&token_from(Ok([0xab; 32]))[..4], "abab");
    }

    /// A connection past the limit is answered 503 and closed cleanly, not
    /// reset. Closing a socket whose request was never read sends a reset on
    /// Windows and Linux, which a browser reports as ERR_CONNECTION_RESET and
    /// never shows the 503 (#181).
    #[test]
    fn a_connection_past_the_limit_is_answered_not_reset() {
        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
        let port = listener.local_addr().unwrap().port();
        let client = std::thread::spawn(move || {
            let mut c = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).unwrap();
            let body = vec![b'x'; 8 * 1024];
            let head = format!(
                "POST /api/x HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: {}\r\n\r\n",
                body.len()
            );
            c.write_all(head.as_bytes()).unwrap();
            c.write_all(&body).unwrap();
            c.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
            let mut answer = Vec::new();
            c.read_to_end(&mut answer).map(|_| answer)
        });
        let (stream, _) = listener.accept().unwrap();
        // Give the client's bytes time to land in the receive buffer, as they
        // have under a real flood.
        std::thread::sleep(Duration::from_millis(100));
        refuse_busy(stream);
        let answer = client.join().unwrap().expect("the client saw a reset");
        assert!(
            String::from_utf8_lossy(&answer).starts_with("HTTP/1.1 503"),
            "{}",
            String::from_utf8_lossy(&answer)
        );
    }

    /// A mistyped URL costs a quarter of a second; a run of guesses costs more
    /// each time, up to four seconds, and the window is a minute long.
    #[test]
    fn a_run_of_refusals_gets_slower() {
        let mut penalty = Penalty::default();
        assert_eq!(penalty.charge(), REFUSAL_DELAY, "the first refusal");
        for _ in 1..REFUSALS_PER_STEP {
            assert_eq!(penalty.charge(), REFUSAL_DELAY);
        }
        // The twenty-first in the window is where it starts doubling.
        assert_eq!(penalty.charge(), REFUSAL_DELAY * 2);
        for _ in 0..REFUSALS_PER_STEP {
            penalty.charge();
        }
        assert_eq!(penalty.charge(), REFUSAL_DELAY * 4);

        for _ in 0..200 {
            penalty.charge();
        }
        assert_eq!(
            penalty.charge(),
            MAX_REFUSAL_DELAY,
            "the delay is capped, because holding a connection forever is the \
             other kind of denial of service"
        );

        // A refusal a minute ago is not part of this run.
        let old = std::time::Instant::now() - Duration::from_secs(61);
        penalty.recent = std::iter::repeat_n(old, 500).collect();
        assert_eq!(penalty.charge(), REFUSAL_DELAY);
    }
}