//! An agent that works on repositories as an account of its own, at one
//! of two jobs. `review` reviews each open pull of the repositories it is
//! given that asks its account for a review, as GitHub's review bots do,
//! and writes the review as that account's comment, with a verdict and
//! comments on lines. `keep` keeps pulls the merge queue turned away
//! moving: one that conflicts with its base it brings up to date, where
//! both sides only added lines, and queues again, or tells its author
//! where they changed the same ones.
//!
//! What reads the pull is a command, run by `sh -c` (`--command`), such as
//! `knotless-claude-review`, which has Claude Code read it: it reads the
//! pull as JSON on stdin, its `repo`, `pull`, `title`, `body`, `author`,
//! `base` and `head` (with the commit it is at), which `round` this is,
//! how many times it may ask for changes (`rounds`) and has (`changes`),
//! the head its last review `approved`, if it did, its own `earlier`
//! reviews of the pull, and what
//! the repository asks of its reviews (`instructions`: its
//! `.knotless/review.md` on the pull's base branch, never the pull's), and
//! writes `{"verdict": "approve" | "changes", "body": ..., "lines": [...]}`
//! on stdout, each of `lines` a comment on lines of the pull's diff as
//! [`LineComment`]s are. It has `--timeout` to answer.
//!
//! With `--all` it reviews every open pull that is not a draft, asked or
//! not, as a GitHub workflow that reviews each pull does.
//!
//! With `--owner <handle|did>` it watches, besides the repositories named,
//! every repository that account holds that it may read: the public ones,
//! listed from its PDS at each read, so one made meanwhile is watched from
//! then on, and the private ones shared with its account, which `knotless
//! repo share` leaves it an invite to at the appview, since a PDS lists
//! no one another's private repositories.
//!
//! A pull is reviewed once at each head, and only one its repository's
//! owner or a collaborator opened: anyone may ask any account for a
//! review, and each spends what the command costs its operator. It stops
//! once it has asked for changes `--rounds` times: that bounds a pull
//! argued back and forth, while one it approved is reviewed at each head
//! it is pushed to after, from the head it approved. A head whose own
//! change is the one last reviewed, as when all it adds is a merge of the
//! base into the branch, is not read again: the last verdict is written
//! at it as it stands. One pushed to while it is read is read again at
//! its new head. One whose title or body its author changed after the
//! agent asked for changes is read again, at that head or a later one,
//! rather than having that verdict carried: what was asked may be
//! answered in what the pull says rather than in its code.
//! One whose review fails is noted and read again at the next sweep.
//!
//! Jetstream is only the doorbell, as it is for knotless-triage: a pull
//! written about a watched repository, or a ref pushed by one whose pull
//! asks it, sends it to read the repository again once `--settle` has
//! passed, as does a repository record an owner it watches writes; it
//! reads every repository on connecting, and every `--sweep` besides. For
//! either job, a queue record about a watched repository rings too.
//! `--once` reads each once, and stops.

use anyhow::{Context, Result, bail};
use knotless::branch_update::{self, Conflicts};
use knotless::env::Ctx;
use knotless::forge::{self, Item, LineComment, Target};
use knotless::jetstream;
use knotless::records::{CommentRecord, PullHead, RepoRecord, RepoRef, nsid};
use knotless::remote::Remote;
use knotless::session;
use knotless::telemetry::{self, Fields, Level};
use knotless::url::{RepoUrl, SCHEME};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::process::Stdio;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::sync::Notify;

/// What a repository asks of its reviews, on the branch a pull merges into.
const REVIEW_MD: &str = ".knotless/review.md";

/// How a review carried to a new head begins: one that read nothing, so
/// counted as no round and given to the command as no earlier review.
const CARRIED: &str = "Carried forward:";

/// How a pull's body says the keeper is to leave it alone, on a line of
/// its own.
const KEEPER_OFF: &str = "Keeper: off";

/// How the keeper's comment on a pull it brought up to date begins: it
/// counts them against `--attempts`.
const UPDATED: &str = "Merged";

