crates/hub/tests/flow.rs at 76f93eaefe45 · andymccall/oxygenhub

# 76f93eaefe45 oxygenhub / crates / hub / tests / flow.rs
260.7 KiB · Rust · last changed Flow priorities: an organisation's level, and a cap on expedited runs 21 days ago source blame history raw
//! OxygenFlow on the hub (hub spec §13.2), end to end: a real `oxy` push
//! starts a run, a runner claims jobs through the API in dependency order,
//! job tokens read only their repository, results become checks and pages.

use axum::body::Body;
use axum::http::{header, Request, StatusCode};
use http_body_util::BodyExt;
use oxy_core::protocol::{open_remote, RemoteOptions};
use oxy_core::repo::Repo;
use oxygenhub::config::Config;
use oxygenhub::repos::RepoName;
use oxygenhub::{db, Shared};
use std::path::Path;
use tower::ServiceExt;

const ROOTFS: &str = "36fc4e3238044e424219a2b9002bf61bbe1805fec14d52cf37e17fae77c9ff16";

fn hermetic() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
    ONCE.call_once(|| {
        let d = tempfile::tempdir().unwrap();
        std::env::set_var("OXY_HOME", d.path());
        DIR.set(d).ok();
    });
}

fn write(root: &Path, rel: &str, content: &str) {
    let p = root.join(rel);
    std::fs::create_dir_all(p.parent().unwrap()).unwrap();
    std::fs::write(p, content).unwrap();
}

struct Hub {
    app: Shared,
    base: String,
    _dir: tempfile::TempDir,
}

async fn hub() -> Hub {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    config.auth.push_tokens = vec!["site".into()];
    config.flow.runner_tokens = vec!["runner-secret".into(), "runner-two".into(), "runner-three".into()];
    config.flow.repos = vec!["andy/*".into()];
    config.registration.mode = "open".into();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    config.server.base_url = format!("http://127.0.0.1:{port}");
    let app = oxygenhub::build(config).await.unwrap();
    let router = oxygenhub::router(app.clone());
    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }
}

async fn call(app: &Shared, method: &str, path: &str, bearer: Option<&str>, json: Option<String>) -> (StatusCode, String) {
    let mut req = Request::builder().method(method).uri(path);
    if let Some(b) = bearer {
        req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));
    }
    let body = match json {
        Some(j) => {
            req = req.header(header::CONTENT_TYPE, "application/json");
            Body::from(j)
        }
        None => Body::empty(),
    };
    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();
    let status = res.status();
    let bytes = res.into_body().collect().await.unwrap().to_bytes();
    (status, String::from_utf8_lossy(&bytes).into_owned())
}

/// A GET's status, `Location` and `Content-Type`.
async fn head(app: &Shared, path: &str) -> (StatusCode, String, String) {
    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    let h = |k: header::HeaderName| res.headers().get(k).and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
    (res.status(), h(header::LOCATION), h(header::CONTENT_TYPE))
}

async fn claim(app: &Shared) -> Option<serde_json::Value> {
    let (st, body) =
        call(app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"test-1","runtimes":["microvm"]}"#.into())).await;
    match st {
        StatusCode::NO_CONTENT => None,
        StatusCode::OK => Some(serde_json::from_str(&body).unwrap()),
        other => panic!("claim: {other} {body}"),
    }
}

fn receipt(job: &str, pipeline_hash: &str, exit: u32) -> serde_json::Value {
    serde_json::json!({
        "job_name": job, "pipeline_hash": pipeline_hash, "scope_tree_hash": "11".repeat(32),
        "inputs": [], "environment": "22".repeat(32), "outputs": [], "exit_code": exit, "duration_ms": 640
    })
}

async fn finish(app: &Shared, id: i64, body: serde_json::Value) -> StatusCode {
    call(app, "POST", &format!("/api/v1/flow/jobs/{id}/finish"), Some("runner-secret"), Some(body.to_string())).await.0
}

async fn create(app: &Shared, owner: &str, name: &str) {
    app.repos.create(&RepoName::parse(&format!("{owner}/{name}")).unwrap(), None).unwrap();
    db::repos::create(&app.db, owner, name, "", hub_api::Visibility::Public).await.unwrap();
}

fn client(url: &str, work: &Path) -> Repo {
    let mut c = Repo::init(work).unwrap();
    c.config.set("remote.origin.url", url).unwrap();
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();
    c
}

fn pipeline(extra: &str) -> String {
    format!(
        "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    depends-on \"build\"\n    exec \"sh\" \"-eu\" \"test.sh\"\n}}\njob \"lint\" {{\n    sandbox \"alpine\"\n    exec \"true\"\n}}\n{extra}"
    )
}

async fn runs_of(app: &Shared, owner: &str, name: &str) -> Vec<db::flow::Run> {
    let row = db::repos::get(&app.db, owner, name).await.unwrap().unwrap();
    for _ in 0..100 {
        let r = db::flow::runs(&app.db, row.id, 10, None).await.unwrap();
        if !r.is_empty() {
            return r;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    vec![]
}

/// The checks on a state once there are `n` of them (they follow the run).
async fn checks_when(app: &Shared, repo_id: i64, state: &str, n: usize) -> Vec<db::hooks::Check> {
    for _ in 0..100 {
        let c = db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap();
        if c.len() >= n {
            return c;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap()
}

#[tokio::test(flavor = "multi_thread")]
async fn a_push_runs_the_pipeline_through_a_runner() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    create(&h.app, "andy", "secret").await;
    create(&h.app, "other", "plain").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The push started a run with three queued jobs, and checks for them.
    let runs = runs_of(&h.app, "andy", "demo").await;
    assert_eq!(runs.len(), 1);
    let run = &runs[0];
    assert_eq!((run.bookmark.as_str(), run.status.as_str(), run.actor.as_str()), ("main", "queued", "site token"));
    assert_eq!(run.pipeline_hash.len(), 64);
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    assert_eq!(jobs.iter().map(|j| (j.name.as_str(), j.depth)).collect::<Vec<_>>(), [("build", 0), ("lint", 0), ("test", 1)]);
    // The checks are reported just after the run is recorded.
    let checks = checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(
        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),
        [("flow/build", "queued"), ("flow/lint", "queued"), ("flow/test", "queued")]
    );

    // Runners need a runner token.
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/claim", Some("site"), Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,
        StatusCode::UNAUTHORIZED
    );
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/claim", None, Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,
        StatusCode::UNAUTHORIZED
    );
    // A runner without the tier gets nothing.
    let (st, _) =
        call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"x","runtimes":["wasm"]}"#.into())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);

    // Claims come in dependency order; `test` waits for `build`.
    let a = claim(&h.app).await.unwrap();
    let b = claim(&h.app).await.unwrap();
    assert_eq!((a["job"]["name"].as_str(), b["job"]["name"].as_str()), (Some("build"), Some("lint")));
    assert!(claim(&h.app).await.is_none(), "test waits for build");
    assert_eq!(a["repo"], "andy/demo");
    assert_eq!(a["clone_url"], url.as_str());
    assert_eq!(a["sandbox"]["rootfs"], ROOTFS);
    assert_eq!(a["job"]["exec"], serde_json::json!(["make"]));
    assert_eq!(a["timeout_s"], 3600);
    // The job is told what it builds (FLOW_STATE, FLOW_BOOKMARK, FLOW_RUN).
    assert_eq!(
        (a["state"].as_str(), a["bookmark"].as_str(), a["run_number"].as_i64()),
        (Some(run.state.as_str()), Some("main"), Some(run.number))
    );
    let heads: Vec<String> = a["heads"].as_array().unwrap().iter().map(|v| v.as_str().unwrap().to_string()).collect();
    assert_eq!(heads.join("+"), run.heads);
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    assert_eq!(checks.iter().find(|c| c.name == "flow/build").unwrap().status, "running");

    // The job token reads this repository over the wire, and nothing else.
    let token = a["token"].as_str().unwrap().to_string();
    assert!(token.starts_with("flowjob_"));
    let opts = RemoteOptions { token: Some(token.clone()), ..Default::default() };
    open_remote(&url, &opts).unwrap().refs().unwrap();
    assert!(open_remote(&format!("{}/andy/secret", h.base), &opts).unwrap().refs().is_err(), "another repository");
    let bad = RemoteOptions { token: Some("flowjob_00".into()), ..Default::default() };
    assert!(open_remote(&url, &bad).unwrap().refs().is_err(), "an unknown job token");
    // … and cannot push.
    c.config.set("remote.origin.token", &token).unwrap();
    c.save_config().unwrap();
    write(work.path(), "more.txt", "x\n");
    c.snapshot("test").unwrap();
    c.commit("More", &[]).unwrap();
    let pushed = c.push("origin", false);
    assert!(pushed.is_err() || pushed.unwrap().iter().all(|r| !r.ok), "a job token cannot push");
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();

    // Logs, then results with receipts.
    let a_id = a["job_id"].as_i64().unwrap();
    let b_id = b["job_id"].as_i64().unwrap();
    let log = serde_json::json!({ "records": [
        { "t": 5, "stream": "flow", "text": "booting" },
        { "t": 120, "stream": "out", "text": "cc -o demo demo.c" },
        { "t": 130, "stream": "weird", "text": "x".repeat(20_000) },
    ]});
    assert_eq!(
        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{a_id}/log"), Some("runner-secret"), Some(log.to_string())).await.0,
        StatusCode::NO_CONTENT
    );
    let lines = db::flow::log(&h.app.db, a_id, 0, 100).await.unwrap();
    assert_eq!(lines.len(), 3);
    assert_eq!((lines[2].stream.as_str(), lines[2].text.len()), ("console", 16 * 1024), "streams are known ones, text is capped");
    let ph = a["pipeline_hash"].as_str().unwrap();
    assert_eq!(
        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("lint", ph, 0) })).await,
        StatusCode::BAD_REQUEST,
        "a receipt for another job"
    );
    assert_eq!(
        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("build", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700 })).await, StatusCode::CONFLICT, "only once");
    assert!(open_remote(&url, &opts).unwrap().refs().is_err(), "the token ends with its job");
    assert_eq!(
        finish(&h.app, b_id, serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": receipt("lint", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    let t = claim(&h.app).await.unwrap();
    assert_eq!(t["job"]["name"], "test");
    let t_id = t["job_id"].as_i64().unwrap();
    assert_eq!(
        finish(&h.app, t_id, serde_json::json!({ "exit_code": 3, "duration_ms": 500, "receipt": receipt("test", ph, 3) })).await,
        StatusCode::NO_CONTENT
    );

    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();
    assert_eq!(run.status, "failed");
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let build = jobs.iter().find(|j| j.name == "build").unwrap();
    assert_eq!((build.status.as_str(), build.exit_code, build.receipt_hash.len()), ("passed", Some(0), 64));
    assert_eq!(jobs.iter().find(|j| j.name == "test").unwrap().status, "failed");
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    assert_eq!(
        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),
        [("flow/build", "passed"), ("flow/lint", "passed"), ("flow/test", "failed")]
    );
    assert!(checks[2].summary.contains("exit 3") && checks[2].url.ends_with(&format!("/andy/demo/flow/{}", run.number)));

    // Pages: the list, the run with its graph and logs, and the live view.
    let (st, list) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(list.contains(&format!("/andy/demo/flow/{}", run.number)) && list.contains("st-failed"), "{list}");
    let (st, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("<svg class=\"flow-graph\""), "graph");
    assert_eq!(page.matches("data-job-node=").count(), 3);
    assert_eq!(page.matches("<path class=\"edge").count(), 1, "one dependency edge");
    assert!(page.contains("cc -o demo demo.c"), "log shown");
    assert!(!page.contains("data-flow-live"), "a finished run does not poll");
    let (st, live) = call(&h.app, "GET", &format!("/andy/demo/flow/{}/live?after={a_id}:1", run.number), None, None).await;
    assert_eq!(st, StatusCode::OK);
    let live: serde_json::Value = serde_json::from_str(&live).unwrap();
    assert_eq!(live["status"], "failed");
    let build_live = live["jobs"].as_array().unwrap().iter().find(|j| j["name"] == "build").unwrap().clone();
    assert_eq!(build_live["lines"].as_array().unwrap().len(), 2, "lines after seq 1");
    assert_eq!(call(&h.app, "GET", "/andy/demo/flow/999", None, None).await.0, StatusCode::NOT_FOUND);

    // A failed dependency skips what follows; an unrun job is an error.
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo changed\n");
    c.snapshot("test").unwrap();
    c.commit("Change the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut second = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 2 {
            second = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let second = second.expect("a second run");
    let x = claim(&h.app).await.unwrap();
    let y = claim(&h.app).await.unwrap();
    let (build_id, lint_id) = if x["job"]["name"] == "build" {
        (x["job_id"].as_i64().unwrap(), y["job_id"].as_i64().unwrap())
    } else {
        (y["job_id"].as_i64().unwrap(), x["job_id"].as_i64().unwrap())
    };
    assert_eq!(
        finish(&h.app, build_id, serde_json::json!({ "exit_code": null, "duration_ms": 10, "error": "rootfs is not on this runner" }))
            .await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(finish(&h.app, lint_id, serde_json::json!({ "exit_code": 0, "duration_ms": 10 })).await, StatusCode::NO_CONTENT);
    assert!(claim(&h.app).await.is_none(), "test was skipped");
    let jobs = db::flow::jobs(&h.app.db, second.id).await.unwrap();
    assert_eq!(
        jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect::<Vec<_>>(),
        [("build", "error"), ("lint", "passed"), ("test", "skipped")]
    );
    assert_eq!(db::flow::run_by_id(&h.app.db, second.id).await.unwrap().unwrap().status, "failed");

    // An invalid pipeline is a run in error with the line.
    write(work.path(), ".flow/pipeline.kdl", &pipeline("job \"broken\" {\n    sandbox \"missing\"\n    exec \"x\"\n}\n"));
    c.snapshot("test").unwrap();
    c.commit("Break the pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut third = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 3 {
            third = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let third = third.expect("a third run");
    assert_eq!(third.status, "error");
    assert!(third.error.contains("line 18") && third.error.contains("\"missing\""), "{}", third.error);
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", third.number), None, None).await;
    assert!(page.contains("The pipeline is not valid"));

    // State stamps (hub spec §13.2) and images in READMEs.
    write(work.path(), "README.md", "# Demo\n\n[![flow](.flow/stamp)](.flow/run)\n\n![logo](img/logo.png)\n");
    std::fs::create_dir_all(work.path().join("img")).unwrap();
    std::fs::write(work.path().join("img/logo.png"), b"\x89PNG\r\n\x1a\n\0\0\0\rIHDR\0\0\0\x01\0\0\0\x01\x08\x06\0\0\0\x1f\x15\xc4\x89")
        .unwrap();
    c.snapshot("test").unwrap();
    c.commit("A README with a stamp", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut fourth = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 4 {
            fourth = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let fourth = fourth.expect("a fourth run");
    let (_, overview) = call(&h.app, "GET", "/andy/demo", None, None).await;
    assert!(overview.contains("src=\"/andy/demo/raw/main/.flow/stamp\""), "{overview}");
    assert!(overview.contains("href=\"/andy/demo/file/main/.flow/run\""));
    assert!(overview.contains("src=\"/andy/demo/raw/main/img/logo.png\""), "images use the raw route");
    let (st, _, ct) = head(&h.app, "/andy/demo/raw/main/img/logo.png").await;
    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/png"));
    let (st, loc, _) = head(&h.app, "/andy/demo/raw/main/.flow/stamp").await;
    assert!(st.is_redirection() && loc == "/andy/demo/flow/stamp/main", "{st} {loc}");
    let (st, loc, _) = head(&h.app, "/andy/demo/flow/stamp/main").await;
    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/stamp/{}", fourth.heads), "a bookmark's stamp names its state: {loc}");
    let (st, _, ct) = head(&h.app, &loc).await;
    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/svg+xml; charset=utf-8"));
    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;
    assert!(svg.starts_with("<svg") && svg.contains("error ·") && svg.contains(&fourth.heads[..12]), "{svg}");
    let (_, svg1) = call(&h.app, "GET", &format!("/andy/demo/flow/stamp/{}", run.heads), None, None).await;
    assert!(svg1.contains("failed ·") && svg1.contains(&format!("run #{} failed", run.id)), "an older state keeps its own result: {svg1}");
    assert!(svg1.contains(&build.receipt_hash), "receipt hashes in the metadata");
    assert_eq!(svg1.matches("text-anchor=\"middle\"").count(), 3, "one glyph per job");
    let (st, loc, _) = head(&h.app, "/andy/demo/file/main/.flow/run").await;
    assert!(st.is_redirection() && loc == "/andy/demo/flow/at/main", "{loc}");
    let (st, loc, _) = head(&h.app, "/andy/demo/flow/at/main").await;
    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/{}", fourth.id), "{loc}");

    // A repository outside [flow] repos starts nothing.
    let work2 = tempfile::tempdir().unwrap();
    let c2 = client(&format!("{}/other/plain", h.base), work2.path());
    write(work2.path(), ".flow/pipeline.kdl", &pipeline(""));
    c2.snapshot("test").unwrap();
    c2.commit("Pipeline", &[]).unwrap();
    assert!(c2.push("origin", false).unwrap().iter().all(|r| r.ok));
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();
    assert!(db::flow::runs(&h.app.db, row.id, 10, None).await.unwrap().is_empty());
    let (_, page) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
    assert!(page.contains("not enabled for this repository"));
    let (_, loc, _) = head(&h.app, "/other/plain/flow/stamp/main").await;
    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;
    assert!(svg.contains("no run ·"), "{svg}");
    let (_, loc, _) = head(&h.app, "/other/plain/flow/at/main").await;
    assert_eq!(loc, "/other/plain/flow");
}

#[tokio::test(flavor = "multi_thread")]
async fn claims_carry_pinned_subrepos_and_the_job_token_reads_only_those() {
    let h = hub().await;
    for n in ["base", "lib", "vault", "other", "app"] {
        create(&h.app, "andy", n).await;
    }
    let push = |name: &str, file: &str| {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/{name}", h.base), work.path());
        write(work.path(), file, "x\n");
        c.snapshot("t").unwrap();
        c.commit("first", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        let heads = c.bookmark_heads().unwrap();
        (work, heads)
    };
    let (_b, base_heads) = push("base", "base.rs");
    // lib pins base in turn: a subrepo of a subrepo.
    let lib_work = tempfile::tempdir().unwrap();
    let lib = client(&format!("{}/andy/lib", h.base), lib_work.path());
    lib.submodule_add(&format!("{}/andy/base", h.base), "deps/base", Some("main")).unwrap();
    write(lib_work.path(), "lib.rs", "x\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib on base", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let lib_heads = lib.bookmark_heads().unwrap();
    let (_v, _) = push("vault", "secret.rs");
    let (_o, _) = push("other", "o.rs");

    // The app pins lib and vault as subrepos and pushes as the site token.
    let work = tempfile::tempdir().unwrap();
    let app = client(&format!("{}/andy/app", h.base), work.path());
    app.submodule_add(&format!("{}/andy/lib", h.base), "deps/lib", Some("main")).unwrap();
    // Vault is private: a checkout with no token for this hub cannot add
    // it, but the app's remote on the same hub lends its token (oxygen#32).
    // The subrepo keeps no copy: its own fetches find the parent's (oxygen#47).
    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'vault'").execute(&h.app.db).await.unwrap();
    let bare_work = tempfile::tempdir().unwrap();
    let mut bare = Repo::init(bare_work.path()).unwrap();
    bare.config.set("remote.origin.url", &format!("{}/andy/bare", h.base)).unwrap();
    bare.save_config().unwrap();
    assert!(bare.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).is_err(), "not without a token");
    app.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).unwrap();
    let nested = Repo::open(&work.path().join("deps/vault")).unwrap();
    assert!(nested.config.remote_tokens.is_empty(), "no token is copied into a subrepo");
    nested.fetch("origin").unwrap();
    // The pusher below is the site token, which has no account: it cannot read vault.
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "Makefile", "all:\n");
    app.snapshot("t").unwrap();
    app.commit("app with deps", &[]).unwrap();
    assert!(app.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "app").await.is_empty());

    let job = claim(&h.app).await.unwrap();
    let subs = job["subrepos"].as_array().cloned().unwrap_or_default();
    assert_eq!(subs.len(), 2, "lib and what lib pins; vault is private and the pusher has no account: {subs:?}");
    assert_eq!(subs[1]["path"], "deps/lib/deps/base", "nested under its parent, after it");
    assert_eq!(subs[1]["heads"], serde_json::json!(base_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));
    let _ = std::fs::remove_dir_all(work.path().join("deps/vault2"));
    assert_eq!(subs[0]["path"], "deps/lib");
    assert_eq!(subs[0]["repo"], "andy/lib");
    assert_eq!(subs[0]["heads"], serde_json::json!(lib_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));
    assert!(subs[0]["clone_url"].as_str().unwrap().ends_with("/andy/lib"));
    // The job's log says vault was left out, and why (oxygenhub#56).
    let job_id = job["job_id"].as_i64().unwrap();
    let lines = db::flow::log(&h.app.db, job_id, 0, 100).await.unwrap();
    assert!(
        lines.iter().any(|l| l.stream == "flow" && l.text.contains("subrepo deps/vault (andy/vault) is left out")),
        "{:?}",
        lines.iter().map(|l| &l.text).collect::<Vec<_>>()
    );

    // The job token reads its repository and the pinned lib, nothing else.
    let token = job["token"].as_str().unwrap().to_string();
    let read = |name: &str| {
        let opts = RemoteOptions { token: Some(token.clone()), ..Default::default() };
        open_remote(&format!("{}/andy/{name}", h.base), &opts).and_then(|r| r.refs()).is_ok()
    };
    assert!(read("app") && read("lib") && read("base"));
    assert!(!read("vault"), "not pinned for this pusher");
    assert!(!read("other"), "public, but not the job's to read");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_log_appends_never_fail_on_a_locked_database() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "Makefile", "all:\n");
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let job = db::flow::jobs(&h.app.db, run.id).await.unwrap().remove(0);
    // Many writers at once, each reading the count and then inserting: a
    // deferred transaction would fail with "database is locked".
    let mut tasks = Vec::new();
    for w in 0..16 {
        let pool = h.app.db.clone();
        tasks.push(tokio::spawn(async move {
            let records: Vec<flow_core::api::Record> =
                (0..20).map(|i| flow_core::api::Record { t: i, stream: "out".into(), text: format!("writer {w} line {i}") }).collect();
            db::flow::append_log(&pool, job.id, &records, 100_000).await
        }));
    }
    for t in tasks {
        assert_eq!(t.await.unwrap().unwrap(), 20, "every append succeeds");
    }
    let lines = db::flow::log(&h.app.db, job.id, 0, 1_000).await.unwrap();
    assert_eq!(lines.len(), 320);
    let seqs: std::collections::BTreeSet<i64> = lines.iter().map(|l| l.seq).collect();
    assert_eq!(seqs.len(), 320, "sequence numbers are unique");
}

/// A repository's own Flow setting overrides hub.toml either way.
#[tokio::test(flavor = "multi_thread")]
async fn a_repository_setting_turns_flow_on_or_off() {
    let h = hub().await;
    create(&h.app, "other", "plain").await;
    create(&h.app, "andy", "quiet").await;
    // A run elsewhere first, so hub-wide ids and per-repository numbers differ.
    create(&h.app, "andy", "first").await;
    {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/first", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        assert_eq!(runs_of(&h.app, "andy", "first").await.len(), 1);
    }
    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();
    db::repos::set_flow(&h.app.db, row.id, "on").await.unwrap();
    let quiet = db::repos::get(&h.app.db, "andy", "quiet").await.unwrap().unwrap();
    db::repos::set_flow(&h.app.db, quiet.id, "off").await.unwrap();
    for (owner, name) in [("other", "plain"), ("andy", "quiet")] {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/{owner}/{name}", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    }
    let plain = runs_of(&h.app, "other", "plain").await;
    assert_eq!(plain.len(), 1, "turned on");
    // andy/first has a job running, so plain's goes next (repositories take
    // turns, #244), and its run says where it stands.
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["repo"], "andy/first");
    let (_, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
    assert!(page.contains("Waiting for a runner: 1st in line") && page.contains("Running now: andy/first #1"), "{page}");
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/other/plain/flow/runs/1", None, None).await;
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["queue_position"], 1);
    let (_, list) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
    assert!(list.contains("queued, 1st in line"), "{list}");
    // Runs are numbered per repository: this is plain's first, whatever
    // other repositories ran before.
    assert_eq!(plain[0].number, 1);
    let checks = checks_when(&h.app, row.id, &plain[0].state, 3).await;
    assert!(checks.iter().all(|c| c.url.ends_with("/other/plain/flow/1")), "{checks:?}");
    let (st, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
    assert!(st == StatusCode::OK && page.contains("Run #1"), "{st}");
    assert!(runs_of(&h.app, "andy", "quiet").await.is_empty(), "turned off despite andy/*");
}

/// "Run pipeline" and "Re-run" (hub spec §13.2): runs that move no
/// bookmark, through the API and the pages.
#[tokio::test(flavor = "multi_thread")]
async fn runs_start_without_a_push_and_finished_runs_re_run() {
    let h = hub().await;
    // The site token acts as the first site administrator.
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let first = runs_of(&h.app, "andy", "demo").await;
    assert_eq!((first.len(), first[0].trigger.as_str()), (1, "push"));

    // Run pipeline on the default bookmark: a new run of the same state.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["number"], 2);
    assert!(v["url"].as_str().unwrap().ends_with("/andy/demo/flow/2"));
    let second = db::flow::run(&h.app.db, first[0].repo_id, 2).await.unwrap().unwrap();
    assert_eq!((second.trigger.as_str(), second.actor.as_str(), second.state.as_str()), ("manual", "andy", first[0].state.as_str()));
    assert_eq!(db::flow::jobs(&h.app.db, second.id).await.unwrap().len(), 3);
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/2", None, None).await;
    assert!(page.contains("started by andy"), "the run page says what started it");

    // Run pipeline on one change: its state, with no bookmark.
    let head = first[0].head_list()[0].to_string();
    let (st, body) =
        call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(format!(r#"{{"change":"{head}"}}"#))).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let on_change = db::flow::run(&h.app.db, first[0].repo_id, 3).await.unwrap().unwrap();
    assert_eq!(
        (on_change.bookmark.as_str(), on_change.heads.as_str(), on_change.state.as_str()),
        ("", head.as_str(), first[0].state.as_str())
    );
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/3", None, None).await;
    assert!(page.contains("title=\"Run on one change, not a bookmark\">change</span>"), "run page");
    let (st, body) =
        call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(format!(r#"{{"change":"{}"}}"#, "ab".repeat(32))))
            .await;
    assert!(st == StatusCode::CONFLICT && body.contains("no change"), "{st} {body}");
    // Viewers who cannot start runs are told why, and get no button.
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert!(page.contains("Sign in with write access to run the pipeline.") && !page.contains("primary-btn"));
    let (_, page) = call(&h.app, "GET", "/andy/demo/bookmarks", None, None).await;
    assert!(!page.contains("/flow/start"), "no run buttons for anonymous viewers");

    // A live run is not re-run; unknown bookmarks and runs are refused.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;
    assert!(st == StatusCode::CONFLICT && body.contains("still queued"), "{st} {body}");
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(r#"{"bookmark":"nope"}"#.into())).await;
    assert_eq!(st, StatusCode::CONFLICT);
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/99/rerun", Some("site"), None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);
    // Anonymous callers cannot start runs.
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", None, Some("{}".into())).await;
    assert_ne!(st, StatusCode::CREATED);

    // A finished run re-runs as a new run of the same state.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    for r in [&first[0], &second] {
        for j in db::flow::jobs(&h.app.db, r.id).await.unwrap() {
            sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE id = $1").bind(j.id).execute(&h.app.db).await.unwrap();
        }
        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = 1 WHERE id = $1").bind(r.id).execute(&h.app.db).await.unwrap();
    }
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let third = db::flow::run(&h.app.db, row.id, 4).await.unwrap().unwrap();
    assert_eq!(
        (third.trigger_label().as_str(), third.status.as_str(), third.state.as_str()),
        ("rerun:1", "queued", first[0].state.as_str())
    );
    assert_eq!(third.trigger, "rerun:1:push", "a rerun keeps the event it repeats");
    let checks = db::hooks::checks_for_state(&h.app.db, row.id, &third.state).await.unwrap();
    assert!(
        checks.iter().all(|c| c.status == "queued" && c.url.ends_with("/flow/4")),
        "the re-run's checks replace the old ones: {checks:?}"
    );
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/4", None, None).await;
    assert!(page.contains("re-run of #1 by andy"));
    assert!(!page.contains("RE-RUN"), "anonymous viewers get no button");

    // oxygenhub#130: one job, with what it depends on; an unknown one is
    // refused, naming the jobs a manual run can start.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(r#"{"jobs":["test"]}"#.into())).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let n = serde_json::from_str::<serde_json::Value>(&body).unwrap()["number"].as_i64().unwrap();
    let only = db::flow::run(&h.app.db, first[0].repo_id, n).await.unwrap().unwrap();
    let mut names: Vec<String> = db::flow::jobs(&h.app.db, only.id).await.unwrap().into_iter().map(|j| j.name).collect();
    names.sort();
    assert_eq!(names, ["build", "test"], "test and the job it depends on, not lint");
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(r#"{"jobs":["deploy"]}"#.into())).await;
    assert!(st == StatusCode::CONFLICT && body.contains("not in the pipeline") && body.contains("build, lint, test"), "{st} {body}");

    // With Flow off, nothing starts.
    db::repos::set_flow(&h.app.db, row.id, "off").await.unwrap();
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert!(st == StatusCode::CONFLICT && body.contains("not enabled"), "{body}");
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert!(page.contains("Flow is off for this repository"), "and the Flow tab says so");
}

/// The Flow API (hub spec §13.2): runs, a run with its jobs, a job's log,
/// checks by the heads a client has, and JSON for unknown API routes.
#[tokio::test(flavor = "multi_thread")]
async fn runs_jobs_and_logs_are_in_the_api() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    let head = c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

    // A job runs and logs.
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    let recs =
        serde_json::json!({ "records": [{ "t": 1, "stream": "stdout", "text": "one" }, { "t": 3, "stream": "stderr", "text": "three" }] });
    let (st, _) = call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(recs.to_string())).await;
    assert!(st.is_success(), "{st}");

    let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs", None, None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let runs: Vec<hub_api::FlowRun> = serde_json::from_str(&body).unwrap();
    assert_eq!(runs.len(), 1);
    assert_eq!((runs[0].number, runs[0].state.as_str(), runs[0].trigger.as_str()), (1, run.state.as_str(), "push"));
    assert!(runs[0].url.ends_with("/andy/demo/flow/1") && runs[0].jobs.is_empty());

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", None, None).await;
    let one: hub_api::FlowRun = serde_json::from_str(&body).unwrap();
    let names: Vec<(&str, &str)> = one.jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect();
    assert_eq!(names.len(), 3);
    let running = one.jobs.iter().find(|j| j.status == "running").unwrap();
    assert_eq!(running.name, a["job"]["name"].as_str().unwrap());
    assert!(one.jobs.iter().any(|j| j.name == "test" && j.depends_on == ["build"]));

    let job = running.name.clone();
    let (st, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log"), None, None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let lines: Vec<hub_api::FlowLogLine> = serde_json::from_str(&body).unwrap();
    assert!(lines.iter().any(|l| l.text == "three"), "{body}");
    let last = lines.last().unwrap().seq;
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log?after={last}"), None, None).await;
    assert_eq!(body, "[]", "nothing after the last line");
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/9", None, None).await.0, StatusCode::NOT_FOUND);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1/jobs/nope/log", None, None).await.0, StatusCode::NOT_FOUND);

    // Checks by the state hash, by the head `oxy log` shows, and by bookmark.
    let by = |key: String| {
        let app = h.app.clone();
        async move {
            let (st, body) = call(&app, "GET", &format!("/api/v1/repos/andy/demo/checks/{key}"), None, None).await;
            assert_eq!(st, StatusCode::OK, "{key}: {body}");
            serde_json::from_str::<Vec<hub_api::Check>>(&body).unwrap().len()
        }
    };
    checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(by(run.state.clone()).await, 3);
    assert_eq!(by(head.to_hex()).await, 3, "a single head names its state");
    assert_eq!(by("main".into()).await, 3);
    let (st, _) = call(&h.app, "GET", "/api/v1/repos/andy/demo/checks/no-such-bookmark", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);

    // An unknown API route answers with a problem document.
    let (st, body) = call(&h.app, "GET", "/api/v1/users/andy/repos", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["status"], 404, "{body}");
}

async fn get_bytes(app: &Shared, path: &str) -> (StatusCode, Vec<u8>) {
    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    let status = res.status();
    (status, res.into_body().collect().await.unwrap().to_bytes().to_vec())
}

#[tokio::test(flavor = "multi_thread")]
async fn signed_receipts_are_checked_kept_and_shown() {
    use oxy_core::hash::{hash_object, ChangeHash, ObjectRef, ObjectType};
    use oxy_core::identity::SecretKey;
    use oxy_core::objects::StateObject;
    use oxy_core::receipt::SignedReceipt;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let state = hash_object(ObjectType::State, &StateObject::new(heads.clone()).encode());
    let repo_id = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().id().unwrap();
    let sk = SecretKey::generate();
    let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);
    let sign = |r: &serde_json::Value, state: ObjectRef, sk: &SecretKey, id: &oxy_core::identity::Identity| {
        let r: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();
        let run_url = format!("{url}/flow/{}", run.number);
        hex::encode(SignedReceipt::sign(r.to_oxy(), repo_id, state, &run_url, id, sk).encode())
    };

    let a = claim(&h.app).await.unwrap();
    let b = claim(&h.app).await.unwrap();
    let ph = a["pipeline_hash"].as_str().unwrap().to_string();
    let (a_id, b_id) = (a["job_id"].as_i64().unwrap(), b["job_id"].as_i64().unwrap());
    // The first signed receipt binds the runner token to its key, and is kept.
    let rb = receipt("build", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": rb, "signed_receipt": sign(&rb, state, &sk, &runner) });
    assert_eq!(finish(&h.app, a_id, body).await, StatusCode::NO_CONTENT);
    // Another key under the same token: the result counts, the receipt is not kept.
    let other = SecretKey::generate();
    let rl = receipt("lint", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": rl,
        "signed_receipt": sign(&rl, state, &other, &other.identity("mallory", "", 1)) });
    assert_eq!(finish(&h.app, b_id, body).await, StatusCode::NO_CONTENT);
    // The right key, but for another state.
    let t = claim(&h.app).await.unwrap();
    let rt = receipt("test", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 500, "receipt": rt,
        "signed_receipt": sign(&rt, ObjectRef([9; 32]), &sk, &runner) });
    assert_eq!(finish(&h.app, t["job_id"].as_i64().unwrap(), body).await, StatusCode::NO_CONTENT);

    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let job = |n: &str| jobs.iter().find(|j| j.name == n).unwrap().clone();
    assert_eq!(job("build").status, "passed");
    assert_eq!((job("build").receipt_attest.len(), job("build").receipt_signer.as_str()), (64, "test-1 (OxygenFlow runner)"));
    assert_eq!(job("lint").status, "passed", "the job's result stands");
    assert!(job("lint").receipt_attest.is_empty());
    assert!(job("lint").receipt_note.contains("not the key this runner signs with"), "{}", job("lint").receipt_note);
    assert!(job("test").receipt_note.contains("another state"), "{}", job("test").receipt_note);

    // The repository keeps it, and anyone who can read the repository can
    // download it and check it on their own.
    let kept = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().receipts(&heads).unwrap();
    assert_eq!(kept.len(), 1);
    let attest = job("build").receipt_attest;
    let (st, bytes) = get_bytes(&h.app, &format!("/andy/demo/receipts/{attest}")).await;
    assert_eq!(st, StatusCode::OK);
    let sr = SignedReceipt::decode(&bytes).unwrap();
    sr.verify().unwrap();
    assert_eq!((sr.state(), sr.receipt.job_name.as_str()), (state, "build"));
    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/receipts/{}", "0".repeat(64))).await.0, StatusCode::NOT_FOUND);

    // Pages: the run names the signer or why a receipt was not kept; the
    // change whose state it is lists it.
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert!(page.contains("signed receipt</a> by test-1"), "run page");
    assert!(page.contains("receipt not kept"), "run page");
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/change/{}", heads[0].to_hex()), None, None).await;
    assert!(page.contains("<dt>Receipts</dt>") && page.contains(&format!("/andy/demo/receipts/{attest}")), "change page");
}

async fn upload(app: &Shared, job: i64, name: &str, bytes: &[u8]) -> (StatusCode, String) {
    let req = Request::post(format!("/api/v1/flow/jobs/{job}/artifacts/{name}"))
        .header(header::AUTHORIZATION, "Bearer runner-secret")
        .header(header::CONTENT_TYPE, "application/octet-stream")
        .body(Body::from(bytes.to_vec()))
        .unwrap();
    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();
    let status = res.status();
    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
}

#[tokio::test(flavor = "multi_thread")]
async fn artifacts_are_kept_named_by_the_receipt_and_served() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Build a tool", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["job"]["outputs"], serde_json::json!([{ "name": "tool", "path": "/src/out/tool" }]), "the runner is told");
    let id = a["job_id"].as_i64().unwrap();

    let body: Vec<u8> = (0..300_000u32).map(|i| (i % 253) as u8).collect();
    let (st, text) = upload(&h.app, id, "other", &body).await;
    assert_eq!(st, StatusCode::BAD_REQUEST, "an undeclared artifact: {text}");
    let (st, text) = upload(&h.app, id, "tool", &body).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    assert_eq!(blob, oxy_core::repo::Repo::blob_ref(&body).to_hex(), "the same blob the runner computes");

    // The receipt must name exactly the uploaded artifacts.
    let mut r = receipt("build", a["pipeline_hash"].as_str().unwrap(), 0);
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r.clone() })).await,
        StatusCode::BAD_REQUEST
    );
    r["outputs"] = serde_json::json!([blob]);
    assert_eq!(finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r })).await, StatusCode::NO_CONTENT);

    // The run page lists it; anyone who can read the repository downloads it.
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    let link = format!("/andy/demo/flow/{}/artifacts/build/tool", run.number);
    assert!(page.contains(&link), "run page lists the artifact");
    let (st, got) = get_bytes(&h.app, &link).await;
    assert_eq!((st, got == body), (StatusCode::OK, true));
    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/flow/{}/artifacts/build/none", run.number)).await.0, StatusCode::NOT_FOUND);

    // The nightly gc keeps it (#126): only the database names it, so with
    // no grace period a gc that did not root it would prune it at once.
    {
        let name = RepoName::parse("andy/demo").unwrap();
        let mut hosted = oxy_core::repo::Repo::open(&h.app.repos.dir(&name)).unwrap();
        hosted.config.set("core.gc-grace", "0").unwrap();
        hosted.save_config().unwrap();
    }
    let done = oxygenhub::maintenance::gc_all(&h.app).await;
    assert!(done.iter().all(|d| !d.failed), "{done:?}");
    let (st, got) = get_bytes(&h.app, &link).await;
    assert_eq!((st, got == body), (StatusCode::OK, true), "the artifact survives gc");

    // A writer attaches the artifact to a release from the run page; the
    // release then names the run and job that built it, as does the API.
    let form = |f: String| {
        Request::post("/register").header(header::CONTENT_TYPE, "application/x-www-form-urlencoded").body(Body::from(f)).unwrap()
    };
    let res = oxygenhub::router(h.app.clone())
        .oneshot(form("username=andy&email=a%40example.com&password=correct-horse-battery".into()))
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let with_cookie = |method: &str, path: &str, body: Option<String>| {
        let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie.clone());
        if body.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()
    };
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    db::releases::create(&h.app.db, row.id, "v1", "One", "", false, None).await.unwrap();
    let res = oxygenhub::router(h.app.clone()).oneshot(with_cookie("GET", &format!("/andy/demo/flow/{}", run.number), None)).await.unwrap();
    let page = String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned();
    let attach = format!("/andy/demo/flow/{}/artifacts/build/tool/attach", run.number);
    assert!(page.contains(&attach), "a writer is offered the release");
    let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
    let csrf = page[i..].split('"').next().unwrap().to_string();
    let release_id = db::releases::list(&h.app.db, row.id).await.unwrap()[0].id;
    let res = oxygenhub::router(h.app.clone())
        .oneshot(with_cookie("POST", &attach, Some(format!("_csrf={csrf}&release={release_id}"))))
        .await
        .unwrap();
    assert_eq!(res.status(), StatusCode::SEE_OTHER);
    let (_, page) = call(&h.app, "GET", "/andy/demo/releases/tag/v1", None, None).await;
    assert!(
        page.contains(">tool</a>") && page.contains(&format!("built by <a href=\"/andy/demo/flow/{0}\">run #{0}</a> build", run.number)),
        "{page}"
    );
    let (_, api) = call(&h.app, "GET", "/api/v1/repos/andy/demo/releases", None, None).await;
    let api: serde_json::Value = serde_json::from_str(&api).unwrap();
    let asset = &api[0]["assets"][0];
    assert_eq!(
        (asset["name"].as_str(), asset["built_by"]["run"].as_i64(), asset["built_by"]["job"].as_str()),
        (Some("tool"), Some(run.number), Some("build"))
    );
    assert!(asset["built_by"]["receipt"].is_null(), "that receipt was not signed");
}

