//! An agent reviewing pulls on the reference PDS: a command, a script
//! here, reviews each open pull that asks the agent's account for a review
//! and that the owner opened, once at each head and only so often, and
//! the review is written as the agent's own comment.

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

use common::{Sandbox, net, wait_until};
use futures::{SinkExt, StreamExt};
use knotless_agent::{Job, Options};
use serde_json::Value;
use std::sync::Arc;
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 {
                let offered = |req: &Request, mut res: Response| {
                    let protocol = req.headers().get("sec-websocket-protocol").cloned();
                    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.
                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 send(&self, did: &str, collection: &str, rkey: &str, record: Value) {
        let payload = serde_json::json!({
            "$type": format!("{KIND}commit"), "seq": 1, "did": did, "time": "2026-10-29T00:00:00.000001Z",
            "rev": "3l3qo2vutsw2b", "operation": "create", "collection": collection, "rkey": rkey,
            "record": record,
        });
        let frame = serde_json::json!({"$type": "message", "payload": payload});
        let frames = [
            Message::Ping(Vec::new().into()),
            Message::Text(frame.to_string().into()),
        ];
        for frame in frames {
            self.frames.send(frame).unwrap();
        }
    }
}

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

/// An appview as [`appview`] makes, and a gateway too when `gateway`, for
/// the service tokens of private repositories.
fn serving(rt: &tokio::runtime::Runtime, s: &Sandbox, gateway: bool) -> String {
    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 = gateway.then(|| knotless_serve::gateway::Gateway::new(&base).unwrap());
    let app = knotless_serve::App::build(common::ctx(s), Duration::ZERO, gateway, Some(index));
    let router = knotless_serve::router(app.clone());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    rt.spawn(knotless_serve::firehose::watch(
        app,
        net().pds.replace("http://", "ws://"),
    ));
    base
}

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

/// Reviews by what the pull says: its first round asks for a change on a
/// line, a later one approves, and one stubborn asks every round; one
/// broken fails, one garbled answers
/// nothing readable, one slow takes too long, and one moving waits until
/// the test says go. It notes each pull it is asked about.
const REVIEW: &str = r#"input=$(cat)
printf '%s\n' "$input" >> "$ASKED"
case "$input" in
  *Broken*) exit 3 ;;
  *Garbled*) echo 'not json' ;;
  *Slow*) sleep 5 ;;
  *Moving*)
    touch "$ASKED.started"
    while [ ! -e "$ASKED.go" ]; do sleep 0.1; done
    echo '{"verdict": "approve", "body": "Fine."}' ;;
  *Stubborn*) echo '{"verdict": "changes", "body": "Still not."}' ;;
  *'"round":1'*) echo '{"verdict": "changes", "body": "Name the file.", "lines": [{"line": {"path": "f", "line": 1}, "body": "Say what f is."}]}' ;;
  *) echo '{"verdict": "approve", "body": "Looks right now."}' ;;
esac
"#;

/// How many times the command was asked about a pull titled `title`.
fn asked(bot: &Sandbox, title: &str) -> usize {
    let asked = std::fs::read_to_string(bot.path("asked")).unwrap_or_default();
    let title = format!("\"title\":\"{title}\"");
    asked.lines().filter(|l| l.contains(&title)).count()
}

/// The last thing the command was given about a pull titled `title`.
fn given(bot: &Sandbox, title: &str) -> Value {
    let asked = std::fs::read_to_string(bot.path("asked")).unwrap();
    let title = format!("\"title\":\"{title}\"");
    let line = asked.lines().rfind(|l| l.contains(&title)).unwrap();
    serde_json::from_str(line).unwrap()
}

/// The reviews `bot` wrote of `pull`, oldest first, and the comments on
/// lines that went with them.
fn reviews(bot: &Sandbox, pull: &str) -> (Vec<Value>, Vec<Value>) {
    let mut said: Vec<(String, Value)> = bot.records("dev.knotless.comment");
    said.retain(|(_, c)| c["subject"] == pull);
    said.sort_by(|a, b| a.0.cmp(&b.0));
    let (lines, reviews): (Vec<_>, Vec<_>) = said
        .into_iter()
        .map(|(_, c)| c)
        .partition(|c| c["line"].is_object());
    (reviews, lines)
}