const USAGE: &str = "usage: knotless-agent review --command <command> [--all] [--rounds <n>] [--timeout <secs>] [--jetstream <wss://host>] [--sweep <secs>] [--settle <secs>] [--once] [--owner <handle|did>]... [<handle>/<name>...]
       knotless-agent keep [--attempts <n>] [--jetstream <wss://host>] [--sweep <secs>] [--settle <secs>] [--once] [--owner <handle|did>]... [<handle>/<name>...]";

/// What the agent does.
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum Job {
    /// Reviews the pulls that ask it.
    Review,
    /// Keeps the pulls the merge queue turned away moving.
    Keep,
}

#[derive(Clone)]
pub struct Options {
    pub job: Job,
    /// The command that reviews a pull, for `sh -c`.
    pub command: String,
    /// Whether it reviews every open pull, not only those asking it.
    pub all: bool,
    /// How many times at most it asks for changes of one pull.
    pub rounds: usize,
    /// How long the command has for a pull.
    pub timeout: Duration,
    pub jetstream: String,
    /// How often every repository is read, whatever Jetstream says.
    pub sweep: Duration,
    /// How long after a ring the repositories are read.
    pub settle: Duration,
    pub once: bool,
    pub repos: Vec<String>,
    /// The accounts whose every repository it watches, by handle or DID.
    pub owners: Vec<String>,
    /// How many times at most the keeper brings one pull up to date.
    pub attempts: usize,
}

/// The options `args` give.
pub fn parse_args(args: &[String]) -> Result<Options, String> {
    let job = match args.get(1).map(String::as_str) {
        Some("review") => Job::Review,
        Some("keep") => Job::Keep,
        _ => return Err(USAGE.into()),
    };
    let mut opts = Options {
        job,
        command: String::new(),
        all: false,
        rounds: 3,
        timeout: Duration::from_secs(1200),
        jetstream: "wss://jetstream2.us-east.bsky.network".into(),
        sweep: Duration::from_secs(300),
        settle: Duration::from_secs(5),
        once: false,
        repos: Vec::new(),
        owners: Vec::new(),
        attempts: 3,
    };
    let mut it = args.iter().skip(2);
    while let Some(flag) = it.next() {
        let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value"));
        match flag.as_str() {
            "--command" => opts.command = value()?,
            "--all" => opts.all = true,
            "--rounds" => {
                let n = value()?.parse().ok().filter(|n| *n > 0);
                opts.rounds = n.ok_or("--rounds takes a count, from 1")?;
            }
            "--timeout" => opts.timeout = secs(flag, value()?)?,
            "--jetstream" => opts.jetstream = value()?,
            "--sweep" => opts.sweep = secs(flag, value()?)?,
            "--settle" => opts.settle = secs(flag, value()?)?,
            "--once" => opts.once = true,
            "--owner" => opts.owners.push(value()?),
            "--attempts" => {
                let n = value()?.parse().ok().filter(|n| *n > 0);
                opts.attempts = n.ok_or("--attempts takes a count, from 1")?;
            }
            other if other.starts_with('-') => return Err(format!("unexpected {other}\n{USAGE}")),
            repo => opts.repos.push(repo.trim_start_matches(SCHEME).to_string()),
        }
    }
    let commanded = job == Job::Keep || !opts.command.is_empty();
    if !commanded || (opts.repos.is_empty() && opts.owners.is_empty()) {
        return Err(format!(
            "name its command and a repository, or --owner\n{USAGE}"
        ));
    }
    Ok(opts)
}

fn secs(flag: &str, value: String) -> Result<Duration, String> {
    let secs = value.parse().map_err(|_| format!("{flag} takes seconds"))?;
    Ok(Duration::from_secs(secs))
}

/// What the command is given of a pull.
#[derive(Serialize)]
struct Asked<'a> {
    /// The repository, as `knotless://<owner>/<name>`.
    repo: String,
    pull: &'a str,
    title: &'a str,
    body: Option<&'a str>,
    /// Who opened it, by handle where it resolves, else by DID.
    author: &'a str,
    /// The branch it merges into.
    base: Option<&'a str>,
    head: Head<'a>,
    /// Which review of the pull this is, from 1.
    round: usize,
    /// How many times at most it may ask for changes of the pull.
    rounds: usize,
    /// How many times it has.
    changes: usize,
    /// The head its last review approved, when it did: what came since is
    /// what is new.
    approved: Option<String>,
    earlier: Vec<Earlier>,
    /// What the repository asks of its reviews.
    instructions: Option<String>,
}

#[derive(Serialize)]
struct Head<'a> {
    repo: String,
    branch: &'a str,
    commit: &'a str,
}