/// oxygenflow#37: a failing job keeps a `when "always"` artifact; its
/// receipt names none, and the run page says the file is outside it.
async fn upload_report_file(app: &Shared, job: i64, name: &str, path: &str, bytes: &[u8]) -> (StatusCode, String) {
    let req = Request::post(format!("/api/v1/flow/jobs/{job}/reports/{name}/files"))
        .header(header::AUTHORIZATION, "Bearer runner-secret")
        .header(header::CONTENT_TYPE, "application/octet-stream")
        .header("X-Flow-Report-Path", path)
        .body(Body::from(bytes.to_vec()))
        .unwrap();
    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();
    let status = res.status();
    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
}

/// #127: a report's files arrive one by one, then its manifest as the
/// report's artifact; the hub indexes the declaration, metrics and files,
/// and gc keeps every file.
#[tokio::test(flavor = "multi_thread")]
async fn reports_are_kept_file_by_file_and_indexed() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"perf\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n        report \"frame-timing\" {{\n            path \"/src/out/report\"\n            when \"always\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Measure frames", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();

    let toml = b"kind = \"performance\"\ntitle = \"Frame timing\"\nsummary = \"Median 16.4 ms\"\n\n[metrics]\nmedian_ms = { value = 16.4, unit = \"ms\", better = \"lower\", fail_above = 20 }\n\n[context]\ntarget_fps = 30\n";
    let page = b"<p>frames</p>".to_vec();
    let data: Vec<u8> = (0..100_000u32).map(|i| (i % 241) as u8).collect();
    let (page_bytes, data_bytes) = (page.clone(), data.clone());
    let files: [(&str, &[u8]); 3] = [("data/frames.csv", &data), ("index.html", &page), ("report.toml", toml)];

    assert_eq!(upload_report_file(&h.app, id, "tool", "report.toml", toml).await.0, StatusCode::BAD_REQUEST, "not a report");
    assert_eq!(upload_report_file(&h.app, id, "frame-timing", "../x", toml).await.0, StatusCode::BAD_REQUEST, "not a clean path");
    let mut listed = vec![];
    for (path, bytes) in files {
        let (st, text) = upload_report_file(&h.app, id, "frame-timing", path, bytes).await;
        assert_eq!(st, StatusCode::OK, "{text}");
        let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
        assert_eq!(blob, oxy_core::repo::Repo::blob_ref(bytes).to_hex());
        listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
    }
    let mut missing = listed.clone();
    missing.push(serde_json::json!({ "path": "assets/x.js", "blob": "ab".repeat(32), "size": 1 }));
    let (st, text) =
        upload(&h.app, id, "frame-timing", &serde_json::to_vec(&serde_json::json!({ "version": 1, "files": missing })).unwrap()).await;
    assert_eq!(st, StatusCode::BAD_REQUEST, "a file never uploaded: {text}");
    assert!(text.contains("assets/x.js"), "{text}");
    let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed })).unwrap();
    let (st, text) = upload(&h.app, id, "frame-timing", &manifest).await;
    assert_eq!(st, StatusCode::OK, "{text}");

    let (kind, title, summary, files_n, bytes): (String, String, String, i64, i64) =
        sqlx::query_as("SELECT kind, title, summary, files, bytes FROM reports WHERE job_id = $1 AND name = 'frame-timing'")
            .bind(id)
            .fetch_one(&h.app.db)
            .await
            .unwrap();
    assert_eq!((kind.as_str(), title.as_str(), summary.as_str()), ("performance", "Frame timing", "Median 16.4 ms"));
    assert_eq!((files_n, bytes), (3, (data.len() + page.len() + toml.len()) as i64));
    let rid: i64 = sqlx::query_scalar("SELECT id FROM reports WHERE job_id = $1").bind(id).fetch_one(&h.app.db).await.unwrap();
    let m = oxygenhub::reports::metrics(&h.app.db, rid).await.unwrap();
    assert_eq!((m[0].key.as_str(), m[0].value, m[0].fail_above), ("median_ms", 16.4, Some(20.0)));
    let log = db::flow::log(&h.app.db, id, 0, 1000).await.unwrap();
    assert!(log.iter().any(|r| r.text.contains("report frame-timing: performance \"Frame timing\", 1 metric")), "{log:?}");

    // The threshold is a check on the run's state, passing at 16.4.
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    let c = checks.iter().find(|c| c.name == "report/frame-timing/median_ms").expect("a check per threshold");
    assert_eq!(c.status, "passed", "{c:?}");
    assert!(c.summary.contains("16.4 ms") && c.url.contains("/reports/frame-timing?id="), "{c:?}");

    // The Reports tab, the report's page, and its own page in a sandbox.
    let (_, page) = call(&h.app, "GET", "/andy/demo", None, None).await;
    assert!(page.contains(">Reports</a>"), "the tab");
    let (st, page) = call(&h.app, "GET", "/andy/demo/reports", None, None).await;
    assert!(st == StatusCode::OK && page.contains("Frame timing") && page.contains("median_ms"), "{st}");
    let (st, page) = call(&h.app, "GET", "/andy/demo/reports/frame-timing", None, None).await;
    assert!(st == StatusCode::OK && page.contains("fails above 20") && page.contains("sandbox=\"allow-scripts\""), "{st}");
    let src = page.split("class=\"rep-frame\" src=\"").nth(1).unwrap().split('"').next().unwrap().to_string();
    assert!(src.ends_with("/index.html"), "{src}");
    let res = oxygenhub::router(h.app.clone()).oneshot(Request::get(&src).body(Body::empty()).unwrap()).await.unwrap();
    assert_eq!(res.status(), StatusCode::OK);
    let csp = res.headers()[header::CONTENT_SECURITY_POLICY].to_str().unwrap().to_string();
    assert!(csp.starts_with("sandbox allow-scripts;") && csp.contains("default-src 'none'"), "{csp}");
    // The site's own policy and frame refusal stay off it: report frames are framed (#222).
    assert!(
        res.headers().get(header::X_FRAME_OPTIONS).is_none() && res.headers().get(header::CONTENT_SECURITY_POLICY_REPORT_ONLY).is_none()
    );
    assert_eq!(res.headers()[header::X_CONTENT_TYPE_OPTIONS], "nosniff");
    assert_eq!(res.headers()[header::ACCESS_CONTROL_ALLOW_ORIGIN], "*");
    // The page as kept, with the frame script that sizes the frame added.
    let served = res.into_body().collect().await.unwrap().to_bytes().to_vec();
    assert!(served.starts_with(&page_bytes), "the report's page");
    assert!(String::from_utf8_lossy(&served).contains("/static/report-frame.js"), "with the frame script");
    let (st, _) = get_bytes(&h.app, "/static/report.css").await;
    assert_eq!(st, StatusCode::OK, "the stylesheet a report may link");
    let data = src.replace("/index.html", "/data/frames.csv");
    assert_eq!(get_bytes(&h.app, &data).await, (StatusCode::OK, data_bytes.clone()));
    // A link with another report's MAC, or a file it lacks, finds nothing.
    // A second run of the report: the timeline has two stops, and the
    // first can be compared with it (RP4).
    let first: i64 = sqlx::query_scalar("SELECT id FROM reports WHERE name = 'frame-timing'").fetch_one(&h.app.db).await.unwrap();
    let second: i64 = sqlx::query_scalar(
        "INSERT INTO reports (repo_id, name, kind, title, summary, context, run_id, job_id, run_number, job, bookmark, state, manifest, files, bytes, created_ms) \
         SELECT repo_id, name, kind, title, 'Median 18.0 ms', context, run_id, NULL, 2, job, bookmark, state, manifest, files, bytes, created_ms + 1000 FROM reports WHERE id = $1 RETURNING id",
    )
    .bind(first)
    .fetch_one(&h.app.db)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO report_metrics (report_id, key, value, unit, better, fail_above) VALUES ($1, 'median_ms', 18.0, 'ms', 'lower', 20)",
    )
    .bind(second)
    .execute(&h.app.db)
    .await
    .unwrap();
    let (_, page) = call(&h.app, "GET", "/andy/demo/reports/frame-timing", None, None).await;
    assert_eq!(page.matches("data-tl-stop").count(), 2, "a stop per run");
    assert!(page.contains("Median 18.0 ms"), "the newest run is shown");
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/reports/frame-timing?id={first}&compare={second}&play=3"), None, None).await;
    assert!(page.contains("This run against run #2") && page.contains("-1.6"), "every metric against the other run");
    let stop = page.split("data-tl-stop").next().unwrap_or("").rsplit("href=").next().unwrap_or("").to_string();
    assert!(page.contains("data-play=\"3\""), "playing");
    assert!(stop.contains("play") && stop.contains('3'), "playing carries on from stop to stop: {stop}");
    assert!(page.contains("data-report-runs="), "the frame interface knows the runs");
    let forged = src.replacen("/files/", "/files/9", 1);
    assert_eq!(get_bytes(&h.app, &forged).await.0, StatusCode::NOT_FOUND);
    assert_eq!(get_bytes(&h.app, &src.replace("/index.html", "/secret.txt")).await.0, StatusCode::NOT_FOUND);
    let _ = run;

    // gc keeps every file: only the database names them.
    {
        let name = RepoName::parse("andy/demo").unwrap();
        let mut hosted = oxy_core::repo::Repo::open(&h.app.repos.dir(&name)).unwrap();
        hosted.config.set("core.gc-grace", "0").unwrap();
        hosted.save_config().unwrap();
    }
    let done = oxygenhub::maintenance::gc_all(&h.app).await;
    assert!(done.iter().all(|d| !d.failed), "{done:?}");
    let hosted = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap();
    for (path, bytes) in files {
        assert_eq!(hosted.read_blob(&oxy_core::repo::Repo::blob_ref(bytes)).unwrap(), bytes, "{path} survives gc");
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn a_failing_job_keeps_an_artifact_outside_its_receipt() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    exec \"make\" \"test\"\n    output {{\n        artifact \"junit\" {{\n            path \"/src/out/junit.xml\"\n            when \"always\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["job"]["outputs"][0]["always"], true, "the runner is told to keep it from a failing job");
    let id = a["job_id"].as_i64().unwrap();
    // A JUnit report (#114): parsed on upload, counted, failures listed.
    let xml = br#"<testsuites><testsuite name="pytest"><testcase classname="tests.test_io" name="test_read" time="0.5"/><testcase classname="tests.test_io" name="test_write" time="1.5"><failure message="disk full">E   OSError: disk full</failure></testcase></testsuite></testsuites>"#;
    let (st, text) = upload(&h.app, id, "junit", xml).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    let mut r = receipt("test", a["pipeline_hash"].as_str().unwrap(), 1);
    r["outputs"] = serde_json::json!([blob]);
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 1, "duration_ms": 700, "receipt": r.clone() })).await,
        StatusCode::BAD_REQUEST,
        "a failing receipt names no outputs"
    );
    r["outputs"] = serde_json::json!([]);
    assert_eq!(finish(&h.app, id, serde_json::json!({ "exit_code": 1, "duration_ms": 700, "receipt": r })).await, StatusCode::NO_CONTENT);
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert!(page.contains(&format!("/andy/demo/flow/{}/artifacts/test/junit", run.number)), "listed");
    assert!(page.contains("not in its receipt"), "marked");
    assert!(page.contains("2 tests: 1 failed"), "counts beside the job");
    assert!(page.contains("tests.test_io::test_write") && page.contains("disk full") && page.contains("OSError"), "the failure");
    assert!(page.contains("tests: 2 tests: 1 failed in 2.0 s"), "said in the job's log");
}