#[test]
fn an_agent_reviews_pulls() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let base = appview(&rt, &Sandbox::new());
    let (mut owner, mut stranger, mut bot) = (Sandbox::new(), Sandbox::new(), Sandbox::new());
    for s in [&mut owner, &mut stranger, &mut bot] {
        let handle = s.account.handle.clone();
        s.set("KNOTLESS_APPVIEW", &base);
        s.set("KNOTLESS_IDENTIFIER", &handle);
    }
    let me = bot.account.handle.clone();
    std::fs::write(bot.path("review.sh"), REVIEW).unwrap();
    let asked_at = bot.path("asked").to_string_lossy().to_string();
    let command = format!("ASKED={asked_at} sh {}", bot.path("review.sh").display());

    let url = owner.url("proj");
    let work = owner.work("w", 1);
    // What it asks of reviews, on the branch pulls merge into.
    std::fs::create_dir_all(work.join(".knotless")).unwrap();
    owner.commit_file(&work, ".knotless/review.md", "Check the tests.\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    // Each pull brings a branch of its own, one commit past main.
    let branch = |name: &str| {
        let sha = owner.commit_file(&work, "f", &format!("{name}\n"));
        owner.git(
            &work,
            &["push", "-q", &url, &format!("HEAD:refs/heads/{name}")],
        );
        owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);
        sha
    };
    let pull = |name: &str, title: &str, more: &[&str]| {
        branch(name);
        let args = [
            &["pull", "create", &url, "main", &url, name, title][..],
            more,
        ]
        .concat();
        ok(&owner, &args)
    };
    let asking = ["--reviewer", me.as_str()];
    let named = pull("named", "Name it", &asking);
    let draft = pull("draft", "Draft one", &[&asking[..], &["--draft"]].concat());
    let unasked = pull("unasked", "Unasked", &[]);
    let failing = [
        pull("broken", "Broken", &asking),
        pull("garbled", "Garbled", &asking),
        pull("slow", "Slow", &asking),
    ];
    let gone = pull("gone", "Gone branch", &asking);
    ok(&owner, &["branch", "delete", &url, "gone"]);
    // A stranger may ask it too, from a fork: it reviews nothing of theirs.
    let fork = ok(&stranger, &["repo", "fork", &url]);
    let theirs = stranger.work("s", 2);
    stranger.git(&theirs, &["push", "-q", &fork, "HEAD:refs/heads/mine"]);
    let args = [
        "pull",
        "create",
        &url,
        "main",
        &fork,
        "mine",
        "From a stranger",
        "--reviewer",
        &me,
    ];
    let strangers = ok(&stranger, &args);
    let all = [&named, &draft, &unasked, &gone, &strangers];
    wait_until(|| {
        let listed = owner.cli(&["pull", "list", &url], b"").1;
        all.iter()
            .copied()
            .chain(&failing)
            .all(|p| listed.contains(p.rsplit('/').next().unwrap()))
    });

    let opts = Options {
        job: Job::Review,
        command: command.clone(),
        all: false,
        rounds: 2,
        timeout: Duration::from_secs(3),
        jetstream: "ws://127.0.0.1:9".into(),
        sweep: Duration::from_secs(600),
        settle: Duration::ZERO,
        once: true,
        repos: vec![format!("{}/proj", owner.account.handle)],
        owners: Vec::new(),
        attempts: 3,
    };
    let once = |opts: &Options| {
        rt.block_on(knotless_agent::run(opts.clone(), common::ctx(&bot)))
            .unwrap()
    };
    once(&opts);
    // The pull that asked it was reviewed, a change asked for on a line,
    // at the head it read.
    let given_named = given(&bot, "Name it");
    assert_eq!(given_named["round"], 1);
    assert_eq!(given_named["rounds"], 2);
    assert_eq!(given_named["changes"], 0);
    assert_eq!(given_named["approved"], Value::Null);
    assert_eq!(given_named["author"], owner.account.handle.as_str());
    assert_eq!(given_named["base"], "main");
    assert_eq!(given_named["head"]["branch"], "named");
    assert_eq!(given_named["earlier"], serde_json::json!([]));
    assert_eq!(given_named["instructions"], "Check the tests.\n");
    let (said, lines) = reviews(&bot, &named);
    assert_eq!(said.len(), 1);
    assert_eq!(said[0]["verdict"], "changes");
    assert_eq!(said[0]["body"], "Name the file.");
    assert_eq!(said[0]["commit"], given_named["head"]["commit"]);
    assert_eq!(lines.len(), 1);
    assert_eq!(lines[0]["body"], "Say what f is.");
    // Nothing that did not ask, or a draft, a stranger's, or one whose
    // branch is gone; and nothing written of what failed.
    for title in ["Draft one", "Unasked", "From a stranger", "Gone branch"] {
        assert_eq!(asked(&bot, title), 0, "{title}");
    }
    for p in &failing {
        assert!(reviews(&bot, p).0.is_empty(), "{p}");
    }
    wait_until(|| ok(&owner, &["pull", "show", &url, &named]).contains("changes"));

    // Read again, by the binary, it reviews only what failed before.
    let ran = std::process::Command::new(env!("CARGO_BIN_EXE_knotless-agent"))
        .args([
            "review",
            "--command",
            &command,
            "--rounds",
            "2",
            "--timeout",
            "3",
        ])
        .args(["--once", &opts.repos[0]])
        .envs(bot.vars())
        .output()
        .unwrap();
    assert!(ran.status.success(), "{ran:?}");
    assert_eq!(asked(&bot, "Name it"), 1);
    assert_eq!(asked(&bot, "Broken"), 2);

    // Pushed to, it is reviewed again, as a second round that knows the
    // first; pushed to once more after it approved, it is reviewed again,
    // from the head it approved: an approval spends none of --rounds.
    let sha = owner.commit_file(&work, "f", "named, as asked\n");
    owner.git(&work, &["push", "-q", "-f", &url, "HEAD:refs/heads/named"]);
    once(&opts);
    let given_named = given(&bot, "Name it");
    assert_eq!(given_named["round"], 2);
    assert_eq!(given_named["changes"], 1);
    assert_eq!(given_named["approved"], Value::Null);
    assert_eq!(given_named["head"]["commit"], sha.as_str());
    assert_eq!(given_named["earlier"][0]["body"], "Name the file.");
    let (said, _) = reviews(&bot, &named);
    assert_eq!(said.len(), 2);
    assert_eq!(said[1]["verdict"], "approve");
    let again = owner.commit_file(&work, "f", "named again\n");
    owner.git(&work, &["push", "-q", "-f", &url, "HEAD:refs/heads/named"]);
    wait_until(|| ok(&owner, &["pull", "show", &url, &named]).contains("approve"));
    once(&opts);
    assert_eq!(asked(&bot, "Name it"), 3);
    let given_named = given(&bot, "Name it");
    assert_eq!(given_named["round"], 3);
    assert_eq!(given_named["changes"], 1);
    assert_eq!(given_named["approved"], sha.as_str());
    assert_eq!(given_named["head"]["commit"], again.as_str());
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~2"]);

    // One it asks changes of at each head stops at --rounds of them.
    let stubborn = pull("stubborn", "Stubborn", &asking);
    wait_until(|| {
        owner
            .cli(&["pull", "list", &url], b"")
            .1
            .contains(stubborn.rsplit('/').next().unwrap())
    });
    once(&opts);
    for n in ["stubborn again\n", "stubborn once more\n"] {
        owner.commit_file(&work, "f", n);
        owner.git(
            &work,
            &["push", "-q", "-f", &url, "HEAD:refs/heads/stubborn"],
        );
        owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);
        once(&opts);
    }
    assert_eq!(asked(&bot, "Stubborn"), 2);
    assert_eq!(reviews(&bot, &stubborn).0.len(), 2);

    // A head that only merges the base into the branch changes nothing the
    // pull does: the last review is carried to it, read by no command and
    // counted as no round. A change of its own is then read as round 2.
    let carried = pull("carried", "Carried", &asking);
    wait_until(|| {
        owner
            .cli(&["pull", "list", &url], b"")
            .1
            .contains(carried.rsplit('/').next().unwrap())
    });
    once(&opts);
    assert_eq!(asked(&bot, "Carried"), 1);
    wait_until(|| ok(&owner, &["pull", "show", &url, &carried]).contains("changes"));
    owner.commit_file(&work, "g", "the base moves on\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    owner.git(&work, &["fetch", "-q", &url, "carried"]);
    owner.git(&work, &["merge", "-q", "--no-edit", "FETCH_HEAD"]);
    let merged = owner.git(&work, &["rev-parse", "HEAD"]);
    owner.git(&work, &["push", "-q", &url, "HEAD:refs/heads/carried"]);
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);
    once(&opts);
    assert_eq!(asked(&bot, "Carried"), 1);
    let (said, lines) = reviews(&bot, &carried);
    assert_eq!(said.len(), 2);
    assert_eq!(said[1]["verdict"], "changes");
    assert_eq!(said[1]["commit"], merged.as_str());
    assert!(
        said[1]["body"]
            .as_str()
            .unwrap()
            .starts_with("Carried forward:")
    );
    assert_eq!(lines.len(), 1);
    // Read again at the same head, it writes nothing more.
    wait_until(|| ok(&owner, &["pull", "show", &url, &carried]).contains(&merged[..12]));
    once(&opts);
    assert_eq!(reviews(&bot, &carried).0.len(), 2);
    let fixed = owner.commit_file(&work, "f", "carried, as asked\n");
    owner.git(
        &work,
        &["push", "-q", "-f", &url, "HEAD:refs/heads/carried"],
    );
    once(&opts);
    assert_eq!(asked(&bot, "Carried"), 2);
    let given_carried = given(&bot, "Carried");
    assert_eq!(given_carried["round"], 2);
    assert_eq!(given_carried["head"]["commit"], fixed.as_str());
    assert_eq!(given_carried["earlier"].as_array().unwrap().len(), 1);
    assert_eq!(given_carried["earlier"][0]["body"], "Name the file.");
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);

    // Asked for changes, a pull whose author changes what it says and
    // then merges the base in is read again at that merge, as a round: the
    // request may be answered, so it is not carried.
    let edited = pull("edited", "Edited", &asking);
    wait_until(|| {
        owner
            .cli(&["pull", "list", &url], b"")
            .1
            .contains(edited.rsplit('/').next().unwrap())
    });
    once(&opts);
    wait_until(|| ok(&owner, &["pull", "show", &url, &edited]).contains("changes"));
    ok(
        &owner,
        &["comment", "edit", &url, &edited, "-m", "Answered in here."],
    );
    wait_until(|| ok(&owner, &["pull", "show", &url, &edited]).contains("Answered in here"));
    owner.commit_file(&work, "g", "the base moves on again\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    owner.git(&work, &["fetch", "-q", &url, "edited"]);
    owner.git(&work, &["merge", "-q", "--no-edit", "FETCH_HEAD"]);
    let merged = owner.git(&work, &["rev-parse", "HEAD"]);
    owner.git(&work, &["push", "-q", &url, "HEAD:refs/heads/edited"]);
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);
    once(&opts);
    assert_eq!(asked(&bot, "Edited"), 2);
    let given_edited = given(&bot, "Edited");
    assert_eq!(given_edited["round"], 2);
    assert_eq!(given_edited["head"]["commit"], merged.as_str());
    let (said, _) = reviews(&bot, &edited);
    assert_eq!(said.len(), 2);
    assert_eq!(said[1]["verdict"], "approve");

    // Asked for changes, a pull whose author then changes what it says is
    // read again at the same head, as a round; once approved, an edit is
    // read no more.
    let answered = pull("answered", "Answered", &asking);
    wait_until(|| {
        owner
            .cli(&["pull", "list", &url], b"")
            .1
            .contains(answered.rsplit('/').next().unwrap())
    });
    once(&opts);
    assert_eq!(asked(&bot, "Answered"), 1);
    let head = given(&bot, "Answered")["head"]["commit"].clone();
    wait_until(|| ok(&owner, &["show", &url, &answered]).contains("changes"));
    once(&opts);
    assert_eq!(asked(&bot, "Answered"), 1);
    ok(
        &owner,
        &[
            "comment",
            "edit",
            &url,
            &answered,
            "-m",
            "The file is named here.",
        ],
    );
    wait_until(|| ok(&owner, &["pull", "show", &url, &answered]).contains("named here"));
    once(&opts);
    assert_eq!(asked(&bot, "Answered"), 2);
    let given_answered = given(&bot, "Answered");
    assert_eq!(given_answered["round"], 2);
    assert_eq!(given_answered["head"]["commit"], head);
    assert_eq!(given_answered["body"], "The file is named here.");
    assert_eq!(given_answered["earlier"][0]["commit"], head);
    let (said, _) = reviews(&bot, &answered);
    assert_eq!(said[1]["verdict"], "approve");
    wait_until(|| ok(&owner, &["pull", "show", &url, &answered]).contains("approve"));
    ok(
        &owner,
        &["comment", "edit", &url, &answered, "-m", "Said again."],
    );
    wait_until(|| ok(&owner, &["pull", "show", &url, &answered]).contains("Said again"));
    once(&opts);
    assert_eq!(asked(&bot, "Answered"), 2);

    // One pushed to as it is read is read again at its new head.
    let moving = pull("moving", "Moving", &asking);
    wait_until(|| {
        owner
            .cli(&["pull", "list", &url], b"")
            .1
            .contains(moving.rsplit('/').next().unwrap())
    });
    let patient = Options {
        timeout: Duration::from_secs(60),
        ..opts.clone()
    };
    let reading = rt.spawn(knotless_agent::run(patient.clone(), common::ctx(&bot)));
    wait_until(|| bot.path("asked.started").exists());
    let moved = owner.commit_file(&work, "f", "moved\n");
    owner.git(&work, &["push", "-q", "-f", &url, "HEAD:refs/heads/moving"]);
    std::fs::write(bot.path("asked.go"), "").unwrap();
    rt.block_on(reading).unwrap().unwrap();
    assert!(reviews(&bot, &moving).0.is_empty());
    once(&patient);
    let (said, _) = reviews(&bot, &moving);
    assert_eq!(said.len(), 1);
    assert_eq!(said[0]["commit"], moved.as_str());
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);

    // Watching, with no Jetstream to hear from, it reads every --sweep.
    let mut watch = Options {
        once: false,
        sweep: Duration::from_millis(200),
        ..opts.clone()
    };
    let sweeping = rt.spawn(knotless_agent::run(watch.clone(), common::ctx(&bot)));
    let swept = pull("swept", "Swept", &asking);
    wait_until(|| reviews(&bot, &swept).0.len() == 1);
    sweeping.abort();

    // With Jetstream, it reads its repositories when a pull about one is
    // written, or a ref by one whose pull asks it: here, one now gone,
    // which it goes on past.
    let gone_repo = owner.url("gone");
    ok(&owner, &["repo", "create", &gone_repo]);
    let js = jetstream(&rt);
    watch.jetstream = js.url.clone();
    watch.sweep = Duration::from_secs(600);
    watch
        .repos
        .insert(0, format!("{}/gone", owner.account.handle));
    let before = asked(&bot, "Broken");
    rt.spawn(knotless_agent::run(watch, common::ctx(&bot)));
    // Connecting, it reads them all.
    wait_until(|| asked(&bot, "Broken") > before);
    ok(&owner, &["repo", "delete", &gone_repo, "--yes"]);
    let heard = pull("heard", "Heard", &asking);
    wait_until(|| {
        owner
            .cli(&["pull", "list", &url], b"")
            .1
            .contains(heard.rsplit('/').next().unwrap())
    });
    let rkey = heard.rsplit('/').next().unwrap();
    let record = owner.record("dev.knotless.pull", rkey);
    let mut elsewhere = record.clone();
    elsewhere["repo"]["key"] = "3zzzzzzzzzzzz".into();
    let did = owner.account.did.clone();
    js.send(&did, "dev.knotless.pull", rkey, elsewhere);
    js.send(&did, "dev.knotless.check", rkey, record.clone());
    js.send(&did, "dev.knotless.pull", rkey, record);
    wait_until(|| reviews(&bot, &heard).0.len() == 1);
    wait_until(|| ok(&owner, &["pull", "show", &url, &heard]).contains("changes"));
    owner.commit_file(&work, "f", "heard again\n");
    owner.git(&work, &["push", "-q", "-f", &url, "HEAD:refs/heads/heard"]);
    js.send(&did, "dev.knotless.ref", "heard", serde_json::json!({}));
    wait_until(|| reviews(&bot, &heard).0.len() == 2);

    wait_until(|| ok(&owner, &["pull", "show", &url, &heard]).contains("approve"));
    // With --all, it reviews the owner's pulls that did not ask it too,
    // but still no draft.
    let all = Options {
        all: true,
        ..opts.clone()
    };
    once(&all);
    assert_eq!(asked(&bot, "Unasked"), 1);
    assert_eq!(reviews(&bot, &unasked).0.len(), 1);
    assert_eq!(asked(&bot, "Draft one"), 0);

    // With --owner, it watches every repository the owner holds, one made
    // after it started too: a repository record the owner writes sends it
    // to list them again. An owner it cannot list is passed by.
    let js = jetstream(&rt);
    let owners = Options {
        once: false,
        jetstream: js.url.clone(),
        sweep: Duration::from_secs(600),
        repos: Vec::new(),
        owners: vec![
            "did:web:nobody.invalid".into(),
            owner.account.handle.clone(),
        ],
        ..opts.clone()
    };
    let before = asked(&bot, "Broken");
    rt.spawn(knotless_agent::run(owners, common::ctx(&bot)));
    wait_until(|| asked(&bot, "Broken") > before);
    let later = owner.url("later");
    owner.git(&work, &["push", "-q", &later, "main"]);
    owner.commit_file(&work, "f", "later\n");
    owner.git(&work, &["push", "-q", &later, "HEAD:refs/heads/late"]);
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);
    let args = ["pull", &later, "main", &later, "late", "Late", "--reviewer"];
    let late = ok(&owner, &[&args[..], &[me.as_str()]].concat());
    let rkey = late.rsplit('/').next().unwrap();
    wait_until(|| owner.cli(&["pull", "list", &later], b"").1.contains(rkey));
    js.send(&did, "dev.knotless.repo", "later", serde_json::json!({}));
    wait_until(|| reviews(&bot, &late).0.len() == 1);
    // A pull about any of the owner's repositories rings for it too.
    wait_until(|| ok(&owner, &["pull", "show", &later, &late]).contains("changes"));
    owner.commit_file(&work, "f", "later again\n");
    owner.git(&work, &["push", "-q", "-f", &later, "HEAD:refs/heads/late"]);
    owner.git(&work, &["reset", "-q", "--hard", "HEAD~1"]);
    let record = owner.record("dev.knotless.pull", rkey);
    js.send(&did, "dev.knotless.pull", rkey, record);
    wait_until(|| reviews(&bot, &late).0.len() == 2);
}