/// One of the agent's own earlier reviews of a pull.
#[derive(Serialize)]
struct Earlier {
    verdict: String,
    body: String,
    /// The head it looked at.
    commit: Option<String>,
    /// When it was written.
    #[serde(skip)]
    at: String,
}

/// What a review read of a pull: its head, and when its title or body
/// had last changed.
type Read = (String, Option<String>);

/// What the command answers: the review to write.
#[derive(Deserialize)]
struct Answer {
    verdict: String,
    body: String,
    #[serde(default)]
    lines: Vec<LineComment>,
}

/// What [`run`] works with.
struct Agent {
    ctx: Ctx,
    opts: Options,
    /// The agent's DID.
    me: String,
    /// The repositories it was named, by owner and key: what records about
    /// each name.
    repos: BTreeMap<(String, String), RepoUrl>,
    /// Each `--owner`'s DID and repositories, as last listed, by the owner
    /// as given.
    owned: Mutex<BTreeMap<String, Owned>>,
    /// Those whose pulls ask it for a review: a ref they push may move one.
    authors: Mutex<BTreeSet<String>>,
    /// The head it last reviewed each pull at, and when the pull's title
    /// or body had last changed then, by the pull: the appview may not
    /// have heard of the review yet as it reads the pull again.
    posted: Mutex<BTreeMap<String, Read>>,
    door: Notify,
}

pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
    let me = session::me(&ctx).await?.did;
    let mut repos = BTreeMap::new();
    for repo in &opts.repos {
        let url = RepoUrl::parse(&format!("{SCHEME}{repo}"))?;
        let t = Target::open(&ctx, &url).await?;
        repos.insert((t.remote.did().to_string(), t.reference().key), url);
    }
    let r = Arc::new(Agent {
        ctx,
        opts,
        me,
        repos,
        owned: Mutex::default(),
        authors: Mutex::default(),
        posted: Mutex::default(),
        door: Notify::new(),
    });
    if r.opts.once {
        for url in r.watched().await {
            r.sweep(&url).await?;
        }
        return Ok(());
    }
    tokio::spawn(follow(r.clone()));
    loop {
        tokio::select! {
            _ = r.door.notified() => tokio::time::sleep(r.opts.settle).await,
            _ = tokio::time::sleep(r.opts.sweep) => {}
        }
        for url in r.watched().await {
            failed(&url, r.sweep(&url).await);
        }
    }
}

/// An `--owner`'s repositories, as last listed.
#[derive(Default)]
struct Owned {
    did: String,
    repos: BTreeMap<(String, String), RepoUrl>,
}

impl Agent {
    /// Every repository it watches: those it was named, and each owner's,
    /// listed again now. An owner whose listing fails keeps its last.
    async fn watched(&self) -> Vec<RepoUrl> {
        for owner in &self.opts.owners {
            match self.list(owner).await {
                Ok(listed) => {
                    self.owned.lock().unwrap().insert(owner.clone(), listed);
                }
                Err(err) => {
                    let f = Fields {
                        did: Some(owner),
                        error: Some(format!("{err:#}")),
                        ..Default::default()
                    };
                    telemetry::note(Level::WARN, "listing an owner's repositories failed", &f);
                }
            }
        }
        let mut all = self.repos.clone();
        for owned in self.owned.lock().unwrap().values() {
            for (key, url) in &owned.repos {
                all.entry(key.clone()).or_insert_with(|| url.clone());
            }
        }
        all.into_values().collect()
    }

    /// `owner`'s repositories: its public ones, from its PDS, and the
    /// private ones it invited this account to, as the appview has their
    /// invites, each that this account reads as a member.
    async fn list(&self, owner: &str) -> Result<Owned> {
        let url = RepoUrl::parse(&format!("{SCHEME}{owner}/_"))?;
        let remote = Remote::connect(&self.ctx, &url).await?;
        let did = remote.did().to_string();
        let public = remote.list_repos().await?.into_iter().map(|r| {
            let url = RepoUrl {
                authority: did.clone(),
                name: r.value.name,
            };
            ((did.clone(), r.value.key), url)
        });
        let mut repos: BTreeMap<_, _> = public.collect();
        // Invites go through the appview, which only a gateway gives a DID
        // to ask with: without them its public ones are still watched.
        let invited = match self.invited(&did).await {
            Ok(found) => found,
            Err(err) => {
                let f = Fields {
                    did: Some(owner),
                    error: Some(format!("{err:#}")),
                    ..Default::default()
                };
                telemetry::note(Level::WARN, "listing the invites left it failed", &f);
                Vec::new()
            }
        };
        for url in &invited {
            // One it no longer reads, unshared since, is passed by.
            if let Ok(t) = Target::open(&self.ctx, url).await {
                repos.insert((did.clone(), t.reference().key), t.url);
            }
        }
        Ok(Owned { repos, did })
    }