/// A release's trust report (hub spec §9.1): a signed tag, an asset built
/// by a job with a signed receipt from the tagged state, and one uploaded
/// by hand; the page, its JSON export and the API.
#[tokio::test(flavor = "multi_thread")]
async fn a_release_is_audited_from_authors_to_receipts() {
    use oxy_core::hash::{hash_object, ChangeHash, ObjectType};
    use oxy_core::identity::SecretKey;
    use oxy_core::objects::StateObject;
    use oxy_core::receipt::SignedReceipt;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&url, work.path());
    c.create_identity("Andy", "[email protected]", oxy_core::repo::IdentityScope::Repo).unwrap();
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Build a tool", &[]).unwrap();
    let heads = c.bookmark_heads().unwrap();
    c.tag_create("v1", &heads, "first").unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The runner builds, uploads the artifact and signs its receipt.
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    let body: Vec<u8> = b"#!/bin/sh\necho tool\n".to_vec();
    let (_, text) = upload(&h.app, id, "tool", &body).await;
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    let mut r = receipt("build", a["pipeline_hash"].as_str().unwrap(), 0);
    r["outputs"] = serde_json::json!([blob]);
    let heads_run: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let state = hash_object(ObjectType::State, &StateObject::new(heads_run).encode());
    let repo_id = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().id().unwrap();
    let sk = SecretKey::generate();
    let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);
    let fr: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();
    let signed = hex::encode(SignedReceipt::sign(fr.to_oxy(), repo_id, state, &format!("{url}/flow/1"), &runner, &sk).encode());
    let done = serde_json::json!({ "exit_code": 0, "duration_ms": 900, "receipt": r, "signed_receipt": signed });
    assert_eq!(finish(&h.app, id, done).await, StatusCode::NO_CONTENT);

    // A release with the built asset and a file uploaded by hand.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let rel = db::releases::create(&h.app.db, row.id, "v1", "One", "", false, None).await.unwrap();
    oxygenhub::web::releases::store_asset(&h.app, rel.id, "tool", "", &body).await.unwrap();
    oxygenhub::web::releases::store_asset(&h.app, rel.id, "notes.txt", "", b"by hand\n").await.unwrap();

    let (st, page) = call(&h.app, "GET", "/andy/demo/releases/trust/v1", None, None).await;
    assert_eq!(st, StatusCode::OK);
    for want in [
        "Tag v1 is signed",
        "The one change in the release is authenticated",
        "tool: signed receipt from run #1 build",
        "a runner key of this hub",
        "tool: built from the tagged state",
        "tool: pinned environment",
        "notes.txt: no build found",
        "<svg class=\"trust-graph\"",
        // oxygenhub#79: a summary of badges, badges in the graph, and the
        // graph drawn at its own size rather than stretched to the page.
        "Tag signed by",
        "1 of 1 changes signed or sealed",
        "1 of 2 assets built and attested by OxygenFlow",
        "attested build",
        "uploaded by hand",
        "receipt verified",
        "class=\"shackle\"",
        "<title>notes.txt:",
    ] {
        assert!(page.contains(want), "missing {want:?}");
    }
    let svg = &page[page.find("<svg class=\"trust-graph\"").unwrap()..];
    assert!(svg[..200].contains(" width=\"") && svg[..200].contains(" height=\""), "the graph has its own size: {}", &svg[..200]);
    let (_, releases) = call(&h.app, "GET", "/andy/demo/releases", None, None).await;
    assert!(releases.contains("/andy/demo/releases/trust/v1"), "the release links its audit");

    // The export carries the signed receipt, which checks on its own.
    let (st, json) = call(&h.app, "GET", "/andy/demo/releases/trust/v1?format=json", None, None).await;
    assert_eq!(st, StatusCode::OK);
    let t: serde_json::Value = serde_json::from_str(&json).unwrap();
    assert_eq!((t["tag"].as_str(), t["tag_signature"].as_str(), t["changes"].as_i64()), (Some("v1"), Some("valid"), Some(1)));
    assert_eq!(t["authors"][0]["who"]["name"], "Andy");
    let tool = t["assets"].as_array().unwrap().iter().find(|a| a["name"] == "tool").unwrap();
    let hex_receipt = tool["build"]["signed_receipt"].as_str().unwrap();
    let sr = SignedReceipt::decode(&hex::decode(hex_receipt).unwrap()).unwrap();
    sr.verify().unwrap();
    assert_eq!(sr.receipt.outputs[0].to_hex(), blob);
    let audit = t["audit"].as_array().unwrap();
    assert!(audit.iter().any(|c| c["verdict"] == "warn" && c["title"].as_str().unwrap().starts_with("notes.txt")));
    assert!(!audit.iter().any(|c| c["verdict"] == "fail"), "{audit:?}");
    // The API gives the same report; an unknown release is a 404.
    let (st, api) = call(&h.app, "GET", "/api/v1/repos/andy/demo/releases/trust/v1", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert_eq!(serde_json::from_str::<serde_json::Value>(&api).unwrap()["assets"].as_array().unwrap().len(), 2);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/releases/trust/v9", None, None).await.0, StatusCode::NOT_FOUND);
    assert_eq!(call(&h.app, "GET", "/andy/demo/releases/trust/v9", None, None).await.0, StatusCode::NOT_FOUND);
}

/// Runners (hub spec §13.2): an administrator makes a registration token,
/// a runner registers with it (binding its key), claims, shows on Admin →
/// Runners with its history, and is drained, disabled and removed.
#[tokio::test(flavor = "multi_thread")]
async fn runners_register_show_their_history_and_are_drained_and_removed() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    // A signed-in site administrator.
    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let admin = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();
    db::accounts::set_admin(&h.app.db, admin.id, true).await.unwrap();
    let web = |method: &'static str, path: String, body: Option<String>| {
        let (app, cookie) = (h.app.clone(), cookie.clone());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            let st = res.status();
            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    let (st, page) = web("GET", "/admin/runners".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("No runners yet"));
    assert_eq!(call(&h.app, "GET", "/admin/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // A one-time registration token, shown once with the command.
    let (_, page) = web("POST", "/admin/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    assert!(token.starts_with("flowreg_"), "{token}");

    // The runner registers, binding its key; the token then is spent.
    let sk = SecretKey::generate();
    let identity = sk.identity("plex (OxygenFlow runner)", "", 1);
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": ["big-disk"], "max_jobs": 2, "memory_mib": 16384, "cpus": 4 });
    let reg = serde_json::json!({ "token": token, "name": "plex", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    assert!(credential.starts_with("flowrunner_"));
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0,
        StatusCode::FORBIDDEN,
        "used once"
    );

    // It claims with its credential; the hub.toml runner shows too.
    let claim_as = |who: String| {
        let app = h.app.clone();
        let info = info.clone();
        async move {
            let body = serde_json::json!({ "runner": "ignored", "runtimes": ["microvm"], "info": info }).to_string();
            call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await
        }
    };
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "nothing queued");
    assert_eq!(claim(&h.app).await, None);
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(
        page.contains(">plex</strong>") && page.contains("st-online") && page.contains("x86_64 · linux") && page.contains("big-disk"),
        "{page}"
    );
    assert!(page.contains(">test-1</strong>") && page.contains("hub.toml"), "the configured runner, by its claimed name");
    let plex = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();
    assert_eq!(plex.key, identity.hash().to_hex(), "bound at registration");

    // A run: plex takes a job, and its history shows it.
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;
    let (st, body) = claim_as(credential.clone()).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let job: serde_json::Value = serde_json::from_str(&body).unwrap();
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    assert!(
        page.contains("andy/demo") && page.contains(job["job"]["name"].as_str().unwrap()) && page.contains("Running now</dt><dd>1 job"),
        "{page}"
    );
    // The queue on Admin → Runners and through the API (#41, part 4).
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(page.contains("id=\"queue\"") && page.contains(">plex</a></td>"), "{page}");
    // Connectivity and what it is doing, kept apart (#243); its status
    // opens the queue filtered to it, and the filters keep and drop jobs.
    assert!(page.contains("running 1 of ") && page.contains("/admin/runners?runner=plex#queue"), "{page}");
    let (_, only) = web("GET", "/admin/runners?runner=plex".into(), None).await;
    assert!(only.contains(">plex</a></td>") && only.contains(" of ") && only.contains(">all</a>"), "{only}");
    let (_, none) = web("GET", "/admin/runners?runner=nobody&status=running".into(), None).await;
    assert!(none.contains("Nothing matches."), "{none}");
    // A busy runner claims nothing; its heartbeat keeps it online.
    sqlx::query("UPDATE flow_runners SET last_seen_ms = 0 WHERE id = $1").bind(plex.id).execute(&h.app.db).await.unwrap();
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(page.contains("st-offline"), "quiet for long: offline");
    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/heartbeat", Some(&credential), None).await.0, StatusCode::NO_CONTENT);
    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/heartbeat", Some("flowrunner_nope"), None).await.0, StatusCode::UNAUTHORIZED);
    let plex_now = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();
    assert_eq!(plex_now.connectivity(), "online");
    let (_, tokens) = web("GET", "/settings/tokens".into(), None).await;
    let (_, made) = web("POST", "/settings/tokens".into(), Some(format!("_csrf={}&name=admin&scopes=user%3Aread", csrf_of(&tokens)))).await;
    let i = made.find("oxyhub_").unwrap();
    let admin_token: String = made[i..].chars().take_while(|c| c.is_ascii_alphanumeric() || *c == '_').collect();
    let (st, body) = call(&h.app, "GET", "/api/v1/runners", Some(&admin_token), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let listed: serde_json::Value = serde_json::from_str(&body).unwrap();
    let plex_row = listed.as_array().unwrap().iter().find(|r| r["name"] == "plex").unwrap();
    assert_eq!((plex_row["running"].as_i64(), plex_row["arch"].as_str()), (Some(1), Some("x86_64")));
    // A runner's own labels and time limit from its page (#98).
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    let (st, _) = web(
        "POST",
        format!("/admin/runners/{}/limits", plex.id),
        Some(format!("_csrf={}&labels=gpu%2C+plex&timeout_minutes=5", csrf_of(&page))),
    )
    .await;
    assert_eq!(st, StatusCode::SEE_OTHER);
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    assert!(page.contains(">gpu<") && page.contains("5 min for its jobs"), "{page}");
    let (_, body) = call(&h.app, "GET", "/api/v1/runners", Some(&admin_token), None).await;
    let listed: serde_json::Value = serde_json::from_str(&body).unwrap();
    let plex_row = listed.as_array().unwrap().iter().find(|r| r["name"] == "plex").unwrap();
    assert!(plex_row["labels"].as_array().unwrap().iter().any(|l| l == "gpu"), "{plex_row}");
    let (st, _) = web(
        "POST",
        format!("/admin/runners/{}/limits", plex.id),
        Some(format!("_csrf={}&labels=bad%20label%21&timeout_minutes=", csrf_of(&page))),
    )
    .await;
    assert_eq!(st, StatusCode::BAD_REQUEST, "a label with a space or punctuation is refused");
    let (_, body) = call(&h.app, "GET", "/api/v1/runners/queue", Some(&admin_token), None).await;
    let queue: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert!(queue.as_array().unwrap().iter().any(|q| q["status"] == "running" && q["runner"] == "plex"), "{queue}");
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/runners/{}/jobs", plex.id), Some(&admin_token), None).await;
    let jobs: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(jobs[0]["repo"], "andy/demo", "{jobs}");
    assert_eq!(call(&h.app, "GET", "/api/v1/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // Draining: no new work; removing while busy is refused.
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    let csrf = csrf_of(&page);
    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=draining"))).await;
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "a draining runner takes nothing");
    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("is running jobs: drain it first"), "{page}");
    // Its job finishes; disabled, it is refused; removed, its credential is dead.
    let id = job["job_id"].as_i64().unwrap();
    let done = call(
        &h.app,
        "POST",
        &format!("/api/v1/flow/jobs/{id}/finish"),
        Some(&credential),
        Some(r#"{"exit_code":0,"duration_ms":5}"#.into()),
    )
    .await;
    assert_eq!(done.0, StatusCode::NO_CONTENT);
    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=disabled"))).await;
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::FORBIDDEN);
    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("Removed plex."), "{page}");
    assert_eq!(claim_as(credential).await.0, StatusCode::UNAUTHORIZED);
}

/// Runner scopes (oxygenhub#41): a runner registered for one repository, or
/// one owner, takes only those jobs; a site runner takes any.
#[tokio::test(flavor = "multi_thread")]
async fn scoped_runners_take_only_their_scope() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    create(&h.app, "andy", "other").await;
    let register = |scope: db::runners::Scope, name: &'static str| {
        let app = h.app.clone();
        async move {
            let token = format!("flowreg_{name}");
            db::runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &scope).await.unwrap();
            let identity = SecretKey::generate().identity(name, "", 1);
            let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
            let reg = serde_json::json!({ "token": token, "name": name, "identity": hex::encode(identity.encode()), "info": info });
            let (st, body) = call(&app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
            assert_eq!(st, StatusCode::OK, "{body}");
            serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string()
        }
    };
    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let only_demo = register(db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) }, "demo-runner").await;
    let carols = register(db::runners::Scope { owner: "carol".into(), repo_id: None }, "carol-runner").await;
    let andys = register(db::runners::Scope { owner: "andy".into(), repo_id: None }, "andy-runner").await;
    let claim_as = |who: String| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"] }).to_string();
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await;
            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["repo"].as_str().unwrap().to_string())
        }
    };
    // Jobs queued in andy/other only.
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/other", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "other").await.is_empty());
    assert_eq!(claim_as(only_demo.clone()).await, None, "nothing of andy/demo is queued");
    assert_eq!(claim_as(carols).await, None, "nothing of carol's");
    assert_eq!(claim_as(andys).await.as_deref(), Some("andy/other"), "the owner's runner takes its repositories' jobs");
    // The site runner (hub.toml) takes any.
    assert_eq!(claim(&h.app).await.unwrap()["repo"], "andy/other");
    assert_eq!(claim_as(only_demo).await, None, "still nothing of its own");
}

/// Settings → Runners (oxygenhub#41, part 2b): a repository's admin adds a
/// runner for that repository alone, sees which runners serve it, and
/// removes only its own.
#[tokio::test(flavor = "multi_thread")]
async fn repository_admins_add_and_remove_their_own_runners() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=andy%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    create(&h.app, "andy", "demo").await;
    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let web = |method: &'static str, path: String, body: Option<String>| {
        let (app, cookie) = (h.app.clone(), cookie.clone());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            let st = res.status();
            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    let (st, page) = web("GET", "/andy/demo/settings".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("<h2>Runners</h2>"), "the owner sees the section");
    let (_, page) = web("POST", "/andy/demo/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let identity = SecretKey::generate().identity("mine", "", 1);
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "mine", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let mine = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "mine").unwrap();
    assert_eq!(mine.scope(), db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) });
    // The site's runner (hub.toml, from its first claim) serves it too, but is not the owner's to remove.
    claim(&h.app).await;
    let site = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "test-1").unwrap();
    let (_, page) = web("GET", "/andy/demo/settings".into(), None).await;
    assert!(page.contains("<strong>mine</strong>") && page.contains("this repository"), "{page}");
    assert!(page.contains("<strong>test-1</strong>") && page.contains("the whole site"), "{page}");
    let csrf = csrf_of(&page);
    let (st, _) = web("POST", format!("/andy/demo/settings/runners/{}/remove", site.id), Some(format!("_csrf={csrf}"))).await;
    assert_eq!(st, StatusCode::NOT_FOUND, "not the repository's own");
    let (_, page) = web("POST", format!("/andy/demo/settings/runners/{}/remove", mine.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("mine removed"), "{page}");
    assert!(db::runners::get(&h.app.db, mine.id).await.unwrap().is_none());

    // The same for an organisation's owner.
    let (_, page) = web("GET", "/orgs/new".into(), None).await;
    web("POST", "/orgs/new".into(), Some(format!("_csrf={}&name=acme&display_name=Acme", csrf_of(&page)))).await;
    let (st, page) = web("GET", "/orgs/acme/settings".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    let (_, page) = web("POST", "/orgs/acme/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let identity = SecretKey::generate().identity("acme-runner", "", 1);
    let reg = serde_json::json!({ "token": token, "name": "acme-runner", "identity": hex::encode(identity.encode()), "info": info });
    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0, StatusCode::OK);
    let acme = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "acme-runner").unwrap();
    assert_eq!(acme.scope(), db::runners::Scope { owner: "acme".into(), repo_id: None });
    let (_, page) = web("GET", "/orgs/acme/settings".into(), None).await;
    assert!(page.contains("<strong>acme-runner</strong>") && page.contains("the organisation's repositories"), "{page}");
    let (_, page) = web("POST", format!("/orgs/acme/settings/runners/{}/remove", acme.id), Some(format!("_csrf={}", csrf_of(&page)))).await;
    assert!(page.contains("acme-runner removed"), "{page}");
}

/// runs-on (oxygenhub#41, part 3): a job that asks for a label goes only to
/// a runner that has it, and waits for one otherwise.
#[tokio::test(flavor = "multi_thread")]
async fn jobs_that_ask_for_labels_wait_for_a_runner_with_them() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    let cross = "job \"cross\" {\n    sandbox \"alpine\"\n    runs-on \"riscv64\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(cross));
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "demo").await.is_empty());
    // The site runner has no labels: it takes the others, never cross.
    let mut taken = Vec::new();
    while let Some(j) = claim(&h.app).await {
        taken.push(j["job"]["name"].as_str().unwrap().to_string());
    }
    assert!(!taken.contains(&"cross".to_string()) && taken.contains(&"lint".to_string()), "{taken:?}");
    let queue = db::flow::queue(&h.app.db).await.unwrap();
    let waiting = queue.iter().find(|q| q.job == "cross").unwrap();
    assert_eq!((waiting.status.as_str(), waiting.needs()), ("queued", vec!["riscv64".to_string()]));
    // A runner registered with the label takes it.
    let token = "flowreg_riscv";
    db::runners::add_registration(&h.app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &Default::default()).await.unwrap();
    let identity = SecretKey::generate().identity("visionfive", "", 1);
    let info = serde_json::json!({ "arch": "riscv64", "os": "linux", "version": "0.1.0", "labels": ["riscv64"], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "visionfive", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    let claim_body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"], "info": info }).to_string();
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(claim_body)).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"], "cross");
}

/// oxygenhub#71: a job its runner stopped reporting (restarted, or lost
/// the hub as the job ended) is ended once it has run past the time limit
/// and a grace: an error with the reason, its dependents skipped and its
/// run settled, as if the runner had reported the error.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_no_runner_reports_is_ended_after_the_time_limit() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "lost").await;
    let url = format!("{}/andy/lost", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "lost").await.len(), 1);
    let job = claim(&h.app).await.expect("a job to claim");
    let id = job["job_id"].as_i64().unwrap();

    // Running, but well within the limit: left alone.
    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 0);
    // Started long ago and never reported: ended.
    let old = db::now_ms() - 24 * 3600 * 1000;
    sqlx::query("UPDATE flow_jobs SET started_ms = $1 WHERE id = $2").bind(old).bind(id).execute(&h.app.db).await.unwrap();
    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 1);
    let j = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert!(j.error.starts_with("lost:"), "{}", j.error);
    let run = &runs_of(&h.app, "andy", "lost").await[0];
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    assert!(jobs.iter().all(|j| j.status != "running"), "nothing left running");
    // A late report, should one come, does not overwrite the ending.
    assert_ne!(finish(&h.app, id, serde_json::json!({ "exit_code": 0 })).await, StatusCode::NO_CONTENT);
    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "error");
}

/// oxygenhub#73: a runner that lists the images it holds is offered only
/// the jobs whose rootfs is among them; one that does not say gets any.
#[tokio::test(flavor = "multi_thread")]
async fn a_runner_is_offered_only_jobs_whose_image_it_holds() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "images").await;
    let url = format!("{}/andy/images", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "images").await.len(), 1);
    let claim_with = |images: Vec<String>| {
        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "images": images }).to_string();
        let app = h.app.clone();
        async move {
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await;
            (st, body)
        }
    };
    let (st, _) = claim_with(vec!["ab".repeat(32)]).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "no job for a runner without the image");
    let jobs = db::flow::jobs(&h.app.db, runs_of(&h.app, "andy", "images").await[0].id).await.unwrap();
    let rootfs: serde_json::Value = serde_json::from_str(&jobs[0].sandbox).unwrap();
    let rootfs = rootfs["rootfs"].as_str().unwrap().to_string();
    let (st, body) = claim_with(vec!["ab".repeat(32), rootfs]).await;
    assert_eq!(st, StatusCode::OK, "{body}");
}

/// oxygenhub#71: a runner that lists the jobs it holds has the others it
/// was given ended at once (it restarted, or never received them), not at
/// the time limit; a held job is left alone.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_its_runner_no_longer_holds_is_ended_at_its_next_claim() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "held").await;
    let url = format!("{}/andy/held", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "held").await.len(), 1);
    let id = claim(&h.app).await.expect("a job to claim")["job_id"].as_i64().unwrap();
    let claim_holding = |holding: Vec<i64>| {
        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "holding": holding }).to_string();
        let app = h.app.clone();
        async move { call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await.0 }
    };
    claim_holding(vec![id]).await;
    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "running", "a held job stays");
    claim_holding(vec![]).await;
    let j = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert!(j.error.starts_with("lost:") && j.error.contains("no longer holds"), "{}", j.error);
    let run = &runs_of(&h.app, "andy", "held").await[0];
    assert!(db::flow::jobs(&h.app.db, run.id).await.unwrap().iter().all(|j| j.status != "running"), "the run settles");
}

