//! `flow runner` (docs/m0.md §5): claim a job from OxygenHub, fetch the
//! repository into a cache with the job's token, write the state into a
//! workspace, run it in its tier, stream the log, report the receipt.
//! One job at a time.

use crate::home::{Home, RunnerConfig, RunnerKey};
use crate::{microvm, workspace};
use anyhow::{bail, Context, Result};
use flow_core::api::{Claim, ClaimRequest, Decline, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};
use flow_core::{scope_tree_hash, Receipt};
use oxy_core::hash::ChangeHash;
use oxy_core::repo::Repo;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

/// A request the hub refused (4xx). On a job's endpoints, 409 means the
/// job no longer runs: it was cancelled, or given up.
#[derive(Debug)]
pub struct Refused {
    pub code: u16,
    pub text: String,
}

impl std::fmt::Display for Refused {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "HTTP {}: {}", self.code, self.text)
    }
}

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

/// Whether `e` is the hub saying the job no longer runs.
fn job_gone(e: &anyhow::Error) -> bool {
    e.downcast_ref::<Refused>().is_some_and(|r| r.code == 409)
}

/// What a job ends with when its run was cancelled: nothing is reported,
/// as the hub has ended it already.
const CANCELLED: &str = "the run was cancelled";

fn cancelled() -> Finish {
    Finish {
        exit_code: None,
        duration_ms: 0,
        receipt: None,
        signed_receipt: None,
        error: Some(CANCELLED.into()),
        peak_disk_mib: None,
        peak_memory_mib: None,
        phases: Default::default(),
    }
}

pub struct Hub {
    base: String,
    token: String,
    agent: ureq::Agent,
}

impl Hub {
    pub fn new(cfg: &RunnerConfig) -> Result<Hub> {
        let tls = oxy_core::protocol::tls_client_config(cfg.ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;
        let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
        Ok(Hub { base: cfg.hub.trim_end_matches('/').to_string(), token: cfg.token.clone(), agent })
    }

    /// POST to the hub. A server error (5xx) means nothing was committed,
    /// so it is tried again after 1, 2 and 4 seconds before giving up: a
    /// busy hub costs a job a few seconds, not its log lines or its claim.
    fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
        self.post_with(path, body, &RETRY_DELAYS)
    }

    fn post_with(&self, path: &str, body: &impl serde::Serialize, delays: &[Duration]) -> Result<Option<ureq::Response>> {
        let url = format!("{}/api/v1{path}", self.base);
        let r = retrying(&url, delays, false, || {
            self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body).map_err(Box::new)
        })?;
        Ok(if r.status() == 204 { None } else { Some(r) })
    }

    /// POST what a job leaves behind (oxygenflow#49): through a hub that is
    /// away for a few minutes, as while a deploy restarts it, so a job's
    /// declines and uploads outlast it.
    fn post_patiently(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
        let url = format!("{}/api/v1{path}", self.base);
        let r = retrying(&url, &PATIENT_DELAYS, true, || {
            self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body).map_err(Box::new)
        })?;
        Ok(if r.status() == 204 { None } else { Some(r) })
    }

    pub fn claim(&self, req: &ClaimRequest) -> Result<Option<Claim>> {
        match self.post("/flow/claim", req)? {
            Some(r) => Ok(Some(r.into_json()?)),
            None => Ok(None),
        }
    }

    /// Hand job `job` back, saying why this runner will not run it.
    pub fn decline(&self, job: i64, reason: &str) -> Result<()> {
        self.post_patiently(&format!("/flow/jobs/{job}/decline"), &Decline { reason: reason.into() })?;
        Ok(())
    }

    /// Hand back a job this runner holds as it leaves (oxygenflow#76): the
    /// hub queues it again ahead of new work for another runner.
    pub fn release(&self, job: i64, reason: &str) -> Result<()> {
        self.post_patiently(&format!("/flow/jobs/{job}/release"), &Decline { reason: reason.into() })?;
        Ok(())
    }

    /// Say this runner is there while it claims nothing: its slots are full,
    /// or it is draining (oxygenhub#243). Once, no retries: the next comes
    /// soon. A hub without the endpoint (404) is not an error.
    /// A ten-minute certificate for `identity`, asked with a job's token
    /// (keyless, oxygenhub#241), and the identity of the hub's key that
    /// issued it, from `/.well-known/oxyhub-key`.
    pub fn certify(
        &self,
        job_token: &str,
        identity: &oxy_core::identity::Identity,
        proof: &oxy_core::identity::Signature,
    ) -> Result<(oxy_core::cert::SignedCertificate, oxy_core::identity::Identity)> {
        use oxy_core::dsse::{base64, base64_decode};
        let body = serde_json::json!({
            "identity": base64(&identity.encode()),
            "proof": base64(&oxy_core::cbor::encode(&proof.to_value())),
            "scope": "seal",
        });
        let url = format!("{}/api/v1/keyless/certify", self.base);
        let r = self
            .agent
            .post(&url)
            .set("Authorization", &format!("Bearer {job_token}"))
            .send_json(body)
            .map_err(|e| anyhow::anyhow!("asking the hub to certify the key: {e}"))?;
        let v: serde_json::Value = r.into_json()?;
        let cert = base64_decode(v["certificate"].as_str().unwrap_or(""))
            .context("the hub's answer has no certificate")
            .and_then(|b| oxy_core::cert::SignedCertificate::decode(&b).map_err(|e| anyhow::anyhow!("{e}")))?;
        let mut bytes = Vec::new();
        std::io::Read::read_to_end(
            &mut self.agent.get(&format!("{}/.well-known/oxyhub-key", self.base)).call()?.into_reader(),
            &mut bytes,
        )?;
        let issuer = oxy_core::identity::Identity::decode(&bytes)?;
        if issuer.hash() != cert.cert.issuer || cert.verify(&issuer).is_err() {
            bail!("the certificate is not signed by the hub's published key");
        }
        if cert.cert.key_id() != identity.hash() {
            bail!("the hub certified another key");
        }
        Ok((cert, issuer))
    }

    pub fn heartbeat(&self) -> Result<()> {
        match self.post_with("/flow/runners/heartbeat", &serde_json::json!({}), &[]) {
            Ok(_) => Ok(()),
            Err(e) if format!("{e:#}").contains("404") => Ok(()),
            Err(e) => Err(e),
        }
    }

    /// The update policy for this runner and the build it should run
    /// (oxygenhub#322); `None` from a hub without runner updates.
    pub fn update_answer(&self) -> Result<Option<flow_core::api::UpdateAnswer>> {
        let url = format!("{}/api/v1/flow/runner/update", self.base);
        match self.agent.get(&url).set("Authorization", &format!("Bearer {}", self.token)).call() {
            Ok(r) => Ok(Some(r.into_json()?)),
            Err(ureq::Error::Status(404, _)) => Ok(None),
            Err(e) => Err(anyhow::anyhow!("{url}: {e}")),
        }
    }

    /// Tell the hub what this runner did about its update.
    pub fn report_update(&self, note: &str) -> Result<()> {
        self.post_with("/flow/runner/update", &flow_core::api::UpdateReport { note: note.to_string() }, &[])?;
        Ok(())
    }

    /// Download an offered build to `to`, refusing it unless its BLAKE3
    /// is the one the hub attests and it is the size it said.
    pub fn download_build(&self, offer: &flow_core::api::UpdateOffer, to: &Path) -> Result<()> {
        use std::io::{Read, Write};
        let url = format!("{}{}", self.base, offer.url);
        let r = self.agent.get(&url).timeout(Duration::from_secs(1800)).call().map_err(|e| anyhow::anyhow!("{url}: {e}"))?;
        let part = to.with_extension("partial");
        let mut out = std::fs::File::create(&part).with_context(|| format!("writing {}", part.display()))?;
        let (mut hasher, mut n, mut buf) = (blake3::Hasher::new(), 0u64, vec![0u8; 1 << 16]);
        let mut body = r.into_reader().take(offer.size + 1);
        loop {
            let k = body.read(&mut buf)?;
            if k == 0 {
                break;
            }
            hasher.update(&buf[..k]);
            out.write_all(&buf[..k])?;
            n += k as u64;
        }
        drop(out);
        let got = hasher.finalize().to_hex().to_string();
        if n != offer.size || got != offer.blake3 {
            let _ = std::fs::remove_file(&part);
            bail!("{} does not match what the hub attests ({n} bytes, BLAKE3 {}…)", offer.asset, &got[..12]);
        }
        std::fs::rename(&part, to)?;
        Ok(())
    }

    pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {
        self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;
        Ok(())
    }

    /// Upload artifact `name` of job `job` (spec §5.2); the hub keeps it as
    /// a blob of the job's repository and answers with its ObjectRef.
    pub fn artifact(&self, job: i64, name: &str, bytes: &[u8]) -> Result<String> {
        let url = format!("{}/api/v1/flow/jobs/{job}/artifacts/{name}", self.base);
        let r = retrying(&url, &PATIENT_DELAYS, true, || {
            self.agent
                .post(&url)
                .timeout(Duration::from_secs(600))
                .set("Authorization", &format!("Bearer {}", self.token))
                .set("Content-Type", "application/octet-stream")
                .send_bytes(bytes)
                .map_err(Box::new)
        })?;
        let v: serde_json::Value = r.into_json()?;
        v["blob"].as_str().map(String::from).context("the hub named no blob")
    }

    /// Keep a checkpoint of job `job` (oxygenflow#77) as its runner hands
    /// it back: the hub answers with its blob, which the next attempt's
    /// claim names.
    pub fn save_checkpoint(&self, job: i64, bytes: &[u8]) -> Result<String> {
        let url = format!("{}/api/v1/flow/jobs/{job}/checkpoint", self.base);
        let r = retrying(&url, &PATIENT_DELAYS, true, || {
            self.agent
                .post(&url)
                .timeout(Duration::from_secs(600))
                .set("Authorization", &format!("Bearer {}", self.token))
                .set("Content-Type", "application/octet-stream")
                .send_bytes(bytes)
                .map_err(Box::new)
        })?;
        let v: serde_json::Value = r.into_json()?;
        v["blob"].as_str().map(String::from).context("the hub named no blob")
    }

    /// The checkpoint an earlier attempt of job `job` saved (oxygenflow#77),
    /// which the caller checks against the blob its claim names.
    pub fn checkpoint(&self, job: i64) -> Result<Vec<u8>> {
        let url = format!("{}/api/v1/flow/jobs/{job}/checkpoint", self.base);
        let r = retrying(&url, &PATIENT_DELAYS, false, || {
            self.agent
                .get(&url)
                .timeout(Duration::from_secs(600))
                .set("Authorization", &format!("Bearer {}", self.token))
                .call()
                .map_err(Box::new)
        })?;
        let mut bytes = Vec::new();
        std::io::Read::read_to_end(&mut std::io::Read::take(r.into_reader(), crate::checkpoint::MAX_BYTES as u64 + 1), &mut bytes)?;
        Ok(bytes)
    }

    /// Input `index` of job `job` (oxygenflow#58): the bytes of the
    /// producing job's artifact, which the caller checks against the blob the
    /// claim names.
    pub fn input(&self, job: i64, index: usize) -> Result<Vec<u8>> {
        let url = format!("{}/api/v1/flow/jobs/{job}/inputs/{index}", self.base);
        let r = retrying(&url, &PATIENT_DELAYS, false, || {
            self.agent
                .get(&url)
                .timeout(Duration::from_secs(600))
                .set("Authorization", &format!("Bearer {}", self.token))
                .call()
                .map_err(Box::new)
        })?;
        let mut bytes = Vec::new();
        let mut reader = r.into_reader();
        std::io::Read::read_to_end(&mut std::io::Read::take(&mut reader, MAX_INPUT_BYTES as u64 + 1), &mut bytes)?;
        if bytes.len() > MAX_INPUT_BYTES {
            bail!("larger than {} MiB", MAX_INPUT_BYTES >> 20);
        }
        Ok(bytes)
    }

    /// Upload file `path` of report `name` of job `job` (oxygenhub#127);
    /// the hub keeps it as a blob and answers with its ObjectRef. The
    /// report's manifest follows through [`Hub::artifact`].
    pub fn report_file(&self, job: i64, name: &str, path: &str, bytes: &[u8]) -> Result<String> {
        let url = format!("{}/api/v1/flow/jobs/{job}/reports/{name}/files", self.base);
        let r = retrying(&url, &PATIENT_DELAYS, true, || {
            self.agent
                .post(&url)
                .timeout(Duration::from_secs(600))
                .set("Authorization", &format!("Bearer {}", self.token))
                .set("Content-Type", "application/octet-stream")
                .set("X-Flow-Report-Path", path)
                .send_bytes(bytes)
                .map_err(Box::new)
        })?;
        let v: serde_json::Value = r.into_json()?;
        v["blob"].as_str().map(String::from).context("the hub named no blob")
    }

    /// Report a job's end until the hub has it (oxygenhub#71): while the hub
    /// cannot be reached or fails (5xx), try again after `first_wait`,
    /// doubling to a minute, for up to `patience`. A refusal (4xx: the job was ended
    /// already, say) is final. A result lost here leaves the job running
    /// on the hub until its time limit.
    pub fn finish_until_accepted(&self, job: i64, f: &Finish, patience: Duration, first_wait: Duration) -> Result<()> {
        let url = format!("{}/api/v1/flow/jobs/{job}/finish", self.base);
        let deadline = std::time::Instant::now() + patience;
        let mut wait = first_wait;
        loop {
            let why = match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(f) {
                Ok(_) => return Ok(()),
                Err(ureq::Error::Status(code, r)) if code < 500 => {
                    bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default())
                }
                Err(ureq::Error::Status(code, _)) => format!("HTTP {code}"),
                Err(e) => e.to_string(),
            };
            if std::time::Instant::now() + wait > deadline {
                bail!("{url}: gave up after {} minutes: {why}", patience.as_secs() / 60);
            }
            eprintln!("job {job}: the hub did not take the result ({why}); trying again in {} s", wait.as_secs());
            std::thread::sleep(wait);
            wait = (wait * 2).min(Duration::from_secs(60));
        }
    }
}

