packset_client/

lib.rs

//! Loopback HTTP client for packsetd.
//!
//! Reads `PACKSET_URL` or `INSIDE_MEMORY_URL`. search/get against packsetd; no SQLite.
//! Does not open LMDB.

use serde::{Deserialize, Serialize};
use std::env;
use std::error::Error as StdError;
use std::path::PathBuf;
use std::time::Duration;

/// How long one request may take: `PACKSET_TIMEOUT_MS`, else thirty seconds.
/// A write that waits behind others on a busy seat is late, not failed. The
/// same budget is sent as `X-Packset-Deadline`, and the writer drops a
/// request that is still waiting when it passes.
fn timeout() -> Duration {
    std::env::var("PACKSET_TIMEOUT_MS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|ms| *ms > 0)
        .map_or(Duration::from_secs(30), Duration::from_millis)
}

/// The wait before a down or busy writer is asked again, tripled each time,
/// and never longer than five seconds. The first waits are 50, 150 and 450 ms.
const BUSY_WAIT: Duration = Duration::from_millis(50);
/// Each wait is this many times the last.
const BUSY_GROWTH: u32 = 3;
/// The longest pause between asks.
const BUSY_WAIT_CAP: Duration = Duration::from_secs(5);

/// Whether another ask still ends inside `budget`.
///
/// `waited` is the sum of the waits already taken, not the clock. A round
/// trip that is slow on one host does not spend the next wait, so every
/// host asks the same number of times for one budget.
fn ask_again(waited: Duration, wait: Duration, budget: Duration) -> bool {
    waited.saturating_add(wait) < budget
}

/// How many times [`send`] asks inside `budget` when the writer sends no
/// `Retry-After`. The first ask costs no wait. Each later ask costs the
/// next backoff, and a wait that would pass the budget is not taken.
#[cfg(test)]
fn attempts_in(budget: Duration) -> u32 {
    let mut waited = Duration::ZERO;
    let mut wait = BUSY_WAIT;
    let mut attempts = 1u32;
    while ask_again(waited, wait, budget) {
        waited = waited.saturating_add(wait);
        wait = (wait * BUSY_GROWTH).min(BUSY_WAIT_CAP);
        attempts = attempts.saturating_add(1);
    }
    attempts
}

/// What [`send`] could not turn into a response.
enum Attempt {
    /// The writer is down or still busy after the budget. One sentence.
    Plain(String),
    /// Any other failure, for the caller to read.
    Ureq(Box<ureq::Error>),
}

/// How long to sleep, and how much of the budget that wait spends. The
/// charge is the full wait, so every host stops at the same attempt. The
/// sleep is half of that plus a fraction of the rest, so a herd does not
/// wake together. `Retry-After` raises the wait when it is longer.
struct Pause {
    sleep: Duration,
    charge: Duration,
}

fn planned_wait(backoff: Duration, retry_after: Option<Duration>) -> Duration {
    match retry_after {
        Some(after) if after > backoff => after,
        _ => backoff,
    }
}

/// Half of `wait`, plus up to the other half. `salt` picks the fraction.
fn jittered(wait: Duration, salt: u64) -> Duration {
    let ms = u64::try_from(wait.as_millis()).unwrap_or(u64::MAX).max(1);
    let half = ms / 2;
    let extra = if half == 0 { 0 } else { salt % (half + 1) };
    Duration::from_millis(half + extra)
}

/// The next wait, when it still ends inside `budget`.
fn pause_for(
    backoff: Duration,
    retry_after: Option<Duration>,
    waited: Duration,
    budget: Duration,
    salt: u64,
) -> Option<Pause> {
    let charge = planned_wait(backoff, retry_after);
    if !ask_again(waited, charge, budget) {
        return None;
    }
    let room = budget.saturating_sub(waited).max(Duration::from_millis(1));
    Some(Pause {
        sleep: jittered(charge, salt)
            .min(room)
            .max(Duration::from_millis(1)),
        charge,
    })
}

fn retry_after_of(header: Option<&str>) -> Option<Duration> {
    let raw = header?.trim();
    if raw.is_empty() {
        return None;
    }
    raw.parse::<u64>().ok().map(Duration::from_secs)
}

fn jitter_salt(asked: u32) -> u64 {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|elapsed| u64::from(elapsed.subsec_nanos()))
        .unwrap_or(0);
    nanos ^ u64::from(asked)
}

/// Send `req`, with `body` when there is one, inside `budget`. A connection
/// refused, a 429, or a 503 is asked again for every wait that fits in the
/// budget. A 503 or 429 that carries `Retry-After` waits at least that long.
/// The waits that are charged are what the budget spends. A slow round trip
/// does not drop an ask. The writer answers 503 before any of the request
/// runs, so a write sent again is not stored twice. Each ask carries
/// `X-Packset-Deadline` at the end of this budget. The first refusal of a
/// loopback writer starts the local daemon once, when this seat has a home
/// and a `packsetd` binary.
fn send(
    req: ureq::Request,
    body: Option<&serde_json::Value>,
    budget: Duration,
) -> Result<ureq::Response, Attempt> {
    let deadline = std::time::SystemTime::now() + budget;
    let deadline_ms = deadline
        .duration_since(std::time::UNIX_EPOCH)
        .map(|elapsed| elapsed.as_millis().to_string())
        .unwrap_or_default();
    let mut waited = Duration::ZERO;
    let mut wait = BUSY_WAIT;
    let mut asked = 0u32;
    let mut started_daemon = false;
    loop {
        asked = asked.saturating_add(1);
        let left = budget.saturating_sub(waited).max(Duration::from_millis(1));
        let mut attempt = req.clone().timeout(left);
        if !deadline_ms.is_empty() {
            attempt = attempt.set("X-Packset-Deadline", &deadline_ms);
        }
        if started_daemon {
            if let Some(token) = token() {
                attempt = attempt.set("Authorization", &format!("Bearer {token}"));
            }
        }
        let answered = match body {
            Some(json) => attempt.send_json(json),
            None => attempt.call(),
        };
        match answered {
            Err(ureq::Error::Status(code, response)) if code == 503 || code == 429 => {
                let after = retry_after_of(response.header("Retry-After"));
                match pause_for(wait, after, waited, budget, jitter_salt(asked)) {
                    Some(pause) => {
                        std::thread::sleep(pause.sleep);
                        waited = waited.saturating_add(pause.charge);
                        wait = (wait * BUSY_GROWTH).min(BUSY_WAIT_CAP);
                    }
                    None => return Err(Attempt::Plain(busy_line(response))),
                }
            }
            Err(err) if connect_refused(&err) => {
                match pause_for(wait, None, waited, budget, jitter_salt(asked)) {
                    Some(pause) => {
                        if !started_daemon {
                            started_daemon = autostart(req.url());
                        }
                        std::thread::sleep(pause.sleep);
                        waited = waited.saturating_add(pause.charge);
                        wait = (wait * BUSY_GROWTH).min(BUSY_WAIT_CAP);
                    }
                    None => {
                        return Err(Attempt::Plain("packsetd is not running.".to_string()));
                    }
                }
            }
            other => return other.map_err(|err| Attempt::Ureq(Box::new(err))),
        }
    }
}