/// A private repository's pulls are reviewed as a public one's, once the
/// owner shares it with the agent's account: it reads them, and what the
/// repository asks of reviews, as itself. These need a PDS on the spaces
/// alpha, and pass trivially without one.
#[test]
fn an_agent_reviews_a_private_repositorys_pulls() {
    if !net().spaces {
        eprintln!("skipped: the PDS is not on the spaces alpha");
        return;
    }
    let rt = tokio::runtime::Runtime::new().unwrap();
    let base = serving(&rt, &Sandbox::new(), true);
    let mut owner = Sandbox::new();
    let owned = owner.account.handle.clone();
    owner.set("KNOTLESS_APPVIEW", &base);
    owner.set("KNOTLESS_IDENTIFIER", &owned);
    // It reads as itself, logged in, as a space credential needs.
    let mut bot = Sandbox::new();
    bot.set("KNOTLESS_PASSWORD", "");
    bot.set("KNOTLESS_APPVIEW", &base);
    let me = bot.account.handle.clone();
    let password = bot.account.password.clone();
    let login = bot.cli(&["login", &me, "--password-stdin"], password.as_bytes());
    assert_eq!(login.0, 0, "{}", login.1);
    std::fs::write(bot.path("review.sh"), REVIEW).unwrap();
    let asked_at = bot.path("asked").to_string_lossy().to_string();
    let command = format!("ASKED={asked_at} sh {}", bot.path("review.sh").display());

    let url = owner.url("hidden");
    ok(
        &owner,
        &["repo", "create", &url, "--private", "--no-runner"],
    );
    ok(&owner, &["repo", "share", &url, &me]);
    let work = owner.work("w", 1);
    std::fs::create_dir_all(work.join(".knotless")).unwrap();
    owner.commit_file(&work, ".knotless/review.md", "Keep it secret.\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    owner.commit_file(&work, "f", "hidden\n");
    owner.git(&work, &["push", "-q", &url, "HEAD:refs/heads/x"]);
    let pull = ok(
        &owner,
        &[
            "pull",
            "create",
            &url,
            "main",
            &url,
            "x",
            "Hidden change",
            "--reviewer",
            &me,
        ],
    );
    wait_until(|| {
        bot.cli(&["pull", "list", &url], b"")
            .1
            .contains(pull.rsplit('/').next().unwrap())
    });

    let opts = Options {
        job: Job::Review,
        command,
        all: false,
        rounds: 2,
        timeout: Duration::from_secs(10),
        jetstream: "ws://127.0.0.1:9".into(),
        sweep: Duration::from_secs(600),
        settle: Duration::ZERO,
        once: true,
        repos: vec![format!("{owned}/hidden")],
        owners: Vec::new(),
        attempts: 3,
    };
    rt.block_on(knotless_agent::run(opts.clone(), common::ctx(&bot)))
        .unwrap();
    let read = given(&bot, "Hidden change");
    assert_eq!(read["instructions"], "Keep it secret.\n");
    // Its review is kept with the repository's comments, which the owner
    // reads.
    wait_until(|| ok(&owner, &["pull", "show", &url, &pull]).contains("Name the file."));

    // With --owner alone it finds the repository too: sharing it left the
    // agent an invite at the appview, which no listing of the owner's PDS
    // would show.
    owner.commit_file(&work, "g", "hidden again\n");
    owner.git(&work, &["push", "-q", &url, "HEAD:refs/heads/y"]);
    let args = ["pull", "create", &url, "main", &url, "y", "Found by owner"];
    let found = ok(&owner, &[&args[..], &["--reviewer", &me]].concat());
    wait_until(|| {
        bot.cli(&["pull", "list", &url], b"")
            .1
            .contains(found.rsplit('/').next().unwrap())
    });
    let owners = Options {
        repos: Vec::new(),
        owners: vec![owned.clone()],
        ..opts
    };
    rt.block_on(knotless_agent::run(owners, common::ctx(&bot)))
        .unwrap();
    assert_eq!(
        given(&bot, "Found by owner")["instructions"],
        "Keep it secret.\n"
    );
}

#[test]
fn the_binary() {
    let bin = env!("CARGO_BIN_EXE_knotless-agent");
    let s = Sandbox::new();
    let run = |args: &[&str]| {
        let out = std::process::Command::new(bin)
            .args(args)
            .envs(s.vars())
            .env("KNOTLESS_IDENTIFIER", &s.account.handle)
            .output()
            .unwrap();
        let err = String::from_utf8_lossy(&out.stderr).to_string();
        (out.status.code(), err)
    };
    let (code, err) = run(&[]);
    assert_eq!(code, Some(2));
    assert!(err.contains("usage: knotless-agent review"), "{err}");
    // A repository its owner does not have.
    let repo = format!("{}/nothing", s.account.handle);
    let (code, err) = run(&["review", "--command", "true", "--once", &repo]);
    assert_eq!(code, Some(1));
    assert!(err.contains("knotless-agent: "), "{err}");
}