/// oxygen#59: a job with a `scope` runs on a push only when the push's
/// changes touch a path it names (a release job, when Cargo.toml's version
/// moves); a bookmark's first run and manual runs have every job.
#[tokio::test(flavor = "multi_thread")]
async fn a_scoped_job_runs_only_when_a_push_touches_its_paths() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "scoped").await;
    let url = format!("{}/andy/scoped", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let extra = "job \"dist\" {\n    sandbox \"alpine\"\n    scope \"Cargo.toml\" \"packaging/**\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    write(work.path(), "Cargo.toml", "version = \"0.1.0\"\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let names = |app: Shared, n: usize| async move {
        let mut runs = Vec::new();
        for _ in 0..100 {
            runs = runs_of(&app, "andy", "scoped").await;
            if runs.len() >= n {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert_eq!(runs.len(), n, "runs");
        let mut j: Vec<String> = db::flow::jobs(&app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
        j.sort();
        // Passed, so the next push is scoped against it, not superseding it.
        sqlx::query("UPDATE flow_runs SET status = 'passed' WHERE id = $1").bind(runs[0].id).execute(&app.db).await.unwrap();
        j
    };
    assert_eq!(names(h.app.clone(), 1).await, ["build", "dist", "lint", "test"], "the first run has every job");
    write(work.path(), "test.sh", "echo still ok\n");
    c.commit("Edit the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(names(h.app.clone(), 2).await, ["build", "lint", "test"], "dist is out of scope");
    write(work.path(), "Cargo.toml", "version = \"0.1.1\"\n");
    c.commit("Release 0.1.1", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(names(h.app.clone(), 3).await, ["build", "dist", "lint", "test"], "a version bump runs dist");
}

/// oxygenflow#32: a run still going is cancelled from the API, the web or
/// by a newer push to its bookmark; its runner is refused from then on,
/// and the next push is scoped past the cancelled run.
#[tokio::test(flavor = "multi_thread")]
async fn runs_are_cancelled_by_hand_or_by_a_newer_push() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "stop").await;
    let url = format!("{}/andy/stop", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let extra = "job \"dist\" {\n    sandbox \"alpine\"\n    scope \"Cargo.toml\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    write(work.path(), "Cargo.toml", "version = \"0.1.0\"\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let wait_runs = |app: Shared, n: usize| async move {
        for _ in 0..100 {
            let runs = runs_of(&app, "andy", "stop").await;
            if runs.len() >= n {
                return runs;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        panic!("no run {n}");
    };
    let first = wait_runs(h.app.clone(), 1).await;
    sqlx::query("UPDATE flow_runs SET status = 'passed' WHERE id = $1").bind(first[0].id).execute(&h.app.db).await.unwrap();
    sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE run_id = $1").bind(first[0].id).execute(&h.app.db).await.unwrap();

    // Run 2 (a version bump) has a job running when run 3 is pushed.
    write(work.path(), "Cargo.toml", "version = \"0.1.1\"\n");
    c.commit("Release 0.1.1", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    wait_runs(h.app.clone(), 2).await;
    let job = claim(&h.app).await.expect("a job of run 2");
    let id = job["job_id"].as_i64().unwrap();
    write(work.path(), "test.sh", "echo still ok\n");
    c.commit("Edit the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let runs = wait_runs(h.app.clone(), 3).await;
    let third = &runs[0];
    let mut second = runs[1].clone();
    for _ in 0..100 {
        second = db::flow::run_by_id(&h.app.db, runs[1].id).await.unwrap().unwrap();
        if !second.live() {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(second.status, "cancelled", "a newer push supersedes a run still going");
    let jobs = db::flow::jobs(&h.app.db, second.id).await.unwrap();
    assert!(jobs.iter().all(|j| j.status == "cancelled" && j.error == "superseded by run #3"), "{jobs:?}");
    let mut names: Vec<String> = db::flow::jobs(&h.app.db, third.id).await.unwrap().into_iter().map(|j| j.name).collect();
    names.sort();
    assert_eq!(names, ["build", "dist", "lint", "test"], "run 3 also checks what run 2 would have: the version bump");
    let mut checks = vec![];
    for _ in 0..100 {
        checks = db::hooks::checks_for_state(&h.app.db, second.repo_id, &second.state).await.unwrap();
        if checks.iter().all(|c| c.status == "cancelled") {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert!(checks.iter().all(|c| c.status == "cancelled"), "{checks:?}");
    // The runner of the cancelled job is refused, so it stops the job.
    let (st, _) =
        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(r#"{"records":[]}"#.into())).await;
    assert_eq!(st, StatusCode::CONFLICT);

    // By hand: the API, then a run already ended is refused.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", Some("site"), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let third = db::flow::run_by_id(&h.app.db, third.id).await.unwrap().unwrap();
    assert_eq!(third.status, "cancelled");
    assert!(db::flow::jobs(&h.app.db, third.id).await.unwrap().iter().all(|j| j.error == "cancelled by andy"));
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", Some("site"), None).await;
    assert!(st == StatusCode::CONFLICT && body.contains("already cancelled"), "{st} {body}");
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", None, None).await;
    assert_ne!(st, StatusCode::OK, "anonymous callers cannot cancel");
    let (_, page) = call(&h.app, "GET", "/andy/stop/flow/3", None, None).await;
    assert!(page.contains("⊘") && !page.contains("CANCEL</button>"), "a cancelled run shows so, with no button");
}

/// oxygenflow#23 and oxygenhub#89: a job runs under its pipeline's
/// timeout, capped at the site's maximum, else the site's default; and the
/// hub gives a job up only past its own limit.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_runs_under_its_own_time_limit_within_the_sites() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    db::accounts::set_setting(&h.app.db, oxygenhub::flow::SETTING_DEFAULT, "30").await.unwrap();
    db::accounts::set_setting(&h.app.db, oxygenhub::flow::SETTING_MAX, "120").await.unwrap();
    create(&h.app, "andy", "limits").await;
    let url = format!("{}/andy/limits", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let extra = "job \"slow\" {\n    sandbox \"alpine\"\n    timeout \"90m\"\n    exec \"true\"\n}\njob \"huge\" {\n    sandbox \"alpine\"\n    timeout \"5h\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "limits").await.len(), 1);
    let mut limits = std::collections::HashMap::new();
    let mut ids = std::collections::HashMap::new();
    while let Some(j) = claim(&h.app).await {
        let name = j["job"]["name"].as_str().unwrap().to_string();
        limits.insert(name.clone(), j["timeout_s"].as_u64().unwrap());
        ids.insert(name, j["job_id"].as_i64().unwrap());
    }
    assert_eq!(limits["slow"], 5400, "its own");
    assert_eq!(limits["huge"], 7200, "cut to the site's maximum");
    assert_eq!(limits["build"], 1800, "the site's default");
    // 40 minutes in: past the default and its grace, within slow's limit.
    let started = db::now_ms() - 40 * 60 * 1000;
    sqlx::query("UPDATE flow_jobs SET started_ms = $1").bind(started).execute(&h.app.db).await.unwrap();
    oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap();
    let status = |id: i64| {
        let app = h.app.clone();
        async move { db::flow::job(&app.db, id).await.unwrap().unwrap().status }
    };
    assert_eq!(status(ids["build"]).await, "error", "past its 30 minutes");
    assert_eq!(status(ids["slow"]).await, "running", "within its 90");
    assert_eq!(status(ids["huge"]).await, "running", "within its 120");
}

/// oxygenflow#24: a job's `on` decides which events start it: a pushed
/// tag runs only the jobs that name it; a push to a bookmark the pipeline
/// does not name starts no run; a rerun repeats its run's event.
#[tokio::test(flavor = "multi_thread")]
async fn on_decides_which_events_start_a_job() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "events").await;
    let url = format!("{}/andy/events", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let text = format!(
        "on {{\n    push \"main\"\n    manual\n}}\n{}job \"dist\" {{\n    sandbox \"alpine\"\n    on {{\n        tag \"v*\"\n    }}\n    exec \"true\"\n}}\n",
        pipeline("")
    );
    // `on` must come after schema-version; pipeline() starts with it.
    let text = text.replacen(
        "on {\n    push \"main\"\n    manual\n}\nschema-version 1\n",
        "schema-version 1\non {\n    push \"main\"\n    manual\n}\n",
        1,
    );
    write(work.path(), ".flow/pipeline.kdl", &text);
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let jobs_of_latest = |app: Shared, n: usize| async move {
        let mut runs = Vec::new();
        for _ in 0..100 {
            runs = runs_of(&app, "andy", "events").await;
            if runs.len() >= n {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert_eq!(runs.len(), n, "runs");
        let mut j: Vec<String> = db::flow::jobs(&app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
        j.sort();
        (runs[0].clone(), j)
    };
    let (_, j) = jobs_of_latest(h.app.clone(), 1).await;
    assert_eq!(j, ["build", "lint", "test"], "a push to main: no dist");
    // A tag: only dist.
    let heads = c.bookmark_heads().unwrap();
    c.tag_create("v1.0.0", &heads, "first").unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let (run, j) = jobs_of_latest(h.app.clone(), 2).await;
    assert_eq!(j, ["dist"], "a tag runs only what names it");
    assert_eq!(run.trigger, "tag:v1.0.0");
    // A rerun repeats the tag's event.
    let row = db::repos::get(&h.app.db, "andy", "events").await.unwrap().unwrap();
    db::flow::finish(
        &h.app.db,
        db::flow::jobs(&h.app.db, run.id).await.unwrap()[0].id,
        db::flow::Outcome { status: "passed", exit_code: Some(0), error: "", receipt: "", receipt_hash: "" },
    )
    .await
    .unwrap();
    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();
    oxygenhub::flow::rerun(&h.app, &row, &run, "andy").await.unwrap();
    let (_, j) = jobs_of_latest(h.app.clone(), 3).await;
    assert_eq!(j, ["dist"], "the rerun is the tag's run again");
    // A bookmark the pipeline does not name starts nothing.
    c.bookmark_create("develop", None).unwrap();
    assert!(c.push_bookmark("origin", "develop", false).unwrap().iter().all(|r| r.ok));
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert_eq!(runs_of(&h.app, "andy", "events").await.len(), 3, "no run for develop");
}

/// oxygenflow#24: a job with a `schedule` line runs on the default
/// bookmark when the line comes due, and only then; other jobs stay out.
#[tokio::test(flavor = "multi_thread")]
async fn scheduled_jobs_run_when_due() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "nightly").await;
    let url = format!("{}/andy/nightly", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    // An odd minute, so the hub's own ticker cannot fire it during the test.
    let extra = "job \"nightly\" {\n    sandbox \"alpine\"\n    on {\n        schedule \"17 4 29 2 *\"\n    }\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "nightly").await.len(), 1);
    let first = runs_of(&h.app, "andy", "nightly").await;
    let names: Vec<String> = db::flow::jobs(&h.app.db, first[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
    assert!(!names.contains(&"nightly".to_string()), "a push does not run a scheduled job");
    use flow_core::pipeline::When;
    let not_yet = When { minute: 16, hour: 4, day: 29, month: 2, weekday: 4 };
    assert_eq!(oxygenhub::flow::run_due(&h.app, not_yet).await, 0);
    let due = When { minute: 17, ..not_yet };
    assert_eq!(oxygenhub::flow::run_due(&h.app, due).await, 1);
    let runs = runs_of(&h.app, "andy", "nightly").await;
    assert_eq!(runs.len(), 2);
    assert_eq!(runs[0].trigger, "schedule");
    let names: Vec<String> = db::flow::jobs(&h.app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
    assert_eq!(names, ["nightly"]);
    assert_eq!(runs[0].bookmark, "main", "on the default bookmark");
}

#[tokio::test(flavor = "multi_thread")]
async fn variables_from_three_levels_reach_a_job_inner_overriding_outer() {
    use oxygenhub::levels::Scope;
    use oxygenhub::variables;
    let h = hub().await;
    let app = &h.app;
    // Andy, the site's administrator, owns the acme organisation.
    let res = oxygenhub::router(app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=a%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let andy = db::accounts::account_by_name(&app.db, "andy").await.unwrap().unwrap();
    db::accounts::create_org(&app.db, "acme", "Acme", andy.id).await.unwrap();
    create(app, "acme", "app").await;
    let page = |path: &str, form: Option<String>| {
        let mut req =
            Request::builder().method(if form.is_some() { "POST" } else { "GET" }).uri(path).header(header::COOKIE, cookie.clone());
        if form.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        let req = req.body(Body::from(form.unwrap_or_default())).unwrap();
        let app = app.clone();
        async move {
            let res = oxygenhub::router(app).oneshot(req).await.unwrap();
            let status = res.status();
            let body = String::from_utf8(res.into_body().collect().await.unwrap().to_bytes().to_vec()).unwrap();
            (status, body)
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };

    // The site sets CC and LANG, the organisation its own CC, the
    // repository RUST_LOG from its settings page.
    variables::set(&app.db, Scope::Site, "CC", "gcc").await.unwrap();
    variables::set(&app.db, Scope::Site, "LANG", "C").await.unwrap();
    let org = db::accounts::org_by_name(&app.db, "acme").await.unwrap().unwrap();
    variables::set(&app.db, Scope::Org(org.id), "CC", "clang").await.unwrap();
    let (st, settings) = page("/acme/app/settings", None).await;
    assert_eq!(st, StatusCode::OK, "{settings}");
    assert!(settings.contains("Inherited") && settings.contains("the acme organisation") && settings.contains("overridden"), "{settings}");
    let csrf = csrf_of(&settings);
    let (_, body) = page("/acme/app/settings/variables", Some(format!("_csrf={csrf}&action=set&name=RUST_LOG&value=debug"))).await;
    assert!(body.contains("RUST_LOG set.") && body.contains("<code>debug</code>"), "{body}");
    let (_, body) = page("/acme/app/settings/variables", Some(format!("_csrf={csrf}&action=set&name=FLOW_REPO&value=x"))).await;
    assert!(body.contains("not FLOW_"), "{body}");
    let repo = db::repos::get(&app.db, "acme", "app").await.unwrap().unwrap();
    db::repos::set_flow(&app.db, repo.id, "on").await.unwrap();
    let env = variables::env_for_repo(&app.db, &repo).await.unwrap();
    assert_eq!(
        env.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect::<Vec<_>>(),
        [("CC", "clang"), ("LANG", "C"), ("RUST_LOG", "debug")],
        "the organisation's CC wins over the site's"
    );

    // A claim carries them.
    let url = format!("{}/acme/app", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let runs = runs_of(app, "acme", "app").await;
    assert_eq!(runs.len(), 1, "{runs:?}");
    let got = claim(app).await.unwrap_or_else(|| panic!("a job: {:?}", runs[0]));
    assert_eq!(got["env"], serde_json::json!({ "CC": "clang", "LANG": "C", "RUST_LOG": "debug" }), "{got}");

    // Removing a site variable from the admin page takes effect at the next claim.
    let (_, admin) = page("/admin", None).await;
    assert!(admin.contains("Variables") && admin.contains("<code>gcc</code>"), "{admin}");
    let (_, body) = page("/admin/variables", Some(format!("_csrf={}&action=delete&name=LANG", csrf_of(&admin)))).await;
    assert!(body.contains("LANG removed."), "{body}");
    let got = claim(app).await.expect("another job");
    assert_eq!(got["env"], serde_json::json!({ "CC": "clang", "RUST_LOG": "debug" }), "{got}");
}

#[tokio::test(flavor = "multi_thread")]
async fn secrets_a_job_declares_come_with_the_claim_and_never_reach_the_log() {
    let h = hub().await;
    let app = &h.app;
    let res = oxygenhub::router(app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=a%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let page = |path: &str, form: Option<String>| {
        let mut req =
            Request::builder().method(if form.is_some() { "POST" } else { "GET" }).uri(path).header(header::COOKIE, cookie.clone());
        if form.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        let req = req.body(Body::from(form.unwrap_or_default())).unwrap();
        let app = app.clone();
        async move {
            let res = oxygenhub::router(app).oneshot(req).await.unwrap();
            let status = res.status();
            let body = String::from_utf8(res.into_body().collect().await.unwrap().to_bytes().to_vec()).unwrap();
            (status, body)
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    create(app, "andy", "demo").await;
    // The site sets deploy-token; the repository overrides it. Neither
    // page shows a value, only that it is set.
    let (_, admin) = page("/admin", None).await;
    let csrf = csrf_of(&admin);
    let (_, body) = page("/admin/secrets", Some(format!("_csrf={csrf}&action=set&name=deploy-token&value=site-level-value"))).await;
    assert!(body.contains("deploy-token set.") && !body.contains("site-level-value"), "{body}");
    let (_, body) = page("/admin/secrets", Some(format!("_csrf={csrf}&action=set&name=bad+name&value=x"))).await;
    assert!(body.contains("name is 1 to 64 of"), "{body}");
    let (_, settings) = page("/andy/demo/settings", None).await;
    assert!(
        settings.contains("Secrets") && settings.contains("<code>deploy-token</code>") && settings.contains("the site</td>"),
        "{settings}"
    );
    let csrf_r = csrf_of(&settings);
    let (_, body) =
        page("/andy/demo/settings/secrets", Some(format!("_csrf={csrf_r}&action=set&name=deploy-token&value=hunter22-repo-value"))).await;
    assert!(body.contains("deploy-token set.") && body.contains("overridden") && !body.contains("hunter22"), "{body}");
    let repo = db::repos::get(&app.db, "andy", "demo").await.unwrap().unwrap();
    assert!(
        !String::from_utf8_lossy(&oxygenhub::secrets::list(&app.db, oxygenhub::levels::Scope::Repo(repo.id)).await.unwrap()[0].value)
            .contains("hunter22"),
        "sealed at rest"
    );

    // A job that declares the secret, and one the levels do not hold.
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &pipeline("job \"deploy\" {\n    sandbox \"alpine\"\n    secret \"deploy-token\" {\n        mount-path \"/run/secrets/token\"\n    }\n    secret \"missing\" {\n        mount-path \"/run/secrets/missing\"\n    }\n    exec \"deploy\"\n}\n"),
    );
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let runs = runs_of(app, "andy", "demo").await;
    let jobs = match runs.first() {
        Some(r) => db::flow::jobs(&app.db, r.id).await.unwrap().iter().map(|j| format!("{}:{}", j.name, j.status)).collect::<Vec<_>>(),
        None => vec![],
    };
    let mut deploy = None;
    for _ in 0..50 {
        let Some(got) = claim(app).await else {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            continue;
        };
        if got["job"]["name"] == "deploy" {
            deploy = Some(got);
            break;
        }
        assert_eq!(got["secrets"], serde_json::Value::Null, "{got}");
    }
    let got = deploy.unwrap_or_else(|| panic!("the deploy job: runs {runs:?} jobs {jobs:?}"));
    assert_eq!(
        got["secrets"],
        serde_json::json!([{ "name": "deploy-token", "path": "/run/secrets/token", "value": "hunter22-repo-value" }]),
        "the repository's value, the missing one left out: {got}"
    );
    let id = got["job_id"].as_i64().unwrap();
    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();
    assert!(log.iter().any(|l| l.stream == "flow" && l.text.contains("secret \"missing\" is not set")), "{log:?}");

    // The value never reaches the stored log, plain or base64.
    let records = serde_json::json!({ "records": [
        { "t": 1, "stream": "out", "text": "token=hunter22-repo-value" },
        { "t": 2, "stream": "err", "text": "b64 aHVudGVyMjItcmVwby12YWx1ZQ== end" },
    ] });
    let (st, _) = call(app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(records.to_string())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);
    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();
    let texts: Vec<&str> = log.iter().map(|l| l.text.as_str()).collect();
    assert!(texts.contains(&"token=***") && texts.contains(&"b64 *** end"), "{texts:?}");
    assert!(!texts.iter().any(|t| t.contains("hunter22")), "{texts:?}");
}

#[tokio::test(flavor = "multi_thread")]
async fn a_job_s_identity_is_a_signed_token_the_discovery_document_lets_anyone_verify() {
    let h = hub().await;
    let app = &h.app;
    let (st, disc) = call(app, "GET", "/.well-known/openid-configuration", None, None).await;
    assert_eq!(st, StatusCode::OK, "{disc}");
    let disc: serde_json::Value = serde_json::from_str(&disc).unwrap();
    assert_eq!(disc["issuer"], h.base.as_str());
    assert_eq!(disc["jwks_uri"], format!("{}/.well-known/jwks.json", h.base));
    let (st, jwks) = call(app, "GET", "/.well-known/jwks.json", None, None).await;
    assert_eq!(st, StatusCode::OK, "{jwks}");
    let jwks: serde_json::Value = serde_json::from_str(&jwks).unwrap();
    assert_eq!(jwks["keys"][0]["alg"], "RS256");
    let kid = jwks["keys"][0]["kid"].as_str().unwrap().to_string();

    create(app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &pipeline("job \"deploy\" {\n    sandbox \"alpine\"\n    identity \"sts.amazonaws.com\" {\n        mount-path \"/run/secrets/aws-token\"\n    }\n    exec \"deploy\"\n}\n"),
    );
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut deploy = None;
    for _ in 0..50 {
        let Some(got) = claim(app).await else {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            continue;
        };
        if got["job"]["name"] == "deploy" {
            deploy = Some(got);
            break;
        }
    }
    let got = deploy.expect("the deploy job");
    let secrets = got["secrets"].as_array().expect("secrets");
    assert_eq!(secrets.len(), 1);
    assert_eq!(
        (secrets[0]["name"].as_str(), secrets[0]["path"].as_str()),
        (Some("identity:sts.amazonaws.com"), Some("/run/secrets/aws-token"))
    );
    let token = secrets[0]["value"].as_str().unwrap().to_string();
    let issuer = app.oidc().await.unwrap();
    assert_eq!(issuer.kid, kid, "the key the JWKS publishes signed it");
    let claims = issuer.verify(&token, "sts.amazonaws.com", db::now_ms()).expect("verifies");
    assert_eq!(claims["sub"], "repo:andy/demo:bookmark:main");
    assert_eq!(claims["repository"], "andy/demo");
    assert_eq!(claims["job"], "deploy");
    assert_eq!(claims["run_number"], 1);
    assert_eq!(claims["runner"], "test-1");
    assert_eq!(claims["iss"], h.base.as_str());
    assert!(claims["state"].as_str().unwrap().len() == 64);
    // A job that prints its token leaves *** in the log.
    let id = got["job_id"].as_i64().unwrap();
    let records = serde_json::json!({ "records": [{ "t": 1, "stream": "out", "text": format!("TOKEN={token} end") }] });
    let (st, _) = call(app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(records.to_string())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);
    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();
    assert!(log.iter().any(|l| l.text == "TOKEN=*** end"), "{log:?}");
}

#[tokio::test(flavor = "multi_thread")]
async fn jobs_are_offered_only_to_runners_of_their_architecture() {
    let h = hub().await;
    let app = &h.app;
    create(app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline("sandbox \"board\" {\n    rootfs \"cas://blob/{ROOTFS}\"\n    arch \"riscv64\"\n}\njob \"flash\" {\n    sandbox \"board\"\n    exec \"flash\"\n}\n").replace("{ROOTFS}", &"ab".repeat(32)));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let claim_as = |arch: &'static str, archs: serde_json::Value| {
        let app = app.clone();
        async move {
            let body = serde_json::json!({
                "runner": format!("{arch}-1"), "runtimes": ["microvm"],
                "info": { "arch": arch, "os": "linux", "version": "t", "labels": [], "max_jobs": 1 },
                "image_archs": archs,
            });
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body.to_string())).await;
            match st {
                StatusCode::NO_CONTENT => None,
                StatusCode::OK => {
                    Some(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"].as_str().unwrap().to_string())
                }
                other => panic!("claim: {other} {body}"),
            }
        }
    };
    // An x86_64 runner that says the alpine image is x86_64: it takes the
    // alpine jobs, never the one that asks for riscv64.
    let rootfs = pipeline("").split("cas://blob/").nth(1).unwrap()[..64].to_string();
    for _ in 0..50 {
        if !runs_of(app, "andy", "demo").await.is_empty() {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
    let runs = runs_of(app, "andy", "demo").await;
    assert_eq!(runs.len(), 1, "{runs:?}");
    let mut taken = vec![];
    while let Some(j) = claim_as("x86_64", serde_json::json!({ rootfs.clone(): "x86_64" })).await {
        taken.push(j);
    }
    assert_eq!(taken, ["build", "lint"], "test waits for build; flash is for riscv64");
    // A riscv64 runner: the alpine image is known to be x86_64, so only flash.
    assert_eq!(claim_as("riscv64", serde_json::json!({})).await.as_deref(), Some("flash"));
    assert_eq!(claim_as("riscv64", serde_json::json!({})).await, None);
    let (_, queue) = call(app, "GET", "/admin/runners", None, None).await;
    let _ = queue;
}

/// OxygenFlow spec §11.5: a runner hands back a job it cannot run. The
/// job is queued again and not offered to that runner again, the decline
/// shows on the run's page and is kept for the runner's, and once three
/// runners have declined it the job ends as unplaceable with their reasons.
#[tokio::test(flavor = "multi_thread")]
async fn a_declined_job_goes_to_another_runner_or_ends_unplaceable() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "declines").await;
    let url = format!("{}/andy/declines", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "declines").await[0].clone();
    let claim_as = |token: &'static str, name: &'static str| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "runner": name, "runtimes": ["microvm"] }).to_string();
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(token), Some(body)).await;
            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap())
        }
    };
    let decline = |token: &'static str, id: i64, reason: &'static str| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "reason": reason }).to_string();
            call(&app, "POST", &format!("/api/v1/flow/jobs/{id}/decline"), Some(token), Some(body)).await.0
        }
    };
    let job = claim_as("runner-secret", "earth-1").await.expect("a job");
    // The other runners are known to the hub before they decline anything.
    assert_eq!(decline("runner-two", job, "not mine").await, StatusCode::CONFLICT, "only the runner it runs on may decline it");
    assert_eq!(decline("runner-secret", job, "the MicroVM tier needs /dev/kvm").await, StatusCode::NO_CONTENT);
    let j = db::flow::job(&h.app.db, job).await.unwrap().unwrap();
    assert_eq!((j.status.as_str(), j.runner.as_str()), ("queued", ""));
    assert_eq!(j.waiting, "declined by earth-1: the MicroVM tier needs /dev/kvm; waiting for another runner");
    // earth-1 is offered the other ready job, not this one, while it stays the same.
    let next = claim_as("runner-secret", "earth-1").await.expect("the other ready job");
    assert_ne!(next, job);
    let page = call(&h.app, "GET", &format!("/andy/declines/flow/{}", run.number), None, None).await.1;
    assert!(page.contains("Waiting: declined by earth-1: the MicroVM tier needs /dev/kvm"), "{page}");
    assert_eq!(claim_as("runner-two", "plex-1").await, Some(job), "another runner takes it");
    assert_eq!(decline("runner-two", job, "rootfs 36fc4e323804 is not on this runner").await, StatusCode::NO_CONTENT);
    assert_eq!(claim_as("runner-three", "narvi").await, Some(job));
    assert_eq!(decline("runner-three", job, "rootfs 36fc4e323804 is for x86_64; this runner is riscv64").await, StatusCode::NO_CONTENT);
    let j = db::flow::job(&h.app.db, job).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert_eq!(
        j.error,
        "unplaceable: declined by 3 runners: earth-1: the MicroVM tier needs /dev/kvm; plex-1: rootfs 36fc4e323804 is not on this runner; narvi: rootfs 36fc4e323804 is for x86_64; this runner is riscv64"
    );
    let r = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex-1").unwrap();
    let declined = db::flow::declines_of_runner(&h.app.db, r.id, 10).await.unwrap();
    assert_eq!(declined.len(), 1);
    assert_eq!((declined[0].job_id, declined[0].reason.as_str()), (job, "rootfs 36fc4e323804 is not on this runner"));
    assert_eq!(r.tier_list(), vec!["microvm"]);
}

/// oxygenhub#113: a job's artifact named `coverage` (LCOV) is read when
/// it is uploaded. The job's log says what it found; the Coverage tab,
/// the file view, the badge and the run page show it.
#[tokio::test(flavor = "multi_thread")]
async fn a_coverage_artifact_becomes_a_coverage_report() {
    let h = hub().await;
    create(&h.app, "andy", "cov").await;
    let url = format!("{}/andy/cov", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"coverage\" {{\n    sandbox \"alpine\"\n    exec \"make\" \"coverage\"\n    output {{\n        artifact \"coverage\" {{\n            path \"/src/lcov.info\"\n        }}\n    }}\n}}\n"
        ),
    );
    write(work.path(), "src/lib.rs", "pub fn add(a: u32, b: u32) -> u32 {\n    a + b\n}\n\npub fn unused() {\n    todo!()\n}\n");
    c.snapshot("test").unwrap();
    c.commit("Code and coverage", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "cov").await.remove(0);
    assert!(!call(&h.app, "GET", "/andy/cov", None, None).await.1.contains(">Reports</a>"), "no tab before a report");
    let id = claim(&h.app).await.unwrap()["job_id"].as_i64().unwrap();

    let (st, text) = upload(&h.app, id, "coverage", b"this is not a report").await;
    assert_eq!(st, StatusCode::OK, "a bad report never fails the upload: {text}");
    let lcov = "TN:\nSF:/src/src/lib.rs\nFN:1,add\nFN:5,unused\nFNDA:3,add\nFNDA:0,unused\nDA:1,3\nDA:2,3\nDA:5,0\nDA:6,0\nend_of_record\n\
SF:/src/src/gone.rs\nDA:1,1\nend_of_record\nSF:/usr/lib/rustlib/x.rs\nDA:1,1\nend_of_record\n";
    let (st, text) = upload(&h.app, id, "coverage", lcov.as_bytes()).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let log: Vec<String> = db::flow::log(&h.app.db, id, 0, 50).await.unwrap().into_iter().map(|l| l.text).collect();
    assert!(log.iter().any(|l| l.contains("coverage: this is not an LCOV report")), "{log:?}");
    assert!(
        log.iter().any(|l| l == "coverage: 50.0% of lines (2 of 4) in 1 files (2 named files are not in this state and were left out)"),
        "{log:?}"
    );

    let (st, page) = call(&h.app, "GET", "/andy/cov/coverage", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(
        page.contains(">Reports</a>")
            && page.contains("50.0%")
            && page.contains("<strong>2</strong> of <strong>4</strong> relevant lines covered"),
        "{page}"
    );
    assert!(page.contains("1 of 2") && page.contains("/andy/cov/coverage/") && page.contains("/file/src/lib.rs"), "{page}");
    let report = crate_report(&h.app, run.id).await;
    let (st, file) = call(&h.app, "GET", &format!("/andy/cov/coverage/{report}/file/src/lib.rs"), None, None).await;
    assert_eq!(st, StatusCode::OK, "{file}");
    assert_eq!(file.matches("class=\"cov-hit\"").count(), 2, "{file}");
    assert_eq!(file.matches("class=\"cov-miss\"").count(), 2, "{file}");
    assert!(file.contains("3×"), "{file}");
    assert_eq!(call(&h.app, "GET", &format!("/andy/cov/coverage/{report}/file/src/gone.rs"), None, None).await.0, StatusCode::NOT_FOUND);
    let (_, badge) = call(&h.app, "GET", "/andy/cov/coverage/badge.svg", None, None).await;
    assert!(badge.contains("coverage: 50.0%"), "{badge}");
    let (_, run_page) = call(&h.app, "GET", &format!("/andy/cov/flow/{}", run.number), None, None).await;
    assert!(run_page.contains("coverage 50.0%"), "{run_page}");
    // The API (#113).
    let (_, list) = call(&h.app, "GET", "/api/v1/repos/andy/cov/coverage", None, None).await;
    let list: serde_json::Value = serde_json::from_str(&list).unwrap();
    assert_eq!((list[0]["percent"].as_f64(), list[0]["format"].as_str()), (Some(50.0), Some("lcov")), "{list}");
    let (_, one) = call(&h.app, "GET", &format!("/api/v1/repos/andy/cov/coverage/{report}"), None, None).await;
    let one: serde_json::Value = serde_json::from_str(&one).unwrap();
    assert_eq!((one["files"][0]["path"].as_str(), one["files"][0]["lines_hit"].as_i64()), (Some("src/lib.rs"), Some(2)), "{one}");

    // The `coverage` check on the run's state: it reports, and with a
    // minimum the repository sets, it fails below it.
    let row = db::repos::get(&h.app.db, "andy", "cov").await.unwrap().unwrap();
    let check = |checks: Vec<db::hooks::Check>| checks.into_iter().find(|c| c.name == "coverage").expect("a coverage check");
    let c = check(db::hooks::checks_for_state(&h.app.db, row.id, &run.state).await.unwrap());
    assert_eq!((c.status.as_str(), c.summary.as_str()), ("passed", "50.0%"));
    assert!(c.url.ends_with(&format!("/andy/cov/coverage?report={report}")), "{}", c.url);
    oxygenhub::coverage::set_rules(&h.app.db, row.id, oxygenhub::coverage::Rules { min: Some(60.0), max_drop: None }).await.unwrap();
    assert_eq!(upload(&h.app, id, "coverage", lcov.as_bytes()).await.0, StatusCode::OK);
    let c = check(db::hooks::checks_for_state(&h.app.db, row.id, &run.state).await.unwrap());
    assert_eq!((c.status.as_str(), c.summary.as_str()), ("failed", "50.0%: below the minimum of 60%"));
}

async fn crate_report(app: &Shared, run_id: i64) -> i64 {
    oxygenhub::coverage::for_run(&app.db, run_id).await.unwrap()[0].id
}

/// The Tests tab (#114): three runs' JUnit reports on a bookmark; a test
/// breaks in the second and is fixed in the third.
#[tokio::test(flavor = "multi_thread")]
async fn junit_reports_make_a_tests_tab() {
    let h = hub().await;
    create(&h.app, "andy", "t").await;
    let url = format!("{}/andy/t", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    exec \"make\" \"test\"\n    output {{\n        artifact \"junit\" {{\n            path \"/src/junit.xml\"\n        }}\n    }}\n}}\n"
        ),
    );
    assert!(!call(&h.app, "GET", "/andy/t", None, None).await.1.contains(">Reports</a>"));
    let case = |name: &str, ok: bool, secs: &str| {
        if ok {
            format!(r#"<testcase classname="t.suite" name="{name}" time="{secs}"/>"#)
        } else {
            format!(
                r#"<testcase classname="t.suite" name="{name}" time="{secs}"><failure message="{name} broke">AssertionError</failure></testcase>"#
            )
        }
    };
    let runs = [
        vec![case("a", true, "0.1"), case("b", true, "2.5"), case("c", true, "0.2")],
        vec![case("a", true, "0.1"), case("b", false, "2.6"), case("c", true, "0.2"), case("d", true, "0.3")],
        vec![case("a", true, "0.1"), case("b", true, "2.4"), case("c", true, "0.2"), case("d", true, "0.3")],
    ];
    for (i, cases) in runs.iter().enumerate() {
        write(work.path(), "n.txt", &i.to_string());
        c.snapshot("test").unwrap();
        c.commit(&format!("Run {i}"), &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        // The run is made after the push answers.
        let mut a = None;
        for _ in 0..100 {
            a = claim(&h.app).await;
            if a.is_some() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        let a = a.expect("a job to claim");
        let id = a["job_id"].as_i64().unwrap();
        let xml = format!("<testsuites><testsuite name=\"t\">{}</testsuite></testsuites>", cases.concat());
        assert_eq!(upload(&h.app, id, "junit", xml.as_bytes()).await.0, StatusCode::OK);
        let failed = cases.iter().any(|c| c.contains("<failure"));
        let mut r = receipt("test", a["pipeline_hash"].as_str().unwrap(), failed as u32);
        if !failed {
            r["outputs"] = serde_json::json!([oxy_core::repo::Repo::blob_ref(xml.as_bytes()).to_hex()]);
        }
        let code = failed as i32;
        assert_eq!(
            finish(&h.app, id, serde_json::json!({ "exit_code": code, "duration_ms": 700, "receipt": r })).await,
            StatusCode::NO_CONTENT
        );
    }
    let (st, page) = call(&h.app, "GET", "/andy/t/tests", None, None).await;
    assert_eq!(st, StatusCode::OK, "{page}");
    if let Some(out) = std::env::var_os("OXYHUB_DUMP_TESTS_PAGE") {
        std::fs::write(out, &page).unwrap();
    }
    assert!(page.contains(">Reports</a>"), "the tab");
    assert!(page.contains("100%") && page.contains("of 4 tests passed"), "the latest run passes");
    assert!(page.contains("1 fixed") && page.contains("t.suite::b"), "b fixed against the run before");
    assert!(page.contains("1 ✕") && page.contains("2 ✓"), "b is flaky: {page}");
    assert!(page.contains("<svg class=\"cov-chart\""), "charts");
    // The slowest first.
    let slow = page.find("Slowest tests").unwrap();
    assert!(page[slow..].find("t.suite::b").unwrap() < page[slow..].find("t.suite::d").unwrap());
    // The second run, picked: b newly failing, its failure listed.
    let row = db::repos::get(&h.app.db, "andy", "t").await.unwrap().unwrap();
    let second = oxygenhub::junit::on_bookmark(&h.app.db, row.id, "main", 10).await.unwrap()[1].id;
    let (_, page) = call(&h.app, "GET", &format!("/andy/t/tests?report={second}"), None, None).await;
    assert!(page.contains("1 failing") && page.contains("Newly failing") && page.contains("b broke"), "{page}");
    assert!(page.contains("1 new tests"), "d is new");

    // The API (#114): the bookmark's reports newest first, and one with
    // its failed cases.
    let (st, list) = call(&h.app, "GET", "/api/v1/repos/andy/t/tests", None, None).await;
    assert_eq!(st, StatusCode::OK, "{list}");
    let list: serde_json::Value = serde_json::from_str(&list).unwrap();
    assert_eq!(list.as_array().unwrap().len(), 3);
    assert_eq!((list[0]["tests"].as_i64(), list[1]["failed"].as_i64()), (Some(4), Some(1)));
    let (_, one) = call(&h.app, "GET", &format!("/api/v1/repos/andy/t/tests/{second}?status=failed"), None, None).await;
    let one: serde_json::Value = serde_json::from_str(&one).unwrap();
    assert_eq!(one["cases"].as_array().unwrap().len(), 1);
    assert_eq!((one["cases"][0]["name"].as_str(), one["cases"][0]["message"].as_str()), (Some("b"), Some("b broke")));
    let (_, all) = call(&h.app, "GET", &format!("/api/v1/repos/andy/t/tests/{second}"), None, None).await;
    assert_eq!(serde_json::from_str::<serde_json::Value>(&all).unwrap()["cases"].as_array().unwrap().len(), 4);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/t/tests/99999", None, None).await.0, StatusCode::NOT_FOUND);

    // The `tests` check names what broke against main.
    let reports = oxygenhub::junit::on_bookmark(&h.app.db, row.id, "main", 10).await.unwrap();
    let check = |state: String| {
        let app = h.app.clone();
        async move { db::hooks::checks_for_state(&app.db, row.id, &state).await.unwrap().into_iter().find(|c| c.name == "tests").unwrap() }
    };
    let c = check(reports[1].state.clone()).await;
    assert_eq!((c.status.as_str(), c.summary.as_str()), ("failed", "1 failing; 1 new against main: t.suite::b"));
    let c = check(reports[0].state.clone()).await;
    assert_eq!((c.status.as_str(), c.summary.as_str()), ("passed", "4 passed"));

    // One test's history: b passed, failed, passed.
    let (_, tab) = call(&h.app, "GET", "/andy/t/tests", None, None).await;
    assert!(
        tab.contains("/andy/t/tests/case?classname=t.suite&amp;name=b") || tab.contains("/andy/t/tests/case?classname=t.suite&name=b"),
        "linked"
    );
    let (st, hist) = call(&h.app, "GET", "/andy/t/tests/case?classname=t.suite&name=b", None, None).await;
    assert_eq!(st, StatusCode::OK, "{hist}");
    assert!(hist.contains("2 passed") && hist.contains("1 failed") && hist.contains("b broke"), "{hist}");
    assert_eq!(call(&h.app, "GET", "/andy/t/tests/case?classname=t.suite&name=nope", None, None).await.0, StatusCode::NOT_FOUND);
}

/// oxygenhub#28: a queued job no runner can take says why, in the API, on
/// the run page (its panel open) and in its check.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_no_runner_can_take_says_why() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

    // One runner, online, holding another image only.
    let r = db::runners::configured(&h.app.db, &blake3::hash(b"elsewhere").to_hex(), "earth-9").await.unwrap();
    db::runners::seen(&h.app.db, r.id, None, None).await.unwrap();
    let other = "ab".repeat(32);
    db::runners::set_offer(&h.app.db, r.id, &["microvm".to_string()], Some(&[other]), false, "p").await.unwrap();
    oxygenhub::flow::explain_waits(&h.app).await.unwrap();

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    let build = v["jobs"].as_array().unwrap().iter().find(|j| j["name"] == "build").unwrap().clone();
    let why = build["waiting"].as_str().unwrap_or_default().to_string();
    assert!(why.contains("earth-9 does not hold image") && why.contains(&ROOTFS[..12]), "{body}");
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/1", None, None).await;
    assert!(page.contains("data-waiting") && page.contains("does not hold image"), "the run page says it");
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    let check = checks.iter().find(|c| c.name == "flow/build").unwrap();
    assert!(check.summary.contains("does not hold image"), "{check:?}");
}

async fn put_bytes(app: &Shared, path: &str, bearer: &str, bytes: Vec<u8>) -> (StatusCode, String) {
    let req = Request::put(path)
        .header(header::AUTHORIZATION, format!("Bearer {bearer}"))
        .header(header::CONTENT_TYPE, "application/octet-stream")
        .body(Body::from(bytes))
        .unwrap();
    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();
    let st = res.status();
    (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
}

/// oxygenflow#39: an image is sent as the chunks the hub lacks and then its
/// manifest, checked against its hash; runners fetch it; and a job whose
/// image only the hub holds is offered to a runner that fetches, not to one
/// that does not.
#[tokio::test(flavor = "multi_thread")]
async fn images_are_sent_by_chunk_and_offered_to_runners_that_fetch() {
    let h = hub().await;
    // The site token acts as the first site administrator.
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    // An image: bytes, a run of zeros, more bytes.
    let mut image: Vec<u8> = (0..400_000u32).map(|i| (i % 251) as u8).collect();
    image.extend(std::iter::repeat_n(0u8, 2 << 20));
    image.extend((0..100_000u32).map(|i| (i % 239) as u8));
    let mut chunks = std::collections::HashMap::new();
    let m = jet::images::scan(&image[..], Some("x86_64".into()), |c, b| {
        chunks.insert(c.hash().to_string(), b.to_vec());
        Ok(())
    })
    .unwrap();
    let hash = m.hash.clone();

    // Which chunks the hub lacks: every one but zeros; a manifest before them is refused.
    let ask = serde_json::json!({ "chunks": m.needed() }).to_string();
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/images/missing", Some("site"), Some(ask.clone())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let missing: Vec<String> = serde_json::from_str::<serde_json::Value>(&body).unwrap()["missing"]
        .as_array()
        .unwrap()
        .iter()
        .map(|v| v.as_str().unwrap().to_string())
        .collect();
    assert_eq!(missing.len(), m.needed().len());
    let manifest = serde_json::to_string(&m).unwrap();
    let (st, body) = call(&h.app, "PUT", &format!("/api/v1/flow/images/{hash}"), Some("site"), Some(manifest.clone())).await;
    assert!(st == StatusCode::CONFLICT && body.contains("was not sent"), "{st} {body}");
    // A chunk is checked against its name; a runner token may not send.
    let first = &missing[0];
    assert_eq!(put_bytes(&h.app, &format!("/api/v1/flow/images/chunks/{first}"), "site", vec![1, 2, 3]).await.0, StatusCode::BAD_REQUEST);
    assert_ne!(
        put_bytes(&h.app, &format!("/api/v1/flow/images/chunks/{first}"), "runner-secret", chunks[first].clone()).await.0,
        StatusCode::NO_CONTENT
    );
    for c in &missing {
        let (st, body) = put_bytes(&h.app, &format!("/api/v1/flow/images/chunks/{c}"), "site", chunks[c].clone()).await;
        assert_eq!(st, StatusCode::NO_CONTENT, "{body}");
    }
    let (_, body) = call(&h.app, "POST", "/api/v1/flow/images/missing", Some("site"), Some(ask)).await;
    assert!(body.contains("\"missing\":[]"), "{body}");
    // A large image's question fits (#147): 60,000 names, ~4 MB of JSON.
    let many: Vec<String> = (0..60_000u32).map(|i| blake3::hash(&i.to_le_bytes()).to_hex().to_string()).collect();
    let (st, body) =
        call(&h.app, "POST", "/api/v1/flow/images/missing", Some("site"), Some(serde_json::json!({ "chunks": many }).to_string())).await;
    assert_eq!(st, StatusCode::OK, "{}", &body[..body.len().min(200)]);
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["missing"].as_array().unwrap().len(), 60_000);
    // A manifest naming another hash is refused; the right one is kept.
    let mut wrong = m.clone();
    wrong.hash = "ab".repeat(32);
    let (st, _) =
        call(&h.app, "PUT", &format!("/api/v1/flow/images/{}", wrong.hash), Some("site"), Some(serde_json::to_string(&wrong).unwrap()))
            .await;
    assert_eq!(st, StatusCode::CONFLICT, "chunks that make another image");
    let (st, body) = call(&h.app, "PUT", &format!("/api/v1/flow/images/{hash}"), Some("site"), Some(manifest)).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");

    // A runner fetches the manifest and the chunks, and the image comes back whole.
    let (st, body) = call(&h.app, "GET", &format!("/api/v1/flow/images/{hash}"), Some("runner-secret"), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let got: jet::images::Manifest = serde_json::from_str(&body).unwrap();
    assert_eq!(got, m);
    let dir = tempfile::tempdir().unwrap();
    let out = dir.path().join("img.ext4");
    let mut fetched = Vec::new();
    for c in got.needed() {
        let req = Request::get(format!("/api/v1/flow/images/chunks/{c}"))
            .header(header::AUTHORIZATION, "Bearer runner-secret")
            .body(Body::empty())
            .unwrap();
        let res = oxygenhub::router(h.app.clone()).oneshot(req).await.unwrap();
        assert_eq!(res.status(), StatusCode::OK);
        let bytes = res.into_body().collect().await.unwrap().to_bytes().to_vec();
        fetched.push((c.to_string(), bytes));
    }
    jet::images::assemble(&got, &out, |c| Ok(fetched.iter().find(|(h, _)| h == c.hash()).unwrap().1.clone())).unwrap();
    assert_eq!(std::fs::read(&out).unwrap(), image);
    assert_eq!(
        call(&h.app, "GET", &format!("/api/v1/flow/images/{}", "cd".repeat(32)), Some("runner-secret"), None).await.0,
        StatusCode::NOT_FOUND
    );
    assert_ne!(call(&h.app, "GET", &format!("/api/v1/flow/images/{hash}"), None, None).await.0, StatusCode::OK, "not anonymous");

    // A job on that image: a runner holding no images and not fetching
    // is offered nothing; one that fetches is offered it.
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!("schema-version 1\nsandbox \"img\" {{\n    rootfs \"cas://blob/{hash}\"\n}}\njob \"build\" {{\n    sandbox \"img\"\n    exec \"make\"\n}}\n"),
    );
    c.commit("Build on the image", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;
    // #137: the run's images are held (by the hub), so no images check.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let run1 = db::flow::run(&h.app.db, row.id, 1).await.unwrap().unwrap();
    let checks = checks_when(&h.app, row.id, &run1.state, 1).await;
    assert!(checks.iter().all(|c| c.name != "pipeline/images"), "{checks:?}");
    let ask = |fetches: bool| format!(r#"{{"runner":"test-1","runtimes":["microvm"],"images":[],"fetches_images":{fetches}}}"#);
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(ask(false))).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "it lacks the image and does not fetch");
    oxygenhub::flow::explain_waits(&h.app).await.unwrap();
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;
    assert!(body.contains("does not hold image"), "{body}");
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(ask(true))).await;
    assert_eq!(st, StatusCode::OK, "it fetches, and the hub holds the image: {body}");
    // oxygenhub#135: once claimed, the old waiting reason is gone.
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;
    assert!(!body.contains("waiting"), "a running job keeps no waiting reason: {body}");
}

/// oxygenhub#136: coverage measured on a workstation is uploaded for a
/// bookmark's state, kept as uploaded by its account, shown so, and sets
/// no check.
#[tokio::test(flavor = "multi_thread")]
async fn coverage_uploaded_from_a_workstation_says_so() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "vk").await;
    let url = format!("{}/andy/vk", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), "src/game.c", "int a;\nint b;\nint c;\nint d;\n");
    c.commit("Game", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let lcov = "SF:src/game.c\nDA:1,3\nDA:2,1\nDA:3,0\nDA:4,5\nend_of_record\nSF:reference/rom.asm\nDA:1,1\nend_of_record\n";
    let body = serde_json::json!({ "bookmark": "main", "content": lcov }).to_string();
    let (st, out) = call(&h.app, "POST", "/api/v1/repos/andy/vk/coverage", Some("site"), Some(body)).await;
    assert_eq!(st, StatusCode::CREATED, "{out}");
    let v: serde_json::Value = serde_json::from_str(&out).unwrap();
    assert_eq!((v["uploaded_by"].as_str(), v["lines_hit"].as_i64(), v["lines_found"].as_i64()), (Some("andy"), Some(3), Some(4)));
    assert_eq!(v["dropped"], 1, "a file outside the state is left out");
    assert_eq!(v["bookmark"], "main");

    let (_, page) = call(&h.app, "GET", "/andy/vk/coverage", None, None).await;
    assert!(page.contains("uploaded by andy") && page.contains("75.0%"), "the Coverage page says where it came from");
    let (_, page) = call(&h.app, "GET", "/andy/vk/reports", None, None).await;
    assert!(page.contains("uploaded by andy"), "and the Reports card");
    let row = db::repos::get(&h.app.db, "andy", "vk").await.unwrap().unwrap();
    let state = v["state"].as_str().unwrap();
    let checks = db::hooks::checks_for_state(&h.app.db, row.id, state).await.unwrap();
    assert!(checks.iter().all(|c| c.name != "coverage"), "an upload sets no check: {checks:?}");

    // Readers cannot upload; bad reports and bookmarks are refused.
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/vk/coverage", None, Some(r#"{"content":"x"}"#.into())).await;
    assert_ne!(st, StatusCode::CREATED);
    let (st, _) = call(
        &h.app,
        "POST",
        "/api/v1/repos/andy/vk/coverage",
        Some("site"),
        Some(r#"{"bookmark":"nope","content":"SF:a\nend_of_record\n"}"#.into()),
    )
    .await;
    assert_eq!(st, StatusCode::BAD_REQUEST);
}

/// oxygenflow#40: what a runner measured at its claim is kept and shown.
#[tokio::test(flavor = "multi_thread")]
async fn a_runners_facts_are_kept_from_its_claim() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    let body = r#"{"runner":"test-1","runtimes":["microvm"],"facts":{"free_disk_mib":42000,"load_x100":112,"speed":100,"speed_class":3,"kernel":"7.0.0"}}"#;
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body.into())).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "nothing to run, but the facts are kept");
    let (st, out) = call(&h.app, "GET", "/api/v1/runners", Some("site"), None).await;
    assert_eq!(st, StatusCode::OK, "{out}");
    let v: serde_json::Value = serde_json::from_str(&out).unwrap();
    let r = &v[0];
    assert_eq!(
        (r["speed"].as_i64(), r["speed_class"].as_i64(), r["free_disk_mib"].as_i64(), r["load_x100"].as_i64()),
        (Some(100), Some(3), Some(42000), Some(112))
    );
    let row = db::runners::list(&h.app.db).await.unwrap().remove(0);
    assert_eq!(row.facts_line(), "speed 100 (class 3) · load 1.12 · 41 GiB free");
}

/// oxygenflow#41: a job's peaks set what its next run needs; a runner
/// without that much free disk is not offered it, and the reason says so.
#[tokio::test(flavor = "multi_thread")]
async fn a_jobs_next_run_needs_the_disk_its_last_took() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!("schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\n"),
    );
    c.commit("Build", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;
    let claim_with = |free: u64| format!(r#"{{"runner":"test-1","runtimes":["microvm"],"facts":{{"free_disk_mib":{free}}}}}"#);

    // The first run has nothing to learn from: any runner takes it.
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(claim_with(500))).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let id = serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap();
    let done = serde_json::json!({ "exit_code": 0, "duration_ms": 10, "peak_disk_mib": 8000, "peak_memory_mib": 900 });
    assert_eq!(finish(&h.app, id, done).await, StatusCode::NO_CONTENT);
    let job = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!((job.peak_disk_mib, job.peak_memory_mib), (Some(8000), Some(900)));

    // The next run needs 8000 MiB and a quarter more.
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::CREATED);
    let repo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let run2 = db::flow::run(&h.app.db, repo.id, 2).await.unwrap().unwrap();
    let next = db::flow::jobs(&h.app.db, run2.id).await.unwrap().remove(0);
    assert_eq!(next.need_disk_mib, Some(8000 + 2000 + 64));
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(claim_with(9000))).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "9000 MiB free is not enough");
    oxygenhub::flow::explain_waits(&h.app).await.unwrap();
    let waiting = db::flow::job(&h.app.db, next.id).await.unwrap().unwrap().waiting;
    assert!(waiting.contains("8.8 GiB free; the job needs ~9.8 GiB"), "{waiting}");
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(claim_with(20_000))).await;
    assert_eq!(st, StatusCode::OK, "a roomier runner takes it: {body}");
}