/// The writer's own reason, as one sentence.
fn busy_line(response: ureq::Response) -> String {
    let text = response.into_string().unwrap_or_default();
    let reason = serde_json::from_str::<serde_json::Value>(&text)
        .ok()
        .and_then(|value| {
            value
                .get("error")
                .and_then(|reason| reason.as_str())
                .map(str::to_string)
        })
        .unwrap_or(text);
    let reason = reason.trim();
    if reason.is_empty() || reason.eq_ignore_ascii_case("busy") {
        "packsetd is busy.".to_string()
    } else if reason.ends_with('.') {
        format!("packsetd is busy: {reason}")
    } else {
        format!("packsetd is busy: {reason}.")
    }
}

/// Whether `err` is a TCP refusal. A reset or a timeout is a different failure.
fn connect_refused(err: &ureq::Error) -> bool {
    let ureq::Error::Transport(transport) = err else {
        return false;
    };
    if transport
        .source()
        .and_then(|source| source.downcast_ref::<std::io::Error>())
        .is_some_and(|io| io.kind() == std::io::ErrorKind::ConnectionRefused)
    {
        return true;
    }
    err.to_string().contains("Connection refused")
}

fn into_error(url: &str, err: Attempt) -> Error {
    match err {
        Attempt::Plain(text) => Error::Unavailable(text),
        Attempt::Ureq(err) => refused(url, err),
    }
}

fn path_seg(id: &str) -> String {
    let mut out = String::with_capacity(id.len());
    for b in id.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

/// The port a writer listens on when nothing names one. The command line,
/// the server and this client agree on it, so a seat needs no variable set.
pub const DEFAULT_PORT: u16 = 8761;

/// Load `~/.config/ljos/env` (KEY=VALUE) when the process has not set
/// those keys. `ljos`, `packset`, and `packset-mcp` then share one pack.
pub fn load_seat_env() {
    let Some(home) = env::var_os("HOME") else {
        return;
    };
    let path = PathBuf::from(home).join(".config/ljos/env");
    let Ok(text) = std::fs::read_to_string(path) else {
        return;
    };
    for line in text.lines() {
        let line = line.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        let Some((k, v)) = line.split_once('=') else {
            continue;
        };
        let k = k.trim();
        if k.is_empty() || env::var_os(k).is_some() {
            continue;
        }
        env::set_var(k, v.trim());
    }
}

/// The workspace the seat's memory lives in: `PACKSET_WORKSPACE` after
/// loading the seat env file, else `seat`. Not `default`, and not the
/// working directory's git remote: those two are how one harness's
/// remember missed the other's sitting.
#[must_use]
pub fn resolved_workspace() -> String {
    load_seat_env();
    env::var("PACKSET_WORKSPACE")
        .ok()
        .map(|w| w.trim().to_string())
        .filter(|w| !w.is_empty())
        .unwrap_or_else(|| "seat".to_string())
}

/// `PACKSET_PORT` (`LJOS_MEM_PORT` is an alias), else [`DEFAULT_PORT`].
#[must_use]
pub fn default_port() -> u16 {
    env::var("PACKSET_PORT")
        .or_else(|_| env::var("LJOS_MEM_PORT"))
        .ok()
        .and_then(|raw| raw.trim().parse().ok())
        .unwrap_or(DEFAULT_PORT)
}

/// The token file a writer leaves in its pack home: `PACKSET_TOKEN_FILE`,
/// else `token` under `PACKSET_HOME` (`LJOS_HOME`,
/// `LJOS_INSIDE_MEMORY_HOME`), else under the default home
/// ([`packset_core::home::resolve`]). packsetd resolves its home the same
/// way.
#[must_use]
pub fn token_path() -> Option<PathBuf> {
    if let Some(file) = env::var_os("PACKSET_TOKEN_FILE").filter(|v| !v.is_empty()) {
        return Some(PathBuf::from(file));
    }
    packset_core::home::resolve().map(|home| home.join(TOKEN_FILE))
}

/// The name of the token file in a pack home.
pub const TOKEN_FILE: &str = "token";

/// The token this seat shows the writer: `PACKSET_TOKEN`, else the first
/// line of [`token_path`]. `None` when there is neither, and the writer then
/// answers 401 to everything but `/health`.
#[must_use]
pub fn token() -> Option<String> {
    if let Ok(t) = env::var("PACKSET_TOKEN") {
        let t = t.trim().to_string();
        if !t.is_empty() {
            return Some(t);
        }
    }
    let text = std::fs::read_to_string(token_path()?).ok()?;
    let t = text.lines().next()?.trim().to_string();
    (!t.is_empty()).then_some(t)
}

/// Start the local writer once. `true` when a process was spawned.
///
/// A home has to be named (`PACKSET_HOME` or an older alias). The binary is
/// `PACKSET_BIN`, else `packsetd` beside this program, else on `PATH`.
fn autostart(url: &str) -> bool {
    use std::sync::atomic::{AtomicBool, Ordering};
    static STARTED: AtomicBool = AtomicBool::new(false);
    if STARTED.load(Ordering::Acquire) {
        return false;
    }
    let Some(port) = loopback_port(url) else {
        return false;
    };
    load_seat_env();
    let Some(home) = packset_core::home::named_home() else {
        return false;
    };
    let Some(bin) = daemon_binary() else {
        return false;
    };
    if STARTED.swap(true, Ordering::AcqRel) {
        return false;
    }
    spawn_daemon(&bin, &home, port)
}

/// `packsetd`, if this seat has one.
fn daemon_binary() -> Option<PathBuf> {
    if let Some(named) = env::var_os("PACKSET_BIN").filter(|value| !value.is_empty()) {
        let path = PathBuf::from(named);
        if path.is_file() {
            return Some(path);
        }
    }
    if let Some(beside) = env::current_exe()
        .ok()
        .and_then(|exe| exe.parent().map(|dir| dir.join("packsetd")))
        .filter(|path| path.is_file())
    {
        return Some(beside);
    }
    let paths = env::var_os("PATH")?;
    env::split_paths(&paths)
        .map(|dir| dir.join("packsetd"))
        .find(|path| path.is_file())
}

/// Spawn `bin` on `port` for `home`. The child leaves the caller's session
/// on Unix so the client's exit does not take the writer with it.
fn spawn_daemon(bin: &std::path::Path, home: &std::path::Path, port: u16) -> bool {
    use std::process::{Command, Stdio};
    let mut cmd = Command::new(bin);
    cmd.arg("--home")
        .arg(home)
        .arg("--port")
        .arg(port.to_string())
        .stdin(Stdio::null());
    if let Ok(log) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(home.join("packsetd.log"))
    {
        if let Ok(errs) = log.try_clone() {
            cmd.stdout(Stdio::from(log)).stderr(Stdio::from(errs));
        }
    } else {
        cmd.stdout(Stdio::null()).stderr(Stdio::null());
    }
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        // SAFETY: `pre_exec` runs in the forked child before exec. `setsid`
        // there detaches the writer from this client's session.
        unsafe {
            cmd.pre_exec(|| {
                extern "C" {
                    fn setsid() -> i32;
                }
                if setsid() == -1 {
                    return Err(std::io::Error::last_os_error());
                }
                Ok(())
            });
        }
    }
    cmd.spawn().is_ok()
}