/// How long a runner keeps trying to report a job's end.
const FINISH_PATIENCE: Duration = Duration::from_secs(2 * 3600);

/// Where a job's result waits until the hub has it, so a runner that
/// restarts still reports it.
fn pending_path(home: &Home, job: i64) -> PathBuf {
    home.root.join("pending").join(format!("{job}.json"))
}

/// Report `f`, keeping it on disk until the hub has it.
fn report_finish(home: &Home, hub: &Hub, job: i64, f: &Finish) {
    let path = pending_path(home, job);
    let kept = std::fs::create_dir_all(path.parent().unwrap())
        .and_then(|_| std::fs::write(&path, serde_json::to_vec(f).unwrap_or_default()))
        .is_ok();
    match hub.finish_until_accepted(job, f, FINISH_PATIENCE, Duration::from_secs(5)) {
        Ok(()) => {}
        Err(e) => eprintln!("job {job}: reporting failed: {e:#}"),
    }
    if kept {
        let _ = std::fs::remove_file(&path);
    }
}

/// Results a previous run of the runner could not report, to send again.
fn report_pending(home: &Home) -> Vec<(i64, Finish)> {
    let Ok(dir) = std::fs::read_dir(home.root.join("pending")) else {
        return Vec::new();
    };
    dir.filter_map(|e| {
        let p = e.ok()?.path();
        let job: i64 = p.file_stem()?.to_str()?.parse().ok()?;
        let f: Finish = serde_json::from_slice(&std::fs::read(&p).ok()?).ok()?;
        Some((job, f))
    })
    .collect()
}

/// `flow runner register`: exchange a one-time registration token from
/// the hub for this runner's credential, binding the runner's signing key,
/// and keep hub, credential and name in runner.toml.
pub fn register(home: &Home, hub: &str, token: &str, name: &str, ca: Option<PathBuf>, labels: Vec<String>) -> Result<Registered> {
    if home.config.runner.is_some() && !home.root.join(crate::home::RUNNER_TOML).is_file() {
        bail!("flow.toml already has a [runner] section; remove it to register this runner instead");
    }
    let key = home.runner_key(name)?;
    let hub = hub.trim_end_matches('/').to_string();
    let tls = oxy_core::protocol::tls_client_config(ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;
    let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
    let req = Register {
        token: token.trim().to_string(),
        name: name.to_string(),
        identity: hex::encode(key.identity.encode()),
        info: RunnerInfo { version: env!("OXY_VERSION_LONG").into(), ..RunnerInfo::here(labels.clone(), 1, None, None) },
    };
    let url = format!("{hub}/api/v1/flow/runners/register");
    let got: Registered = match agent.post(&url).send_json(&req) {
        Ok(r) => r.into_json()?,
        Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
        Err(e) => bail!("{url}: {e}"),
    };
    let mut text = format!(
        "# Written by `flow runner register`; the credential is secret.\nhub = {hub:?}\ntoken = {:?}\nname = {:?}\n",
        got.credential, got.name
    );
    if let Some(ca) = &ca {
        let kept = keep_ca(&home.root, ca)?;
        text.push_str(&format!("ca = {:?}\n", kept.display().to_string()));
    }
    if !labels.is_empty() {
        text.push_str(&format!("labels = {labels:?}\n"));
    }
    let path = home.root.join(crate::home::RUNNER_TOML);
    {
        use std::io::Write;
        let tmp = path.with_extension("toml.tmp");
        let mut f = crate::portable::private(std::fs::OpenOptions::new().write(true).create(true).truncate(true)).open(&tmp)?;
        f.write_all(text.as_bytes())?;
        f.sync_all()?;
        std::fs::rename(&tmp, &path)?;
    }
    Ok(got)
}

/// Copy the hub's CA certificate `ca` into the Flow home `root` (#98) and
/// give the copy's absolute path, for runner.toml: a service starts in
/// another directory, and the file given may move or go.
fn keep_ca(root: &Path, ca: &Path) -> Result<PathBuf> {
    let kept = root.join(crate::home::CA_PEM);
    let same = matches!((std::fs::canonicalize(ca), std::fs::canonicalize(&kept)), (Ok(a), Ok(b)) if a == b);
    if !same {
        std::fs::copy(ca, &kept).with_context(|| format!("copy {} into {}", ca.display(), root.display()))?;
    }
    Ok(std::fs::canonicalize(&kept)?)
}

/// Waits before trying a hub request again after a server error.
const RETRY_DELAYS: [Duration; 3] = [Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(4)];

/// Waits for what a job leaves behind (oxygenflow#49): three minutes in
/// all, longer than a hub takes to restart.
/// Check an input's bytes against the blob the producing job's receipt
/// names, then write them at its path under `tree` (the workspace's `/src`).
fn place_input(tree: &Path, i: &flow_core::api::ArtifactInput, bytes: &[u8]) -> Result<[u8; 32]> {
    let Some(rel) = i.path.strip_prefix("/src/").filter(|r| r.split('/').all(|p| !p.is_empty() && p != "." && p != ".." && p != ".oxy"))
    else {
        bail!("path {:?} is not a clean path under /src", i.path);
    };
    let blob = Repo::blob_ref(bytes);
    if blob.to_hex() != i.blob {
        bail!("it is not what {}'s receipt names (blob {}, not {})", i.from, &blob.to_hex()[..12], &i.blob[..i.blob.len().min(12)]);
    }
    let dest = tree.join(rel);
    if let Some(dir) = dest.parent() {
        std::fs::create_dir_all(dir)?;
    }
    std::fs::write(&dest, bytes)?;
    Ok(blob.0)
}

/// The largest input a job takes: the hub's artifact limit.
pub const MAX_INPUT_BYTES: usize = 512 << 20;

const PATIENT_DELAYS: [Duration; 15] = {
    let mut d = [Duration::from_secs(15); 15];
    d[0] = Duration::from_secs(1);
    d[1] = Duration::from_secs(2);
    d[2] = Duration::from_secs(4);
    d[3] = Duration::from_secs(8);
    d
};

/// Send a request until the hub answers, waiting `delays` in turn after a
/// server error (5xx: nothing was committed) and, when `connection`, after
/// the hub could not be reached at all. A refusal (4xx) is final, as a
/// [`Refused`] the caller can look at.
fn retrying(
    url: &str,
    delays: &[Duration],
    connection: bool,
    mut send: impl FnMut() -> Result<ureq::Response, Box<ureq::Error>>,
) -> Result<ureq::Response> {
    let mut delays = delays.iter();
    loop {
        let why = match send().map_err(|e| *e) {
            Ok(r) => return Ok(r),
            Err(ureq::Error::Status(code, r)) if code < 500 => {
                let text = r.into_string().unwrap_or_default();
                return Err(anyhow::Error::new(Refused { code, text }).context(url.to_string()));
            }
            Err(ureq::Error::Status(code, r)) => format!("HTTP {code}: {}", r.into_string().unwrap_or_default()),
            Err(e) if connection => e.to_string(),
            Err(e) => bail!("{url}: {e}"),
        };
        match delays.next() {
            Some(d) => std::thread::sleep(*d),
            None => bail!("{url}: {why}"),
        }
    }
}

/// What the runner may run at once (flow.toml `[runner]`): at most
/// `max_jobs` jobs, whose sandboxes together fit `memory_mib` and `cpus`.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Budget {
    pub max_jobs: usize,
    pub memory_mib: Option<u32>,
    pub cpus: Option<u32>,
}

/// What is in use now.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Used {
    pub jobs: usize,
    pub memory_mib: u32,
    pub cpus: u32,
}

impl Budget {
    pub fn of(cfg: &RunnerConfig) -> Budget {
        Budget { max_jobs: cfg.max_jobs.unwrap_or(1).max(1), memory_mib: cfg.memory_mib, cpus: cfg.cpus }
    }

    /// The room to ask the hub for, or `None` when no job more can start.
    /// A lone job may use the whole budget even if it asks for more, so a
    /// large sandbox is never starved.
    pub fn room(&self, used: Used) -> Option<(Option<u32>, Option<u32>)> {
        if used.jobs >= self.max_jobs {
            return None;
        }
        if used.jobs == 0 {
            return Some((None, None));
        }
        let mem = self.memory_mib.map(|m| m.saturating_sub(used.memory_mib));
        let cpus = self.cpus.map(|c| c.saturating_sub(used.cpus));
        if mem == Some(0) || cpus == Some(0) {
            return None;
        }
        Some((mem, cpus))
    }
}