/// oxygenflow#42: a job another free runner would do better (it holds the
/// image) is held from a runner that would fetch it, the better runner
/// takes it, and the job says why it ran there.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_waits_briefly_for_a_runner_that_holds_its_image() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    // A second runner, with a token of its own.
    db::runners::configured(&h.app.db, &blake3::hash(b"other-secret").to_hex(), "earth-9").await.unwrap();
    let ask = |images: &str| {
        format!(
            r#"{{"runner":"r","runtimes":["microvm"],"images":[{images}],"fetches_images":true,"facts":{{"speed":100,"speed_class":3,"free_disk_mib":50000}}}}"#
        )
    };
    let holder = ask(&format!("\"{ROOTFS}\""));
    // earth-9 holds the image and is polling; nothing is queued yet.
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(holder.clone())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);
    // The hub holds the image too, so test-1 could fetch it.
    sqlx::query("INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES ($1, 1, 'x86_64', 1, 'andy', 0)")
        .bind(ROOTFS)
        .execute(&h.app.db)
        .await
        .unwrap();

    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!("schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\n"),
    );
    c.commit("Build", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(ask(""))).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "held for the runner that holds the image");
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(holder)).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let id = serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap();
    let placed = db::flow::job(&h.app.db, id).await.unwrap().unwrap().placed;
    assert!(placed.starts_with("holds the image, idle, speed 100 (class 3)"), "{placed}");
    assert!(placed.contains("held for a better runner"), "{placed}");
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;
    assert!(body.contains("\"placed\":\"holds the image"), "{body}");
}

/// oxygenflow#51: a job waits briefly for the runner holding its build
/// cache, even when the other one holds its image too.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_waits_briefly_for_the_runner_holding_its_build_cache() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    db::runners::configured(&h.app.db, &blake3::hash(b"other-secret").to_hex(), "earth-9").await.unwrap();
    let slot = flow_core::api::cache_slot("andy/demo", "build");
    let ask = |caches: &str| {
        format!(
            r#"{{"runner":"r","runtimes":["microvm"],"images":["{ROOTFS}"],"fetches_images":true,"caches":[{caches}],"facts":{{"speed":100,"speed_class":3,"free_disk_mib":50000}}}}"#
        )
    };
    let holder = ask(&format!("\"{slot}\""));
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(holder.clone())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);

    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!("schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\n"),
    );
    c.commit("Build", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(ask(""))).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "held for the runner holding the build cache");
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(holder)).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let id = serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap();
    let placed = db::flow::job(&h.app.db, id).await.unwrap().unwrap().placed;
    assert!(placed.starts_with("holds its build cache, holds the image"), "{placed}");
}

/// oxygenflow#54: when both runners hold a job's build cache, the job waits
/// briefly for the one whose copy is fresher.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_waits_briefly_for_the_fresher_build_cache() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    db::runners::configured(&h.app.db, &blake3::hash(b"other-secret").to_hex(), "earth-9").await.unwrap();
    let slot = flow_core::api::cache_slot("andy/demo", "build");
    let now = db::now_ms();
    let ask = |kept: i64| {
        format!(
            r#"{{"runner":"r","runtimes":["microvm"],"images":["{ROOTFS}"],"fetches_images":true,"caches":["{slot}"],"cache_times":{{"{slot}":{kept}}},"facts":{{"speed":100,"speed_class":3,"free_disk_mib":50000}}}}"#
        )
    };
    // earth-9 kept its copy a minute ago; the other runner's is a day old.
    let fresh = ask(now - 60_000);
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(fresh.clone())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);

    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!("schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\n"),
    );
    c.commit("Build", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(ask(now - 86_400_000))).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "held for the fresher copy");
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(fresh)).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let id = serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap();
    let placed = db::flow::job(&h.app.db, id).await.unwrap().unwrap().placed;
    assert!(placed.starts_with("holds its build cache (kept 1 min ago), holds the image"), "{placed}");
}