    /// The private repositories of `owner` (a DID) that invited this
    /// account, as `knotless repo share` leaves an invite: read from the
    /// first each time, as it keeps nothing.
    async fn invited(&self, owner: &str) -> Result<Vec<RepoUrl>> {
        let me = session::me(&self.ctx).await?;
        let mut found = Vec::new();
        let mut cursor = None;
        loop {
            let page = knotless::invites::list(&self.ctx, &me, cursor.as_deref()).await?;
            if page.invites.is_empty() {
                return Ok(found);
            }
            let theirs = page.invites.into_iter().filter(|i| i.did == owner);
            found.extend(theirs.map(|i| RepoUrl {
                authority: i.did,
                name: i.name,
            }));
            cursor = page.cursor;
        }
    }

    /// Does its job for each of `url`'s pulls waiting for it.
    async fn sweep(&self, url: &RepoUrl) -> Result<()> {
        match self.opts.job {
            Job::Review => self.review_all(url).await,
            Job::Keep => self.keep_all(url).await,
        }
    }

    /// Reviews each of `url`'s pulls waiting for it.
    async fn review_all(&self, url: &RepoUrl) -> Result<()> {
        let t = Target::open(&self.ctx, url).await?;
        let items = forge::list(&self.ctx, url).await?;
        let asking = items
            .iter()
            .filter(|i| self.asks(i, t.remote.did(), &t.record));
        let asking: Vec<&Item> = asking.collect();
        let authors = asking.iter().map(|i| i.author.clone());
        self.authors.lock().unwrap().extend(authors);
        for pull in asking {
            let reviewed = self.review(url, pull).await;
            noted(url, &pull.uri, reviewed, "reviewed", "reviewing failed");
        }
        Ok(())
    }

    /// Keeps each of `url`'s pulls the merge queue turned away moving.
    async fn keep_all(&self, url: &RepoUrl) -> Result<()> {
        let t = Target::open(&self.ctx, url).await?;
        let items = forge::list(&self.ctx, url).await?;
        let (owner, key) = (t.remote.did(), t.reference().key);
        let ejected = items.iter().filter(|i| keeps(i, owner, &key, &t.record));
        // Queueing is refused once its branch's rule has dropped the queue,
        // so one brought up to date could not go back in.
        let ejected = ejected.filter(|i| knotless::queue::takes(&t.record, owner, &items, i));
        for pull in ejected {
            let kept = self.keep(url, &t, pull).await;
            let kept = kept.map(|did| did.map(String::from));
            noted(url, &pull.uri, kept, "kept", "keeping failed");
        }
        Ok(())
    }