/// The isolation tiers this machine can honour now (spec §11.1): the
/// MicroVM tier when `/dev/kvm` opens for reading and writing and
/// flow.toml pins Firecracker; the syscall tier when this binary may use
/// user namespaces, Landlock and seccomp and manage a delegated cgroup.
/// `Err` says why there are none; the reasons for missing tiers come back
/// with the ones there are.
pub fn tiers(home: &Home) -> std::result::Result<(Vec<String>, Vec<String>), String> {
    let mut have = Vec::new();
    let mut missing = Vec::new();
    let kvm = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm");
    match (kvm, home.config.microvm.is_some()) {
        _ if !cfg!(target_os = "linux") => missing.push("microvm: the tier runs only on Linux (Firecracker and KVM)".to_string()),
        (Ok(_), true) => have.push("microvm".to_string()),
        (Ok(_), false) => missing.push("microvm: flow.toml has no [microvm] section".to_string()),
        (Err(e), _) => missing.push(format!(
            "microvm: /dev/kvm cannot be opened ({e}); the tier needs hardware virtualisation and this user in the kvm group"
        )),
    }
    match crate::syscall::available() {
        Ok(()) => have.push("syscall".to_string()),
        Err(e) => missing.push(format!("syscall: {e}")),
    }
    // wasmtime runs on every machine this binary does (oxygenflow#75).
    have.push("wasm".to_string());
    if have.is_empty() {
        return Err(format!("no isolation tier can run here: {}", missing.join("; ")));
    }
    Ok((have, missing))
}

/// Fetch `claim`'s image from the hub when this home lacks it
/// (oxygenflow#39), saying so in the job's log. A hub that does not hold it
/// leaves it missing, and the preflight declines the job as before.
fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) -> Option<u64> {
    let hash = &claim.sandbox.rootfs;
    if hash.is_empty() || home.rootfs(hash).is_file() || !jet::images::is_hash(hash) {
        return None;
    }
    let note = |text: String| {
        eprintln!("job {}: {text}", claim.job_id);
        hub.log(claim.job_id, vec![Record { t: 0, stream: "flow".into(), text }]).ok();
    };
    // One fetch of an image at a time: two jobs wanting it wait for one.
    static FETCHING: std::sync::Mutex<()> = std::sync::Mutex::new(());
    let _one = FETCHING.lock().unwrap_or_else(|p| p.into_inner());
    if home.rootfs(hash).is_file() {
        return None;
    }
    let client = match crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref()) {
        Ok(c) => c,
        Err(e) => {
            note(format!("image {}: cannot reach the hub: {e:#}", &hash[..12]));
            return None;
        }
    };
    let fetching = Instant::now();
    let fetched = match swarm {
        Some(sw) => {
            note(format!("image {} is not on this runner: fetching it from its peers and the hub", &hash[..12]));
            sw.fetch(home, hash)
        }
        None => {
            note(format!("image {} is not on this runner: fetching it from the hub", &hash[..12]));
            client.fetch(home, hash)
        }
    };
    let took = fetching.elapsed().as_millis() as u64;
    match fetched {
        Ok(Some(m)) => note(format!(
            "image {}: fetched {} chunks, {}{} in {:.1} s",
            &hash[..12],
            m.chunks,
            crate::images::human(m.bytes),
            if swarm.is_some() { format!(" ({} from peers)", crate::images::human(m.peer_bytes)) } else { String::new() },
            m.elapsed.as_secs_f64()
        )),
        Ok(None) => note(format!("image {}: the hub does not hold it either", &hash[..12])),
        Err(e) => note(format!("image {}: fetch failed: {e:#}", &hash[..12])),
    }
    Some(took)
}

/// Why this runner cannot run `claim` now (spec §11.5), checked before
/// anything is fetched: the tier, the architecture, the image and the
/// memory. The hub chose it from what the runner last said, which may be
/// out of date.
fn preflight(home: &Home, claim: &Claim) -> Option<String> {
    let sb = &claim.sandbox;
    match sb.runtime {
        flow_core::Runtime::Microvm => {
            if let Err(e) = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm") {
                return Some(format!("the MicroVM tier needs /dev/kvm, which cannot be opened here ({e})"));
            }
            if home.config.microvm.is_none() {
                return Some("the MicroVM tier is not set up here (flow.toml has no [microvm] section)".into());
            }
        }
        flow_core::Runtime::Syscall => {
            if let Err(e) = crate::syscall::available() {
                return Some(format!("the syscall tier cannot run here: {e}"));
            }
        }
        // No image and no architecture: only the memory matters.
        flow_core::Runtime::Wasm => {
            return available_memory_mib()
                .filter(|free| *free < u64::from(sb.memory_mib))
                .map(|free| format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));
        }
    }
    sandbox_problem(home, sb, std::env::consts::ARCH, available_memory_mib())
}

/// What stops `sb` running on this home's images, on a machine of
/// architecture `here` with `free_mib` of memory free.
fn sandbox_problem(home: &Home, sb: &flow_core::Sandbox, here: &str, free_mib: Option<u64>) -> Option<String> {
    if let Some(want) = sb.arch.as_deref().filter(|a| *a != here) {
        return Some(format!("the sandbox needs {want}; this runner is {here}"));
    }
    let short = &sb.rootfs[..12.min(sb.rootfs.len())];
    let image = home.rootfs(&sb.rootfs);
    if !image.is_file() {
        return Some(format!("rootfs {short} is not on this runner"));
    }
    if let Some(arch) = home.image_arch(&sb.rootfs).filter(|a| a != here) {
        return Some(format!("rootfs {short} is for {arch}; this runner is {here}"));
    }
    if let Err(e) = home.verified(&image, &sb.rootfs) {
        return Some(format!("rootfs {short} does not verify here: {e:#}"));
    }
    if let Some(free) = free_mib.filter(|free| *free < u64::from(sb.memory_mib)) {
        return Some(format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));
    }
    None
}

/// The memory to claim with: the budget's room, never more than the host
/// has free now. An idle runner's budget is unlimited (a lone job may use
/// it all), so without this it was offered sandboxes it then declined for
/// want of memory, one after another, having fetched their images first
/// (an 8 GiB job on a 7 GB machine).
fn claim_memory(room: Option<u32>, host_free_mib: Option<u64>) -> Option<u32> {
    match (room, host_free_mib) {
        (Some(r), Some(h)) => Some(r.min(u32::try_from(h).unwrap_or(u32::MAX))),
        (None, Some(h)) => Some(u32::try_from(h).unwrap_or(u32::MAX)),
        (room, None) => room,
    }
}

/// `MemAvailable` from /proc/meminfo, in MiB.
fn available_memory_mib() -> Option<u64> {
    let info = std::fs::read_to_string("/proc/meminfo").ok()?;
    let kib: u64 = info.lines().find_map(|l| l.strip_prefix("MemAvailable:"))?.trim().trim_end_matches("kB").trim().parse().ok()?;
    Some(kib / 1024)
}

pub fn serve(home: &Home) -> Result<()> {
    drain_on_stop();
    crate::update::note_runner(&home.root);
    // `flow runner update` swapped a new build in and asks this one to
    // leave with the notice it gave (#85).
    #[cfg(unix)]
    {
        let handler = on_update as extern "C" fn(libc::c_int) as libc::sighandler_t;
        unsafe { libc::signal(libc::SIGUSR1, handler) };
    }
    if let Some(p) = crate::update::pending(&home.root) {
        // On trial: a build that does not reach the hub in time goes back.
        let root = home.root.clone();
        std::thread::spawn(move || {
            std::thread::sleep(crate::update::PROVE_WITHIN);
            if crate::update::pending(&root).is_some_and(|now| now.to == p.to && now.at_ms == p.at_ms) {
                let why = format!("did not reach the hub within {} minutes", crate::update::PROVE_WITHIN.as_secs() / 60);
                if crate::update::roll_back(&root, &p, &why).is_ok() {
                    std::process::exit(1);
                }
            }
        });
    }
    let served = serve_until_drained(home);
    crate::update::forget_runner(&home.root);
    // Left for a new build: on Windows, end as a failure, so a Task
    // Scheduler task set to restart on failure starts it (systemd's
    // Restart=always starts it after a clean exit).
    if cfg!(windows) && served.is_ok() && LEFT_FOR_UPDATE.load(std::sync::atomic::Ordering::SeqCst) {
        eprintln!("flow runner: left for the new build; exiting 75 so it is started");
        std::process::exit(75);
    }
    served
}

/// Set once the runner leaves for a new build.
static LEFT_FOR_UPDATE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

/// Set by SIGUSR1: a new build is in place; leave for it.
static UPDATE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

#[cfg(unix)]
extern "C" fn on_update(_: libc::c_int) {
    UPDATE.store(true, std::sync::atomic::Ordering::SeqCst);
}

/// The runner as a library, as the Android app runs it (oxygenhub-android#8):
/// no signal handlers; [`stop`] drains it, and this returns once the jobs
/// it holds are done and reported. It may be started again afterwards.
pub fn serve_embedded(home: &Home) -> Result<()> {
    STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = None;
    serve_until_drained(home)
}

/// A departure ([`leave_within`]): why, and when the runner must be gone.
#[derive(Clone, Debug)]
struct Leave {
    reason: String,
    by: Instant,
}

/// Set once the runner is leaving: a job whose expected rest fits before
/// the deadline finishes; the others are stopped and handed back
/// (oxygenflow#76).
static LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);

fn leaving() -> Option<String> {
    LEAVING.lock().unwrap_or_else(|p| p.into_inner()).as_ref().map(|l| l.reason.clone())
}

/// Ask a runner started with [`serve_embedded`] to leave now: claim nothing
/// more, stop the jobs it holds and hand them back to the hub, which queues
/// them ahead of new work for another runner (oxygenflow#76), then return.
/// For a departure with no time to finish: a phone off the allowed network.
pub fn leave(reason: &str) {
    leave_within(reason, Duration::ZERO);
}

/// [`leave`], with `notice`: a job the hub expects to finish within it
/// (the claim's `expected_ms`, less the time it has run) finishes and is
/// reported; the others, and any still running when the notice is up, are
/// handed back. For a phone taken off charge, a spot reclaim.
pub fn leave_within(reason: &str, notice: Duration) {
    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = Some(Leave { reason: reason.to_string(), by: Instant::now() + notice });
    stop();
}

/// Ask a runner started with [`serve_embedded`] to drain: claim nothing
/// more, finish and report what it holds, then return (oxygenflow#76's
/// drain, for a phone taken off charge or off the allowed network).
pub fn stop() {
    STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}

