//! A runner watching a repository on the reference PDS through a stand-in
//! Jetstream: a push rings, the runner reads the refs back from the PDS,
//! runs the workflows at the new commit and files a check for each in its
//! own account.

#[path = "../../knotless/tests/common/mod.rs"]
mod common;

use common::{Sandbox, net, wait_until};
use futures::{SinkExt, StreamExt};
use knotless::keys::{ref_hash, ref_rkey};
use knotless_run::check;
use knotless_run::executor::Executor;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};

const KIND: &str = "network.bsky.jetstream.subscribeEvents#";

/// A Jetstream that sends what the test hands it, to whoever is connected.
struct Jetstream {
    url: String,
    frames: tokio::sync::mpsc::UnboundedSender<Message>,
}

// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
fn jetstream(rt: &tokio::runtime::Runtime) -> Jetstream {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let url = format!("ws://{}", listener.local_addr().unwrap());
    let (frames, rx) = tokio::sync::mpsc::unbounded_channel::<Message>();
    let rx = Arc::new(Mutex::new(rx));
    rt.spawn(async move {
        loop {
            let (tcp, _) = listener.accept().await.unwrap();
            let rx = rx.clone();
            tokio::spawn(async move {
                // Jetstream answers only a client that offers its framing.
                let offered = |req: &Request, mut res: Response| {
                    let protocol = req.headers().get("sec-websocket-protocol").cloned();
                    assert_eq!(protocol.as_ref().unwrap(), "xrpc.v1.json");
                    res.headers_mut()
                        .insert("sec-websocket-protocol", protocol.unwrap());
                    Ok(res)
                };
                let Ok(mut ws) = tokio_tungstenite::accept_hdr_async(tcp, offered).await else {
                    return;
                };
                // Reading answers the client's pings, as Jetstream does: one
                // left unanswered gives the connection up, and takes the
                // next frame with it.
                let mut rx = rx.lock().await;
                loop {
                    tokio::select! {
                        frame = rx.recv() => {
                            let Some(frame) = frame else { return };
                            if ws.send(frame).await.is_err() {
                                return;
                            }
                        }
                        read = ws.next() => {
                            if !matches!(read, Some(Ok(_))) {
                                return;
                            }
                        }
                    }
                }
            });
        }
    });
    Jetstream { url, frames }
}

impl Jetstream {
    fn ring(&self, did: &str, rkey: &str) {
        self.send(did, "dev.knotless.ref", rkey, None);
    }

    fn send(&self, did: &str, collection: &str, rkey: &str, record: Option<serde_json::Value>) {
        let mut payload = serde_json::json!({
            "$type": format!("{KIND}commit"), "seq": 1, "did": did, "time": "2026-10-21T00:00:00.000001Z",
            "rev": "3l3qo2vutsw2b", "operation": "update", "collection": collection, "rkey": rkey,
        });
        if let Some(record) = record {
            payload["record"] = record;
        }
        let frame = serde_json::json!({"$type": "message", "payload": payload});
        self.frames
            .send(Message::Text(frame.to_string().into()))
            .unwrap();
    }
}

const CI: &str = r#"
on:
  push:
    refs: [refs/heads/main]
secrets: [TOKEN, UNGRANTED, MAIN_ONLY, TAGS_ONLY]
env:
  GREETING: hello
steps:
  - name: look around
    run: |
      set -e
      echo "$GREETING from $KNOTLESS_RUN_REF at $KNOTLESS_RUN_SHA of $KNOTLESS_RUN_REPO"
      echo "token=$TOKEN ungranted=${UNGRANTED:-none} other=${OTHER:-none} password=${KNOTLESS_PASSWORD:-none}"
      echo "main=$MAIN_ONLY tags=${TAGS_ONLY:-none}"
      test "$(cat f)" = "$(git rev-parse HEAD~1)"
  - run: test -f f
"#;

/// Fails, leaving pictures and what is not kept of them as outputs.
const FAIL: &str = r#"
steps:
  - name: break
    run: |
      out=$KNOTLESS_RUN_OUTPUTS
      echo iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg== | base64 -d > "$out/page 00.png"
      for i in $(seq -w 1 32); do cp "$out/page 00.png" "$out/page $i.png"; done
      printf x >> "$out/page 31.png"
      head -c 1000001 /dev/zero > "$out/big.png"
      echo notes > "$out/notes.txt"
      mkdir "$out/nested"
      echo broken; exit 4
  - name: after
    run: echo never
"#;
const TAGS: &str = "on: {push: {refs: [refs/tags/v*]}}\nsteps: [{run: 'true'}]\n";

fn runner_opts(runner: &Sandbox, owner: &Sandbox, js: &Jetstream) -> knotless_run::Options {
    let secrets = runner.path("secrets.json");
    let label = format!("{}/proj", owner.account.handle);
    std::fs::write(
        &secrets,
        serde_json::json!({ label.clone(): {
            "TOKEN": "s3cret",
            "OTHER": "nope",
            "MAIN_ONLY": {"value": "m4inonly", "refs": ["refs/heads/main"]},
            "TAGS_ONLY": {"value": "t4gsonly", "refs": ["refs/tags/*"]},
        } })
        .to_string(),
    )
    .unwrap();
    knotless_run::Options {
        jetstream: js.url.clone(),
        repos: vec![label],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: Some(secrets),
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        step_timeout: Duration::from_secs(60),
        run_timeout: None,
        jobs: 4,
        pull_jobs: 4,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: Some(runner.path("caches")),
        sweep: Duration::from_secs(600),
        stop_wait: Duration::from_secs(60),
        pulls: false,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: Some(runner.path("paused.json")),
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    }
}

fn check_of(runner: &Sandbox, owner: &Sandbox, key: &str, workflow: &str) -> serde_json::Value {
    check_on(runner, owner, key, "refs/heads/main", workflow)
}

fn blob(did: &str, cid: &str) -> String {
    reqwest::blocking::get(format!(
        "{}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}",
        net().pds
    ))
    .unwrap()
    .text()
    .unwrap()
}

