| 1 | //! `g1t work`: coordinating work on issues. Who is on an issue (claims and |
| 2 | //! sessions), what it waits on, handing it over, and following what happens |
| 3 | //! in the repository as it happens. |
| 4 | |
| 5 | use serde_json::{Value, json}; |
| 6 | use std::collections::{BTreeSet, HashSet, VecDeque}; |
| 7 | use std::time::{Duration, Instant}; |
| 8 | |
| 9 | use crate::cli::{ |
| 10 | WorkBlockArgs, WorkClaimArgs, WorkCommand, WorkDispatchArgs, WorkEventsArgs, WorkHandoffArgs, |
| 11 | WorkNumber, WorkReleaseArgs, |
| 12 | }; |
| 13 | use crate::client::Client; |
| 14 | use crate::context::Ctx; |
| 15 | use crate::error::{Error, Kind, Result}; |
| 16 | use crate::output::{Output, ago, now, parse_time, s}; |
| 17 | use crate::repo::Repo; |
| 18 | |
| 19 | /// How long a follow asks the API to hold each call, in seconds. |
| 20 | const WAIT_SECONDS: u64 = 25; |
| 21 | /// The longest a follow waits before trying again after a failure. |
| 22 | const MAX_BACKOFF_SECONDS: u64 = 30; |
| 23 | /// How many event ids a follow remembers, to print none twice. |
| 24 | const REMEMBERED: usize = 1_000; |
| 25 | |
| 26 | pub fn run(ctx: &Ctx, command: WorkCommand) -> Result<()> { |
| 27 | match command { |
| 28 | WorkCommand::Context(args) => context(ctx, args), |
| 29 | WorkCommand::Claim(args) => claim(ctx, args), |
| 30 | WorkCommand::Release(args) => release(ctx, args), |
| 31 | WorkCommand::Handoff(args) => handoff(ctx, args), |
| 32 | WorkCommand::List => list(ctx), |
| 33 | WorkCommand::Block(args) => block(ctx, args, true), |
| 34 | WorkCommand::Unblock(args) => block(ctx, args, false), |
| 35 | WorkCommand::Events(args) => events(ctx, args), |
| 36 | WorkCommand::Dispatch(args) => dispatch(ctx, args), |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | /// `in 27m`, `in 2h5m`, or `expired` for a time still to come. |
| 41 | pub fn until(text: &str) -> String { |
| 42 | match parse_time(text) { |
| 43 | Some(then) if then > now() => format!("in {}", crate::output::duration(then - now())), |
| 44 | Some(_) => "expired".to_owned(), |
| 45 | None => text.to_owned(), |
| 46 | } |
| 47 | } |
| 48 | |
| 49 | /// One claim as a line: who, with what, since when, until when, and why. |
| 50 | pub fn claim_line(out: &Output, claim: &Value) -> String { |
| 51 | let who = super::login_of(claim); |
| 52 | let mut line = out.bold(if who.is_empty() { "someone" } else { &who }); |
| 53 | if let Some(harness) = claim["harness"].as_str().filter(|h| !h.is_empty()) { |
| 54 | line.push_str(&out.dim(&format!(" ({harness})"))); |
| 55 | } |
| 56 | if let Some(issue) = claim["issue"].as_u64() { |
| 57 | line.push_str(&format!(" on #{issue}")); |
| 58 | } |
| 59 | let state = s(claim, "state"); |
| 60 | if !state.is_empty() && state != "active" { |
| 61 | line.push_str(&format!(" {}", out.state(state))); |
| 62 | } |
| 63 | if let Some(created) = claim["created_at"].as_str() { |
| 64 | line.push_str(&out.dim(&format!(" since {}", ago(created)))); |
| 65 | } |
| 66 | if let Some(expires) = claim["expires_at"].as_str().filter(|_| state.is_empty() || state == "active") { |
| 67 | line.push_str(&out.dim(&format!(", expires {}", until(expires)))); |
| 68 | } |
| 69 | if let Some(to) = claim["handoff_to"].as_str() { |
| 70 | line.push_str(&format!(", handed to {to}")); |
| 71 | } |
| 72 | if let Some(note) = claim["note"].as_str().filter(|n| !n.trim().is_empty()) { |
| 73 | line.push_str(&format!(": {}", note.trim())); |
| 74 | } |
| 75 | line |
| 76 | } |
| 77 | |
| 78 | /// An issue another one waits on, as a line. |
| 79 | fn blocker_line(out: &Output, blocker: &Value) -> String { |
| 80 | let mut line = format!("#{} {} {}", blocker["number"], s(blocker, "title"), out.state(s(blocker, "state"))); |
| 81 | let claims = blocker["claims"].as_array().map(Vec::as_slice).unwrap_or_default(); |
| 82 | let who: Vec<String> = claims.iter().map(super::login_of).filter(|w| !w.is_empty()).collect(); |
| 83 | if !who.is_empty() { |
| 84 | line.push_str(&out.dim(&format!(", claimed by {}", who.join(", ")))); |
| 85 | } |
| 86 | let pulls: Vec<String> = blocker["pulls"] |
| 87 | .as_array() |
| 88 | .into_iter() |
| 89 | .flatten() |
| 90 | .map(|p| match p["status"].as_str().or_else(|| p["state"].as_str()) { |
| 91 | Some(status) => format!("#{} {status}", p["number"]), |
| 92 | None => format!("#{}", p["number"]), |
| 93 | }) |
| 94 | .collect(); |
| 95 | if !pulls.is_empty() { |
| 96 | line.push_str(&out.dim(&format!(", pull requests {}", pulls.join(", ")))); |
| 97 | } |
| 98 | line |
| 99 | } |
| 100 | |
| 101 | fn heading(out: &Output, title: &str) { |
| 102 | println!(); |
| 103 | println!("{}", out.bold(title)); |
| 104 | } |
| 105 | |
| 106 | fn context(ctx: &Ctx, args: WorkNumber) -> Result<()> { |
| 107 | let (repo, client) = super::coordination_client(ctx, super::QUICK)?; |
| 108 | let work = client.op("get_work").repo(&repo).arg("number", args.number).send()?; |
| 109 | if ctx.out.structured() { |
| 110 | return ctx.out.emit(&work); |
| 111 | } |
| 112 | print_context(&ctx.out, &repo, args.number, &work); |
| 113 | Ok(()) |
| 114 | } |
| 115 | |
| 116 | fn list_of<'a>(value: &'a Value, key: &str) -> &'a [Value] { |
| 117 | value[key].as_array().map(Vec::as_slice).unwrap_or_default() |
| 118 | } |
| 119 | |
| 120 | pub fn print_context(out: &Output, repo: &Repo, number: u64, work: &Value) { |
| 121 | let issue = &work["issue"]; |
| 122 | println!("{} {}", out.bold(s(issue, "title")), out.dim(&format!("{repo}#{number}"))); |
| 123 | let mut facts = vec![out.state(s(issue, "state"))]; |
| 124 | let assignees: Vec<String> = issue["assignees"] |
| 125 | .as_array() |
| 126 | .into_iter() |
| 127 | .flatten() |
| 128 | .filter_map(|a| a.as_str().map(str::to_owned).or_else(|| a["username"].as_str().map(str::to_owned))) |
| 129 | .collect(); |
| 130 | if !assignees.is_empty() { |
| 131 | facts.push(format!("assigned to {}", assignees.join(", "))); |
| 132 | } |
| 133 | println!("{}", facts.join(" · ")); |
| 134 | // get_work gives the items under the issue's "Definition of done". |
| 135 | let items: Vec<&str> = list_of(work, "definition_of_done").iter().filter_map(Value::as_str).collect(); |
| 136 | if !items.is_empty() { |
| 137 | heading(out, "Definition of done"); |
| 138 | for item in items { |
| 139 | println!(" - {item}"); |
| 140 | } |
| 141 | } else if let Some(done) = issue["definition_of_done"].as_str().filter(|d| !d.trim().is_empty()) { |
| 142 | heading(out, "Definition of done"); |
| 143 | for line in done.trim().lines() { |
| 144 | println!(" {line}"); |
| 145 | } |
| 146 | } |
| 147 | heading(out, "Claims"); |
| 148 | let claims = list_of(work, "claims"); |
| 149 | if claims.is_empty() { |
| 150 | println!(" {}", out.dim("Nobody has claimed it.")); |
| 151 | } |
| 152 | for claim in claims { |
| 153 | println!(" {}", claim_line(out, claim)); |
| 154 | } |
| 155 | let sessions = list_of(work, "sessions"); |
| 156 | if !sessions.is_empty() { |
| 157 | heading(out, "Sessions"); |
| 158 | for session in sessions { |
| 159 | let mut line = format!( |
| 160 | "{} {} {}", |
| 161 | s(session, "id"), |
| 162 | out.state(s(session, "state")), |
| 163 | super::login_of(session) |
| 164 | ); |
| 165 | if let Some(harness) = session["harness"].as_str() { |
| 166 | line.push_str(&out.dim(&format!(" ({harness})"))); |
| 167 | } |
| 168 | if let Some(branch) = session["branch"].as_str().filter(|b| !b.is_empty()) { |
| 169 | line.push_str(&format!(" on {branch}")); |
| 170 | } |
| 171 | if let Some(seen) = session["last_seen_at"].as_str() { |
| 172 | line.push_str(&out.dim(&format!(", last seen {}", ago(seen)))); |
| 173 | } |
| 174 | println!(" {line}"); |
| 175 | } |
| 176 | } |
| 177 | let pulls = list_of(work, "pulls"); |
| 178 | heading(out, "Pull requests"); |
| 179 | if pulls.is_empty() { |
| 180 | println!(" {}", out.dim("None yet.")); |
| 181 | } |
| 182 | for pull in pulls { |
| 183 | let status = pull["status"].as_str().or_else(|| pull["state"].as_str()).unwrap_or(""); |
| 184 | let mut line = format!("#{} {} {}", pull["number"], s(pull, "title"), out.state(status)); |
| 185 | let checks = pull["checks"]["status"] |
| 186 | .as_str() |
| 187 | .or_else(|| pull["checks"].as_str()) |
| 188 | .or_else(|| pull["check_status"].as_str()); |
| 189 | if let Some(checks) = checks { |
| 190 | line.push_str(&format!(", checks {}", out.state(checks))); |
| 191 | } |
| 192 | if let Some(position) = pull["queue_position"].as_u64().or_else(|| pull["queue"]["position"].as_u64()) { |
| 193 | line.push_str(&format!(", queued {position}")); |
| 194 | } |
| 195 | println!(" {line}"); |
| 196 | for overlap in list_of(pull, "overlaps") { |
| 197 | let other = overlap["pull"].as_u64().or_else(|| overlap["number"].as_u64()).unwrap_or(0); |
| 198 | let files: Vec<&str> = list_of(overlap, "files").iter().filter_map(Value::as_str).collect(); |
| 199 | println!( |
| 200 | " {} #{other}{}", |
| 201 | out.yellow("overlaps"), |
| 202 | if files.is_empty() { String::new() } else { format!(": {}", files.join(", ")) } |
| 203 | ); |
| 204 | } |
| 205 | } |
| 206 | let blocked_by = list_of(work, "blocked_by"); |
| 207 | if !blocked_by.is_empty() { |
| 208 | heading(out, "Blocked by"); |
| 209 | for blocker in blocked_by { |
| 210 | println!(" {}", blocker_line(out, blocker)); |
| 211 | } |
| 212 | } |
| 213 | let blocks = list_of(work, "blocks"); |
| 214 | if !blocks.is_empty() { |
| 215 | heading(out, "Blocks"); |
| 216 | for blocked in blocks { |
| 217 | println!(" #{} {} {}", blocked["number"], s(blocked, "title"), out.state(s(blocked, "state"))); |
| 218 | } |
| 219 | } |
| 220 | let failure = &work["last_integration_failure"]; |
| 221 | if failure.is_object() { |
| 222 | heading(out, "Last integration failure"); |
| 223 | println!(" {}", event_summary(failure.get("data").unwrap_or(failure))); |
| 224 | } |
| 225 | let decisions = list_of(work, "decisions"); |
| 226 | if !decisions.is_empty() { |
| 227 | heading(out, "Decisions"); |
| 228 | for decision in decisions { |
| 229 | println!(" {}", decision_line(out, decision)); |
| 230 | } |
| 231 | } |
| 232 | let recent = list_of(work, "recent"); |
| 233 | if !recent.is_empty() { |
| 234 | heading(out, "Recent"); |
| 235 | for event in recent { |
| 236 | println!(" {}", event_line(out, event)); |
| 237 | } |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | /// A decision as a line: how sure, what kind, what it says, who and when. |
| 242 | pub fn decision_line(out: &Output, decision: &Value) -> String { |
| 243 | let confidence = s(decision, "confidence"); |
| 244 | let shown = match confidence { |
| 245 | "approved" => out.green(confidence), |
| 246 | "tested" => out.cyan(confidence), |
| 247 | _ => out.dim(confidence), |
| 248 | }; |
| 249 | let mut line = format!("[{shown}] {}: {}", s(decision, "kind"), s(decision, "summary")); |
| 250 | let who = super::login_of(decision); |
| 251 | let when = decision["created_at"].as_str().map(ago).unwrap_or_default(); |
| 252 | let by = [who, when].into_iter().filter(|p| !p.is_empty()).collect::<Vec<_>>().join(" "); |
| 253 | if !by.is_empty() { |
| 254 | line.push_str(&out.dim(&format!(" ({by})"))); |
| 255 | } |
| 256 | line |
| 257 | } |
| 258 | |
| 259 | fn claim(ctx: &Ctx, args: WorkClaimArgs) -> Result<()> { |
| 260 | let (repo, client) = super::coordination_client(ctx, super::QUICK)?; |
| 261 | let found = client |
| 262 | .op("claim_issue") |
| 263 | .repo(&repo) |
| 264 | .arg("number", args.number) |
| 265 | .field_opt("session_id", super::session::current()) |
| 266 | .field_opt("note", args.note.clone()) |
| 267 | .field_opt("ttl_minutes", args.ttl) |
| 268 | .send()?; |
| 269 | if ctx.out.structured() { |
| 270 | return ctx.out.emit(&found); |
| 271 | } |
| 272 | let out = &ctx.out; |
| 273 | let claim = &found["claim"]; |
| 274 | match claim["expires_at"].as_str() { |
| 275 | Some(expires) => println!("Claimed #{}; it expires {} unless renewed.", args.number, until(expires)), |
| 276 | None => println!("Claimed #{}.", args.number), |
| 277 | } |
| 278 | let others = list_of(&found, "others"); |
| 279 | if !others.is_empty() { |
| 280 | println!("{}", out.yellow(&format!("#{} is claimed by others too:", args.number))); |
| 281 | for other in others { |
| 282 | println!(" {}", claim_line(out, other)); |
| 283 | } |
| 284 | } |
| 285 | let waiting: Vec<&Value> = list_of(&found, "blocked_by") |
| 286 | .iter() |
| 287 | .filter(|b| !matches!(b["state"].as_str(), Some("closed" | "merged"))) |
| 288 | .collect(); |
| 289 | if !waiting.is_empty() { |
| 290 | println!("{}", out.yellow(&format!("#{} waits on:", args.number))); |
| 291 | for blocker in waiting { |
| 292 | println!(" {}", blocker_line(out, blocker)); |
| 293 | } |
| 294 | } |
| 295 | Ok(()) |
| 296 | } |
| 297 | |
| 298 | /// The time a dispatch may take: starting a Managed Agents session makes |
| 299 | /// several calls to the provider. |
| 300 | const DISPATCH_TIMEOUT: Duration = Duration::from_secs(60); |
| 301 | |
| 302 | fn dispatch(ctx: &Ctx, args: WorkDispatchArgs) -> Result<()> { |
| 303 | let (repo, client) = super::coordination_client(ctx, DISPATCH_TIMEOUT)?; |
| 304 | let handoff = if args.handoff { |
| 305 | Some(true) |
| 306 | } else if args.managed { |
| 307 | Some(false) |
| 308 | } else { |
| 309 | None |
| 310 | }; |
| 311 | let found = client |
| 312 | .op("start_agent_session") |
| 313 | .repo(&repo) |
| 314 | .field_opt("issue", args.number) |
| 315 | .field_opt("title", args.title.clone()) |
| 316 | .field_opt("body", args.body.clone()) |
| 317 | .field_opt("instruction", args.instruction.clone()) |
| 318 | .field_opt("mode", args.mode.map(|mode| mode.name())) |
| 319 | .field_opt("handoff", handoff) |
| 320 | .send()?; |
| 321 | if ctx.out.structured() { |
| 322 | return ctx.out.emit(&found); |
| 323 | } |
| 324 | let issue = found["issue"].as_u64().unwrap_or_default(); |
| 325 | if found["handoff"].as_bool() == Some(true) { |
| 326 | eprintln!("Run this where the repository is checked out, with g1t's Claude Code plugin, to work on #{issue}:"); |
| 327 | println!("{}", s(&found, "command")); |
| 328 | return Ok(()); |
| 329 | } |
| 330 | let out = &ctx.out; |
| 331 | println!("{}", s(&found, "session_url")); |
| 332 | eprintln!( |
| 333 | "Started {} on #{issue} on the workspace's Anthropic account.", |
| 334 | found["session"]["id"].as_str().unwrap_or("a session") |
| 335 | ); |
| 336 | if let Some(url) = found["provider_session_url"].as_str() { |
| 337 | eprintln!("{} {url}", out.dim("On Anthropic:")); |
| 338 | } |
| 339 | Ok(()) |
| 340 | } |
| 341 | |
| 342 | fn release(ctx: &Ctx, args: WorkReleaseArgs) -> Result<()> { |
| 343 | let (repo, client) = super::coordination_client(ctx, super::QUICK)?; |
| 344 | let found = client |
| 345 | .op("release_claim") |
| 346 | .repo(&repo) |
| 347 | .arg("number", args.number) |
| 348 | .field_opt("reason", args.reason.clone()) |
| 349 | .field_opt("session_id", super::session::current()) |
| 350 | .send()?; |
| 351 | if ctx.out.structured() { |
| 352 | return ctx.out.emit(&found); |
| 353 | } |
| 354 | eprintln!("Released your claim on #{}.", args.number); |
| 355 | Ok(()) |
| 356 | } |
| 357 | |
| 358 | fn handoff(ctx: &Ctx, args: WorkHandoffArgs) -> Result<()> { |
| 359 | if args.note.trim().is_empty() { |
| 360 | return Err(Error::usage("Say where it stands with --note.")); |
| 361 | } |
| 362 | let (repo, client) = super::coordination_client(ctx, super::QUICK)?; |
| 363 | let to = match &args.to { |
| 364 | Some(login) => Some(super::who(ctx, &client, login)?), |
| 365 | None => None, |
| 366 | }; |
| 367 | let found = client |
| 368 | .op("hand_off_issue") |
| 369 | .repo(&repo) |
| 370 | .arg("number", args.number) |
| 371 | .field_opt("to", to.clone()) |
| 372 | .field("note", args.note.trim()) |
| 373 | .field_opt("session_id", super::session::current()) |
| 374 | .send()?; |
| 375 | if ctx.out.structured() { |
| 376 | return ctx.out.emit(&found); |
| 377 | } |
| 378 | match to { |
| 379 | Some(login) => eprintln!("Handed #{} to {login}.", args.number), |
| 380 | None => eprintln!("Handed #{} to whoever takes it next.", args.number), |
| 381 | } |
| 382 | Ok(()) |
| 383 | } |
| 384 | |
| 385 | pub fn active_claims(client: &Client, repo: &Repo) -> Result<Vec<Value>> { |
| 386 | let found = client.op("list_claims").repo(repo).query("state", "active").send()?; |
| 387 | Ok(super::items(found, "claims")) |
| 388 | } |
| 389 | |
| 390 | fn list(ctx: &Ctx) -> Result<()> { |
| 391 | let (repo, client) = super::coordination_client(ctx, super::QUICK)?; |
| 392 | let (claims, sessions) = std::thread::scope(|scope| { |
| 393 | let claims = scope.spawn(|| active_claims(&client, &repo)); |
| 394 | let sessions = scope.spawn(|| -> Result<Vec<Value>> { |
| 395 | let found = client.op("list_sessions").repo(&repo).query("state", "active").send()?; |
| 396 | Ok(super::items(found, "sessions")) |
| 397 | }); |
| 398 | (claims.join(), sessions.join()) |
| 399 | }); |
| 400 | let failed = || Error::failed("A lookup failed."); |
| 401 | let claims = claims.map_err(|_| failed())??; |
| 402 | let sessions = sessions.map_err(|_| failed())??; |
| 403 | if ctx.out.structured() { |
| 404 | return ctx.out.emit(&json!({ "claims": claims, "sessions": sessions })); |
| 405 | } |
| 406 | let out = &ctx.out; |
| 407 | println!("{}", out.bold(&format!("Claims in {repo}"))); |
| 408 | if claims.is_empty() { |
| 409 | println!(" {}", out.dim("None.")); |
| 410 | } |
| 411 | for claim in &claims { |
| 412 | println!(" {}", claim_line(out, claim)); |
| 413 | } |
| 414 | println!(); |
| 415 | println!("{}", out.bold(&format!("Active sessions in {repo}"))); |
| 416 | if sessions.is_empty() { |
| 417 | println!(" {}", out.dim("None.")); |
| 418 | return Ok(()); |
| 419 | } |
| 420 | super::session::print_list(out, &sessions, super::session::current().as_deref()) |
| 421 | } |
| 422 | |
| 423 | /// The issues `number` waits on once `by` is added to `current` |
| 424 | /// (`add`) or taken from it. |
| 425 | pub fn blocked_set(number: u64, current: &[u64], by: &[u64], add: bool) -> Result<Vec<u64>> { |
| 426 | if add && by.contains(&number) { |
| 427 | return Err(Error::usage(format!("#{number} cannot wait on itself."))); |
| 428 | } |
| 429 | let mut set: BTreeSet<u64> = current.iter().copied().collect(); |
| 430 | for n in by { |
| 431 | if add { |
| 432 | set.insert(*n); |
| 433 | } else { |
| 434 | set.remove(n); |
| 435 | } |
| 436 | } |
| 437 | Ok(set.into_iter().collect()) |
| 438 | } |
| 439 | |
| 440 | fn block(ctx: &Ctx, args: WorkBlockArgs, add: bool) -> Result<()> { |
| 441 | let (repo, client) = super::coordination_client(ctx, super::QUICK)?; |
| 442 | let found = client.op("get_issue").repo(&repo).arg("number", args.number).send()?; |
| 443 | let issue = super::unwrap_key(found, "issue"); |
| 444 | let current: Vec<u64> = issue["blocked_by"] |
| 445 | .as_array() |
| 446 | .into_iter() |
| 447 | .flatten() |
| 448 | .filter_map(|n| n.as_u64().or_else(|| n["number"].as_u64())) |
| 449 | .collect(); |
| 450 | let wanted = blocked_set(args.number, ¤t, &args.by, add)?; |
| 451 | let updated = client |
| 452 | .op("update_issue") |
| 453 | .repo(&repo) |
| 454 | .arg("number", args.number) |
| 455 | .field("blocked_by", json!(wanted)) |
| 456 | .send()?; |
| 457 | if ctx.out.structured() { |
| 458 | return ctx.out.emit(&super::unwrap_key(updated, "issue")); |
| 459 | } |
| 460 | let list: Vec<String> = wanted.iter().map(|n| format!("#{n}")).collect(); |
| 461 | if list.is_empty() { |
| 462 | eprintln!("#{} waits on nothing.", args.number); |
| 463 | } else { |
| 464 | eprintln!("#{} waits on {}.", args.number, list.join(", ")); |
| 465 | } |
| 466 | Ok(()) |
| 467 | } |
| 468 | |
| 469 | // --- events ------------------------------------------------------------- |
| 470 | |
| 471 | /// `15m`, `2h`, `1d`, `1w` ago as an RFC 3339 time; anything else as given. |
| 472 | pub fn since_time(text: &str, now_secs: i64) -> String { |
| 473 | let ms = super::logs::since(text, now_secs * 1000); |
| 474 | if ms == text.trim() { |
| 475 | return ms; |
| 476 | } |
| 477 | match ms.parse::<i64>() { |
| 478 | Ok(ms) => format!("{}Z", super::logs::clock(ms * 1000).replacen(' ', "T", 1)), |
| 479 | Err(_) => ms, |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | /// The issue or pull request an event is about. |
| 484 | fn event_number(data: &Value) -> Option<u64> { |
| 485 | data["number"] |
| 486 | .as_u64() |
| 487 | .or_else(|| data["issue"].as_u64()) |
| 488 | .or_else(|| data["pull"].as_u64()) |
| 489 | .or_else(|| data["issue"]["number"].as_u64()) |
| 490 | .or_else(|| data["pull"]["number"].as_u64()) |
| 491 | .or_else(|| data["claim"]["issue"].as_u64()) |
| 492 | } |
| 493 | |
| 494 | /// What an event says, in a few words. |
| 495 | pub fn event_summary(data: &Value) -> String { |
| 496 | let nums = |key: &str| -> String { |
| 497 | data[key] |
| 498 | .as_array() |
| 499 | .into_iter() |
| 500 | .flatten() |
| 501 | .filter_map(|n| n.as_u64().or_else(|| n["number"].as_u64())) |
| 502 | .map(|n| format!("#{n}")) |
| 503 | .collect::<Vec<_>>() |
| 504 | .join(", ") |
| 505 | }; |
| 506 | if let Some(check) = data["check"].as_object() { |
| 507 | let mut line = format!( |
| 508 | "{} {}", |
| 509 | check.get("name").and_then(Value::as_str).unwrap_or("check"), |
| 510 | check.get("conclusion").and_then(Value::as_str).unwrap_or("") |
| 511 | ); |
| 512 | let combination = nums("combination"); |
| 513 | if !combination.is_empty() { |
| 514 | line.push_str(&format!(" with {combination}")); |
| 515 | } |
| 516 | let landed = nums("landed"); |
| 517 | if !landed.is_empty() { |
| 518 | line.push_str(&format!(" (since it branched: {landed} landed)")); |
| 519 | } |
| 520 | return line.trim().to_owned(); |
| 521 | } |
| 522 | if let Some(other) = data["other_pull"].as_u64() { |
| 523 | let files: Vec<&str> = data["files"].as_array().into_iter().flatten().filter_map(Value::as_str).collect(); |
| 524 | return format!("overlaps #{other}: {}", files.join(", ")); |
| 525 | } |
| 526 | if !data["blocked_by"].is_null() && data["blocked_by"].is_array() { |
| 527 | let list = nums("blocked_by"); |
| 528 | return if list.is_empty() { "waits on nothing".to_owned() } else { format!("waits on {list}") }; |
| 529 | } |
| 530 | for path in [ |
| 531 | &data["title"], |
| 532 | &data["summary"], |
| 533 | &data["decision"]["summary"], |
| 534 | &data["note"], |
| 535 | &data["claim"]["note"], |
| 536 | &data["session"]["title"], |
| 537 | &data["reason"], |
| 538 | ] { |
| 539 | if let Some(text) = path.as_str().filter(|t| !t.trim().is_empty()) { |
| 540 | return text.trim().lines().next().unwrap_or("").to_owned(); |
| 541 | } |
| 542 | } |
| 543 | String::new() |
| 544 | } |
| 545 | |
| 546 | /// One event as the cells of its line: time, type, number, summary, id. |
| 547 | pub fn event_cells(event: &Value) -> [String; 5] { |
| 548 | let time = s(event, "time"); |
| 549 | let time = time.get(..19).unwrap_or(time).replacen('T', " ", 1); |
| 550 | let data = &event["data"]; |
| 551 | let number = event_number(data).map(|n| format!("#{n}")).unwrap_or_default(); |
| 552 | [time, s(event, "type").to_owned(), number, event_summary(data), s(event, "id").to_owned()] |
| 553 | } |
| 554 | |
| 555 | /// One event as a line: on a terminal spaced and colored, piped |
| 556 | /// tab-separated. |
| 557 | pub fn event_line(out: &Output, event: &Value) -> String { |
| 558 | let [time, kind, number, summary, id] = event_cells(event); |
| 559 | if !out.tty { |
| 560 | return [time, kind, number, summary, id].join("\t"); |
| 561 | } |
| 562 | let painted = if kind.contains("failed") || kind.ends_with("expired") { |
| 563 | out.red(&kind) |
| 564 | } else if kind.starts_with("overlap") || kind.starts_with("dependency") || kind.contains("handoff") { |
| 565 | out.yellow(&kind) |
| 566 | } else if kind.ends_with("passed") || kind.ends_with("merged") { |
| 567 | out.green(&kind) |
| 568 | } else { |
| 569 | out.cyan(&kind) |
| 570 | }; |
| 571 | let mut line = format!("{} {painted}", out.dim(&time)); |
| 572 | if !number.is_empty() { |
| 573 | line.push_str(&format!(" {number}")); |
| 574 | } |
| 575 | if !summary.is_empty() { |
| 576 | line.push_str(&format!(" {summary}")); |
| 577 | } |
| 578 | line.push_str(&format!(" {}", out.dim(&id))); |
| 579 | line |
| 580 | } |
| 581 | |
| 582 | /// The ids a follow has printed, the oldest forgotten first. |
| 583 | #[derive(Default)] |
| 584 | pub struct Seen { |
| 585 | ids: HashSet<String>, |
| 586 | order: VecDeque<String>, |
| 587 | } |
| 588 | |
| 589 | impl Seen { |
| 590 | /// Whether `id` is new, remembering it. |
| 591 | pub fn insert(&mut self, id: &str) -> bool { |
| 592 | if !self.ids.insert(id.to_owned()) { |
| 593 | return false; |
| 594 | } |
| 595 | self.order.push_back(id.to_owned()); |
| 596 | if self.order.len() > REMEMBERED |
| 597 | && let Some(old) = self.order.pop_front() |
| 598 | { |
| 599 | self.ids.remove(&old); |
| 600 | } |
| 601 | true |
| 602 | } |
| 603 | |
| 604 | pub fn contains(&self, id: &str) -> bool { |
| 605 | self.ids.contains(id) |
| 606 | } |
| 607 | } |
| 608 | |
| 609 | /// How long to wait after `failures` failures in a row: 1, 2, 4... 30 s. |
| 610 | pub fn backoff(failures: u32) -> Duration { |
| 611 | let secs = 1u64 << failures.saturating_sub(1).min(5); |
| 612 | Duration::from_secs(secs.min(MAX_BACKOFF_SECONDS)) |
| 613 | } |
| 614 | |
| 615 | struct Query<'a> { |
| 616 | client: &'a Client, |
| 617 | repo: &'a Repo, |
| 618 | types: String, |
| 619 | numbers: String, |
| 620 | since: Option<String>, |
| 621 | } |
| 622 | |
| 623 | impl Query<'_> { |
| 624 | fn fetch(&self, after: Option<&str>, wait: Option<u64>, limit: usize, filtered: bool) -> Result<Vec<Value>> { |
| 625 | let mut call = self |
| 626 | .client |
| 627 | .op("list_events") |
| 628 | .repo(self.repo) |
| 629 | .query_opt("after", after) |
| 630 | .query_opt("wait", wait) |
| 631 | .query("limit", limit.clamp(1, 200)); |
| 632 | if filtered { |
| 633 | if !self.types.is_empty() { |
| 634 | call = call.query("types", &self.types); |
| 635 | } |
| 636 | if !self.numbers.is_empty() { |
| 637 | call = call.query("numbers", &self.numbers); |
| 638 | } |
| 639 | call = call.query_opt("since", self.since.clone()); |
| 640 | } |
| 641 | Ok(super::items(call.send()?, "events")) |
| 642 | } |
| 643 | } |
| 644 | |
| 645 | fn events(ctx: &Ctx, args: WorkEventsArgs) -> Result<()> { |
| 646 | let (repo, client) = super::coordination_client(ctx, std::time::Duration::from_secs(WAIT_SECONDS + 35))?; |
| 647 | let query = Query { |
| 648 | client: &client, |
| 649 | repo: &repo, |
| 650 | types: args.types.iter().map(|t| t.trim()).filter(|t| !t.is_empty()).collect::<Vec<_>>().join(","), |
| 651 | numbers: args.issues.iter().map(u64::to_string).collect::<Vec<_>>().join(","), |
| 652 | since: args.since.as_deref().map(|t| since_time(t, now())), |
| 653 | }; |
| 654 | let out = &ctx.out; |
| 655 | let print = |event: &Value| -> Result<()> { |
| 656 | if out.structured() { |
| 657 | out.emit_line(event) |
| 658 | } else { |
| 659 | println!("{}", event_line(out, event)); |
| 660 | Ok(()) |
| 661 | } |
| 662 | }; |
| 663 | let mut seen = Seen::default(); |
| 664 | // Where following starts, read before the past events so that nothing |
| 665 | // written between the two is missed: the newest event now, or for a |
| 666 | // repository with none yet, the start of the log. |
| 667 | let mut cursor = match &args.after { |
| 668 | Some(after) => Some(after.clone()), |
| 669 | None if args.follow => Some( |
| 670 | query |
| 671 | .fetch(None, None, 1, false)? |
| 672 | .first() |
| 673 | .map(|event| s(event, "id").to_owned()) |
| 674 | .unwrap_or_else(|| "evt_0".to_owned()), |
| 675 | ), |
| 676 | None => None, |
| 677 | }; |
| 678 | let past = match &args.after { |
| 679 | // Resuming: what came after it, oldest first. |
| 680 | Some(after) => query.fetch(Some(after), None, args.limit, true)?, |
| 681 | None => { |
| 682 | let mut past = query.fetch(None, None, args.limit, true)?; |
| 683 | past.truncate(args.limit); |
| 684 | past.reverse(); |
| 685 | past |
| 686 | } |
| 687 | }; |
| 688 | for event in &past { |
| 689 | seen.insert(s(event, "id")); |
| 690 | print(event)?; |
| 691 | } |
| 692 | if !args.follow { |
| 693 | if past.is_empty() && !out.structured() { |
| 694 | eprintln!("No events in {repo} that match."); |
| 695 | } |
| 696 | return Ok(()); |
| 697 | } |
| 698 | if args.after.is_some() |
| 699 | && let Some(last) = past.last() |
| 700 | { |
| 701 | cursor = Some(s(last, "id").to_owned()); |
| 702 | } |
| 703 | let mut failures = 0u32; |
| 704 | loop { |
| 705 | let asked = Instant::now(); |
| 706 | match query.fetch(cursor.as_deref(), Some(WAIT_SECONDS), 200, true) { |
| 707 | Ok(batch) => { |
| 708 | if failures > 0 && !out.structured() { |
| 709 | eprintln!("g1t: following {repo} again."); |
| 710 | } |
| 711 | failures = 0; |
| 712 | // A server that ignores `after` answers with what it has |
| 713 | // already sent, over and over. |
| 714 | if !batch.is_empty() && batch.iter().all(|event| seen.contains(s(event, "id"))) { |
| 715 | return Err(Error::failed(format!( |
| 716 | "{} cannot follow events: the g1t server is older than this CLI.", |
| 717 | client.host.api |
| 718 | ))); |
| 719 | } |
| 720 | for event in &batch { |
| 721 | if seen.insert(s(event, "id")) { |
| 722 | print(event)?; |
| 723 | } |
| 724 | } |
| 725 | if let Some(last) = batch.last() { |
| 726 | cursor = Some(s(last, "id").to_owned()); |
| 727 | } else if asked.elapsed() < Duration::from_secs(1) { |
| 728 | // Answered at once with nothing: not held, so not |
| 729 | // asked again at once. |
| 730 | std::thread::sleep(Duration::from_secs(2)); |
| 731 | } |
| 732 | } |
| 733 | Err(error) if matches!(error.kind, Kind::Auth | Kind::Usage) => return Err(error), |
| 734 | Err(error) => { |
| 735 | failures += 1; |
| 736 | let wait = backoff(failures); |
| 737 | eprintln!("g1t: {} Trying again in {}s.", error.message.trim_end(), wait.as_secs()); |
| 738 | std::thread::sleep(wait); |
| 739 | } |
| 740 | } |
| 741 | } |
| 742 | } |
| 743 | |
| 744 | #[cfg(test)] |
| 745 | mod tests { |
| 746 | use super::*; |
| 747 | |
| 748 | #[test] |
| 749 | fn blocking_adds_and_unblocking_removes() { |
| 750 | assert_eq!(blocked_set(12, &[9], &[10, 9], true).unwrap(), vec![9, 10]); |
| 751 | assert_eq!(blocked_set(12, &[9, 10], &[9], false).unwrap(), vec![10]); |
| 752 | assert_eq!(blocked_set(12, &[], &[3], false).unwrap(), Vec::<u64>::new()); |
| 753 | assert!(blocked_set(12, &[], &[12], true).is_err()); |
| 754 | } |
| 755 | |
| 756 | #[test] |
| 757 | fn events_read_as_one_line_each() { |
| 758 | let out = Output::default(); |
| 759 | let failed = json!({ |
| 760 | "id": "evt_2", "type": "integration.failed", "time": "2026-10-12T09:01:02.345Z", |
| 761 | "data": { "number": 14, "issue": 12, "check": { "name": "test", "conclusion": "failure" }, "combination": [14, 15] }, |
| 762 | }); |
| 763 | assert_eq!( |
| 764 | event_line(&out, &failed), |
| 765 | "2026-10-12 09:01:02\tintegration.failed\t#14\ttest failure with #14, #15\tevt_2" |
| 766 | ); |
| 767 | let claimed = json!({ |
| 768 | "id": "evt_3", "type": "work.claimed", "time": "2026-10-12T09:02:00Z", |
| 769 | "data": { "number": 12, "claim": { "issue": 12, "note": "taking the parser" } }, |
| 770 | }); |
| 771 | assert_eq!(event_cells(&claimed)[3], "taking the parser"); |
| 772 | let overlap = json!({ "type": "overlap.detected", "data": { "pull": 14, "other_pull": 15, "files": ["src/a.rs"] } }); |
| 773 | assert_eq!(event_cells(&overlap)[2], "#14"); |
| 774 | assert_eq!(event_cells(&overlap)[3], "overlaps #15: src/a.rs"); |
| 775 | let dependency = json!({ "type": "dependency.changed", "data": { "number": 12, "blocked_by": [9, 10] } }); |
| 776 | assert_eq!(event_cells(&dependency)[3], "waits on #9, #10"); |
| 777 | } |
| 778 | |
| 779 | #[test] |
| 780 | fn json_events_are_one_object_a_line() { |
| 781 | let out = Output { json: Some(vec!["id".into(), "type".into()]), ..Output::default() }; |
| 782 | let line = crate::output::json_line(&out, &json!({ "type": "push", "id": "evt_1", "data": {} })).unwrap(); |
| 783 | assert_eq!(line, r#"{"id":"evt_1","type":"push"}"#); |
| 784 | let all = Output { json: Some(vec![]), tty: true, ..Output::default() }; |
| 785 | let line = crate::output::json_line(&all, &json!({ "b": { "c": 1 }, "a": 2 })).unwrap(); |
| 786 | assert_eq!(line, r#"{"a":2,"b":{"c":1}}"#, "compact on a terminal too"); |
| 787 | } |
| 788 | |
| 789 | #[test] |
| 790 | fn a_follow_backs_off_to_thirty_seconds() { |
| 791 | let waits: Vec<u64> = (1..=8).map(|n| backoff(n).as_secs()).collect(); |
| 792 | assert_eq!(waits, vec![1, 2, 4, 8, 16, 30, 30, 30]); |
| 793 | } |
| 794 | |
| 795 | #[test] |
| 796 | fn a_follow_remembers_what_it_printed_but_not_forever() { |
| 797 | let mut seen = Seen::default(); |
| 798 | assert!(seen.insert("evt_1")); |
| 799 | assert!(!seen.insert("evt_1")); |
| 800 | for n in 0..REMEMBERED { |
| 801 | seen.insert(&format!("evt_x{n}")); |
| 802 | } |
| 803 | assert!(!seen.contains("evt_1"), "the oldest is forgotten"); |
| 804 | assert_eq!(seen.ids.len(), REMEMBERED); |
| 805 | } |
| 806 | |
| 807 | #[test] |
| 808 | fn since_takes_spans_and_times() { |
| 809 | // 2026-10-12T10:00:00Z |
| 810 | let now = 1_791_799_200; |
| 811 | assert_eq!(since_time("2h", now), "2026-10-12T08:00:00Z"); |
| 812 | assert_eq!(since_time("15m", now), "2026-10-12T09:45:00Z"); |
| 813 | assert_eq!(since_time("2026-10-01T00:00:00Z", now), "2026-10-01T00:00:00Z"); |
| 814 | } |
| 815 | |
| 816 | #[test] |
| 817 | fn claims_say_who_until_when_and_why() { |
| 818 | let out = Output::default(); |
| 819 | let line = claim_line(&out, &json!({ "actor": "ana", "harness": "claude_code", "issue": 12, "state": "active", "note": "parser" })); |
| 820 | assert_eq!(line, "ana (claude_code) on #12: parser"); |
| 821 | assert_eq!(until("2000-01-01T00:00:00Z"), "expired"); |
| 822 | } |
| 823 | } |