fn serve_until_drained(home: &Home) -> Result<()> {
    let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
    // A runner offers only what it can run: a machine without KVM claims
    // nothing rather than taking MicroVM jobs and failing them (#36).
    let (runtimes, missing) = tiers(home).map_err(|why| anyhow::anyhow!("flow runner: {why}"))?;
    for why in &missing {
        eprintln!("flow runner: not offering {why}");
    }
    let hub = Hub::new(&cfg)?;
    let name = cfg.name.clone().unwrap_or_else(|| "flow".into());
    let budget = Budget::of(&cfg);
    let key = home.runner_key(&name)?;
    // The swarm (docs/jetstream.md §8.1): a runner that cannot serve its
    // peers still runs jobs, fetching images from the hub alone.
    let swarm = crate::swarm::Swarm::start(home, &cfg).unwrap_or_else(|e| {
        eprintln!("flow runner: not joining the swarm: {e:#}");
        None
    });
    eprintln!(
        "flow runner {name}: polling {} (up to {} jobs at once, tiers {}), signing receipts as {}",
        cfg.hub,
        budget.max_jobs,
        runtimes.join(", "),
        key.identity.hash().to_hex()
    );
    if let Some(addr) = &cfg.metrics {
        crate::metrics::serve(addr, cfg.metrics_token.clone(), name.clone());
    }
    // The manifests of images held here that the hub keeps none of, so
    // peers in this runner's site can fetch them without anyone pushing
    // them (oxygenflow#94): at start and every ten minutes, which also
    // catches an image imported while it runs. Only a runner that serves
    // its peers has anything to offer.
    if swarm.is_some() {
        let (root, cfg) = (home.root.clone(), cfg.clone());
        std::thread::spawn(move || loop {
            let offered = Home::locate(Some(&root))
                .and_then(|home| crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref())?.offer_held(&home));
            match offered {
                Ok(0) => {}
                Ok(n) => eprintln!("flow runner: sent the hub the manifests of {n} image(s) only runners hold"),
                Err(e) => eprintln!("flow runner: offering held images: {e:#}"),
            }
            std::thread::sleep(Duration::from_secs(600));
        });
    }
    let used = std::sync::Mutex::new(Used::default());
    let mut idle = Duration::from_secs(2);
    let pending = report_pending(home);
    // The jobs held now, told to the hub at each claim so it can end the
    // ones this runner lost: a job is held from its claim until its result
    // is reported.
    let held = std::sync::Mutex::new(std::collections::BTreeSet::<i64>::new());
    held.lock().unwrap().extend(pending.iter().map(|(job, _)| *job));
    std::thread::scope(|scope| {
        for (job, f) in pending {
            let (hub, held) = (&hub, &held);
            scope.spawn(move || {
                eprintln!("job {job}: reporting the result kept from before the restart");
                report_finish(home, hub, job, &f);
                held.lock().unwrap().remove(&job);
            });
        }
        // The speed (spec §11.1): measured now, and again every 15 minutes
        // while no job runs, so a job never shares the CPU with it.
        let measure = || {
            let (speed, ms) = crate::facts::bench();
            eprintln!("flow runner {name}: speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
            speed
        };
        let mut speed = measure();
        let mut benched = Instant::now();
        let mut said_draining = false;
        // When the hub last heard from us (oxygenhub#243): a claim, or a
        // heartbeat while there is no room to claim.
        let mut heard = Instant::now();
        let beat = |heard: &mut Instant| {
            if heard.elapsed() >= HUB_HEARTBEAT {
                if let Err(e) = hub.heartbeat() {
                    eprintln!("flow runner {name}: heartbeat: {e:#}");
                }
                *heard = Instant::now();
            }
        };
        let mut on_trial = crate::update::pending(&home.root).is_some();
        // Runner updates (oxygenhub#322): asked of the hub once it has
        // answered a claim, then every 15 minutes.
        let mut update_asked: Option<Instant> = None;
        let mut update_said = String::new();
        loop {
            if !stopping() && crate::update::drain_requested(&home.root) {
                eprintln!("flow runner {name}: asked to drain (flow runner drain)");
                stop();
            }
            if UPDATE.swap(false, std::sync::atomic::Ordering::SeqCst) || crate::update::requested(&home.root) {
                let p = crate::update::pending(&home.root);
                let notice = Duration::from_secs(p.as_ref().map_or(0, |p| p.notice_secs));
                let to = p.map_or_else(|| "a new build".to_string(), |p| format!("flow {}", p.to));
                eprintln!("flow runner {name}: updating to {to}: leaving within {} s", notice.as_secs());
                LEFT_FOR_UPDATE.store(true, std::sync::atomic::Ordering::SeqCst);
                leave_within(&format!("updating to {to}"), notice);
            }
            // Draining (oxygenflow#50): told to stop, claim nothing more and
            // stop once every job held is done and reported.
            if stopping() {
                let (jobs, holding) = (used.lock().unwrap().jobs, held.lock().unwrap().len());
                if jobs == 0 && holding == 0 {
                    eprintln!("flow runner {name}: drained, stopping");
                    break;
                }
                if !said_draining {
                    eprintln!("flow runner {name}: draining: claiming nothing more, stopping when the {jobs} job(s) held are done (stop again to stop now)");
                    said_draining = true;
                }
                beat(&mut heard);
                std::thread::sleep(Duration::from_millis(500));
                continue;
            }
            if used.lock().unwrap().jobs == 0 && benched.elapsed() >= Duration::from_secs(15 * 60) {
                speed = measure();
                benched = Instant::now();
            }
            let room = budget.room(*used.lock().unwrap());
            let Some((free_memory_mib, free_cpus)) = room else {
                beat(&mut heard);
                std::thread::sleep(Duration::from_millis(500));
                continue;
            };
            heard = Instant::now();
            let cache_times = crate::cache::Pool::new(&home.root, cfg.cache_gb).slot_times();
            let req = ClaimRequest {
                runner: name.clone(),
                runtimes: runtimes.clone(),
                free_memory_mib: claim_memory(free_memory_mib, available_memory_mib()),
                free_cpus,
                info: Some(RunnerInfo {
                    version: env!("OXY_VERSION_LONG").into(),
                    ..RunnerInfo::here(cfg.offered_labels(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)
                }),
                // Read each time: an image imported while the runner runs counts.
                images: Some(home.images()),
                image_archs: Some(home.image_archs()),
                holding: Some(held.lock().unwrap().iter().copied().collect()),
                fetches_images: true,
                facts: Some(crate::facts::now(home, Some(speed))),
                swarm: swarm.as_ref().map(|s| s.addr.clone()),
                caches: Some(cache_times.keys().cloned().collect()),
                cache_times: Some(cache_times),
            };
            let answer = hub.claim(&req);
            if on_trial && answer.is_ok() {
                crate::update::proven(&home.root);
                on_trial = false;
            }
            if answer.is_ok() && update_asked.is_none_or(|t| t.elapsed() >= UPDATE_EVERY) {
                update_asked = Some(Instant::now());
                if let Some(note) = crate::update::take_note(&home.root) {
                    let _ = hub.report_update(&note);
                }
                consider_update(home, &hub, &mut update_said);
            }
            match answer {
                Ok(Some(claim)) => {
                    idle = Duration::from_secs(2);
                    crate::metrics::CLAIMS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);
                    {
                        let mut u = used.lock().unwrap();
                        u.jobs += 1;
                        u.memory_mib += mem;
                        u.cpus += cpus;
                    }
                    held.lock().unwrap().insert(claim.job_id);
                    let (home, hub, cfg, used, key, held, swarm) = (home, &hub, &cfg, &used, &key, &held, swarm.as_ref());
                    scope.spawn(move || {
                        let id = claim.job_id;
                        let image_ms = fetch_image(home, hub, cfg, swarm, &claim);
                        if let Some(why) = preflight(home, &claim) {
                            eprintln!("job {id}: declined: {why}");
                            crate::metrics::DECLINED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            if let Err(e) = hub.decline(id, &why) {
                                eprintln!("job {id}: decline: {e:#}");
                            }
                            held.lock().unwrap().remove(&id);
                            let mut u = used.lock().unwrap();
                            u.jobs -= 1;
                            u.memory_mib -= mem;
                            u.cpus -= cpus;
                            return;
                        }
                        eprintln!(
                            "job {id}: {} {} on {}",
                            claim.repo,
                            claim.job.name,
                            claim.heads.first().map(|h| &h[..12]).unwrap_or("-")
                        );
                        crate::metrics::RUNNING.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        let finish = run_claim(home, hub, cfg, key, &claim, image_ms).unwrap_or_else(|e| Finish {
                            exit_code: None,
                            duration_ms: 0,
                            receipt: None,
                            signed_receipt: None,
                            error: Some(format!("{e:#}")),
                            peak_disk_mib: None,
                            peak_memory_mib: None,
                            phases: Default::default(),
                        });
                        eprintln!(
                            "job {id}: exit {:?}{}",
                            finish.exit_code,
                            finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default()
                        );
                        crate::metrics::RUNNING.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
                        let cancelled = finish.error.as_deref() == Some(CANCELLED);
                        crate::metrics::finished(finish.exit_code, finish.error.as_deref(), finish.duration_ms, cancelled);
                        match leaving() {
                            // Stopped as the runner leaves: another runner takes it.
                            Some(why) if cancelled => {
                                eprintln!("job {id}: handed back: {why}");
                                if let Err(e) = hub.release(id, &why) {
                                    eprintln!("job {id}: handing back: {e:#}; the hub requeues it once this runner's lease lapses");
                                }
                            }
                            _ if !cancelled => report_finish(home, hub, id, &finish),
                            _ => {}
                        }
                        held.lock().unwrap().remove(&id);
                        let mut u = used.lock().unwrap();
                        u.jobs -= 1;
                        u.memory_mib -= mem;
                        u.cpus -= cpus;
                    });
                }
                Ok(None) => std::thread::sleep(idle),
                Err(e) => {
                    eprintln!("claim: {e:#}");
                    std::thread::sleep(idle);
                    idle = (idle * 2).min(Duration::from_secs(60));
                }
            }
        }
        Ok(())
    })
}

/// Pack and keep a handed-back job's checkpoints (oxygenflow#77); what
/// happened goes to its log, and a failure only means the next attempt
/// starts again.
#[allow(clippy::too_many_arguments)]
fn save_checkpoint(hub: &Hub, id: i64, job: &flow_core::Job, scratch: &Path, dir_tier: bool, wasm: bool, mut say: impl FnMut(String)) {
    if !dir_tier {
        say("checkpoint: only the WASM and syscall tiers save one yet; the next attempt starts again".into());
        return;
    }
    let dirs: Vec<(String, PathBuf)> = job
        .checkpoints
        .iter()
        .filter_map(|c| crate::checkpoint::host_dir(c, job, &scratch.join("tree"), scratch, wasm).map(|d| (c.clone(), d)))
        .collect();
    match crate::checkpoint::pack(&dirs).and_then(|(bytes, n)| Ok((hub.save_checkpoint(id, &bytes)?, n))) {
        Ok((blob, n)) => {
            say(format!("checkpoint {}: {} files, {} bytes, kept for the next attempt", &blob[..12.min(blob.len())], n.files, n.bytes))
        }
        Err(e) => say(format!("checkpoint: not kept ({e:#}); the next attempt starts again")),
    }
}

/// How many times the runner was told to stop (SIGTERM or SIGINT).
static STOPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

extern "C" fn on_stop(_: libc::c_int) {
    // A second stop means now: only async-signal-safe calls here.
    if STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst) >= 1 {
        unsafe { libc::_exit(143) };
    }
}

/// Drain instead of dying on the first SIGTERM or SIGINT (oxygenflow#50),
/// so restarting a runner never kills the job it runs: the unit's
/// `KillMode=mixed` sends the signal to the runner alone, not its microVMs.
fn drain_on_stop() {
    let handler = on_stop as extern "C" fn(libc::c_int) as libc::sighandler_t;
    unsafe {
        libc::signal(libc::SIGTERM, handler);
        libc::signal(libc::SIGINT, handler);
    }
}

fn stopping() -> bool {
    STOPS.load(std::sync::atomic::Ordering::SeqCst) > 0
}

/// Batches records to the hub every second (or every 200 lines). A quiet
/// job's sink still asks the hub every few seconds whether the job runs,
/// with an empty batch; once the hub answers 409 the job is not wanted.
struct LogSink<'a> {
    hub: &'a Hub,
    job: i64,
    buf: Vec<Record>,
    last: Instant,
    gone: bool,
    /// Set when the runner leaves ([`LEAVING`]): the job stops, to be
    /// handed back, unless its expected rest fits the notice.
    leaving: &'a std::sync::Mutex<Option<Leave>>,
    /// When the job started, and how long the hub expects it to take.
    started: Instant,
    expected: Option<Duration>,
}

/// The most log lines sent at once, as a busy job's sink flushes them.
const BATCH: usize = 200;

/// The most log lines kept while the hub cannot be reached.
const MAX_UNSENT: usize = 50_000;