    /// Keeps `pull`, which the merge queue turned away, moving, when it
    /// conflicts with its base: brings its branch up to date, where both
    /// sides only added lines, and queues it again; or says, once at each
    /// head and base, where both sides changed the same lines, for its
    /// author to merge. Not once its author has pushed to it since, nor
    /// when it was turned away for any other reason, its own to fix, nor
    /// after `--attempts` updates. Answers what it did.
    async fn keep(&self, url: &RepoUrl, t: &Target, pull: &Item) -> Result<Option<&'static str>> {
        let head = pull.head.as_ref().context("a pull has a head")?;
        let base = pull.base.as_deref().context("a pull has a base")?;
        let Some(commit) = self.head(head).await? else {
            return Ok(None);
        };
        let turned_away = pull.queue.as_ref().and_then(|q| q.commit.as_deref());
        let tip = t.snap.find_ref(&format!("refs/heads/{base}"));
        let tip = tip.and_then(|r| r.value.target.clone()).unwrap_or_default();
        let at = format!("`{}` on {base} at `{}`", short(&commit), short(&tip));
        let thread = forge::show(&self.ctx, url, &pull.uri).await?;
        let mine = thread.comments.iter().filter(|c| c.author == self.me);
        let said: Vec<&str> = mine.filter_map(|c| c.value["body"].as_str()).collect();
        // Brought up to date but not queued again, as when queueing failed
        // once it had said so: queued now, with nothing more said.
        let made = format!("making its head `{}`", short(&commit));
        let unqueued = said
            .last()
            .is_some_and(|b| b.starts_with(UPDATED) && b.contains(&made));
        if turned_away != Some(commit.as_str()) && unqueued {
            knotless::queue::enqueue(&self.ctx, url, &pull.uri).await?;
            return Ok(Some("queued"));
        }
        let updates = said.iter().filter(|b| b.starts_with(UPDATED)).count();
        // Its author is told of a head that conflicts once, however often
        // its base moves on.
        let here = format!("`{}` on {base}", short(&commit));
        let seen = said.iter().any(|b| b.contains(&here));
        if turned_away != Some(commit.as_str()) || seen || updates >= self.opts.attempts {
            return Ok(None);
        }
        let words = knotless::fix::Words::Cli;
        let merge = knotless::landing::mergeable(&self.ctx, url, &pull.uri, &self.me, words);
        let merge = merge.await?;
        if merge.state != "conflicting" {
            return Ok(None);
        }
        let stuck = match branch_update::update_branch(&self.ctx, url, &pull.uri).await {
            Ok(updated) => {
                let made = updated.commit.as_deref().map(short).unwrap_or_default();
                let added = updated.added.join(", ");
                let body = format!(
                    "{UPDATED} {base} into this pull, making its head `{made}`, as its head {at} conflicted with it when the merge queue turned it away. Both sides only added lines in {added}, so {base}'s come first and this pull's after. Queued it again."
                );
                // Said before it is queued, so a sweep after a failure to
                // queue it finds what it did, and queues it then.
                forge::comment(&self.ctx, url, &pull.uri, &body, None, &[], &[]).await?;
                knotless::queue::enqueue(&self.ctx, url, &pull.uri).await?;
                return Ok(Some("updated"));
            }
            Err(err) => err.downcast::<Conflicts>()?.0.join(", "),
        };
        let body = format!(
            "Its head {at} conflicts with {base} in {stuck}, where both sides changed the same lines, so it is left as it is. Merge {base} into it, then queue it again."
        );
        forge::comment(&self.ctx, url, &pull.uri, &body, None, &[], &[]).await?;
        Ok(Some("conflicted"))
    }

    /// Whether `item` is an open pull asking the agent for a review, or
    /// any with `--all`, from `owner` or one of `record`'s collaborators.
    fn asks(&self, item: &Item, owner: &str, record: &RepoRecord) -> bool {
        let trusted = item.author == owner || record.is_collaborator(&item.author);
        item.kind == "pull"
            && item.state == "open"
            && !item.draft
            && item.author != self.me
            && (self.opts.all || item.reviewers.contains(&self.me))
            && trusted
    }

    /// Reviews `pull` at its head, unless it did already or has reviewed
    /// it as often as it may: answers the verdict, if it wrote one.
    async fn review(&self, url: &RepoUrl, pull: &Item) -> Result<Option<String>> {
        let head = pull.head.as_ref().context("a pull has a head")?;
        let Some(commit) = self.head(head).await? else {
            return Ok(None);
        };
        let edited = pull.updated_at.as_deref();
        let read: Read = (commit.clone(), pull.updated_at.clone());
        let posted = self.posted.lock().unwrap().get(&pull.uri) == Some(&read);
        // Its verdict at this head stands, unless it asked for changes and
        // the title or body may since have answered them: the thread says.
        let stands = |r: &knotless::reviews::Review| {
            r.by == self.me
                && r.at_head(&commit)
                && (r.verdict != knotless::reviews::CHANGES || edited.is_none())
        };
        if posted || pull.reviews.iter().any(stands) {
            return Ok(None);
        }
        let thread = forge::show(&self.ctx, url, &pull.uri).await?;
        let mut earlier = earlier(&thread.comments, &self.me);
        // The thread may know of a review at this head the pull does not yet.
        let at_head = (earlier.iter()).rfind(|e| e.commit.as_deref() == Some(commit.as_str()));
        let seen = at_head.is_some_and(|e| !answered(e, edited));
        // A head bringing nothing of its own since the last review keeps it.
        // Not a request for changes its title or body has since answered:
        // that is read again, as at its own head.
        if let Some(last) = earlier.last()
            && !answered(last, edited)
            && let Some(at) = last.commit.as_deref().filter(|at| *at != commit)
            && self.same_change(url, pull, at, &commit).await?
            && self.head(head).await?.as_ref() == Some(&commit)
        {
            let body = format!(
                "{CARRIED} the change this pull makes at `{commit}` is the one reviewed at `{at}`, \
                 so that review stands. What is new comes from its base."
            );
            let verdict = last.verdict.clone();
            forge::comment(&self.ctx, url, &pull.uri, &body, Some(&verdict), &[], &[]).await?;
            self.posted.lock().unwrap().insert(pull.uri.clone(), read);
            return Ok(Some(verdict));
        }
        // Where its last verdict, carried or not, approved: what is new is
        // what came since.
        let approved = earlier
            .last()
            .filter(|e| e.verdict == knotless::reviews::APPROVE);
        let approved = approved.and_then(|e| e.commit.clone());
        earlier.retain(|e| !e.body.starts_with(CARRIED));
        let changes = earlier
            .iter()
            .filter(|e| e.verdict == knotless::reviews::CHANGES);
        let changes = changes.count();
        if seen || changes >= self.opts.rounds {
            return Ok(None);
        }
        let asked = Asked {
            repo: url.to_string(),
            pull: &pull.uri,
            title: &pull.title,
            body: pull.body.as_deref(),
            author: pull.handle.as_deref().unwrap_or(&pull.author),
            base: pull.base.as_deref(),
            head: Head {
                repo: forge::repo_url(&head.repo.uri)?.to_string(),
                branch: &head.branch,
                commit: &commit,
            },
            round: earlier.len() + 1,
            rounds: self.opts.rounds,
            changes,
            approved,
            earlier,
            instructions: self.instructions(url, pull.base.as_deref()).await,
        };
        let answered = run_command(&self.opts.command, &asked);
        let answered = tokio::time::timeout(self.opts.timeout, answered).await;
        let answer = answered.context("the reviewer took too long")??;
        // One pushed to as it was read is read again, at its new head.
        if self.head(head).await?.as_ref() != Some(&commit) {
            return Ok(None);
        }
        let (body, verdict) = (&answer.body, Some(answer.verdict.as_str()));
        forge::comment(&self.ctx, url, &pull.uri, body, verdict, &[], &answer.lines).await?;
        self.posted.lock().unwrap().insert(pull.uri.clone(), read);
        Ok(Some(answer.verdict))
    }

    /// Whether `pull` changes at `now` what it changed at `before`: the
    /// same hunks against its base, wherever they fall, as a merge of the
    /// base into the branch leaves them. Not, when `before` is not to be
    /// had, as after a force-push that dropped it, or when there is no
    /// change at all.
    async fn same_change(
        &self,
        url: &RepoUrl,
        pull: &Item,
        before: &str,
        now: &str,
    ) -> Result<bool> {
        let base = pull.base.as_deref().context("a pull has a base")?;
        let head = pull.head.as_ref().context("a pull has a head")?;
        let git = knotless::mirror::synced(&self.ctx, url).await?;
        let fetched = forge::head_of(&self.ctx, &git, head).await?;
        fetched.map_err(anyhow::Error::msg)?;
        let base = git.rev_parse(&format!("refs/heads/{base}")).await?;
        // One it cannot read, as when it is gone, is no match.
        let before = change_id(&git, &base, before).await.ok().flatten();
        Ok(before.is_some() && before == change_id(&git, &base, now).await?)
    }

    /// What `.knotless/review.md` says on `url`'s branch `base`, if it is
    /// there: the base's, so a pull cannot change how it is reviewed.
    async fn instructions(&self, url: &RepoUrl, base: Option<&str>) -> Option<String> {
        let git = knotless::mirror::synced(&self.ctx, url).await.ok()?;
        let rev = format!("refs/heads/{}", base?);
        let read = knotless::read::blob(&git, &rev, REVIEW_MD).await.ok()?;
        String::from_utf8(read.1).ok()
    }

    /// The commit `head`'s branch points at, if it is still there.
    async fn head(&self, head: &PullHead) -> Result<Option<String>> {
        let source = Target::open(&self.ctx, &forge::repo_url(&head.repo.uri)?).await?;
        let branch = source.snap.find_ref(&format!("refs/heads/{}", head.branch));
        Ok(branch.and_then(|r| r.value.target.clone()))
    }

    /// Whether `event` writes a pull or a queue record about a repository
    /// it watches, a
    /// ref of one whose pull asks it for a review, or a repository record
    /// of an owner it watches.
    fn rings(&self, event: &jetstream::Event) -> bool {
        let did = event.did.as_deref().unwrap_or_default();
        let authors = self.authors.lock().unwrap();
        let owned = self.owned.lock().unwrap();
        let owner = |did: &str| owned.values().any(|o| o.did == did);
        event.ops.iter().any(|op| match op.collection.as_str() {
            nsid::PULL | nsid::QUEUE => (op.record.as_ref()).is_some_and(|r| {
                let about = serde_json::from_value::<RepoRef>(r["repo"].clone());
                about.is_ok_and(|about| owner(about.did()))
                    || self
                        .repos
                        .keys()
                        .any(|(owner, key)| forge::is_about(r, owner, key))
            }),
            nsid::REF => authors.contains(did),
            nsid::REPO => owner(did),
            _ => false,
        })
    }
}