/// `request` with this seat's token as a bearer header. A caller that speaks
/// to the writer outside [`PacksetClient`] wraps its request in this.
///
/// The token stays home when another user's process holds the writer's
/// loopback port: anyone can answer `/health` with `packsetd ok`, and a
/// listener that got there first would otherwise collect the token.
pub fn authorize(request: ureq::Request) -> ureq::Request {
    if foreign_listener(request.url()) {
        return request;
    }
    match token() {
        Some(t) => request.set("Authorization", &format!("Bearer {t}")),
        None => request,
    }
}

/// The port of a loopback `url`, `None` for any other host.
///
/// The foreign-listener check is Linux only. Starting a local writer uses
/// the same parse on every host.
fn loopback_port(url: &str) -> Option<u16> {
    let rest = url
        .strip_prefix("http://")
        .or_else(|| url.strip_prefix("https://"))?;
    let authority = rest.split(['/', '?', '#']).next()?;
    let (host, port) = match authority.rsplit_once(':') {
        Some((host, port)) if !port.contains(']') => (host, port.parse().ok()?),
        _ => (authority, 80),
    };
    matches!(host, "127.0.0.1" | "localhost" | "[::1]").then_some(port)
}

/// The owners of the sockets listening on `port`, from one `/proc/net/tcp`
/// or `tcp6` table: local address is the second column, state the fourth,
/// uid the eighth.
#[cfg(target_os = "linux")]
fn listen_uids(table: &str, port: u16) -> Vec<u32> {
    table
        .lines()
        .skip(1)
        .filter_map(|line| {
            let fields: Vec<&str> = line.split_whitespace().collect();
            let local = fields.get(1)?;
            if fields.get(3) != Some(&"0A") {
                return None;
            }
            let hex = local.rsplit(':').next()?;
            if u16::from_str_radix(hex, 16).ok()? != port {
                return None;
            }
            fields.get(7)?.parse().ok()
        })
        .collect()
}

/// Whether every socket listening on the loopback port of `url` belongs to
/// a user other than this one. Unknown is not foreign: off Linux, or when
/// the tables cannot be read, the token goes as before.
fn foreign_listener(url: &str) -> bool {
    #[cfg(target_os = "linux")]
    {
        use std::os::unix::fs::MetadataExt;
        let Some(port) = loopback_port(url) else {
            return false;
        };
        let Ok(me) = std::fs::metadata("/proc/self").map(|m| m.uid()) else {
            return false;
        };
        let owners: Vec<u32> = ["/proc/net/tcp", "/proc/net/tcp6"]
            .iter()
            .filter_map(|t| std::fs::read_to_string(t).ok())
            .flat_map(|text| listen_uids(&text, port))
            .collect();
        !owners.is_empty() && owners.iter().all(|uid| *uid != me)
    }
    #[cfg(not(target_os = "linux"))]
    {
        let _ = url;
        false
    }
}

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("packset url missing")]
    NoUrl,
    #[error("http: {0}")]
    Http(#[from] Box<ureq::Error>),
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("json: {0}")]
    Json(#[from] serde_json::Error),
    #[error("bad response: {0}")]
    Bad(String),
    /// The writer was down or busy for the whole budget. One sentence.
    #[error("{0}")]
    Unavailable(String),
}

#[derive(Debug, Clone)]
pub struct PacksetClient {
    base: String,
    /// A workspace pinned by the caller; `None` reads the environment and
    /// the working directory.
    workspace: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Hit {
    pub id: Option<String>,
    pub text: String,
    #[serde(default)]
    pub score: f64,
    #[serde(default)]
    pub kind: String,
    /// When the memory was written, the writer's clock, RFC 3339. Absent
    /// on card paragraphs, which have no clock.
    #[serde(default)]
    pub ts: Option<String>,
    /// How many of the panel's ballots named this hit, and how many ran.
    /// Two of three is agreement; one of three is one scorer's opinion.
    /// The atom's entities: `seat:<name>` names the seat that wrote it,
    /// `persona:<name>` a persona's own claim, `habit:<name>` a reading.
    #[serde(default)]
    pub entities: Vec<String>,
    #[serde(default)]
    pub ballots: Option<u32>,
    #[serde(default)]
    pub of: Option<u32>,
    /// The cross-encoder score when the second stage scored this hit.
    /// Absent, or null, when it did not.
    #[serde(default)]
    pub rerank_score: Option<f64>,
    /// Labels on the claim, `flag:credential-shaped` and the caller's own.
    /// Empty when the hit has none, including a writer that omits the field.
    #[serde(default)]
    pub labels: Vec<String>,
}

/// A refusal, carrying the reason the writer gave in its body.
fn refused(url: &str, e: Box<ureq::Error>) -> Error {
    match *e {
        ureq::Error::Status(code, response) => {
            let text = response.into_string().unwrap_or_default();
            let reason = serde_json::from_str::<serde_json::Value>(&text)
                .ok()
                .and_then(|v| v.get("error").and_then(|r| r.as_str()).map(str::to_string))
                .unwrap_or(text);
            let reason = reason.trim();
            if code == 401 {
                return Error::Bad(format!(
                    "{url}: 401: {reason}; this seat's token does not match the writer's, \
                     so it is another user's writer or PACKSET_HOME names another pack"
                ));
            }
            if reason.is_empty() {
                Error::Bad(format!("{url}: status code {code}"))
            } else {
                Error::Bad(format!("{url}: {code}: {reason}"))
            }
        }
        other => Error::Http(Box::new(other)),
    }
}

impl PacksetClient {
    pub fn new(base: impl Into<String>) -> Self {
        let mut base = base.into();
        while base.ends_with('/') {
            base.pop();
        }
        Self {
            base,
            workspace: None,
        }
    }

    /// Pin the workspace this client speaks for, ahead of `PACKSET_WORKSPACE`
    /// and the working directory. A seat that is one memory across every
    /// repository it works in sets this once.
    #[must_use]
    pub fn with_workspace(mut self, workspace: impl Into<String>) -> Self {
        let workspace = workspace.into();
        self.workspace = (!workspace.is_empty()).then_some(workspace);
        self
    }