/// How often a quiet job asks whether it was cancelled.
const HEARTBEAT: Duration = Duration::from_secs(5);

impl microvm::Sink for LogSink<'_> {
    fn record(&mut self, r: Record) {
        self.push(r)
    }

    fn cancelled(&mut self) -> bool {
        let leave = self.leaving.lock().unwrap_or_else(|p| p.into_inner()).clone();
        if let Some(l) = leave {
            // Finished if what is left of it fits the notice; else handed
            // back as the runner leaves, what the job said so far first.
            let now = Instant::now();
            let fits = self.expected.is_some_and(|e| now + e.saturating_sub(self.started.elapsed()) <= l.by);
            if !fits || now >= l.by {
                self.flush();
                return true;
            }
        }
        if !self.gone && self.last.elapsed() >= HEARTBEAT {
            self.send();
        }
        self.gone
    }
}

impl LogSink<'_> {
    fn push(&mut self, mut r: Record) {
        if self.gone {
            return;
        }
        if r.text.len() > MAX_RECORD_TEXT {
            let mut cut = MAX_RECORD_TEXT;
            while !r.text.is_char_boundary(cut) {
                cut -= 1;
            }
            r.text.truncate(cut);
        }
        self.buf.push(r);
        if self.buf.len() >= BATCH || self.last.elapsed() >= Duration::from_secs(1) {
            self.flush();
        }
    }

    fn flush(&mut self) {
        if !self.buf.is_empty() {
            self.send();
        }
        self.last = Instant::now();
    }

    /// Send what is buffered, even nothing, in batches of at most
    /// `BATCH` lines.
    fn send(&mut self) {
        loop {
            if self.gone {
                return;
            }
            let n = self.buf.len().min(BATCH);
            let batch: Vec<Record> = self.buf.drain(..n).collect();
            match self.hub.log(self.job, batch.clone()) {
                Ok(()) => {}
                Err(e) if job_gone(&e) => self.gone = true,
                Err(e) => {
                    // The hub is away (a restart, say): keep the lines for
                    // the next batch, up to a bound, so the log loses none
                    // (oxygenflow#49).
                    eprintln!("job {}: log: {e:#}; kept for the next batch", self.job);
                    let mut kept = batch;
                    kept.append(&mut self.buf);
                    let over = kept.len().saturating_sub(MAX_UNSENT);
                    if over > 0 {
                        kept.drain(..over);
                        kept.insert(
                            0,
                            Record {
                                t: 0,
                                stream: "flow".into(),
                                text: format!("({over} log lines lost while the hub could not be reached)"),
                            },
                        );
                    }
                    self.buf = kept;
                    break;
                }
            }
            if self.buf.is_empty() {
                break;
            }
        }
        self.last = Instant::now();
    }
}

/// How often a runner with no room to claim tells the hub it is there:
/// well inside the hub's 90 s before it shows a runner offline.
const HUB_HEARTBEAT: Duration = Duration::from_secs(30);

/// How often a runner asks the hub whether it should run another build.
const UPDATE_EVERY: Duration = Duration::from_secs(15 * 60);

/// Notice a runner gives itself to leave for an update the hub offered.
const UPDATE_NOTICE: Duration = Duration::from_secs(600);

/// Ask the hub for this runner's update, and act on it: download the build
/// it offers, check it against the hub's BLAKE3 and as a runner build, and
/// install it, or for a dry run say what would happen. What it did goes to
/// the hub once per outcome (`said`).
fn consider_update(home: &Home, hub: &Hub, said: &mut String) {
    let answer = match hub.update_answer() {
        Ok(Some(a)) => a,
        Ok(None) => return,
        Err(e) => {
            eprintln!("flow runner: asking for an update: {e:#}");
            return;
        }
    };
    let Some(offer) = answer.offer else { return };
    let mut say = |note: String| {
        if *said != note {
            eprintln!("flow runner: {note}");
            if let Err(e) = hub.report_update(&note) {
                eprintln!("flow runner: telling the hub: {e:#}");
            }
            *said = note;
        }
    };
    if home.config.runner.as_ref().and_then(|r| r.update.as_ref()).is_some_and(|u| !u.allow) {
        return say(format!("refuses flow {}: flow.toml says [runner.update] allow = false", offer.version));
    }
    let dir = home.root.join("updates");
    let file = dir.join(format!("flow-{}", offer.version));
    let fresh = std::fs::read(&file).is_ok_and(|b| blake3::hash(&b).to_hex().as_str() == offer.blake3);
    if !fresh {
        if let Err(e) = std::fs::create_dir_all(&dir).map_err(anyhow::Error::from).and_then(|_| hub.download_build(&offer, &file)) {
            return say(format!("flow {}: not fetched: {e:#}", offer.version));
        }
    }
    let exe = match std::env::current_exe() {
        Ok(e) => e,
        Err(e) => return say(format!("flow {}: cannot find its own binary: {e}", offer.version)),
    };
    match crate::update::install(home, &exe, &file, UPDATE_NOTICE, answer.policy.dry_run, offer.older) {
        Ok(done) if answer.policy.dry_run => say(format!("dry run: flow {} checks out here and would replace flow {}", done.to, done.from)),
        Ok(done) => {
            let _ = std::fs::remove_file(&file);
            say(format!("installed flow {} (was {}); leaving within {} s to start it", done.to, done.from, UPDATE_NOTICE.as_secs()))
        }
        Err(e) => say(format!("flow {}: not installed: {e:#}", offer.version)),
    }
}

/// Held while a job takes its cache disks and images its workspace (see
/// `run_claim`): one job at a time on the runner's disk and cache pool.
static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// Held while the first clone of any repository passes its token through
/// the process environment.
static CLONE: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// One lock per cached repository: fetches into it, and exports from it,
/// never overlap, while jobs of other repositories go on. A slow repository
/// (one kept in a bucket, say) once held every job on the runner waiting
/// behind a single global lock.
fn repo_lock(name: &str) -> std::sync::Arc<std::sync::Mutex<()>> {
    static LOCKS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, std::sync::Arc<std::sync::Mutex<()>>>>> =
        std::sync::OnceLock::new();
    let mut locks = LOCKS.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner());
    locks.entry(name.to_string()).or_default().clone()
}

/// Seconds, to one decimal place, for the job's log.
fn secs(d: std::time::Duration) -> String {
    format!("{:.1} s", d.as_secs_f64())
}

