~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

knotless-agent keep updates turned-away pulls and queues them again

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:agent/keeper
.changeset/keeper.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: minor
---
# A keeper for the merge queue
`knotless-agent keep` is an account that keeps the pulls the merge queue turned away moving. One turned away for a conflict that is only lines both sides added it brings up to date and queues again. Where both sides changed the same lines, it tells the author on the pull which files. `Keeper: off` on a line of a pull's body has it left alone.
crates/knotless-agent/src/lib.rs+196−37
View file
@@ -1,8 +1,11 @@
//! An agent that works on repositories as an account of its own. Its one
//! job so far is `review`: it 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.
//! 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
@@ -37,10 +40,12 @@
//! 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.
//! 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;
@@ -64,10 +69,29 @@ const REVIEW_MD: &str = ".knotless/review.md";
/// counted as no round and given to the command as no earlier review.
const CARRIED: &str = "Carried forward:";
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>...]";
/// 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.
@@ -85,14 +109,19 @@ pub struct Options {
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> {
if args.get(1).map(String::as_str) != Some("review") {
return Err(USAGE.into());
}
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,
@@ -103,6 +132,7 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
once: false,
repos: Vec::new(),
owners: Vec::new(),
attempts: 3,
};
let mut it = args.iter().skip(2);
while let Some(flag) = it.next() {
@@ -120,11 +150,16 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
"--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()),
}
}
if opts.command.is_empty() || (opts.repos.is_empty() && opts.owners.is_empty()) {
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}"
));
@@ -191,7 +226,7 @@ struct Answer {
}
/// What [`run`] works with.
struct Reviewer {
struct Agent {
ctx: Ctx,
opts: Options,
/// The agent's DID.
@@ -219,7 +254,7 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
let t = Target::open(&ctx, &url).await?;
repos.insert((t.remote.did().to_string(), t.reference().key), url);
}
let r = Arc::new(Reviewer {
let r = Arc::new(Agent {
ctx,
opts,
me,
@@ -254,7 +289,7 @@ struct Owned {
repos: BTreeMap<(String, String), RepoUrl>,
}
impl Reviewer {
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> {
@@ -300,8 +335,16 @@ impl Reviewer {
})
}
/// Reviews each of `url`'s pulls waiting for it.
/// 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
@@ -312,27 +355,89 @@ impl Reviewer {
self.authors.lock().unwrap().extend(authors);
for pull in asking {
let reviewed = self.review(url, pull).await;
let repo = url.to_string();
let mut f = Fields {
repo: Some(&repo),
pull: Some(&pull.uri),
..Default::default()
};
match reviewed {
Ok(Some(verdict)) => {
f.reason = Some(&verdict);
telemetry::note(Level::INFO, "reviewed", &f);
}
Ok(None) => {}
Err(err) => {
f.error = Some(format!("{err:#}"));
telemetry::note(Level::WARN, "reviewing failed", &f);
}
}
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));
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 merge = knotless::landing::mergeable(&self.ctx, url, &pull.uri, &self.me).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 {
@@ -458,7 +563,8 @@ impl Reviewer {
Ok(branch.and_then(|r| r.value.target.clone()))
}
/// Whether `event` writes a pull about a repository it watches, a
/// 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 {
@@ -467,7 +573,7 @@ impl Reviewer {
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 => (op.record.as_ref()).is_some_and(|r| {
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
@@ -489,6 +595,50 @@ 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);
@@ -552,8 +702,8 @@ fn failed(url: &RepoUrl, done: Result<()>) {
}
/// Rings `r`'s door on connecting to Jetstream, and on each event that
/// [`Reviewer::rings`] for, reconnecting whenever the connection drops.
fn follow(r: Arc<Reviewer>) -> impl Future<Output = ()> {
/// [`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 })
}
@@ -561,7 +711,7 @@ fn follow(r: Arc<Reviewer>) -> impl Future<Output = ()> {
/// 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<Reviewer>,
r: Arc<Agent>,
}
impl jetstream::Follower for Doorbell {
@@ -598,6 +748,10 @@ mod tests {
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",
))
@@ -628,6 +782,11 @@ mod tests {
"--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}");
crates/knotless-agent/tests/keep.rsnew+236−0
View file
@@ -0,0 +1,236 @@
//! An agent keeping pulls the merge queue turned away moving, on the
//! reference PDS: one that conflicts with its base only where both sides
//! added lines it brings up to date and queues again, one where they
//! changed the same lines it tells its author of, once, and any other it
//! leaves alone.
#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::{Sandbox, net, wait_until};
use knotless::forge::Target;
use knotless::queue::Ejected;
use knotless::url::RepoUrl;
use knotless_agent::{Job, Options};
use serde_json::Value;
use std::time::Duration;
/// An appview over the reference PDS, following its firehose.
fn appview(rt: &tokio::runtime::Runtime, s: &Sandbox) -> String {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let base = format!("http://localhost:{}", listener.local_addr().unwrap().port());
let index = knotless_serve::index::Index::open(&s.path("index.db")).unwrap();
let app = knotless_serve::App::build(common::ctx(s), Duration::ZERO, None, 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
}
/// The comments `who` wrote on `pull`, oldest first.
fn said(who: &Sandbox, pull: &str) -> Vec<String> {
let mut said: Vec<(String, Value)> = who.records("dev.knotless.comment");
said.retain(|(_, c)| c["subject"] == pull);
said.sort_by(|a, b| a.0.cmp(&b.0));
let bodies = said
.into_iter()
.map(|(_, c)| c["body"].as_str().unwrap().to_string());
bodies.collect()
}
#[test]
fn an_agent_keeps_turned_away_pulls_moving() {
let rt = tokio::runtime::Runtime::new().unwrap();
let base = appview(&rt, &Sandbox::new());
let mut owner = Sandbox::new();
let handle = owner.account.handle.clone();
owner.set("KNOTLESS_APPVIEW", &base);
owner.set("KNOTLESS_IDENTIFIER", &handle);
let url = owner.url("kept");
let work = owner.work("w", 1);
owner.commit_file(&work, "notes", "one\n");
owner.git(&work, &["push", "-q", &url, "main"]);
// A runner named, so its pulls may be queued.
ok(&owner, &["runners", &url, "add", &handle]);
// `branch` as main is now, with `file` holding `text`, pushed.
let write = |branch: &str, from: &str, file: &str, text: &str| {
owner.git(&work, &["fetch", "-q", &url, from]);
owner.git(&work, &["checkout", "-q", "-B", branch, "FETCH_HEAD"]);
let sha = owner.commit_file(&work, file, text);
owner.git(&work, &["push", "-q", "-f", &url, branch]);
sha
};
let ctx = common::ctx(&owner);
let repo = RepoUrl::parse(&url).unwrap();
// Turns `pull` away from the queue at `head`, as a runner does.
let eject = |pull: &str, head: &str| {
let t = rt.block_on(Target::open(&ctx, &repo)).unwrap();
let ejected = Ejected {
reason: Some("it conflicts in notes"),
head: Some(head),
checks: &[],
};
rt.block_on(knotless::queue::eject(&ctx, &t, pull, ejected))
.unwrap();
};
let pull = |branch: &str, title: &str, body: &str| {
let pull = ok(
&owner,
&["pull", &url, "main", &url, branch, title, "-m", body],
);
wait_until(|| ok(&owner, &["list", &url]).contains(&pull));
pull
};
let shown = |pull: &str| ok(&owner, &["show", &url, pull]);
// Pulls cut from main, then main moved on.
let added = write("added", "main", "notes", "one\nadded\n");
let changed = write("changed", "main", "notes", "uno\n");
let apart = write("apart", "main", "apart", "apart\n");
let off = write("off", "main", "notes", "one\noff\n");
let moved = write("moved", "main", "notes", "one\nmoved\n");
let gone = write("gone", "main", "notes", "one\ngone\n");
// A branch of another of the owner's repositories: not the keeper's
// to push to.
let other = owner.url("elsewhere");
let away = write("away", "main", "notes", "one\naway\n");
owner.git(&work, &["push", "-q", &other, "away:main", "away"]);
let from_elsewhere = || {
let args = [
"pull",
&url,
"main",
&other,
"away",
"Away",
"-m",
"From elsewhere",
];
let pull = ok(&owner, &args);
wait_until(|| ok(&owner, &["list", &url]).contains(&pull));
pull
};
let pulls = [
(pull("added", "Added", "More notes"), added),
(pull("changed", "Changed", "Other notes"), changed),
(pull("apart", "Apart", "Elsewhere"), apart),
(pull("off", "Off", "Mine to fix\n\nKeeper: off"), off),
(pull("moved", "Moved", "Pushed to since"), moved),
(pull("gone", "Gone", "Its branch deleted"), gone),
(from_elsewhere(), away),
];
write("main", "main", "notes", "one\nmain\n");
for (pull, head) in &pulls {
eject(pull, head);
}
// Pushed to since it was turned away, it is its author's again.
write("moved", "moved", "notes", "one\nmoved, again\n");
ok(&owner, &["branch", &url, "gone", "--delete"]);
for (pull, _) in &pulls {
wait_until(|| shown(pull).contains("ejected"));
}
let added_at = pulls[0].1.clone();
let [added, changed, apart, off, moved, gone, away] = pulls.map(|(p, _)| p);
let opts = Options {
job: Job::Keep,
command: String::new(),
all: false,
rounds: 3,
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!("{handle}/kept")],
owners: Vec::new(),
attempts: 2,
};
let once = |opts: &Options| {
rt.block_on(knotless_agent::run(opts.clone(), common::ctx(&owner)))
.unwrap()
};
once(&opts);
// Lines both sides added: brought up to date, main's first, and queued.
owner.git(&work, &["fetch", "-q", &url, "added"]);
let notes = owner.git(&work, &["show", "FETCH_HEAD:notes"]);
assert_eq!(notes, "one\nmain\nadded");
let told = said(&owner, &added);
assert_eq!(told.len(), 1, "{told:?}");
assert!(
told[0].starts_with("Merged main into this pull"),
"{}",
told[0]
);
assert!(
told[0].contains("Both sides only added lines in notes"),
"{}",
told[0]
);
wait_until(|| shown(&added).contains("merge queue: queued"));
// The same line changed: its author is told where, and it stays.
let told = said(&owner, &changed);
assert_eq!(told.len(), 1, "{told:?}");
assert!(
told[0].contains("conflicts with main in notes"),
"{}",
told[0]
);
owner.git(&work, &["fetch", "-q", &url, "changed"]);
assert_eq!(owner.git(&work, &["show", "FETCH_HEAD:notes"]), "uno");
// Turned away for another reason, asked to be left, pushed to, its
// branch gone or elsewhere: nothing.
for pull in [&apart, &off, &moved, &gone, &away] {
assert!(said(&owner, pull).is_empty(), "{pull}");
}
// Read again, it says nothing twice.
wait_until(|| shown(&changed).matches("conflicts with main").count() == 1);
once(&opts);
assert_eq!(said(&owner, &changed).len(), 1);
assert_eq!(said(&owner, &added).len(), 1);
// Left turned away at the head it had before it was brought up to
// date, as when queueing it failed: queued again, nothing more said.
eject(&added, &added_at);
wait_until(|| shown(&added).contains("merge queue: ejected"));
once(&opts);
wait_until(|| shown(&added).contains("merge queue: queued"));
assert_eq!(said(&owner, &added).len(), 1);
// Turned away again once main moved on, it is brought up to date
// again, but no more often than it may be.
owner.git(&work, &["fetch", "-q", &url, "added"]);
let head = owner.git(&work, &["rev-parse", "FETCH_HEAD"]);
write("main", "main", "notes", "one\nmain\nmain again\n");
eject(&added, &head);
wait_until(|| shown(&added).contains("merge queue: ejected"));
let once_more = Options {
attempts: 1,
..opts.clone()
};
once(&once_more);
assert_eq!(said(&owner, &added).len(), 1);
once(&opts);
let told = said(&owner, &added);
assert_eq!(told.len(), 2, "{told:?}");
owner.git(&work, &["fetch", "-q", &url, "added"]);
let notes = owner.git(&work, &["show", "FETCH_HEAD:notes"]);
assert_eq!(notes, "one\nmain\nmain again\nadded");
// Its author was told of the head that conflicts once, though main
// moved on.
assert_eq!(said(&owner, &changed).len(), 1);
}
crates/knotless-agent/tests/review.rs+5−1
View file
@@ -8,7 +8,7 @@ mod common;
use common::{Sandbox, net, wait_until};
use futures::{SinkExt, StreamExt};
use knotless_agent::Options;
use knotless_agent::{Job, Options};
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
@@ -239,6 +239,7 @@ fn an_agent_reviews_pulls() {
});
let opts = Options {
job: Job::Review,
command: command.clone(),
all: false,
rounds: 2,
@@ -249,6 +250,7 @@ fn an_agent_reviews_pulls() {
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)))
@@ -589,6 +591,7 @@ fn an_agent_reviews_a_private_repositorys_pulls() {
});
let opts = Options {
job: Job::Review,
command,
all: false,
rounds: 2,
@@ -599,6 +602,7 @@ fn an_agent_reviews_a_private_repositorys_pulls() {
once: true,
repos: vec![format!("{owned}/hidden")],
owners: Vec::new(),
attempts: 3,
};
rt.block_on(knotless_agent::run(opts, common::ctx(&bot)))
.unwrap();
deploy/docker-compose.yml+35−0
View file
@@ -319,6 +319,41 @@ services:
cpus: "1"
memory: 1g
# Keeps the pulls the merge queue turned away moving (knotless-agent
# keep): one that conflicts only where both sides added lines it brings
# up to date and queues again. /etc/knotless/keep.env holds its
# account's KNOTLESS_IDENTIFIER and KNOTLESS_PASSWORD, an app password;
# the account is a collaborator of each repository it keeps. It runs
# no model and keeps nothing but its session.
keep:
image: knotless:current
profiles: [keep]
restart: always
logging: *logging
command:
- sh
- -c
- >-
exec knotless-agent keep
--jetstream ${KNOTLESS_JETSTREAM:-wss://jetstream.us-east.bsky.network}
${KNOTLESS_KEEP_ARGS:-}
${KNOTLESS_KEEP_REPOS:-}
environment:
KNOTLESS_APPVIEW: http://serve:2590
KNOTLESS_CONFIG_DIR: /bots
KNOTLESS_CACHE_DIR: /tmp/knotless-keep/cache
RUST_LOG: ${KNOTLESS_LOG:-info}
env_file:
- path: /etc/knotless/keep.env
required: false
volumes:
- ${KNOTLESS_DATA:-/var/lib/knotless}/bots:/bots
deploy:
resources:
limits:
cpus: "0.5"
memory: 512m
networks:
proxy:
name: ${KNOTLESS_PROXY_NETWORK:-knotless}
deploy/README.md+1−1
View file
@@ -5,7 +5,7 @@ One Docker host, a reverse proxy in front of it for TLS, and a domain.
| | |
| --- | --- |
| `../Dockerfile` | one image, `knotless:current`: knotless-serve, knotless-run, knotless-triage, knotless-agent, knotless-mcp with Claude Code, the knotless CLI, git, the docker client and the built pages |
| `docker-compose.yml` | `/etc/knotless/docker-compose.yml`: `serve` (knotless-serve, on :2590), `pages` (the SvelteKit app, on :2591) and, if asked, `run` (knotless-run), `triage` (knotless-triage) and `review` (knotless-agent review) |
| `docker-compose.yml` | `/etc/knotless/docker-compose.yml`: `serve` (knotless-serve, on :2590), `pages` (the SvelteKit app, on :2591) and, if asked, `run` (knotless-run), `triage` (knotless-triage), `review` (knotless-agent review) and `keep` (knotless-agent keep) |
| `deploy.sh` | builds the image on the host over SSH, writes the compose file and its `.env`, and restarts what changed; or one of those steps, `build`, `up` or `drop` |
| `rollout.sh` | run on the host by `deploy.sh`: moves `pages` and `serve` to a new container each, with no gap |
| `handover.sh` | run on the host by `deploy.sh`: moves `run` to a new container, the old one handing over to it as it finishes its runs |
docs/design/merge-queue.md+32−0
View file
@@ -521,6 +521,38 @@ while it is open:
- A time already past reads as any minute now. With no batch landed to
go by, nothing is said.
## Keeping turned-away pulls moving
A pull the queue turns away for a conflict waits for its author to merge
the branch into it and queue it again, and most such conflicts are two
pulls adding lines at the same place in a page both touch.
`knotless-agent keep` does that part for them: an account of its own,
the owner or a collaborator, that reads each repository it watches as
the review agent does, on Jetstream's queue and pull records and every
`--sweep`.
For each open pull from a branch of the repository itself, turned away
at the head that branch is still at, it
asks whether the pull conflicts with its base now. If it does, it merges
the base in, as [updating a pull's
branch](collaboration.md#updating-a-pulls-branch) does, keeping the
lines both sides only added, and queues the pull again at the merge.
It says so on the pull, naming the new head, before it queues it, so a
sweep after a failure to queue it finds the pull at that head and
queues it then. Its pre-flights then run at the new head, and a batch
checks it as any other. Where both sides changed the same lines it
pushes nothing, and says so on the pull once at that head, however
often the base moves on, for its author.
It never fixes a pull. One turned away for a check that failed is its
author's, and so is one pushed to since, as its author is at work on it.
A pull whose body says `Keeper: off` on a line of its own is left alone,
and one it has brought up to date `--attempts` times is left too, as a
pull that keeps conflicting has something to settle. It writes nothing
it would need to read back but its comments, so it keeps no state of its
own. Merging where both sides added lines takes no judgment, so the
keeper runs no model.
## Measuring it
`just queue-stats` says how the queue did, by day (UTC), from what it
docs/reference/cli.md+24−2
View file
@@ -645,6 +645,28 @@ setup-token`) or `ANTHROPIC_API_KEY`, `KNOTLESS_REVIEW_MODEL` for
Claude's `--model`,
`KNOTLESS_MCP` for the `knotless-mcp` to run and `CLAUDE` for the `claude`.
`knotless-agent keep [flags] [<handle>/<name>...]` keeps the pulls the
merge queue turned away moving, for the same repositories, with the same
`--jetstream`, `--sweep`, `--settle`, `--once` and `--owner` flags, and
`--attempts <n>` (3), the most times it brings one pull up to date. It
takes no command. Its account must be the owner or a collaborator, as it
pushes to the pull's branch and queues it. See [keeping turned-away
pulls moving](../design/merge-queue.md#keeping-turned-away-pulls-moving).
It reads a pull that is open and not a draft, that the owner or a
collaborator opened from a branch of the repository itself, that the queue turned away at the head its branch
is at now, and whose body has no line `Keeper: off`. When it conflicts
with its base, the keeper merges the base in as `knotless update-branch`
does, comments, beginning `Merged` and naming the new head, and queues
it again; one it finds still turned away at the head it gave it, as when
queueing failed, it queues with nothing more said. Where both sides
changed the same lines it pushes nothing, and comments once at that
head, naming the paths. A pull turned away for another reason, pushed
to since, or brought up to date `--attempts` times, it leaves alone.
Each pull it updates, queues or comments on is logged as `kept`, with
`updated`, `queued` or `conflicted`; one it fails on, as `keeping
failed`.
A usage error exits 2 and prints the usage, as `name its command and a
repository, or --owner`; an error at run time exits 1, as `knotless-agent: <error>`.
Watching, it stops at the start if a repository cannot be opened;
@@ -652,8 +674,8 @@ afterwards one that cannot be read is logged as `reading the repository
failed`, and it goes on. With `--once`, any such error stops it.
It follows Jetstream as knotless-triage does. It reads every repository
on connecting; a pull written about a watched repository or any of an
owner's, a repository record an owner writes, or a ref pushed by an
on connecting; a pull or a queue record written about a watched
repository or any of an owner's, a repository record an owner writes, or a ref pushed by an
account whose pull asks it, has it read the repositories again after
`--settle`.