/// oxygenhub#127 §3.3: report files expire beyond the newest 50 runs of a
/// report on a bookmark other than the default; the default keeps all,
/// and an expired report keeps its metrics.
#[tokio::test(flavor = "multi_thread")]
async fn report_files_expire_under_retention() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let repo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let add = |bookmark: &'static str, at: i64| {
        let db = h.app.db.clone();
        async move {
            let id: i64 = sqlx::query_scalar(
                "INSERT INTO reports (repo_id, name, kind, title, summary, context, job, bookmark, state, files, bytes, created_ms) \
                 VALUES ($1, 'perf', 'performance', 'Perf', '', '{}', 'perf', $2, 's', 1, 1, $3) RETURNING id",
            )
            .bind(repo.id)
            .bind(bookmark)
            .bind(at)
            .fetch_one(&db)
            .await
            .unwrap();
            sqlx::query("INSERT INTO report_files (report_id, path, blob, size) VALUES ($1, 'report.toml', $2, 1)")
                .bind(id)
                .bind(format!("{:064x}", id))
                .execute(&db)
                .await
                .unwrap();
            sqlx::query("INSERT INTO report_metrics (report_id, key, value, unit, better) VALUES ($1, 'ms', 1, '', '')")
                .bind(id)
                .execute(&db)
                .await
                .unwrap();
            id
        }
    };
    let main_old = add("main", 1).await;
    add("main", 2).await;
    let mut feature = Vec::new();
    for at in 0..52 {
        feature.push(add("feature", 10 + at).await);
    }
    assert_eq!(oxygenhub::reports::expire(&h.app.db).await.unwrap(), 2, "the two oldest feature runs");
    let files = |id: i64| {
        let db = h.app.db.clone();
        async move {
            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM report_files WHERE report_id = $1").bind(id).fetch_one(&db).await.unwrap()
        }
    };
    assert_eq!((files(feature[0]).await, files(feature[1]).await, files(feature[2]).await), (0, 0, 1));
    assert_eq!(files(main_old).await, 1, "the default bookmark keeps every run");
    assert_eq!(oxygenhub::reports::metrics(&h.app.db, feature[0]).await.unwrap().len(), 1, "metrics are kept");
    assert_eq!(oxygenhub::reports::expire(&h.app.db).await.unwrap(), 0, "once");
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/reports/perf?id={}", feature[0]), None, None).await;
    assert!(page.contains("files have expired"), "the page says so");
}

/// #137: a run pinning an image neither the hub nor any runner holds sets a
/// failing `pipeline/images` check saying how to send it; once it is held,
/// a run of the same state turns the check to passed.
#[tokio::test(flavor = "multi_thread")]
async fn a_pipeline_pinning_an_image_nobody_holds_fails_a_check() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    // The one runner reports holding no images.
    let none = r#"{"runner":"test-1","runtimes":["microvm"],"images":[]}"#;
    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(none.into())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let checks = checks_when(&h.app, run.repo_id, &run.state, 4).await;
    let check = checks.iter().find(|c| c.name == "pipeline/images").expect("an images check");
    assert!(check.status == "failed" && check.summary.contains(&ROOTFS[..12]) && check.summary.contains("flow rootfs push"), "{check:?}");
    // Once a runner holds it, a run of the state passes the check.
    let holds = format!(r#"{{"runner":"test-1","runtimes":["microvm"],"images":["{ROOTFS}"]}}"#);
    call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(holds)).await;
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::CREATED);
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    let check = checks.iter().find(|c| c.name == "pipeline/images").unwrap();
    assert_eq!(check.status, "passed", "{check:?}");
}