/// `image_ms`: how long fetching its image took, when the runner lacked it.
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim, image_ms: Option<u64>) -> Result<Finish> {
    let started = Instant::now();
    let mut sink = LogSink {
        hub,
        job: claim.job_id,
        buf: Vec::new(),
        last: Instant::now(),
        gone: false,
        leaving: &LEAVING,
        started: Instant::now(),
        expected: claim.expected_ms.map(Duration::from_millis),
    };
    let note =
        |sink: &mut LogSink, text: String| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
    let heads: Vec<ChangeHash> = claim.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
    note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(|h| &h[..12]).collect::<Vec<_>>().join("+")));
    // Preparing a workspace: fetches into a cached repository, and exports
    // from it, are one job at a time per repository (`repo_lock`); taking
    // cache disks and imaging the workspace are one job at a time on the
    // runner (`PREPARE`); a first clone, whose token goes through the process
    // environment, one at a time (`CLONE`).
    let main_lock = repo_lock(&claim.repo);
    let asked = Instant::now();
    let mut main_guard = Some(main_lock.lock().unwrap_or_else(|p| p.into_inner()));
    if asked.elapsed().as_secs() >= 2 {
        note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), claim.repo));
    }
    let fetching = Instant::now();
    let repo = cached_repo(home, cfg, claim)?;
    let fetched = fetching.elapsed();
    let mut preparing = None;
    let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
    let _ = std::fs::remove_dir_all(&scratch);
    std::fs::create_dir_all(scratch.join("tree"))?;
    let result = (|| -> Result<Finish> {
        let exporting = Instant::now();
        let mut files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
        main_guard.take();
        note(&mut sink, format!("fetched in {}, {} files exported in {}", secs(fetched), files.len(), secs(exporting.elapsed())));
        // Pinned subrepos (Oxygen spec §7.8), at their paths, as part of the
        // workspace and of the scope hash.
        for sub in &claim.subrepos {
            if sub.path.split('/').any(|p| p.is_empty() || p == "." || p == ".." || p == ".oxy") {
                bail!("subrepo path {:?} is not a clean relative path", sub.path);
            }
            note(
                &mut sink,
                format!(
                    "fetching subrepo {} at {} for {}",
                    sub.repo,
                    sub.heads.iter().map(|h| &h[..12]).collect::<Vec<_>>().join("+"),
                    sub.path
                ),
            );
            let sub_lock = repo_lock(&sub.repo);
            let asked = Instant::now();
            let sub_guard = sub_lock.lock().unwrap_or_else(|p| p.into_inner());
            if asked.elapsed().as_secs() >= 2 {
                note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), sub.repo));
            }
            let sub_repo = cached_repo_at(home, cfg, &sub.repo, &sub.clone_url, &claim.token)?;
            let sub_heads: Vec<ChangeHash> =
                sub.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad subrepo head")).collect::<Result<_>>()?;
            let dir = scratch.join("tree").join(&sub.path);
            std::fs::create_dir_all(&dir)?;
            let sub_files = workspace::export_state(&sub_repo, &sub_heads, &dir)?;
            drop(sub_guard);
            files.extend(sub_files.into_iter().map(|(p, h)| (format!("{}/{p}", sub.path), h)));
        }
        let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(|(p, h)| (p.as_str(), *h)));
        // Inputs (oxygenflow#58): each producing job's artifact, checked
        // against the blob its receipt names, then put at its path. Not part
        // of the scope: they are the run's, not the repository's.
        let mut input_blobs = Vec::new();
        for (index, i) in claim.inputs.iter().enumerate() {
            let what = format!("input {} from {}", i.artifact, i.from);
            let bytes = hub.input(claim.job_id, index).with_context(|| format!("{what}: fetching it"))?;
            let blob = place_input(&scratch.join("tree"), i, &bytes).with_context(|| what.clone())?;
            note(&mut sink, format!("{what}: {} bytes at {}, blob {} as its receipt names", bytes.len(), i.path, &i.blob[..12]));
            input_blobs.push(blob);
        }
        // The syscall tier runs on the tree itself; a microVM gets an image.
        let syscall_tier = claim.sandbox.runtime == flow_core::Runtime::Syscall;
        let wasm_tier = claim.sandbox.runtime == flow_core::Runtime::Wasm;
        // The syscall and WASM tiers see the job's files as a directory;
        // the MicroVM tier as an ext4 image.
        let dir_tier = syscall_tier || wasm_tier;
        let job = claim.job.in_sandbox(&claim.sandbox).with_flow_env(&claim.flow_env());
        // Disk mounts that name a cache start from a kept disk (#28): a
        // microVM takes an image already in place as it is, the syscall tier
        // a directory (#70). Taken before the workspace is imaged: the
        // sources the disk was built from give unchanged files their old
        // times back (#52).
        let asked = Instant::now();
        preparing = Some(PREPARE.lock().unwrap_or_else(|p| p.into_inner()));
        if asked.elapsed().as_secs() >= 2 {
            note(&mut sink, format!("waited {} for another job imaging its workspace", secs(asked.elapsed())));
        }
        let pool = crate::cache::Pool::new(&home.root, cfg.cache_gb);
        let form = if dir_tier { crate::cache::Form::Dir } else { crate::cache::Form::Image };
        let cache_ctx =
            crate::cache::Ctx { repo: &claim.repo, bookmark: &claim.bookmark, job: &claim.job.name, image: &claim.sandbox.rootfs, form };
        let mut cached: Vec<(usize, String, String)> = Vec::new();
        let mut built_from: Option<crate::cache::Sources> = None;
        for (i, m) in job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).enumerate() {
            let Some(name) = &m.cache else { continue };
            let key = crate::cache::Pool::key(&cache_ctx, m, |p| files.iter().find(|(f, _)| f == p).map(|(_, h)| hex::encode(h)));
            let how = match pool.take(&cache_ctx, m, &key, &crate::cache::disk_path(form, &scratch, i)) {
                Some(t) => {
                    if built_from.is_none() {
                        built_from = t.sources;
                    }
                    t.how
                }
                None => format!("cache {name}: none kept yet, an empty disk"),
            };
            note(&mut sink, format!("{} ({})", how, m.target));
            // What the job reads is seen when the disk is pruned (#72).
            if let Err(e) = crate::cache::forget_reads(form, &crate::cache::disk_path(form, &scratch, i)) {
                note(&mut sink, format!("cache {name}: {e:#}"));
            }
            cached.push((i, key, how));
        }
        // What the job reads or writes on its cache from now on is what it
        // used; the rest may be pruned before the cache is kept (#72).
        let cache_started = if cached.is_empty() { std::time::SystemTime::now() } else { flow_init::prune::start_clock() };
        let sources = if cached.is_empty() {
            crate::cache::Sources::new()
        } else {
            let (sources, restored) = workspace::restore_times(&scratch.join("tree"), &files, built_from.as_ref());
            if built_from.is_some() {
                note(&mut sink, format!("cache: {restored} of {} files unchanged since the cached build, their times kept", sources.len()));
            }
            sources
        };
        // An earlier attempt's checkpoint (oxygenflow#77): back in place,
        // over the cache, before the same exec runs again; the receipt
        // lists it among the job's inputs.
        if let Some(blob) = &claim.checkpoint {
            if dir_tier {
                let bytes = hub.checkpoint(claim.job_id).context("fetching the checkpoint")?;
                if Repo::blob_ref(&bytes).to_hex() != *blob {
                    bail!("the checkpoint fetched is not blob {}, as the claim names", &blob[..12.min(blob.len())]);
                }
                let place = |root: &str| crate::checkpoint::host_dir(root, &job, &scratch.join("tree"), &scratch, wasm_tier);
                let n = crate::checkpoint::unpack(&bytes, &job.checkpoints, place)?;
                note(
                    &mut sink,
                    format!("restored checkpoint {}: {} files, {} bytes, from the attempt handed back", &blob[..12], n.files, n.bytes),
                );
                input_blobs.push(Repo::blob_ref(&bytes).0);
            } else {
                note(&mut sink, "a checkpoint was saved, but only the WASM and syscall tiers restore one yet: starting again".into());
            }
        }
        if !dir_tier {
            workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
            std::fs::remove_dir_all(scratch.join("tree"))?;
        }
        preparing.take();
        // Cancelled while it was being prepared: no sandbox.
        if microvm::Sink::cancelled(&mut sink) {
            return Ok(cancelled());
        }
        if wasm_tier {
            note(&mut sink, format!("starting WASM (wasmtime {}): {} MiB", crate::wasm::WASMTIME, claim.sandbox.memory_mib));
        } else {
            note(
                &mut sink,
                format!(
                    "{}: {} {}, {} MiB, rootfs {}",
                    if syscall_tier { "starting a syscall sandbox" } else { "booting a microVM" },
                    claim.sandbox.cpus,
                    if syscall_tier { "CPU" } else { "vCPU" },
                    claim.sandbox.memory_mib,
                    &claim.sandbox.rootfs[..12]
                ),
            );
        }
        let timeout = Duration::from_secs(claim.timeout_s);
        // Where the time goes (oxygenflow#35): what came before the
        // sandbox, the sandbox, and what comes after.
        let prepare_ms = started.elapsed().as_millis() as u64;
        let sandbox_started = Instant::now();
        let mut outcome = if wasm_tier {
            crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)?
        } else if syscall_tier {
            let group = format!("job-{}", claim.job_id);
            crate::syscall::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("tree"), &scratch, timeout, &mut sink, &group)?
        } else {
            microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?
        };
        let run_ms = sandbox_started.elapsed().as_millis() as u64;
        if outcome.cancelled {
            // Handed back as the runner leaves: its checkpoints first, for
            // the next attempt (oxygenflow#77).
            if leaving().is_some() && !job.checkpoints.is_empty() {
                save_checkpoint(hub, claim.job_id, &job, &scratch, dir_tier, wasm_tier, |t| note(&mut sink, t));
            }
            return Ok(cancelled());
        }
        // The receipt says what a cached disk started from.
        let disk_mounts: Vec<&flow_core::pipeline::Mount> =
            job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).collect();
        for (i, key, how) in &cached {
            outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));
        }
        // A passing job's artifacts, read from its disks and uploaded; a
        // declared artifact that cannot be kept fails the job. A failing
        // job keeps only its `when "always"` ones (oxygenflow#37), outside
        // its receipt, and one it cannot keep is only noted.
        let mut outputs = vec![];
        let mut output_error = None;
        let passed = outcome.exit_code == Some(0);
        let wanted = claim.job.outputs.iter().filter(|o| passed || o.always);
        for o in wanted {
            if o.kind == flow_core::OutputKind::Report {
                let read = if dir_tier {
                    let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
                    crate::outputs::read_report_dir(&dir, &inner)
                } else {
                    let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
                    crate::outputs::read_report(&image, &inner, &scratch)
                };
                let kept = read.and_then(|files| upload_report(hub, claim.job_id, &o.name, &files));
                match kept {
                    Ok((blob, files, bytes)) => {
                        let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
                        let hex = blob.to_hex();
                        note(&mut sink, format!("report {}: {files} files, {bytes} bytes, manifest {}{unattested}", o.name, &hex[..12]));
                        if passed {
                            outputs.push(blob.0);
                        }
                    }
                    Err(e) if passed => {
                        output_error = Some(format!("report {}: {e:#}", o.name));
                        break;
                    }
                    Err(e) => note(&mut sink, format!("report {}: not kept: {e:#}", o.name)),
                }
                continue;
            }
            let read = if dir_tier {
                let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
                crate::outputs::read_dir_file(&dir, &inner)
            } else {
                let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
                crate::outputs::read(&image, &inner, &scratch)
            };
            let kept = read.and_then(|bytes| {
                let blob = Repo::blob_ref(&bytes);
                let named = hub.artifact(claim.job_id, &o.name, &bytes)?;
                if named != blob.to_hex() {
                    bail!("the hub kept it as {named}, not {}", blob.to_hex());
                }
                let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
                note(&mut sink, format!("artifact {}: {} bytes, blob {}{unattested}", o.name, bytes.len(), &named[..12]));
                Ok(blob.0)
            });
            match kept {
                Ok(b) if passed => outputs.push(b),
                Ok(_) => {}
                Err(e) if passed => {
                    output_error = Some(format!("artifact {}: {e:#}", o.name));
                    break;
                }
                Err(e) => note(&mut sink, format!("artifact {}: not kept: {e:#}", o.name)),
            }
        }
        if let Some(e) = &output_error {
            note(&mut sink, e.clone());
        }
        let receipt = outcome.exit_code.filter(|_| output_error.is_none()).map(|code| Receipt {
            job_name: claim.job.name.clone(),
            pipeline_hash: hex32(&claim.pipeline_hash),
            scope_tree_hash: scope_hash,
            inputs: input_blobs.clone(),
            environment: outcome.environment.hash(),
            outputs,
            exit_code: code as u32,
            duration_ms: outcome.duration_ms,
        });
        let signed_receipt = match &receipt {
            Some(r) => {
                let mut signed = sign_receipt(&repo, &heads, cfg, key, claim, r)?;
                if cfg.keyless {
                    match sign_keyless(&repo, &heads, cfg, hub, claim, r) {
                        Ok(k) => signed = k,
                        Err(e) => {
                            note(&mut sink, format!("receipt: signed with the runner's key; a short-lived key was not certified: {e:#}"))
                        }
                    }
                }
                Some(hex::encode(signed.encode()))
            }
            None => None,
        };
        // Measured before a cached disk leaves the scratch directory.
        let peak_disk_mib = crate::facts::allocated_mib(&scratch);
        // A passing job's cached disks are pruned and kept (#72); a
        // failing job's are thrown away, so nothing it wrote reaches a
        // later job.
        let kept = passed && output_error.is_none();
        for (i, key, _) in &cached {
            let disk = crate::cache::disk_path(form, &scratch, *i);
            let m = disk_mounts[*i];
            let prune = crate::cache::Prune {
                started: cache_started,
                percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),
                guest: outcome.pruned.iter().find(|(t, _)| *t == m.target).map(|(_, p)| p),
            };
            match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {
                Ok(Some(line)) => note(&mut sink, line),
                Ok(None) => {}
                Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),
            }
        }
        let total_ms = started.elapsed().as_millis() as u64;
        let mut phases: std::collections::BTreeMap<String, u64> =
            [("prepare", prepare_ms), ("run", run_ms), ("finish", total_ms.saturating_sub(prepare_ms + run_ms))]
                .into_iter()
                .map(|(k, v)| (k.to_string(), v))
                .collect();
        if let Some(ms) = image_ms {
            phases.insert("image".into(), ms);
        }
        Ok(Finish {
            phases,
            exit_code: outcome.exit_code,
            duration_ms: total_ms,
            receipt,
            signed_receipt,
            error: output_error.or_else(|| outcome.timed_out.then(|| "stopped at the time limit".to_string())),
            // What the job took (oxygenflow#41): its disks are sparse files
            // in the scratch directory, so their blocks now are their peak.
            peak_disk_mib,
            peak_memory_mib: outcome.peak_memory_mib,
        })
    })();
    let _ = workspace::remove_tree(&scratch);
    sink.flush();
    result
}

/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
/// runner (Oxygen spec §14.6). The state is computed here, not taken from
/// the claim, so the runner vouches only for what it fetched.
/// Upload a report's files, then its manifest under the report's name, and
/// check the hub kept each as the blob it should (oxygenhub#127). Answers
/// the manifest's blob, the file count and their total size.
fn upload_report(hub: &Hub, job: i64, name: &str, files: &[(String, Vec<u8>)]) -> Result<(oxy_core::hash::ObjectRef, usize, u64)> {
    let mut listed = Vec::with_capacity(files.len());
    let mut total = 0u64;
    for (path, bytes) in files {
        let blob = Repo::blob_ref(bytes).to_hex();
        let named = hub.report_file(job, name, path, bytes).with_context(|| format!("uploading {path}"))?;
        if named != blob {
            bail!("the hub kept {path} as {named}, not {blob}");
        }
        total += bytes.len() as u64;
        listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
    }
    let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed }))?;
    let blob = Repo::blob_ref(&manifest);
    let named = hub.artifact(job, name, &manifest)?;
    if named != blob.to_hex() {
        bail!("the hub kept the manifest as {named}, not {}", blob.to_hex());
    }
    Ok((blob, files.len(), total))
}

fn sign_receipt(
    repo: &Repo,
    heads: &[ChangeHash],
    cfg: &RunnerConfig,
    key: &RunnerKey,
    claim: &Claim,
    receipt: &Receipt,
) -> Result<oxy_core::receipt::SignedReceipt> {
    let state = state_of(heads);
    let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
    Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
}