/// Whether `review` asked for changes and the pull's title or body
/// changed after it, at `edited`: both are written as RFC 3339 in UTC to
/// the millisecond, so the later is the greater.
fn answered(review: &Earlier, edited: Option<&str>) -> bool {
    review.verdict == knotless::reviews::CHANGES && edited.is_some_and(|at| at > review.at.as_str())
}

/// Whether `item` is an open pull, not a draft, that the merge queue
/// turned away, from `owner` or one of `record`'s collaborators, whose
/// branch is in the repository itself, `owner`'s `key`, and whose body
/// does not say `Keeper: off`.
fn keeps(item: &Item, owner: &str, key: &str, record: &RepoRecord) -> bool {
    let trusted = item.author == owner || record.is_collaborator(&item.author);
    let here = item.head.as_ref().is_some_and(|h| {
        h.repo.key == key && forge::parse_uri(&h.repo.uri).is_ok_and(|(did, ..)| did == owner)
    });
    let ejected =
        (item.queue.as_ref()).is_some_and(|q| q.state == knotless::records::queue::EJECTED);
    let body = item.body.as_deref().unwrap_or_default();
    let off = body.lines().any(|l| l.trim() == KEEPER_OFF);
    let open = item.kind == "pull" && item.state == "open" && !item.draft;
    open && trusted && here && ejected && !off
}

