chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
run log shows what a check prints while it runs
.changeset/run-log.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: minor | ||
--- | ||
| ||
# `run log` shows what a check prints while it runs | ||
| ||
`knotless run log <url>`, with the flags `run list` filters by, prints the newest check's output: while it runs, what its steps printed so far, as the appview holds it, ending with the step going; once it ended, its log. `--follow` prints what it prints next, until it ends, then its result. The MCP tool `run_log` gives the same, with `since`, the cursor an earlier answer gave, and `wait`, to wait for more. Both read the appview's new `dev.knotless.getRunOutput` query, so they need an appview (`KNOTLESS_APPVIEW`). | ||
crates/knotless/src/checks.rs+126−3
| @@ -8,19 +8,34 @@ use crate::aturi::{self, AtUri}; | ||
use crate::env::Ctx; | ||
use crate::identity; | ||
use crate::keys::{check_prefix, checks_of, queue_name}; | ||
use crate::records::{CheckOutput, CheckRecord, MinutesRecord, RepoRecord, nsid, status}; | ||
use crate::records::{ | ||
CheckOutput, CheckRecord, CheckStep, MinutesRecord, RepoRecord, nsid, status, | ||
}; | ||
use crate::remote::{Remote, Snapshot, Stored}; | ||
use crate::space::{self, Space}; | ||
use crate::url::RepoUrl; | ||
use crate::xrpc::{Xrpc, XrpcError}; | ||
use crate::xrpc::{Auth, Xrpc, XrpcError}; | ||
use anyhow::{Context, Result, bail}; | ||
use serde::Serialize; | ||
use serde::{Deserialize, Serialize}; | ||
use serde_json::Value; | ||
use std::collections::HashMap; | ||
use std::future::Future; | ||
use std::time::Duration; | ||
use tokio::time::Instant; | ||
| ||
/// The appview's query of what a check's run printed so far. | ||
pub const RUN_OUTPUT: &str = "dev.knotless.getRunOutput"; | ||
| ||
/// The longest the appview waits for more of a run's output at once. | ||
pub const ASK: Duration = Duration::from_secs(25); | ||
| ||
/// How long a run just asked for is waited for to show, when followed. | ||
pub const FIND: Duration = Duration::from_secs(60); | ||
| ||
/// How long after a run's steps are over its record is looked for again: | ||
/// it is filed in a moment. | ||
pub const SETTLE: Duration = Duration::from_secs(1); | ||
| ||
/// Which checks [`list`] gives. | ||
#[derive(Debug, Clone, Copy, Default)] | ||
pub struct Filter<'a> { | ||
| @@ -400,6 +415,114 @@ pub async fn wait( | ||
} | ||
} | ||
| ||
/// The newest check `filter` finds, looked for again every | ||
/// `ctx.env.poll` while there is none, for at most `wait`: a run just | ||
/// asked for takes a moment to show. Checks of several workflows are an | ||
/// error naming them, as one is to be named, and finding none is one too. | ||
pub async fn newest(ctx: &Ctx, url: &RepoUrl, filter: Filter<'_>, wait: Duration) -> Result<Check> { | ||
let deadline = Instant::now() + wait; | ||
let found = loop { | ||
let found = retrying(ctx.env.poll, deadline, || list(ctx, url, filter, false)).await?; | ||
let left = deadline.saturating_duration_since(Instant::now()); | ||
if !found.is_empty() || left.is_zero() { | ||
break found; | ||
} | ||
tokio::time::sleep(left.min(ctx.env.poll)).await; | ||
}; | ||
let mut workflows: Vec<&str> = found.iter().map(|c| c.record.workflow.as_str()).collect(); | ||
workflows.sort_unstable(); | ||
workflows.dedup(); | ||
if workflows.len() > 1 { | ||
let several = workflows.join(", "); | ||
bail!( | ||
"the checks of {} are of several workflows, {several}: name one", | ||
asked(&filter) | ||
); | ||
} | ||
let none = format!("no checks of {}", asked(&filter)); | ||
found.into_iter().next().context(none) | ||
} | ||
| ||
/// The check `c`, found by `filter`, as its runner has it now, with its | ||
/// log if `logs` and it has one. | ||
pub async fn again( | ||
ctx: &Ctx, | ||
url: &RepoUrl, | ||
filter: Filter<'_>, | ||
c: &Check, | ||
logs: bool, | ||
) -> Result<Check> { | ||
let filter = Filter { | ||
commit: Some(&c.record.commit), | ||
workflow: Some(&c.record.workflow), | ||
..filter | ||
}; | ||
let found = list(ctx, url, filter, logs).await?; | ||
let gone = format!("{} is gone", c.uri); | ||
found.into_iter().find(|f| f.uri == c.uri).context(gone) | ||
} | ||
| ||
/// What a running check printed, as the appview holds it | ||
/// (`dev.knotless.getRunOutput`). | ||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] | ||
#[serde(rename_all = "camelCase")] | ||
pub struct Output { | ||
/// What its steps printed after the cursor asked from. | ||
pub text: String, | ||
/// Where `text` ends: asked from, it gives what comes after. | ||
pub cursor: String, | ||
/// The steps that ended so far. | ||
pub steps: Vec<CheckStep>, | ||
/// The step going, by name: none between steps. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub going: Option<String>, | ||
/// Whether its steps are over: its record is filed in a moment. | ||
pub ended: bool, | ||
} | ||
| ||
/// What the check `c`'s run printed after `cursor`, as the appview holds | ||
/// it while the run goes: asked again, [`ASK`] at most at a time, until | ||
/// there is some, the run ended, or `wait` is up. A private repository's | ||
/// is read as a member, with a delegation token for its space. | ||
pub async fn output(ctx: &Ctx, c: &Check, cursor: Option<&str>, wait: Duration) -> Result<Output> { | ||
let appview = ctx.env.appview.as_deref().context( | ||
"a running check's output comes from an appview, and none is set: set KNOTLESS_APPVIEW", | ||
)?; | ||
let xrpc = Xrpc::new(ctx.http.clone(), appview); | ||
let deadline = Instant::now() + wait; | ||
let mut cursor = cursor.map(str::to_string); | ||
loop { | ||
let left = deadline.saturating_duration_since(Instant::now()); | ||
let waited = left.min(ASK).as_secs().to_string(); | ||
let mut params = vec![ | ||
("repo", c.record.repo.uri.as_str()), | ||
("uri", c.uri.as_str()), | ||
("wait", waited.as_str()), | ||
]; | ||
params.extend(cursor.as_deref().map(|at| ("cursor", at))); | ||
let auth = match &c.space { | ||
Some(uri) => Some(member(ctx, uri).await?), | ||
None => None, | ||
}; | ||
let said: Output = xrpc.query(RUN_OUTPUT, ¶ms, auth.as_ref()).await?; | ||
if !said.text.is_empty() || said.ended || left <= ASK { | ||
return Ok(said); | ||
} | ||
cursor = Some(said.cursor); | ||
} | ||
} | ||
| ||
/// What proves whoever is logged in a member of the space `uri` to an | ||
/// appview reading it: a delegation token from their PDS, good once. | ||
async fn member(ctx: &Ctx, uri: &str) -> Result<Auth> { | ||
let (_, me) = space::reader(ctx, uri).await?; | ||
let token = space::delegation(ctx, uri, &me).await?; | ||
Ok(Auth::Headers(vec![( | ||
space::DELEGATION_HEADER.into(), | ||
token, | ||
)])) | ||
} | ||
| ||
/// Where a pull's checks stand at its head merged with its base, for the | ||
/// workflows that run on it. | ||
#[derive(Debug, PartialEq)] | ||
crates/knotless/src/cli.rs+91−9
| @@ -386,6 +386,10 @@ run | ||
run list <url> [--ref <name>] [--commit <sha>] [--pull <at-uri>] [--queue <branch>] [--workflow <name>] [--log] [--images <dir>] [--wait <secs>] [--json] | ||
what the runners report, newest first; | ||
--queue, of the merge queue's batches | ||
run log <url> [--ref <name>] [--commit <sha>] [--pull <at-uri>] [--queue <branch>] [--workflow <name>] [--follow] | ||
the newest check's output: so far, as | ||
the appview holds it, while it runs, | ||
else its log; --follow, until it ends | ||
run rerun <url> --ref <name> | --pull <at-uri> [--workflow <name>] | ||
ask the runners to run it again, or | ||
only one of its workflows | ||
| @@ -1104,6 +1108,68 @@ fn text(value: &str, stdin: &mut dyn std::io::Read) -> Result<String> { | ||
Ok(text) | ||
} | ||
| ||
/// A check as `run list` prints it: its status, workflow, the pull or the | ||
/// ref, the commit's first 12 characters and when it started, then why a | ||
/// workflow the change left out was, tab-separated. | ||
fn check_line(c: &crate::checks::Check) -> String { | ||
let r = &c.record; | ||
let sha = r.commit.get(..12).unwrap_or(&r.commit); | ||
// A pull's run is of the pull, whichever branch it brings. | ||
let of = r.pull.as_deref().unwrap_or(&r.ref_name); | ||
let skipped = r.skipped.as_deref().map(|why| format!("\t{why}")); | ||
let skipped = skipped.unwrap_or_default(); | ||
format!( | ||
"{}\t{}\t{of}\t{sha}\t{}{skipped}", | ||
r.status, r.workflow, r.started_at | ||
) | ||
} | ||
| ||
/// `knotless run log`: what the newest check `filter` finds printed, as | ||
/// the appview holds it while it runs, else its log; with `follow`, what | ||
/// it prints next until it ends. Ends with the check's line, as `run list` | ||
/// prints it, and while it runs the step going. | ||
async fn run_log( | ||
ctx: &Ctx, | ||
url: &RepoUrl, | ||
filter: crate::checks::Filter<'_>, | ||
follow: bool, | ||
out: &mut dyn Write, | ||
) -> Result<()> { | ||
use crate::checks::{self, ASK, FIND, SETTLE}; | ||
use std::time::Duration; | ||
let find = if follow { FIND } else { Duration::ZERO }; | ||
let mut check = checks::newest(ctx, url, filter, find).await?; | ||
let (mut cursor, mut going) = (None, None); | ||
while records::status::going(&check.record.status) { | ||
let wait = if follow { ASK } else { Duration::ZERO }; | ||
let said = checks::output(ctx, &check, cursor.as_deref(), wait).await?; | ||
write!(out, "{}", said.text)?; | ||
out.flush()?; | ||
(cursor, going) = (Some(said.cursor), said.going); | ||
if !follow { | ||
break; | ||
} | ||
// Quiet a while, or over: its record says whether it ended, and | ||
// once its steps are over it is filed in a moment. | ||
if said.ended || said.text.is_empty() { | ||
let settle = if said.ended { SETTLE } else { Duration::ZERO }; | ||
tokio::time::sleep(settle).await; | ||
check = checks::again(ctx, url, filter, &check, false).await?; | ||
} | ||
} | ||
// Ended before it was followed: its log, when it has one. | ||
if cursor.is_none() { | ||
check = checks::again(ctx, url, filter, &check, true).await?; | ||
let log = check.log_text.as_deref(); | ||
let log = log.map(|log| format!("{}\n", log.trim_end())); | ||
write!(out, "{}", log.unwrap_or_default())?; | ||
} | ||
let going = going.filter(|_| records::status::going(&check.record.status)); | ||
let step = going.map(|s| format!("\tstep {s}")).unwrap_or_default(); | ||
writeln!(out, "{}{step}", check_line(&check))?; | ||
Ok(()) | ||
} | ||
| ||
/// `knotless notifications settings`: changes them as the flags say, then | ||
/// prints each reason, on or off, and whom they are heard of from. | ||
async fn notification_settings(ctx: &Ctx, args: &mut Args<'_>, out: &mut dyn Write) -> Result<()> { | ||
| @@ -2388,15 +2454,7 @@ async fn command( | ||
writeln!(out, "{text}")?; | ||
} else { | ||
for c in &checks { | ||
let r = &c.record; | ||
let sha = r.commit.get(..12).unwrap_or(&r.commit); | ||
// A pull's run is of the pull, whichever branch it brings. | ||
let of = r.pull.as_deref().unwrap_or(&r.ref_name); | ||
let line = format!("{}\t{}\t{of}\t{sha}", r.status, r.workflow); | ||
// A workflow the change left out says so after. | ||
let skipped = r.skipped.as_deref().map(|why| format!("\t{why}")); | ||
let skipped = skipped.unwrap_or_default(); | ||
writeln!(out, "{line}\t{}{skipped}", r.started_at)?; | ||
writeln!(out, "{}", check_line(c))?; | ||
if let Some(text) = &c.log_text { | ||
writeln!(out, "{}", text.trim_end())?; | ||
} | ||
| @@ -2406,6 +2464,30 @@ async fn command( | ||
} | ||
} | ||
} | ||
("run", "log") => { | ||
let url = args.url()?; | ||
let (mut refname, mut commit, mut pull, mut workflow) = (None, None, None, None); | ||
let (mut queue, mut follow) = (None, false); | ||
while let Some(flag) = args.rest.next() { | ||
match flag.as_str() { | ||
"--queue" => queue = Some(format!("refs/heads/{}", args.next("<branch>")?)), | ||
"--ref" => refname = Some(args.next("<ref>")?), | ||
"--commit" => commit = Some(crate::checks::commit(args.next("<commit>")?)?), | ||
"--pull" => pull = Some(args.next("<at-uri>")?), | ||
"--workflow" => workflow = Some(args.next("<name>")?), | ||
"--follow" => follow = true, | ||
other => bail!("unexpected {other}"), | ||
} | ||
} | ||
let filter = crate::checks::Filter { | ||
refname, | ||
commit: commit.as_deref(), | ||
pull, | ||
workflow, | ||
queue: queue.as_deref(), | ||
}; | ||
run_log(ctx, &url, filter, follow, out).await?; | ||
} | ||
("run", "rerun") => { | ||
let url = args.url()?; | ||
let (mut refname, mut pull, mut workflow) = (None, None, None); | ||
crates/knotless/src/space.rs+10−4
| @@ -9,7 +9,7 @@ | ||
use crate::crypto::{self, P256Key}; | ||
use crate::env::Ctx; | ||
use crate::fix::{Fix, Refused}; | ||
use crate::identity; | ||
use crate::identity::{self, Identity}; | ||
use crate::keys; | ||
use crate::session::{self, Session}; | ||
use crate::store; | ||
| @@ -406,14 +406,20 @@ fn kept_at(ctx: &Ctx, uri: &str, did: &str) -> std::path::PathBuf { | ||
/// Joins the space `uri` as whoever is logged in, to read its members' | ||
/// repos: its authority is the account it is under. | ||
pub async fn joined(ctx: &Ctx, uri: &str) -> Result<Space> { | ||
let (id, me) = reader(ctx, uri).await?; | ||
let host = Xrpc::new(ctx.http.clone(), &id.pds); | ||
Space::join(ctx, &host, uri, &id.did, &me).await | ||
} | ||
| ||
/// The authority of the space `uri`, and whoever reads it: its owner, if | ||
/// that is who is logged in, else whoever is. | ||
pub async fn reader(ctx: &Ctx, uri: &str) -> Result<(Identity, Session)> { | ||
let authority = uri.trim_start_matches("at://").split('/').next(); | ||
let authority = authority.unwrap_or_default(); | ||
let id = identity::resolve(ctx, authority).await?; | ||
// As whoever reads it: its owner, if that is who is logged in. | ||
let me = session::reader(ctx, &id).await; | ||
let me = me.ok_or_else(|| Refused::new("not logged in", [Fix::Login]))?; | ||
let host = Xrpc::new(ctx.http.clone(), &id.pds); | ||
Space::join(ctx, &host, uri, &id.did, &me).await | ||
Ok((id, me)) | ||
} | ||
| ||
/// Shows the appview `did`'s private repository `name`, as whoever is | ||
crates/knotless-mcp/src/lib.rs+65−1
| @@ -89,6 +89,7 @@ pub const READS: &[&str] = &[ | ||
"theme_list", | ||
"theme_check", | ||
"run_list", | ||
"run_log", | ||
"workflow_list", | ||
"runner_list", | ||
"runner_minutes", | ||
| @@ -412,6 +413,33 @@ pub struct Checks { | ||
pub wait: Option<u64>, | ||
} | ||
| ||
#[derive(Deserialize, schemars::JsonSchema)] | ||
pub struct RunLog { | ||
/// `handle/name`, `did/name` or `knotless://handle/name`. | ||
pub repo: String, | ||
/// Only checks of pushes to this ref, and runs asked for there, such | ||
/// as `refs/heads/main`, or a branch or tag named bare, as `main`. | ||
#[serde(rename = "ref")] | ||
pub refname: Option<String>, | ||
/// Only checks of this commit: its hex object ID, or at least its | ||
/// first 4 digits. | ||
pub commit: Option<String>, | ||
/// Only the runs of this pull's head, by the pull's AT-URI. | ||
pub pull: Option<String>, | ||
/// Only the runs of this workflow, by name: needed when the others | ||
/// find checks of several. | ||
pub workflow: Option<String>, | ||
/// Only the runs of batches of this branch's merge queue, by the | ||
/// branch's name, such as `main`. | ||
pub queue: Option<String>, | ||
/// The `cursor` an earlier answer gave: only what the run printed | ||
/// after it. | ||
pub since: Option<String>, | ||
/// Seconds to wait, at most 300, for the check to show and for it to | ||
/// print more after `since`, rather than asking again. | ||
pub wait: Option<u64>, | ||
} | ||
| ||
#[derive(Deserialize, schemars::JsonSchema)] | ||
pub struct RepoHandle { | ||
/// `handle/name`, `did/name` or `knotless://handle/name`. | ||
| @@ -3758,6 +3786,42 @@ impl Server { | ||
Ok(answered) | ||
} | ||
| ||
#[tool( | ||
description = "The output of the newest check of a repository's workflows that ref, pull or queue, workflow and commit find: give workflow when they find several. While it runs, what it printed so far, as the appview holds it (`text`, its last 8000 bytes), with `steps`, those that ended, `going`, the step running, `ended`, whether its steps are over, and `cursor`: pass it as since to get only what it printed after, and wait, seconds, to wait for more. Once it finished, the end of its log as `text`. Each answer has the check as run_list gives it." | ||
)] | ||
async fn run_log(&self, Parameters(p): Parameters<RunLog>) -> Result<CallToolResult, McpError> { | ||
answer( | ||
async { | ||
let queue = p.queue.as_ref().map(|b| format!("refs/heads/{b}")); | ||
let commit = p.commit.as_deref().map(checks::commit).transpose()?; | ||
let filter = checks::Filter { | ||
refname: p.refname.as_deref(), | ||
commit: commit.as_deref(), | ||
pull: p.pull.as_deref(), | ||
workflow: p.workflow.as_deref(), | ||
queue: queue.as_deref(), | ||
}; | ||
let wait = Duration::from_secs(p.wait.unwrap_or(0).min(MOST_WAIT)); | ||
let deadline = tokio::time::Instant::now() + wait; | ||
let url = repo_url(&p.repo)?; | ||
let check = checks::newest(&self.ctx, &url, filter, wait).await?; | ||
if !records::status::going(&check.record.status) { | ||
let mut check = checks::again(&self.ctx, &url, filter, &check, true).await?; | ||
let text = check.log_text.take().map(tail).unwrap_or_default(); | ||
return Ok(json!({"check": check, "text": text})); | ||
} | ||
let left = deadline.saturating_duration_since(tokio::time::Instant::now()); | ||
let since = p.since.as_deref(); | ||
let mut said = checks::output(&self.ctx, &check, since, left).await?; | ||
said.text = tail(said.text); | ||
let mut said = serde_json::to_value(said)?; | ||
said["check"] = serde_json::to_value(check)?; | ||
Ok(said) | ||
} | ||
.await, | ||
) | ||
} | ||
| ||
#[tool( | ||
description = "Ask a repository's runners, as its owner, to run a ref's or a pull's workflows again, where it points now: give one of ref or pull. workflow runs only that one: of a pull's, such as one that failed alone; at a ref, as one that runs only when asked (`on: {asked: {}}`) does, its steps seeing env. Their new checks follow under run_list." | ||
)] | ||
| @@ -4275,7 +4339,7 @@ Collaboration: pull_create (write it from pull_template), pull_list, pull_show, | ||
| ||
Resources: a repository's pulls and issues, a pull or an issue, a pull's diff and a file at a revision can be read as MCP resources too, by URIs that mirror their pages on knotless.dev (as knotless://handle/name/pull/<did>/<rkey>: see the resource templates), each answered as the tool that reads it answers. | ||
| ||
Workflows are YAML files under .knotless/workflows/, run by the runners a repository names (runner_list, runner_add, runner_remove, runner_minutes) on the commit a push or a pull brings: `on: {push: {refs: [refs/heads/*]}, pull: {base: [refs/heads/main], paths: [web/**]}}` (a file with no `on` runs on every push to a branch; `paths`, globs, runs it only on a change touching one, and passes it as skipped otherwise), `image` (the one container all the steps run in, in turn, so a step sees what earlier ones installed), `env`, `secrets` (names; never given to a pull; see secret_set, secret_list and secret_remove), and `steps: [{name, run, timeout}]`, each `run` a shell script, stopping at the first that fails. After a push, run_list says whether they passed, workflow_list which a ref has and what runs each, and run_rerun asks for them again, or, as owner, for one of a pull's workflows by name, or one at a ref, with variables: one with `on: {asked: {refs: [refs/heads/main]}}` runs only so. Agents, accounts whose pulls the runners check as code no one vouches for: agent_list, agent_add, agent_remove. | ||
Workflows are YAML files under .knotless/workflows/, run by the runners a repository names (runner_list, runner_add, runner_remove, runner_minutes) on the commit a push or a pull brings: `on: {push: {refs: [refs/heads/*]}, pull: {base: [refs/heads/main], paths: [web/**]}}` (a file with no `on` runs on every push to a branch; `paths`, globs, runs it only on a change touching one, and passes it as skipped otherwise), `image` (the one container all the steps run in, in turn, so a step sees what earlier ones installed), `env`, `secrets` (names; never given to a pull; see secret_set, secret_list and secret_remove), and `steps: [{name, run, timeout}]`, each `run` a shell script, stopping at the first that fails. After a push, run_list says whether they passed, run_log what one running prints as it goes, workflow_list which a ref has and what runs each, and run_rerun asks for them again, or, as owner, for one of a pull's workflows by name, or one at a ref, with variables: one with `on: {asked: {refs: [refs/heads/main]}}` runs only so. Agents, accounts whose pulls the runners check as code no one vouches for: agent_list, agent_add, agent_remove. | ||
| ||
Moderation: mod_hide, mod_block, mod_blocks and mod_report. Accounts: account_follow, account_follows, account_followers, account_search, account_profile (how you show on knotless), account_email, account_email_add, account_email_remove (which addresses show commits as yours), account_delegate (let a gateway write for you) and account_delete_data. Themes for knotless's pages: theme_check, theme_publish, theme_use, theme_delete and theme_list. What others did that concerns you: notification_list; choose which you hear of with notification_settings, and watch a repository, or subscribe to or mute a pull or an issue, with notification_watch, notification_subscribe, notification_mute and notification_unsubscribe (see notification_subscriptions). Packages: release_publish a version from a repository, see its versions with release_list, read a file of one with release_read, and withdraw one with release_yank. Organizations: org_add, org_remove, org_join, org_leave, org_members, org_list and org_delete, once you have delegated to a gateway with account_delegate. Operators of a server: admin_takedown, admin_status, admin_reports, admin_stats and admin_logs. status says whether a server is well. An error naming a `knotless` command names a tool here: `knotless org join` is org_join, `knotless pull merge` is pull_merge, and so on, but for login, logout, upgrade, repo mirror, workflow pin and release download, which act on the machine they run on."; | ||
| ||
crates/knotless-mcp/tests/mcp.rs+137−0
| @@ -2696,6 +2696,143 @@ fn agents_make_themes() { | ||
} | ||
} | ||
| ||
/// An agent reads what a check prints as it runs, from the appview: so | ||
/// far, then only what came after a cursor, waited for; and once a check | ||
/// ended, its log. | ||
#[test] | ||
fn agents_follow_a_run() { | ||
let rt = tokio::runtime::Runtime::new().unwrap(); | ||
let (base, _appview, _proxy) = appview(&rt, ""); | ||
let mut owner = logged_in(); | ||
owner.set("KNOTLESS_APPVIEW", &base); | ||
owner.set("KNOTLESS_DID_WEB_SCHEME", "http"); | ||
let o = Client::new(&owner); | ||
let repo = format!("{}/watched", owner.account.handle); | ||
let start = json!({"repo": repo, "branch": "main", "message": "start", "changes": [{"path": "a", "content": "a\n"}]}); | ||
o.json("repo_commit", start); | ||
// Its owner runs its own. | ||
let handle = owner.account.handle.clone(); | ||
o.json("runner_add", json!({"repo": repo, "handle": handle})); | ||
let did = owner.account.did.clone(); | ||
let key = owner.record("dev.knotless.repo", "watched")["key"] | ||
.as_str() | ||
.unwrap() | ||
.to_string(); | ||
let at = format!("at://{did}/dev.knotless.repo/watched"); | ||
let (commit, started) = ("a".repeat(40), "2026-11-09T00:00:00.000Z"); | ||
let file = |workflow: &str, status: &str, log: Option<Value>| { | ||
let main = "refs/heads/main"; | ||
let rkey = knotless::keys::check_rkey(&did, &key, main, &commit, None, workflow); | ||
let mut check = json!({ | ||
"$type": "dev.knotless.check", "repo": {"uri": at, "key": key}, "ref": main, | ||
"commit": commit, "workflow": workflow, "status": status, "startedAt": started, | ||
}); | ||
if let Some(log) = log { | ||
check["log"] = log; | ||
} | ||
owner.put_record("dev.knotless.check", &rkey, check); | ||
format!("at://{did}/dev.knotless.check/{rkey}") | ||
}; | ||
let log = owner.upload_blob(b"==> lint\nclean\n", "text/plain"); | ||
file("lint", "success", Some(log)); | ||
let ci = file("ci", "running", None); | ||
// Once the appview has it, as it shows only the checks it has. | ||
let shown = format!("{base}/xrpc/dev.knotless.getCheck?repo={at}&uri={ci}"); | ||
wait_until(|| reqwest::blocking::get(&shown).unwrap().status() == 200); | ||
// Its runner sends the appview what it prints. | ||
let appview_did = reqwest::blocking::get(format!("{base}/.well-known/did.json")) | ||
.unwrap() | ||
.json::<Value>() | ||
.unwrap()["id"] | ||
.as_str() | ||
.unwrap() | ||
.to_string(); | ||
let token = rt.block_on(async { | ||
let ctx = ctx(&owner); | ||
let me = knotless::session::me(&ctx).await.unwrap(); | ||
let output = "dev.knotless.putRunOutput"; | ||
let token = knotless::auth::service_token(&ctx, &me, &appview_did, output); | ||
token.await.unwrap() | ||
}); | ||
let send = |offset: u64, text: &str, going: Option<&str>, ended: bool| { | ||
let body = json!({ | ||
"check": ci, "startedAt": started, "offset": offset, "text": text, | ||
"steps": [], "going": going, "ended": ended, | ||
}); | ||
let sent = reqwest::blocking::Client::new() | ||
.post(format!("{base}/xrpc/dev.knotless.putRunOutput")) | ||
.bearer_auth(&token) | ||
.json(&body) | ||
.send() | ||
.unwrap(); | ||
assert_eq!(sent.status(), 200); | ||
}; | ||
| ||
// Checks of several workflows: one is to be named. | ||
let (text, err) = o.call("run_log", json!({"repo": repo, "ref": "main"})); | ||
assert!(err, "{text}"); | ||
assert!( | ||
text.contains("are of several workflows, ci, lint: name one"), | ||
"{text}" | ||
); | ||
// Running, what it printed so far, and the step going. | ||
send(0, "==> build\nhello\n", Some("build"), false); | ||
let ci_log = json!({"repo": repo, "workflow": "ci"}); | ||
let first = o.json("run_log", ci_log.clone()); | ||
assert_eq!(first["text"], "==> build\nhello\n", "{first}"); | ||
assert_eq!( | ||
(&first["going"], &first["ended"]), | ||
(&json!("build"), &json!(false)) | ||
); | ||
assert_eq!(first["check"]["status"], "running"); | ||
assert_eq!(first["check"]["uri"], ci.as_str()); | ||
let cursor = first["cursor"].as_str().unwrap(); | ||
// After it, nothing more yet, though waited for longer than the | ||
// appview waits at once: asked again for the rest. | ||
let after = |wait: u64| { | ||
let mut args = ci_log.clone(); | ||
args["since"] = json!(cursor); | ||
args["wait"] = json!(wait); | ||
o.json("run_log", args) | ||
}; | ||
let none = after(27); | ||
assert_eq!( | ||
(&none["text"], &none["cursor"]), | ||
(&json!(""), &json!(cursor)) | ||
); | ||
// Then only what came after, and that it ended. | ||
send(16, "more\n", None, true); | ||
let next = after(0); | ||
assert_eq!( | ||
(&next["text"], &next["ended"]), | ||
(&json!("more\n"), &json!(true)) | ||
); | ||
assert!(next.get("going").is_none(), "{next}"); | ||
// Ended, its log. | ||
let lint = o.json("run_log", json!({"repo": repo, "workflow": "lint"})); | ||
assert_eq!(lint["text"], "==> lint\nclean\n", "{lint}"); | ||
assert_eq!(lint["check"]["status"], "success"); | ||
// None found, once waited for; a merge queue's batches likewise. | ||
let none = "b".repeat(40); | ||
let (text, err) = o.call("run_log", json!({"repo": repo, "commit": none, "wait": 1})); | ||
assert!( | ||
err && text.contains(&format!("no checks of commit {none}")), | ||
"{text}" | ||
); | ||
let (text, err) = o.call("run_log", json!({"repo": repo, "queue": "main"})); | ||
assert!( | ||
err && text.contains("no checks of queue refs/heads/main"), | ||
"{text}" | ||
); | ||
let (text, err) = o.call("run_log", json!({"repo": repo, "commit": "main"})); | ||
assert!(err && text.contains("a commit is its hex ID"), "{text}"); | ||
// Without an appview there is no one to ask while it runs. | ||
let mut alone = logged_in(); | ||
alone.set("KNOTLESS_APPVIEW", ""); | ||
let (why, err) = Client::new(&alone).call("run_log", ci_log); | ||
assert!(err && why.contains("KNOTLESS_APPVIEW"), "{why}"); | ||
} | ||
| ||
/// An agent reacts to an issue and a comment, sees the counts the appview | ||
/// keeps in issue_show, and takes a reaction back; the command line does | ||
/// the same. | ||
crates/knotless-run/tests/run.rs+87−1
| @@ -5001,6 +5001,32 @@ fn a_private_repository_runs_in_its_space() { | ||
.unwrap(); | ||
assert_eq!(public, r#"{"records":[]}"#); | ||
| ||
// What a run prints shows to a member as it runs, read from the | ||
// appview as one, with a delegation token for the space. | ||
let go = runner.path("go"); | ||
std::fs::create_dir_all(&go).unwrap(); | ||
let hold = format!( | ||
"steps:\n - name: hold\n run: |\n echo held\n until [ -f {}/go ]; do sleep 0.1; done\n", | ||
go.display() | ||
); | ||
std::fs::write(work.join(".knotless/workflows/hold.yml"), hold).unwrap(); | ||
owner.git(&work, &["add", "."]); | ||
owner.git(&work, &["commit", "-q", "-m", "hold"]); | ||
owner.git(&work, &["push", "-q", &url, "main"]); | ||
let log = ["run", "log", &url, "--workflow", "hold"]; | ||
wait_until(|| owner.cli(&log, b"").1.contains("held\n")); | ||
let (code, out) = owner.cli(&log, b""); | ||
assert_eq!(code, 0, "{out}"); | ||
assert!(out.ends_with("\tstep hold"), "{out}"); | ||
std::fs::write(go.join("go"), "").unwrap(); | ||
let (code, out) = owner.cli(&[&log[..], &["--follow"][..]].concat(), b""); | ||
assert_eq!(code, 0, "{out}"); | ||
let last = out.lines().last().unwrap(); | ||
assert!( | ||
last.starts_with("success\thold\trefs/heads/main\t"), | ||
"{out}" | ||
); | ||
| ||
// Deleted, its space with it, it is left rather than asked forever. | ||
// The space is asked itself once a minute while the appview passes on | ||
// its writes, so this may take longer than a wait. | ||
| @@ -5519,7 +5545,67 @@ fn a_run_shows_as_it_runs() { | ||
.contains("event: going\ndata: \"first\"") | ||
}); | ||
assert!(!said.lock().unwrap().contains("event: ended")); | ||
std::fs::write(go.join("go"), "").unwrap(); | ||
// The command line reads it from the appview too: what it printed so | ||
// far, masked, then the check's line and the step going. | ||
let url = owner.url("proj"); | ||
let log = ["run", "log", &url, "--ref", "main", "--workflow", "live"]; | ||
let (code, out) = owner.cli(&log, b""); | ||
assert_eq!(code, 0, "{out}"); | ||
assert!( | ||
out.contains("first ***\n") && !out.contains("s3cret"), | ||
"{out}" | ||
); | ||
let last = out.lines().last().unwrap(); | ||
assert!( | ||
last.starts_with("running\tlive\trefs/heads/main\t"), | ||
"{out}" | ||
); | ||
assert!(last.ends_with("\tstep first"), "{out}"); | ||
// Followed, it prints what comes until the run ends, then its result. | ||
let follow_log = [&log[..], &["--follow"][..]].concat(); | ||
let (code, out) = std::thread::scope(|s| { | ||
let following = s.spawn(|| owner.cli(&follow_log, b"")); | ||
std::thread::sleep(Duration::from_secs(1)); | ||
std::fs::write(go.join("go"), "").unwrap(); | ||
following.join().unwrap() | ||
}); | ||
assert_eq!(code, 0, "{out}"); | ||
assert!( | ||
out.contains("==> second") && out.contains("second\n"), | ||
"{out}" | ||
); | ||
let last = out.lines().last().unwrap(); | ||
assert!( | ||
last.starts_with("success\tlive\trefs/heads/main\t"), | ||
"{out}" | ||
); | ||
assert!(!last.contains("step"), "{out}"); | ||
// Ended, it prints its log. | ||
let (code, out) = owner.cli(&log, b""); | ||
assert_eq!(code, 0, "{out}"); | ||
assert!( | ||
out.contains("first ***\n") && out.contains("second\n"), | ||
"{out}" | ||
); | ||
assert!(out.lines().last().unwrap().starts_with("success\tlive\t")); | ||
// It takes the flags `run list` filters by, and none other. | ||
let none = [ | ||
"run", | ||
"log", | ||
&url, | ||
"--ref", | ||
"main", | ||
"--commit", | ||
"abcd", | ||
"--pull", | ||
&uri, | ||
"--queue", | ||
"main", | ||
"--workflow", | ||
"live", | ||
]; | ||
assert_eq!(owner.cli(&none, b"").0, 1); | ||
assert_eq!(owner.cli(&["run", "log", &url, "--bogus"], b"").0, 1); | ||
wait_until(|| said.lock().unwrap().contains("event: ended\ndata: null")); | ||
let said = said.lock().unwrap().clone(); | ||
assert!(!said.contains("s3cret"), "{said}"); | ||
crates/knotless-serve/src/web/live.rs+64−4
| @@ -3,18 +3,19 @@ | ||
//! again, and a running check's output as its runner sends it | ||
//! ([`crate::live`]). Each is followed only by whoever may read it, as its | ||
//! page is, and for ten minutes at most: the browser then asks again, and is | ||
//! proved again. | ||
//! proved again. A client that follows no events, as the command line, | ||
//! asks for a check's output as a query instead (`getRunOutput`). | ||
| ||
use super::{CheckParams, Result, indexed, named_check, not_found, repo}; | ||
use super::{CheckParams, Result, answer, indexed, named_check, not_found, repo}; | ||
use crate::App; | ||
use crate::live::Run; | ||
use crate::live::{Kept, Run}; | ||
use axum::extract::{Query, State}; | ||
use axum::http::HeaderMap; | ||
use axum::response::sse::{Event, KeepAlive, Sse}; | ||
use axum::response::{IntoResponse, Response}; | ||
use futures::{Stream, StreamExt, stream}; | ||
use serde::Deserialize; | ||
use serde_json::Value; | ||
use serde_json::{Value, json}; | ||
use std::convert::Infallible; | ||
use std::sync::Arc; | ||
use std::time::Duration; | ||
| @@ -31,6 +32,9 @@ const NOTHING: &str = "null"; | ||
/// as a push or a run makes, draws a page again once. | ||
const SETTLE: Duration = Duration::from_millis(250); | ||
| ||
/// The longest `dev.knotless.getRunOutput` waits for more. | ||
const MOST_WAIT: Duration = Duration::from_secs(25); | ||
| ||
#[derive(Deserialize)] | ||
pub struct RepoParams { | ||
repo: String, | ||
| @@ -135,6 +139,62 @@ impl Following { | ||
} | ||
} | ||
| ||
#[derive(Deserialize)] | ||
pub struct Since { | ||
cursor: Option<String>, | ||
wait: Option<u64>, | ||
} | ||
| ||
/// `dev.knotless.getRunOutput`: what `/_live/check` sends of a check's | ||
/// run, as one answer, for a client that follows no events: what it | ||
/// printed after `cursor`, all that is kept with none, the steps that | ||
/// ended, the one going, and whether it ended. With nothing new and the | ||
/// run going, it waits up to `wait` seconds for more. Its cursor names the | ||
/// run, by when it started, and the byte where what it gave ends. | ||
pub async fn get_run_output( | ||
State(app): State<Arc<App>>, | ||
Query(p): Query<CheckParams>, | ||
Query(since): Query<Since>, | ||
) -> Result<Response> { | ||
named_check(&app, &p).await?; | ||
let full = not_found(anyhow::anyhow!("too many runs followed")); | ||
let run = app.live.run(&p.uri).ok_or(full)?; | ||
let mut news = run.news.subscribe(); | ||
let cursor = since.cursor.as_deref(); | ||
let fresh = |kept: &Kept| kept.ended || !kept.since(had(kept, cursor)).1.is_empty(); | ||
if !fresh(&run.kept()) { | ||
let wait = Duration::from_secs(since.wait.unwrap_or(0)).min(MOST_WAIT); | ||
let _ = tokio::time::timeout(wait, news.recv()).await; | ||
} | ||
// A rerun at the same commit, started meanwhile, is the run now. | ||
let run = app.live.run(&p.uri).unwrap_or(run); | ||
let kept = run.kept(); | ||
let (at, text) = kept.since(had(&kept, cursor)); | ||
let started = kept.started.as_deref().unwrap_or_default(); | ||
let end = at + text.len() as u64; | ||
let steps = kept.steps.as_array().cloned().unwrap_or_default(); | ||
let mut said = json!({ | ||
"text": String::from_utf8_lossy(text), | ||
"cursor": format!("{started}/{end}"), | ||
"steps": steps, | ||
"ended": kept.ended, | ||
}); | ||
// Between steps, none is going. | ||
if let Some(going) = &kept.going { | ||
said["going"] = json!(going); | ||
} | ||
Ok(answer(said)) | ||
} | ||
| ||
/// Where `cursor`, as [`get_run_output`] gives it, says what was had of | ||
/// the run `kept` ends: nowhere for none, or for one of another run of the | ||
/// check, which is read from its start. | ||
fn had(kept: &Kept, cursor: Option<&str>) -> Option<u64> { | ||
let (started, at) = cursor?.rsplit_once('/')?; | ||
let other = kept.started.as_deref().is_some_and(|s| s != started); | ||
at.parse().ok().filter(|_| !other) | ||
} | ||
| ||
/// `events` as server-sent events, for [`FOLLOW`] or until the server | ||
/// stops. | ||
fn events(app: &App, events: impl Stream<Item = Event> + Send + 'static) -> Response { | ||
crates/knotless-serve/src/live.rs+5−4
| @@ -5,9 +5,10 @@ | ||
//! a blob of the end, so a page showing a check that runs would show | ||
//! nothing of it. Its runner sends this server what its steps print as they | ||
//! print it (`dev.knotless.putRunOutput`), and the server keeps the tail of | ||
//! each in memory for the pages that follow it ([`crate::web`]'s `/_live`), | ||
//! never in the index: the record and its log stay what is kept. A runner | ||
//! signs for itself with a service token, and sends only its own checks'. | ||
//! each in memory for the pages that follow it ([`crate::web`]'s `/_live`) | ||
//! and the clients that ask for it (`dev.knotless.getRunOutput`), never in | ||
//! the index: the record and its log stay what is kept. A runner signs for | ||
//! itself with a service token, and sends only its own checks'. | ||
//! | ||
//! Nothing written in a space reaches Jetstream, so a runner of a private | ||
//! repository would have to ask the space whether anything was. The | ||
| @@ -113,7 +114,7 @@ pub struct Kept { | ||
runner: String, | ||
heard: Option<Instant>, | ||
/// When the run sending started, as its runner says. | ||
started: Option<String>, | ||
pub started: Option<String>, | ||
} | ||
| ||
impl Kept { | ||
crates/knotless-serve/src/web.rs+1−0
| @@ -221,6 +221,7 @@ pub fn queries() -> Router<Arc<App>> { | ||
.route("/xrpc/dev.knotless.getQueue", get(queue::get_queue)) | ||
.route("/xrpc/dev.knotless.listRuns", get(runs::list_runs)) | ||
.route("/xrpc/dev.knotless.getCheck", get(get_check)) | ||
.route("/xrpc/dev.knotless.getRunOutput", get(live::get_run_output)) | ||
.route("/xrpc/dev.knotless.getAttachment", get(get_attachment)) | ||
.route("/xrpc/dev.knotless.getCheckImage", get(get_check_image)) | ||
.route("/xrpc/dev.knotless.getAvatar", get(get_avatar)) | ||