    /// The writer the seat talks to, with nothing set: `PACKSET_URL`
    /// (`INSIDE_MEMORY_URL` is an alias), else the loopback port the command
    /// line starts a writer on, `PACKSET_PORT` (`LJOS_MEM_PORT`) or 8761.
    /// `PACKSET_URL=off` is the one way to have no pack.
    pub fn from_env() -> Result<Self, Error> {
        load_seat_env();
        let url = env::var("PACKSET_URL")
            .or_else(|_| env::var("INSIDE_MEMORY_URL"))
            .ok()
            .filter(|url| !url.is_empty());
        match url {
            Some(url) if url == "off" => Err(Error::NoUrl),
            Some(url) => Ok(Self::new(url)),
            None => Ok(Self::new(format!("http://127.0.0.1:{}", default_port()))),
        }
    }

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

    pub fn workspace(&self) -> String {
        if let Some(w) = &self.workspace {
            return w.clone();
        }
        if let Ok(w) = env::var("PACKSET_WORKSPACE") {
            if !w.is_empty() {
                return w;
            }
        }
        let cwd = env::var("HOSTSEAT_WORKSPACE")
            .ok()
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .map(std::path::PathBuf::from)
            .or_else(|| env::current_dir().ok())
            .unwrap_or_else(|| std::path::PathBuf::from("."));
        self.workspace_for_cwd(&cwd)
    }

    /// Workspace id from `/v1/identity` for `cwd`, or `dir:<abs>` if that call fails.
    pub fn workspace_for_cwd(&self, cwd: &std::path::Path) -> String {
        let abs = cwd.canonicalize().unwrap_or_else(|_| cwd.to_path_buf());
        let url = format!("{}/v1/identity", self.base);
        let body = send(
            authorize(ureq::get(&url)).query("cwd", abs.to_string_lossy().as_ref()),
            None,
            timeout(),
        )
        .ok()
        .and_then(|r| r.into_string().ok());
        if let Some(body) = body {
            if let Ok(val) = serde_json::from_str::<serde_json::Value>(&body) {
                if let Some(ws) = val.get("workspace").and_then(|v| v.as_str()) {
                    if !ws.is_empty() {
                        return ws.to_string();
                    }
                }
            }
        }
        format!("dir:{}", abs.display())
    }

    pub fn health(&self) -> Result<String, Error> {
        let url = format!("{}/health", self.base);
        let body = send(authorize(ureq::get(&url)), None, timeout())
            .map_err(|e| into_error(&url, e))?
            .into_string()?;
        Ok(body)
    }

    /// Queue depth, shed counts and the current limits. `line` is `ok`, or
    /// `busy, retrying` when work is waiting.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn stats(&self) -> Result<serde_json::Value, Error> {
        let url = format!("{}/stats", self.base);
        let body: serde_json::Value = send(authorize(ureq::get(&url)), None, timeout())
            .map_err(|e| into_error(&url, e))?
            .into_json()?;
        Ok(body)
    }

    pub fn get_atom(&self, workspace: &str, id: &str) -> Result<serde_json::Value, Error> {
        let encoded = path_seg(id);
        let url = format!("{}/v1/atoms/{encoded}", self.base);
        let resp = match send(
            authorize(ureq::get(&url)).query("workspace", workspace),
            None,
            timeout(),
        ) {
            Ok(resp) => resp,
            Err(Attempt::Ureq(e)) if matches!(*e, ureq::Error::Status(404, _)) => {
                return Err(Error::Bad(format!("no atom {id}")));
            }
            Err(e) => return Err(into_error(&url, e)),
        };
        Ok(resp.into_json()?)
    }

    pub fn list_atoms(&self, workspace: &str) -> Result<Vec<serde_json::Value>, Error> {
        self.atoms_as_of(workspace, None)
    }