#[test]
fn a_push_runs_its_workflows() {
    let owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);

    let work = owner.work("proj", 1);
    std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
    for (file, text) in [
        ("ci.yml", CI),
        ("fail.yaml", FAIL),
        ("tags.yml", TAGS),
        ("bad.yml", "stpes: []\n"),
        ("notes.txt", "not a workflow"),
    ] {
        std::fs::write(work.join(".knotless/workflows").join(file), text).unwrap();
    }
    // Those naming runners run only on those, by handle or DID.
    let did = runner.account.did.clone();
    for (file, runners) in [
        ("by-handle.yml", format!("[nobody.invalid, {handle}]")),
        ("by-did.yml", format!("['{did}']")),
        (
            "elsewhere.yml",
            "['did:plc:elsewhere', nobody.invalid]".to_string(),
        ),
    ] {
        let text = format!(
            "on: {{push: {{refs: [refs/heads/main]}}}}\nrunners: {runners}\nsteps: [{{run: 'true'}}]\n"
        );
        std::fs::write(work.join(".knotless/workflows").join(file), text).unwrap();
    }
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let key = owner.record("dev.knotless.repo", "proj")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let ref_key = ref_rkey(&key, "refs/heads/main");

    let rt = tokio::runtime::Runtime::new().unwrap();
    let js = jetstream(&rt);
    let opts = knotless_run::Options {
        allowance: Some(1),
        unmetered: vec![owner.account.did.clone()],
        trust: Vec::new(),
        ..runner_opts(&runner, &owner, &js)
    };
    rt.spawn(knotless_run::run(opts.clone(), common::ctx(&runner)));

    // Watching starts where the refs are: nothing runs for what was there.
    wait_until(|| runner.path("state.json").exists());
    // It made a key, kept beside its state, and published its public half
    // for owners to seal secrets to.
    let keys = knotless::sealed::RunnerKeys::open_or_make(&runner.path("secrets-key.json"));
    let sealing = keys.unwrap().public().to_did_key();
    let published = runner.record(knotless::sealed::KEY_NSID, knotless::sealed::KEY_RKEY);
    assert_eq!(published["key"], sealing.as_str(), "{published}");
    js.ring(&owner.account.did, &ref_key);
    std::thread::sleep(Duration::from_millis(500));
    assert!(check_of(&runner, &owner, &key, "ci").is_null());

    // A push, and the ring that follows it, run the workflows at its commit.
    let sha = owner.commit_file(&work, "f", "");
    let sha = {
        std::fs::write(work.join("f"), &sha).unwrap();
        owner.git(&work, &["commit", "-q", "-am", "f names its parent"]);
        owner.git(&work, &["rev-parse", "HEAD"])
    };
    owner.git(&work, &["push", "-q", "origin", "main"]);
    js.ring(&owner.account.did, &ref_key);
    // Side by side, each finishing in its own time.
    wait_until(|| {
        ["ci", "fail"]
            .iter()
            .all(|wf| check_of(&runner, &owner, &key, wf)["finishedAt"].is_string())
    });

    let ci = check_of(&runner, &owner, &key, "ci");
    assert_eq!(ci["status"], "success", "{ci}");
    assert_eq!(ci["commit"], sha.as_str());
    wait_until(|| {
        ["by-handle", "by-did"]
            .iter()
            .all(|wf| check_of(&runner, &owner, &key, wf)["status"] == "success")
    });
    assert!(check_of(&runner, &owner, &key, "elsewhere").is_null());
    // The owner's minutes are filed in the runner's account, under its DID.
    let minutes = runner.record("dev.knotless.minutes", &owner.account.did);
    assert_eq!(minutes["owner"], owner.account.did.as_str(), "{minutes}");
    assert!(minutes["minutes"].is_u64(), "{minutes}");
    assert!(minutes["updatedAt"].is_string(), "{minutes}");
    assert_eq!(ci["ref"], "refs/heads/main");
    assert_eq!(ci["repo"]["key"], key.as_str());
    assert_eq!(ci["steps"][0]["name"], "look around");
    assert_eq!(ci["steps"][1]["name"], "step 2");
    assert_eq!(ci["steps"][1]["exitCode"], 0);
    let log = blob(
        &runner.account.did,
        ci["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.contains(&format!(
            "hello from refs/heads/main at {sha} of {}/proj",
            owner.account.handle
        )),
        "{log}"
    );
    // Only what the workflow asks for and the runner grants, and never the
    // runner's own credentials; a granted secret is masked in the log.
    assert!(
        log.contains("token=*** ungranted=none other=none password=none"),
        "{log}"
    );
    assert!(!log.contains("s3cret"), "{log}");
    // A secret granted to some refs alone reaches only their pushes.
    assert!(log.contains("main=*** tags=none"), "{log}");
    // The log says which it asked for and was not given.
    let label = format!("{}/proj", owner.account.handle);
    for name in ["UNGRANTED", "TAGS_ONLY"] {
        let said = format!("==> {name} is not granted to {label} at refs/heads/main\n");
        assert!(log.contains(&said), "{log}");
    }
    assert!(!log.contains("TOKEN is not granted"), "{log}");

    let fail = check_of(&runner, &owner, &key, "fail");
    assert_eq!(fail["status"], "failure");
    assert_eq!(fail["steps"][0]["exitCode"], 4);
    assert_eq!(fail["steps"][1]["status"], "skipped");
    // What it left to be looked at, failing: the first 32 pictures, each
    // named for its file; the rest of what it left, in its log, and why.
    let outputs = fail["outputs"].as_array().unwrap();
    assert_eq!(outputs.len(), 32);
    assert_eq!(outputs[0]["$type"], "dev.knotless.check#image");
    assert_eq!(outputs[0]["alt"], "page 00");
    assert_eq!(outputs[31]["alt"], "page 31");
    assert_eq!(outputs[0]["image"]["mimeType"], "image/png");
    // Files alike share one blob, uploaded once; each keeps its place.
    let cid = |i: usize| outputs[i]["image"]["ref"]["$link"].as_str().unwrap();
    assert_eq!(cid(0), cid(30));
    assert_ne!(cid(30), cid(31));
    let log = blob(
        &runner.account.did,
        fail["log"]["ref"]["$link"].as_str().unwrap(),
    );
    for left in [
        "big.png: not kept, being over 1000000 bytes",
        "nested: not kept, as only images are outputs",
        "notes.txt: not kept, as only images are outputs",
        "page 32.png: not kept, past 32 outputs",
    ] {
        assert!(log.contains(&format!("==> {left}\n")), "{log}");
    }
    let bad = check_of(&runner, &owner, &key, "bad");
    assert_eq!(bad["status"], "error");
    assert!(
        bad["error"]
            .as_str()
            .unwrap()
            .contains("unknown field `stpes`"),
        "{bad}"
    );
    assert!(check_of(&runner, &owner, &key, "tags").is_null());

    // Named as a runner, what it wrote is what a reader finds.
    let me = runner.account.handle.clone();
    assert_eq!(
        owner
            .cli(&["runners", &owner.url("proj"), "add", &me], b"")
            .0,
        0
    );
    let (code, out) = owner.cli(
        &["checks", &owner.url("proj"), "--ref", "refs/heads/main"],
        b"",
    );
    assert_eq!(code, 0, "{out}");
    let lines: Vec<&str> = out.lines().collect();
    assert_eq!(lines.len(), 5, "{out}");
    // And the minutes it counts for the owner, with no allowance for an
    // owner unmetered.
    let (code, out) = owner.cli(&["minutes", &owner.url("proj")], b"");
    assert_eq!(code, 0, "{out}");
    assert!(out.starts_with(&format!("{me}\t")), "{out}");
    assert!(out.ends_with(" min in 30 days, no allowance"), "{out}");
    // Its pictures, saved to open, each named for its workflow and what it
    // shows: after the listing, or beside the JSON.
    let saved = owner.path("pictures");
    let dir = saved.to_str().unwrap();
    let (code, out) = owner.cli(
        &[
            "checks",
            &owner.url("proj"),
            "--workflow",
            "fail",
            "--images",
            dir,
        ],
        b"",
    );
    assert_eq!(code, 0, "{out}");
    let shown: Vec<&str> = out.lines().filter(|l| l.starts_with("image\t")).collect();
    assert_eq!(shown.len(), 32, "{out}");
    assert_eq!(shown[0], format!("image\t{dir}/fail page 00.png"));
    let first = std::fs::read(saved.join("fail page 00.png")).unwrap();
    assert!(first.starts_with(b"\x89PNG"));
    let (code, out) = owner.cli(
        &[
            "checks",
            &owner.url("proj"),
            "--workflow",
            "fail",
            "--images",
            dir,
            "--json",
        ],
        b"",
    );
    assert_eq!(code, 0, "{out}");
    assert!(
        serde_json::from_str::<serde_json::Value>(&out).is_ok(),
        "{out}"
    );
    // With their logs, each after its line.
    let (_, out) = owner.cli(
        &[
            "checks",
            &owner.url("proj"),
            "--ref",
            "refs/heads/main",
            "--log",
        ],
        b"",
    );
    assert!(out.contains("==> break\nbroken\n"), "{out}");
    assert!(
        lines
            .iter()
            .any(|l| l.starts_with("success\tci\trefs/heads/main\t")),
        "{out}"
    );

    // A ring for a ref that has not moved runs nothing again.
    let started = ci["startedAt"].clone();
    js.ring(&owner.account.did, &ref_key);
    std::thread::sleep(Duration::from_millis(500));
    assert_eq!(check_of(&runner, &owner, &key, "ci")["startedAt"], started);

    // Unless its owner asks: requests that are not one are passed over.
    let junk = serde_json::json!({"$type": "dev.knotless.rerun"});
    owner.put_record("dev.knotless.rerun", &format!("{key}.zzzz"), junk.clone());
    owner.put_record(
        "dev.knotless.rerun",
        &format!("{key}.{}", knotless::keys::tid()),
        junk,
    );
    let (code, asked) = owner.cli(
        &["rerun", &owner.url("proj"), "--ref", "refs/heads/main"],
        b"",
    );
    assert_eq!(code, 0, "{asked}");
    let asked = asked.rsplit('/').next().unwrap().to_string();
    js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
    wait_until(|| {
        let ci = check_of(&runner, &owner, &key, "ci");
        ci["startedAt"] != started && ci["finishedAt"].is_string()
    });
    assert_eq!(
        check_of(&runner, &owner, &key, "ci")["commit"],
        sha.as_str()
    );
    // Once.
    let started = check_of(&runner, &owner, &key, "ci")["startedAt"].clone();
    js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
    std::thread::sleep(Duration::from_millis(500));
    assert_eq!(check_of(&runner, &owner, &key, "ci")["startedAt"], started);

    // A push while the runner is down runs when it is back, rung or not.
    drop(rt);
    let next = owner.commit_file(&work, "f", "elsewhere\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let rt = tokio::runtime::Runtime::new().unwrap();
    let js = jetstream(&rt);
    let opts = knotless_run::Options {
        jetstream: js.url.clone(),
        ..opts
    };
    rt.spawn(knotless_run::run(opts.clone(), common::ctx(&runner)));
    wait_until(|| check_of(&runner, &owner, &key, "ci")["commit"] == next.as_str());
    wait_until(|| check_of(&runner, &owner, &key, "ci")["finishedAt"].is_string());
    // The checkout is that commit, whose f no longer names its parent.
    let again = check_of(&runner, &owner, &key, "ci");
    assert_eq!(again["status"], "failure", "{again}");
    // The commit before keeps its own check.
    let before = check::rkey(
        &owner.account.did,
        &key,
        "refs/heads/main",
        &sha,
        None,
        "ci",
    );
    let before = runner.record(check::NSID, &before);
    assert_eq!(before["commit"], sha.as_str());
    assert_eq!(before["status"], "success", "{before}");

    // A newer commit's run of a workflow on the ref cancels the older
    // one's still going: its step is killed, the rest skipped.
    let long = "steps:\n  - name: wait\n    run: test -f quick || sleep 30\n  - run: 'true'\n";
    std::fs::write(work.join(".knotless/workflows/long.yml"), long).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "long"]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let slow = owner.git(&work, &["rev-parse", "HEAD"]);
    js.ring(&owner.account.did, &ref_key);
    let slow_key = check::rkey(
        &owner.account.did,
        &key,
        "refs/heads/main",
        &slow,
        None,
        "long",
    );
    wait_until(|| runner.record(check::NSID, &slow_key)["status"] == "running");
    std::thread::sleep(Duration::from_millis(500));
    let quick = owner.commit_file(&work, "quick", "");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    js.ring(&owner.account.did, &ref_key);
    let quick_key = check::rkey(
        &owner.account.did,
        &key,
        "refs/heads/main",
        &quick,
        None,
        "long",
    );
    wait_until(|| runner.record(check::NSID, &quick_key)["status"] == "success");
    wait_until(|| runner.record(check::NSID, &slow_key)["finishedAt"].is_string());
    let cancelled = runner.record(check::NSID, &slow_key);
    assert_eq!(cancelled["status"], "cancelled", "{cancelled}");
    assert_eq!(cancelled["steps"][0]["status"], "cancelled");
    assert_eq!(cancelled["steps"][1]["status"], "skipped");
    let log = blob(
        &runner.account.did,
        cancelled["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.ends_with("==> cancelled for a newer commit's run\n"),
        "{log}"
    );

    // One whose runs finish keeps going, and the newer commit's waits its
    // turn; one waiting is cancelled for a newer still, so one waits.
    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let finish = format!(
        "superseded: finish\nsteps:\n  - name: hold\n    run: until [ -f {}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n",
        go.display()
    );
    std::fs::write(work.join(".knotless/workflows/finish.yml"), finish).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "finish"]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let first = owner.git(&work, &["rev-parse", "HEAD"]);
    js.ring(&owner.account.did, &ref_key);
    let at = |sha: &str| {
        let rkey = check::rkey(
            &owner.account.did,
            &key,
            "refs/heads/main",
            sha,
            None,
            "finish",
        );
        runner.record(check::NSID, &rkey)
    };
    wait_until(|| at(&first)["status"] == "running");
    std::thread::sleep(Duration::from_millis(500));
    let second = owner.commit_file(&work, "f", "second\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    js.ring(&owner.account.did, &ref_key);
    // The second, waiting for the first to end, is filed queued.
    wait_until(|| at(&second)["status"] == "queued");
    std::thread::sleep(Duration::from_millis(500));
    assert!(at(&first)["finishedAt"].is_null(), "{}", at(&first));
    let third = owner.commit_file(&work, "f", "third\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    js.ring(&owner.account.did, &ref_key);
    wait_until(|| at(&second)["finishedAt"].is_string());
    let never = at(&second);
    assert_eq!(never["status"], "cancelled", "{never}");
    assert!(never.get("steps").is_none(), "{never}");
    let log = blob(
        &runner.account.did,
        never["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert_eq!(
        log,
        "==> cancelled for a newer commit's run before it started\n"
    );
    // The third waits for the first, even free to end at once.
    std::fs::write(go.join(&third), "").unwrap();
    std::thread::sleep(Duration::from_millis(500));
    assert_eq!(at(&third)["status"], "queued", "{}", at(&third));
    assert!(at(&first)["finishedAt"].is_null(), "{}", at(&first));
    std::fs::write(go.join(&first), "").unwrap();
    wait_until(|| at(&third)["finishedAt"].is_string());
    let (first, third) = (at(&first), at(&third));
    assert_eq!(first["status"], "success", "{first}");
    assert_eq!(third["status"], "success", "{third}");
    assert!(
        first["finishedAt"].as_str() <= third["finishedAt"].as_str(),
        "{first} {third}"
    );

    // Runs of a workflow with a cache, on two branches, go side by side,
    // each over its own layers, which go when they end. Each writes to
    // every run's upper layer going, as the shell executor mounts none.
    std::fs::remove_file(work.join(".knotless/workflows/finish.yml")).unwrap();
    let caches = runner.path("caches");
    let hold = format!(
        "cache: [target]\nsteps:\n  - name: hold\n    run: |\n      b=${{KNOTLESS_RUN_REF##*/}}\n      for u in {caches}/runs/*/*/upper; do echo $b > $u/$b; done\n      touch {go}/started-vol-$b\n      until [ -f {go}/vol-$b ]; do sleep 0.1; done\n",
        go = go.display(),
        caches = caches.display()
    );
    std::fs::write(work.join(".knotless/workflows/vol.yml"), hold).unwrap();
    owner.git(&work, &["add", "-A"]);
    owner.git(&work, &["commit", "-q", "-m", "volumes"]);
    owner.git(
        &work,
        &[
            "push",
            "-q",
            "origin",
            "HEAD:refs/heads/a",
            "HEAD:refs/heads/b",
        ],
    );
    js.ring(&owner.account.did, &ref_rkey(&key, "refs/heads/a"));
    let started = || {
        ["a", "b"]
            .into_iter()
            .filter(|b| go.join(format!("started-vol-{b}")).exists())
            .count()
    };
    wait_until(|| started() == 2);
    let runs = caches.join("runs");
    assert_eq!(std::fs::read_dir(&runs).unwrap().count(), 2);
    for name in ["vol-a", "vol-b", "vol-main"] {
        std::fs::write(go.join(name), "").unwrap();
    }
    let vol_done = |b: &str| {
        let r = format!("refs/heads/{b}");
        wait_until(|| check_on(&runner, &owner, &key, &r, "vol")["finishedAt"].is_string());
        let done = check_on(&runner, &owner, &key, &r, "vol");
        assert_eq!(done["status"], "success", "{done}");
    };
    vol_done("a");
    vol_done("b");
    wait_until(|| std::fs::read_dir(&runs).unwrap().count() == 0);
    // What they wrote went with them: the base they started from is as
    // empty as it was.
    let base = || {
        let cache = std::fs::read_dir(caches.join("bases")).unwrap();
        let cache = cache.map(|c| c.unwrap().path()).next().unwrap();
        let newest = std::fs::read_dir(cache).unwrap().map(|g| g.unwrap().path());
        let newest = newest.max().unwrap();
        let mut held: Vec<String> = std::fs::read_dir(&newest)
            .unwrap()
            .map(|f| f.unwrap().file_name().to_string_lossy().into_owned())
            .collect();
        held.sort();
        held
    };
    assert!(base().is_empty(), "{:?}", base());
    // A push to main that passes keeps what it wrote, for the runs after.
    owner.git(&work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
    js.ring(&owner.account.did, &ref_key);
    vol_done("main");
    wait_until(|| std::fs::read_dir(&runs).unwrap().count() == 0);
    assert_eq!(base(), ["main"]);

    // An owner its operator pauses has its runs filed as errors, saying
    // why, until they are taken off the list.
    let paused = runner.path("paused.json");
    let list = serde_json::json!({ owner.account.did.as_str(): "mining, by the looks of it" });
    std::fs::write(&paused, list.to_string()).unwrap();
    let sha = owner.commit_file(&work, "f", "paused\n");
    owner.git(&work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
    js.ring(&owner.account.did, &ref_key);
    wait_until(|| check_of(&runner, &owner, &key, "ci")["commit"] == sha.as_str());
    let ci = check_of(&runner, &owner, &key, "ci");
    assert_eq!(ci["status"], "error", "{ci}");
    let why = "paused by this runner's operator: mining, by the looks of it";
    assert_eq!(ci["error"], why, "{ci}");
    std::fs::remove_file(&paused).unwrap();

    // A repository whose name now holds another one is not run: the one
    // it was is gone, and its checks with it.
    let mut repo = owner.record("dev.knotless.repo", "proj");
    repo["key"] = "3l3qo2vuowo2z".into();
    owner.put_record("dev.knotless.repo", "proj", repo);
    owner.commit_file(&work, "f", "again\n");
    js.ring(&owner.account.did, &ref_key);
    wait_until(|| check_of(&runner, &owner, &key, "ci").is_null());
    std::thread::sleep(Duration::from_millis(500));
    assert!(check_of(&runner, &owner, "3l3qo2vuowo2z", "ci").is_null());

    // One it cannot find beside one it can is passed over.
    let nope = format!("{}/nope", owner.account.handle);
    let both = knotless_run::Options {
        repos: vec![nope, opts.repos[0].clone()],
        ..opts.clone()
    };
    let started = knotless_run::Runner::start(both, common::ctx(&runner));
    let (_, watches) = rt.block_on(started).unwrap();
    assert_eq!(watches.len(), 1);
    assert_eq!(watches[0].label, opts.repos[0]);

    // Alone, it stops it before it watches anything.
    let missing = knotless_run::Options {
        repos: vec![format!("{}/nope", owner.account.handle)],
        ..opts
    };
    let err = rt
        .block_on(knotless_run::run(missing, common::ctx(&runner)))
        .unwrap_err();
    assert!(
        err.to_string()
            .ends_with("/nope: no repository by that name this runner can read"),
        "{err}"
    );
}

/// The newest of `runner`'s checks of `wf` on `r`, of any commit, for the
/// repository keyed `key` of `owner`'s: null if there is none.
fn check_on(runner: &Sandbox, owner: &Sandbox, key: &str, r: &str, wf: &str) -> serde_json::Value {
    let prefix = knotless::keys::check_prefix(&owner.account.did, key);
    let (of, named) = (format!("{prefix}.{}.", ref_hash(r)), format!(".{wf}"));
    runner
        .records(check::NSID)
        .into_iter()
        .filter(|(k, _)| k.starts_with(&of) && k.ends_with(&named))
        .map(|(_, v)| v)
        .max_by(|a, b| a["startedAt"].as_str().cmp(&b["startedAt"].as_str()))
        .unwrap_or_default()
}

#[test]
fn what_goes_wrong_is_reported() {
    let owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let work = owner.work("tools", 1);
    owner.git(&work, &["remote", "add", "origin", &owner.url("tools")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let label = format!("{}/tools", owner.account.handle);
    std::fs::write(runner.path("secrets.json"), "{ not json").unwrap();
    let opts = knotless_run::Options {
        jetstream: "ws://127.0.0.1:9".into(),
        repos: vec![label],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: Some(runner.path("secrets.json")),
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: None,
        step_timeout: Duration::from_secs(1),
        // Past the one step that is out of its own time first, and short
        // of the two that run out of the run's.
        run_timeout: Some(Duration::from_millis(1200)),
        jobs: 2,
        pull_jobs: 2,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: None,
        sweep: Duration::from_secs(600),
        stop_wait: Duration::from_secs(60),
        pulls: false,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    };
    let rt = tokio::runtime::Runtime::new().unwrap();
    let (run, watches) = rt
        .block_on(knotless_run::Runner::start(
            opts.clone(),
            common::ctx(&runner),
        ))
        .unwrap();
    let watch = &watches[0];
    rt.block_on(run.reconcile(watch)).unwrap();

    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    for (file, text) in [
        (
            "slow.yml",
            "steps: [{run: 'echo going; sleep 30'}, {run: 'true'}]\n",
        ),
        (
            "long.yml",
            "steps: [{run: 'sleep 0.5'}, {run: 'echo late; sleep 30'}, {run: 'true'}]\n",
        ),
        ("needs.yml", "secrets: [X]\nsteps: [{run: 'true'}]\n"),
        ("twin-a.yml", "name: twin\nsteps: []\n"),
        ("twin-b.yml", "name: twin\nsteps: []\n"),
        ("bad name!.yml", "{"),
    ] {
        std::fs::write(dir.join(file), text).unwrap();
    }
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    rt.block_on(run.reconcile(watch)).unwrap();
    rt.block_on(run.idle());
    let main = "refs/heads/main";
    let key = watch.key.clone();

    // Out of time: killed, with no exit code, and said so in the log.
    let slow = check_on(&runner, &owner, &key, main, "slow");
    assert_eq!(slow["status"], "failure", "{slow}");
    assert!(slow["steps"][0].get("exitCode").is_none(), "{slow}");
    assert_eq!(slow["steps"][1]["status"], "skipped");
    let log = blob(
        &runner.account.did,
        slow["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(log.ends_with("going\n==> killed after 1s\n"), "{log}");
    // Out of the run's time, each step in its own: the step going is
    // killed, and the rest skipped.
    let long = check_on(&runner, &owner, &key, main, "long");
    assert_eq!(long["status"], "failure", "{long}");
    assert_eq!(long["steps"][0]["status"], "success", "{long}");
    assert!(long["steps"][1].get("exitCode").is_none(), "{long}");
    assert_eq!(long["steps"][2]["status"], "skipped");
    let log = blob(
        &runner.account.did,
        long["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.ends_with("late\n==> killed as the run's 1s were up\n"),
        "{log}"
    );
    // A workflow that asks for secrets the runner cannot read does not run.
    let needs = check_on(&runner, &owner, &key, main, "needs");
    assert_eq!(needs["status"], "error");
    assert!(
        needs["error"]
            .as_str()
            .unwrap()
            .starts_with("reading the runner's secrets"),
        "{needs}"
    );
    let twin = check_on(&runner, &owner, &key, main, "twin");
    assert_eq!(
        twin["error"],
        "more than one workflow is named twin, twin-b.yml among them"
    );
    let invalid = check_on(&runner, &owner, &key, main, "invalid");
    assert!(
        invalid["error"]
            .as_str()
            .unwrap()
            .starts_with("bad name!.yml: "),
        "{invalid}"
    );

    // A branch without the directory has no workflows to run.
    owner.git(&work, &["checkout", "-q", "--orphan", "bare"]);
    owner.git(&work, &["rm", "-rqf", "."]);
    owner.commit_file(&work, "g", "g\n");
    owner.git(&work, &["push", "-q", "origin", "bare"]);
    rt.block_on(run.reconcile(watch)).unwrap();
    assert!(check_on(&runner, &owner, &key, "refs/heads/bare", "slow").is_null());

    // A name that comes to hold another repository, the one it held kept
    // under another name, is no longer watched by that name, its checks
    // kept.
    let renamed = ["rename", &owner.url("tools"), "elsewhere"];
    assert_eq!(owner.cli(&renamed, b"").0, 0);
    let mut repo = owner.record("dev.knotless.repo", "elsewhere");
    repo["key"] = "3l3qo2vuowo2z".into();
    owner.put_record("dev.knotless.repo", "tools", repo);
    let err = rt.block_on(run.reconcile(watch)).unwrap_err();
    assert!(
        err.to_string()
            .ends_with("/tools no longer names the repository it did"),
        "{err}"
    );
    assert!(!check_on(&runner, &owner, &key, main, "slow").is_null());

    // The one it held gone, the checks of it go with it.
    owner.delete_record("dev.knotless.repo", "elsewhere");
    rt.block_on(run.reconcile(watch)).unwrap();
    let of_it = knotless::keys::check_prefix(&owner.account.did, &key);
    let filed = runner.records("dev.knotless.check");
    assert!(filed.iter().all(|(rkey, _)| !rkey.starts_with(&of_it)));

    // A check it left running, as a runner stopped mid-run leaves one, is
    // ended when it starts again; one it finished is left.
    let prefix = knotless::keys::check_prefix(&owner.account.did, "3l3qo2vuowo2z");
    let mut stale = slow.clone();
    // Its log went with the checks it was of.
    let fields = stale.as_object_mut().unwrap();
    fields.retain(|_, v| v["$type"] != "blob");
    stale["repo"]["key"] = "3l3qo2vuowo2z".into();
    stale["status"] = "running".into();
    runner.put_record(
        "dev.knotless.check",
        &format!("{prefix}.a.b.stale"),
        stale.clone(),
    );
    stale["status"] = "queued".into();
    runner.put_record(
        "dev.knotless.check",
        &format!("{prefix}.a.b.waiting"),
        stale.clone(),
    );
    stale["status"] = "success".into();
    runner.put_record("dev.knotless.check", &format!("{prefix}.a.b.done"), stale);

    // A Jetstream it cannot reach: it reads the repositories anyway, and
    // tries again later.
    rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
    let ended = || runner.record("dev.knotless.check", &format!("{prefix}.a.b.stale"));
    wait_until(|| ended()["status"] == "error");
    assert_eq!(
        ended()["error"],
        "the runner stopped before this run finished"
    );
    let waiting = runner.record("dev.knotless.check", &format!("{prefix}.a.b.waiting"));
    assert_eq!(waiting["status"], "error");
    assert_eq!(
        waiting["error"],
        "the runner stopped before this run started"
    );
    let done = runner.record("dev.knotless.check", &format!("{prefix}.a.b.done"));
    assert_eq!(done["status"], "success");
}

/// A run of a workflow whose runs finish is never cut off by a newer
/// commit's, even while it waits for a job: with the one job held, the
/// first deploy has its turn on main and waits; the second waits behind it
/// and is cancelled for the third, which runs once the first has finished.
#[test]
fn a_deploy_is_never_cut_off() {
    let owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let work = owner.work("ship", 1);
    owner.git(&work, &["remote", "add", "origin", &owner.url("ship")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let opts = knotless_run::Options {
        jetstream: "ws://127.0.0.1:9".into(),
        repos: vec![format!("{}/ship", owner.account.handle)],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: None,
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        step_timeout: Duration::from_secs(60),
        run_timeout: None,
        jobs: 1,
        pull_jobs: 1,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: Some(runner.path("caches")),
        sweep: Duration::from_secs(600),
        stop_wait: Duration::from_secs(60),
        pulls: false,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    };
    let rt = tokio::runtime::Runtime::new().unwrap();
    let (run, watches) = rt
        .block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
        .unwrap();
    let watch = &watches[0];
    let key = watch.key.clone();
    rt.block_on(run.reconcile(watch)).unwrap();

    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    let block = format!(
        "on: {{push: {{refs: [refs/heads/block]}}}}\nsteps:\n  - run: |\n      touch {go}/started-block\n      until [ -f {go}/block ]; do sleep 0.1; done\n",
        go = go.display()
    );
    let deploy = format!(
        "superseded: finish\non: {{push: {{refs: [refs/heads/main]}}}}\nsteps:\n  - run: until [ -f {}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n",
        go.display()
    );
    std::fs::write(dir.join("block.yml"), block).unwrap();
    std::fs::write(dir.join("deploy.yml"), deploy).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    // The one job, held.
    owner.git(&work, &["push", "-q", "origin", "HEAD:block"]);
    rt.block_on(run.reconcile(watch)).unwrap();
    wait_until(|| go.join("started-block").exists());

    let mut shas = Vec::new();
    for n in 0..3 {
        if n > 0 {
            owner.commit_file(&work, "f", &format!("{n}\n"));
        }
        owner.git(&work, &["push", "-q", "origin", "HEAD:main"]);
        shas.push(owner.git(&work, &["rev-parse", "HEAD"]));
        rt.block_on(run.reconcile(watch)).unwrap();
    }
    let at = |sha: &str| {
        let rkey = check::rkey(
            &owner.account.did,
            &key,
            "refs/heads/main",
            sha,
            None,
            "deploy",
        );
        runner.record(check::NSID, &rkey)
    };
    let (first, second, third) = (&shas[0], &shas[1], &shas[2]);
    wait_until(|| at(second)["finishedAt"].is_string());
    let never = at(second);
    assert_eq!(never["status"], "cancelled", "{never}");
    let log = blob(
        &runner.account.did,
        never["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert_eq!(
        log,
        "==> cancelled for a newer commit's run before it started\n"
    );
    // The first, still waiting for the job, is not: it is filed queued,
    // and filed running once it has the job, started then.
    let waiting = at(first);
    assert_eq!(waiting["status"], "queued", "{waiting}");
    std::fs::write(go.join("block"), "").unwrap();
    wait_until(|| at(first)["status"] == "running");
    let started = at(first)["startedAt"].clone();
    assert!(
        started.as_str() > waiting["startedAt"].as_str(),
        "{started}"
    );
    for name in [first.as_str(), third.as_str()] {
        std::fs::write(go.join(name), "").unwrap();
    }
    rt.block_on(run.idle());
    let (first, third) = (at(first), at(third));
    assert_eq!(first["status"], "success", "{first}");
    assert_eq!(first["startedAt"], started, "{first}");
    assert_eq!(third["status"], "success", "{third}");
    assert!(
        first["finishedAt"].as_str() <= third["finishedAt"].as_str(),
        "{first} {third}"
    );
}

/// A stop waits only for the runs whose steps have started, a deploy's
/// among them: those waiting their turn are dropped, their checks filed
/// as cancelled, not deleted, and run when the runner starts again, and
/// only they. Those
/// of a going deploy's commit are kept until it ends.
#[test]
fn a_stop_drops_what_has_not_started() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let base = appview(&rt, &Sandbox::new());
    owner.set("KNOTLESS_APPVIEW", &base);
    let mine = owner.account.handle.clone();
    owner.set("KNOTLESS_IDENTIFIER", &mine);
    runner.set("KNOTLESS_APPVIEW", &base);
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let work = owner.work("ship", 1);
    owner.git(&work, &["remote", "add", "origin", &owner.url("ship")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let opts = knotless_run::Options {
        jetstream: "ws://127.0.0.1:9".into(),
        repos: vec![format!("{}/ship", owner.account.handle)],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: None,
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        step_timeout: Duration::from_secs(60),
        run_timeout: None,
        jobs: 1,
        pull_jobs: 1,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: Some(runner.path("caches")),
        sweep: Duration::from_secs(600),
        stop_wait: Duration::from_secs(60),
        pulls: true,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    };
    let start = || {
        let started = knotless_run::Runner::start(opts.clone(), common::ctx(&runner));
        let (run, mut watches) = rt.block_on(started).unwrap();
        (run, watches.remove(0))
    };
    let (run, watch) = start();
    let key = watch.key.clone();
    rt.block_on(run.reconcile(&watch)).unwrap();

    // `block` holds the one job until `go/block` is there, so the runs
    // after it wait in line, where the first ranked goes first: the deploy,
    // which then holds the job until its commit's file is there. `then`
    // waits for `go/then`, and notes each run of it.
    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    let block = format!(
        "on: {{push: {{refs: [refs/heads/block]}}}}\nsteps:\n  - run: |\n      touch {go}/started-block\n      until [ -f {go}/block ]; do sleep 0.1; done\n",
        go = go.display()
    );
    std::fs::write(dir.join("block.yml"), block).unwrap();
    let deploy = format!(
        "superseded: finish\non: {{push: {{refs: [refs/heads/main]}}}}\nsteps:\n  - run: |\n      echo $KNOTLESS_RUN_SHA >> {go}/deploys\n      until [ -f {go}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n",
        go = go.display()
    );
    let then = format!(
        "on: {{push: {{refs: [refs/heads/main]}}, pull: {{base: [refs/heads/main]}}}}\nsteps:\n  - run: |\n      echo \"$KNOTLESS_RUN_SHA ${{KNOTLESS_RUN_PULL:-push}}\" >> {go}/thens\n      until [ -f {go}/then ]; do sleep 0.1; done\n",
        go = go.display()
    );
    std::fs::write(dir.join("deploy.yml"), deploy).unwrap();
    std::fs::write(dir.join("then.yml"), then).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["push", "-q", "origin", "HEAD:block"]);
    rt.block_on(run.reconcile(&watch)).unwrap();
    wait_until(|| go.join("started-block").exists());
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let main = owner.git(&work, &["rev-parse", "HEAD"]);
    rt.block_on(run.reconcile(&watch)).unwrap();
    // Both of main's runs are in line by the time the job is let go: a
    // run is in line once its task first runs, a moment after it is filed.
    std::thread::sleep(Duration::from_millis(500));
    std::fs::write(go.join("block"), "").unwrap();
    let ran = |name: &str| std::fs::read_to_string(go.join(name)).unwrap_or_default();
    wait_until(|| ran("deploys") == format!("{main}\n"));
    // A pull, its run waiting behind the deploy too.
    let topic = owner.commit_file(&work, "t", "t\n");
    owner.git(&work, &["push", "-q", "origin", "HEAD:topic"]);
    let (code, pull) = owner.cli(
        &[
            "pull",
            &owner.url("ship"),
            "main",
            &owner.url("ship"),
            "topic",
            "t",
        ],
        b"",
    );
    assert_eq!(code, 0, "{pull}");
    let pull = pull.trim().to_string();
    let of = |r: &str, wf: &str| check_on(&runner, &owner, &key, r, wf);
    wait_until(|| {
        rt.block_on(run.reconcile(&watch)).unwrap();
        of(&pull, "then")["status"] == "queued"
    });
    assert_eq!(of("refs/heads/main", "then")["status"], "queued");

    // Stopped, it cancels the pull's run and reads no more, but waits
    // for the deploy going, and keeps the push's run of its commit, which
    // a deploy may wait for, until the deploy ends.
    let stopping = rt.spawn({
        let run = run.clone();
        async move { run.stop().await }
    });
    wait_until(|| of(&pull, "then")["status"] == "cancelled");
    rt.block_on(run.reconcile(&watch)).unwrap();
    std::thread::sleep(Duration::from_millis(500));
    assert!(!stopping.is_finished());
    assert_eq!(of("refs/heads/main", "then")["status"], "queued");
    std::fs::write(go.join(&main), "").unwrap();
    promptly(&rt, stopping);
    assert_eq!(of("refs/heads/main", "deploy")["status"], "success");
    assert_eq!(of("refs/heads/main", "then")["status"], "cancelled");
    assert_eq!(ran("thens"), "");

    // Started again, it runs what it dropped, and not the deploy that
    // finished: one of the two takes the job. Stopped while that one goes,
    // it waits for it, an ordinary run, and drops the other again.
    let (run, watch) = start();
    rt.block_on(run.reconcile(&watch)).unwrap();
    let (push, pulled) = (format!("{main} push\n"), format!("{topic} {pull}\n"));
    wait_until(|| [&push, &pulled].contains(&&ran("thens")));
    let (first, second) = match ran("thens") == push {
        true => ("refs/heads/main", pull.as_str()),
        false => (pull.as_str(), "refs/heads/main"),
    };
    wait_until(|| of(second, "then")["status"] == "queued");
    let stopping = rt.spawn({
        let run = run.clone();
        async move { run.stop().await }
    });
    wait_until(|| of(second, "then")["status"] == "cancelled");
    assert!(!stopping.is_finished());
    std::fs::write(go.join("then"), "").unwrap();
    promptly(&rt, stopping);
    assert_eq!(of(first, "then")["status"], "success");

    // And the other on the start after.
    let (run, watch) = start();
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.block_on(run.idle());
    let ran_second = of(second, "then");
    assert_eq!(ran_second["status"], "success", "{ran_second}");
    let mut thens: Vec<String> = ran("thens").lines().map(str::to_string).collect();
    thens.sort();
    let mut both = vec![push.trim().to_string(), pulled.trim().to_string()];
    both.sort();
    assert_eq!(thens, both);
    assert_eq!(ran("deploys"), format!("{main}\n"));
}

/// A stand-in for the runner's PDS, in front of the reference one: it fails
/// the next `fail` writes, holds the next `hang` past any wait for an
/// answer, as the PDS did one night for a few minutes, and refuses the next
/// `limit` as a rate limit does, saying it lets writes in a second on.
struct Flaky {
    url: String,
    fail: Arc<AtomicUsize>,
    hang: Arc<AtomicUsize>,
    limit: Arc<AtomicUsize>,
}

fn flaky(rt: &tokio::runtime::Runtime) -> Flaky {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let (fail, hang) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
    let limit = Arc::new(AtomicUsize::new(0));
    let (failing, hanging, limiting) = (fail.clone(), hang.clone(), limit.clone());
    let pass = move |req: axum::extract::Request| {
        let (fail, hang, limit) = (failing.clone(), hanging.clone(), limiting.clone());
        async move {
            let path = req.uri().path_and_query().unwrap().as_str().to_string();
            let write = ["putRecord", "deleteRecord", "uploadBlob"]
                .iter()
                .any(|w| path.ends_with(w));
            let take = |n: &AtomicUsize| {
                write && n.fetch_update(SeqCst, SeqCst, |n| n.checked_sub(1)).is_ok()
            };
            if take(&hang) {
                tokio::time::sleep(Duration::from_secs(30)).await;
            }
            if take(&fail) {
                let busy = r#"{"error":"InternalServerError","message":"busy"}"#;
                let mut res = axum::response::Response::new(axum::body::Body::from(busy));
                *res.status_mut() = axum::http::StatusCode::INTERNAL_SERVER_ERROR;
                return res;
            }
            if take(&limit) {
                let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
                let reset = (now.unwrap().as_secs() + 1).to_string();
                let refused = r#"{"error":"RateLimitExceeded","message":"Rate Limit Exceeded"}"#;
                let mut res = axum::response::Response::new(axum::body::Body::from(refused));
                *res.status_mut() = axum::http::StatusCode::TOO_MANY_REQUESTS;
                let reset = axum::http::HeaderValue::from_str(&reset).unwrap();
                res.headers_mut().insert("ratelimit-reset", reset);
                return res;
            }
            let (parts, body) = req.into_parts();
            let body = axum::body::to_bytes(body, usize::MAX).await.unwrap();
            let method = reqwest::Method::from_bytes(parts.method.as_str().as_bytes()).unwrap();
            let mut out = reqwest::Client::new().request(method, format!("{}{path}", net().pds));
            for (k, v) in parts.headers.iter().filter(|(k, _)| *k != "host") {
                out = out.header(k.as_str(), v.as_bytes());
            }
            let res = out.body(body).send().await.unwrap();
            let status = res.status().as_u16();
            let kind = res.headers().get("content-type").cloned();
            let mut back =
                axum::response::Response::new(axum::body::Body::from(res.bytes().await.unwrap()));
            *back.status_mut() = axum::http::StatusCode::from_u16(status).unwrap();
            if let Some(kind) = kind {
                let kind = axum::http::HeaderValue::from_bytes(kind.as_bytes()).unwrap();
                back.headers_mut().insert("content-type", kind);
            }
            back
        }
    };
    let router = axum::Router::new().fallback(pass);
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    Flaky {
        url,
        fail,
        hang,
        limit,
    }
}

/// Writes the PDS fails, or never answers, wedge nothing: a read that
/// failed for them is made again by itself, a write that hung is given up
/// on and made again, one a rate limit refused waits until it lets writes
/// in, and the runs after start as ever. A stop waits for a
/// run that hangs only so long, then cancels it, killing its steps. And a
/// push Jetstream never rang for is found by the sweep.
#[test]
fn a_failing_pds_wedges_nothing() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let work = owner.work("flaky", 1);
    owner.git(&work, &["remote", "add", "origin", &owner.url("flaky")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    std::fs::write(
        dir.join("quick.yml"),
        "on: {push: {refs: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n",
    )
    .unwrap();
    let stuck = format!(
        "on: {{push: {{refs: [refs/heads/stuck]}}}}\nsteps:\n  - run: |\n      echo $$ > {go}/stuck\n      sleep 600\n",
        go = go.display()
    );
    std::fs::write(dir.join("stuck.yml"), stuck).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    let js = jetstream(&rt);
    let opts = knotless_run::Options {
        jetstream: js.url.clone(),
        repos: vec![format!("{}/flaky", owner.account.handle)],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: None,
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        step_timeout: Duration::from_secs(60),
        run_timeout: None,
        jobs: 1,
        pull_jobs: 1,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: Some(runner.path("caches")),
        sweep: Duration::from_secs(1),
        stop_wait: Duration::from_secs(2),
        pulls: false,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    };
    let pds = flaky(&rt);
    let ctx = common::ctx(&runner);
    let reporter = rt.block_on(check::Reporter::new(&ctx)).unwrap();
    // Its writes alone wait only so long for an answer: past what a busy
    // PDS takes for one that passes, as 3 s was not, short of the hang.
    let mut env = runner.env();
    env.timeout = Duration::from_secs(10);
    let impatient = knotless::env::Ctx::new(env);
    let reporter = reporter.through(&impatient, &pds.url);
    let started = knotless_run::Runner::start_as(opts, ctx, reporter);
    let (run, watches) = rt.block_on(started).unwrap();
    let watch = watches[0].clone();
    let key = watch.key.clone();
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.spawn(knotless_run::watch(run.clone(), watches));
    let of = |r: &str, sha: &str, wf: &str| {
        let rkey = check::rkey(&owner.account.did, &key, r, sha, None, wf);
        runner.record(check::NSID, &rkey)
    };

    // Every write failing, the push's read fails, each write tried a few
    // times; once they pass, it is read again, rung by nothing, and runs.
    pds.fail.store(1000, SeqCst);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let first = owner.git(&work, &["rev-parse", "HEAD"]);
    wait_until(|| pds.fail.load(SeqCst) <= 1000 - 8);
    pds.fail.store(0, SeqCst);
    wait_until(|| of("refs/heads/main", &first, "quick")["status"] == "success");

    // Writes that never answer are given up on and made again.
    pds.hang.store(1, SeqCst);
    let second = owner.commit_file(&work, "f", "f\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    wait_until(|| of("refs/heads/main", &second, "quick")["status"] == "success");
    assert_eq!(pds.hang.load(SeqCst), 0);

    // Writes a rate limit refuses, more of them than a busy PDS's are
    // tried, wait until it lets writes in, and the run passes.
    pds.limit.store(6, SeqCst);
    let third = owner.commit_file(&work, "g", "g\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    wait_until(|| of("refs/heads/main", &third, "quick")["status"] == "success");
    assert_eq!(pds.limit.load(SeqCst), 0);

    // A stop waits for a run that hangs only so long, then cancels it, and
    // its steps are killed.
    owner.git(&work, &["push", "-q", "origin", "HEAD:stuck"]);
    wait_until(|| go.join("stuck").exists());
    let stopping = rt.spawn({
        let run = run.clone();
        async move { run.stop().await }
    });
    promptly(&rt, stopping);
    let cut = of("refs/heads/stuck", &third, "stuck");
    assert_eq!(cut["status"], "cancelled", "{cut}");
    let log = blob(
        &runner.account.did,
        cut["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.ends_with("==> cancelled as the runner stopped, past what a stop waits for\n"),
        "{log}"
    );
    let pid = std::fs::read_to_string(go.join("stuck")).unwrap();
    assert!(!alive(pid.trim()));
}

/// Whether the process `pid` is there.
fn alive(pid: &str) -> bool {
    let signalled = std::process::Command::new("kill")
        .args(["-0", pid])
        .stderr(std::process::Stdio::null())
        .status();
    signalled.unwrap().success()
}

/// Waits for `stopping`, a stop that has only its last few seconds left,
/// rather than a backlog's minutes.
fn promptly(rt: &tokio::runtime::Runtime, stopping: tokio::task::JoinHandle<()>) {
    let within = async { tokio::time::timeout(Duration::from_secs(20), stopping).await };
    rt.block_on(within).expect("the stop waited").unwrap();
}

#[test]
fn the_binary() {
    let s = Sandbox::new();
    let exe = env!("CARGO_BIN_EXE_knotless-run");
    let out = s.command(exe, s.dir.path()).output().unwrap();
    assert_eq!(out.status.code(), Some(2));
    assert!(String::from_utf8_lossy(&out.stderr).starts_with("name a repository"));
    // Nobody to run as.
    let out = s
        .command(exe, s.dir.path())
        .arg(format!("{}/proj", s.account.handle))
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(err.contains("knotless-run: not logged in"), "{err}");
    // Said as a log line too, as JSON.
    assert!(err.contains(r#""message":"knotless-run failed""#), "{err}");

    // Logged in and watching, it stops cleanly when asked to.
    let mut s = s;
    let handle = s.account.handle.clone();
    s.set("KNOTLESS_IDENTIFIER", &handle);
    let work = s.work("proj", 1);
    s.git(&work, &["push", "-q", &s.url("proj"), "main"]);
    let mut child = s
        .command(exe, s.dir.path())
        .args(["--jetstream", "ws://127.0.0.1:9", "--executor", "shell"])
        .arg(format!("{handle}/proj"))
        .stderr(std::process::Stdio::piped())
        .spawn()
        .unwrap();
    // It logs, as JSON lines, that it started, and then what it watches.
    let lines = std::io::BufRead::lines(std::io::BufReader::new(child.stderr.take().unwrap()));
    let mut lines = lines.map(Result::unwrap);
    let first: serde_json::Value = serde_json::from_str(&lines.next().unwrap()).unwrap();
    assert_eq!(first["message"], "knotless-run starting");
    assert!(first["version"].is_string());
    let watching = lines
        .find(|l| l.contains(r#""message":"watching""#))
        .unwrap();
    assert!(watching.contains(&format!("{handle}/proj")), "{watching}");
    let killed = std::process::Command::new("kill")
        .args(["-TERM", &child.id().to_string()])
        .status()
        .unwrap();
    assert!(killed.success());
    assert_eq!(child.wait().unwrap().code(), Some(0));
}

/// An appview over the reference PDS, following its firehose, at the URL
/// returned.
fn appview(rt: &tokio::runtime::Runtime, s: &Sandbox) -> String {
    listening(rt, s).0
}

/// An appview as [`appview`] makes, and its following, which stops when
/// aborted: the appview then hears of nothing written after.
fn listening(rt: &tokio::runtime::Runtime, s: &Sandbox) -> (String, tokio::task::JoinHandle<()>) {
    serving(rt, s, None)
}

/// An appview as [`listening`] makes, naming `runner` for new
/// repositories, and then a gateway too, for the service tokens of
/// invites to private ones.
fn serving(
    rt: &tokio::runtime::Runtime,
    s: &Sandbox,
    runner: Option<&str>,
) -> (String, tokio::task::JoinHandle<()>) {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    // A gateway's did:web names a host, which an IP address is not.
    let base = format!("http://localhost:{}", listener.local_addr().unwrap().port());
    let index = knotless_serve::index::Index::open(&s.path("index.db")).unwrap();
    let gateway = runner.map(|_| knotless_serve::gateway::Gateway::new(&base).unwrap());
    let mut app = knotless_serve::App::build(common::ctx(s), Duration::ZERO, gateway, Some(index));
    Arc::get_mut(&mut app).unwrap().default_runner = runner.map(str::to_string);
    let router = knotless_serve::router(app.clone());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    let ears = rt.spawn(knotless_serve::firehose::watch(
        app,
        net().pds.replace("http://", "ws://"),
    ));
    (base, ears)
}

const PULLS: &str = r#"
on: {pull: {base: [refs/heads/main]}}
secrets: [TOKEN]
cache: [target]
steps:
  - run: |
      echo "pull $KNOTLESS_RUN_PULL at $KNOTLESS_RUN_REF on $KNOTLESS_RUN_BASE token=${TOKEN:-none}"
      for u in CACHES/runs/*/*/upper; do echo pull > $u/pull; done
      test -f f
      test ! -f h || echo "with h"
"#;

#[test]
fn a_pull_runs_its_head() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut contrib = Sandbox::new();
    let mut runner = Sandbox::new();
    let base = appview(&rt, &Sandbox::new());
    owner.set("KNOTLESS_APPVIEW", &base);
    contrib.set("KNOTLESS_APPVIEW", &base);
    let theirs = contrib.account.handle.clone();
    contrib.set("KNOTLESS_IDENTIFIER", &theirs);
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);

    let work = owner.work("proj", 1);
    std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
    let caches = runner.path("caches");
    let pulls = PULLS.replace("CACHES", &caches.display().to_string());
    std::fs::write(work.join(".knotless/workflows/pulls.yml"), pulls).unwrap();
    // A push's, of a branch nothing here pushes: no push runs.
    std::fs::write(
        work.join(".knotless/workflows/push.yml"),
        "on: {push: {refs: [refs/heads/release]}}\nsteps: [{run: 'true'}]\n",
    )
    .unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let key = owner.record("dev.knotless.repo", "proj")["key"]
        .as_str()
        .unwrap()
        .to_string();

    let (code, fork) = contrib.cli(&["fork", &owner.url("proj")], b"");
    assert_eq!(code, 0, "{fork}");
    let fork = fork.trim().to_string();
    let fork_name = fork.rsplit('/').next().unwrap().to_string();
    let fork_key = contrib.record("dev.knotless.repo", &fork_name)["key"]
        .as_str()
        .unwrap()
        .to_string();
    // The pulls of whoever may push to it run as they move: the owner's,
    // and a collaborator's, from their own fork.
    let collaborators = |named: serde_json::Value| {
        let mut record = owner.record("dev.knotless.repo", "proj");
        record["collaborators"] = named;
        owner.put_record("dev.knotless.repo", "proj", record);
    };
    collaborators(serde_json::json!([{"did": contrib.account.did}]));
    let clone = contrib.path("clone");
    contrib.git(contrib.dir.path(), &["clone", "-q", &fork, "clone"]);
    let pull = |branch: &str| {
        let (code, out) = contrib.cli(
            &["pull", &owner.url("proj"), "main", &fork, branch, branch],
            b"",
        );
        assert_eq!(code, 0, "{out}");
        out.trim().to_string()
    };
    contrib.commit_file(&clone, "f", "f\n");
    contrib.git(&clone, &["push", "-q", "origin", "HEAD:topic"]);
    let topic = pull("topic");
    // Into a branch the workflow does not run on.
    owner.git(&work, &["push", "-q", "origin", "main:side"]);
    contrib.git(&clone, &["push", "-q", "origin", "HEAD:side"]);
    let (code, out) = contrib.cli(
        &["pull", &owner.url("proj"), "side", &fork, "side", "side"],
        b"",
    );
    assert_eq!((code, out.is_empty()), (0, false), "{out}");
    let lists = |uri: &str| {
        let (_, out) = owner.cli(&["list", &owner.url("proj"), "--pulls"], b"");
        out.contains(uri.rsplit('/').next().unwrap())
    };
    wait_until(|| lists(&topic));
    let main = owner.git(&work, &["rev-parse", "main"]);

    let js = jetstream(&rt);
    let mut opts = runner_opts(&runner, &owner, &js);
    opts.pulls = true;
    // It finds pulls through an appview, and has none: KNOTLESS_APPVIEW is empty.
    let err = rt
        .block_on(knotless_run::run(opts.clone(), common::ctx(&runner)))
        .unwrap_err();
    assert!(
        err.to_string().contains("KNOTLESS_APPVIEW is set empty"),
        "{err}"
    );
    runner.set("KNOTLESS_APPVIEW", &base);
    rt.spawn(knotless_run::run(opts.clone(), common::ctx(&runner)));

    let check = |uri: &str| check_on(&runner, &owner, &key, uri, "pulls");
    // Watching starts where the pulls are: the open one has not run.
    wait_until(|| {
        std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&topic))
    });
    assert!(check(&topic).is_null());

    // Its head moving, in the fork, runs it: the pull's workflows, without
    // the secrets a push gets.
    let sha = contrib.commit_file(&clone, "g", "g\n");
    contrib.git(&clone, &["push", "-q", "origin", "HEAD:topic"]);
    let fork_ref = ref_rkey(&fork_key, "refs/heads/topic");
    wait_until(|| {
        js.ring(&contrib.account.did, &fork_ref);
        check(&topic)["finishedAt"].is_string()
    });
    let ran = check(&topic);
    assert_eq!(ran["status"], "success", "{ran}");
    assert_eq!(ran["pull"], topic.as_str());
    assert_eq!(ran["ref"], "refs/heads/topic");
    assert_eq!(ran["commit"], sha.as_str());
    assert_eq!(ran["baseCommit"], main.as_str());
    assert_eq!(ran["repo"]["key"], key.as_str());
    let log = blob(
        &runner.account.did,
        ran["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.contains(&format!(
            "pull {topic} at refs/heads/topic on {main} token=none"
        )),
        "{log}"
    );
    // A push workflow does not run on a pull.
    assert!(check_on(&runner, &owner, &key, &topic, "push").is_null());
    // What it wrote to its cache went with it, its layers too: the base
    // is as the default branch left it, with nothing.
    let held = |dir: &std::path::Path| {
        let all = std::fs::read_dir(dir).unwrap();
        let mut all: Vec<_> = all.map(|e| e.unwrap().path()).collect();
        all.sort();
        all
    };
    wait_until(|| held(&caches.join("runs")).is_empty());
    let bases = || {
        let all = held(&caches.join("bases"));
        all.iter().flat_map(|c| held(c)).collect::<Vec<_>>()
    };
    assert_eq!(bases().len(), 1);
    assert!(held(&bases()[0]).is_empty());

    // The base moving runs it again, at its head merged with the base, as
    // merging it would leave the base.
    let main_ref = ref_rkey(&key, "refs/heads/main");
    let at_base = |tip: &str| {
        wait_until(|| {
            js.ring(&owner.account.did, &main_ref);
            let c = check(&topic);
            c["baseCommit"] == tip && c["finishedAt"].is_string()
        });
        check(&topic)
    };
    // A pull whose base is deleted has nothing to merge into, and is passed
    // over.
    owner.git(&work, &["push", "-q", "origin", ":side"]);
    let moved = owner.commit_file(&work, "h", "h\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let ran = at_base(&moved);
    assert_eq!(
        (&ran["status"], &ran["commit"]),
        (&"success".into(), &sha.as_str().into()),
        "{ran}"
    );
    let log = blob(
        &runner.account.did,
        ran["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(log.contains("with h"), "{log}");
    // One that conflicts with it fails without running, naming where.
    let clash = owner.commit_file(&work, "f", "the owner's f\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let failed = at_base(&clash);
    assert_eq!(failed["status"], "failure", "{failed}");
    let error = failed["error"].as_str().unwrap();
    let said = format!("conflicts with main at {} in f: merge main", &clash[..12]);
    assert!(error.contains(&said), "{error}");

    // A pull opened rings for the repository it is into, and runs.
    contrib.git(&clone, &["checkout", "-q", "-b", "bare", "HEAD~2"]);
    contrib.git(&clone, &["rm", "-q", "f"]);
    contrib.git(&clone, &["commit", "-q", "-m", "no f"]);
    contrib.git(&clone, &["push", "-q", "origin", "bare"]);
    let bare = pull("bare");
    let record = serde_json::json!({"repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key}});
    wait_until(|| {
        js.send(
            &contrib.account.did,
            "dev.knotless.pull",
            "p",
            Some(record.clone()),
        );
        check(&bare)["finishedAt"].is_string()
    });
    assert_eq!(check(&bare)["status"], "failure");

    // The owner's checks of it, by the runner the repository names.
    let me = runner.account.handle.clone();
    assert_eq!(
        owner
            .cli(&["runners", &owner.url("proj"), "add", &me], b"")
            .0,
        0
    );
    let (code, out) = owner.cli(&["checks", &owner.url("proj"), "--pull", &bare], b"");
    assert_eq!(code, 0, "{out}");
    assert!(
        out.starts_with(&format!("failure\tpulls\t{bare}\t")),
        "{out}"
    );

    // Asked to, it runs a pull's head again.
    let finished = check(&bare)["finishedAt"].clone();
    let (code, asked) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &bare], b"");
    assert_eq!(code, 0, "{asked}");
    let asked = asked.rsplit('/').next().unwrap().to_string();
    wait_until(|| {
        js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
        // Read once: read twice, the first could be the last run's end and
        // the second this run's start, whose finishedAt is null.
        let now = check(&bare)["finishedAt"].clone();
        now.is_string() && now != finished
    });

    // Noted again once it has run.
    wait_until(|| {
        std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&bare))
    });

    // Asked for one of its workflows by name, it runs that one where it is.
    let finished = check(&bare)["finishedAt"].clone();
    let url = owner.url("proj");
    let args = ["rerun", &url, "--pull", &bare, "--workflow", "pulls"];
    let (code, asked) = owner.cli(&args, b"");
    assert_eq!(code, 0, "{asked}");
    let asked = asked.rsplit('/').next().unwrap().to_string();
    wait_until(|| {
        js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
        let now = check(&bare)["finishedAt"].clone();
        now.is_string() && now != finished
    });
    let args = [
        "rerun",
        &url,
        "--pull",
        &bare,
        "--workflow",
        "pulls",
        "-e",
        "A=1",
    ];
    // Variables go only to a run at a ref.
    assert_eq!(owner.cli(&args, b"").0, 1);

    // Closed, it drops out. Reopened where it ran, rung by its state
    // record alone, it is noted, not run again: its checks of that head
    // and base are there. Reopened at a head that moved while it was
    // closed, it runs.
    let noted =
        || std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&bare));
    let set_state = |cmd: &str, noted_after: bool| {
        let (code, out) = contrib.cli(&[cmd, &owner.url("proj"), &bare], b"");
        assert_eq!(code, 0, "{out}");
        wait_until(|| {
            js.send(
                &contrib.account.did,
                "dev.knotless.state",
                "s",
                Some(record.clone()),
            );
            noted() == noted_after
        });
    };
    wait_until(|| check(&bare)["finishedAt"].is_string());
    let finished = check(&bare)["finishedAt"].clone();
    set_state("close", false);
    set_state("reopen", true);
    std::thread::sleep(Duration::from_millis(500));
    assert_eq!(check(&bare)["finishedAt"], finished);
    set_state("close", false);
    let reopened = contrib.commit_file(&clone, "r", "r\n");
    contrib.git(&clone, &["push", "-q", "origin", "bare"]);
    set_state("reopen", true);
    wait_until(|| {
        js.send(
            &contrib.account.did,
            "dev.knotless.state",
            "s",
            Some(record.clone()),
        );
        let c = check(&bare);
        c["commit"] == reopened.as_str() && c["finishedAt"].is_string()
    });

    // A head branch deleted has nothing to run, and drops out. Any run
    // still going ends first, so what follows is only what the deletion
    // does.
    wait_until(|| check(&bare)["finishedAt"].is_string());
    let finished = check(&bare)["finishedAt"].clone();
    contrib.git(&clone, &["push", "-q", "origin", ":bare"]);
    wait_until(|| {
        js.ring(
            &contrib.account.did,
            &ref_rkey(&fork_key, "refs/heads/bare"),
        );
        std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| !s.contains(&bare))
    });
    assert_eq!(check(&bare)["finishedAt"], finished);
    // Asked to run it now, with no branch to run, it lets the request go
    // rather than keep it for a pull that cannot run.
    let (code, gone) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &bare], b"");
    assert_eq!(code, 0, "{gone}");
    let gone = gone.rsplit('/').next().unwrap().to_string();
    wait_until(|| {
        js.send(&owner.account.did, "dev.knotless.rerun", &gone, None);
        std::fs::read_to_string(runner.path("state.json"))
            .is_ok_and(|s| s.contains(&gone) && !s.contains(&bare))
    });
    assert_eq!(check(&bare)["finishedAt"], finished);

    // Anyone else's pull is noted but not run, its code unvouched for,
    // until the owner asks; then it runs where it is.
    collaborators(serde_json::json!([]));
    contrib.git(&clone, &["fetch", "-q", &owner.url("proj"), "main"]);
    contrib.git(&clone, &["checkout", "-q", "-b", "late", "FETCH_HEAD"]);
    contrib.commit_file(&clone, "l", "late\n");
    contrib.git(&clone, &["push", "-q", "origin", "late"]);
    let late = pull("late");
    wait_until(|| {
        js.send(
            &contrib.account.did,
            "dev.knotless.pull",
            "p",
            Some(record.clone()),
        );
        std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&late))
    });
    assert!(check(&late).is_null());
    let (code, asked) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &late], b"");
    assert_eq!(code, 0, "{asked}");
    let asked = asked.rsplit('/').next().unwrap().to_string();
    wait_until(|| {
        js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
        check(&late)["finishedAt"].is_string()
    });
    assert_eq!(check(&late)["status"], "success");
    // Its workflow asks for secrets, which a push's run of it may have left
    // in its cache: anyone else's pull starts from a cache of its own,
    // which nothing writes.
    assert_eq!(held(&caches.join("bases")).len(), 2);
    assert!(bases().iter().all(|b| held(b).is_empty()));
    // Nor is what anyone else's pull wrote ever kept for main to take.
    assert!(held(&caches.join("kept")).is_empty());

    // An agent the repository names runs its pulls by opening them, as a
    // collaborator does, and even one that is a collaborator too runs as
    // code no one vouches for: what it writes is never kept, even by a
    // workflow asking for no secrets, whose trusted pass would be.
    let agents = |named: serde_json::Value| {
        let mut record = owner.record("dev.knotless.repo", "proj");
        record["agents"] = named;
        owner.put_record("dev.knotless.repo", "proj", record);
    };
    agents(serde_json::json!([{"did": contrib.account.did}]));
    let bot_yml = format!(
        "on: {{pull: {{base: [refs/heads/main]}}}}\ncache: [target]\nsteps:\n  - run: for u in {}/runs/*/*/upper; do echo bot > $u/bot; done\n",
        caches.display()
    );
    for (branch, collaborating) in [("bot", false), ("bot-too", true)] {
        let named = if collaborating {
            serde_json::json!([{"did": contrib.account.did}])
        } else {
            serde_json::json!([])
        };
        collaborators(named);
        contrib.git(&clone, &["checkout", "-q", "-b", branch, "late"]);
        std::fs::write(clone.join(".knotless/workflows/bot.yml"), &bot_yml).unwrap();
        contrib.git(&clone, &["add", "."]);
        contrib.git(&clone, &["commit", "-q", "-m", branch]);
        contrib.git(&clone, &["push", "-q", "origin", branch]);
        let bot = pull(branch);
        let ran = |wf: &str| check_on(&runner, &owner, &key, &bot, wf);
        wait_until(|| {
            js.send(
                &contrib.account.did,
                "dev.knotless.pull",
                "p",
                Some(record.clone()),
            );
            ["pulls", "bot"]
                .iter()
                .all(|wf| ran(wf)["finishedAt"].is_string())
        });
        for wf in ["pulls", "bot"] {
            assert_eq!(ran(wf)["status"], "success", "{branch} {wf}");
        }
        wait_until(|| held(&caches.join("runs")).is_empty());
        assert!(bases().iter().all(|b| held(b).is_empty()), "{branch}");
        assert!(held(&caches.join("kept")).is_empty(), "{branch}");
    }
    agents(serde_json::json!([]));
    collaborators(serde_json::json!([]));
    contrib.git(&clone, &["checkout", "-q", "late"]);

    // A collaborator's pull whose lock adds shared steps from an account the
    // repository does not vouch for runs as code no one vouches for, and
    // says so; the owner's own add nothing.
    collaborators(serde_json::json!([{"did": contrib.account.did}]));
    contrib.git(&clone, &["checkout", "-q", "-b", "locked", "late"]);
    let (mine, theirs) = (&owner.account.handle, &runner.account.handle);
    let commit = "a".repeat(40);
    let lock = format!(
        "{mine}/steps/a@1: {{version: 1.0.0, commit: {commit}}}\n{theirs}/steps/b@1: {{version: 1.0.0, commit: {commit}}}\n"
    );
    std::fs::write(clone.join(".knotless/uses.lock"), lock).unwrap();
    contrib.git(&clone, &["add", "."]);
    contrib.git(&clone, &["commit", "-q", "-m", "locked"]);
    contrib.git(&clone, &["push", "-q", "origin", "locked"]);
    let locked = pull("locked");
    wait_until(|| {
        js.send(
            &contrib.account.did,
            "dev.knotless.pull",
            "p",
            Some(record.clone()),
        );
        check(&locked)["finishedAt"].is_string()
    });
    let ran = check(&locked);
    assert_eq!(ran["status"], "success", "{ran}");
    let log = blob(
        &runner.account.did,
        ran["log"]["ref"]["$link"].as_str().unwrap(),
    );
    let said = format!(
        "==> runs as code no one vouches for, as its lock adds shared steps from {theirs}\n"
    );
    assert!(log.contains(&said), "{log}");
    collaborators(serde_json::json!([]));
    contrib.git(&clone, &["checkout", "-q", "late"]);

    // A pull's run of a head and base it is no longer at is cancelled, as
    // when its base moves on, which runs anyone else's pull no further.
    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    for (wf, extra) in [("hold", ""), ("keep", "superseded: finish\n")] {
        let text = format!(
            "{extra}on: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n  - name: hold\n    run: |\n      echo $$ > {go}/pid-{wf}\n      touch {go}/started-{wf}\n      until [ -f {go}/{wf} ]; do sleep 0.1; done\n",
            go = go.display()
        );
        std::fs::write(clone.join(format!(".knotless/workflows/{wf}.yml")), text).unwrap();
    }
    contrib.git(&clone, &["add", "."]);
    contrib.git(&clone, &["commit", "-q", "-m", "hold"]);
    contrib.git(&clone, &["push", "-q", "origin", "late"]);
    let (code, asked) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &late], b"");
    assert_eq!(code, 0, "{asked}");
    let asked = asked.rsplit('/').next().unwrap().to_string();
    wait_until(|| {
        js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
        ["hold", "keep"]
            .iter()
            .all(|wf| go.join(format!("started-{wf}")).exists())
    });
    owner.commit_file(&work, "m", "m\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let of = |wf: &str| check_on(&runner, &owner, &key, &late, wf);
    // Both, even one whose runs finish, which a newer commit's run never
    // cuts off.
    wait_until(|| {
        js.ring(&owner.account.did, &main_ref);
        ["hold", "keep"]
            .iter()
            .all(|wf| of(wf)["finishedAt"].is_string())
    });
    for wf in ["hold", "keep"] {
        let moved = of(wf);
        assert_eq!(moved["status"], "cancelled", "{moved}");
        let log = blob(
            &runner.account.did,
            moved["log"]["ref"]["$link"].as_str().unwrap(),
        );
        assert!(
            log.ends_with("==> cancelled as the pull is no longer at that head and base\n"),
            "{log}"
        );
        // Its steps are killed, not left to run on.
        let pid = std::fs::read_to_string(go.join(format!("pid-{wf}"))).unwrap();
        assert!(!alive(pid.trim()), "{wf} still runs");
    }
}

/// A pull merged as it was checked leaves its base at the tree its run
/// passed at: the push that follows files those checks as passed, naming
/// the pull's run, rather than running them again. A tree no pull passed
/// at runs, as does a deploy, which runs for what it does, and one that
/// asks for secrets, which no pull's run gets.
#[test]
fn a_merged_pull_passes_its_push() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    // At a did:web, for the service tokens an owner's secrets are left
    // and read with.
    let did = runner.account.did.clone();
    let (base, _ears) = serving(&rt, &Sandbox::new(), Some(&did));
    owner.set("KNOTLESS_APPVIEW", &base);
    owner.set("KNOTLESS_DID_WEB_SCHEME", "http");
    runner.set("KNOTLESS_DID_WEB_SCHEME", "http");
    let mine = owner.account.handle.clone();
    owner.set("KNOTLESS_IDENTIFIER", &mine);
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    runner.set("KNOTLESS_APPVIEW", &base);

    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let work = owner.work("proj", 1);
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    // Each run notes the commit it is of; ci fails while `break` is there,
    // and writes it to its cache, which each run's layer is under `runs`.
    let both = "on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/main]}}";
    let caches = runner.path("caches");
    let warm = format!(
        "  - name: warm\n    run: for u in {}/runs/*/*/upper; do echo \"$KNOTLESS_RUN_SHA\" > $u/sha; done\n",
        caches.display()
    );
    for (wf, extra) in [
        ("ci", format!("{both}\ncache: [target]\n")),
        ("secret", format!("{both}\nsecrets: [TOKEN]\n")),
        (
            "deploy",
            "on: {push: {refs: [refs/heads/main]}}\nsuperseded: finish\n".to_string(),
        ),
    ] {
        let text = format!(
            "{extra}steps:\n  - name: note\n    run: |\n      echo \"$KNOTLESS_RUN_SHA\" >> {go}/{wf}\n      test ! -f {go}/break\n",
            go = go.display()
        );
        let text = match wf {
            "ci" => text + &warm,
            "secret" => text + "  - name: token\n    run: echo \"token=$TOKEN\"\n",
            _ => text,
        };
        std::fs::write(dir.join(format!("{wf}.yml")), text).unwrap();
    }
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let url = owner.url("proj");
    assert_eq!(owner.cli(&["protect", &url, "main"], b"").0, 0);
    assert_eq!(owner.cli(&["runners", &url, "add", &handle], b"").0, 0);

    let opts = knotless_run::Options {
        jetstream: "ws://127.0.0.1:9".into(),
        repos: vec![format!("{}/proj", owner.account.handle)],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: None,
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        step_timeout: Duration::from_secs(60),
        run_timeout: None,
        jobs: 4,
        pull_jobs: 4,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: Some(runner.path("caches")),
        sweep: Duration::from_secs(600),
        stop_wait: Duration::from_secs(60),
        pulls: true,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    };
    let (run, watches) = rt
        .block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
        .unwrap();
    let watch = &watches[0];
    let key = watch.key.clone();
    let pass = || {
        rt.block_on(run.reconcile(watch)).unwrap();
        rt.block_on(run.idle());
    };
    pass();
    // Its owner seals it a secret, which only a push to main reaches, and
    // another its workflows do not ask for.
    for (name, value) in [("TOKEN", "t0ken\n"), ("UNASKED", "u")] {
        let set = ["secret", &url, "set", name, "--ref", "refs/heads/main"];
        let (code, out) = owner.cli(&set, value.as_bytes());
        assert_eq!(code, 0, "{out}");
    }

    let ran = |wf: &str| std::fs::read_to_string(go.join(wf)).unwrap_or_default();
    let at = |sha: &str, wf: &str| {
        let rkey = check::rkey(&owner.account.did, &key, "refs/heads/main", sha, None, wf);
        runner.record(check::NSID, &rkey)
    };
    let log_of = |check: &serde_json::Value| {
        blob(
            &runner.account.did,
            check["log"]["ref"]["$link"].as_str().unwrap(),
        )
    };
    let lists = |uri: &str| {
        let (_, out) = owner.cli(&["list", &url, "--pulls"], b"");
        out.contains(uri.rsplit('/').next().unwrap())
    };
    // A pull from a branch of main as it is now, run once it is listed.
    let open = |branch: &str, file: &str| {
        owner.git(&work, &["checkout", "-q", "-b", branch, "main"]);
        let head = owner.commit_file(&work, file, &format!("{file}\n"));
        owner.git(&work, &["push", "-q", "origin", branch]);
        let (code, uri) = owner.cli(&["pull", &url, "main", &url, branch, branch], b"");
        assert_eq!(code, 0, "{uri}");
        wait_until(|| lists(&uri));
        pass();
        (uri, head)
    };
    // Main as a merge left it, checked out to work on.
    let merged = |args: &[&str]| {
        let (code, out) = owner.cli(args, b"");
        assert_eq!(code, 0, "{out}");
        owner.git(&work, &["fetch", "-q", "origin", "main"]);
        owner.git(&work, &["checkout", "-q", "-B", "main", "FETCH_HEAD"]);
        let sha = owner.git(&work, &["rev-parse", "HEAD"]);
        pass();
        sha
    };

    // The runs whose layers are kept, and what ci's cache now starts from.
    let held = |dir: std::path::PathBuf| {
        let all = std::fs::read_dir(dir).into_iter().flatten();
        let mut all: Vec<_> = all.map(|e| e.unwrap().path()).collect();
        all.sort();
        all
    };
    let kept = || held(caches.join("kept"));
    let base_sha = || {
        let cache = held(caches.join("bases")).pop().unwrap();
        let newest = held(cache).into_iter().max_by_key(|g| {
            let name = g.file_name().unwrap().to_string_lossy().into_owned();
            name.parse::<u64>().unwrap()
        });
        std::fs::read_to_string(newest.unwrap().join("sha")).unwrap()
    };

    let (topic, head) = open("topic", "a");
    let pulled = check_on(&runner, &owner, &key, &topic, "ci");
    assert_eq!(pulled["status"], "success", "{pulled}");
    assert_eq!(ran("ci"), format!("{head}\n"));
    // A trusted pull's run that passed keeps what it wrote to its cache.
    assert_eq!(kept().len(), 1);
    // A pull's run of what asks for secrets gets none, and says so.
    let secret = log_of(&check_on(&runner, &owner, &key, &topic, "secret"));
    let none = "==> TOKEN is not granted to a pull's or a merge-queue batch's run, which gets no secrets\n";
    assert!(secret.contains(none), "{secret}");

    // Squashed into main: its tree is the one the pull ran at.
    let squashed = merged(&["merge", &url, &topic]);
    assert_ne!(squashed, head);
    let ci = at(&squashed, "ci");
    assert_eq!(ci["status"], "success", "{ci}");
    assert_eq!(ci["commit"], squashed.as_str());
    assert!(ci.get("pull").is_none(), "{ci}");
    assert_eq!(ci["steps"], pulled["steps"]);
    let log = log_of(&ci);
    assert!(log.starts_with("==> passed: the same tree, "), "{log}");
    assert!(
        log.contains(&format!("as pull {topic} at {head} merged with ")),
        "{log}"
    );
    assert!(log.contains("; not run again\n"), "{log}");
    assert_eq!(ran("ci"), format!("{head}\n"), "not run again");
    // And what the pull's run wrote to its cache is what main's runs start
    // from now, as main's own run's would be.
    let said = "==> caches: what that run wrote to /work/target is kept, as this run's would be\n";
    assert!(log.contains(said), "{log}");
    // Then the pull's log, so main's check stands alone once the pull's is
    // pruned.
    let theirs = log_of(&pulled);
    assert!(log.ends_with(&format!("{said}{theirs}")), "{log}");
    assert_eq!(base_sha(), format!("{head}\n"));
    assert!(kept().is_empty());
    // What a deploy's wait for the commit's checks reads.
    let (code, out) = owner.cli(
        &[
            "checks",
            &url,
            "--ref",
            "refs/heads/main",
            "--commit",
            &squashed,
            "--workflow",
            "ci",
        ],
        b"",
    );
    assert_eq!(code, 0, "{out}");
    assert!(out.starts_with("success\tci\trefs/heads/main\t"), "{out}");
    // The deploy runs, and so does what asks for secrets.
    for wf in ["deploy", "secret"] {
        let pushed = at(&squashed, wf);
        assert_eq!(pushed["status"], "success", "{pushed}");
        assert!(ran(wf).contains(&squashed), "{wf}: {}", ran(wf));
        assert!(!log_of(&pushed).contains("not run again"));
    }
    // The push's run opens the secret its owner sealed to the runner, and
    // masks it in the log.
    let log = log_of(&at(&squashed, "secret"));
    assert!(log.contains("token=***\n"), "{log}");
    assert!(
        !log.contains("t0ken") && !log.contains("not granted"),
        "{log}"
    );

    // A tree no pull ran at runs.
    let direct = owner.commit_file(&work, "b", "b\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    pass();
    assert_eq!(at(&direct, "ci")["status"], "success");
    assert!(ran("ci").contains(&direct), "{}", ran("ci"));
    assert_eq!(base_sha(), format!("{direct}\n"));

    // A pull that moves on drops what its run there kept, for its run
    // where it is now.
    let (moving, _) = open("moving", "d");
    let first = kept();
    assert_eq!(first.len(), 1);
    let moved = owner.commit_file(&work, "e", "e\n");
    owner.git(&work, &["push", "-q", "origin", "moving"]);
    pass();
    let pulled = check_on(&runner, &owner, &key, &moving, "ci");
    assert_eq!(pulled["commit"], moved.as_str(), "{pulled}");
    let then = kept();
    assert_eq!(then.len(), 1);
    assert_ne!(then, first);
    // Main run again meanwhile moves its cache's base on, and what the
    // pull's run wrote over the one before is not taken when it merges.
    let again = ["rerun", &url, "--ref", "refs/heads/main"];
    assert_eq!(owner.cli(&again, b"").0, 0);
    pass();
    assert_eq!(ran("ci").matches(&direct).count(), 2, "{}", ran("ci"));
    let taken = merged(&["merge", &url, &moving]);
    let log = log_of(&at(&taken, "ci"));
    assert!(log.contains("; not run again\n"), "{log}");
    let said = "==> caches: what that run wrote to /work/target is not kept: the base it was written over has moved on since, and the next run here warms it\n";
    // Before the pull's own log, which it copies.
    assert!(log.contains(said), "{log}");
    assert_eq!(base_sha(), format!("{direct}\n"));
    assert!(kept().is_empty());

    // A pull whose run passed and then failed when run again, at the same
    // head and base, is not reused: its check says what it last did.
    let (late, _) = open("late", "c");
    assert_eq!(
        check_on(&runner, &owner, &key, &late, "ci")["status"],
        "success"
    );
    std::fs::write(go.join("break"), "").unwrap();
    assert_eq!(owner.cli(&["rerun", &url, "--pull", &late], b"").0, 0);
    pass();
    let failed = check_on(&runner, &owner, &key, &late, "ci");
    assert_eq!(failed["status"], "failure", "{failed}");
    std::fs::remove_file(go.join("break")).unwrap();
    let forced = merged(&["merge", &url, &late, "--force"]);
    let ci = at(&forced, "ci");
    assert_eq!(ci["status"], "success", "{ci}");
    assert!(ran("ci").contains(&forced), "{}", ran("ci"));
    assert!(!log_of(&ci).contains("not run again"));
}

#[test]
fn pulls_warm_where_pushes_run_elsewhere() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let base = appview(&rt, &Sandbox::new());
    owner.set("KNOTLESS_APPVIEW", &base);
    let mine = owner.account.handle.clone();
    owner.set("KNOTLESS_IDENTIFIER", &mine);
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    runner.set("KNOTLESS_APPVIEW", &base);

    // ci notes the commit it is of in its cache, whose layers are each
    // run's under `runs`.
    let work = owner.work("proj", 1);
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    let caches = runner.path("caches");
    let ci = format!(
        "on: {{push: {{refs: [refs/heads/main]}}, pull: {{base: [refs/heads/main]}}}}\ncache: [target]\nsteps:\n  - run: for u in {}/runs/*/*/upper; do echo \"$KNOTLESS_RUN_SHA\" > $u/sha; done\n",
        caches.display()
    );
    std::fs::write(dir.join("ci.yml"), ci).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let url = owner.url("proj");

    // Its pulls alone: another runner runs its pushes.
    let repos = vec![format!("{mine}/proj")];
    let opts = knotless_run::Options {
        jetstream: "ws://127.0.0.1:9".into(),
        pulls_only: repos.clone(),
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        caches: Some(caches.clone()),
        pulls: true,
        keep_checks: None,
        ..knotless_run::Options::new(repos)
    };
    let (run, watches) = rt
        .block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
        .unwrap();
    let watch = &watches[0];
    let key = watch.key.clone();
    let pass = || {
        rt.block_on(run.reconcile(watch)).unwrap();
        rt.block_on(run.idle());
    };
    pass();
    let lists = |uri: &str| {
        let (_, out) = owner.cli(&["list", &url, "--pulls"], b"");
        out.contains(uri.rsplit('/').next().unwrap())
    };
    let open = |branch: &str, file: &str| {
        owner.git(&work, &["checkout", "-q", "-b", branch, "main"]);
        let head = owner.commit_file(&work, file, &format!("{file}\n"));
        owner.git(&work, &["push", "-q", "origin", branch]);
        let (code, uri) = owner.cli(&["pull", &url, "main", &url, branch, branch], b"");
        assert_eq!(code, 0, "{uri}");
        wait_until(|| lists(&uri));
        pass();
        (uri, head)
    };
    let held = |dir: std::path::PathBuf| {
        let all = std::fs::read_dir(dir).into_iter().flatten();
        let mut all: Vec<_> = all.map(|e| e.unwrap().path()).collect();
        all.sort();
        all
    };
    let base_sha = || {
        let cache = held(caches.join("bases")).pop().unwrap();
        let newest = held(cache).into_iter().max_by_key(|g| {
            let name = g.file_name().unwrap().to_string_lossy().into_owned();
            name.parse::<u64>().unwrap()
        });
        std::fs::read_to_string(newest.unwrap().join("sha")).unwrap()
    };

    // A trusted pull's run that passes is what the next runs start from,
    // as a push's would be where pushes run: none here takes kept layers.
    let (topic, head) = open("topic", "a");
    let pulled = check_on(&runner, &owner, &key, &topic, "ci");
    assert_eq!(pulled["status"], "success", "{pulled}");
    let log = blob(
        &runner.account.did,
        pulled["log"]["ref"]["$link"].as_str().unwrap(),
    );
    let said =
        "==> caches: what this run wrote to /work/target is what the runs here start from now\n";
    assert!(log.contains(said), "{log}");
    assert_eq!(base_sha(), format!("{head}\n"));
    assert!(held(caches.join("kept")).is_empty());
    // The next pull goes on from it.
    let (_, other) = open("other", "b");
    assert_eq!(base_sha(), format!("{other}\n"));
    // A push to main runs nothing here, nor the pulls it moves the base
    // of, as the queue checks them against it.
    owner.git(&work, &["checkout", "-q", "main"]);
    let tip = owner.commit_file(&work, "c", "c\n");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    pass();
    assert!(check_of(&runner, &owner, &key, "ci").is_null());
    assert_eq!(check_on(&runner, &owner, &key, &topic, "ci"), pulled);
    // Its head moving runs it, merged with main as it is now, and goes on
    // from the base as it was.
    owner.git(&work, &["checkout", "-q", "topic"]);
    let moved = owner.commit_file(&work, "a2", "a2\n");
    owner.git(&work, &["push", "-q", "origin", "topic"]);
    pass();
    let again = check_on(&runner, &owner, &key, &topic, "ci");
    assert_eq!(again["baseCommit"], tip.as_str(), "{again}");
    let now = base_sha();
    assert!(
        [moved, other].iter().any(|s| now == format!("{s}\n")),
        "{now}"
    );
}

/// Runs on every push to main, and on a pull only when it changes `web`.
const WEB: &str = r#"
on:
  push: {refs: [refs/heads/main]}
  pull: {base: [refs/heads/main], paths: [web/**, .knotless/workflows/web.yml]}
steps:
  - name: build
    run: test -d web
  - name: check
    run: 'true'
"#;

#[test]
fn paths_leave_out_what_a_change_misses() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let base = appview(&rt, &Sandbox::new());
    owner.set("KNOTLESS_APPVIEW", &base);
    let mine = owner.account.handle.clone();
    owner.set("KNOTLESS_IDENTIFIER", &mine);
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    runner.set("KNOTLESS_APPVIEW", &base);

    let work = owner.work("proj", 1);
    for dir in [".knotless/workflows", "web", "crates"] {
        std::fs::create_dir_all(work.join(dir)).unwrap();
    }
    for (file, text) in [
        ("web.yml", WEB),
        // Every pull, whatever it changes, as check and test do.
        (
            "all.yml",
            "on: {pull: {base: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n",
        ),
        // A push to any branch that changes `web`, or one new.
        (
            "branches.yml",
            "on: {push: {paths: [web/**]}}\nsteps: [{name: look, run: 'true'}]\n",
        ),
    ] {
        std::fs::write(work.join(".knotless/workflows").join(file), text).unwrap();
    }
    std::fs::write(work.join("web/page.svelte"), "page\n").unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    let url = owner.url("proj");
    owner.git(&work, &["remote", "add", "origin", &url]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let key = owner.record("dev.knotless.repo", "proj")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let main_ref = ref_rkey(&key, "refs/heads/main");

    let js = jetstream(&rt);
    let mut opts = runner_opts(&runner, &owner, &js);
    opts.pulls = true;
    rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
    // Watching starts where the refs and pulls are.
    let id = format!("{}/{key}", owner.account.did);
    wait_until(|| {
        let state = std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
        let state: serde_json::Value = serde_json::from_str(&state).unwrap_or_default();
        state["pulls"][&id].is_object()
    });

    let record = serde_json::json!({"repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key}});
    // A branch pushed, and a pull of it opened, each rung for.
    let open = |branch: &str, file: &str| {
        owner.git(&work, &["checkout", "-q", "-b", branch, "main"]);
        let sha = owner.commit_file(&work, file, &format!("{branch}\n"));
        owner.git(&work, &["push", "-q", "origin", branch]);
        let (code, pull) = owner.cli(&["pull", &url, "main", &url, branch, branch], b"");
        assert_eq!(code, 0, "{pull}");
        let pull = pull.trim().to_string();
        let ran = |wf: &str| check_on(&runner, &owner, &key, &pull, wf)["finishedAt"].is_string();
        wait_until(|| {
            js.ring(
                &owner.account.did,
                &ref_rkey(&key, &format!("refs/heads/{branch}")),
            );
            js.send(
                &owner.account.did,
                "dev.knotless.pull",
                "p",
                Some(record.clone()),
            );
            ran("web") && ran("all")
        });
        (pull, sha)
    };

    // A pull that changes only Rust passes `web` without running it, saying
    // why, its steps each skipped; what names no paths runs.
    let (rust, rust_sha) = open("rust", "crates/a.rs");
    let web = check_on(&runner, &owner, &key, &rust, "web");
    assert_eq!(web["status"], "success", "{web}");
    assert_eq!(web["commit"], rust_sha.as_str());
    assert_eq!(
        web["skipped"],
        "skipped: nothing changed in web/**, .knotless/workflows/web.yml"
    );
    let steps: Vec<_> = web["steps"].as_array().unwrap().iter().collect();
    assert_eq!(steps.len(), 2, "{web}");
    assert!(steps.iter().all(|s| s["status"] == "skipped"), "{web}");
    assert!(web["log"].is_null(), "{web}");
    let all = check_on(&runner, &owner, &key, &rust, "all");
    assert_eq!(all["status"], "success", "{all}");
    assert!(all["skipped"].is_null() && all["log"].is_object(), "{all}");
    // A branch new to the runner is all change: a push filtered runs on it.
    let branch = check_on(&runner, &owner, &key, "refs/heads/rust", "branches");
    assert!(
        branch["skipped"].is_null() && branch["finishedAt"].is_string(),
        "{branch}"
    );

    // One that changes `web` runs it.
    let (pages, _) = open("pages", "web/other.svelte");
    let web = check_on(&runner, &owner, &key, &pages, "web");
    assert_eq!(web["status"], "success", "{web}");
    assert!(web["skipped"].is_null() && web["log"].is_object(), "{web}");
    assert_eq!(web["steps"][0]["status"], "success", "{web}");

    // The merge gate takes one left out as passed, and the CLI says why.
    let me = runner.account.handle.clone();
    assert_eq!(owner.cli(&["runners", &url, "add", &me], b"").0, 0);
    let (code, out) = owner.cli(&["checks", &url, "--pull", &rust, "--workflow", "web"], b"");
    assert_eq!(code, 0, "{out}");
    assert!(out.starts_with(&format!("success\tweb\t{rust}\t")), "{out}");
    assert!(
        out.trim_end()
            .ends_with("\tskipped: nothing changed in web/**, .knotless/workflows/web.yml"),
        "{out}"
    );
    let (code, merged) = owner.cli(&["merge", &url, &rust], b"");
    assert_eq!(code, 0, "{merged}");
    let tip = merged.lines().next().unwrap().to_string();

    // Main moving runs what a push to it runs, whatever it changed; a push
    // filtered to `web` is left out, as only Rust changed since.
    let of = |wf: &str| check_on(&runner, &owner, &key, "refs/heads/main", wf);
    wait_until(|| {
        js.ring(&owner.account.did, &main_ref);
        ["web", "branches"]
            .iter()
            .all(|wf| of(wf)["commit"] == tip.as_str() && of(wf)["finishedAt"].is_string())
    });
    let web = of("web");
    assert!(web["skipped"].is_null() && web["log"].is_object(), "{web}");
    assert_eq!(web["status"], "success", "{web}");
    let branches = of("branches");
    assert_eq!(branches["status"], "success", "{branches}");
    assert_eq!(branches["skipped"], "skipped: nothing changed in web/**");
}

/// What runs on a pull into main, in a repository whose merge queue a
/// test lands: it fails one that brings `bad`, unless `good` too, and
/// otherwise leaves
/// `go/ran.<commit>` and holds until `go/go` is there.
fn held(go: &std::path::Path) -> String {
    let go = go.display();
    format!(
        "on: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n  - run: |\n      test ! -f bad || test -f good\n      touch {go}/ran.$KNOTLESS_RUN_SHA\n      until [ -f {go}/go ]; do sleep 0.1; done\n",
    )
}

/// A repository whose merge queue a runner lands as its owner, and the
/// owner, in a checkout of it, opening pulls and queueing them.
struct Line {
    owner: Sandbox,
    url: String,
    work: std::path::PathBuf,
    go: std::path::PathBuf,
    /// The appview, and its following of the PDS.
    base: String,
    ears: tokio::task::JoinHandle<()>,
}

impl Line {
    fn new(rt: &tokio::runtime::Runtime) -> Line {
        Line::served(rt, None)
    }

    /// A [`Line`] whose appview is a gateway too if it names `runner`.
    fn served(rt: &tokio::runtime::Runtime, runner: Option<&str>) -> Line {
        let mut owner = Sandbox::new();
        let (base, ears) = serving(rt, &Sandbox::new(), runner);
        owner.set("KNOTLESS_APPVIEW", &base);
        let handle = owner.account.handle.clone();
        owner.set("KNOTLESS_IDENTIFIER", &handle);
        let url = owner.url("line");
        let go = owner.path("go");
        std::fs::create_dir_all(&go).unwrap();
        std::fs::write(go.join("go"), "").unwrap();
        let work = owner.work("line", 1);
        let flows = work.join(".knotless/workflows");
        std::fs::create_dir_all(&flows).unwrap();
        std::fs::write(flows.join("ci.yml"), held(&go)).unwrap();
        // Not a pull's, so no batch runs it.
        std::fs::write(flows.join("push.yml"), "steps: [{run: 'true'}]\n").unwrap();
        owner.git(&work, &["add", "."]);
        owner.git(&work, &["commit", "-q", "-m", "workflows"]);
        owner.git(&work, &["remote", "add", "origin", &url]);
        owner.git(&work, &["push", "-q", "origin", "main"]);
        let line = Line {
            owner,
            url,
            work,
            go,
            base,
            ears,
        };
        line.ok(&["runners", &line.url, "add", &handle]);
        line
    }

    fn ok(&self, args: &[&str]) -> String {
        let (code, out) = self.owner.cli(args, b"");
        assert_eq!(code, 0, "{args:?}: {out}");
        out
    }

    /// Commits `file` on `branch`, off main as it is now unless `on`.
    fn branch(&self, branch: &str, file: &str, text: &str, on: bool) {
        let o = &self.owner;
        o.git(&self.work, &["fetch", "-q", "origin"]);
        let from = if on { branch } else { "origin/main" };
        o.git(&self.work, &["checkout", "-q", "-B", branch, from]);
        o.commit_file(&self.work, file, text);
        o.git(&self.work, &["push", "-q", "-f", "origin", branch]);
    }

    /// A pull of a branch off main as it is now that writes `file`.
    fn pull(&self, branch: &str, file: &str, text: &str) -> String {
        self.branch(branch, file, text, false);
        self.ok(&["pull", &self.url, "main", &self.url, branch, branch])
    }

    /// Queues pulls, and waits until the queue as read has them all.
    fn queue(&self, pulls: &[&str]) {
        for pull in pulls {
            let out = self.ok(&["merge", &self.url, pull, "--queue"]);
            assert!(out.starts_with("queued into main: "), "{out}");
        }
        wait_until(|| {
            let line = self.ok(&["queue", &self.url]);
            pulls.iter().all(|p| line.contains(&format!("\t{p}\t")))
        });
    }

    fn main(&self) -> String {
        let refs = self.ok(&["refs", &self.url]);
        let main = refs.lines().find(|l| l.ends_with("\trefs/heads/main"));
        main.unwrap().split('\t').next().unwrap().to_string()
    }

    /// The subjects of main's newest `n` commits, newest first.
    fn log(&self, n: usize) -> Vec<String> {
        let o = &self.owner;
        o.git(&self.work, &["fetch", "-q", "origin"]);
        let count = format!("-{n}");
        let out = o.git(&self.work, &["log", &count, "--format=%s", "origin/main"]);
        out.lines().map(str::to_string).collect()
    }

    /// Waits until `knotless show` of `pull` says `what`.
    fn shows(&self, pull: &str, what: &str) -> String {
        wait_until(|| self.ok(&["show", &self.url, pull]).contains(what));
        self.ok(&["show", &self.url, pull])
    }

    /// The queue's batches' checks, newest first.
    fn batches(&self) -> Vec<serde_json::Value> {
        let out = self.ok(&["checks", &self.url, "--queue", "main", "--json"]);
        serde_json::from_str::<Vec<serde_json::Value>>(&out).unwrap()
    }

    /// Holds the runs of what is queued from here on, forgetting which
    /// ran before.
    fn hold(&self) {
        for ran in self.ran() {
            std::fs::remove_file(self.go.join(format!("ran.{ran}"))).unwrap();
        }
        std::fs::remove_file(self.go.join("go")).unwrap();
    }

    /// The commits whose runs have started since the last hold.
    fn ran(&self) -> Vec<String> {
        let names = std::fs::read_dir(&self.go).unwrap();
        let names = names.map(|e| e.unwrap().file_name().to_string_lossy().to_string());
        names
            .filter_map(|n| Some(n.strip_prefix("ran.")?.to_string()))
            .collect()
    }

    /// The check of the newest batch on `base`.
    fn batch_on(&self, base: &str) -> serde_json::Value {
        let all = self.batches().into_iter();
        let on = all.filter(|c| c["baseCommit"] == base);
        let newest = on.max_by_key(|c| c["startedAt"].as_str().unwrap().to_string());
        newest.unwrap()
    }

    fn release(&self) {
        std::fs::write(self.go.join("go"), "").unwrap();
    }

    fn opts(&self, state: &str) -> knotless_run::Options {
        let label = format!("{}/line", self.owner.account.handle);
        let mut opts = knotless_run::Options::new(vec![label]);
        opts.jetstream = "ws://127.0.0.1:9".into();
        opts.state = Some(self.owner.path(state));
        opts.work = Some(self.owner.path("work"));
        opts.queue = true;
        opts
    }

    /// A runner as the owner, taking batches of `batch` to `depth`.
    fn runner(&self, rt: &tokio::runtime::Runtime, batch: usize, depth: usize) -> Lander {
        self.lander(rt, "state.json", |opts| {
            (opts.batch, opts.depth) = (batch, depth);
        })
    }

    /// A runner as the owner, keeping its state in `state`, as `with`
    /// sets it up.
    fn lander(
        &self,
        rt: &tokio::runtime::Runtime,
        state: &str,
        with: impl FnOnce(&mut knotless_run::Options),
    ) -> Lander {
        let mut opts = self.opts(state);
        opts.pulls = true;
        with(&mut opts);
        let ctx = common::ctx(&self.owner);
        let (run, mut watches) = rt.block_on(knotless_run::Runner::start(opts, ctx)).unwrap();
        // Watching starts where the repository is.
        rt.block_on(run.reconcile(&watches[0])).unwrap();
        Lander {
            run,
            watch: watches.remove(0),
        }
    }
}

struct Lander {
    run: Arc<knotless_run::Runner>,
    watch: knotless_run::Watch,
}

impl Lander {
    /// A pass over the repository, with what is going still going.
    fn step(&self, rt: &tokio::runtime::Runtime) {
        rt.block_on(self.run.reconcile(&self.watch)).unwrap();
    }

    /// A pass, once what is going has ended.
    fn pass(&self, rt: &tokio::runtime::Runtime) {
        rt.block_on(self.run.idle());
        self.step(rt);
    }
}

/// A runner of a repository's pulls alone leaves its queue, and its
/// pushes, to the runner that lands it, and runs pulls as that one would:
/// main moving runs no open pull again until its head moves, and a pull
/// queued runs only with its batch.
#[test]
fn a_runner_of_pulls_alone_leaves_the_queue_be() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let url = q.url.clone();
    let l = q.lander(&rt, "pulls.json", |opts| {
        opts.queue = false;
        opts.pulls_only = opts.repos.clone();
    });
    let key = q.owner.record("dev.knotless.repo", "line")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let ci = |pull: &str| check_on(&q.owner, &q.owner, &key, pull, "ci");
    let lists = |pull: &str| {
        let rkey = pull.rsplit('/').next().unwrap();
        wait_until(|| q.ok(&["list", &url, "--pulls"]).contains(rkey));
    };
    let twice = || {
        l.pass(&rt);
        l.pass(&rt);
    };

    // A pull opened runs, at its head merged with main.
    let start = q.main();
    let a = q.pull("a", "a", "a\n").trim().to_string();
    lists(&a);
    twice();
    let ran = ci(&a);
    assert_eq!(ran["status"], "success", "{ran}");
    assert_eq!(ran["baseCommit"], start.as_str());

    // Main moving runs it no further, nor a push.
    q.branch("main", "m", "m\n", false);
    let moved = q.main();
    twice();
    assert_eq!(ci(&a), ran);
    let push = check_on(&q.owner, &q.owner, &key, "refs/heads/main", "push");
    assert!(push.is_null(), "{push}");

    // Its head moving runs it, merged with main as it is now.
    q.branch("a", "a2", "a2\n", true);
    twice();
    let again = ci(&a);
    assert_eq!(again["status"], "success", "{again}");
    assert_eq!(again["baseCommit"], moved.as_str());
    assert_ne!(again["commit"], ran["commit"]);

    // A pull queued is noted, and left to run with its batch.
    let b = q.pull("b", "b", "b\n").trim().to_string();
    q.queue(&[&b]);
    twice();
    let state = std::fs::read_to_string(q.owner.path("pulls.json")).unwrap();
    assert!(state.contains(&b), "{state}");
    assert!(ci(&b).is_null());
}

#[test]
fn a_merge_queue_lands_in_batches() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let url = q.url.clone();
    let l = q.runner(&rt, 8, 1);

    // A runner that may not push here leaves the queue be.
    let mut other = Sandbox::new();
    let theirs = other.account.handle.clone();
    other.set("KNOTLESS_IDENTIFIER", &theirs);
    let mut opts = q.opts("stranger.json");
    opts.state = Some(other.path("state.json"));
    let (stranger, watches) = rt
        .block_on(knotless_run::Runner::start(opts, common::ctx(&other)))
        .unwrap();

    // Three pulls land together, each a commit of its own, in the order
    // they were queued, once one run of the batch passes.
    assert_eq!(q.ok(&["queue", &url]), "the merge queue is empty");
    let start = q.main();
    // Two of them name an issue to close, which landing them closes.
    let issue = q.ok(&["issue", &url, "Needs a"]);
    let a = q.pull("a", "a", "a\n");
    q.branch("b", "b", "b\n", false);
    let fixes = format!("Fixes {issue}");
    let b = q.ok(&["pull", &url, "main", &url, "b", "b", "-m", &fixes]);
    q.branch("c", "c", "c\n", false);
    let c = q.ok(&["pull", &url, "main", &url, "c", "c", "-m", &fixes]);
    q.queue(&[&a, &b, &c]);
    rt.block_on(stranger.reconcile(&watches[0])).unwrap();
    assert!(!rt.block_on(stranger.noticed(&watches[0])).unwrap());
    l.pass(&rt);
    assert_eq!(q.main(), start);
    l.pass(&rt);
    assert_eq!(q.log(4), ["c", "b", "a", "workflows"]);
    let tip = q.main();
    let ran = &q.batches()[0];
    assert_eq!(ran["status"], "success", "{ran}");
    assert_eq!(
        (&ran["commit"], &ran["baseCommit"]),
        (&tip.as_str().into(), &start.as_str().into())
    );
    let pulls: Vec<&str> = (ran["batch"].as_array().unwrap().iter())
        .map(|b| b["pull"].as_str().unwrap())
        .collect();
    assert_eq!(pulls, [a.as_str(), b.as_str(), c.as_str()]);
    assert_eq!(ran["batch"][2]["commit"], tip.as_str());
    assert_eq!(ran["ref"], "refs/heads/main");
    assert_eq!(q.batches().len(), 1, "push.yml is not a pull's");
    for pull in [&a, &b, &c] {
        q.shows(pull, "merged pull");
    }
    let closed = q.shows(&issue, "closed issue");
    assert!(closed.contains(&format!("\nclosed by {b}")), "{closed}");
    // Their branches went with them, and none ran alone: the batch is
    // their run.
    let refs = q.ok(&["refs", &url]);
    assert!(!refs.contains("refs/heads/a"), "{refs}");
    let key = q.owner.record("dev.knotless.repo", "line")["key"]
        .as_str()
        .unwrap()
        .to_string();
    assert!(check_on(&q.owner, &q.owner, &key, &a, "ci").is_null());

    // One that conflicts with those ahead of it is ejected at once, naming
    // where; one that breaks the branch is found by halving the batch.
    let d = q.pull("d", "f", "d\n");
    let e = q.pull("e", "f", "e\n");
    let x = q.pull("x", "bad", "x\n");
    let g = q.pull("g", "g", "g\n");
    q.queue(&[&d, &e, &x, &g]);
    l.pass(&rt);
    let shown = q.shows(&e, "merge queue: ejected");
    assert!(
        shown.contains(&format!(
            "it conflicts in f with {d} (d), queued ahead of it on main at "
        )),
        "{shown}"
    );
    assert!(shown.contains(". Merge that branch into it"), "{shown}");
    // [d, x, g] fails, then [d, x]; [d] passes and lands, so x alone is
    // what broke it, and is ejected without running again; then g lands.
    for _ in 0..4 {
        l.pass(&rt);
    }
    let shown = q.shows(&x, "merge queue: ejected");
    assert!(shown.contains("Halving the batch found it"), "{shown}");
    assert!(shown.contains("\nlogs: knotless checks "), "{shown}");
    assert!(shown.contains(" --queue main --log"), "{shown}");
    assert!(!shown.contains("Their logs"), "{shown}");
    assert!(shown.contains(". Failed: ci (failure)"), "{shown}");
    // It names the check that failed, as the runner filed it.
    let listed = q.ok(&["queue", &url, "--json"]);
    let listed: Vec<serde_json::Value> = serde_json::from_str(&listed).unwrap();
    let ejected = listed.iter().find(|w| w["uri"] == x.as_str()).unwrap();
    let failed = &ejected["queue"]["checks"][0];
    assert_eq!(
        (&failed["workflow"], &failed["status"]),
        (&"ci".into(), &"failure".into())
    );
    let uris: Vec<serde_json::Value> = q.batches().iter().map(|c| c["uri"].clone()).collect();
    assert!(uris.contains(&failed["check"]), "{failed} {uris:?}");
    assert_eq!(q.log(3), ["g", "d", "c"]);
    let line = q.ok(&["queue", &url]);
    assert!(
        line.contains(&format!("-\tmain\tejected\t{x}\tx\n\t")),
        "{line}"
    );

    // One alone that fails is ejected at the head it was: here, as a
    // workflow it brings does not load.
    let y = q.pull("y", ".knotless/workflows/broken.yml", "steps: 3\n");
    q.queue(&[&y]);
    l.pass(&rt);
    l.pass(&rt);
    let shown = q.shows(&y, "merge queue: ejected");
    assert!(shown.contains("its batch failed its checks at"), "{shown}");
    assert!(shown.contains("broken (error)"), "{shown}");

    // One whose branch goes once queued has nothing to land; queued again,
    // it is turned away.
    let gone = q.pull("gone", "gone", "gone\n");
    q.queue(&[&gone]);
    q.owner.git(&q.work, &["push", "-q", "origin", ":gone"]);
    l.pass(&rt);
    q.shows(&gone, "has no branch gone");
    let (queued, _, err) = q.owner.knotless(&["merge", &url, &gone, "--queue"], b"");
    assert!(!queued && err.contains("has no branch gone"), "{err}");

    // While its batch runs, the queue says so. The branch moving any
    // other way drops the batch, which is built again on it.
    q.hold();
    let h = q.pull("h", "h", "h\n");
    q.queue(&[&h]);
    let first = q.main();
    l.step(&rt);
    wait_until(|| q.ok(&["queue", &url]).contains("1\tmain\tchecking at "));
    q.owner
        .git(&q.work, &["checkout", "-q", "-B", "main", "origin/main"]);
    q.owner.commit_file(&q.work, "m", "m\n");
    q.owner.git(&q.work, &["push", "-q", "origin", "main"]);
    l.step(&rt);
    q.release();
    l.pass(&rt);
    assert_eq!(q.log(2), ["h", "write m"]);
    let batches = q.batches();
    let dropped = batches
        .iter()
        .find(|c| c["baseCommit"] == first.as_str())
        .unwrap();
    assert_eq!(dropped["status"], "cancelled", "{dropped}");
    let log = blob(
        &q.owner.account.did,
        dropped["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.ends_with("==> cancelled as the merge queue dropped its batch\n"),
        "{log}"
    );

    // A pull whose head moves while its batch runs is not landed as it
    // was, but built again as it is.
    q.hold();
    let k = q.pull("k", "k", "k\n");
    q.queue(&[&k]);
    l.step(&rt);
    q.branch("k", "k2", "k2\n", true);
    q.release();
    l.pass(&rt);
    l.pass(&rt);
    assert_eq!(q.log(1), ["k"]);
    q.owner.git(&q.work, &["checkout", "-q", "origin/main"]);
    assert!(q.work.join("k2").exists());

    // One taken out of the queue while its batch runs is not landed.
    q.hold();
    let out = q.pull("out", "out", "out\n");
    q.queue(&[&out]);
    l.step(&rt);
    assert_eq!(
        q.ok(&["queue", &url, "--leave", &out]),
        "left the merge queue"
    );
    wait_until(|| !q.ok(&["queue", &url]).contains(&out));
    l.step(&rt);
    q.release();
    l.pass(&rt);
    assert_eq!(q.log(1), ["k"]);
    q.shows(&out, "open pull");

    // A queue into a branch that is gone waits.
    let side = ["push", "-q", "origin", "origin/main:refs/heads/side"];
    q.owner.git(&q.work, &side);
    q.branch("sidework", "s", "s\n", false);
    let into = q.ok(&["pull", &url, "side", &url, "sidework", "Into side"]);
    let out = q.ok(&["merge", &url, &into, "--queue"]);
    assert!(out.starts_with("queued into side: 1 of 1\n"), "{out}");
    wait_until(|| q.ok(&["queue", &url]).contains(&into));
    q.owner.git(&q.work, &["push", "-q", "origin", ":side"]);
    l.pass(&rt);
    q.shows(&into, "merge queue: queued");
}

/// A runner the repository names that may not push lands its queue
/// through the appview's gateway, which takes a batch from it only once the
/// owner has delegated writes there: until then, each batch it would land
/// is ejected, saying so, rather than checked and refused for ever.
#[test]
fn a_runner_named_lands_through_the_gateway() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut named = Sandbox::new();
    let handle = named.account.handle.clone();
    let q = Line::served(&rt, Some(&named.account.did));
    named.set("KNOTLESS_IDENTIFIER", &handle);
    named.set("KNOTLESS_APPVIEW", &q.base);
    // The gateway's did:web names localhost, served over http.
    named.set("KNOTLESS_DID_WEB_SCHEME", "http");
    let url = q.url.clone();
    q.ok(&["runners", &url, "add", &handle]);
    q.ok(&["runners", &url, "remove", &q.owner.account.handle]);
    let mut opts = q.opts("named.json");
    opts.state = Some(named.path("state.json"));
    opts.work = Some(named.path("work"));
    opts.pulls = true;
    let ctx = common::ctx(&named);
    let (run, mut watches) = rt.block_on(knotless_run::Runner::start(opts, ctx)).unwrap();
    let l = Lander {
        run,
        watch: watches.remove(0),
    };
    l.step(&rt);

    let start = q.main();
    let a = q.pull("a", "a", "a\n");
    q.queue(&[&a]);
    for _ in 0..3 {
        l.pass(&rt);
    }
    assert_eq!(q.main(), start);
    let shown = q.shows(&a, "merge queue: ejected");
    assert!(
        shown.contains("the gateway would not land its batch: "),
        "{shown}"
    );
    assert!(shown.contains("has not delegated writes"), "{shown}");
    assert!(
        shown.contains("under Delegated writes in their settings"),
        "{shown}"
    );
}

#[test]
fn a_merge_queue_waits_for_preflights() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let url = q.url.clone();
    // A pre-flight on main: it fails a pull that brings `nobuild`, and
    // holds one that brings `slow` until `fly` is there.
    let fly = q.go.join("fly");
    let first = format!(
        "queue: first\non: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n  - run: |\n      test ! -f nobuild\n      until [ ! -f slow ] || [ -f {} ]; do sleep 0.1; done\n",
        fly.display()
    );
    q.branch("main", ".knotless/workflows/first.yml", &first, false);
    let l = q.runner(&rt, 8, 1);
    // The newest run of a pull's pre-flight, by its status.
    let first_of = |pull: &str| {
        let args = ["checks", &url, "--pull", pull, "--workflow", "first"];
        let out = q.ok(&[&args[..], &["--json"]].concat());
        let all: Vec<serde_json::Value> = serde_json::from_str(&out).unwrap();
        all.first().cloned().unwrap_or_default()
    };
    let in_batch = |pull: &str| {
        (q.batches().iter())
            .flat_map(|c| c["batch"].as_array().cloned().unwrap_or_default())
            .any(|b| b["pull"] == pull)
    };

    // One that comes to conflict with main, one whose branch goes, one
    // that does not build, one held, and one that passes, queued in that
    // order: none has run. Queueing turns away the first two as they end
    // up, so main moves and the branch goes once they are in.
    let c = q.pull("c", "m", "c\n");
    let gone = q.pull("gone", "gone", "gone\n");
    let p = q.pull("p", "nobuild", "p\n");
    let s = q.pull("s", "slow", "s\n");
    let r = q.pull("r", "r", "r\n");
    q.queue(&[&c, &gone, &p, &s, &r]);
    q.branch("main", "m", "main\n", false);
    q.owner.git(&q.work, &["push", "-q", "origin", ":gone"]);
    let start = q.main();

    // The lander runs their pre-flights itself, as pulls' runs, and
    // batches none yet. The one that conflicts it does not run, nor the
    // one with no head: each is ejected as the batch is built, saying why.
    l.step(&rt);
    q.shows(&c, "it does not apply on main");
    q.shows(&gone, "has no branch gone");
    assert!(first_of(&c).is_null());
    wait_until(|| {
        let line = q.ok(&["queue", &url]);
        line.contains(&format!("\tmain\tpre-flight first\t{s}\t"))
    });
    wait_until(|| first_of(&p)["status"] == "failure" && first_of(&r)["status"] == "success");
    let ran = first_of(&r);
    assert_eq!(ran["queue"], "first", "{ran}");
    assert_eq!(ran["baseCommit"], start.as_str(), "{ran}");
    assert!(q.batches().is_empty(), "no batch has run");
    q.shows(&s, "Queued, 2nd of 3, running its checks");

    // One whose pre-flight failed is ejected without a batch, naming it;
    // one still running holds up none behind it, which goes on alone.
    l.step(&rt);
    let shown = q.shows(&p, "merge queue: ejected");
    assert!(
        shown.contains("its pre-flight failed at its head "),
        "{shown}"
    );
    assert!(
        shown.contains(&format!(
            "\nlogs: knotless checks {} --pull {p} --log",
            q.url
        )),
        "{shown}"
    );
    assert!(shown.contains(". Failed: first (failure)"), "{shown}");
    assert!(in_batch(&r) && !in_batch(&s) && !in_batch(&p));

    // Once it passes, it goes in too, after the one behind it landed.
    std::fs::write(&fly, "").unwrap();
    l.pass(&rt);
    assert_eq!(q.log(2), ["r", "write m"]);
    l.pass(&rt);
    assert_eq!(q.log(3), ["s", "r", "write m"]);
    assert!(!in_batch(&p));
}

/// Into a branch whose rule asks for reviews, a pull queued before it
/// did, and not reviewed as it asks, is ejected saying what it lacks; one
/// whose branch went is ejected as a batch is built.
#[test]
fn a_merge_queue_ejects_what_is_not_reviewed() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let url = q.url.clone();
    let l = q.runner(&rt, 8, 1);
    let a = q.pull("a", "a", "a\n");
    let gone = q.pull("gone", "gone", "gone\n");
    q.queue(&[&a, &gone]);
    q.owner.git(&q.work, &["push", "-q", "origin", ":gone"]);
    q.ok(&["protect", &url, "main", "--reviews", "1"]);
    l.step(&rt);
    let shown = q.shows(&a, "merge queue: ejected");
    assert!(
        shown.contains(&format!("{a} is not reviewed as main asks: reviews 0 of 1")),
        "{shown}"
    );
    q.shows(&gone, "has no branch gone");
}

/// Another runner's pre-flight only queued, waiting for a job there, is
/// run by the lander too; and whichever starts second leaves it to the
/// other's, taking its own check back rather than filing it cancelled.
#[test]
fn a_preflight_only_queued_elsewhere_is_run_here_once() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let url = q.url.clone();
    // A pre-flight that holds a pull that brings `slow` until `fly` is
    // there, and leaves `flew.<sha>` once it runs.
    let fly = q.go.join("fly");
    let first = format!(
        "queue: first\non: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n  - run: |\n      touch {go}/flew.$KNOTLESS_RUN_SHA\n      until [ ! -f slow ] || [ -f {fly} ]; do sleep 0.1; done\n",
        go = q.go.display(),
        fly = fly.display()
    );
    q.branch("main", ".knotless/workflows/first.yml", &first, false);
    // Another runner the repository names, whose checks are written here.
    let other = Sandbox::new();
    q.ok(&["runners", &url, "add", &other.account.handle]);
    let l = q.lander(&rt, "state.json", |opts| opts.jobs = 1);
    let key = q.owner.record("dev.knotless.repo", "line")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let repo = format!("at://{}/dev.knotless.repo/line", q.owner.account.did);

    // One held takes the one job.
    let s = q.pull("s", "slow", "s\n");
    q.queue(&[&s]);
    l.step(&rt);
    let flew = |sha: &str| q.go.join(format!("flew.{sha}")).exists();
    let tip = q.main();
    let s_head = q.owner.git(&q.work, &["rev-parse", "s"]);
    wait_until(|| flew(&s_head));

    // Another's run of the next one's is only queued: the lander runs it
    // too, and it waits for the job.
    let d = q.pull("d", "d", "d\n");
    let head = q.owner.git(&q.work, &["rev-parse", "d"]);
    let rkey = check::rkey(&q.owner.account.did, &key, &d, &head, Some(&tip), "first");
    let theirs = |status: &str| {
        let record = serde_json::json!({
            "$type": check::NSID,
            "repo": {"uri": repo, "key": key},
            "ref": "refs/heads/d", "commit": head, "baseCommit": tip, "pull": d,
            "queue": "first", "workflow": "first", "status": status,
            "startedAt": "2026-01-01T00:00:00.000Z",
        });
        other.put_record(check::NSID, &rkey, record);
    };
    theirs("queued");
    q.queue(&[&d]);
    l.step(&rt);
    let mine = || q.owner.record(check::NSID, &rkey);
    wait_until(|| mine()["status"] == "queued");
    q.shows(&d, "running its checks");

    // The other's starts first: given the job, the lander's is not run,
    // and its check goes, so the queue waits for the other's.
    theirs("running");
    std::fs::write(&fly, "").unwrap();
    rt.block_on(l.run.idle());
    assert!(mine().is_null(), "{}", mine());
    assert!(!flew(&head));
    l.step(&rt);
    q.shows(&d, "running its checks");

    // Once it passes, the pull goes into a batch.
    theirs("success");
    l.pass(&rt);
    l.pass(&rt);
    assert!(q.log(2).contains(&"d".to_string()), "{:?}", q.log(2));
}

#[test]
fn a_merge_queue_runs_ahead() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let l = q.runner(&rt, 2, 2);

    // While a batch runs, the next runs on it: both have started before
    // either lands. The second lands on the commit the first landed as.
    let start = q.main();
    q.hold();
    let (n1, n2, n3) = (
        q.pull("n1", "n1", "1\n"),
        q.pull("n2", "n2", "2\n"),
        q.pull("n3", "n3", "3\n"),
    );
    q.queue(&[&n1, &n2, &n3]);
    l.step(&rt);
    wait_until(|| q.ran().len() == 2);
    assert_eq!(q.main(), start);
    let first = q.batch_on(&start);
    let ahead = q.batch_on(first["commit"].as_str().unwrap());
    q.release();
    l.pass(&rt);
    assert_eq!(q.log(3), ["n3", "n2", "n1"]);
    assert_eq!(q.batches().len(), 2);
    let first = q.batch_on(&start);
    let landed = first["commit"].as_str().unwrap();
    let second = q.batch_on(landed);
    assert_eq!(second["status"], "success", "{second}");
    assert_eq!(second["startedAt"], ahead["startedAt"]);
    assert_eq!(second["commit"], q.main().as_str());
    // The measure of it: the second started before the first ended.
    let (began, ended) = (second["startedAt"].as_str(), first["finishedAt"].as_str());
    assert!(began < ended, "{second} {first}");

    // Nothing lands but on the very commit its batch was built on: not on
    // one the branch has moved on from, even under the batch's tip.
    let stale = knotless::queue::Stack {
        base: start.clone(),
        stacked: Vec::new(),
        conflicts: Vec::new(),
    };
    let url = knotless::url::RepoUrl::parse(&q.url).unwrap();
    let git = knotless::git::Git::new(q.work.join(".git"));
    let ctx = common::ctx(&q.owner);
    let landing = knotless::queue::land(&ctx, &url, &git, "main", &stale);
    let refused = rt.block_on(landing).unwrap_err().to_string();
    let (at, base) = (&q.main()[..12], &start[..12]);
    let said = format!("main is at {at}, not at {base}, where the batch was built");
    assert_eq!(refused, said);
    assert_eq!(q.log(1), ["n3"]);

    // A pull leaving the batch being checked drops the one ahead on it
    // too, cancelling its run, and what is left is built again.
    q.hold();
    let tip = q.main();
    let (o1, o2, o3) = (
        q.pull("o1", "o1", "1\n"),
        q.pull("o2", "o2", "2\n"),
        q.pull("o3", "o3", "3\n"),
    );
    q.queue(&[&o1, &o2, &o3]);
    l.step(&rt);
    wait_until(|| q.ran().len() == 2);
    let ahead = q.batch_on(q.batch_on(&tip)["commit"].as_str().unwrap());
    q.ok(&["queue", &q.url, "--leave", &o1]);
    wait_until(|| !q.ok(&["queue", &q.url]).contains(&o1));
    l.step(&rt);
    q.release();
    l.pass(&rt);
    assert_eq!(q.log(2), ["o3", "o2"]);
    q.shows(&o1, "open pull");
    let dropped = q.batches().into_iter().find(|c| c["uri"] == ahead["uri"]);
    let dropped = dropped.unwrap();
    assert_eq!(dropped["status"], "cancelled", "{dropped}");
    let did = &q.owner.account.did;
    let log = blob(did, dropped["log"]["ref"]["$link"].as_str().unwrap());
    assert!(
        log.ends_with("cancelled as the merge queue dropped its batch\n"),
        "{log}"
    );
    let rebuilt = q.batch_on(&tip);
    assert_eq!(rebuilt["batch"].as_array().unwrap().len(), 2, "{rebuilt}");

    // The branch moving any other way drops both, and they are built
    // again on it.
    q.hold();
    let (r1, r2, r3) = (
        q.pull("r1", "r1", "1\n"),
        q.pull("r2", "r2", "2\n"),
        q.pull("r3", "r3", "3\n"),
    );
    q.queue(&[&r1, &r2, &r3]);
    l.step(&rt);
    wait_until(|| q.ran().len() == 2);
    q.owner
        .git(&q.work, &["checkout", "-q", "-B", "main", "origin/main"]);
    q.owner.commit_file(&q.work, "moved", "moved\n");
    q.owner.git(&q.work, &["push", "-q", "origin", "main"]);
    let moved = q.main();
    l.step(&rt);
    q.release();
    l.pass(&rt);
    assert_eq!(q.log(4), ["r3", "r2", "r1", "write moved"]);
    let again = q.batch_on(&moved);
    assert_eq!(again["status"], "success", "{again}");
    let cancelled = q
        .batches()
        .into_iter()
        .filter(|c| c["status"] == "cancelled");
    assert_eq!(cancelled.count(), 4, "the two before, and these two");

    // One that conflicts with a batch ahead waits for it, rather than be
    // ejected for what may not land; once it has, it is.
    let (n4, n5, n6) = (
        q.pull("n4", "c", "4\n"),
        q.pull("n5", "n5", "5\n"),
        q.pull("n6", "c", "6\n"),
    );
    q.queue(&[&n4, &n5, &n6]);
    l.pass(&rt);
    q.shows(&n6, "merge queue: queued");
    l.pass(&rt);
    assert_eq!(q.log(2), ["n5", "n4"]);
    q.shows(&n6, "merge queue: ejected");

    // Halving goes no further once one suspected leaves: those left may
    // all be sound.
    let (m1, m2) = (q.pull("m1", "bad", "m1\n"), q.pull("m2", "m2", "m2\n"));
    q.queue(&[&m1, &m2]);
    l.pass(&rt);
    l.pass(&rt);
    q.ok(&["queue", &q.url, "--leave", &m1]);
    wait_until(|| !q.ok(&["queue", &q.url]).contains(&m1));
    l.pass(&rt);
    l.pass(&rt);
    assert_eq!(q.log(1), ["m2"]);

    // A batch that fails drops the one on it, cancelling its run, and
    // halving it does not run ahead. n9 brings good, so the batch ahead,
    // with n7's bad, holds rather than fail too.
    q.hold();
    let tip = q.main();
    let (n7, n8, n9) = (
        q.pull("n7", "bad", "7\n"),
        q.pull("n8", "n8", "8\n"),
        q.pull("n9", "good", "9\n"),
    );
    q.queue(&[&n7, &n8, &n9]);
    l.step(&rt);
    wait_until(|| q.batch_on(&tip)["status"] == "failure" && q.ran().len() == 1);
    let ahead = q.batch_on(q.batch_on(&tip)["commit"].as_str().unwrap());
    l.step(&rt);
    q.release();
    l.pass(&rt);
    let dropped = q.batches().into_iter().find(|c| c["uri"] == ahead["uri"]);
    let dropped = dropped.unwrap();
    assert_eq!(dropped["status"], "cancelled", "{dropped}");
    l.pass(&rt);
    q.shows(&n7, "merge queue: ejected");
    l.pass(&rt);
    assert_eq!(q.log(2), ["n9", "n8"]);
}

#[test]
fn a_merge_queue_runs_ahead_with_a_job_to_spare() {
    // With one job, the batch ahead waits for the one it is on to let it
    // go: it ranks as a pull's run, after the batch that lands next.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let l = q.lander(&rt, "state.json", |opts| {
        (opts.batch, opts.jobs) = (1, 1);
    });
    let start = q.main();
    q.hold();
    let (j1, j2) = (q.pull("j1", "j1", "1\n"), q.pull("j2", "j2", "2\n"));
    q.queue(&[&j1, &j2]);
    l.step(&rt);
    wait_until(|| q.ran().len() == 1);
    let first = q.batch_on(&start);
    let ahead = q.batch_on(first["commit"].as_str().unwrap());
    assert_eq!(ahead["status"], "queued", "{ahead}");
    std::thread::sleep(Duration::from_millis(500));
    assert_eq!(q.ran(), [first["commit"].as_str().unwrap()]);
    q.release();
    l.pass(&rt);
    assert_eq!(q.log(2), ["j2", "j1"]);
    let second = q.batch_on(first["commit"].as_str().unwrap());
    assert_eq!(second["uri"], ahead["uri"]);
    assert_eq!(second["status"], "success", "{second}");
}

#[test]
fn a_batch_passing_as_the_runner_stops_lands() {
    // A batch whose runs pass once the runner is stopping lands before it
    // stops, rather than be checked again when it starts: and nothing
    // else goes on, no batch after it started.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let l = q.runner(&rt, 1, 1);
    q.hold();
    let (s1, s2) = (q.pull("s1", "s1", "1\n"), q.pull("s2", "s2", "2\n"));
    q.queue(&[&s1, &s2]);
    l.step(&rt);
    wait_until(|| q.ran().len() == 1);
    let stopping = rt.spawn({
        let run = l.run.clone();
        async move { run.stop().await }
    });
    // Its stop begun, the batch's run passes.
    std::thread::sleep(Duration::from_millis(300));
    q.release();
    promptly(&rt, stopping);
    assert_eq!(q.log(2), ["s1", "workflows"]);
    q.shows(&s1, "merged pull");
    assert_eq!(q.batches().len(), 1);
}

#[test]
fn a_merge_queue_notices_what_was_missed() {
    // A pull queued is noticed from its writer's repo, though neither
    // Jetstream nor the appview hears of it, and lands; merged, it is
    // not landed again by a runner starting afresh.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let z = q.pull("z", "z", "z\n");
    wait_until(|| q.ok(&["list", &q.url]).contains(&z));
    let l = q.lander(&rt, "state.json", |opts| {
        opts.notice = Duration::from_millis(500);
    });
    q.ears.abort();
    let door = Arc::new(tokio::sync::Notify::new());
    let (run, watch) = (l.run.clone(), l.watch.clone());
    rt.spawn(knotless_run::notice(run, watch, door.clone()));
    let wait = Duration::from_secs(20);
    let rung = || rt.block_on(async { tokio::time::timeout(wait, door.notified()).await });
    // The first look rings, as nothing was seen before it.
    rung().unwrap();
    let out = q.ok(&["merge", &q.url, &z, "--queue"]);
    assert!(out.starts_with("queued into main: 1 of 1"), "{out}");
    rung().unwrap();
    l.pass(&rt);
    l.pass(&rt);
    assert_eq!(q.log(1), ["z"]);
    assert_eq!(q.ok(&["queue", &q.url]), "the merge queue is empty");
    let tip = q.main();
    let again = q.lander(&rt, "again.json", |_| ());
    again.pass(&rt);
    again.pass(&rt);
    assert_eq!(q.main(), tip);
}

#[test]
fn a_merge_queue_goes_on_by_itself() {
    // Queued, a pull lands with nobody waiting: each batch's runs ending
    // ring the repository's door, and its queue goes on.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let z = q.pull("z", "z", "z\n");
    q.queue(&[&z]);
    // Each pass over it that runs past a few milliseconds says where it is.
    let mut opts = q.opts("state.json");
    opts.watchdog = Duration::from_millis(5);
    rt.spawn(knotless_run::run(opts, common::ctx(&q.owner)));
    q.shows(&z, "merged pull");
    assert_eq!(q.log(1), ["z"]);
}

#[test]
fn a_merge_queue_goes_first() {
    // Landing the queue, a pull not in it runs as its head moves, after
    // the queue's runs, and not again each time the queue moves main.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let l = q.runner(&rt, 8, 1);
    let key = q.owner.record("dev.knotless.repo", "line")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let ran = |p: &str| check_on(&q.owner, &q.owner, &key, p, "ci");
    let p = q.pull("p", "p", "p\n");
    l.pass(&rt);
    l.pass(&rt);
    let first = ran(&p);
    assert_eq!(first["status"], "success", "{first}");
    // What it lands brings a workflow pushes to main run too: the push
    // that lands the batch takes the batch's pass.
    let both = "on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n";
    let z = q.pull("z", ".knotless/workflows/both.yml", both);
    q.queue(&[&z]);
    l.pass(&rt);
    l.pass(&rt);
    assert_eq!(q.log(1), ["z"]);
    l.pass(&rt);
    assert_eq!(ran(&p)["startedAt"], first["startedAt"]);
    let pushed = check_on(&q.owner, &q.owner, &key, "refs/heads/main", "both");
    assert_eq!(
        (&pushed["status"], &pushed["commit"]),
        (&"success".into(), &q.main().into())
    );
    let log = blob(
        &q.owner.account.did,
        pushed["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(log.contains("as the merge queue's batch at "), "{log}");
    q.branch("p", "p2", "p2\n", true);
    l.pass(&rt);
    l.pass(&rt);
    let again = ran(&p);
    assert_ne!(again["commit"], first["commit"], "{again}");
    assert_eq!(again["status"], "success", "{again}");
}

#[test]
fn a_workflow_asked_for_by_name() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let mine = owner.account.handle.clone();
    owner.set("KNOTLESS_IDENTIFIER", &mine);
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let work = owner.work("proj", 1);
    let url = owner.url("proj");
    owner.git(&work, &["remote", "add", "origin", &url]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    assert_eq!(owner.cli(&["runners", &url, "add", &handle], b"").0, 0);
    let label = format!("{mine}/proj");
    let secrets = runner.path("secrets.json");
    let granted = serde_json::json!({ label.clone(): {
        "MAIN_ONLY": {"value": "m4inonly", "refs": ["refs/heads/main"]},
    } });
    std::fs::write(&secrets, granted.to_string()).unwrap();
    let opts = knotless_run::Options {
        jetstream: "ws://127.0.0.1:9".into(),
        repos: vec![label],
        pulls_only: Vec::new(),
        discover: false,
        executor: Executor::Shell,
        secrets: Some(secrets),
        secrets_key: None,
        state: Some(runner.path("state.json")),
        work: Some(runner.path("work")),
        // Shorter than a step of `costly`, which only a run asked for may
        // take past it.
        step_timeout: Duration::from_secs(2),
        run_timeout: None,
        jobs: 4,
        pull_jobs: 2,
        owner_jobs: 0,
        allowance: None,
        private_allowance: None,
        unmetered: Vec::new(),
        trust: Vec::new(),
        cache_max: 1 << 30,
        disk_floor: 0,
        caches: Some(runner.path("caches")),
        sweep: Duration::from_secs(600),
        stop_wait: Duration::from_secs(60),
        pulls: false,
        queue: false,
        batch: 8,
        depth: 1,
        notice: Duration::from_secs(30),
        poll: Duration::from_secs(10),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
    };
    let start = || {
        let started = knotless_run::Runner::start(opts.clone(), common::ctx(&runner));
        let (run, mut watches) = rt.block_on(started).unwrap();
        (run, watches.remove(0))
    };
    let (run, watch) = start();
    let key = watch.key.clone();
    rt.block_on(run.reconcile(&watch)).unwrap();

    // `ci` runs on every push; `costly` and `wait` only when asked for at
    // main, and `capped` on a push, its own timeout held to the runner's.
    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let go = go.display();
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    for (file, text) in [
        (
            "ci.yml",
            format!("steps: [{{run: 'echo $KNOTLESS_RUN_REF >> {go}/ci'}}]\n"),
        ),
        (
            "costly.yml",
            "on: {asked: {refs: [refs/heads/main]}}\nsecrets: [MAIN_ONLY]\nenv: {TRACKS: all, KEPT: 'yes'}\nsteps:\n  - name: show\n    run: echo \"tracks=$TRACKS kept=$KEPT ref=$KNOTLESS_RUN_REF main=$MAIN_ONLY\"\n  - name: long\n    timeout: 1\n    run: sleep 3\n".into(),
        ),
        (
            "capped.yml",
            "on: {push: {refs: [refs/heads/main]}}\nsteps: [{run: sleep 3, timeout: 1}]\n".into(),
        ),
        (
            "wait.yml",
            format!("on: {{asked: {{refs: [refs/heads/main]}}}}\nsteps:\n  - run: |\n      echo $TRACKS >> {go}/waits\n      until [ -f {go}/$TRACKS ]; do sleep 0.1; done\n    timeout: 1\n"),
        ),
    ] {
        std::fs::write(dir.join(file), text).unwrap();
    }
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["push", "-q", "origin", "main", "HEAD:other"]);
    let sha = owner.git(&work, &["rev-parse", "HEAD"]);
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.block_on(run.idle());
    let ran =
        |name: &str| std::fs::read_to_string(runner.path("go").join(name)).unwrap_or_default();
    let of = |r: &str, wf: &str| check_on(&runner, &owner, &key, r, wf);
    let main = "refs/heads/main";
    assert_eq!(of(main, "ci")["status"], "success");
    let mut pushed: Vec<String> = ran("ci").lines().map(str::to_string).collect();
    pushed.sort();
    assert_eq!(pushed, ["refs/heads/main", "refs/heads/other"]);
    let capped = of(main, "capped");
    assert_eq!(capped["status"], "failure", "{capped}");
    assert!(capped["steps"][0]["exitCode"].is_null(), "{capped}");
    let log = blob(
        &runner.account.did,
        capped["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(log.contains("==> killed after 2s"), "{log}");
    // A push runs none only asked for.
    assert!(of(main, "costly").is_null());
    assert!(of(main, "wait").is_null());

    // Asked for, only it runs, with the variables given over its own, and
    // its secrets; none named as the runner's own or as a secret.
    let asked = serde_json::json!({
        "repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key},
        "ref": main,
        "workflow": "costly",
        "env": [
            {"name": "TRACKS", "value": "team"},
            {"name": "KNOTLESS_RUN_REF", "value": "x"},
            {"name": "MAIN_ONLY", "value": "leak"},
        ],
        "createdAt": knotless::records::now(),
    });
    let rkey = format!("{key}.{}", knotless::keys::tid());
    owner.put_record("dev.knotless.rerun", &rkey, asked);
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.block_on(run.idle());
    let costly = of(main, "costly");
    assert_eq!(costly["status"], "success", "{costly}");
    assert_eq!(costly["commit"], sha.as_str());
    // Its long step past the runner's limit, as its own allows.
    assert_eq!(costly["steps"][1]["status"], "success", "{costly}");
    let log = blob(
        &runner.account.did,
        costly["log"]["ref"]["$link"].as_str().unwrap(),
    );
    assert!(
        log.starts_with("==> asked for by name TRACKS=team\n"),
        "{log}"
    );
    assert!(
        log.contains("tracks=team kept=yes ref=refs/heads/main main=***"),
        "{log}"
    );
    assert_eq!(ran("ci").lines().count(), 2);
    assert!(of(main, "wait").is_null());

    // One that runs on a push may be asked for by name too; one not there,
    // or not to be asked for at that ref, says so; a ref that is not there
    // is refused by the CLI, and, asked for anyway, as a branch since
    // deleted may be, forgotten.
    let rerun = |r: &str, wf: &str| {
        let (code, out) = owner.cli(&["rerun", &url, "--ref", r, "--workflow", wf], b"");
        assert_eq!(code, 0, "{out}");
    };
    rerun(main, "ci");
    rerun(main, "nope");
    rerun("refs/heads/other", "costly");
    let gone = [
        "rerun",
        &url,
        "--ref",
        "refs/heads/gone",
        "--workflow",
        "ci",
    ];
    // Refused, saying so on stderr, which `cli` does not keep.
    let (code, out) = owner.cli(&gone, b"");
    assert_eq!(code, 1, "{out}");
    let asked = serde_json::json!({
        "repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key},
        "ref": "refs/heads/gone",
        "workflow": "ci",
        "createdAt": knotless::records::now(),
    });
    let rkey = format!("{key}.{}", knotless::keys::tid());
    owner.put_record("dev.knotless.rerun", &rkey, asked);
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.block_on(run.idle());
    assert_eq!(ran("ci").lines().count(), 3);
    let nope = of(main, "nope");
    assert_eq!(nope["status"], "error", "{nope}");
    assert_eq!(
        nope["error"],
        format!(
            "asked for, but there is no workflow named nope at {}",
            &sha[..12]
        )
    );
    let other = of("refs/heads/other", "costly");
    assert_eq!(other["status"], "error", "{other}");
    assert_eq!(
        other["error"],
        "asked for, but it runs neither on a push to refs/heads/other nor when asked there"
    );
    let state = std::fs::read_to_string(runner.path("state.json")).unwrap();
    let state: serde_json::Value = serde_json::from_str(&state).unwrap();
    let id = format!("{}/{key}", owner.account.did);
    assert_eq!(state["wanted"][&id], serde_json::json!({}), "{state}");

    // A stop does not wait for one asked for: it is cancelled, and runs
    // again on the next start.
    let (code, out) = owner.cli(
        &[
            "rerun",
            &url,
            "--ref",
            main,
            "--workflow",
            "wait",
            "-e",
            "TRACKS=w1",
        ],
        b"",
    );
    assert_eq!(code, 0, "{out}");
    rt.block_on(run.reconcile(&watch)).unwrap();
    wait_until(|| ran("waits") == "w1\n");
    let stopping = rt.spawn({
        let run = run.clone();
        async move { run.stop().await }
    });
    promptly(&rt, stopping);
    let wait = of(main, "wait");
    assert_eq!(wait["status"], "cancelled", "{wait}");
    let (run, watch) = start();
    rt.block_on(run.reconcile(&watch)).unwrap();
    wait_until(|| ran("waits") == "w1\nw1\n");
    std::fs::write(runner.path("go").join("w1"), "").unwrap();
    rt.block_on(run.idle());
    assert_eq!(of(main, "wait")["status"], "success");
    // Once.
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.block_on(run.idle());
    assert_eq!(ran("waits"), "w1\nw1\n");
    assert_eq!(ran("ci").lines().count(), 3);
}

/// A private repository's runs, read in its space and filed in the
/// runner's repo there, nothing of them public: a push rings no
/// Jetstream, so the runner hears of it asking the space, and of the
/// repository from its invite. These need a PDS on the spaces alpha, and
/// pass trivially without one.
#[test]
fn a_private_repository_runs_in_its_space() {
    if !net().spaces {
        eprintln!("skipped: the PDS is not on the spaces alpha");
        return;
    }
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    // It reads as itself, logged in, as a space credential needs.
    let mut runner = Sandbox::new();
    runner.set("KNOTLESS_PASSWORD", "");
    let handle = runner.account.handle.clone();
    let password = runner.account.password.clone();
    let login = runner.cli(&["login", &handle, "--password-stdin"], password.as_bytes());
    assert_eq!(login.0, 0, "{}", login.1);
    // The appview names it for new repositories, and passes on invites.
    let base = serving(&rt, &Sandbox::new(), Some(&runner.account.did)).0;
    owner.set("KNOTLESS_APPVIEW", &base);
    runner.set("KNOTLESS_APPVIEW", &base);

    // Made private, it names the runner, a member, to read it and file
    // there, and invited.
    let url = owner.url("proj");
    let (code, out) = owner.cli(&["create", &url, "--private"], b"");
    assert_eq!(code, 0, "{out}");
    assert!(
        out.contains(&format!("{handle} reads it, to run its checks")),
        "{out}"
    );
    assert!(!out.contains("warning"), "{out}");
    let work = owner.work("proj", 1);
    std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
    std::fs::write(work.join(".knotless/workflows/fail.yaml"), FAIL).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["push", "-q", &url, "main"]);
    // One named once made is invited then.
    let later = owner.url("later");
    let made = owner.cli(&["create", &later, "--private", "--no-runner"], b"");
    assert_eq!(made.0, 0, "{}", made.1);
    let (code, out) = owner.cli(&["runners", &later, "add", &handle], b"");
    assert_eq!(code, 0, "{out}");
    assert_eq!(out, "added");
    // With no appview to tell, the runner is named and a member still.
    let alone = Sandbox::with(owner.account.clone());
    let lone = owner.url("lone");
    let (code, out) = alone.cli(&["create", &lone, "--private", "--runner", &handle], b"");
    assert_eq!(code, 0, "{out}");
    assert!(
        out.contains("was not told of it: invites go through an appview"),
        "{out}"
    );
    let (code, out) = alone.cli(&["runners", &lone, "add", &handle], b"");
    assert_eq!(code, 0, "{out}");
    assert!(out.starts_with("added\nwarning: "), "{out}");

    // Its account reads it.
    let (code, out) = runner.cli(&["refs", &url], b"");
    assert_eq!(code, 0, "{out}");
    let js = jetstream(&rt);
    let opts = knotless_run::Options {
        repos: Vec::new(),
        discover: true,
        sweep: Duration::from_secs(1),
        poll: Duration::from_secs(1),
        paused: None,
        keep_checks: None,
        prune_grace: Duration::ZERO,
        prune_every: Duration::ZERO,
        watchdog: Duration::from_secs(300),
        ..runner_opts(&runner, &owner, &js)
    };
    rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
    let state = || std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
    let did = &owner.account.did;
    wait_until(|| {
        let state = state();
        ["proj", "later"]
            .iter()
            .all(|n| state.contains(&format!("{did}/{n}")))
    });
    let read = || owner.cli(&["checks", &url, "--json"], b"").1;
    // Found, the commit that made it runs.
    wait_until(|| read().contains("\"finishedAt\""));

    // Pushed, with no ring: the space says something was written.
    let sha = owner.commit_file(&work, "f", "moved\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    wait_until(|| {
        read().contains(&format!("\"commit\": \"{sha}\"")) && !read().contains("\"running\"")
    });
    let checks: serde_json::Value = serde_json::from_str(&read()).unwrap();
    assert_eq!(checks[0]["status"], "failure", "{checks}");
    assert_eq!(checks[0]["commit"], sha.as_str());
    assert_eq!(checks[0]["runner"], runner.account.did.as_str());

    // Its log and pictures are read from the space too.
    let saved = owner.path("pictures");
    let dir = saved.to_str().unwrap();
    let (code, out) = owner.cli(&["checks", &url, "--log", "--images", dir], b"");
    assert_eq!(code, 0, "{out}");
    assert!(out.contains("==> break\nbroken\n"), "{out}");
    // Of both runs: the commit that made it, as found, and the push.
    let shown = out.lines().filter(|l| l.starts_with("image\t")).count();
    assert_eq!(shown, 64, "{out}");

    // Nothing of it is in the runner's public repo.
    let public = reqwest::blocking::get(format!(
        "{}/xrpc/com.atproto.repo.listRecords?repo={}&collection=dev.knotless.check",
        net().pds,
        runner.account.did
    ))
    .unwrap()
    .text()
    .unwrap();
    assert_eq!(public, r#"{"records":[]}"#);

    // Deleted, its space with it, it is left rather than asked forever.
    // The space is asked itself once a minute while the appview passes on
    // its writes, so this may take longer than a wait.
    let (code, out) = owner.cli(&["delete", &later, "--yes"], b"");
    assert_eq!(code, 0, "{out}");
    let until = std::time::Instant::now() + 3 * common::WAIT;
    while state().contains(&format!("{did}/later")) {
        assert!(std::time::Instant::now() < until, "never left: {}", state());
        std::thread::sleep(Duration::from_millis(500));
    }
    assert!(state().contains(&format!("{did}/proj")), "{}", state());
    // Its poll, waking, finds it left and ends.
    std::thread::sleep(3 * Duration::from_secs(1));
}

#[test]
fn a_dropped_pull_runs_on_a_base_moved_since() {
    // Landing the queue, a runner runs a pull again as its head moves,
    // not its base; but one whose run was dropped as the runner stopped
    // runs when it starts again, on the base as it is then.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let one = |opts: &mut knotless_run::Options| {
        opts.jobs = 1;
        opts.stop_wait = Duration::ZERO;
    };
    let l = q.lander(&rt, "state.json", one);
    q.hold();
    // One pull's run holds the only job; the other's waits for it.
    let a = q.pull("a", "a", "a\n");
    wait_until(|| {
        l.step(&rt);
        q.ran().len() == 1
    });
    let b = q.pull("b", "b", "b\n");
    let checks = |pull: &str| {
        let out = q.ok(&["checks", &q.url, "--pull", pull.trim(), "--json"]);
        serde_json::from_str::<Vec<serde_json::Value>>(&out).unwrap()
    };
    wait_until(|| {
        l.step(&rt);
        !checks(&b).is_empty()
    });
    rt.block_on(l.run.stop());
    let dropped = &checks(&b)[0];
    assert_eq!(dropped["status"], "cancelled", "{dropped}");
    drop(l);

    // Main moves on while the runner is down.
    q.branch("main", "elsewhere", "moved\n", true);
    let moved = q.main();
    q.release();
    let l = q.lander(&rt, "state.json", one);
    rt.block_on(l.run.idle());
    let ran = &checks(&b)[0];
    assert_eq!(ran["baseCommit"], moved.as_str(), "{ran}");
    assert_eq!(ran["status"], "success", "{ran}");
    // The one cut off, not dropped, is not run again until its head moves.
    assert_ne!(checks(&a)[0]["baseCommit"], moved.as_str());
}

#[test]
fn checks_of_no_more_use_are_pruned() {
    // Here a runner prunes on every pass, with no grace, keeping of main
    // only each workflow's newest.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let q = Line::new(&rt);
    let secrets = q.owner.path("secrets.json");
    let label = format!("{}/line", q.owner.account.handle);
    let granted = serde_json::json!({ label.clone(): {"TOKEN": "s3cret"}, "x.test/other": {"TOKEN": "kept"} });
    std::fs::write(&secrets, granted.to_string()).unwrap();
    let l = q.lander(&rt, "state.json", |opts| {
        opts.keep_checks = Some(Duration::ZERO);
        (opts.prune_grace, opts.prune_every) = (Duration::ZERO, Duration::ZERO);
        opts.secrets = Some(secrets.clone());
    });
    let checks = |by: &str, of: &str| {
        let out = q.ok(&["checks", &q.url, by, of, "--json"]);
        serde_json::from_str::<Vec<serde_json::Value>>(&out).unwrap()
    };
    let workflows = |found: &[serde_json::Value]| {
        let mut names: Vec<&str> = found
            .iter()
            .map(|c| c["workflow"].as_str().unwrap())
            .collect();
        names.sort();
        names.join(" ")
    };
    let a = q.pull("a", "a", "a\n");
    let b = q.pull("b", "b", "b\n");
    l.pass(&rt);
    assert_eq!(workflows(&checks("--pull", &a)), "ci");
    q.queue(&[&a]);
    l.pass(&rt);
    l.pass(&rt);
    assert_eq!(q.log(2), ["a", "workflows"]);
    // b's head moves, and runs again.
    q.branch("b", "b2", "b2\n", true);
    l.pass(&rt);
    l.pass(&rt);
    // The batch's run is gone, as is the merged pull's; the open one keeps
    // only its newest.
    assert!(q.batches().is_empty());
    assert!(checks("--pull", &a).is_empty());
    let of_b = checks("--pull", &b);
    assert_eq!(workflows(&of_b), "ci");
    let o = &q.owner;
    o.git(&q.work, &["fetch", "-q", "origin"]);
    let head = o.git(&q.work, &["rev-parse", "origin/b"]);
    assert_eq!(of_b[0]["commit"], head.trim(), "{of_b:?}");
    // A branch keeps its newest of each workflow, as main does here: its
    // pull's and batches' runs are theirs, not the branch's, and ci runs
    // only on those.
    assert_eq!(workflows(&checks("--ref", "refs/heads/b")), "push");
    let main = checks("--ref", "refs/heads/main");
    assert_eq!(workflows(&main), "push");
    assert!(
        main.iter().all(|c| c["commit"] == q.main().as_str()),
        "{main:?}"
    );

    let filed = || q.owner.records("dev.knotless.check").len();
    let before = filed();
    assert!(before > 0);
    // Deleted, the checks filed of it go, and the secrets granted to its
    // name, which a repository made under it later is not to have; it is
    // left, and a pass over it again finds nothing to do.
    q.ok(&["delete", &q.url, "--yes"]);
    l.pass(&rt);
    assert_eq!(filed(), 0);
    let left: serde_json::Value =
        serde_json::from_str(&std::fs::read_to_string(&secrets).unwrap()).unwrap();
    assert_eq!(left, serde_json::json!({"x.test/other": {"TOKEN": "kept"}}));
    l.step(&rt);
}

#[test]
fn a_run_asked_for_writes_back() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mine = owner.account.handle.clone();
    owner.set("KNOTLESS_IDENTIFIER", &mine);
    let work = owner.work("proj", 1);
    let url = owner.url("proj");
    let go = owner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    // `take` writes a file it may write back and one it may not; `held`
    // waits to be let go first.
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    let take = "on: {asked: {commit: [shots/**]}}\nsteps:\n  - run: mkdir -p shots && echo $SHOT > shots/a.txt && echo x > stray.txt\nenv: {SHOT: one}\n";
    std::fs::write(dir.join("take.yml"), take).unwrap();
    let held = format!(
        "on: {{asked: {{commit: [shots/**]}}}}\nsteps:\n  - run: |\n      touch {go}/started\n      until [ -f {go}/go ]; do sleep 0.1; done\n      mkdir -p shots && echo held > shots/a.txt\n",
        go = go.display()
    );
    std::fs::write(dir.join("held.yml"), held).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["remote", "add", "origin", &url]);
    owner.git(&work, &["push", "-q", "origin", "main", "main:topic"]);
    let ok = |args: &[&str]| {
        let (code, out) = owner.cli(args, b"");
        assert_eq!(code, 0, "{args:?}: {out}");
        out
    };
    ok(&["runners", &url, "add", &mine]);
    let label = format!("{mine}/proj");
    let mut opts = knotless_run::Options::new(vec![label]);
    opts.jetstream = "ws://127.0.0.1:9".into();
    opts.state = Some(owner.path("state.json"));
    opts.work = Some(owner.path("work"));
    let started = knotless_run::Runner::start(opts, common::ctx(&owner));
    let (run, mut watches) = rt.block_on(started).unwrap();
    let watch = watches.remove(0);
    let key = watch.key.clone();
    rt.block_on(run.reconcile(&watch)).unwrap();
    rt.block_on(run.idle());
    let tip = |branch: &str| {
        let refs = ok(&["refs", &url]);
        let line = refs
            .lines()
            .find(|l| l.ends_with(&format!("\trefs/heads/{branch}")));
        line.unwrap().split('\t').next().unwrap().to_string()
    };
    let asked = |r: &str, wf: &str| {
        ok(&["rerun", &url, "--ref", r, "--workflow", wf]);
        rt.block_on(run.reconcile(&watch)).unwrap();
    };
    let of = |r: &str, wf: &str| check_on(&owner, &owner, &key, r, wf);
    let log_of = |c: &serde_json::Value| {
        blob(
            &owner.account.did,
            c["log"]["ref"]["$link"].as_str().unwrap(),
        )
    };
    let topic = "refs/heads/topic";
    let was = tip("topic");

    // Asked for at a branch, what it wrote among its globs is committed
    // onto it, and only that.
    asked(topic, "take");
    rt.block_on(run.idle());
    let took = of(topic, "take");
    assert_eq!(took["status"], "success", "{took}");
    let now = tip("topic");
    assert_ne!(now, was);
    assert!(log_of(&took).contains(&format!("==> wrote back 1 file to topic as {}", &now[..12])));
    owner.git(&work, &["fetch", "-q", "origin"]);
    let parent = owner.git(&work, &["rev-parse", "origin/topic^"]);
    assert_eq!(parent, was);
    let files = owner.git(&work, &["diff", "--name-only", &was, "origin/topic"]);
    assert_eq!(files, "shots/a.txt");
    let said = owner.git(&work, &["log", "-1", "--format=%an %s", "origin/topic"]);
    assert_eq!(said, "knotless-run take: what it wrote, as asked");
    assert_eq!(tip("main"), was);

    // Asked again, it writes the same, so nothing.
    asked(topic, "take");
    rt.block_on(run.idle());
    let again = of(topic, "take");
    assert_eq!(again["status"], "success", "{again}");
    assert!(log_of(&again).contains("==> nothing to write back in shots/**"));
    assert_eq!(tip("topic"), now);

    // Never the default branch, nor a protected one.
    asked("refs/heads/main", "take");
    rt.block_on(run.idle());
    let main = of("refs/heads/main", "take");
    assert_eq!(main["status"], "failure", "{main}");
    let error = main["error"].as_str().unwrap();
    assert!(error.starts_with("not written back: main is "), "{error}");
    assert!(error.ends_with("'s default branch"), "{error}");
    // From before what was written, so there is something to write.
    let from = format!("{was}:refs/heads/kept");
    owner.git(&work, &["push", "-q", "origin", &from]);
    ok(&["protect", &url, "kept"]);
    asked("refs/heads/kept", "take");
    rt.block_on(run.idle());
    let kept = of("refs/heads/kept", "take");
    assert_eq!(
        kept["error"], "not written back: kept is protected",
        "{kept}"
    );

    // Nor over a push made while it ran.
    asked(topic, "held");
    wait_until(|| go.join("started").exists());
    owner.git(&work, &["checkout", "-q", "-B", "topic", "origin/topic"]);
    owner.commit_file(&work, "b.txt", "b\n");
    owner.git(&work, &["push", "-q", "origin", "topic"]);
    let pushed = owner.git(&work, &["rev-parse", "HEAD"]);
    std::fs::write(go.join("go"), "").unwrap();
    rt.block_on(run.idle());
    let held = of(topic, "held");
    assert_eq!(held["status"], "failure", "{held}");
    assert_eq!(
        held["error"],
        format!("not written back: topic has moved on from {}", &now[..12])
    );
    assert_eq!(tip("topic"), pushed);
}

#[test]
fn a_runner_finds_the_repositories_naming_it() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    // The appview names it for new repositories.
    let named = [runner.account.did.clone()];
    let base = serving(&rt, &Sandbox::new(), Some(&named[0])).0;
    owner.set("KNOTLESS_APPVIEW", &base);
    runner.set("KNOTLESS_APPVIEW", &base);
    let me = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &me);
    let did = owner.account.did.clone();
    let runners = |name: &str| owner.record("dev.knotless.repo", name)["runners"].clone();
    // A repository made names those, unless its owner names others.
    let made = owner.url("made");
    assert_eq!(owner.cli(&["create", &made], b"").0, 0);
    let dids = |r: serde_json::Value| -> Vec<String> {
        let r = r.as_array().cloned().unwrap_or_default();
        r.iter()
            .map(|r| r["did"].as_str().unwrap().to_string())
            .collect()
    };
    assert_eq!(dids(runners("made")), named);
    let own = [
        "create",
        &owner.url("own"),
        "--runner",
        &owner.account.handle,
    ];
    assert_eq!(owner.cli(&own, b"").0, 0);
    assert_eq!(dids(runners("own")), std::slice::from_ref(&did));
    assert_eq!(
        owner
            .cli(&["create", &owner.url("none"), "--no-runner"], b"")
            .0,
        0
    );
    assert!(dids(runners("none")).is_empty());
    // Each below is made by its first push, so names the runner too, and
    // has a workflow every push runs.
    let make = |name: &str| {
        let url = owner.url(name);
        let work = owner.work(name, 1);
        std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
        let flow = work.join(".knotless/workflows/ci.yml");
        std::fs::write(flow, "steps: [{run: 'true'}]\n").unwrap();
        owner.git(&work, &["add", "."]);
        owner.git(&work, &["commit", "-q", "-m", "workflows"]);
        owner.git(&work, &["remote", "add", "origin", &url]);
        owner.git(&work, &["push", "-q", "origin", "main"]);
        assert_eq!(dids(runners(name)), named);
        let record = owner.record("dev.knotless.repo", name);
        (work, record["key"].as_str().unwrap().to_string())
    };
    let state = || -> serde_json::Value {
        let text = std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
        serde_json::from_str(&text).unwrap_or_default()
    };
    let found = || -> Vec<String> {
        let state = state();
        let labels = state["discovered"].as_array().cloned().unwrap_or_default();
        labels
            .iter()
            .map(|l| l.as_str().unwrap().to_string())
            .collect()
    };
    let js = jetstream(&rt);
    // Once it finds one, the commit that made it runs, and then a push.
    let ran = |key: &str, sha: &str| {
        wait_until(|| {
            let check = check_of(&runner, &owner, key, "ci");
            check["commit"] == sha && check["status"] == "success"
        });
    };
    let runs = |work: &std::path::Path, key: &str| {
        ran(key, &owner.git(work, &["rev-parse", "HEAD"]));
        let sha = owner.commit_file(work, "f", "f\n");
        owner.git(work, &["push", "-q", "origin", "main"]);
        js.ring(&did, &ref_rkey(key, "refs/heads/main"));
        ran(key, &sha);
    };

    // Those named before it starts the appview lists, but one named on
    // the command line is watched as that.
    let (first, first_key) = make("first");
    make("both");
    let listing = format!(
        "{base}/xrpc/dev.knotless.listRepos?runner={}",
        runner.account.did
    );
    let listed = || reqwest::blocking::get(&listing).unwrap().text().unwrap();
    wait_until(|| {
        ["first", "both", "made"]
            .iter()
            .all(|n| listed().contains(&format!("\"{n}\"")))
    });
    let opts = knotless_run::Options {
        repos: vec![format!("{}/both", owner.account.handle)],
        discover: true,
        sweep: Duration::from_secs(1),
        notice: Duration::from_millis(200),
        ..runner_opts(&runner, &owner, &js)
    };
    rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
    let at = |names: &[&str]| -> Vec<String> {
        let mut labels: Vec<String> = names.iter().map(|n| format!("{did}/{n}")).collect();
        labels.sort();
        labels
    };
    wait_until(|| found() == at(&["first", "made"]));
    runs(&first, &first_key);

    // One named once it has started is found as its record is written; a
    // record of no repository is passed over.
    let (second, second_key) = make("second");
    let record = owner.record("dev.knotless.repo", "second");
    js.send(&did, "dev.knotless.repo", "second", Some(record));
    js.send(
        &did,
        "dev.knotless.repo",
        "nothing",
        Some(serde_json::json!({})),
    );
    wait_until(|| found().contains(&format!("{did}/second")));
    runs(&second, &second_key);

    // Renamed, it is watched by its new name.
    let (code, out) = owner.cli(&["rename", &owner.url("first"), "renamed"], b"");
    assert_eq!(code, 0, "{out}");
    let record = owner.record("dev.knotless.repo", "renamed");
    js.send(&did, "dev.knotless.repo", "renamed", Some(record));
    wait_until(|| {
        let found = found();
        found.contains(&format!("{did}/renamed")) && !found.contains(&format!("{did}/first"))
    });

    // No longer naming it, it is left.
    let (code, out) = owner.cli(&["runners", &owner.url("second"), "remove", &me], b"");
    assert_eq!(code, 0, "{out}");
    let record = owner.record("dev.knotless.repo", "second");
    js.send(&did, "dev.knotless.repo", "second", Some(record));
    wait_until(|| found() == at(&["made", "renamed"]));
    // Its loops end with it, each at its next look.
    std::thread::sleep(Duration::from_millis(500));
}

/// The server-sent events at `url`, gathered as they come, from
/// `last` if it says where the last were had to.
fn listen(url: &str, last: Option<&str>) -> Arc<std::sync::Mutex<String>> {
    use std::io::Read;
    let client = reqwest::blocking::Client::builder()
        .timeout(None)
        .build()
        .unwrap();
    let mut req = client.get(url);
    if let Some(last) = last {
        req = req.header("last-event-id", last);
    }
    let mut res = req.send().unwrap();
    assert_eq!(res.status(), 200, "{url}");
    let heard = Arc::new(std::sync::Mutex::new(String::new()));
    let gathered = heard.clone();
    std::thread::spawn(move || {
        let mut buf = [0u8; 4096];
        while let Ok(n @ 1..) = res.read(&mut buf) {
            let more = String::from_utf8_lossy(&buf[..n]);
            gathered.lock().unwrap().push_str(&more);
        }
    });
    heard
}

/// What a run prints shows at the appview as it prints it, masked as its
/// log is, for whoever may read its repository, and its steps as each
/// ends; its repository's pages hear of each write about it. A page that
/// lost its place goes on from it.
#[test]
fn a_run_shows_as_it_runs() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let base = serving(&rt, &Sandbox::new(), Some(&runner.account.did)).0;
    owner.set("KNOTLESS_APPVIEW", &base);
    runner.set("KNOTLESS_APPVIEW", &base);
    let go = runner.path("go");
    std::fs::create_dir_all(&go).unwrap();
    let flow = format!(
        "secrets: [TOKEN]\nsteps:\n  - name: first\n    run: |\n      echo \"first $TOKEN\"\n      until [ -f {go}/go ]; do sleep 0.1; done\n  - name: second\n    run: echo second\n",
        go = go.display()
    );
    let work = owner.work("proj", 1);
    std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
    std::fs::write(
        work.join(".knotless/workflows/live.yml"),
        "steps: [{run: 'true'}]\n",
    )
    .unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "workflows"]);
    owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let key = owner.record("dev.knotless.repo", "proj")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let ref_key = ref_rkey(&key, "refs/heads/main");
    let js = jetstream(&rt);
    rt.spawn(knotless_run::run(
        runner_opts(&runner, &owner, &js),
        common::ctx(&runner),
    ));
    wait_until(|| runner.path("state.json").exists());

    // Its pages follow it; a push is heard of.
    let repo = format!("at://{}/dev.knotless.repo/proj", owner.account.did);
    let changed = listen(&format!("{base}/_live/repo?repo={repo}"), None);
    std::fs::write(work.join(".knotless/workflows/live.yml"), flow).unwrap();
    owner.git(&work, &["commit", "-q", "-am", "it holds"]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    js.ring(&owner.account.did, &ref_key);
    // With data, as a browser hears no event without.
    wait_until(|| {
        changed
            .lock()
            .unwrap()
            .contains("event: changed\ndata: null")
    });

    // Running, what it printed shows, masked; and not before.
    let running = || {
        let checks = runner.records(check::NSID);
        let live = checks
            .iter()
            .find(|(_, c)| c["workflow"] == "live" && c["status"] == "running");
        live.map(|(k, _)| format!("at://{}/{}/{k}", runner.account.did, check::NSID))
    };
    wait_until(|| running().is_some());
    let uri = running().unwrap();
    // Once the appview has it, as it shows only the checks it has.
    let shown = format!("{base}/xrpc/dev.knotless.getCheck?repo={repo}&uri={uri}");
    wait_until(|| reqwest::blocking::get(&shown).unwrap().status() == 200);
    let follow = format!("{base}/_live/check?repo={repo}&uri={uri}");
    let said = listen(&follow, None);
    wait_until(|| said.lock().unwrap().contains(r#"first ***\n"#));
    // The step going is named, as its `==>` line may be gone from the tail.
    wait_until(|| {
        said.lock()
            .unwrap()
            .contains("event: going\ndata: \"first\"")
    });
    assert!(!said.lock().unwrap().contains("event: ended"));
    std::fs::write(go.join("go"), "").unwrap();
    wait_until(|| said.lock().unwrap().contains("event: ended\ndata: null"));
    let said = said.lock().unwrap().clone();
    assert!(!said.contains("s3cret"), "{said}");
    assert!(said.contains("event: steps"), "{said}");
    assert!(said.contains(r#""name":"second""#), "{said}");
    // Each piece is named by where it ends: a page that had it all is sent
    // only that it ended.
    let last = said
        .lines()
        .rev()
        .find_map(|l| l.strip_prefix("id: "))
        .unwrap();
    let again = listen(&follow, Some(last));
    wait_until(|| again.lock().unwrap().contains("event: ended"));
    assert!(!again.lock().unwrap().contains("event: output"));

    // A check its repository's runners did not file is no one's to follow.
    let other = format!("at://{}/{}/x", owner.account.did, check::NSID);
    let status = |url: String| reqwest::blocking::get(url).unwrap().status().as_u16();
    assert_eq!(
        status(format!("{base}/_live/check?repo={repo}&uri={other}")),
        404
    );
    // Nor does a runner send another's.
    let did = reqwest::blocking::get(format!("{base}/.well-known/did.json"))
        .unwrap()
        .json::<serde_json::Value>()
        .unwrap()["id"]
        .as_str()
        .unwrap()
        .to_string();
    let ctx = common::ctx(&runner);
    let token = rt.block_on(async {
        let me = knotless::session::me(&ctx).await.unwrap();
        knotless::auth::service_token(&ctx, &me, &did, "dev.knotless.putRunOutput")
            .await
            .unwrap()
    });
    let sent = reqwest::blocking::Client::new()
        .post(format!("{base}/xrpc/dev.knotless.putRunOutput"))
        .bearer_auth(&token)
        .json(&serde_json::json!({ "check": other, "offset": 0, "text": "forged" }))
        .send()
        .unwrap();
    assert_eq!(sent.status(), 400);

    // Run again, as a rerun at the same commit is, the check's output
    // starts anew rather than showing the last run's.
    let anew = serde_json::json!({
        "check": uri, "startedAt": "2099-01-01T00:00:00.000Z", "offset": 0, "text": "anew\n"
    });
    let sent = reqwest::blocking::Client::new()
        .post(format!("{base}/xrpc/dev.knotless.putRunOutput"))
        .bearer_auth(&token)
        .json(&anew)
        .send()
        .unwrap();
    assert_eq!(sent.status(), 200);
    let fresh = listen(&follow, None);
    wait_until(|| fresh.lock().unwrap().contains(r#""anew\n""#));
    assert!(
        !fresh.lock().unwrap().contains("first"),
        "{}",
        fresh.lock().unwrap()
    );
}

/// A step another repository shares: it takes inputs, needs a command, may
/// be given one secret, and runs a script kept beside it.
const GREET: &str = r#"
description: Says who it greets.
inputs:
  who: {required: true}
  how: {default: hello}
needs: [sh]
secrets: [TOKEN]
steps:
  - name: say
    run: |
      echo "$INPUT_HOW $INPUT_WHO token=${TOKEN:-none} other=${OTHER:-none}"
  - name: script
    run: sh "$KNOTLESS_STEP_DIR/greet.sh"
"#;

#[test]
fn a_workflow_uses_a_shared_step() {
    let owner = Sandbox::new();
    let mut runner = Sandbox::new();
    let handle = runner.account.handle.clone();
    runner.set("KNOTLESS_IDENTIFIER", &handle);
    let h = owner.account.handle.clone();

    // The shared steps, in a repository of their own, at the commit the
    // lock pins.
    let steps = owner.work("steps", 1);
    let files = steps.join(".knotless/steps");
    std::fs::create_dir_all(&files).unwrap();
    let write = |name: &str, text: &str| std::fs::write(files.join(name), text).unwrap();
    write("greet.yml", GREET);
    std::fs::write(steps.join("greet.sh"), "echo a script of its own\n").unwrap();
    // One that uses another, passing on what it is given; one that uses
    // itself; and a chain past the deepest.
    let outer = format!(
        "secrets: [TOKEN]\nsteps:\n  - {{uses: {h}/steps/greet@1, with: {{who: nested}}, secrets: [TOKEN]}}\n"
    );
    write("outer.yml", &outer);
    // One asking for a secret by a name of its own, which says how long
    // what it sees is, as the log hides the value; and one passing it a
    // secret of its own under that name.
    write(
        "keyed.yml",
        "secrets: [KEY]\nsteps:\n  - run: echo \"key=${KEY:-none} len=${#KEY} token=${TOKEN:-none} other=${OTHER:-none}\"\n",
    );
    let renaming = format!(
        "secrets: [INNER]\nsteps:\n  - {{uses: {h}/steps/keyed@1, secrets: {{KEY: INNER}}}}\n"
    );
    write("renaming.yml", &renaming);
    // One with a step of its own for main alone.
    write(
        "gated.yml",
        "steps:\n  - {name: any, run: echo inner any}\n  - {name: main, run: echo inner main, on: {refs: [refs/heads/main]}}\n",
    );
    write(
        "loop.yml",
        &format!("steps: [{{uses: {h}/steps/loop@1}}]\n"),
    );
    for d in 1..=5 {
        let next = format!("steps: [{{uses: {h}/steps/d{}@1}}]\n", d + 1);
        write(&format!("d{d}.yml"), &next);
    }
    owner.git(&steps, &["add", "."]);
    owner.git(&steps, &["commit", "-q", "-m", "steps"]);
    let pinned = owner.git(&steps, &["rev-parse", "HEAD"]);
    owner.git(&steps, &["remote", "add", "origin", &owner.url("steps")]);
    owner.git(&steps, &["push", "-q", "origin", "main"]);
    // What it says later never runs where the lock pins the earlier.
    let later = "steps: [{run: exit 9}]\n";
    owner.commit_file(&steps, ".knotless/steps/greet.yml", later);
    owner.git(&steps, &["push", "-q", "origin", "main"]);

    let uses = format!("{h}/steps/greet@1");
    let work = owner.work("proj", 1);
    let dir = work.join(".knotless/workflows");
    std::fs::create_dir_all(&dir).unwrap();
    let on = "on: {push: {refs: [refs/heads/main]}}\nsecrets: [TOKEN, OTHER]\n";
    let shared = format!(
        "{on}steps:\n  - run: echo \"own token=${{TOKEN:-none}} other=${{OTHER:-none}}\"\n  - uses: {uses}\n    with: {{who: world}}\n    secrets: [TOKEN]\n  - {{name: quiet, uses: {uses}, with: {{who: you, how: hi}}}}\n"
    );
    std::fs::write(dir.join("shared.yml"), shared).unwrap();
    // Steps for some refs: a shared step's `on` holds for each of its
    // steps, beside their own.
    let gated = format!(
        "on: {{push: {{refs: [refs/heads/main, refs/heads/x]}}}}\nsteps:\n  - run: echo own any\n  - {{name: own main, run: echo own main, on: {{refs: [refs/heads/main]}}}}\n  - {{uses: {uses}, with: {{who: x}}, on: {{refs: [refs/heads/x]}}}}\n  - {{uses: {h}/steps/gated@1, on: {{refs: [refs/heads/*]}}}}\n"
    );
    std::fs::write(dir.join("gated.yml"), gated).unwrap();
    let using =
        |step: &str, more: &str| format!("{on}steps:\n  - {{uses: {h}/steps/{step}{more}}}\n");
    let workflows = [
        ("unpinned", using("other@1", "")),
        ("newer", using("greet@2", ", with: {who: x}")),
        ("gone", using("gone@1", "")),
        ("nested", using("outer@1", ", secrets: [TOKEN]")),
        ("looped", using("loop@1", "")),
        ("deep", using("d1@1", "")),
        ("renamed", using("keyed@1", ", secrets: {KEY: OTHER}")),
        (
            "renamings",
            using("renaming@1", ", secrets: {INNER: TOKEN}"),
        ),
        (
            "unrenamed",
            format!(
                "on: {{push: {{refs: [refs/heads/main]}}}}\nsecrets: [MISSING]\nsteps:\n  - {{uses: {h}/steps/keyed@1, secrets: {{KEY: MISSING}}}}\n"
            ),
        ),
    ];
    for (name, text) in &workflows {
        std::fs::write(dir.join(format!("{name}.yml")), text).unwrap();
    }
    let mut lock = String::new();
    let pin = |lock: &mut String, step: &str, version: &str| {
        let line = format!("{h}/steps/{step}: {{version: {version}, commit: {pinned}}}\n");
        lock.push_str(&line);
    };
    pin(&mut lock, "greet@1", "1.0.0");
    pin(&mut lock, "greet@2", "1.0.0");
    pin(&mut lock, "gone@1", "1.1.0");
    let pinned_steps = [
        "outer", "gated", "loop", "d1", "d2", "d3", "d4", "d5", "keyed", "renaming",
    ];
    for step in pinned_steps {
        pin(&mut lock, &format!("{step}@1"), "1.0.0");
    }
    std::fs::write(work.join(".knotless/uses.lock"), lock).unwrap();
    owner.git(&work, &["add", "."]);
    owner.git(&work, &["commit", "-q", "-m", "shared steps"]);
    owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let key = owner.record("dev.knotless.repo", "proj")["key"]
        .as_str()
        .unwrap()
        .to_string();
    let ref_key = ref_rkey(&key, "refs/heads/main");

    let rt = tokio::runtime::Runtime::new().unwrap();
    let js = jetstream(&rt);
    let opts = runner_opts(&runner, &owner, &js);
    rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
    wait_until(|| runner.path("state.json").exists());
    owner.commit_file(&work, "f", "again");
    owner.git(&work, &["push", "-q", "origin", "main"]);
    js.ring(&owner.account.did, &ref_key);
    let all = [
        "shared",
        "unpinned",
        "newer",
        "gone",
        "nested",
        "looped",
        "deep",
        "gated",
        "renamed",
        "renamings",
        "unrenamed",
    ];
    wait_until(|| {
        all.iter()
            .all(|wf| check_of(&runner, &owner, &key, wf)["finishedAt"].is_string())
    });

    let ran = check_of(&runner, &owner, &key, "shared");
    assert_eq!(ran["status"], "success", "{ran}");
    let names: Vec<&str> = ran["steps"]
        .as_array()
        .unwrap()
        .iter()
        .map(|s| s["name"].as_str().unwrap())
        .collect();
    assert_eq!(
        names,
        [
            "step 1",
            "greet / needs",
            "greet / say",
            "greet / script",
            "quiet / needs",
            "quiet / say",
            "quiet / script",
        ]
    );
    let log_of = |check: &serde_json::Value| {
        blob(
            &runner.account.did,
            check["log"]["ref"]["$link"].as_str().unwrap(),
        )
    };
    let log = log_of(&ran);
    // Its own step sees every secret granted; a shared one only those it
    // is given, and none it is not.
    assert!(log.contains("own token=*** other=***"), "{log}");
    assert!(log.contains("hello world token=*** other=none"), "{log}");
    assert!(log.contains("hi you token=none other=none"), "{log}");
    assert!(log.contains("a script of its own"), "{log}");

    // A shared step's own shared steps, given what it passes on.
    let nested = check_of(&runner, &owner, &key, "nested");
    assert_eq!(nested["status"], "success", "{nested}");
    assert_eq!(nested["steps"][1]["name"], "outer / greet / say");
    let log = log_of(&nested);
    assert!(log.contains("hello nested token=*** other=none"), "{log}");

    // A secret passed under the name the step file asks for is seen by
    // that name alone: OTHER's four letters as KEY, and no OTHER.
    let renamed = check_of(&runner, &owner, &key, "renamed");
    assert_eq!(renamed["status"], "success", "{renamed}");
    let log = log_of(&renamed);
    assert!(log.contains("key=*** len=4 token=none other=none"), "{log}");
    // Through a step file passing on its own name, TOKEN's six as KEY.
    let renamings = check_of(&runner, &owner, &key, "renamings");
    assert_eq!(renamings["status"], "success", "{renamings}");
    let log = log_of(&renamings);
    assert!(log.contains("key=*** len=6 token=none other=none"), "{log}");
    // One the runner does not grant is named as the workflow names it.
    let unrenamed = check_of(&runner, &owner, &key, "unrenamed");
    assert_eq!(unrenamed["status"], "success", "{unrenamed}");
    let log = log_of(&unrenamed);
    assert!(log.contains("==> MISSING is not granted to"), "{log}");
    assert!(log.contains("key=none len=0"), "{log}");

    // On main, what is for main runs, and what is for x passes unrun.
    let gated = check_of(&runner, &owner, &key, "gated");
    assert_eq!(gated["status"], "success", "{gated}");
    let steps = gated["steps"].as_array().unwrap().iter();
    let unrun = steps.filter(|s| s["status"] == "skipped");
    let unrun: Vec<&str> = unrun.map(|s| s["name"].as_str().unwrap()).collect();
    assert_eq!(unrun, ["greet / needs", "greet / say", "greet / script"]);
    let log = log_of(&gated);
    assert!(log.contains("own any") && !log.contains("==> own main\n==> skipped"));
    assert!(
        log.contains("inner any") && log.contains("inner main"),
        "{log}"
    );
    assert!(!log.contains("hello x"), "{log}");
    let why = "==> greet / say\n==> skipped, as refs/heads/main is not refs/heads/x\n";
    assert!(log.contains(why), "{log}");
    // On x, the other way about.
    owner.git(&work, &["push", "-q", "origin", "HEAD:refs/heads/x"]);
    js.ring(&owner.account.did, &ref_rkey(&key, "refs/heads/x"));
    wait_until(|| {
        check_on(&runner, &owner, &key, "refs/heads/x", "gated")["finishedAt"].is_string()
    });
    let on_x = check_on(&runner, &owner, &key, "refs/heads/x", "gated");
    assert_eq!(on_x["status"], "success", "{on_x}");
    let log = log_of(&on_x);
    assert!(
        log.contains("hello x") && log.contains("inner any"),
        "{log}"
    );
    assert!(!log.contains("inner main"), "{log}");
    let why = "==> own main\n==> skipped, as refs/heads/x is not refs/heads/main\n";
    assert!(log.contains(why), "{log}");
    let inner = "==> gated / main\n==> skipped, as refs/heads/x is not refs/heads/main\n";
    assert!(log.contains(inner), "{log}");

    let said = |wf: &str| {
        let check = check_of(&runner, &owner, &key, wf);
        assert_eq!(check["status"], "error", "{check}");
        check["error"].as_str().unwrap().to_string()
    };
    for (wf, says) in [
        ("unpinned", "does not pin".to_string()),
        ("newer", "which @2 does not take".to_string()),
        ("gone", format!(".knotless/steps/gone.yml at {pinned} of")),
        ("looped", "loop@1 uses itself, through".to_string()),
        ("deep", "d5@1 is past 4 shared steps deep".to_string()),
    ] {
        let error = said(wf);
        assert!(error.contains(&says), "{wf}: {error}");
    }

    // Without a lock, nothing shared runs.
    owner.git(&work, &["rm", "-q", ".knotless/uses.lock"]);
    owner.git(&work, &["commit", "-q", "-m", "no lock"]);
    owner.git(&work, &["push", "-q", "origin", "main"]);
    let unlocked = owner.git(&work, &["rev-parse", "HEAD"]);
    js.ring(&owner.account.did, &ref_key);
    wait_until(|| {
        let check = check_of(&runner, &owner, &key, "shared");
        check["commit"] == unlocked.as_str() && check["finishedAt"].is_string()
    });
    let error = said("shared");
    assert!(error.contains("there is no .knotless/uses.lock"), "{error}");
}