/// Sign `receipt` with a fresh key the hub certifies for this job now,
/// just before signing, so the certificate covers the moment the hub
/// accepts it (keyless, oxygenhub#241). The key is dropped with the
/// receipt signed: nothing long-lived signs.
fn sign_keyless(
    repo: &Repo,
    heads: &[ChangeHash],
    cfg: &RunnerConfig,
    hub: &Hub,
    claim: &Claim,
    receipt: &Receipt,
) -> Result<oxy_core::receipt::SignedReceipt> {
    let sk = oxy_core::identity::SecretKey::generate();
    let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
    let name = format!("{} job {}", cfg.name.as_deref().unwrap_or("runner"), claim.job_id);
    let id = sk.identity(&name, "", now);
    let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge(&claim.token)).map_err(|e| anyhow::anyhow!("{e}"))?;
    let (cert, issuer) = hub.certify(&claim.token, &id, &proof)?;
    let state = state_of(heads);
    let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
    Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &id, &sk).with_certificate(cert, issuer))
}

/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
    use oxy_core::hash::{hash_object, ObjectType};
    hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
}

fn hex32(s: &str) -> [u8; 32] {
    let mut out = [0; 32];
    if let Ok(v) = hex::decode(s) {
        if v.len() == 32 {
            out.copy_from_slice(&v);
        }
    }
    out
}

/// Whether a network error is worth another try: DNS, a refused or reset
/// connection, a timeout, or a gateway answering 502, 503 or 504 while the
/// hub restarts (oxygenhub#111).
fn transient(msg: &str) -> bool {
    let m = msg.to_lowercase();
    [
        "dns failed",
        "name resolution",
        "connection refused",
        "connection reset",
        "timed out",
        "timeout",
        "http 502",
        "http 503",
        "http 504",
        "broken pipe",
        "unexpected eof",
    ]
    .iter()
    .any(|k| m.contains(k))
}

/// Run `f` again after a transient network error, waiting `waits` between
/// tries (about a minute in all), and give up on any other error at once.
fn with_retries<T>(waits: &[Duration], mut f: impl FnMut() -> Result<T>) -> Result<T> {
    let mut tries = waits.iter();
    loop {
        match f() {
            Ok(v) => return Ok(v),
            Err(e) => match tries.next() {
                Some(w) if transient(&format!("{e:#}")) => std::thread::sleep(*w),
                _ => return Err(e),
            },
        }
    }
}

/// The waits between tries of a fetch: 2, 4, 8, 16 and 30 seconds.
const FETCH_WAITS: [Duration; 5] =
    [Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];

/// The runner's copy of a repository, fetched up to date with the job's token.
fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
    cached_repo_at(home, cfg, &claim.repo, &claim.clone_url, &claim.token)
}

/// The runner's copy of `repo`, cloned or fetched with `token`.
fn cached_repo_at(home: &Home, cfg: &RunnerConfig, repo_name: &str, clone_url: &str, token: &str) -> Result<Repo> {
    let dir: PathBuf = home.root.join("repos").join(repo_name.replace('/', "__"));
    let set_remote = |repo: &mut Repo| -> Result<()> {
        repo.config.set("remote.origin.url", clone_url).map_err(|e| anyhow::anyhow!("{e}"))?;
        repo.config.set("remote.origin.token", token).map_err(|e| anyhow::anyhow!("{e}"))?;
        if let Some(ca) = &cfg.ca {
            repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(|e| anyhow::anyhow!("{e}"))?;
        }
        repo.save_config().map_err(|e| anyhow::anyhow!("{e}"))
    };
    if dir.join(".oxy").is_dir() {
        let mut repo = Repo::open(&dir).map_err(|e| anyhow::anyhow!("{e}"))?;
        set_remote(&mut repo)?;
        with_retries(&FETCH_WAITS, || repo.fetch("origin").map(|_| ()).map_err(|e| anyhow::anyhow!("fetch {clone_url}: {e}")))?;
        return Ok(repo);
    }
    // First job for this repository: clone with the job's token, which goes
    // through the process environment, so one clone at a time.
    let _clone = CLONE.lock().unwrap_or_else(|p| p.into_inner());
    std::env::set_var("OXY_TOKEN", token);
    if let Some(ca) = &cfg.ca {
        std::env::set_var("OXY_CA", ca);
    }
    let cloned = with_retries(&FETCH_WAITS, || {
        let _ = std::fs::remove_dir_all(&dir);
        Repo::clone(clone_url, &dir).map_err(|e| anyhow::anyhow!("{e}"))
    });
    std::env::remove_var("OXY_TOKEN");
    std::env::remove_var("OXY_CA");
    let mut repo = cloned.map_err(|e| {
        let _ = std::fs::remove_dir_all(&dir);
        anyhow::anyhow!("clone {}: {e}", clone_url)
    })?;
    set_remote(&mut repo)?;
    Ok(repo)
}

#[cfg(test)]
mod tests {

    /// #98: register keeps a copy of the CA in the home, by absolute path,
    /// and keeping the home's own copy again is a no-op.
    #[test]
    fn the_ca_is_kept_in_the_home_by_absolute_path() {
        let home = tempfile::tempdir().unwrap();
        let elsewhere = tempfile::tempdir().unwrap();
        let given = elsewhere.path().join("ca.pem");
        std::fs::write(&given, "PEM").unwrap();
        let kept = super::keep_ca(home.path(), &given).unwrap();
        assert!(kept.is_absolute() && kept.ends_with(crate::home::CA_PEM), "{kept:?}");
        assert_eq!(std::fs::read_to_string(&kept).unwrap(), "PEM");
        std::fs::remove_file(&given).unwrap();
        assert_eq!(super::keep_ca(home.path(), &kept).unwrap(), kept, "the home's copy again");
        assert_eq!(std::fs::read_to_string(&kept).unwrap(), "PEM");
    }

