johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! Triages repositories' issues: has a classifier choose each open one's
//! labels, and writes them as the triager's own `dev.knotless.label`
//! records, with what it judged.
//!
//! The classifier is a decision model (`--decide`, see [`decide`]), as
//! Jev or Clef, or any command, run by `sh -c` (`--classify`): it reads an issue as JSON on stdin, its
//! `subject`, `author`, `title` and `body` and the `labels` to choose
//! among, and writes `{"labels": [...], "model": ..., "reason": ...}` on
//! stdout. Either has `--timeout` to answer. It is asked only of an open
//! issue not judged
//! at its current version, and only for the scopes asked for (`--scopes`)
//! where nobody else has put a label on, so it never second-guesses a
//! person. One that fails, or answers nothing readable, is noted and asked
//! again at the next read.
//!
//! Jetstream is only the doorbell, as it is for knotless-run: an issue
//! written about a watched repository, or the repository's record, sends
//! it to read the repository again once `--settle` has passed, for the
//! appview to index the issue; it reads every repository on connecting,
//! and every `--sweep` besides. `--once` reads each once, and stops.
//!
//! Its labels count only once the owner names its account among the
//! repository's triagers, for those scopes. Until then they count for
//! nothing, and `knotless triager eval <url> <handle>` says how often
//! they agree with people's.
//!
//! With `--discover` it triages, besides the repositories named, every one
//! whose record names it among its triagers, in the scopes it names it a
//! triager of, as knotless-run finds those naming it among their runners:
//! as the appview lists them and private ones' invites say, at the start
//! and every `--sweep`, and as Jetstream brings a repository's record. One
//! that no longer names it is left.
pub mod decide;
use anyhow::{Context, Result, bail};
use decide::Decider;
use knotless::env::Ctx;
use knotless::forge;
use knotless::jetstream;
use knotless::records::nsid;
use knotless::remote::Remote;
use knotless::session;
use knotless::telemetry::{self, Fields, Level};
use knotless::triage::{Choice, Pending, Triage};
use knotless::url::{RepoUrl, SCHEME};
use knotless::xrpc::Xrpc;
use serde::Deserialize;
use std::collections::BTreeMap;
use std::process::Stdio;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::sync::Notify;
const USAGE: &str = "usage: knotless-triage (--decide <url> --model <name> [--threshold <p>] | --classify <command>) --scopes <scope>[,<scope>...] [--jetstream <wss://host>] [--sweep <secs>] [--settle <secs>] [--timeout <secs>] [--once] [--discover] [<handle>/<name>...]";
/// What chooses an issue's labels.
#[derive(Clone, PartialEq)]
pub enum Classifier {
Decide(Decider),
/// A command for `sh -c`.
Command(String),
}
#[derive(Clone)]
pub struct Options {
pub classify: Classifier,
/// The scopes of labels it chooses, as `kind` for `kind/bug`.
pub scopes: Vec<String>,
pub jetstream: String,
/// How often every repository is read, whatever Jetstream says.
pub sweep: Duration,
/// How long after a ring the repository is read.
pub settle: Duration,
/// How long the classifier has for an issue.
pub timeout: Duration,
pub once: bool,
/// Whether to triage, besides `repos`, every repository that names
/// the triager among its triagers.
pub discover: bool,
pub repos: Vec<String>,
}
/// The options `args` give, with a decision model's token from `env`:
/// `KNOTLESS_DECISION_TOKEN`.
pub fn parse_args(
args: &[String],
env: &dyn Fn(&str) -> Option<String>,
) -> Result<Options, String> {
let (mut command, mut endpoint, mut model, mut threshold) = (None, None, None, 0.5);
let mut opts = Options {
classify: Classifier::Command(String::new()),
scopes: Vec::new(),
jetstream: "wss://jetstream2.us-east.bsky.network".into(),
sweep: Duration::from_secs(600),
settle: Duration::from_secs(5),
timeout: Duration::from_secs(120),
once: false,
discover: false,
repos: Vec::new(),
};
let mut it = args.iter().skip(1);
while let Some(flag) = it.next() {
let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value"));
match flag.as_str() {
"--classify" => command = Some(value()?),
"--decide" => endpoint = Some(value()?),
"--model" => model = Some(value()?),
"--threshold" => {
let p = value()?.parse().ok().filter(|p| (0.0..=1.0).contains(p));
threshold = p.ok_or("--threshold takes a probability, from 0 to 1")?;
}
"--scopes" => {
let scopes = value()?;
let scopes = scopes.split(',').filter(|s| !s.is_empty());
opts.scopes = scopes.map(String::from).collect();
}
"--jetstream" => opts.jetstream = value()?,
"--sweep" => opts.sweep = secs(flag, value()?)?,
"--settle" => opts.settle = secs(flag, value()?)?,
"--timeout" => opts.timeout = secs(flag, value()?)?,
"--once" => opts.once = true,
"--discover" => opts.discover = true,
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.scopes.is_empty() || (opts.repos.is_empty() && !opts.discover) {
return Err(format!(
"name its scopes and a repository, or --discover\n{USAGE}"
));
}
opts.classify = match (command, endpoint) {
(Some(command), None) => Classifier::Command(command),
(None, Some(endpoint)) => Classifier::Decide(Decider {
endpoint,
model: model.ok_or("--decide takes --model")?,
token: env("KNOTLESS_DECISION_TOKEN")
.ok_or("--decide takes KNOTLESS_DECISION_TOKEN")?,
threshold,
}),
_ => {
return Err(format!(
"name one classifier, --decide or --classify\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))
}
/// A repository it triages, by its name, and whether it was found naming
/// the triager rather than named on the command line.
#[derive(Clone)]
struct Watch {
url: RepoUrl,
found: bool,
}
/// What [`run`] works with.
struct Triager {
ctx: Ctx,
opts: Options,
/// The triager's DID.
me: String,
/// What it watches, by owner and key: what records about each name.
watched: Mutex<BTreeMap<(String, String), Watch>>,
door: Notify,
}
pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
if opts.discover && ctx.env.appview.is_none() {
bail!(
"--discover finds repositories through an appview, and KNOTLESS_APPVIEW is set empty"
);
}
let me = session::me(&ctx).await?.did;
let t = Arc::new(Triager {
ctx,
opts,
me,
watched: Mutex::default(),
door: Notify::new(),
});
for repo in &t.opts.repos {
let url = RepoUrl::parse(&format!("{SCHEME}{repo}"))?;
let id = Triage::open(&t.ctx, &url).await?.repo();
let watch = Watch { url, found: false };
t.watched.lock().unwrap().insert(id, watch);
}
if t.opts.once {
if t.opts.discover {
t.discover().await;
}
for watch in t.watching() {
t.sweep(&watch).await?;
}
return Ok(());
}
tokio::spawn(follow(t.clone()));
if t.opts.discover {
tokio::spawn(discover(t.clone()));
}
loop {
tokio::select! {
_ = t.door.notified() => tokio::time::sleep(t.opts.settle).await,
_ = tokio::time::sleep(t.opts.sweep) => {}
}
for watch in t.watching() {
failed(&watch.url, t.sweep(&watch).await);
}
}
}
impl Triager {
/// What it watches now.
fn watching(&self) -> Vec<Watch> {
self.watched.lock().unwrap().values().cloned().collect()
}
/// Has the classifier judge each of `watch`'s issues waiting for it,
/// in the scopes asked for: of one found, only those the repository
/// names it a triager of. One found that no longer names it, or that
/// cannot be read, is left: discovery finds it again if it still does.
async fn sweep(&self, watch: &Watch) -> Result<()> {
let scopes = &self.opts.scopes;
let opened = Triage::open(&self.ctx, &watch.url).await;
if !watch.found {
return judge(&self.ctx, &self.opts, &opened?, scopes).await;
}
let granted = opened.as_ref().ok().and_then(|t| t.granted(scopes));
let (Ok(triage), Some(granted)) = (opened, granted) else {
self.watched
.lock()
.unwrap()
.retain(|_, w| w.url != watch.url);
said(&watch.url, "left");
return Ok(());
};
judge(&self.ctx, &self.opts, &triage, &granted).await
}
/// Watches `did`'s repository `name` if its record names this
/// triager and it is not watched already by that name: by another, it
/// was renamed, and is watched by this one instead.
async fn consider(&self, did: &str, name: &str) -> Result<()> {
let url = RepoUrl {
authority: did.to_string(),
name: name.to_string(),
};
let remote = Remote::open(&self.ctx, &url).await?;
let Some(repo) = remote.repo().await? else {
return Ok(());
};
if repo.value.triager(&self.me).is_none() {
return Ok(());
}
let url = RepoUrl {
authority: did.to_string(),
name: repo.rkey,
};
{
let mut watched = self.watched.lock().unwrap();
let id = (did.to_string(), repo.value.key);
// One on the command line is watched as it is.
if watched.get(&id).is_some_and(|w| !w.found || w.url == url) {
return Ok(());
}
let watch = Watch {
url: url.clone(),
found: true,
};
watched.insert(id, watch);
}
said(&url, "found");
self.door.notify_one();
Ok(())
}
/// Considers each repository the appview lists as naming this
/// triager, and each private one that invited it.
async fn discover(&self) {
let listed = self.listed().await;
let invited = self.invited().await;
for found in [listed, invited] {
match found {
Ok(found) => {
for (did, name) in found {
let url = RepoUrl {
authority: did.clone(),
name: name.clone(),
};
failed(&url, self.consider(&did, &name).await);
}
}
Err(err) => {
let f = Fields {
error: Some(format!("{err:#}")),
..Default::default()
};
telemetry::note(Level::WARN, "finding repositories failed", &f);
}
}
}
}
/// The repositories the appview lists as naming this triager, by
/// owner and name.
async fn listed(&self) -> Result<Vec<(String, String)>> {
let appview = self.ctx.env.appview.clone().unwrap_or_default();
let xrpc = Xrpc::new(self.ctx.http.clone(), &appview);
let mut found = Vec::new();
// The page to ask for next, if any: the first, from no cursor.
let mut next: Option<Option<String>> = Some(None);
while let Some(cursor) = next {
let mut params = vec![("triager", self.me.as_str()), ("limit", "500")];
params.extend(std::iter::repeat("cursor").zip(cursor.as_deref()));
let page: Listing = xrpc.query("dev.knotless.listRepos", ¶ms, None).await?;
found.extend(page.repos.into_iter().map(|r| (r.did, r.name)));
next = page.cursor.map(Some);
}
Ok(found)
}
/// The private repositories that invited this triager, all of them:
/// it keeps nothing, so reads them from the first each time, and one
/// left a while ago is found again.
async fn invited(&self) -> Result<Vec<(String, String)>> {
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);
}
found.extend(page.invites.into_iter().map(|i| (i.did, i.name)));
cursor = page.cursor;
}
}
/// Whether `event` writes an issue about a repository it watches, or a
/// watched owner's repository record, as when labels are defined.
fn rings(&self, event: &jetstream::Event) -> bool {
let watched = self.watched.lock().unwrap();
let did = event.did.as_deref().unwrap_or_default();
event.ops.iter().any(|op| match op.collection.as_str() {
nsid::ISSUE => (op.record.as_ref()).is_some_and(|r| {
watched
.keys()
.any(|(owner, key)| forge::is_about(r, owner, key))
}),
nsid::REPO => watched.keys().any(|(owner, _)| owner == did),
_ => false,
})
}
/// With `--discover`, considers each repository whose record `event`
/// writes, which may name this triager now.
fn found(self: &Arc<Self>, event: &jetstream::Event) {
let Some(did) = event.did.clone().filter(|_| self.opts.discover) else {
return;
};
let written = event.ops.iter().filter(|op| op.record.is_some());
for op in written.filter(|op| op.collection == nsid::REPO) {
let (t, did, name) = (self.clone(), did.clone(), op.rkey.clone());
tokio::spawn(async move {
let url = RepoUrl {
authority: did.clone(),
name: name.clone(),
};
failed(&url, t.consider(&did, &name).await);
});
}
}
}
/// Considers, at the start and every `--sweep`, each repository that
/// names the triager: Jetstream brings public ones named meanwhile
/// ([`Triager::found`]), but not always, and never private ones.
async fn discover(t: Arc<Triager>) {
loop {
t.discover().await;
tokio::time::sleep(t.opts.sweep).await;
}
}
/// Logs `message` of `url`.
fn said(url: &RepoUrl, message: &str) {
let repo = url.to_string();
let f = Fields {
repo: Some(&repo),
..Default::default()
};
telemetry::note(Level::INFO, message, &f);
}
/// 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);
}
}
/// A page of the appview's `dev.knotless.listRepos`, as discovery reads it.
#[derive(Deserialize)]
struct Listing {
repos: Vec<Listed>,
cursor: Option<String>,
}
#[derive(Deserialize)]
struct Listed {
did: String,
name: String,
}
/// Has the classifier judge each of `triage`'s issues waiting for it in
/// `scopes`, and writes what it chose.
async fn judge(ctx: &Ctx, opts: &Options, triage: &Triage, scopes: &[String]) -> Result<()> {
if scopes.is_empty() {
return Ok(());
}
let repo = triage.url().to_string();
for issue in triage.pending(ctx, scopes).await? {
let judged = match classify(ctx, &opts.classify, opts.timeout, &issue).await {
Ok(choice) => triage.judge(&issue, &choice).await,
Err(err) => Err(err),
};
let mut f = Fields {
repo: Some(&repo),
key: Some(&issue.subject),
..Default::default()
};
match judged {
Ok(labels) => {
let labels = labels.join(",");
f.reason = Some(&labels);
telemetry::note(Level::INFO, "judged", &f);
}
Err(err) => {
f.error = Some(format!("{err:#}"));
telemetry::note(Level::WARN, "judging failed", &f);
}
}
}
Ok(())
}
/// What `classifier` chooses for `issue`, in `timeout` or not at all.
async fn classify(
ctx: &Ctx,
classifier: &Classifier,
timeout: Duration,
issue: &Pending,
) -> Result<Choice> {
let chosen = async {
match classifier {
Classifier::Decide(d) => decide::classify(&ctx.http, d, issue).await,
Classifier::Command(command) => run_command(command, issue).await,
}
};
let chosen = tokio::time::timeout(timeout, chosen).await;
chosen.context("the classifier took too long")?
}
/// What `command` chooses for `issue`, given it on stdin.
async fn run_command(command: &str, issue: &Pending) -> Result<Choice> {
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(issue)?).await;
drop(stdin);
let out = child.wait_with_output().await?;
if !out.status.success() {
bail!("the classifier failed: {}", out.status);
}
serde_json::from_slice(&out.stdout)
.context("the classifier's answer is not {labels, model, reason}")
}
/// Rings `t`'s door on connecting to Jetstream, and on each event that
/// [`Triager::rings`] for, reconnecting whenever the connection drops;
/// with `--discover`, has it consider each repository record written.
fn follow(t: Arc<Triager>) -> impl Future<Output = ()> {
let base = t.opts.jetstream.clone();
jetstream::follow(base, jetstream::IDLE, Doorbell { t })
}
/// Jetstream as the triager hears it: a doorbell for the repositories it
/// reads again, so each connection starts at the stream's present.
struct Doorbell {
t: Arc<Triager>,
}
impl jetstream::Follower for Doorbell {
fn connected(&mut self, up: bool) {
if up {
self.t.door.notify_one();
}
}
fn event(&mut self, event: jetstream::Event, _frame: &str) {
if self.t.rings(&event) {
self.t.door.notify_one();
}
self.t.found(&event);
}
}
#[cfg(test)]
mod tests {
use super::*;
fn args(line: &str) -> Vec<String> {
line.split(' ').map(String::from).collect()
}
#[test]
fn options() {
let none = |_: &str| None;
let line = "knotless-triage --classify ./c --scopes kind,,area --jetstream ws://j --sweep 9 --settle 1 --timeout 3 --once knotless://a.example/p b.example/q";
let opts = parse_args(&args(line), &none).unwrap();
assert!(opts.classify == Classifier::Command("./c".into()));
assert_eq!(opts.scopes, ["kind", "area"]);
assert_eq!(opts.jetstream, "ws://j");
let durations = [opts.sweep, opts.settle, opts.timeout];
assert_eq!(durations.map(|d| d.as_secs()), [9, 1, 3]);
assert!(opts.once && !opts.discover);
assert_eq!(opts.repos, ["a.example/p", "b.example/q"]);
let line = "knotless-triage --classify ./c --scopes kind --discover";
let found = parse_args(&args(line), &none).unwrap();
assert!(found.discover && found.repos.is_empty());
let env = |k: &str| (k == "KNOTLESS_DECISION_TOKEN").then(|| "t0ken".to_string());
let line = "knotless-triage --decide https://d --model jev-latest --threshold 0.8 --scopes kind a.example/p";
let decider = Decider {
endpoint: "https://d".into(),
token: "t0ken".into(),
model: "jev-latest".into(),
threshold: 0.8,
};
let opts = parse_args(&args(line), &env).unwrap();
assert!(opts.classify == Classifier::Decide(decider));
for (bad, says) in [
(
"knotless-triage --classify ./c --scopes kind",
"name its scopes",
),
(
"knotless-triage --classify ./c a.example/p",
"name its scopes",
),
(
"knotless-triage --scopes kind a.example/p",
"name one classifier",
),
(
"knotless-triage --decide u --classify ./c --scopes k a.example/p",
"name one classifier",
),
(
"knotless-triage --decide u --scopes k a.example/p",
"--decide takes --model",
),
(
"knotless-triage --decide u --model m --scopes k a.example/p",
"--decide takes KNOTLESS_DECISION_TOKEN",
),
(
"knotless-triage --threshold 2",
"--threshold takes a probability",
),
(
"knotless-triage --threshold x",
"--threshold takes a probability",
),
("knotless-triage --classify", "--classify takes a value"),
("knotless-triage --sweep soon", "--sweep takes seconds"),
("knotless-triage --bogus", "unexpected --bogus"),
] {
let err = parse_args(&args(bad), &none).err().unwrap();
assert!(err.starts_with(says), "{bad}: {err}");
}
}
}