/// Logs what a job did of `pull`, as `done` says: `did`, with what, or
/// `failed`, with why.
fn noted(url: &RepoUrl, pull: &str, done: Result<Option<String>>, did: &str, failed: &str) {
    let repo = url.to_string();
    let mut f = Fields {
        repo: Some(&repo),
        pull: Some(pull),
        ..Default::default()
    };
    match &done {
        Ok(Some(what)) => {
            f.reason = Some(what);
            telemetry::note(Level::INFO, did, &f);
        }
        Ok(None) => {}
        Err(err) => {
            f.error = Some(format!("{err:#}"));
            telemetry::note(Level::WARN, failed, &f);
        }
    }
}

/// A commit's first 12 characters, as a comment names it.
fn short(sha: &str) -> &str {
    sha.get(..12).unwrap_or(sha)
}

/// The reviews `me` wrote among `comments`, oldest first.
fn earlier(comments: &[forge::Entry], me: &str) -> Vec<Earlier> {
    let mine = comments.iter().filter(|c| c.author == me);
    let records =
        mine.filter_map(|c| serde_json::from_value::<CommentRecord>(c.value.clone()).ok());
    let reviews = records.filter_map(|c| {
        let verdict = c.verdict?;
        Some(Earlier {
            verdict,
            body: c.body,
            commit: c.commit,
            at: c.created_at,
        })
    });
    reviews.collect()
}

/// What `head` changes against where it left `base`, as `git patch-id`
/// names it: its hunks with no context and no line numbers, so a base
/// that moved around them leaves it the same. None for no change.
async fn change_id(git: &knotless::git::Git, base: &str, head: &str) -> Result<Option<String>> {
    let range = format!("{base}...{head}");
    let args = ["diff", "-U0", "--no-color", "--no-ext-diff", &range, "--"];
    let diff = git.run(&args, None).await?;
    // Of no change, patch-id names none.
    let id = git.text(&["patch-id", "--stable"], Some(&diff)).await?;
    Ok(id.split_whitespace().next().map(String::from))
}