    /// Live-now atoms, or the ones that were live at `as_of`.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn atoms_as_of(
        &self,
        workspace: &str,
        as_of: Option<&str>,
    ) -> Result<Vec<serde_json::Value>, Error> {
        let url = format!("{}/v1/atoms", self.base);
        let mut req = authorize(ureq::get(&url)).query("workspace", workspace);
        if let Some(at) = as_of {
            req = req.query("as_of", at);
        }
        let mut body: serde_json::Value = send(req, None, timeout())
            .map_err(|e| into_error(&url, e))?
            .into_json()?;
        let atoms = body
            .get_mut("atoms")
            .map(serde_json::Value::take)
            .unwrap_or(serde_json::Value::Array(vec![]));
        Ok(serde_json::from_value(atoms)?)
    }

    /// [`Self::atoms_as_of`] without the dense vectors, for a reader of
    /// texts, review clocks or rules: the vectors are most of the bytes. A
    /// writer older than the `embedding=omit` query sends them anyway.
    ///
    /// # Errors
    ///
    /// The pack not answering, or an answer that is not atoms.
    pub fn atoms_without_vectors(
        &self,
        workspace: &str,
        as_of: Option<&str>,
    ) -> Result<Vec<serde_json::Value>, Error> {
        let url = format!("{}/v1/atoms", self.base);
        let mut req = authorize(ureq::get(&url))
            .query("workspace", workspace)
            .query("embedding", "omit");
        if let Some(at) = as_of {
            req = req.query("as_of", at);
        }
        let mut body: serde_json::Value = send(req, None, timeout())
            .map_err(|e| into_error(&url, e))?
            .into_json()?;
        let atoms = body
            .get_mut("atoms")
            .map(serde_json::Value::take)
            .unwrap_or(serde_json::Value::Array(vec![]));
        Ok(serde_json::from_value(atoms)?)
    }

    pub fn search(&self, workspace: &str, q: &str, limit: u32) -> Result<Vec<Hit>, Error> {
        self.search_as_of(workspace, q, limit, None)
    }

    /// Ranked hits, optionally over the atoms that were live at `as_of`.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn search_as_of(
        &self,
        workspace: &str,
        q: &str,
        limit: u32,
        as_of: Option<&str>,
    ) -> Result<Vec<Hit>, Error> {
        self.search_opts(workspace, q, limit, as_of, false)
    }

    /// Ranked hits, optionally dated and optionally through the measured
    /// cross-encoder stage.
    ///
    /// Off by default. On, the writer spends a forward pass per candidate and
    /// the request waits for that rather than the usual five-second budget.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not a hit list.
    pub fn search_opts(
        &self,
        workspace: &str,
        q: &str,
        limit: u32,
        as_of: Option<&str>,
        rerank: bool,
    ) -> Result<Vec<Hit>, Error> {
        self.search_ranked(workspace, q, limit, as_of, rerank, None)
    }

    /// [`Self::search_opts`] with a rank. `Some("relevance")` asks for the
    /// top k by lexical, BM25 and cosine, with no half-life and no
    /// retrievability. `None` leaves the shipped order. An older writer
    /// ignores the parameter; `GET /v1/status` lists `search.ranks` when
    /// this writer honours it.
    ///
    /// # Errors
    ///
    /// The request's, including a rank the writer refuses, or a body that
    /// is not a hit list.
    pub fn search_ranked(
        &self,
        workspace: &str,
        q: &str,
        limit: u32,
        as_of: Option<&str>,
        rerank: bool,
        rank: Option<&str>,
    ) -> Result<Vec<Hit>, Error> {
        let url = format!("{}/v1/search", self.base);
        let budget = if rerank {
            Duration::from_secs(60).max(timeout())
        } else {
            timeout()
        };
        let mut req = authorize(ureq::get(&url))
            .query("workspace", workspace)
            .query("q", q)
            .query("limit", &limit.to_string());
        if let Some(at) = as_of {
            req = req.query("as_of", at);
        }
        if rerank {
            req = req.query("rerank", "1");
        }
        if let Some(rank) = rank.map(str::trim).filter(|s| !s.is_empty()) {
            req = req.query("rank", rank);
        }
        let body: serde_json::Value = send(req, None, budget)
            .map_err(|e| into_error(&url, e))?
            .into_json()?;
        let hits = body
            .get("hits")
            .cloned()
            .unwrap_or(serde_json::Value::Array(vec![]));
        Ok(serde_json::from_value(hits)?)
    }

    /// Seat home, atom counts by kind, pin, index and embedder.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn status(&self, workspace: Option<&str>) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/status", self.base);
        let mut req = authorize(ureq::get(&url));
        if let Some(workspace) = workspace {
            req = req.query("workspace", workspace);
        }
        Ok(send(req, None, timeout())
            .map_err(|e| into_error(&url, e))?
            .into_json()?)
    }

    /// The set a workspace is pinned to.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn pin(&self, workspace: &str) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/pin", self.base);
        Ok(send(
            authorize(ureq::get(&url)).query("workspace", workspace),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?)
    }

    /// Pin a workspace to a set.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn set_pin(&self, workspace: &str, name: &str) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/pin", self.base);
        Ok(send(
            authorize(ureq::put(&url)),
            Some(&serde_json::json!({ "workspace": workspace, "name": name })),
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?)
    }

    /// The deed accessions a workspace's live atoms cite, sorted.
    ///
    /// The accession is the only identifier crossing the tracker, the pack and
    /// the deed store, so this is what `deedar evidence -` reads.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn accessions(&self, workspace: &str) -> Result<Vec<String>, Error> {
        let url = format!("{}/v1/accessions", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::get(&url)).query("workspace", workspace),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        let found = body
            .get("accessions")
            .cloned()
            .unwrap_or(serde_json::Value::Array(vec![]));
        Ok(serde_json::from_value(found)?)
    }
    /// Every live atom in a workspace.
    ///
    /// The bodies, not the join keys: this is what a handover carries when
    /// somebody is given what the seat learned rather than only what it cites.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn atoms(&self, workspace: &str) -> Result<Vec<serde_json::Value>, Error> {
        self.atoms_as_of(workspace, None)
    }

    /// The live atoms of one kind in a workspace: personas, trust rows,
    /// habits, without the rest of the pack.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn atoms_of_kind(
        &self,
        workspace: &str,
        kind: &str,
    ) -> Result<Vec<serde_json::Value>, Error> {
        let url = format!("{}/v1/atoms", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::get(&url))
                .query("workspace", workspace)
                .query("kind", kind),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body
            .get("atoms")
            .and_then(serde_json::Value::as_array)
            .cloned()
            .unwrap_or_default())
    }

    /// The live atoms in a workspace that cite one deed accession.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn citers(
        &self,
        workspace: &str,
        accession: &str,
    ) -> Result<Vec<serde_json::Value>, Error> {
        let url = format!("{}/v1/citers", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::get(&url))
                .query("workspace", workspace)
                .query("accession", accession),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        let found = body
            .get("atoms")
            .cloned()
            .unwrap_or(serde_json::Value::Array(vec![]));
        Ok(serde_json::from_value(found)?)
    }

    /// Tombstone one atom. The daemon keeps the record and drops the index
    /// entry, so a forgotten atom stops being recalled without the pack losing
    /// the fact that it once held it.
    ///
    /// `why` is the deed accession that withdrew the claim, or the plain-text
    /// reason when the seat has no accession. An accession is stored as
    /// `retracted_by`. Plain text is stored as `why_text`, and the answer
    /// carries one `warning` line. The retraction and what it retracted read
    /// back together either way.
    ///
    /// # Errors
    ///
    /// [`Error::Bad`] when the workspace does not hold that atom, else the
    /// request's or a body that is not JSON.
    pub fn delete_atom(
        &self,
        workspace: &str,
        id: &str,
        why: Option<&str>,
    ) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/atoms/delete", self.base);
        let mut body = serde_json::json!({
            "workspace": workspace,
            "id": id,
        });
        if let Some(accession) = why {
            body["why"] = serde_json::Value::String(accession.to_string());
        }
        let resp = match send(authorize(ureq::post(&url)), Some(&body), timeout()) {
            Ok(resp) => resp,
            Err(Attempt::Ureq(e)) if matches!(*e, ureq::Error::Status(404, _)) => {
                return Err(Error::Bad(format!("no atom {id}")));
            }
            Err(e) => return Err(into_error(&url, e)),
        };
        Ok(resp.into_json()?)
    }

    /// Move one atom along the review clock: recalled, or lapsed.
    pub fn grade(
        &self,
        workspace: &str,
        id: &str,
        recalled: bool,
    ) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/grade", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::post(&url)),
            Some(&serde_json::json!({
                "workspace": workspace,
                "id": id,
                "recalled": recalled,
            })),
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    /// The link graph's communities, largest first.
    /// The claims the link graph turns on, highest first.
    pub fn hubs(&self, workspace: &str, limit: usize) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/hubs", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::get(&url))
                .query("workspace", workspace)
                .query("limit", &limit.to_string()),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    pub fn islands(&self, workspace: &str) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/islands", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::get(&url)).query("workspace", workspace),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    /// Claims that fired together: their links gain weight.
    pub fn fire(&self, workspace: &str, ids: &[String]) -> Result<serde_json::Value, Error> {
        self.fire_as(workspace, ids, None)
    }

    /// [`Self::fire`] through a persona's lens: the weights move under its
    /// name and the seat's stand.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn fire_as(
        &self,
        workspace: &str,
        ids: &[String],
        lens: Option<&str>,
    ) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/fire", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::post(&url)),
            Some(
                &serde_json::json!({"workspace": workspace, "ids": ids, "as": lens.unwrap_or("")}),
            ),
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    /// Consolidate the workspace: every claim that replaces an earlier one
    /// closes it (the write-time rule, run over what is held). `apply`
    /// false reports the pairs and writes nothing.
    pub fn consolidate(&self, workspace: &str, apply: bool) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/consolidate", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::post(&url)),
            Some(&serde_json::json!({"workspace": workspace, "apply": apply})),
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    /// Sweep a workspace for neglect: reviews left due past twice their
    /// interval lapse, and the third miss forgets a never-recalled lesson.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn sweep(&self, workspace: &str) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/sweep", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::post(&url)),
            Some(&serde_json::json!({"workspace": workspace})),
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    /// Drop a scratch workspace's live atoms, no deed; its tombstones stay.
    /// Retraction stays [`Self::delete_atom`]; this is what a per-run
    /// scratch workspace calls on its way out.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn forget_workspace(&self, workspace: &str) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/forget", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::post(&url)),
            Some(&serde_json::json!({"workspace": workspace})),
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body)
    }

    /// The memories a cue activates, strongest first; with `fire`, the top
    /// of them fire together.
    pub fn activate(
        &self,
        workspace: &str,
        q: &str,
        limit: u32,
        fire: bool,
    ) -> Result<serde_json::Value, Error> {
        self.activate_as(workspace, q, limit, fire, None)
    }

    /// [`Self::activate`] through a persona's lens: the spread follows the
    /// weights that persona wrote, and a fire writes them.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn activate_as(
        &self,
        workspace: &str,
        q: &str,
        limit: u32,
        fire: bool,
        lens: Option<&str>,
    ) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/activate", self.base);
        let mut req = authorize(ureq::get(&url))
            .query("workspace", workspace)
            .query("q", q)
            .query("limit", &limit.to_string())
            .query("fire", if fire { "1" } else { "0" });
        if let Some(name) = lens.filter(|n| !n.trim().is_empty()) {
            req = req.query("as", name);
        }
        let body: serde_json::Value = send(req, None, timeout())
            .map_err(|e| into_error(&url, e))?
            .into_json()?;
        Ok(body)
    }

    /// The live atoms of one named set in a workspace: a persona's own
    /// conclusions, a pinned slice.
    ///
    /// # Errors
    ///
    /// The request's, or a body that is not JSON.
    pub fn atoms_in_set(
        &self,
        workspace: &str,
        set: &str,
    ) -> Result<Vec<serde_json::Value>, Error> {
        let url = format!("{}/v1/pack", self.base);
        let body: serde_json::Value = send(
            authorize(ureq::get(&url))
                .query("workspace", workspace)
                .query("set", set),
            None,
            timeout(),
        )
        .map_err(|e| into_error(&url, e))?
        .into_json()?;
        Ok(body
            .get("atoms")
            .and_then(serde_json::Value::as_array)
            .cloned()
            .unwrap_or_default())
    }

    pub fn post_atom(&self, atom: &serde_json::Value) -> Result<serde_json::Value, Error> {
        let url = format!("{}/v1/atoms", self.base);
        let body: serde_json::Value = send(authorize(ureq::post(&url)), Some(atom), timeout())
            .map_err(|e| into_error(&url, e))?
            .into_json()?;
        Ok(body)
    }
}

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

    // Both workspace tests below rewrite process-global environment, so
    // they hold this lock: without it the runner's parallel threads
    // interleave one test's set-and-restore with the other's read, and
    // the suite fails one run in five on a polluted read.
    static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[test]
    fn only_a_loopback_url_has_a_port_to_check() {
        assert_eq!(loopback_port("http://127.0.0.1:8761/health"), Some(8761));
        assert_eq!(loopback_port("http://localhost:18462"), Some(18462));
        assert_eq!(loopback_port("http://[::1]:8761/v1/search?q=a"), Some(8761));
        assert_eq!(loopback_port("http://127.0.0.1/health"), Some(80));
        assert_eq!(loopback_port("http://example.org:8761/"), None);
        assert_eq!(loopback_port("http://[::1]/"), Some(80));
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn the_owner_of_a_listening_port_is_read_off_the_table() {
        let table = "  sl  local_address rem_address   st tx_queue rx_queue tr tm->when retrnsmt   uid  timeout inode
   0: 0100007F:223A 00000000:0000 0A 00000000:00000000 00:00000000 00000000  1001        0 4242 1 0 100 0 0 10 0
   1: 0100007F:223A 0100007F:9C40 01 00000000:00000000 00:00000000 00000000  1000        0 4343 1 0 20 4 30 10 -1
   2: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000  1000        0 4444 1 0 100 0 0 10 0
";
        assert_eq!(listen_uids(table, 8762), vec![1001], "only the listener");
        assert_eq!(listen_uids(table, 8080), vec![1000]);
        assert!(listen_uids(table, 9).is_empty());
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn a_port_this_user_holds_is_not_foreign() {
        let held = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = held.local_addr().unwrap().port();
        assert!(!foreign_listener(&format!(
            "http://127.0.0.1:{port}/health"
        )));
        drop(held);
        assert!(
            !foreign_listener(&format!("http://127.0.0.1:{port}/health")),
            "nobody listening is not foreign"
        );
    }

    #[test]
    fn resolved_workspace_reads_ljos_env_not_default() {
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let dir = std::env::temp_dir().join(format!("packset-ljos-env-{}", std::process::id()));
        std::fs::create_dir_all(dir.join(".config/ljos")).unwrap();
        std::fs::write(
            dir.join(".config/ljos/env"),
            "PACKSET_WORKSPACE=git:example.com/seat/notes\n",
        )
        .unwrap();
        let old_home = env::var("HOME").ok();
        let old_ws = env::var("PACKSET_WORKSPACE").ok();
        unsafe {
            env::remove_var("PACKSET_WORKSPACE");
            env::set_var("HOME", &dir);
        }
        let got = resolved_workspace();
        unsafe {
            match old_home {
                Some(h) => env::set_var("HOME", h),
                None => env::remove_var("HOME"),
            }
            match old_ws {
                Some(w) => env::set_var("PACKSET_WORKSPACE", w),
                None => env::remove_var("PACKSET_WORKSPACE"),
            }
        }
        assert_eq!(got, "git:example.com/seat/notes");
    }

    #[test]
    fn resolved_workspace_without_env_is_seat_not_default() {
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let dir = std::env::temp_dir().join(format!("packset-no-ljos-env-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let old_home = env::var("HOME").ok();
        let old_ws = env::var("PACKSET_WORKSPACE").ok();
        unsafe {
            env::remove_var("PACKSET_WORKSPACE");
            env::set_var("HOME", &dir);
        }
        let got = resolved_workspace();
        unsafe {
            match old_home {
                Some(h) => env::set_var("HOME", h),
                None => env::remove_var("HOME"),
            }
            match old_ws {
                Some(w) => env::set_var("PACKSET_WORKSPACE", w),
                None => env::remove_var("PACKSET_WORKSPACE"),
            }
        }
        assert_eq!(got, "seat");
        assert_ne!(got, "default");
    }

    /// A writer that answers busy to its first `busy` requests and counts every
    /// request. `retry_after` is the `Retry-After` value, in seconds, when
    /// the writer sends one.
    fn busy_for(
        busy: usize,
        retry_after: Option<u64>,
    ) -> (
        PacksetClient,
        std::sync::Arc<std::sync::atomic::AtomicUsize>,
    ) {
        use std::io::{BufRead, BufReader, Read, Write};
        use std::sync::atomic::Ordering;
        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
        let client = PacksetClient::new(format!("http://{}", listener.local_addr().unwrap()));
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counted = std::sync::Arc::clone(&seen);
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                let mut reader = BufReader::new(&stream);
                let mut length = 0;
                let mut line = String::new();
                while reader.read_line(&mut line).is_ok_and(|n| n > 2) {
                    if let Some(v) = line.to_ascii_lowercase().strip_prefix("content-length:") {
                        length = v.trim().parse().unwrap_or(0);
                    }
                    line.clear();
                }
                let mut body = vec![0u8; length];
                let _ = reader.read_exact(&mut body);
                let (status, text) = if counted.fetch_add(1, Ordering::SeqCst) < busy {
                    ("503 Service Unavailable", r#"{"error":"busy"}"#)
                } else {
                    ("200 OK", r#"{"ok":true}"#)
                };
                let retry = retry_after
                    .map(|secs| format!("Retry-After: {secs}\r\n"))
                    .unwrap_or_default();
                let _ = (&stream).write_all(
                    format!(
                        "HTTP/1.1 {status}\r\n{retry}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}",
                        text.len()
                    )
                    .as_bytes(),
                );
            }
        });
        (client, seen)
    }

    #[test]
    fn a_busy_writer_is_asked_again_for_a_read_and_a_write() {
        use std::sync::atomic::Ordering;
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let (client, seen) = busy_for(2, None);
        assert_eq!(client.status(None).unwrap()["ok"], true);
        assert_eq!(seen.load(Ordering::SeqCst), 3);
        let (client, seen) = busy_for(1, None);
        let stored = client
            .post_atom(&serde_json::json!({"text": "one write", "workspace": "w"}))
            .unwrap();
        assert_eq!(stored["ok"], true);
        assert_eq!(
            seen.load(Ordering::SeqCst),
            2,
            "the write was not sent again"
        );
    }

    #[test]
    fn a_writer_that_stays_busy_fails_in_one_sentence() {
        use std::sync::atomic::Ordering;
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let old = env::var_os("PACKSET_TIMEOUT_MS");
        unsafe { env::set_var("PACKSET_TIMEOUT_MS", "700") };
        let (client, seen) = busy_for(usize::MAX, None);
        let err = client.status(None).unwrap_err().to_string();
        unsafe {
            match old {
                Some(value) => env::set_var("PACKSET_TIMEOUT_MS", value),
                None => env::remove_var("PACKSET_TIMEOUT_MS"),
            }
        }
        assert_eq!(err, "packsetd is busy.", "{err}");
        assert!(
            !err.contains("os error") && !err.contains("status code"),
            "{err}"
        );
        // The count is the waits that fit in 700 ms, not how long this host
        // took to read each 503.
        assert_eq!(
            seen.load(Ordering::SeqCst),
            usize::try_from(attempts_in(Duration::from_millis(700))).unwrap(),
            "{err}"
        );
    }

    #[test]
    fn a_refused_connection_is_retried_and_then_one_sentence() {
        use std::time::Instant;
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let dir = std::env::temp_dir().join(format!("packset-down-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let saved = save_env(&[
            "HOME",
            "PACKSET_HOME",
            "LJOS_HOME",
            "LJOS_INSIDE_MEMORY_HOME",
            "PACKSET_BIN",
            "PACKSET_TIMEOUT_MS",
        ]);
        unsafe {
            env::set_var("HOME", &dir);
            env::remove_var("PACKSET_HOME");
            env::remove_var("LJOS_HOME");
            env::remove_var("LJOS_INSIDE_MEMORY_HOME");
            env::remove_var("PACKSET_BIN");
            env::set_var("PACKSET_TIMEOUT_MS", "250");
        }
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        drop(listener);
        let client = PacksetClient::new(format!("http://127.0.0.1:{port}"));
        let started = Instant::now();
        let err = client.status(None).unwrap_err().to_string();
        let took = started.elapsed();
        restore_env(saved);
        let _ = std::fs::remove_dir_all(&dir);
        assert_eq!(err, "packsetd is not running.", "{err}");
        assert!(took >= Duration::from_millis(40), "no backoff: {took:?}");
        assert!(took < Duration::from_secs(2), "{took:?}");
    }

    #[test]
    fn a_down_local_writer_is_started_once() {
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let dir = std::env::temp_dir().join(format!(
            "packset-autostart-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let home = dir.join("home");
        std::fs::create_dir_all(&home).unwrap();
        let log = dir.join("starts");
        let stub = dir.join("packsetd-stub");
        std::fs::write(
            &stub,
            format!(
                "#!/bin/sh\necho start >> \"{}\"\necho pid $$ >> \"{}\"\nexec python3 - \"$@\" << 'PY'\nimport sys\nargs = sys.argv[1:]\nport = int(args[args.index('--port') + 1])\nhome = args[args.index('--home') + 1]\nfrom http.server import BaseHTTPRequestHandler, HTTPServer\nclass Handler(BaseHTTPRequestHandler):\n    def do_GET(self):\n        body = b'{{\"ok\":true}}'\n        self.send_response(200)\n        self.send_header('Content-Type', 'application/json')\n        self.send_header('Content-Length', str(len(body)))\n        self.end_headers()\n        self.wfile.write(body)\n    def log_message(self, *args):\n        pass\nHTTPServer(('127.0.0.1', port), Handler).serve_forever()\nPY\n",
                log.display(),
                log.display(),
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut mode = std::fs::metadata(&stub).unwrap().permissions();
            mode.set_mode(0o755);
            std::fs::set_permissions(&stub, mode).unwrap();
        }
        let saved = save_env(&[
            "HOME",
            "PACKSET_HOME",
            "LJOS_HOME",
            "LJOS_INSIDE_MEMORY_HOME",
            "PACKSET_BIN",
            "PACKSET_TIMEOUT_MS",
            "PACKSET_TOKEN",
        ]);
        unsafe {
            env::set_var("HOME", &dir);
            env::set_var("PACKSET_HOME", &home);
            env::remove_var("LJOS_HOME");
            env::remove_var("LJOS_INSIDE_MEMORY_HOME");
            env::remove_var("PACKSET_TOKEN");
            env::set_var("PACKSET_BIN", &stub);
            env::set_var("PACKSET_TIMEOUT_MS", "5000");
        }
        let reap = Reap(log.clone());
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        drop(listener);
        let client = PacksetClient::new(format!("http://127.0.0.1:{port}"));
        let body = client.status(None).expect("the stub should answer");
        let text = std::fs::read_to_string(&log).unwrap_or_default();
        drop(reap);
        restore_env(saved);
        let _ = std::fs::remove_dir_all(&dir);
        assert_eq!(body["ok"], true, "{body}");
        assert_eq!(
            text.lines().filter(|line| *line == "start").count(),
            1,
            "{text}"
        );
    }

    fn save_env(keys: &[&str]) -> Vec<(String, Option<std::ffi::OsString>)> {
        keys.iter()
            .map(|key| ((*key).to_string(), env::var_os(key)))
            .collect()
    }

    fn restore_env(saved: Vec<(String, Option<std::ffi::OsString>)>) {
        for (key, value) in saved {
            unsafe {
                match value {
                    Some(value) => env::set_var(key, value),
                    None => env::remove_var(key),
                }
            }
        }
    }

    /// Kills every pid the autostart stub wrote, including when the test fails.
    struct Reap(std::path::PathBuf);

    impl Drop for Reap {
        fn drop(&mut self) {
            let Ok(text) = std::fs::read_to_string(&self.0) else {
                return;
            };
            for line in text.lines() {
                let Some(pid) = line.strip_prefix("pid ") else {
                    continue;
                };
                let _ = std::process::Command::new("kill")
                    .args(["-9", pid.trim()])
                    .status();
            }
        }
    }

    /// Room for the first wait and not the second: 50 ms fits in 100 ms,
    /// and 50 + 150 ms does not.
    const BUDGET: Duration = Duration::from_millis(100);

    /// A 100 ms budget leaves no room for the 150 ms wait. The time handed
    /// in is the waits already taken. A round trip does not move it.
    #[test]
    fn the_retries_stop_where_the_budget_does() {
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let old = env::var_os("PACKSET_TIMEOUT_MS");
        unsafe { env::set_var("PACKSET_TIMEOUT_MS", BUDGET.as_millis().to_string()) };
        let budget = timeout();
        unsafe {
            match old {
                Some(v) => env::set_var("PACKSET_TIMEOUT_MS", v),
                None => env::remove_var("PACKSET_TIMEOUT_MS"),
            }
        }
        assert_eq!(budget, BUDGET);
        let second = BUSY_WAIT * BUSY_GROWTH;
        assert!(
            ask_again(Duration::ZERO, BUSY_WAIT, budget),
            "the first wait did not fit"
        );
        assert!(!ask_again(BUSY_WAIT, second, budget), "the second wait fit");
        assert!(
            !ask_again(budget - BUSY_WAIT, BUSY_WAIT, budget),
            "a first round trip that spent the room was followed by a wait"
        );
    }

    /// 50 ms fits in 51 and not in 50. 50 + 150 + 450 fits in 700, and the
    /// next wait, 1350 ms, does not. The same numbers on every host.
    #[test]
    fn the_attempt_count_is_the_waits_that_fit_the_budget() {
        assert_eq!(attempts_in(Duration::from_millis(50)), 1);
        assert_eq!(attempts_in(Duration::from_millis(51)), 2);
        assert_eq!(attempts_in(Duration::from_millis(100)), 2);
        assert_eq!(attempts_in(Duration::from_millis(700)), 4);
        assert_eq!(attempts_in(Duration::from_secs(30)), 10);
    }

    /// `Retry-After` longer than the budget is not waited. The shorter
    /// backoff is not used in its place.
    #[test]
    fn a_retry_after_past_the_budget_is_not_waited() {
        let budget = Duration::from_millis(700);
        assert!(pause_for(
            BUSY_WAIT,
            Some(Duration::from_secs(1)),
            Duration::ZERO,
            budget,
            0
        )
        .is_none());
        let inside = pause_for(BUSY_WAIT, None, Duration::ZERO, budget, 0).unwrap();
        assert_eq!(inside.charge, BUSY_WAIT);
        assert!(inside.sleep >= BUSY_WAIT / 2 && inside.sleep <= BUSY_WAIT);
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let old = env::var_os("PACKSET_TIMEOUT_MS");
        unsafe { env::set_var("PACKSET_TIMEOUT_MS", "700") };
        let (client, seen) = busy_for(usize::MAX, Some(1));
        let err = client.status(None).unwrap_err().to_string();
        unsafe {
            match old {
                Some(value) => env::set_var("PACKSET_TIMEOUT_MS", value),
                None => env::remove_var("PACKSET_TIMEOUT_MS"),
            }
        }
        assert_eq!(err, "packsetd is busy.", "{err}");
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "{err}");
    }

    /// One second fits in a three-second budget, so the second ask waits
    /// for it. Jitter keeps the wait under that second and over half of it.
    #[test]
    fn a_retry_after_inside_the_budget_is_waited() {
        use std::time::Instant;
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let old = env::var_os("PACKSET_TIMEOUT_MS");
        unsafe { env::set_var("PACKSET_TIMEOUT_MS", "3000") };
        let (client, seen) = busy_for(1, Some(1));
        let started = Instant::now();
        let body = client.status(None).unwrap();
        let took = started.elapsed();
        unsafe {
            match old {
                Some(value) => env::set_var("PACKSET_TIMEOUT_MS", value),
                None => env::remove_var("PACKSET_TIMEOUT_MS"),
            }
        }
        assert_eq!(body["ok"], true, "{body}");
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
        assert!(
            took >= Duration::from_millis(400),
            "Retry-After was not waited: {took:?}"
        );
        assert!(took < Duration::from_secs(3), "{took:?}");
    }

    #[test]
    fn a_request_carries_the_end_of_its_budget() {
        use std::io::{BufRead, BufReader, Write};
        let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
        let client = PacksetClient::new(format!("http://{}", listener.local_addr().unwrap()));
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let (stream, _) = listener.accept().unwrap();
            let mut reader = BufReader::new(&stream);
            let mut deadline = String::new();
            let mut line = String::new();
            while reader.read_line(&mut line).is_ok_and(|n| n > 2) {
                let lower = line.to_ascii_lowercase();
                if let Some(value) = lower.strip_prefix("x-packset-deadline:") {
                    deadline = value.trim().to_string();
                }
                line.clear();
            }
            let _ = tx.send(deadline);
            let text = r#"{"ok":true}"#;
            let _ = (&stream).write_all(
                format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}",
                    text.len()
                )
                .as_bytes(),
            );
        });
        let before = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis();
        let body = client.status(None).unwrap();
        let header = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        let stamp: u128 = header.parse().unwrap();
        assert_eq!(body["ok"], true, "{body}");
        assert!(stamp > before, "{stamp} <= {before}");
        assert!(stamp < before + 60_000, "{stamp} is not this budget");
    }
}