use std::io::Write;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use super::command_record::{CommandRecord, CommandTree};
use super::effects::OnMachine;
use super::reading::{self, Snapshot};
use crate::config::UserConfig;
#[derive(Clone, Copy)]
pub enum Ending {
Triage,
Requeue,
}
#[derive(Deserialize, Serialize)]
pub struct Request {
pub who: String,
pub requeue: bool,
#[serde(default)]
pub from_console: bool,
}
impl Request {
pub(super) fn source(&self) -> &'static str {
if self.from_console { "Console" } else { "CLI" }
}
pub fn ending(&self) -> Ending {
if self.requeue {
Ending::Requeue
} else {
Ending::Triage
}
}
}
fn request_path(record: &Path) -> PathBuf {
record.with_extension("stop.json")
}
pub(super) fn requested(record: &Path) -> Option<Request> {
serde_json::from_slice(&std::fs::read(request_path(record)).ok()?).ok()
}
fn read(record: &Path) -> Result<CommandTree> {
serde_json::from_slice(&std::fs::read(record)?)
.with_context(|| format!("could not read Command record {}", record.display()))
}
fn require_live(record: &CommandRecord) -> Result<()> {
if let Some(outcome) = &record.outcome {
bail!("cannot stop {}: Command ended: {outcome}", record.name);
}
if !record.live(&mut OnMachine) {
bail!(
"cannot stop {}: Command died (its process is gone or its identity changed)",
record.name
);
}
Ok(())
}
pub fn stop(input: &str, requeue: bool, json: bool) -> Result<()> {
let user = std::process::Command::new("id")
.arg("-un")
.output()
.context("could not identify the Unix user asking for Stop")?;
if !user.status.success() {
bail!("could not identify the Unix user asking for Stop");
}
let request = Request {
who: String::from_utf8(user.stdout)?.trim().to_owned(),
requeue,
from_console: false,
};
let path = request_stop(input, request)?;
wait(
&path,
&mut WallClock(Instant::now()),
&mut std::io::stdout(),
json,
)
}
pub(crate) fn request_stop(input: &str, request: Request) -> Result<PathBuf> {
let config = UserConfig::load()?;
let snapshot = Snapshot::load(&config.logs_dir, chrono::Local::now())?;
let path = snapshot.resolve(input)?.stop_target(input)?;
let tree = read(&path)?;
require_live(&tree.command)?;
let mut file =
tempfile::NamedTempFile::new_in(path.parent().context("Command record has no directory")?)?;
file.write_all(&serde_json::to_vec(&request)?)?;
file.persist_noclobber(request_path(&path))
.context("could not write Stop request (a Stop may already have been requested)")?;
let signal = (|| {
require_live(&read(&path)?.command)?;
super::process_identity::terminate(tree.command.process_id, &tree.command.process_start)
})();
if let Err(error) = signal {
let _ = std::fs::remove_file(request_path(&path));
return Err(error);
}
Ok(path)
}
pub(crate) struct Inspection {
pub details: serde_json::Value,
pub can_stop: bool,
pub stopping: bool,
pub branches: Vec<String>,
pub pull_requests: Vec<String>,
}
pub(crate) fn inspect(config: &UserConfig, input: &str) -> Result<Inspection> {
let snapshot = Snapshot::load_with_silence(
&config.logs_dir,
chrono::Local::now(),
config.silent_interval,
)?;
let selected = snapshot.resolve(input)?;
let pending = selected
.owner
.is_some_and(|owner| requested(&owner.record_path).is_some());
let details = serde_json::to_value(selected.observation.details()?)?;
let stopping = pending && !selected.process_exited() && details["outcome"].is_null();
let work = if selected.process_exited() || !details["outcome"].is_null() {
selected
.owner
.map(|owner| reading::retained_work(&owner.record_path, chrono::Local::now()))
.transpose()?
.unwrap_or_default()
} else {
reading::RetainedWork::default()
};
Ok(Inspection {
details,
can_stop: selected.stop_target(input).is_ok() && !pending,
stopping,
branches: work.branches.into_iter().collect(),
pull_requests: work.pull_requests.into_iter().collect(),
})
}
trait Clock {
fn elapsed(&self) -> Duration;
fn sleep(&mut self);
}
struct WallClock(Instant);
impl Clock for WallClock {
fn elapsed(&self) -> Duration {
self.0.elapsed()
}
fn sleep(&mut self) {
std::thread::sleep(Duration::from_millis(100));
}
}
fn wait(path: &Path, clock: &mut impl Clock, output: &mut impl Write, json: bool) -> Result<()> {
let mut reminder = clock.elapsed();
loop {
let tree = read(path)?;
if tree.command.outcome.is_some() || !tree.command.live(&mut OnMachine) {
print_ending(path, &tree, output, json)?;
return Ok(());
}
if !json && clock.elapsed().saturating_sub(reminder) >= Duration::from_secs(60) {
writeln!(output, "{} still ending", tree.command.name)?;
output.flush()?;
reminder = clock.elapsed();
}
clock.sleep();
}
}
fn print_ending(
path: &Path,
tree: &CommandTree,
output: &mut impl Write,
json: bool,
) -> Result<()> {
let work = reading::retained_work(path, chrono::Local::now())?;
if json {
writeln!(
output,
"{}",
serde_json::json!({
"version": 1,
"name": tree.command.name,
"outcome": tree.command.outcome.as_deref().unwrap_or("died before recording an outcome"),
"issue_branches": work.branches,
"pull_requests": work.pull_requests,
})
)?;
return Ok(());
}
writeln!(
output,
"{}: {}",
tree.command.name,
tree.command
.outcome
.as_deref()
.unwrap_or("died before recording an outcome")
)?;
for branch in work.branches {
writeln!(output, "Issue branch left: {branch}")?;
}
for pr in work.pull_requests {
writeln!(output, "Pull request left: {pr}")?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::harness::{Choice, ChosenBy, Harness};
use crate::issue::IssueUrl;
use crate::logs::Work;
#[test]
fn waiting_reports_still_ending_each_minute_then_the_outcome_and_resources() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("example.command.json");
let issue = IssueUrl::parse("https://github.com/acme/widgets/issues/7").unwrap();
let mut command = CommandRecord::new(
Work::Run(&issue),
"stamp",
&Choice {
harness: Harness::Claude,
model: None,
effort: None,
chosen_by: ChosenBy::Default,
},
chrono::Local::now().fixed_offset(),
);
command.process_start = super::super::process_identity::start(command.process_id).unwrap();
let tree = CommandTree {
command,
dispatched: Vec::new(),
};
std::fs::write(&path, serde_json::to_vec(&tree).unwrap()).unwrap();
struct Minutes {
elapsed: u64,
path: PathBuf,
}
impl Clock for Minutes {
fn elapsed(&self) -> Duration {
Duration::from_secs(self.elapsed)
}
fn sleep(&mut self) {
self.elapsed += 1;
if self.elapsed == 125 {
let mut tree = read(&self.path).unwrap();
tree.command.outcome = Some("interrupted".into());
tree.command.issue_branch = Some("issue-7".into());
tree.command.pull_request =
Some("https://github.com/acme/widgets/pull/1".into());
std::fs::write(&self.path, serde_json::to_vec(&tree).unwrap()).unwrap();
}
}
}
let mut output = Vec::new();
wait(
&path,
&mut Minutes {
elapsed: 0,
path: path.clone(),
},
&mut output,
false,
)
.unwrap();
assert_eq!(
String::from_utf8(output).unwrap(),
"acme/widgets#7 still ending\nacme/widgets#7 still ending\nacme/widgets#7: interrupted\nIssue branch left: issue-7\nPull request left: https://github.com/acme/widgets/pull/1\n"
);
}
}