/// The swarm's tracker (#144): a fetching runner is given the peers in its
/// site that hold the image or are fetching it, and a ticket signed with
/// the hub key; a runner not in a swarm is told how to join one.
#[tokio::test(flavor = "multi_thread")]
async fn the_tracker_names_peers_in_the_same_site_and_signs_a_ticket() {
    let h = hub().await;
    let app = &h.app;
    let image = "ab".repeat(32);
    let claim_as = |token: &'static str, name: &'static str, body: serde_json::Value| async move {
        let mut b = serde_json::json!({ "runner": name, "runtimes": ["microvm"] });
        b.as_object_mut().unwrap().extend(body.as_object().unwrap().clone());
        let (st, out) = call(app, "POST", "/api/v1/flow/claim", Some(token), Some(b.to_string())).await;
        assert_eq!(st, StatusCode::NO_CONTENT, "{out}");
    };
    let announce = |token: &'static str, state: &'static str| {
        let body = serde_json::json!({ "image": image, "state": state }).to_string();
        async move { call(app, "POST", "/api/v1/flow/swarm/announce", Some(token), Some(body)).await }
    };
    // A holds the image; B and C hold nothing; all three are in site "lab".
    claim_as(
        "runner-secret",
        "a",
        serde_json::json!({ "images": [image], "swarm": { "addr": "10.0.0.1:7401", "site": "lab", "cert": "11".repeat(32) } }),
    )
    .await;
    claim_as(
        "runner-two",
        "b",
        serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.2:7401", "site": "lab", "cert": "22".repeat(32) } }),
    )
    .await;
    claim_as(
        "runner-three",
        "c",
        serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "lab", "cert": "33".repeat(32) } }),
    )
    .await;

    let (st, body) = announce("runner-two", "fetching").await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();
    let names: Vec<(&str, bool)> = a.peers.iter().map(|p| (p.addr.as_str(), p.complete)).collect();
    assert_eq!(names, vec![("10.0.0.1:7401", true)], "only A holds it so far");
    assert_eq!(a.peers[0].cert, "11".repeat(32), "with the fingerprint peers check it by");
    let t = jet::swarm::Ticket::open(&a.ticket, &app.hub_key.identity, oxygenhub::db::now_ms() as u64).unwrap();
    assert_eq!((t.name.as_str(), t.site.as_str()), ("b", "lab"));

    // C, fetching too, is told of A (whole) and B (fetching).
    let (_, body) = announce("runner-three", "fetching").await;
    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();
    let mut peers: Vec<(&str, bool)> = a.peers.iter().map(|p| (p.addr.as_str(), p.complete)).collect();
    peers.sort();
    assert_eq!(peers, vec![("10.0.0.1:7401", true), ("10.0.0.2:7401", false)]);

    // B is done: no longer offered as fetching. C moves site: no peers.
    announce("runner-two", "done").await;
    claim_as(
        "runner-three",
        "c",
        serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "elsewhere", "cert": "33".repeat(32) } }),
    )
    .await;
    let (_, body) = announce("runner-three", "fetching").await;
    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();
    assert!(a.peers.is_empty(), "{:?}", a.peers);

    // A runner without a certificate (an older flow) is not in the swarm.
    claim_as("runner-secret", "a", serde_json::json!({ "images": [image], "swarm": { "addr": "10.0.0.1:7401", "site": "elsewhere" } }))
        .await;
    let (st, _) = announce("runner-secret", "fetching").await;
    assert_eq!(st, StatusCode::CONFLICT);

    // A runner that left the swarm is told how to join; nonsense is refused.
    claim_as("runner-two", "b", serde_json::json!({ "images": [] })).await;
    let (st, body) = announce("runner-two", "fetching").await;
    assert_eq!(st, StatusCode::CONFLICT);
    assert!(body.contains("[runner.swarm]"), "{body}");
    let (st, _) =
        call(app, "POST", "/api/v1/flow/swarm/announce", Some("runner-secret"), Some(r#"{"image":"nope","state":"fetching"}"#.into()))
            .await;
    assert_eq!(st, StatusCode::BAD_REQUEST);
    let (st, _) = call(
        app,
        "POST",
        "/api/v1/flow/swarm/announce",
        None,
        Some(serde_json::json!({ "image": image, "state": "fetching" }).to_string()),
    )
    .await;
    assert_eq!(st, StatusCode::UNAUTHORIZED);
}

/// Admin › Runners › Images (C0): each image, where it is and what uses it.
#[tokio::test(flavor = "multi_thread")]
async fn the_images_page_says_where_each_image_is_and_what_uses_it() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    // A runner holds another image and shares it; the hub keeps a third.
    let held = "cd".repeat(32);
    let kept = "ef".repeat(32);
    let body = serde_json::json!({ "runner": "earth-1", "runtimes": ["wasm"], "images": [held], "swarm": { "addr": "10.0.0.1:7401", "site": "home", "cert": "11".repeat(32) } });
    let (st, out) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body.to_string())).await;
    assert!(st == StatusCode::NO_CONTENT || st == StatusCode::OK, "{st} {out}");
    sqlx::query("INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES ($1, 1048576, 'x86_64', 9, 'andy', 1)")
        .bind(&kept)
        .execute(&h.app.db)
        .await
        .unwrap();

    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let carol = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();
    db::accounts::set_admin(&h.app.db, carol.id, true).await.unwrap();
    let res = oxygenhub::router(h.app.clone())
        .oneshot(Request::get("/admin/runners/images").header(header::COOKIE, cookie).body(Body::empty()).unwrap())
        .await
        .unwrap();
    assert_eq!(res.status(), StatusCode::OK);
    let page = String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned();
    let row = |hash: &str| {
        let at = page.find(&format!("title=\"{hash}\"")).unwrap_or_else(|| panic!("{hash} is not on the page"));
        let end = page[at..].find("</tr>").unwrap();
        page[at..at + end].to_string()
    };
    let pinned = row(ROOTFS);
    assert!(pinned.contains("nowhere") && pinned.contains("andy/demo") && pinned.contains("3 jobs"), "{pinned}");
    let on_runner = row(&held);
    assert!(on_runner.contains("img-holds") && on_runner.contains("earth-1") && !on_runner.contains("nowhere"), "{on_runner}");
    let on_hub = row(&kept);
    assert!(on_hub.contains("img-hub") && on_hub.contains("1.0 MiB") && on_hub.contains("x86_64"), "{on_hub}");
    assert!(page.contains("earth-1 (home)"), "the swarm is named");
    // The swarm backdrop (docs/backgrounds.md §2): the hub and the runners.
    assert!(page.contains(r#"{"images":"#) && page.contains(r#""scene":"swarm""#) && page.contains(r#""earth-1""#), "the swarm's data");
    assert_eq!(call(&h.app, "GET", "/admin/runners/images", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");
}

/// Diffusion D1 (#148): the reverse pin index. A repository that pins
/// another as a subrepo is a consumer of it; the upstream's API and page
/// say who, at which state, and how far behind its default bookmark.
#[tokio::test(flavor = "multi_thread")]
async fn consumers_of_a_subrepo_are_indexed_with_how_far_behind_they_are() {
    let h = hub().await;
    // The site token acts as the first site administrator.
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "api").await;
    create(&h.app, "andy", "game").await;
    let api_work = tempfile::tempdir().unwrap();
    let api = client(&format!("{}/andy/api", h.base), api_work.path());
    write(api_work.path(), "api.h", "v1\n");
    api.snapshot("t").unwrap();
    api.commit("api v1", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));

    let game_work = tempfile::tempdir().unwrap();
    let game = client(&format!("{}/andy/game", h.base), game_work.path());
    game.submodule_add(&format!("{}/andy/api", h.base), "deps/api", Some("main")).unwrap();
    write(game_work.path(), "game.c", "int main;\n");
    game.snapshot("t").unwrap();
    game.commit("game on api v1", &[]).unwrap();
    assert!(game.push("origin", false).unwrap().iter().all(|r| r.ok));

    let consumers = || async {
        let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/api/consumers", Some("site"), None).await;
        assert_eq!(st, StatusCode::OK, "{body}");
        serde_json::from_str::<Vec<serde_json::Value>>(&body).unwrap()
    };
    // The index is written by the push's after-work, a moment after the
    // client hears the push succeeded.
    let settle = |want: &'static (dyn Fn(&[serde_json::Value]) -> bool + Sync)| {
        let consumers = &consumers;
        async move {
            for _ in 0..100 {
                let list = consumers().await;
                if want(&list) {
                    return list;
                }
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            }
            consumers().await
        }
    };
    let list = settle(&|l| !l.is_empty()).await;
    assert_eq!(list.len(), 1, "{list:?}");
    assert_eq!(
        (list[0]["name"].as_str(), list[0]["path"].as_str(), list[0]["bookmark"].as_str()),
        (Some("game"), Some("deps/api"), Some("main"))
    );
    assert_eq!(list[0]["behind"], 0);

    // The upstream moves on: the consumer is one change behind.
    write(api_work.path(), "api.c", "v2\n");
    api.snapshot("t").unwrap();
    api.commit("api v2", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(settle(&|l| l.first().is_some_and(|c| c["behind"] == 1)).await[0]["behind"], 1);
    let (_, page) = call(&h.app, "GET", "/andy/api", None, None).await;
    assert!(page.contains("Diffuses to") && page.contains("andy/game") && page.contains("1 behind"), "the page shows it");

    // A private consumer is not shown to someone who cannot read it.
    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'game'").execute(&h.app.db).await.unwrap();
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/api/consumers", None, None).await;
    assert_eq!(body.trim(), "[]");
    let (_, page) = call(&h.app, "GET", "/andy/api", None, None).await;
    assert!(!page.contains("andy/game"));
    sqlx::query("UPDATE repositories SET visibility = 'public' WHERE name = 'game'").execute(&h.app.db).await.unwrap();

    // A bump (DF3): one signed change on the target moving only the pin, on
    // `diffusion/deps/api`; again with the same state, nothing new.
    let game_row = db::repos::get(&h.app.db, "andy", "game").await.unwrap().unwrap();
    let api_row = db::repos::get(&h.app.db, "andy", "api").await.unwrap().unwrap();
    let api_heads = api.bookmark_heads().unwrap();
    let bumped = oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, api_heads.clone()).await.unwrap();
    let oxygenhub::diffusion::Bumped::Moved { bookmark, heads, .. } = bumped else { panic!("{bumped:?}") };
    assert_eq!(bookmark, "diffusion/deps/api");
    game.fetch("origin").unwrap();
    let bump_state = game.load_state(&heads).unwrap().materialize();
    let pin = match &bump_state.files["deps/api"].content {
        oxy_core::state::FileContent::Opaque { blobs } => oxy_core::subrepo::Pin::decode(&game.read_blob(&blobs[0]).unwrap()).unwrap(),
        other => panic!("{other:?}"),
    };
    assert!(
        matches!(&pin, oxy_core::subrepo::Pin::Oxy { heads, .. } if { let mut a = heads.clone(); a.sort(); let mut b = api_heads.clone(); b.sort(); a == b }),
        "{pin:?}"
    );
    assert!(bump_state.files.contains_key("game.c"), "the rest of the game is as it was");
    assert!(matches!(
        oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, api_heads.clone()).await.unwrap(),
        oxygenhub::diffusion::Bumped::Already { .. }
    ));
    // A state lacking changes the pin has (as when the consumer pinned an
    // upstream proposal not landed yet) is combined with the pin, never
    // dropping what it relies on.
    write(api_work.path(), "other.c", "v3\n");
    api.snapshot("t").unwrap();
    let v3 = api.commit("api v3, on its own", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    let only_v3 = api.load_state(&[v3]).unwrap().heads().to_vec();
    let combined = oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, only_v3).await.unwrap();
    let oxygenhub::diffusion::Bumped::Moved { heads: ch, .. } = combined else { panic!("{combined:?}") };
    game.fetch("origin").unwrap();
    let st = game.load_state(&ch).unwrap().materialize();
    let pin = match &st.files["deps/api"].content {
        oxy_core::state::FileContent::Opaque { blobs } => oxy_core::subrepo::Pin::decode(&game.read_blob(&blobs[0]).unwrap()).unwrap(),
        other => panic!("{other:?}"),
    };
    let oxy_core::subrepo::Pin::Oxy { heads: pinned, .. } = pin else { panic!() };
    let all = api.load_state(&pinned).unwrap();
    // The game's main still pins v1 alone; combined with v3, it is both.
    assert!(all.contains(&v3) && all.len() == 2, "v1 and v3, {} changes", all.len());
    // The upstream moves again: the same bookmark moves.
    let newer = api.bookmark_heads().unwrap();
    let again = oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, newer).await.unwrap();
    assert!(matches!(again, oxygenhub::diffusion::Bumped::Moved { ref bookmark, .. } if bookmark == "diffusion/deps/api"), "{again:?}");

    // Following is off by default; owners turn it on (DF2).
    let put =
        |body: &'static str, who: Option<&'static str>| call(&h.app, "PUT", "/api/v1/repos/andy/game/diffusion", who, Some(body.into()));
    let (st, body) = put(r#"{"follow":"green"}"#, Some("site")).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert!(body.contains("\"follow\":\"green\"") && body.contains("\"from\":\"setting\""), "{body}");
    assert_eq!(put(r#"{"follow":"sometimes"}"#, Some("site")).await.0, StatusCode::BAD_REQUEST);
    assert_ne!(put(r#"{"follow":"any"}"#, None).await.0, StatusCode::OK, "not without a token");
    let (_, body) = put(r#"{"follow":"default"}"#, Some("site")).await;
    assert!(body.contains("\"follow\":\"off\"") && body.contains("\"from\":\"site\""), "{body}");

    // Dropping the subrepo drops the consumer.
    game.submodule_remove("deps/api").unwrap();
    std::fs::remove_dir_all(game_work.path().join("deps/api")).unwrap();
    game.snapshot("t").unwrap();
    game.commit("game without api", &[]).unwrap();
    assert!(game.push("origin", false).unwrap().iter().all(|r| r.ok));
    let left = settle(&|l| l.is_empty()).await;
    assert!(left.is_empty(), "{left:?}");
}

/// Diffusion D1 end to end (#148): a consumer that follows every move gets
/// a bump proposal on a push; one that follows green gets it when the
/// upstream's run passes. One proposal, kept up to date.
#[tokio::test(flavor = "multi_thread")]
async fn diffusion_proposes_a_bump_when_the_upstream_moves() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "api").await;
    create(&h.app, "andy", "game").await;
    // Run every job of the upstream's newest run to a pass, keeping each claim.
    let seen: std::sync::Mutex<Vec<serde_json::Value>> = std::sync::Mutex::new(Vec::new());
    let seen = &seen;
    let hub_app = &h.app;
    let pass_all = move || async move {
        let h_app = hub_app;
        for _ in 0..20 {
            let Some(a) = claim(h_app).await else {
                tokio::time::sleep(std::time::Duration::from_millis(30)).await;
                if !jobs_left(h_app).await {
                    break;
                }
                continue;
            };
            seen.lock().unwrap().push(a.clone());
            let id = a["job_id"].as_i64().unwrap();
            let (name, ph) = (a["job"]["name"].as_str().unwrap().to_string(), a["pipeline_hash"].as_str().unwrap().to_string());
            assert_eq!(
                finish(h_app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt(&name, &ph, 0) })).await,
                StatusCode::NO_CONTENT
            );
        }
    };
    let api_work = tempfile::tempdir().unwrap();
    let api = client(&format!("{}/andy/api", h.base), api_work.path());
    write(api_work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(api_work.path(), "test.sh", "echo ok\n");
    write(api_work.path(), "api.h", "v1\n");
    api.snapshot("t").unwrap();
    api.commit("api v1", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    pass_all().await;

    let game_work = tempfile::tempdir().unwrap();
    let game = client(&format!("{}/andy/game", h.base), game_work.path());
    game.submodule_add(&format!("{}/andy/api", h.base), "deps/api", Some("main")).unwrap();
    write(game_work.path(), "game.c", "int main;\n");
    write(game_work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(game_work.path(), "test.sh", "echo ok\n");
    game.snapshot("t").unwrap();
    game.commit("game on api v1", &[]).unwrap();
    assert!(game.push("origin", false).unwrap().iter().all(|r| r.ok));
    let game_row = db::repos::get(&h.app.db, "andy", "game").await.unwrap().unwrap();
    let bump = || async {
        for _ in 0..100 {
            let open = db::proposals::list(&h.app.db, game_row.id, &["open"]).await.unwrap();
            if let Some(p) = open.into_iter().find(|p| p.source_bookmark == "diffusion/deps/api") {
                return Some(p);
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        None
    };

    // Off by default: an upstream move proposes nothing.
    write(api_work.path(), "a2.c", "v2\n");
    api.snapshot("t").unwrap();
    api.commit("api v2: faster sprites", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    pass_all().await;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert!(db::proposals::list(&h.app.db, game_row.id, &[]).await.unwrap().is_empty(), "off proposes nothing");

    // Turned on to follow every move: bumped at once to where the upstream
    // is (v2), by the diffusion account; the next push updates it.
    let put = |body: &'static str| call(&h.app, "PUT", "/api/v1/repos/andy/game/diffusion", Some("site"), Some(body.into()));
    assert_eq!(put(r#"{"follow":"any"}"#).await.0, StatusCode::OK);
    let first = bump().await.expect("caught up when turned on");
    assert!(first.title.contains("1 change") && first.body.contains("api v2: faster sprites"), "{:?}", first.title);
    // The bump's pipeline runs, as a push of it would, so the proposal is tested.
    let mut ran = false;
    for _ in 0..100 {
        if runs_of(&h.app, "andy", "game").await.iter().any(|r| r.bookmark == "diffusion/deps/api") {
            ran = true;
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert!(ran, "a run of the bump");
    pass_all().await;
    write(api_work.path(), "a3.c", "v3\n");
    api.snapshot("t").unwrap();
    api.commit("api v3: save slots past ten", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut p = first.clone();
    for _ in 0..100 {
        p = bump().await.unwrap();
        if p.body.contains("api v3") {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(p.number, first.number, "the same proposal");
    assert_eq!((p.author.as_str(), p.target.as_str()), ("diffusion", "main"));
    assert!(p.title.contains("deps/api") && p.title.contains("andy/api") && p.title.contains("2 changes"), "{}", p.title);
    assert!(p.body.contains("api v2: faster sprites") && p.body.contains("api v3: save slots past ten"), "{}", p.body);
    let labels = db::issues::proposal_labels(&h.app.db, p.id).await.unwrap();
    assert!(labels.iter().any(|l| l.name == "dependencies"));
    pass_all().await;

    // Following green: a push alone does not move it; the passing run does,
    // and the same proposal is brought up to date.
    assert_eq!(put(r#"{"follow":"green"}"#).await.0, StatusCode::OK);
    write(api_work.path(), "a4.c", "v4\n");
    api.snapshot("t").unwrap();
    api.commit("api v4: rsm_set cheaper", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert!(!bump().await.unwrap().body.contains("api v4"), "not before its run passes");
    pass_all().await;
    let mut updated = None;
    for _ in 0..100 {
        let p2 = bump().await.unwrap();
        if p2.body.contains("api v4") {
            updated = Some(p2);
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let p2 = updated.expect("brought up to date once the run passed");
    assert_eq!(p2.number, p.number, "the same proposal");
    assert!(p2.title.contains("3 changes"), "{}", p2.title);
    assert_eq!(db::proposals::list(&h.app.db, game_row.id, &[]).await.unwrap().len(), 1, "never a pile of bumps");

    // The views (#150): the upstream's Diffusion tab lists the game with its
    // status; the game's tab and overview say what it follows.
    pass_all().await;
    let (st, page) = call(&h.app, "GET", "/andy/api/diffusion", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("Diffuses to") && page.contains("andy/game") && page.contains(&format!("#{}", p2.number)), "the bump is listed");
    assert!(page.contains("df-green"), "a passing run of a checked upstream is green");
    // The ripples backdrop: the page's own repositories and tones.
    assert!(
        page.contains(r#""scene":"ripples""#) && page.contains(r#"["andy/game","#) && page.contains(r#""repo":"andy/api""#),
        "the ripples' data"
    );
    let (_, page) = call(&h.app, "GET", "/andy/game/diffusion", None, None).await;
    assert!(page.contains("Takes from") && page.contains("andy/api"));
    let (_, overview) = call(&h.app, "GET", "/andy/game", None, None).await;
    assert!(overview.contains("Takes from") && overview.contains(">Diffusion</a>"), "the tab and the panel");
    // Both directions through the API (#153), as the extension reads them.
    let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/game/diffusion", Some("site"), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let d: hub_api::Diffusion = serde_json::from_str(&body).unwrap();
    assert!(d.diffuses_to.is_empty(), "nothing pins the game");
    let pin = d.takes_from.first().expect("the game takes from the api");
    assert_eq!((pin.upstream.as_str(), pin.consumer.as_str()), ("andy/api", "andy/game"));
    assert_eq!(pin.bump, Some(p2.number));
    assert_eq!(pin.upstream_state.tone, "green", "{body}");
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/api/diffusion", Some("site"), None).await;
    let d: hub_api::Diffusion = serde_json::from_str(&body).unwrap();
    assert!(d.takes_from.is_empty() && d.diffuses_to.iter().any(|p| p.consumer == "andy/game"), "{body}");
    let (st, _) = call(&h.app, "GET", "/admin/diffusion", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND, "administrators only");

    // Landing the bump on the hub moves the game's pin as a push would:
    // the index follows, so it is no longer behind; even when the first
    // write fails, as a busy database's would (#210: it is tried again).
    let game = oxygenhub::db::repos::get(&h.app.db, "andy", "game").await.unwrap().unwrap();
    oxygenhub::diffusion::fail_next_records(&h.app, game.id, 1);
    let (st, body) =
        call(&h.app, "POST", &format!("/api/v1/repos/andy/game/proposals/{}/land", p2.number), Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let mut behind = serde_json::Value::Null;
    for _ in 0..100 {
        let (_, list) = call(&h.app, "GET", "/api/v1/repos/andy/api/consumers", Some("site"), None).await;
        let list: Vec<serde_json::Value> = serde_json::from_str(&list).unwrap();
        behind = list.iter().find(|c| c["name"] == "game").map(|c| c["behind"].clone()).unwrap_or_default();
        if behind == 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(behind, 0, "the index followed the landing");
    assert_eq!(oxygenhub::diffusion::forced_left(&h.app, game.id), 0, "the first write did fail");
    // The reconciler (#210): an index left behind its bookmark (as when
    // both writes failed) is mended from the bookmark itself, and a pass
    // with nothing to mend changes nothing.
    let pins_of = || async {
        sqlx::query_scalar::<_, String>("SELECT heads FROM diffusion_pins WHERE consumer_id = $1")
            .bind(game.id)
            .fetch_one(&h.app.db)
            .await
            .unwrap()
    };
    let right = pins_of().await;
    sqlx::query("UPDATE diffusion_pins SET heads = 'stale' WHERE consumer_id = $1").bind(game.id).execute(&h.app.db).await.unwrap();
    sqlx::query("DELETE FROM diffusion_indexed WHERE consumer_id = $1").bind(game.id).execute(&h.app.db).await.unwrap();
    let r = oxygenhub::diffusion::reconcile(&h.app).await;
    assert!(r.reindexed >= 1, "{r:?}");
    assert_eq!(pins_of().await, right, "the pin read again from the game's bookmark");
    assert_eq!(oxygenhub::diffusion::reconcile(&h.app).await.reindexed, 0, "nothing left to mend");

    // A private upstream: the bump's run (started by the diffusion account,
    // which cannot read it) still gets the subrepo it bumps.
    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'api'").execute(&h.app.db).await.unwrap();
    seen.lock().unwrap().clear();
    write(api_work.path(), "a6.c", "v6\n");
    api.snapshot("t").unwrap();
    api.commit("api v6", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    // The bump's run starts in the background: wait for its job, under a
    // loaded test run as well as alone.
    let mut bump_job = None;
    for _ in 0..100 {
        pass_all().await;
        bump_job = seen.lock().unwrap().iter().find(|c| c["repo"] == "andy/game" && c["bookmark"] == "diffusion/deps/api").cloned();
        if bump_job.is_some() {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
    let bump_job = bump_job.expect("the bump's job ran");
    let subs: Vec<&str> = bump_job["subrepos"].as_array().unwrap().iter().filter_map(|s| s["path"].as_str()).collect();
    assert_eq!(subs, vec!["deps/api"], "{bump_job}");

    // A consumer turning Diffusion on is bumped at once to the upstream's
    // newest passing state, without waiting for it to move again.
    create(&h.app, "andy", "game2").await;
    let g2_work = tempfile::tempdir().unwrap();
    let g2 = client(&format!("{}/andy/game2", h.base), g2_work.path());
    g2.submodule_add(&format!("{}/andy/api", h.base), "deps/api", Some("main")).unwrap();
    g2.snapshot("t").unwrap();
    g2.commit("game2 on api", &[]).unwrap();
    assert!(g2.push("origin", false).unwrap().iter().all(|r| r.ok));
    // Its pin is api's newest state already: move api on, pass the run.
    write(api_work.path(), "a5.c", "v5\n");
    api.snapshot("t").unwrap();
    api.commit("api v5", &[]).unwrap();
    let api_row = db::repos::get(&h.app.db, "andy", "api").await.unwrap().unwrap();
    let runs_before = db::flow::runs(&h.app.db, api_row.id, 100, None).await.unwrap().len();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    // The push's run is made in the background: wait for it, or pass_all
    // can finish before it exists and v5 never passes (a flake under load).
    for _ in 0..300 {
        if db::flow::runs(&h.app.db, api_row.id, 100, None).await.unwrap().len() > runs_before {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
    pass_all().await;
    let g2_row = db::repos::get(&h.app.db, "andy", "game2").await.unwrap().unwrap();
    for _ in 0..100 {
        if sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM diffusion_pins WHERE consumer_id = $1")
            .bind(g2_row.id)
            .fetch_one(&h.app.db)
            .await
            .unwrap()
            > 0
        {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
    }
    assert!(db::proposals::list(&h.app.db, g2_row.id, &[]).await.unwrap().is_empty(), "off until turned on");
    let (st, _) = call(&h.app, "PUT", "/api/v1/repos/andy/game2/diffusion", Some("site"), Some(r#"{"follow":"green"}"#.into())).await;
    assert_eq!(st, StatusCode::OK);
    let mut caught = None;
    // Up to 15 s: under a full parallel test run the bump can take a while.
    for _ in 0..300 {
        if let Some(p) = db::proposals::list(&h.app.db, g2_row.id, &["open"]).await.unwrap().into_iter().next() {
            caught = Some(p);
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let caught = caught.expect("bumped as soon as it was turned on");
    assert!(caught.body.contains("api v5") && caught.title.contains("1 change"), "{:?} {:?}", caught.title, caught.body);
    // A bump someone closed is a "no": the reconciler does not open it again.
    sqlx::query("UPDATE proposals SET state = 'closed' WHERE id = $1").bind(caught.id).execute(&h.app.db).await.unwrap();
    // (The bump's own runs finishing may still propose again, as Diffusion
    // does on an upstream's moves; the reconciler itself opens nothing.)
    let r = oxygenhub::diffusion::reconcile(&h.app).await;
    assert_eq!(r.caught_up, 0, "the reconciler respects the closed bump: {r:?}");
}

/// Whether any job is queued or running.
async fn jobs_left(app: &Shared) -> bool {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM flow_jobs WHERE status IN ('queued', 'running')")
        .fetch_one(&app.db)
        .await
        .unwrap_or(0)
        > 0
}

/// An upstream with no checks never turns green: a consumer following green
/// still gets its bumps, marked amber, `upstream unverified` (#148).
#[tokio::test(flavor = "multi_thread")]
async fn bumps_from_an_upstream_without_checks_are_marked_unverified() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "lib").await;
    create(&h.app, "andy", "app").await;
    let lib_work = tempfile::tempdir().unwrap();
    let lib = client(&format!("{}/andy/lib", h.base), lib_work.path());
    write(lib_work.path(), "lib.h", "v1\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib v1", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let app_work = tempfile::tempdir().unwrap();
    let app = client(&format!("{}/andy/app", h.base), app_work.path());
    app.submodule_add(&format!("{}/andy/lib", h.base), "deps/lib", Some("main")).unwrap();
    // It watches lib.c: upstream changes touching it are flagged.
    let modules = std::fs::read_to_string(app_work.path().join(".oxymodules")).unwrap();
    write(app_work.path(), ".oxymodules", &format!("{modules}watch = lib.c\n"));
    app.snapshot("t").unwrap();
    app.commit("app on lib", &[]).unwrap();
    assert!(app.push("origin", false).unwrap().iter().all(|r| r.ok));
    let app_row = db::repos::get(&h.app.db, "andy", "app").await.unwrap().unwrap();
    for _ in 0..100 {
        if sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM diffusion_pins WHERE consumer_id = $1")
            .bind(app_row.id)
            .fetch_one(&h.app.db)
            .await
            .unwrap()
            > 0
        {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
    }
    let (st, _) = call(&h.app, "PUT", "/api/v1/repos/andy/app/diffusion", Some("site"), Some(r#"{"follow":"green"}"#.into())).await;
    assert_eq!(st, StatusCode::OK);
    write(lib_work.path(), "lib.c", "v2\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib v2", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut found = None;
    for _ in 0..100 {
        if let Some(p) = db::proposals::list(&h.app.db, app_row.id, &["open"]).await.unwrap().into_iter().next() {
            found = Some(p);
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let p = found.expect("a bump although the upstream never goes green");
    assert!(p.body.contains("Upstream unverified"), "{}", p.body);
    assert!(p.body.contains("Touches what you watch") && p.body.contains("touches `lib.c`") && p.body.contains("⚑"), "{}", p.body);
    let mut labels = Vec::new();
    for _ in 0..100 {
        labels = db::issues::proposal_labels(&h.app.db, p.id).await.unwrap().into_iter().map(|l| l.name).collect::<Vec<String>>();
        if labels.len() == 2 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
    }
    assert!(labels.contains(&"upstream unverified".to_string()) && labels.contains(&"dependencies".to_string()), "{labels:?}");
}

/// #134: when a proposal's target moves, its projected state is new; the
/// hub runs the pipeline on it, so its checks are never for an older one.
#[tokio::test(flavor = "multi_thread")]
async fn a_moved_target_runs_its_open_proposals_again() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&format!("{}/andy/demo", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("t").unwrap();
    let base = c.commit("base", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    // Two proposals, each main plus one change.
    for (b, file) in [("fix-a", "a.txt"), ("fix-b", "b.txt")] {
        c.bookmark_create(b, Some(&[base])).unwrap();
        c.bookmark_switch(b, true).unwrap();
        write(work.path(), file, "x\n");
        c.snapshot("t").unwrap();
        c.commit(&format!("add {file}"), &[]).unwrap();
        assert!(c.push_bookmark("origin", b, false).unwrap().iter().all(|r| r.ok));
        let body = serde_json::json!({ "title": b, "target": "main", "source": b }).to_string();
        let (st, out) = call(&h.app, "POST", "/api/v1/repos/andy/demo/proposals", Some("site"), Some(body)).await;
        assert_eq!(st, StatusCode::CREATED, "{out}");
    }
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let props = db::proposals::list(&h.app.db, row.id, &["open"]).await.unwrap();
    let a = props.iter().find(|p| p.source_bookmark == "fix-a").unwrap().number;
    let b = props.iter().find(|p| p.source_bookmark == "fix-b").unwrap().number;
    // Landing fix-a moves main: fix-b's projected state gets a run.
    let (st, out) = call(&h.app, "POST", &format!("/api/v1/repos/andy/demo/proposals/{a}/land"), Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::OK, "{out}");
    let label = format!("proposal/{b}");
    let mut found = None;
    for _ in 0..100 {
        if let Some(r) = runs_of(&h.app, "andy", "demo").await.into_iter().find(|r| r.bookmark == label) {
            found = Some(r);
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let run = found.expect("a run of fix-b's new projected state");
    let projected = db::proposals::get(&h.app.db, row.id, b).await.unwrap().unwrap();
    assert_eq!((run.trigger.as_str(), run.state.as_str()), ("proposal", projected.projected_state.as_str()), "on its projected state");
    // Moving main again by a push does it once more; the same state is not run twice.
    let before = runs_of(&h.app, "andy", "demo").await.len();
    oxygenhub::web::proposals::settle_after_push(&h.app, &row, "main", "andy").await;
    assert_eq!(runs_of(&h.app, "andy", "demo").await.len(), before, "not again for the same state");
}

/// Send image bytes to the hub as their chunks and then a manifest that
/// names no architecture; the manifest as the hub answers it.
async fn send_image(app: &Shared, image: &[u8]) -> (String, serde_json::Value) {
    let mut chunks = std::collections::HashMap::new();
    let m = jet::images::scan(image, None, |c, b| {
        chunks.insert(c.hash().to_string(), b.to_vec());
        Ok(())
    })
    .unwrap();
    for c in m.needed() {
        let (st, body) = put_bytes(app, &format!("/api/v1/flow/images/chunks/{c}"), "site", chunks[c].clone()).await;
        assert_eq!(st, StatusCode::NO_CONTENT, "{body}");
    }
    let (st, body) =
        call(app, "PUT", &format!("/api/v1/flow/images/{}", m.hash), Some("site"), Some(serde_json::to_string(&m).unwrap())).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    (m.hash.clone(), serde_json::from_str(&body).unwrap())
}

/// oxygenhub#173 (OxygenFlow spec §11.2, Andy's "detect and learn"): an
/// image sent with no architecture is read from its own files, so a
/// riscv64 runner is never offered an x86_64 image's jobs; an image
/// nobody can read takes the architecture its first pass ran on.
#[tokio::test(flavor = "multi_thread")]
async fn an_image_names_its_own_architecture_and_a_pass_teaches_the_rest() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    // An ext4 image whose shell is x86_64, as `flow rootfs import` makes one.
    let dir = tempfile::tempdir().unwrap();
    let tree = dir.path().join("tree");
    std::fs::create_dir_all(tree.join("bin")).unwrap();
    let mut elf = vec![0u8; 64];
    elf[..7].copy_from_slice(b"\x7fELF\x02\x01\x01");
    elf[18..20].copy_from_slice(&0x3Eu16.to_le_bytes());
    std::fs::write(tree.join("bin/sh"), &elf).unwrap();
    let img = dir.path().join("rootfs.ext4");
    let made = std::process::Command::new("mkfs.ext4").args(["-q", "-F", "-d"]).arg(&tree).arg(&img).arg("8M").status();
    if !made.is_ok_and(|s| s.success()) {
        eprintln!("mkfs.ext4 not found: skipped");
        return;
    }
    let (x86, answer) = send_image(&h.app, &std::fs::read(&img).unwrap()).await;
    assert_eq!(answer["arch"], "x86_64", "read from its shell: {answer}");
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/flow/images/{x86}"), Some("runner-secret"), None).await;
    assert_eq!(serde_json::from_str::<jet::images::Manifest>(&body).unwrap().arch.as_deref(), Some("x86_64"), "runners fetch it named");
    // One nobody can read: no ext4, no shell.
    let blob: Vec<u8> = (0..300_000u32).map(|i| (i % 241) as u8).collect();
    let (unread, answer) = send_image(&h.app, &blob).await;
    assert!(answer["arch"].is_null(), "{answer}");
    let archs = db::flow::image_archs(&h.app.db).await.unwrap();
    assert_eq!((archs.get(&x86).map(String::as_str), archs.get(&unread)), (Some("x86_64"), None));
    // Kept images with none are read once at start.
    sqlx::query("UPDATE jet_images SET arch = ''").execute(&h.app.db).await.unwrap();
    sqlx::query("DELETE FROM flow_images").execute(&h.app.db).await.unwrap();
    assert_eq!(oxygenhub::images::learn_archs(&h.app).await.unwrap(), 1);
    assert_eq!(db::flow::image_archs(&h.app.db).await.unwrap().get(&x86).map(String::as_str), Some("x86_64"));

    // A job on each image.
    create(&h.app, "andy", "arch").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/arch", h.base), work.path());
    let kdl = format!(
        "schema-version 1\nsandbox \"unread\" {{\n    rootfs \"cas://blob/{unread}\"\n}}\nsandbox \"x86\" {{\n    rootfs \"cas://blob/{x86}\"\n}}\n\
         job \"first\" {{\n    sandbox \"unread\"\n    exec \"make\"\n}}\njob \"second\" {{\n    sandbox \"x86\"\n    exec \"make\"\n    depends-on \"first\"\n}}\n\
         job \"third\" {{\n    sandbox \"unread\"\n    exec \"make\"\n    depends-on \"first\"\n}}\n"
    );
    write(work.path(), ".flow/pipeline.kdl", &kdl);
    c.commit("Jobs on two images", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "arch").await;

    // The x86_64 runner passes the first job: its image is x86_64 now.
    let x86_info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
    let x86_claim =
        serde_json::json!({ "runner": "earth", "runtimes": ["microvm"], "images": [], "fetches_images": true, "info": x86_info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(x86_claim.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let first: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(first["job"]["name"], "first");
    let ph = first["pipeline_hash"].as_str().unwrap();
    let id = first["job_id"].as_i64().unwrap();
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("first", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(db::flow::image_archs(&h.app.db).await.unwrap().get(&unread).map(String::as_str), Some("x86_64"), "learned");

    // A riscv64 runner is offered neither: one read, one learned.
    let token = "flowreg_narvi";
    db::runners::add_registration(&h.app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &Default::default()).await.unwrap();
    let identity = SecretKey::generate().identity("narvi", "", 1);
    let info = serde_json::json!({ "arch": "riscv64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "narvi", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    let narvi =
        serde_json::json!({ "runner": "narvi", "runtimes": ["microvm"], "images": [], "fetches_images": true, "info": info }).to_string();
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(narvi)).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "the second job's image is x86_64 and the third's learned so: {body}");
    for _ in 0..2 {
        let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(x86_claim.to_string())).await;
        assert_eq!(st, StatusCode::OK, "both go to x86_64: {body}");
    }
}

/// Diffusion auto-land (docs/diffusion.md §4.2; Andy chose "opt-in, per
/// repository", 2026-10-13): a consumer that chose `land = auto` has a bump
/// landed by itself, as the diffusion account, once both halves are green;
/// a bump from an upstream with no checks (amber) always waits.
#[tokio::test(flavor = "multi_thread")]
async fn a_consumer_that_chose_auto_lands_green_bumps_itself() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "api").await;
    create(&h.app, "andy", "game").await;
    let hub_app = &h.app;
    // Run every queued job to a pass.
    let pass_all = move || async move {
        for _ in 0..20 {
            let Some(a) = claim(hub_app).await else {
                tokio::time::sleep(std::time::Duration::from_millis(30)).await;
                if !jobs_left(hub_app).await {
                    break;
                }
                continue;
            };
            let id = a["job_id"].as_i64().unwrap();
            let (name, ph) = (a["job"]["name"].as_str().unwrap().to_string(), a["pipeline_hash"].as_str().unwrap().to_string());
            assert_eq!(
                finish(hub_app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt(&name, &ph, 0) })).await,
                StatusCode::NO_CONTENT
            );
        }
    };
    let api_work = tempfile::tempdir().unwrap();
    let api = client(&format!("{}/andy/api", h.base), api_work.path());
    write(api_work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(api_work.path(), "test.sh", "echo ok\n");
    api.snapshot("t").unwrap();
    api.commit("api v1", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    pass_all().await;
    let game_work = tempfile::tempdir().unwrap();
    let game = client(&format!("{}/andy/game", h.base), game_work.path());
    game.submodule_add(&format!("{}/andy/api", h.base), "deps/api", Some("main")).unwrap();
    write(game_work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(game_work.path(), "test.sh", "echo ok\n");
    game.snapshot("t").unwrap();
    game.commit("game on api v1", &[]).unwrap();
    assert!(game.push("origin", false).unwrap().iter().all(|r| r.ok));
    pass_all().await;
    let game_row = db::repos::get(&h.app.db, "andy", "game").await.unwrap().unwrap();
    for _ in 0..100 {
        let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM diffusion_pins WHERE consumer_id = $1")
            .bind(game_row.id)
            .fetch_one(&h.app.db)
            .await
            .unwrap();
        if n > 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
    }
    let (st, body) =
        call(&h.app, "PUT", "/api/v1/repos/andy/game/diffusion", Some("site"), Some(r#"{"follow":"green","land":"auto"}"#.into())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert!(body.contains(r#""land":"auto""#), "{body}");
    let (st, _) =
        call(&h.app, "PUT", "/api/v1/repos/andy/game/diffusion", Some("site"), Some(r#"{"follow":"green","land":"soon"}"#.into())).await;
    assert_eq!(st, StatusCode::BAD_REQUEST);

    // The upstream moves and passes: the bump is proposed, its run passes,
    // and it lands by itself.
    write(api_work.path(), "a2.c", "v2\n");
    api.snapshot("t").unwrap();
    api.commit("api v2: faster sprites", &[]).unwrap();
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut landed = None;
    for _ in 0..100 {
        pass_all().await;
        let all = db::proposals::list(&h.app.db, game_row.id, &[]).await.unwrap();
        if let Some(p) = all.into_iter().find(|p| p.source_bookmark == "diffusion/deps/api" && p.state == "landed") {
            landed = Some(p);
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let p = landed.expect("landed by itself once both halves were green");
    assert!(p.body.contains("lands its bumps itself"), "{}", p.body);
    let events = db::proposals::events(&h.app.db, p.id).await.unwrap();
    let landing = events.iter().find(|e| e.event == "landed").expect("a landing on the timeline");
    assert_eq!(landing.actor, "diffusion");
    assert!(landing.detail.contains("Diffusion, both halves green"), "{}", landing.detail);
    // The game's main now pins api v2.
    let mut behind = serde_json::Value::Null;
    for _ in 0..100 {
        let (_, list) = call(&h.app, "GET", "/api/v1/repos/andy/api/consumers", Some("site"), None).await;
        let list: Vec<serde_json::Value> = serde_json::from_str(&list).unwrap();
        behind = list.iter().find(|c| c["name"] == "game").map(|c| c["behind"].clone()).unwrap_or_default();
        if behind == 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(behind, 0, "main moved to the bump");

    // An upstream with no checks: its bump passes in the consumer, but
    // amber never lands by itself, whatever the setting.
    create(&h.app, "andy", "lib").await;
    create(&h.app, "andy", "app").await;
    let lib_work = tempfile::tempdir().unwrap();
    let lib = client(&format!("{}/andy/lib", h.base), lib_work.path());
    write(lib_work.path(), "lib.h", "v1\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib v1", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let app_work = tempfile::tempdir().unwrap();
    let app = client(&format!("{}/andy/app", h.base), app_work.path());
    app.submodule_add(&format!("{}/andy/lib", h.base), "deps/lib", Some("main")).unwrap();
    let modules = std::fs::read_to_string(app_work.path().join(".oxymodules")).unwrap();
    write(app_work.path(), ".oxymodules", &format!("{modules}follow = green\nland = auto\n"));
    write(app_work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(app_work.path(), "test.sh", "echo ok\n");
    app.snapshot("t").unwrap();
    app.commit("app on lib", &[]).unwrap();
    assert!(app.push("origin", false).unwrap().iter().all(|r| r.ok));
    pass_all().await;
    let app_row = db::repos::get(&h.app.db, "andy", "app").await.unwrap().unwrap();
    for _ in 0..100 {
        let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM diffusion_pins WHERE consumer_id = $1")
            .bind(app_row.id)
            .fetch_one(&h.app.db)
            .await
            .unwrap();
        if n > 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
    }
    write(lib_work.path(), "lib.c", "v2\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib v2", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut passed = false;
    for _ in 0..100 {
        pass_all().await;
        let s: Option<String> = sqlx::query_scalar(
            "SELECT status FROM flow_runs WHERE repo_id = $1 AND bookmark = 'diffusion/deps/lib' ORDER BY id DESC LIMIT 1",
        )
        .bind(app_row.id)
        .fetch_optional(&h.app.db)
        .await
        .unwrap();
        if s.as_deref() == Some("passed") {
            passed = true;
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert!(passed, "the app's run on the bump passed");
    assert_eq!(oxygenhub::diffusion::auto_land(&h.app, &app_row, "deps/lib").await, Err(oxygenhub::diffusion::Held::Unverified));
    let open = db::proposals::list(&h.app.db, app_row.id, &["open"]).await.unwrap();
    assert!(open.iter().any(|p| p.source_bookmark == "diffusion/deps/lib"), "amber waits for a person");
}

/// oxygenflow#31: one finished job runs again in the same run, with the
/// jobs that depend on it; what it depends on keeps its result, so a flaky
/// test reruns without its build. Not while the run is live, and only a
/// job the run has.
#[tokio::test(flavor = "multi_thread")]
async fn one_job_reruns_with_what_depends_on_it() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "flaky").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/flaky", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "flaky").await[0].clone();
    // build and lint pass; test fails, as a flaky one does.
    let (st, _) =
        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/test/rerun", run.number), Some("site"), None).await;
    assert_eq!(st, StatusCode::CONFLICT, "not while the run is live");
    while let Some(j) = claim(&h.app).await {
        let id = j["job_id"].as_i64().unwrap();
        let (name, ph) = (j["job"]["name"].as_str().unwrap().to_string(), j["pipeline_hash"].as_str().unwrap().to_string());
        let exit = if name == "test" { 1 } else { 0 };
        let log = serde_json::json!({ "records": [{ "t": 1, "stream": "out", "text": format!("{name} ran") }] });
        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(log.to_string())).await;
        assert_eq!(
            finish(&h.app, id, serde_json::json!({ "exit_code": exit, "duration_ms": 10, "receipt": receipt(&name, &ph, exit as u32) }))
                .await,
            StatusCode::NO_CONTENT
        );
    }
    let status = |n: &'static str| {
        let (app, run_id) = (h.app.clone(), run.id);
        async move { db::flow::jobs(&app.db, run_id).await.unwrap().into_iter().find(|j| j.name == n).unwrap() }
    };
    assert_eq!(db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap().status, "failed");
    let failed_test = status("test").await;

    // The test again: build and lint keep their results.
    let (st, body) =
        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/test/rerun", run.number), Some("site"), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert!(body.contains(r#""queued":1"#), "{body}");
    assert_eq!(
        (status("build").await.status, status("lint").await.status, status("test").await.status),
        ("passed".into(), "passed".into(), "queued".into())
    );
    assert!(db::flow::log(&h.app.db, failed_test.id, 0, 100).await.unwrap().is_empty(), "its old log is gone");
    assert!(db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap().live());
    let checks = checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(checks.iter().find(|c| c.name == "flow/test").map(|c| c.status.as_str()), Some("queued"));
    let j = claim(&h.app).await.expect("the test, queued again");
    assert_eq!(j["job"]["name"], "test");
    let ph = j["pipeline_hash"].as_str().unwrap().to_string();
    let id = j["job_id"].as_i64().unwrap();
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt("test", &ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap().status, "passed", "the run is whole again");
    // Andy owns it: told it failed, then that it is fixed (#183).
    let andy_id = db::accounts::account_by_name(&h.app.db, "andy").await.unwrap().unwrap().id;
    let kinds: Vec<String> = db::notify::inbox(&h.app.db, andy_id, 20).await.unwrap().into_iter().map(|n| n.kind).collect();
    assert!(kinds.contains(&"flow.failed".to_string()) && kinds.contains(&"flow.fixed".to_string()), "{kinds:?}");

    // The build again brings the test that depends on it; lint stays.
    let (_, body) =
        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/build/rerun", run.number), Some("site"), None).await;
    assert!(body.contains(r#""queued":2"#), "{body}");
    assert_eq!(
        (status("build").await.status, status("lint").await.status, status("test").await.status),
        ("queued".into(), "passed".into(), "queued".into())
    );
    let (st, _) =
        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/deploy/rerun", run.number), Some("site"), None).await;
    assert_eq!(st, StatusCode::CONFLICT, "live again, and no such job either way");
}

/// oxygenflow#30: a failed run tells its pusher and the repository's
/// watchers; a runner whose last five jobs ended in infrastructure errors
/// is drained and the administrators told, counting only since an
/// administrator last made it active.
#[tokio::test(flavor = "multi_thread")]
async fn failed_runs_are_told_and_an_erroring_runner_is_drained() {
    let h = hub().await;
    let andy = db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    let bob = db::accounts::create_account(&h.app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();
    create(&h.app, "andy", "alerts").await;
    let row = db::repos::get(&h.app.db, "andy", "alerts").await.unwrap().unwrap();
    db::notify::set_watch(&h.app.db, bob.id, row.id, true).await.unwrap();
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/alerts", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "exit 1\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "alerts").await[0].clone();
    sqlx::query("UPDATE flow_runs SET actor = 'andy' WHERE id = $1").bind(run.id).execute(&h.app.db).await.unwrap();
    while let Some(j) = claim(&h.app).await {
        let id = j["job_id"].as_i64().unwrap();
        let (name, ph) = (j["job"]["name"].as_str().unwrap().to_string(), j["pipeline_hash"].as_str().unwrap().to_string());
        let exit = if name == "test" { 1 } else { 0 };
        finish(&h.app, id, serde_json::json!({ "exit_code": exit, "duration_ms": 10, "receipt": receipt(&name, &ph, exit as u32) })).await;
    }
    let told = |id: i64| {
        let db = h.app.db.clone();
        async move { db::notify::inbox(&db, id, 20).await.unwrap().into_iter().filter(|n| n.kind == "flow.failed").collect::<Vec<_>>() }
    };
    let (for_andy, for_bob) = (told(andy.id).await, told(bob.id).await);
    assert_eq!((for_andy.len(), for_bob.len()), (1, 1), "the pusher and a watcher");
    assert!(for_andy[0].title.contains(&format!("run #{} on main failed (test)", run.number)), "{}", for_andy[0].title);
    assert_eq!(for_andy[0].url, format!("/andy/alerts/flow/{}", run.number));

    // Five infrastructure errors in a row drain the runner.
    let runner_id = || {
        let db = h.app.db.clone();
        async move { sqlx::query_scalar::<_, i64>("SELECT id FROM flow_runners WHERE name = 'test-1'").fetch_one(&db).await.unwrap() }
    };
    let mut errors = 0;
    while errors < 5 {
        let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/alerts/flow/runs", Some("site"), Some("{}".into())).await;
        assert_eq!(st, StatusCode::CREATED);
        while let Some(j) = claim(&h.app).await {
            let id = j["job_id"].as_i64().unwrap();
            let st = finish(
                &h.app,
                id,
                serde_json::json!({ "exit_code": null, "duration_ms": 10, "error": "the microVM did not start: /dev/kvm busy" }),
            )
            .await;
            assert_eq!(st, StatusCode::NO_CONTENT);
            errors += 1;
            if errors == 5 {
                break;
            }
        }
    }
    let state = |id: i64| {
        let db = h.app.db.clone();
        async move { db::runners::get(&db, id).await.unwrap().unwrap().state }
    };
    let id = runner_id().await;
    assert_eq!(state(id).await, "draining");
    let drained: Vec<_> =
        db::notify::inbox(&h.app.db, andy.id, 50).await.unwrap().into_iter().filter(|n| n.kind == "flow.runner_drained").collect();
    assert_eq!(drained.len(), 1, "the administrator is told");
    assert!(drained[0].title.contains("/dev/kvm busy"), "{}", drained[0].title);
    assert!(
        !told(bob.id).await.is_empty()
            && db::notify::inbox(&h.app.db, bob.id, 50).await.unwrap().iter().all(|n| n.kind != "flow.runner_drained")
    );
    // Made active again, past errors no longer count.
    db::runners::set_state(&h.app.db, id, "active").await.unwrap();
    assert_eq!(db::runners::erroring(&h.app.db, id, 5).await.unwrap(), None);
}

/// #225: an ordinary user turns Flow on for their repository on their own
/// runner, and the site's runners never take its jobs.
#[tokio::test(flavor = "multi_thread")]
async fn owners_run_flow_on_their_own_runners_only() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    let register = |name: &'static str| {
        let app = h.app.clone();
        async move {
            let body = format!("username={name}&email={name}%40example.com&password=correct-horse-battery");
            let res = oxygenhub::router(app)
                .oneshot(
                    Request::post("/register")
                        .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                        .body(Body::from(body))
                        .unwrap(),
                )
                .await
                .unwrap();
            res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string()
        }
    };
    register("andy").await; // the first account: the site's administrator
    let bob = register("bob").await;
    create(&h.app, "bob", "game").await;
    let web = |method: &'static str, path: &str, body: Option<String>| {
        let (app, cookie, path) = (h.app.clone(), bob.clone(), path.to_string());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned()
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    // No runner of his own yet: he is told how, and cannot switch it on.
    let page = web("GET", "/bob/game/settings", None).await;
    assert!(page.contains("register a runner for this repository") && !page.contains("value=\"own\""), "{page}");
    let page = web("POST", "/bob/game/settings/flow", Some(format!("_csrf={}&flow=own", csrf_of(&page)))).await;
    assert!(page.contains("Register a runner for this repository"), "{page}");
    // He registers one.
    let page = web("GET", "/bob/game/settings", None).await;
    let page = web("POST", "/bob/game/settings/runners/token", Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
    let identity = SecretKey::generate().identity("bobs", "", 1);
    let reg = serde_json::json!({ "token": token, "name": "bobs", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    // The site's runners stay a site administrator's to grant; his own he may choose.
    let page = web("GET", "/bob/game/settings", None).await;
    assert!(page.contains("value=\"own\"") && !page.contains("value=\"on\""), "{page}");
    let page = web("POST", "/bob/game/settings/flow", Some(format!("_csrf={}&flow=on", csrf_of(&page)))).await;
    assert!(page.contains("Only site administrators let the site's runners"), "{page}");
    let page = web("POST", "/bob/game/settings/flow", Some(format!("_csrf={}&flow=own", csrf_of(&page)))).await;
    assert!(page.contains("value=\"own\" checked") && page.contains("start runs"), "{page}");
    assert_eq!(db::repos::get(&h.app.db, "bob", "game").await.unwrap().unwrap().flow, "own");
    // A push starts a run; the site's runner is not offered it, his is.
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/bob/game", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("t").unwrap();
    c.commit("A pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "bob", "game").await.len(), 1, "a run started");
    assert!(claim(&h.app).await.is_none(), "the site's runner is not offered it");
    let body = serde_json::json!({ "runner": "ignored", "runtimes": ["microvm"], "info": info }).to_string();
    let (st, job) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(body)).await;
    assert_eq!(st, StatusCode::OK, "{job}");
    assert_eq!(serde_json::from_str::<serde_json::Value>(&job).unwrap()["repo"], "bob/game");
}

#[tokio::test(flavor = "multi_thread")]
async fn a_job_takes_the_artifact_its_producer_kept() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"plan\" {{\n    sandbox \"alpine\"\n    exec \"plan\"\n    output {{\n        artifact \"tfplan\" {{\n            path \"/src/out/tfplan\"\n        }}\n    }}\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n    depends-on \"plan\"\n    input {{\n        artifact \"tfplan\" from \"plan\"\n        path \"/src/tfplan\"\n    }}\n    exec \"apply\"\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Plan, then apply that plan", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The plan job runs and keeps its plan, named by its receipt.
    let runs = runs_of(&h.app, "andy", "demo").await;
    assert_eq!(runs.len(), 1, "a run");
    let jobs = db::flow::jobs(&h.app.db, runs[0].id).await.unwrap();
    assert!(!jobs.is_empty(), "jobs: {:?}", runs[0]);
    let p = claim(&h.app)
        .await
        .unwrap_or_else(|| panic!("nothing to claim: {:?}", jobs.iter().map(|j| (&j.name, &j.status, &j.error)).collect::<Vec<_>>()));
    assert_eq!(p["job"]["name"], "plan");
    let plan_id = p["job_id"].as_i64().unwrap();
    let plan = b"the reviewed plan".to_vec();
    let (st, text) = upload(&h.app, plan_id, "tfplan", &plan).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let blob = oxy_core::repo::Repo::blob_ref(&plan).to_hex();
    let mut r = receipt("plan", p["pipeline_hash"].as_str().unwrap(), 0);
    r["outputs"] = serde_json::json!([blob]);
    assert_eq!(
        finish(&h.app, plan_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r })).await,
        StatusCode::NO_CONTENT
    );

    // The apply job is told what to take and the blob to expect, and the
    // runner fetches exactly those bytes.
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["job"]["name"], "apply");
    let apply_id = a["job_id"].as_i64().unwrap();
    assert_eq!(
        a["inputs"],
        serde_json::json!([{ "artifact": "tfplan", "from": "plan", "path": "/src/tfplan", "blob": blob }]),
        "the claim names the producer's blob"
    );
    let req = Request::get(format!("/api/v1/flow/jobs/{apply_id}/inputs/0")).header(header::AUTHORIZATION, "Bearer runner-secret");
    let res = oxygenhub::router(h.app.clone()).oneshot(req.body(Body::empty()).unwrap()).await.unwrap();
    assert_eq!(res.status(), StatusCode::OK);
    assert_eq!(res.into_body().collect().await.unwrap().to_bytes().to_vec(), plan);
    let (st, _) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{apply_id}/inputs/1"), Some("runner-secret"), None).await;
    assert_eq!(st, StatusCode::NOT_FOUND, "no second input");
    let (st, _) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{apply_id}/inputs/0"), None, None).await;
    assert_eq!(st, StatusCode::UNAUTHORIZED, "runners only");
    let (st, _) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{plan_id}/inputs/0"), Some("runner-secret"), None).await;
    assert_eq!(st, StatusCode::CONFLICT, "only a running job's");
}

#[tokio::test(flavor = "multi_thread")]
async fn a_job_waits_for_approval_and_is_approved_rejected_or_expires() {
    let h = hub().await;
    let andy = db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    let bob = db::accounts::create_account(&h.app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();
    let carol = db::accounts::create_account(&h.app.db, "carol", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "gated").await;
    let row = db::repos::get(&h.app.db, "andy", "gated").await.unwrap().unwrap();
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/gated", h.base), work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"plan\" {{\n    sandbox \"alpine\"\n    exec \"plan\"\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n    depends-on \"plan\"\n    approval {{\n        by \"maintain\"\n        others-only #true\n        sealed #false\n    }}\n    environment \"prod\"\n    identity \"https://sts.example\" {{\n        mount-path \"/run/secrets/cloud\"\n    }}\n    exec \"apply\"\n}}\n"
        ),
    );
    let mut n = 0;
    let mut push = || {
        n += 1;
        write(work.path(), "n.txt", &format!("{n}\n"));
        c.snapshot("test").unwrap();
        c.commit(&format!("Push {n}"), &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    };
    // Runs `plan` to a pass; returns the run, its actor set to andy.
    let plan_passes = |number: i64| {
        let app = h.app.clone();
        let row = row.clone();
        async move {
            let mut run = None;
            for _ in 0..100 {
                if let Some(r) = db::flow::run(&app.db, row.id, number).await.unwrap() {
                    run = Some(r);
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(30)).await;
            }
            let run = run.expect("the run");
            sqlx::query("UPDATE flow_runs SET actor = 'andy' WHERE id = $1").bind(run.id).execute(&app.db).await.unwrap();
            let j = claim(&app).await.expect("plan is claimed");
            assert_eq!(j["job"]["name"], "plan");
            let (id, ph) = (j["job_id"].as_i64().unwrap(), j["pipeline_hash"].as_str().unwrap().to_string());
            finish(&app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt("plan", &ph, 0) })).await;
            db::flow::run(&app.db, row.id, number).await.unwrap().unwrap()
        }
    };

    // Run 1: apply waits; only someone with the role, other than the pusher, approves it.
    push();
    let run = runs_of(&h.app, "andy", "gated").await[0].clone();
    // Live pages hear of the push (oxygenhub#203).
    let heard = |topic: String| match h.app.events.live.since(0, &[topic]) {
        oxygenhub::events::Replay::Events(e) => e.into_iter().map(|e| e.kind).collect::<Vec<_>>(),
        oxygenhub::events::Replay::Reload => vec!["reload".into()],
    };
    assert!(heard(oxygenhub::events::topic::repo(row.id)).contains(&"push".to_string()), "the push is on the repository's topic");
    sqlx::query("UPDATE flow_runs SET actor = 'andy' WHERE id = $1").bind(run.id).execute(&h.app.db).await.unwrap();
    let early = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, true).await.unwrap_err();
    assert!(early.contains("waits for plan first"), "{early}");
    let run = plan_passes(run.number).await;
    assert!(claim(&h.app).await.is_none(), "apply waits for a person, so no runner gets it");
    assert!(heard(oxygenhub::events::topic::run(run.id)).contains(&"jobs".to_string()), "the run's topic hears its jobs");
    // Those who may approve are told, once; the pusher (others-only) and
    // someone without the role are not.
    let told = |id: i64| {
        let db = h.app.db.clone();
        async move { db::notify::inbox(&db, id, 20).await.unwrap().into_iter().filter(|n| n.kind == "flow.approval").count() }
    };
    assert_eq!((told(carol.id).await, told(andy.id).await, told(bob.id).await), (1, 0, 0));
    let (_, page) = call(&h.app, "GET", &format!("/andy/gated/flow/{}", run.number), None, None).await;
    assert!(page.contains("waits for approval by someone with the <em>maintain</em> role, other than andy"), "the run page says so");
    let own = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &andy, true).await.unwrap_err();
    assert!(own.contains("someone other than andy"), "{own}");
    let no_role = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &bob, true).await.unwrap_err();
    assert!(no_role.contains("takes the maintain role"), "{no_role}");
    assert_eq!(oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, true).await.unwrap(), "apply: approved by carol");
    assert_eq!(told(carol.id).await, 1, "told once");
    let j = claim(&h.app).await.expect("approved, it runs");
    assert_eq!(j["job"]["name"], "apply");
    // Approved, its identity token names the environment (oxygenflow#59).
    let token =
        j["secrets"].as_array().unwrap().iter().find(|s| s["path"] == "/run/secrets/cloud").unwrap()["value"].as_str().unwrap().to_string();
    let claims = h.app.oidc().await.unwrap().verify(&token, "https://sts.example", db::now_ms()).expect("a valid token");
    assert_eq!(claims["environment"], "prod");
    let (id, ph) = (j["job_id"].as_i64().unwrap(), j["pipeline_hash"].as_str().unwrap().to_string());
    finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt("apply", &ph, 0) })).await;
    let audit = db::accounts::audit_rows(&h.app.db, 50).await.unwrap();
    assert!(audit.iter().any(|r| r.action == "flow.approved" && r.actor == "carol" && r.target.ends_with("#1 apply")), "audited");
    let (_, page) = call(&h.app, "GET", &format!("/andy/gated/flow/{}", run.number), None, None).await;
    assert!(page.contains("approved by carol"), "the run says who");
    let (_, page) = call(&h.app, "GET", "/andy/gated/deployments", None, None).await;
    assert!(page.contains("<strong>prod</strong>") && page.contains(">#1</a>") && page.contains("<td>carol</td>"), "the Deployments page");
    let (_, list) = call(&h.app, "GET", "/api/v1/repos/andy/gated/deployments", Some("site"), None).await;
    let list: serde_json::Value = serde_json::from_str(&list).unwrap();
    assert_eq!((list[0]["environment"].as_str(), list[0]["approved_by"].as_str()), (Some("prod"), Some("carol")));

    // Run 2: rejected through the API's rules: the job ends cancelled, the run ends.
    push();
    let run = plan_passes(2).await;
    assert_eq!(oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, false).await.unwrap(), "apply: rejected by carol");
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let apply = jobs.iter().find(|j| j.name == "apply").unwrap();
    assert_eq!((apply.status.as_str(), apply.error.as_str()), ("cancelled", "rejected by carol"));
    assert!(claim(&h.app).await.is_none());

    // Run 3: nobody decides in time.
    push();
    let run = plan_passes(3).await;
    assert_eq!(oxygenhub::flow::expire_approvals(&h.app, db::now_ms() + 3_600_000).await.unwrap(), 0, "not yet");
    assert_eq!(oxygenhub::flow::expire_approvals(&h.app, db::now_ms() + 25 * 3_600_000).await.unwrap(), 1);
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let apply = jobs.iter().find(|j| j.name == "apply").unwrap();
    assert_eq!((apply.status.as_str(), apply.error.as_str()), ("cancelled", "no approval within 24h"));
}

#[tokio::test(flavor = "multi_thread")]
async fn live_streams_only_topics_the_viewer_may_hear_and_replays() {
    let h = hub().await;
    create(&h.app, "andy", "open").await;
    h.app.repos.create(&RepoName::parse("andy/shut").unwrap(), None).unwrap();
    db::repos::create(&h.app.db, "andy", "shut", "", hub_api::Visibility::Private).await.unwrap();
    let open = db::repos::get(&h.app.db, "andy", "open").await.unwrap().unwrap();
    let shut = db::repos::get(&h.app.db, "andy", "shut").await.unwrap().unwrap();
    let live = &h.app.events.live;
    let first = live.publish(&oxygenhub::events::topic::repo(open.id), "push");
    live.publish(&oxygenhub::events::topic::repo(shut.id), "push");
    live.publish(&oxygenhub::events::topic::proposal(open.id, 3), "review");

    // The first bytes of a stream, within a moment.
    let read = |path: String, last: Option<u64>| {
        let app = h.app.clone();
        async move {
            let mut req = Request::get(&path);
            if let Some(l) = last {
                req = req.header("last-event-id", l.to_string());
            }
            let res = oxygenhub::router(app).oneshot(req.body(Body::empty()).unwrap()).await.unwrap();
            let status = res.status();
            let mut body = res.into_body();
            let mut text = String::new();
            while let Ok(Some(Ok(frame))) = tokio::time::timeout(std::time::Duration::from_millis(300), body.frame()).await {
                if let Some(d) = frame.data_ref() {
                    text.push_str(&String::from_utf8_lossy(d));
                }
            }
            (status, text)
        }
    };
    let topics = format!("/live?topics=repo:{},repo:{}/proposal:3", open.id, open.id);
    // From the first event on: what came after it on these topics only.
    let (st, text) = read(topics.clone(), Some(first.id)).await;
    assert_eq!(st, StatusCode::OK);
    assert!(text.contains("event: change") && text.contains(&format!("\"topic\":\"repo:{}/proposal:3\"", open.id)), "{text}");
    assert!(!text.contains(&format!("\"topic\":\"repo:{}\"", shut.id)), "never another topic: {text}");
    assert!(!text.contains(&format!("id: {}\n", first.id)), "not what was already seen: {text}");
    // A private repository's topic, or someone else's, is refused whole.
    let (st, _) = read(format!("/live?topics=repo:{},repo:{}", open.id, shut.id), None).await;
    assert_eq!(st, StatusCode::FORBIDDEN);
    assert_eq!(read("/live?topics=user:1".into(), None).await.0, StatusCode::FORBIDDEN);
    assert_eq!(read("/live?topics=nonsense".into(), None).await.0, StatusCode::BAD_REQUEST);
    assert_eq!(read("/live?topics=".into(), None).await.0, StatusCode::BAD_REQUEST);
    // Too far behind: told to reload.
    for _ in 0..oxygenhub::events::REPLAY {
        live.publish("repo:999999", "push");
    }
    let (_, text) = read(topics, Some(first.id)).await;
    assert!(text.contains("event: reload"), "{text}");
}

#[tokio::test(flavor = "multi_thread")]
async fn a_sealed_approval_releases_exactly_the_reviewed_plan() {
    use oxy_core::approval::{SignedApproval, Statement};
    use oxy_core::hash::ObjectRef;
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    let carol = db::accounts::create_account(&h.app.db, "carol", "[email protected]", "x", "salt", true).await.unwrap();
    let sk = SecretKey::generate();
    let carol_id = sk.identity("Carol", "[email protected]", 1);
    db::accounts::link_identity(&h.app.db, carol.id, &carol_id.hash().to_hex(), "Carol", "[email protected]", &carol_id.encode())
        .await
        .unwrap();
    create(&h.app, "andy", "infra").await;
    let row = db::repos::get(&h.app.db, "andy", "infra").await.unwrap().unwrap();
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/infra", h.base), work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"plan\" {{\n    sandbox \"alpine\"\n    exec \"plan\"\n    output {{\n        artifact \"tfplan\" {{\n            path \"/src/out/tfplan\"\n        }}\n    }}\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n    depends-on \"plan\"\n    environment \"prod\"\n    approval\n    input {{\n        artifact \"tfplan\" from \"plan\"\n        path \"/src/tfplan\"\n    }}\n    identity \"https://sts.example\" {{\n        mount-path \"/run/secrets/cloud\"\n    }}\n    exec \"apply\"\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Plan, then apply", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "infra").await[0].clone();
    let p = claim(&h.app).await.unwrap();
    let plan_id = p["job_id"].as_i64().unwrap();
    let plan = b"the reviewed plan".to_vec();
    upload(&h.app, plan_id, "tfplan", &plan).await;
    let mut r = receipt("plan", p["pipeline_hash"].as_str().unwrap(), 0);
    r["outputs"] = serde_json::json!([oxy_core::repo::Repo::blob_ref(&plan).to_hex()]);
    finish(&h.app, plan_id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": r })).await;
    let run = db::flow::run(&h.app.db, row.id, run.number).await.unwrap().unwrap();

    // An environment's gate takes a sealed approval: pressing Approve is not enough.
    let plain = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, true).await.unwrap_err();
    assert!(plain.contains(&format!("oxy seal --flow {}/andy/infra/flow/{} --job apply", h.base, run.number)), "{plain}");
    // What there is to sign.
    let (st, body) =
        call(&h.app, "GET", &format!("/api/v1/repos/andy/infra/flow/runs/{}/jobs/apply/approval", run.number), Some("site"), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["environment"], "prod");
    assert_eq!(v["inputs"], serde_json::json!([oxy_core::repo::Repo::blob_ref(&plan).to_hex()]), "the reviewed plan");
    let mut repo = [0u8; 16];
    repo.copy_from_slice(&hex::decode(v["repo"].as_str().unwrap()).unwrap());
    let oref = |k: &str| ObjectRef::from_hex(v[k].as_str().unwrap()).unwrap();
    let statement = Statement {
        repo,
        run: v["run"].as_str().unwrap().into(),
        job: "apply".into(),
        job_hash: oref("job_hash"),
        state: oref("state"),
        inputs: v["inputs"].as_array().unwrap().iter().map(|i| ObjectRef::from_hex(i.as_str().unwrap()).unwrap()).collect(),
        environment: "prod".into(),
        approve: true,
    };
    let seal = |s: &Statement, key: &SecretKey, id: &oxy_core::identity::Identity| {
        let body = serde_json::json!({ "signed": hex::encode(SignedApproval::sign(s, id, key).encode()) }).to_string();
        let app = h.app.clone();
        let path = format!("/api/v1/repos/andy/infra/flow/runs/{}/jobs/apply/seal", run.number);
        async move { call(&app, "POST", &path, Some("site"), Some(body)).await }
    };
    // Another plan, or a key linked to nobody (an agent's), is refused.
    let other_plan = Statement { inputs: vec![ObjectRef([9; 32])], ..statement.clone() };
    let (st, body) = seal(&other_plan, &sk, &carol_id).await;
    assert!(st == StatusCode::CONFLICT && body.contains("approves something else"), "{st} {body}");
    let agent = SecretKey::generate();
    let agent_id = agent.identity("Claude Code", "[email protected]", 1);
    let (st, body) = seal(&statement, &agent, &agent_id).await;
    assert!(st == StatusCode::CONFLICT && body.contains("not linked to an account"), "{st} {body}");
    assert!(claim(&h.app).await.is_none(), "still waiting");

    // Carol's seal releases it; it is kept in the repository; the token names her and it.
    let (st, body) = seal(&statement, &sk, &carol_id).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert!(body.contains("approved by carol, sealed"), "{body}");
    let name = oxygenhub::repos::RepoName::parse("andy/infra").unwrap();
    let state = statement.state;
    let kept = h.app.repos.read(&name, move |r, _| Ok(r.store().approvals_for(&state)?)).await.unwrap();
    assert_eq!(kept.len(), 1, "the approval is in the repository");
    let j = claim(&h.app).await.expect("released");
    assert_eq!(j["job"]["name"], "apply");
    let token =
        j["secrets"].as_array().unwrap().iter().find(|s| s["path"] == "/run/secrets/cloud").unwrap()["value"].as_str().unwrap().to_string();
    let claims = h.app.oidc().await.unwrap().verify(&token, "https://sts.example", db::now_ms()).unwrap();
    assert_eq!((claims["environment"].as_str(), claims["approved_by"].as_str()), (Some("prod"), Some("carol")));
    assert_eq!(claims["approval"].as_str(), Some(kept[0].to_hex().as_str()));
    let (_, page) = call(&h.app, "GET", &format!("/andy/infra/flow/{}", run.number), None, None).await;
    assert!(page.contains("sealed with their key"), "the run page says so");
}

/// Repositories take turns (oxygenhub#244): a repository with a job running
/// waits behind one with none, so a burst can't hold up the rest.
#[tokio::test(flavor = "multi_thread")]
async fn repositories_take_turns_on_the_runners() {
    let h = hub().await;
    for name in ["busy", "quiet"] {
        create(&h.app, "andy", name).await;
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/{name}", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.snapshot("test").unwrap();
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        assert_eq!(runs_of(&h.app, "andy", name).await.len(), 1);
    }
    // busy pushed first: its first job goes first; then quiet's turn, though
    // busy's other independent job is older; then busy's again.
    let order: Vec<String> = {
        let mut out = Vec::new();
        for _ in 0..3 {
            let c = claim(&h.app).await.expect("a job to claim");
            out.push(c["repo"].as_str().unwrap().to_string());
        }
        out
    };
    assert_eq!(order, ["andy/busy", "andy/quiet", "andy/busy"]);
}

/// Priority comes before taking turns (oxygenhub#244): a low repository's
/// jobs wait for a normal one's, and an expedited run goes first of all.
#[tokio::test(flavor = "multi_thread")]
async fn priority_then_turns_and_an_expedited_run_first() {
    let h = hub().await;
    for name in ["early", "later"] {
        create(&h.app, "andy", name).await;
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/{name}", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.snapshot("test").unwrap();
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        assert_eq!(runs_of(&h.app, "andy", name).await.len(), 1);
    }
    let early = db::repos::get(&h.app.db, "andy", "early").await.unwrap().unwrap();
    db::repos::set_flow_priority(&h.app.db, early.id, "low").await.unwrap();
    // early pushed first, but it is low: later's jobs go first.
    let repo_of = |c: serde_json::Value| c["repo"].as_str().unwrap().to_string();
    assert_eq!(repo_of(claim(&h.app).await.unwrap()), "andy/later");
    assert_eq!(repo_of(claim(&h.app).await.unwrap()), "andy/later");
    // Expedited, early's run goes before anything else.
    let run = &runs_of(&h.app, "andy", "early").await[0];
    assert!(db::flow::expedite(&h.app.db, run.id, "expedited by andy: an incident").await.unwrap());
    // What the cap on expedited runs counts: andy's live urgent runs.
    assert_eq!(db::flow::expedited_live(&h.app.db, "andy").await.unwrap(), 1);
    assert_eq!(db::flow::expedited_live(&h.app.db, "someone-else").await.unwrap(), 0);
    assert_eq!(repo_of(claim(&h.app).await.unwrap()), "andy/early");
    let q = db::flow::queue(&h.app.db).await.unwrap();
    assert!(
        q.iter().any(|e| e.name == "early" && e.priority() == "urgent" && e.priority_note.contains("an incident")),
        "{:?}",
        q.iter().map(|e| (&e.name, e.priority())).collect::<Vec<_>>()
    );
    // The site's level is the default for everyone else.
    db::accounts::set_setting(&h.app.db, "flow_priority", "high").await.unwrap();
    let q = db::flow::queue(&h.app.db).await.unwrap();
    assert!(q.iter().filter(|e| e.name == "later").all(|e| e.priority() == "high"));
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