    #[test]
    fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {
        let dir = tempfile::tempdir().unwrap();
        let bytes = b"the reviewed plan";
        let blob = Repo::blob_ref(bytes).to_hex();
        let input = |path: &str, blob: &str| flow_core::api::ArtifactInput {
            artifact: "tfplan".into(),
            from: "plan".into(),
            path: path.into(),
            blob: blob.into(),
        };
        let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();
        assert_eq!(hex::encode(got), blob);
        assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);
        let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();
        assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");
        assert!(!dir.path().join("other").exists(), "nothing written when it does not match");
        for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {
            assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");
        }
    }

    use super::*;

    #[test]
    fn a_runner_names_what_stops_a_sandbox() {
        let dir = tempfile::tempdir().unwrap();
        let home = Home::locate(Some(dir.path())).unwrap();
        let bytes = b"not really ext4";
        let hash = blake3::hash(bytes).to_hex().to_string();
        let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
            "name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"
        }))
        .unwrap();
        let problem = |sb: &flow_core::Sandbox, free| sandbox_problem(&home, sb, "x86_64", free);
        assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");
        std::fs::write(home.rootfs(&hash), bytes).unwrap();
        sb.rootfs = hash.clone();
        assert_eq!(problem(&sb, Some(4096)), None);
        assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");
        std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();
        assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));
        sb.arch = Some("aarch64".into());
        assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");
        sb.arch = None;
        std::fs::remove_file(home.image_manifest(&hash)).unwrap();
        std::fs::write(home.rootfs(&hash), b"changed").unwrap();
        assert!(problem(&sb, None).unwrap().contains("does not verify"));
    }

    #[test]
    fn transient_network_errors_are_tried_again_and_others_are_not() {
        assert!(transient("fetch x: Dns Failed: resolve dns name 'oxygenhub.io:443': Temporary failure in name resolution"));
        assert!(transient("HTTP 503: the hub is restarting"));
        assert!(!transient("HTTP 404: no such repository"));
        let mut n = 0;
        let got: Result<u32> = with_retries(&[Duration::ZERO, Duration::ZERO], || {
            n += 1;
            if n < 3 {
                Err(anyhow::anyhow!("connection refused"))
            } else {
                Ok(7)
            }
        });
        assert_eq!((got.unwrap(), n), (7, 3), "two retries, then success");
        let mut m = 0;
        let got: Result<u32> = with_retries(&[Duration::ZERO; 4], || {
            m += 1;
            Err(anyhow::anyhow!("HTTP 403: forbidden"))
        });
        assert!(got.is_err());
        assert_eq!(m, 1, "a refusal is not tried again");
        let mut k = 0;
        let got: Result<u32> = with_retries(&[Duration::ZERO; 2], || {
            k += 1;
            Err(anyhow::anyhow!("timed out"))
        });
        assert!(got.is_err());
        assert_eq!(k, 3, "gives up after the last wait");
    }

    /// A hub that answers each request with the next of `codes`, counting them.
    fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
        fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)
    }

    /// A hub at `listener` answering each request with the next of `codes`.
    fn fake_hub_on(listener: std::net::TcpListener, codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
        use std::io::{Read, Write};
        let addr = listener.local_addr().unwrap();
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let count = seen.clone();
        std::thread::spawn(move || {
            for code in codes {
                let (mut s, _) = listener.accept().unwrap();
                let mut buf = [0u8; 8192];
                let _ = s.read(&mut buf);
                count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let reason = if code == 204 { "No Content" } else { "Error" };
                let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
            }
        });
        (format!("http://{addr}"), seen)
    }

    /// A hub that answers one request with `body` (status 200) and hands
    /// back what it was sent.
    fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
        use std::io::{Read, Write};
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let (mut s, _) = listener.accept().unwrap();
            let mut req = Vec::new();
            let mut buf = [0u8; 65536];
            loop {
                let n = s.read(&mut buf).unwrap_or(0);
                if n == 0 {
                    break;
                }
                req.extend_from_slice(&buf[..n]);
                let text = String::from_utf8_lossy(&req);
                if let Some(end) = text.find("\r\n\r\n") {
                    let len = text[..end]
                        .lines()
                        .find_map(|l| {
                            l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
                        })
                        .unwrap_or(0);
                    if req.len() >= end + 4 + len {
                        break;
                    }
                }
            }
            tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
            let _ = write!(
                s,
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            );
        });
        (format!("http://{addr}"), rx)
    }

    #[test]
    fn registering_binds_the_key_and_keeps_the_credential_private() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().unwrap();
        let home = Home::locate(Some(dir.path())).unwrap();
        let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
        let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
        assert_eq!((got.id, got.name.as_str()), (7, "plex"));
        let req = sent.recv().unwrap();
        assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
        let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
        assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
        let key = home.runner_key("plex").unwrap();
        assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
        let path = dir.path().join(crate::home::RUNNER_TOML);
        assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
        // The runner then starts from runner.toml.
        let again = Home::locate(Some(dir.path())).unwrap();
        let cfg = again.config.runner.unwrap();
        assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
        assert_eq!(cfg.labels, ["gpu"]);
    }

    /// A hub that answers certify with a certificate `sign_as` signs for
    /// `cert_key`, and publishes `published` as its key: two requests.
    fn fake_certify_hub(
        sign_as: (oxy_core::identity::Identity, oxy_core::identity::SecretKey),
        cert_key: oxy_core::identity::Identity,
        published: oxy_core::identity::Identity,
    ) -> String {
        use std::io::{Read, Write};
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            for _ in 0..2 {
                let Ok((mut s, _)) = listener.accept() else { return };
                let mut req = Vec::new();
                let mut buf = [0u8; 65536];
                loop {
                    let n = s.read(&mut buf).unwrap_or(0);
                    if n == 0 {
                        break;
                    }
                    req.extend_from_slice(&buf[..n]);
                    let text = String::from_utf8_lossy(&req);
                    if let Some(end) = text.find("\r\n\r\n") {
                        let len = text[..end]
                            .lines()
                            .find_map(|l| {
                                l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
                            })
                            .unwrap_or(0);
                        if req.len() >= end + 4 + len {
                            break;
                        }
                    }
                }
                let line = String::from_utf8_lossy(&req).lines().next().unwrap_or("").to_string();
                let body: Vec<u8> = if line.contains("/keyless/certify") {
                    let cert = oxy_core::cert::SignedCertificate::sign(
                        oxy_core::cert::Certificate {
                            key: cert_key.clone(),
                            issuer: sign_as.0.hash(),
                            subject: "repo:andy/demo:bookmark:main".into(),
                            claims: vec![],
                            scope: oxy_core::identity::Scope::SEAL,
                            not_before: 1,
                            expires: 600_001,
                            grant: None,
                        },
                        &sign_as.0,
                        &sign_as.1,
                    );
                    serde_json::json!({ "certificate": oxy_core::dsse::base64(&cert.encode()) }).to_string().into_bytes()
                } else {
                    published.encode()
                };
                let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
                let _ = s.write_all(&body);
            }
        });
        format!("http://{addr}")
    }

    /// keyless (oxygenhub#241): the runner takes a certificate only when it
    /// is for its fresh key and signed by the key the hub publishes.
    #[test]
    fn a_certificate_is_taken_only_for_this_key_from_the_hubs_key() {
        use oxy_core::identity::SecretKey;
        let hub_sk = SecretKey::generate();
        let hub_id = hub_sk.identity("hub", "", 1);
        let sk = SecretKey::generate();
        let id = sk.identity("earth-1 job 7", "", 1);
        let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge("flowjob_x")).unwrap();

        let base = fake_certify_hub((hub_id.clone(), hub_sk.clone()), id.clone(), hub_id.clone());
        let (cert, issuer) = hub_at(base).certify("flowjob_x", &id, &proof).unwrap();
        assert_eq!((cert.cert.key_id(), issuer.hash()), (id.hash(), hub_id.hash()));

        let mallory_sk = SecretKey::generate();
        let mallory = mallory_sk.identity("mallory", "", 1);
        let base = fake_certify_hub((mallory.clone(), mallory_sk), id.clone(), hub_id.clone());
        let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
        assert!(e.contains("not signed by the hub's published key"), "{e}");

        let other = SecretKey::generate().identity("other", "", 1);
        let base = fake_certify_hub((hub_id.clone(), hub_sk), other, hub_id);
        let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
        assert!(e.contains("another key"), "{e}");
    }

    fn hub_at(base: String) -> Hub {
        Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
    }

    /// oxygenhub#322: an offered build is kept only if it is the bytes the
    /// hub attests; a peer or a proxy cannot change what is installed.
    #[test]
    fn a_build_is_kept_only_if_it_matches_the_hubs_blake3() {
        let (body, other) = (b"#!flow build 0.3.2".to_vec(), b"#!something else!!".to_vec());
        let serve = |bytes: Vec<u8>| {
            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            let addr = listener.local_addr().unwrap();
            std::thread::spawn(move || {
                use std::io::{Read, Write};
                let (mut s, _) = listener.accept().unwrap();
                let mut buf = [0u8; 4096];
                let _ = s.read(&mut buf);
                let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", bytes.len());
                let _ = s.write_all(&bytes);
            });
            format!("http://{addr}")
        };
        let offer = flow_core::api::UpdateOffer {
            version: "0.3.2".into(),
            asset: "flow-linux-x86_64".into(),
            url: "/andy/oxygenflow/asset/1/flow-linux-x86_64".into(),
            blake3: blake3::hash(&body).to_hex().to_string(),
            size: body.len() as u64,
            ..Default::default()
        };
        let dir = tempfile::tempdir().unwrap();
        let to = dir.path().join("flow-0.3.2");
        hub_at(serve(body.clone())).download_build(&offer, &to).unwrap();
        assert_eq!(std::fs::read(&to).unwrap(), body);
        std::fs::remove_file(&to).unwrap();
        let e = hub_at(serve(other)).download_build(&offer, &to).unwrap_err();
        assert!(e.to_string().contains("does not match what the hub attests"), "{e:#}");
        assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing of it is left");
    }

    /// oxygenhub#71: a job's result is offered until the hub takes it,
    /// through server errors, and a refusal ends the trying at once.
    #[test]
    fn a_result_is_reported_until_the_hub_takes_it() {
        let f = Finish {
            exit_code: Some(0),
            duration_ms: 1,
            receipt: None,
            signed_receipt: None,
            error: None,
            peak_disk_mib: None,
            peak_memory_mib: None,
            phases: Default::default(),
        };
        let quick = Duration::from_millis(10);
        let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
        hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");
        let (base, seen) = fake_hub(vec![409]);
        assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
    }

    #[test]
    fn server_errors_are_tried_again_and_client_errors_are_not() {
        let quick = [Duration::from_millis(10); 3];
        let (base, seen) = fake_hub(vec![500, 503, 204]);
        let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
        assert!(r.is_none());
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");

        let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
        let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
        assert!(e.contains("HTTP 500"), "{e}");
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");

        let (base, seen) = fake_hub(vec![401]);
        assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
    }

    static NOT_LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);

    /// oxygenflow#76: once the runner leaves, a job's sink says it is
    /// cancelled, having sent what the job said so far, and the job is
    /// handed back to the hub.
    #[test]
    fn a_leaving_runner_stops_its_job_and_hands_it_back() {
        use microvm::Sink;
        let leaving = std::sync::Mutex::new(None);
        let (base, seen) = fake_hub(vec![204, 204]);
        let hub = hub_at(base);
        let mut sink = LogSink {
            hub: &hub,
            job: 7,
            buf: Vec::new(),
            last: Instant::now(),
            gone: false,
            leaving: &leaving,
            started: Instant::now(),
            expected: None,
        };
        sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
        assert!(!sink.cancelled());
        *leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });
        assert!(sink.cancelled());
        assert!(sink.buf.is_empty(), "flushed before the job stops");
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
        hub.release(7, "off Wi-Fi").unwrap();
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
    }

    /// oxygenflow#76: with notice, a job the hub expects to finish within
    /// it runs on; one that will not, or whose time is unknown, is handed
    /// back at once, and any job still running when the notice is up.
    #[test]
    fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {
        use microvm::Sink;
        let (base, _) = fake_hub(vec![204; 8]);
        let hub = hub_at(base);
        let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));
        let sink = |expected: Option<u64>, ran: u64| LogSink {
            hub: &hub,
            job: 1,
            buf: Vec::new(),
            last: Instant::now(),
            gone: false,
            leaving: &leaving,
            started: Instant::now() - Duration::from_secs(ran),
            expected: expected.map(Duration::from_secs),
        };
        assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");
        assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");
        assert!(sink(None, 0).cancelled(), "no idea how long: handed back");
        *leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });
        assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");
    }

    /// oxygenflow#32: a quiet job asks the hub every few seconds whether
    /// it still runs; a 409 marks it cancelled, and nothing more is sent.
    #[test]
    fn a_quiet_job_learns_that_its_run_was_cancelled() {
        use microvm::Sink;
        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);
        let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
        let (base, seen) = fake_hub(vec![204]);
        let hub = hub_at(base);
        let mut sink = LogSink {
            hub: &hub,
            job: 1,
            buf: Vec::new(),
            last: quiet,
            gone: false,
            leaving: &NOT_LEAVING,
            started: Instant::now(),
            expected: None,
        };
        assert!(!sink.cancelled());
        assert!(!sink.cancelled(), "asked again only after the heartbeat");
        assert_eq!(n(&seen), 1);

        let (base, seen) = fake_hub(vec![409]);
        let hub = hub_at(base);
        let mut sink = LogSink {
            hub: &hub,
            job: 1,
            buf: Vec::new(),
            last: quiet,
            gone: false,
            leaving: &NOT_LEAVING,
            started: Instant::now(),
            expected: None,
        };
        assert!(sink.cancelled());
        sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
        sink.flush();
        assert!(sink.cancelled());
        assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
    }

    /// oxygenflow#49: a hub that is away for a while (a deploy restarting
    /// it) costs a job's uploads a wait, not the upload, and costs its log
    /// no lines.
    #[test]
    fn a_job_outlasts_a_hub_restart() {
        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);
        // A port nobody listens on yet: the hub is restarting.
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
        let base = format!("http://127.0.0.1:{port}");
        let hub = hub_at(base.clone());

        // Log lines sent while it is away are kept, and go when it is back.
        let mut sink = LogSink {
            hub: &hub,
            job: 1,
            buf: Vec::new(),
            last: Instant::now(),
            gone: false,
            leaving: &NOT_LEAVING,
            started: Instant::now(),
            expected: None,
        };
        for i in 0..450 {
            sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
        }
        sink.send();
        assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");
        let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);
        sink.send();
        assert!(sink.buf.is_empty() && !sink.gone);
        assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");

        // An upload waits for the hub to come back, through refused
        // connections and server errors alike.
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
        let url = format!("http://127.0.0.1:{port}/x");
        let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
        let later = std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(150));
            fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])
        });
        let quick = [Duration::from_millis(50); 20];
        let r = retrying(&url, &quick, true, || agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();
        assert_eq!(r.status(), 204);
        assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");
        // Without `connection`, a hub that cannot be reached ends it at once.
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
        let url = format!("http://127.0.0.1:{port}/x");
        assert!(retrying(&url, &quick, false, || agent.post(&url).send_string("{}").map_err(Box::new)).is_err());
        assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");
    }

    /// oxygenflow#50: the first stop drains (the second, not called here,
    /// exits at once).
    #[test]
    fn a_stop_drains_the_runner() {
        assert!(!stopping());
        on_stop(libc::SIGTERM);
        assert!(stopping(), "told to stop, it claims nothing more");
        // An embedded runner starts afresh, and stop() drains it the same way.
        STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
        stop();
        assert!(stopping());
        STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
    }

    /// Lines kept for a hub that stays away are bounded, saying how many went.
    #[test]
    fn unsent_log_lines_are_bounded() {
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
        let hub = hub_at(format!("http://127.0.0.1:{port}"));
        let mut sink = LogSink {
            hub: &hub,
            job: 1,
            buf: Vec::new(),
            last: Instant::now(),
            gone: false,
            leaving: &NOT_LEAVING,
            started: Instant::now(),
            expected: None,
        };
        sink.buf = (0..MAX_UNSENT as u64 + 10).map(|i| Record { t: i, stream: "out".into(), text: String::new() }).collect();
        sink.send();
        assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
        assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");
        assert_eq!(sink.buf[1].t, 10, "the oldest go first");
    }

    #[test]
    fn the_budget_admits_jobs_that_fit() {
        let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
        let used = |jobs, memory_mib, cpus| Used { jobs, memory_mib, cpus };
        assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
        assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
        assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
        assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
        let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
        assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
        assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
        let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
        assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
    }

    #[test]
    fn a_claim_never_asks_for_more_memory_than_the_host_has_free() {
        assert_eq!(claim_memory(None, Some(6541)), Some(6541), "idle: the host's free memory, not unlimited");
        assert_eq!(claim_memory(Some(16_384), Some(6541)), Some(6541));
        assert_eq!(claim_memory(Some(4096), Some(6541)), Some(4096), "the budget when it is smaller");
        assert_eq!(claim_memory(None, None), None, "nothing known: no limit, as before");
        assert_eq!(claim_memory(Some(4096), None), Some(4096));
    }
}