/// What `command` answers of `asked`, given it on stdin.
async fn run_command(command: &str, asked: &Asked<'_>) -> Result<Answer> {
    let mut child = tokio::process::Command::new("sh")
        .args(["-c", command])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .kill_on_drop(true)
        .spawn()?;
    let mut stdin = child.stdin.take().expect("stdin is piped");
    // One that reads none of it answers all the same.
    let _ = stdin.write_all(&serde_json::to_vec(asked)?).await;
    drop(stdin);
    let out = child.wait_with_output().await?;
    if !out.status.success() {
        bail!("the reviewer failed: {}", out.status);
    }
    serde_json::from_slice(&out.stdout)
        .context("the reviewer's answer is not {verdict, body, lines}")
}

/// Logs that reading `url` failed, if `done` says it did.
fn failed(url: &RepoUrl, done: Result<()>) {
    if let Err(err) = done {
        let repo = url.to_string();
        let f = Fields {
            repo: Some(&repo),
            error: Some(format!("{err:#}")),
            ..Default::default()
        };
        telemetry::note(Level::WARN, "reading the repository failed", &f);
    }
}

/// Rings `r`'s door on connecting to Jetstream, and on each event that
/// [`Agent::rings`] for, reconnecting whenever the connection drops.
fn follow(r: Arc<Agent>) -> impl Future<Output = ()> {
    let base = r.opts.jetstream.clone();
    jetstream::follow(base, jetstream::IDLE, Doorbell { r })
}

/// Jetstream as the agent hears it: a doorbell for the repositories it
/// reads again, so each connection starts at the stream's present.
struct Doorbell {
    r: Arc<Agent>,
}

impl jetstream::Follower for Doorbell {
    fn connected(&mut self, up: bool) {
        if up {
            self.r.door.notify_one();
        }
    }

    fn event(&mut self, event: jetstream::Event, _frame: &str) {
        if self.r.rings(&event) {
            self.r.door.notify_one();
        }
    }
}

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

    fn args(line: &str) -> Vec<String> {
        line.split(' ').map(String::from).collect()
    }

    #[test]
    fn options() {
        let line = "knotless-agent review --command ./r --rounds 2 --timeout 3 --jetstream ws://j --sweep 9 --settle 1 --once knotless://a.example/p b.example/q";
        let opts = parse_args(&args(line)).unwrap();
        assert_eq!(opts.command, "./r");
        assert!(!opts.all);
        assert_eq!(opts.rounds, 2);
        assert_eq!(opts.jetstream, "ws://j");
        let durations = [opts.timeout, opts.sweep, opts.settle];
        assert_eq!(durations.map(|d| d.as_secs()), [3, 9, 1]);
        assert!(opts.once);
        assert_eq!(opts.repos, ["a.example/p", "b.example/q"]);
        assert_eq!((opts.job, opts.attempts), (Job::Review, 3));
        // Keeping needs no command.
        let opts = parse_args(&args("knotless-agent keep --attempts 2 a.example/p")).unwrap();
        assert_eq!((opts.job, opts.attempts), (Job::Keep, 2));
        let opts = parse_args(&args(
            "knotless-agent review --all --command ./r a.example/p",
        ))
        .unwrap();
        assert!(opts.all);
        assert_eq!((opts.rounds, opts.timeout.as_secs()), (3, 1200));
        assert!(!opts.once);
        let opts = parse_args(&args(
            "knotless-agent review --command ./r --owner a.example --owner did:plc:b",
        ))
        .unwrap();
        assert!(opts.repos.is_empty());
        assert_eq!(opts.owners, ["a.example", "did:plc:b"]);
        for (bad, says) in [
            ("knotless-agent", "usage:"),
            ("knotless-agent label --command ./r a.example/p", "usage:"),
            ("knotless-agent review a.example/p", "name its command"),
            ("knotless-agent review --command ./r", "name its command"),
            ("knotless-agent review --command", "--command takes a value"),
            (
                "knotless-agent review --command ./r --owner",
                "--owner takes a value",
            ),
            ("knotless-agent review --rounds 0", "--rounds takes a count"),
            ("knotless-agent review --rounds x", "--rounds takes a count"),
            (
                "knotless-agent review --sweep soon",
                "--sweep takes seconds",
            ),
            ("knotless-agent review --bogus", "unexpected --bogus"),
            ("knotless-agent keep", "name its command"),
            (
                "knotless-agent keep --attempts 0",
                "--attempts takes a count",
            ),
        ] {
            let err = parse_args(&args(bad)).err().unwrap();
            assert!(err.starts_with(says), "{bad}: {err}");
        }
    }
}