johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! 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}");
}
}
}