From 1ac81a9ac263f905f4ef92f42b420bd895531933 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Mon, 19 Oct 2026 06:27:20 -0400 Subject: [PATCH] Deepen Command reading around selection and ownership (#793) --- docs/code-navigation.md | 6 +- src/logs.rs | 1 + src/logs/reading.rs | 367 +++++++++++++++++++ src/logs/{status => reading}/history.rs | 30 +- src/logs/reading/selection.rs | 425 ++++++++++++++++++++++ src/logs/reading/tests.rs | 292 +++++++++++++++ src/logs/status.rs | 425 ++++------------------ src/logs/status/inspect.rs | 456 ++---------------------- src/logs/stop.rs | 14 +- tests/show_logs.rs | 58 +++ 10 files changed, 1255 insertions(+), 819 deletions(-) create mode 100644 src/logs/reading.rs rename src/logs/{status => reading}/history.rs (93%) create mode 100644 src/logs/reading/selection.rs create mode 100644 src/logs/reading/tests.rs diff --git a/docs/code-navigation.md b/docs/code-navigation.md index 6d67d35d8..dc0a89727 100644 --- a/docs/code-navigation.md +++ b/docs/code-navigation.md @@ -19,9 +19,9 @@ Start with the issue's terms in `CONTEXT.md`, then use this index to find their | Session log, Resume | `src/session.rs` and `src/session/` | `src/session/execution_tests.rs`, `tests/background_work.rs` | | Command log, Activity log | `src/logs.rs`; `src/logs/recording.rs` — `Record` | `src/logs/recording/tests.rs`, `tests/command_logs.rs`, `tests/activity_log.rs` | | Command-log allocation, Session naming, child inheritance and filename decoding | `src/logs/owner.rs` — `allocate`, `Owner`, `decode`, `session_matches`; `src/session/recording.rs` — captured Model recording | `src/logs/owner/tests.rs`, `src/logs/recording/tests.rs`, `src/session/recording/tests.rs`, `tests/command_logs.rs`, `tests/status.rs`, `tests/show_logs.rs` | -| Command record, status | `src/logs/recording.rs`; `src/logs/command_record.rs`; `src/logs/status.rs` and `src/logs/status/history.rs`; `src/logs/process_identity.rs` | `tests/status.rs`, `src/logs/recording/tests.rs` | -| Command naming, show, logs | `src/logs/status/inspect.rs` — Command resolution, detailed state, Session selection and log following | `tests/show_logs.rs` | -| Stop | `src/logs/stop.rs` — request, signaling and waiting; `src/claim.rs` — Stop endings | `tests/stop.rs`; inline Claim and Stop tests | +| Command record, status | `src/logs/recording.rs`; `src/logs/command_record.rs`; `src/logs/reading.rs` — `Snapshot`, exact-parent trees and observations; `src/logs/reading/history.rs` — historical recovery; `src/logs/status.rs` — formatting; `src/logs/process_identity.rs` | `src/logs/reading/tests.rs`, `tests/status.rs`, `src/logs/recording/tests.rs` | +| Command naming, Session selection, show, logs | `src/logs/reading/selection.rs` — resolution, captured outer owner and child-refusal target, Session inventory/selection and bounded latest events; `src/logs/status/inspect.rs` — formatting and log streaming | `src/logs/reading/tests.rs`, `tests/show_logs.rs` | +| Stop | `src/logs/reading.rs` — `retained_work` reload by captured record; `src/logs/reading/selection.rs` — `Selection::stop_target`; `src/logs/stop.rs` — request, signaling and waiting; `src/claim.rs` — Stop endings | `src/logs/reading/tests.rs`, `tests/stop.rs`; inline Claim and Stop tests | | User config, Setup | `src/config.rs`; `src/setup.rs` | `tests/user_config.rs`, `tests/setup.rs` | | Architecture review, Architect run | `src/architect.rs` | `tests/architect_run.rs` | | Security audit, Security review, Security run | `src/security.rs` and `src/security/` | `tests/security_review.rs`, `tests/security_run.rs` | diff --git a/src/logs.rs b/src/logs.rs index 11d316821..6d88ab1ab 100644 --- a/src/logs.rs +++ b/src/logs.rs @@ -16,6 +16,7 @@ mod command_record; mod effects; pub(crate) mod owner; mod process_identity; +mod reading; mod recording; pub mod status; pub mod stop; diff --git a/src/logs/reading.rs b/src/logs/reading.rs new file mode 100644 index 000000000..ecfc7ae13 --- /dev/null +++ b/src/logs/reading.rs @@ -0,0 +1,367 @@ +//! Filesystem observations of Commands, their Runs and exact ownership. +//! Loading uses one sampled time; reading never prints or performs Stop effects. + +mod history; +mod selection; +#[cfg(test)] +mod tests; + +use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::fs; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Local}; +use serde::Serialize; + +use super::command_record::{CommandRecord, CommandTree}; +use super::effects::OnMachine; +use super::{Pass, activity}; +pub(super) use history::Command as HistoricalCommand; +pub(super) use selection::{Detailed, Observation}; + +pub(super) struct Snapshot { + commands: Vec, + pub(super) idle_passes: Vec, + now: DateTime, +} + +impl Snapshot { + pub(super) fn load(root: &Path, now: DateTime) -> Result { + let mut records = BTreeMap::new(); + let mut commands = Vec::new(); + let mut idle_passes = Vec::new(); + for owner in directories(root)? { + for repo in directories(&owner)? { + let mut covered = HashSet::new(); + idle_passes.extend(read_idle_passes(&repo)?); + for folder in directories(&repo.join("commands"))? { + for (path, tree) in read_records(&folder)? { + let record = &tree.command; + if record.parent_record.is_none() { + covered.insert(command_log(&path)); + } + for session in std::iter::once(record) + .chain(&tree.dispatched) + .flat_map(CommandRecord::session_references) + { + covered.insert(session.clone()); + } + records.insert(path, tree); + } + } + commands.extend( + history::read(&repo, &covered)? + .into_iter() + .map(|command| Entry::History(Box::new(command))), + ); + } + } + let roots: Vec<_> = records + .iter() + .filter(|(_, tree)| { + let record = &tree.command; + record.parent_record.is_none() + }) + .map(|(path, _)| path.clone()) + .collect(); + commands.extend( + roots + .into_iter() + .map(|path| Entry::Record(Box::new(tree(&path, &mut records, &now)))), + ); + idle_passes.sort_by(|a, b| (&a.repository, &a.kind).cmp(&(&b.repository, &b.kind))); + Ok(Self { + commands, + idle_passes, + now, + }) + } + + pub(super) fn list(&self, all: bool) -> Vec<&Entry> { + let mut commands: Vec<_> = self + .commands + .iter() + .filter(|entry| match entry { + Entry::Record(command) => { + all || command.record.outcome.is_none() + || recent(command.record.ended_at.as_deref(), self.now) + } + Entry::History(command) => all || recent(command.ended_at.as_deref(), self.now), + }) + .collect(); + commands.sort_by(|a, b| a.ordering(all).cmp(&b.ordering(all))); + commands + } + + pub(super) fn resolve(&self, input: &str) -> Result> { + selection::resolve(&self.commands, input) + } +} + +/// Reload only the captured Command's folder, after Stop cleanup has finished. +pub(super) fn retained_work(path: &Path, now: DateTime) -> Result { + let mut records = read_records(path.parent().context("Command record has no directory")?)?; + anyhow::ensure!(records.contains_key(path), "Command record is unavailable"); + let command = tree(path, &mut records, &now); + let mut work = RetainedWork::default(); + fn collect(command: &Command, work: &mut RetainedWork) { + work.branches + .extend(command.record.issue_branch.iter().cloned()); + work.pull_requests + .extend(command.record.pull_request.iter().cloned()); + for child in command.dispatched.iter().chain(&command.children) { + collect(child, work); + } + } + collect(&command, &mut work); + Ok(work) +} + +#[derive(Default)] +pub(super) struct RetainedWork { + pub(super) branches: BTreeSet, + pub(super) pull_requests: BTreeSet, +} + +#[derive(Serialize)] +pub(super) struct IdlePass { + pub(super) repository: String, + pub(super) kind: String, + pub(super) reason: String, +} + +#[derive(Serialize)] +#[serde(untagged)] +pub(super) enum Entry { + Record(Box), + History(Box), +} + +impl Entry { + fn ordering( + &self, + all: bool, + ) -> ( + bool, + std::cmp::Reverse>>, + &str, + ) { + let (state, started, repo) = match self { + Self::Record(command) => ( + command.state, + command.record.started_at.as_str(), + command.record.repository.as_str(), + ), + Self::History(command) => ( + command.state, + command.started_at.as_str(), + command.repository.as_str(), + ), + }; + ( + !all && state != "live", + std::cmp::Reverse(DateTime::parse_from_rfc3339(started).ok()), + repo, + ) + } +} + +#[derive(Serialize)] +pub(super) struct Command { + #[serde(skip)] + pub(super) record_path: PathBuf, + #[serde(flatten)] + pub(super) record: CommandRecord, + pub(super) state: &'static str, + pub(super) stage: Option, + pub(super) children: Vec, + pub(super) source: &'static str, + pub(super) command_log: PathBuf, + pub(super) dispatched: Vec, +} + +#[derive(Serialize)] +pub(super) struct CurrentStage { + pub(super) name: String, + pub(super) began_at: String, + pub(super) age_seconds: i64, +} + +impl Command { + fn new(record: CommandRecord, record_path: PathBuf, now: DateTime) -> Self { + let state = if record.outcome.is_some() { + "ended" + } else if record.live(&mut OnMachine) { + "live" + } else { + "died" + }; + let stage = record.current_stage().map(|stage| CurrentStage { + name: stage.name.clone(), + began_at: stage.began_at.clone(), + age_seconds: DateTime::parse_from_rfc3339(&stage.began_at) + .map(|began| now.signed_duration_since(began).num_seconds().max(0)) + .unwrap_or(0), + }); + Self { + command_log: command_log(&record_path), + record_path, + record, + state, + stage, + children: Vec::new(), + source: "command_record", + dispatched: Vec::new(), + } + } +} + +fn read_records(folder: &Path) -> Result> { + let mut records = BTreeMap::new(); + for entry in entries(folder)? { + let name = entry.file_name(); + let name = name.to_string_lossy(); + if !entry.file_type()?.is_file() + || (!name.ends_with(".command.json") && !name.ends_with(".child.json")) + { + continue; + } + let path = entry.path(); + let bytes = + fs::read(&path).with_context(|| format!("could not read {}", path.display()))?; + let Ok(mut tree) = serde_json::from_slice::(&bytes) else { + continue; + }; + let record = &mut tree.command; + if record.version != 1 { + continue; + } + // Records written before completion times existed were last + // replaced when their outcome was recorded. + if record.outcome.is_some() && record.ended_at.is_none() { + record.ended_at = entry + .metadata()? + .modified() + .ok() + .map(|time| DateTime::::from(time).to_rfc3339()); + } + records.insert(path, tree); + } + Ok(records) +} + +fn command_log(path: &Path) -> PathBuf { + super::owner::record_log(path) +} + +/// Keep the filename spelling alongside its parsed time for naming and history. +fn command_stamp(path: &Path) -> Option<(&str, DateTime)> { + let decoded = super::owner::decode(path)?; + Some((decoded.identity.stamp, decoded.time())) +} + +fn tree( + path: &Path, + records: &mut BTreeMap, + now: &DateTime, +) -> Command { + let tree = records.remove(path).expect("a discovered record exists"); + let mut command = Command::new(tree.command, path.to_owned(), *now); + command.children = children(path, None, records, now); + command.dispatched = tree + .dispatched + .into_iter() + .map(|record| { + let mut dispatched = Command::new(record, path.to_owned(), *now); + let issue = dispatched + .record + .issue + .as_ref() + .map(|issue| issue.url.as_str()); + dispatched.children = children(path, issue, records, now); + dispatched + }) + .collect(); + command +} + +fn children( + path: &Path, + dispatch: Option<&str>, + records: &mut BTreeMap, + now: &DateTime, +) -> Vec { + let paths: Vec<_> = records + .iter() + .filter(|(_, child)| { + child.command.parent_record.as_deref() == Some(path) + && child.command.parent_dispatch.as_deref() == dispatch + }) + .map(|(path, _)| path.clone()) + .collect(); + let mut children: Vec<_> = paths + .into_iter() + .map(|path| tree(&path, records, now)) + .collect(); + children.sort_by_key(|child| child.record.issue.as_ref().map(|issue| issue.number)); + children +} + +fn recent(ended: Option<&str>, now: DateTime) -> bool { + ended + .and_then(|time| DateTime::parse_from_rfc3339(time).ok()) + .is_some_and(|ended| ended >= now - chrono::Duration::hours(24)) +} + +fn read_idle_passes(repo: &Path) -> Result> { + let path = repo.join(activity::FILE); + let activity = match fs::read_to_string(&path) { + Ok(activity) => activity, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => { + return Err(error).with_context(|| format!("could not read {}", path.display())); + } + }; + let repository = format!( + "{}/{}", + repo.parent() + .unwrap() + .file_name() + .unwrap() + .to_string_lossy(), + repo.file_name().unwrap().to_string_lossy() + ); + let mut idle = Vec::new(); + for pass in [Pass::Architect, Pass::Pickup, Pass::Security] { + let kind = activity::Kind::Pass(pass); + if let Some(line) = activity::last_of(&activity, kind) + && let Some(reason) = line.strip_prefix(&format!("{kind} skipped: ")) + { + idle.push(IdlePass { + repository: repository.clone(), + kind: kind.to_string(), + reason: reason.into(), + }); + } + } + Ok(idle) +} + +fn entries(path: &Path) -> Result> { + match fs::read_dir(path) { + Ok(entries) => entries.collect::>().map_err(Into::into), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()), + Err(error) => Err(error).with_context(|| format!("could not list {}", path.display())), + } +} + +fn directories(path: &Path) -> Result> { + let mut paths = Vec::new(); + for entry in entries(path)? { + if entry.file_type()?.is_dir() { + paths.push(entry.path()); + } + } + Ok(paths) +} diff --git a/src/logs/status/history.rs b/src/logs/reading/history.rs similarity index 93% rename from src/logs/status/history.rs rename to src/logs/reading/history.rs index 819d9bbc1..b542ec7d6 100644 --- a/src/logs/status/history.rs +++ b/src/logs/reading/history.rs @@ -14,18 +14,18 @@ use crate::logs::owner; use crate::progress::ChildLine; #[derive(Clone, Serialize)] -pub(super) struct Command { +pub(in crate::logs) struct Command { pub repository: String, - pub(super) kind: String, - pub(super) issue: Option, + pub(in crate::logs) kind: String, + pub(in crate::logs) issue: Option, harness: Option, model: Option, effort: Option, pub started_at: String, pub ended_at: Option, - outcome: Option, - pub(super) command_log: Option, - pub(super) session_logs: Vec, + pub(in crate::logs) outcome: Option, + pub(in crate::logs) command_log: Option, + pub(in crate::logs) session_logs: Vec, stages: [(); 0], stage: Option<()>, source: &'static str, @@ -62,22 +62,6 @@ impl Command { state: "unknown", } } - - pub fn print(&self) { - let issue = self - .issue - .as_ref() - .map(|issue| format!("#{}", issue.number)) - .unwrap_or_default(); - println!( - "{}{issue} {} · {} → {} · {}", - self.repository, - self.kind, - self.started_at, - self.ended_at.as_deref().unwrap_or("unknown end"), - self.outcome.as_deref().unwrap_or("unknown outcome") - ); - } } struct Started { @@ -131,7 +115,7 @@ impl Started { } } -pub(super) fn read(repo: &Path, covered: &HashSet) -> Result> { +pub(in crate::logs) fn read(repo: &Path, covered: &HashSet) -> Result> { let repository = format!( "{}/{}", repo.parent() diff --git a/src/logs/reading/selection.rs b/src/logs/reading/selection.rs new file mode 100644 index 000000000..f43f242d3 --- /dev/null +++ b/src/logs/reading/selection.rs @@ -0,0 +1,425 @@ +//! Shared naming, Session inventory and captured selection ownership. + +use std::collections::BTreeSet; +use std::io::{Read, Seek, SeekFrom}; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, bail}; +use serde::Serialize; + +use super::{Command, Entry}; +use crate::issue::IssueUrl; +use crate::logs::effects::Outside; +use crate::logs::owner; + +enum Name { + Issue { repository: String, number: u64 }, + Pass(String), +} + +impl Name { + fn parse(name: &str) -> Result { + if name.starts_with("https://") { + let issue = IssueUrl::parse(name)?; + return Ok(Self::Issue { + repository: issue.repo_slug(), + number: issue.number, + }); + } + if let Some((repository, number)) = name.split_once('#') { + let number = number + .parse::() + .with_context(|| format!("invalid Command name: {name}"))? + .get(); + if !repository.is_empty() { + return Ok(Self::Issue { + repository: repository.into(), + number, + }); + } + } + if ["architect-", "pickup-", "secure-"] + .iter() + .any(|prefix| name.starts_with(prefix)) + { + return Ok(Self::Pass(name.into())); + } + bail!("invalid Command name: {name}; use repo#number, an Issue URL or kind-stamp") + } + + fn matches_repo(&self, repo: &str) -> bool { + match self { + Self::Issue { repository, .. } => { + repo.eq_ignore_ascii_case(repository) + || (!repository.contains('/') + && repo + .rsplit('/') + .next() + .is_some_and(|name| name.eq_ignore_ascii_case(repository))) + } + Self::Pass(_) => true, + } + } +} + +#[derive(Clone, Copy, Serialize)] +#[serde(untagged)] +pub(in crate::logs) enum Observation<'a> { + Record(&'a Command), + History(&'a super::history::Command), +} + +impl<'a> Observation<'a> { + fn repository(self) -> &'a str { + match self { + Self::Record(command) => &command.record.repository, + Self::History(command) => &command.repository, + } + } + + fn matches(self, name: &Name) -> bool { + if !name.matches_repo(self.repository()) { + return false; + } + match name { + Name::Issue { number, .. } => { + let issue = match self { + Self::Record(command) => command.record.issue.as_ref(), + Self::History(command) => command.issue.as_ref(), + }; + issue.is_some_and(|issue| issue.number == *number) + } + Name::Pass(name) => match self { + Self::Record(command) => command.record.name == *name, + Self::History(command) => command.command_log.as_ref().is_some_and(|path| { + let Some(kind) = pass_prefix(&command.kind) else { + return false; + }; + super::command_stamp(path) + .is_some_and(|(stamp, _)| *name == format!("{kind}-{stamp}")) + }), + }, + } + } + + fn ordering( + self, + ) -> ( + bool, + std::cmp::Reverse>>, + ) { + let (state, started) = match self { + Self::Record(command) => (command.state, command.record.started_at.as_str()), + Self::History(command) => (command.state, command.started_at.as_str()), + }; + ( + state != "live", + std::cmp::Reverse(chrono::DateTime::parse_from_rfc3339(started).ok()), + ) + } + + pub(in crate::logs) fn command_log(self) -> Option<&'a Path> { + match self { + Self::Record(command) => Some(&command.command_log), + Self::History(command) => command.command_log.as_deref(), + } + } + + pub(in crate::logs) fn sessions(self) -> Result> { + let (mut paths, issue, kind) = match self { + Self::Record(command) => { + let mut paths = Vec::new(); + for path in command.record.session_references() { + if !paths.contains(path) { + paths.push(path.clone()); + } + } + ( + paths, + command.record.issue.as_ref(), + command.record.kind.as_str(), + ) + } + Self::History(command) => ( + command.session_logs.clone(), + command.issue.as_ref(), + command.kind.as_str(), + ), + }; + // A stage can cover several sessions (for example Architecture review + // and its Resume), so stage references alone are not a log inventory. + if let Some(log) = self.command_log() + && owner::decode(log).is_some() + { + let prefix = issue + .map(|issue| issue.number.to_string()) + .or_else(|| pass_prefix(kind).map(String::from)) + .context("cannot name Session logs for this Command kind")?; + for entry in super::entries(log.parent().context("Command log has no directory")?)? { + let path = entry.path(); + if entry.file_type()?.is_file() + && owner::decode(&path).is_some_and(|decoded| { + decoded.session_kind.is_some() && decoded.subject == prefix + }) + && owner::session_matches(&path, log) + && !paths.contains(&path) + { + paths.push(path); + } + } + } + let mut dated = paths + .into_iter() + .map(|path| { + let modified = std::fs::metadata(&path) + .and_then(|metadata| metadata.modified()) + .unwrap_or(std::time::SystemTime::UNIX_EPOCH); + (modified, path) + }) + .collect::>(); + dated.sort(); + Ok(dated.into_iter().map(|(_, path)| path).collect()) + } + + pub(in crate::logs) fn session(self, kind: &str) -> Result { + self.sessions()? + .into_iter() + .rev() + .find(|path| { + let Some(session) = owner::decode(path).and_then(|decoded| decoded.session_kind) + else { + return false; + }; + let original = session.strip_suffix("-resume").unwrap_or(session); + session == kind + || original == kind + || (kind == "repair" + && original + .strip_prefix("repair-") + .is_some_and(|number| number.parse::().is_ok())) + }) + .with_context(|| format!("no {kind} Session log for this Command")) + } +} + +fn pass_prefix(kind: &str) -> Option<&'static str> { + match kind { + "Architect run" => Some("architect"), + "Pickup run" => Some("pickup"), + "Security run" => Some("secure"), + _ => None, + } +} + +/// A selected Run and its captured outer process/record. Child refusal is an +/// observation of the enclosing Run/Spec, independent of signal authority. +#[derive(Clone, Copy)] +pub(in crate::logs) struct Selection<'a> { + pub(in crate::logs) observation: Observation<'a>, + pub(in crate::logs) owner: Option<&'a Command>, + pub(in crate::logs) child_owner: Option<&'a Command>, +} + +impl Selection<'_> { + pub(in crate::logs) fn stop_target(self, input: &str) -> Result { + if let Some(owner) = self.child_owner { + bail!( + "cannot stop {input}: child Run; stop {} {} instead", + owner.record.kind, + owner.record.name + ); + } + match self.observation { + Observation::Record(command) if command.state == "live" => Ok(self + .owner + .context("Command has no owner")? + .record_path + .clone()), + Observation::Record(command) => bail!( + "cannot stop {input}: Command {}{}", + command.state, + command + .record + .outcome + .as_ref() + .map(|outcome| format!(": {outcome}")) + .unwrap_or_default() + ), + Observation::History(command) => { + bail!("cannot stop {input}: Command {}", command.state) + } + } + } + + /// Outcomes precede final output. Recheck the captured process, not state. + pub(in crate::logs) fn process_exited(self) -> bool { + self.owner.is_none_or(|command| { + let record = &command.record; + record.process_start.is_empty() + || crate::logs::effects::OnMachine + .process_start(record.process_id) + .as_deref() + != Some(&record.process_start) + }) + } +} + +fn records<'a>( + command: &'a Command, + outer: &'a Command, + enclosing: &'a Command, + found: &mut Vec>, +) { + found.push(Selection { + observation: Observation::Record(command), + owner: Some(outer), + child_owner: command.record.parent_record.as_ref().map(|_| enclosing), + }); + for dispatch in &command.dispatched { + records(dispatch, outer, dispatch, found); + } + for child in &command.children { + records(child, outer, enclosing, found); + } +} + +pub(super) fn resolve<'a>(entries: &'a [Entry], input: &str) -> Result> { + let name = Name::parse(input)?; + let mut candidates = Vec::new(); + for entry in entries { + match entry { + Entry::Record(command) => records(command, command, command, &mut candidates), + Entry::History(command) => candidates.push(Selection { + observation: Observation::History(command), + owner: None, + child_owner: None, + }), + } + } + let repositories: BTreeSet<_> = candidates + .iter() + .filter(|command| name.matches_repo(command.observation.repository())) + .map(|command| command.observation.repository().to_ascii_lowercase()) + .collect(); + if matches!(&name, Name::Issue { repository, .. } if !repository.contains('/')) + && repositories.len() > 1 + { + bail!( + "ambiguous Command name {input}; matching repositories: {}", + repositories.into_iter().collect::>().join(", ") + ); + } + candidates.retain(|command| command.observation.matches(&name)); + if matches!(name, Name::Pass(_)) { + let repositories: BTreeSet<_> = candidates + .iter() + .map(|command| command.observation.repository()) + .collect(); + if repositories.len() > 1 { + bail!( + "ambiguous Command name {input}; matching repositories: {}", + repositories.into_iter().collect::>().join(", ") + ); + } + } + candidates.sort_by_key(|command| command.observation.ordering()); + candidates + .into_iter() + .next() + .with_context(|| format!("no Command matches {input}")) +} + +fn latest_event(command: &Command) -> Option { + if command.state != "live" { + return None; + } + let path = command.record.session_log.as_ref()?; + let mut file = std::fs::File::open(path).ok()?; + // A show reads at most 64 KiB, independent of the transcript's size. + let start = file.metadata().ok()?.len().saturating_sub(64 * 1024); + file.seek(SeekFrom::Start(start)).ok()?; + let mut bytes = Vec::new(); + file.take(64 * 1024).read_to_end(&mut bytes).ok()?; + let tail = String::from_utf8_lossy(&bytes); + let mut lines = tail.lines(); + if start > 0 { + lines.next(); + } + let harness = crate::harness::Harness::named(&command.record.harness)?; + let mut stream = harness.adapter().interpretation(Path::new(""), ""); + let event = lines.flat_map(|line| stream.condense(line)).last()?; + let event = event + .strip_prefix("$ ") + .map_or_else(|| event.clone(), |command| format!("Bash: {command}")); + Some(event.lines().collect::>().join(" ")) +} + +#[derive(Serialize)] +#[serde(untagged)] +pub(in crate::logs) enum Detailed<'a> { + Record(Box>), + History(Box), +} + +/// The detailed view adds a complete Session inventory and a live event to +/// each Run while retaining the status reader's version 1 field names. +#[derive(Serialize)] +pub(in crate::logs) struct Details<'a> { + #[serde(flatten)] + record: &'a crate::logs::command_record::CommandRecord, + state: &'static str, + stage: &'a Option, + source: &'static str, + command_log: &'a Path, + session_logs: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + latest_event: Option, + children: Vec>, + dispatched: Vec>, +} + +impl<'a> Details<'a> { + fn new(command: &'a Command) -> Result { + Ok(Self { + record: &command.record, + state: command.state, + stage: &command.stage, + source: command.source, + command_log: &command.command_log, + session_logs: Observation::Record(command).sessions()?, + latest_event: latest_event(command), + children: command + .children + .iter() + .map(Self::new) + .collect::>()?, + dispatched: command + .dispatched + .iter() + .map(Self::new) + .collect::>()?, + }) + } +} + +impl<'a> Observation<'a> { + pub(in crate::logs) fn details(self) -> Result> { + Ok(match self { + Self::Record(command) => Detailed::Record(Box::new(Details::new(command)?)), + Self::History(command) => { + let mut details = command.clone(); + details.session_logs = self.sessions()?; + Detailed::History(Box::new(details)) + } + }) + } +} + +impl Command { + pub(in crate::logs) fn sessions(&self) -> Result> { + Observation::Record(self).sessions() + } + pub(in crate::logs) fn latest_event(&self) -> Option { + latest_event(self) + } +} diff --git a/src/logs/reading/tests.rs b/src/logs/reading/tests.rs new file mode 100644 index 000000000..4305584d8 --- /dev/null +++ b/src/logs/reading/tests.rs @@ -0,0 +1,292 @@ +use std::fs; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Local}; +use serde_json::{Value, json}; + +use super::*; + +fn now() -> DateTime { + DateTime::parse_from_rfc3339("2026-10-19T12:00:00-04:00") + .unwrap() + .with_timezone(&Local) +} + +fn record(number: u64) -> Value { + json!({ + "version": 1, "kind": "Run", "repository": "acme/widgets", + "name": format!("acme/widgets#{number}"), + "issue": {"number": number, "url": format!("https://github.com/acme/widgets/issues/{number}"), "title": "Widgets"}, + "hostname": "factory", "process_id": std::process::id(), + "process_start": crate::logs::process_identity::start(std::process::id()).unwrap(), + "harness": "claude", "model": null, "effort": null, + "started_at": "2026-10-19T11:00:00-04:00", "ended_at": null, + "stages": [], "session_log": null, "outcome": null, "cause": null, + "issue_branch": format!("issue-{number}"), "pull_request": null + }) +} + +fn write(root: &Path, name: &str, value: &Value) -> PathBuf { + let path = root.join("acme/widgets/commands/issue").join(name); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(&path, serde_json::to_vec(value).unwrap()).unwrap(); + path +} + +#[test] +fn direct_run_selection_captures_its_owner_and_reloads_retained_work() { + let dir = tempfile::tempdir().unwrap(); + let path = write( + dir.path(), + "7-20261019T110000-0400~direct.command.json", + &record(7), + ); + let snapshot = Snapshot::load(dir.path(), now()).unwrap(); + assert_eq!(snapshot.list(true).len(), 1); + let selected = snapshot.resolve("WIDGETS#7").unwrap(); + assert_eq!(selected.owner.unwrap().record_path, path); + assert!(selected.child_owner.is_none()); + assert_eq!(selected.stop_target("WIDGETS#7").unwrap(), path); + assert!(!selected.process_exited()); + let mut ended = record(7); + ended["issue_branch"] = "issue-7-preserved".into(); + ended["pull_request"] = "https://github.com/acme/widgets/pull/9".into(); + fs::write(&path, serde_json::to_vec(&ended).unwrap()).unwrap(); + let retained = retained_work(&path, now()).unwrap(); + assert_eq!(retained.branches, ["issue-7-preserved".to_string()].into()); + assert_eq!( + retained.pull_requests, + ["https://github.com/acme/widgets/pull/9".to_string()].into() + ); +} + +#[test] +fn pass_dispatches_and_equal_numbered_children_keep_exact_ownership() { + let dir = tempfile::tempdir().unwrap(); + let stem = "20261019T110000-0400~security"; + let mut pass = record(7); + pass["kind"] = "Security run".into(); + pass["name"] = "secure-20261019T110000-0400".into(); + pass["issue"] = Value::Null; + pass["issue_branch"] = "issue-20".into(); + pass["pull_request"] = "https://github.com/acme/widgets/pull/20".into(); + let path = write(dir.path(), &format!("{stem}.command.json"), &pass); + let mut dispatches = Vec::new(); + for number in [20, 30] { + let mut dispatch = record(number); + if number == 20 { + dispatch["kind"] = "Spec run".into(); + } + dispatches.push(dispatch); + let mut child = record(21); + child["parent_record"] = json!(path); + child["parent_dispatch"] = + format!("https://github.com/acme/widgets/issues/{number}").into(); + child["issue_branch"] = format!("dispatch-{number}-ticket-21").into(); + child["pull_request"] = format!("https://github.com/acme/widgets/pull/{number}").into(); + child["outcome"] = "merged".into(); + child["ended_at"] = "2026-10-19T11:30:00-04:00".into(); + if number == 30 { + child["started_at"] = "2026-10-19T11:10:00-04:00".into(); + } + write( + dir.path(), + &format!("{stem}.command.dispatch-{number}-issue-21.child.json"), + &child, + ); + } + pass["dispatched"] = dispatches.into(); + fs::write(&path, serde_json::to_vec(&pass).unwrap()).unwrap(); + // A concurrent Command with the same subject and stamp owns its own child. + let separate = write( + dir.path(), + "20261019T110000-0400~separate.command.json", + &record(20), + ); + let mut unrelated = record(99); + unrelated["parent_record"] = json!(separate); + write(dir.path(), "separate-99.child.json", &unrelated); + // A parent-dispatch link to a dispatch absent from this Command is not adopted. + unrelated["parent_record"] = json!(path); + unrelated["parent_dispatch"] = "https://github.com/acme/widgets/issues/999".into(); + write(dir.path(), "unmatched-99.child.json", &unrelated); + + let snapshot = Snapshot::load(dir.path(), now()).unwrap(); + let selected = snapshot.resolve("secure-20261019T110000-0400").unwrap(); + let outer = selected.owner.unwrap(); + assert_eq!(outer.record_path, path); + assert_eq!(outer.children.len(), 0); + assert_eq!(outer.dispatched.len(), 2); + for (index, number) in [20, 30].into_iter().enumerate() { + let dispatch = &outer.dispatched[index]; + assert_eq!(dispatch.record.issue.as_ref().unwrap().number, number); + assert_eq!(dispatch.children.len(), 1); + assert_eq!( + dispatch.children[0].record.issue_branch.as_deref(), + Some(format!("dispatch-{number}-ticket-21").as_str()) + ); + } + let dispatched = snapshot.resolve("widgets#30").unwrap(); + assert_eq!(dispatched.stop_target("widgets#30").unwrap(), path); + assert!(dispatched.child_owner.is_none()); + let child = snapshot + .resolve("https://github.com/acme/widgets/issues/21") + .unwrap(); + assert_eq!(child.owner.unwrap().record_path, path); + assert_eq!(child.child_owner.unwrap().record.name, "acme/widgets#30"); + assert_eq!( + child.stop_target("widgets#21").unwrap_err().to_string(), + "cannot stop widgets#21: child Run; stop Run acme/widgets#30 instead" + ); + // A child's durable outcome does not end following its outer process. + assert!(!child.process_exited()); + let retained = retained_work(&path, now()).unwrap(); + assert_eq!( + retained.branches.into_iter().collect::>(), + [ + "dispatch-20-ticket-21", + "dispatch-30-ticket-21", + "issue-20", + "issue-30" + ] + ); + assert_eq!( + retained.pull_requests.into_iter().collect::>(), + [ + "https://github.com/acme/widgets/pull/20", + "https://github.com/acme/widgets/pull/30" + ] + ); + assert_eq!( + retained_work(&separate, now()) + .unwrap() + .branches + .into_iter() + .collect::>(), + ["issue-20", "issue-99"] + ); + assert_eq!(fs::read(&path).unwrap(), serde_json::to_vec(&pass).unwrap()); + assert!(!path.with_extension("stop.json").exists()); +} + +#[test] +fn ticket_session_inventory_includes_stage_current_resume_and_numbered_repairs() { + let dir = tempfile::tempdir().unwrap(); + let mut spec = record(20); + spec["kind"] = "Spec run".into(); + let path = write( + dir.path(), + "20-20261019T110000-0400~spec.command.json", + &spec, + ); + let folder = path.parent().unwrap(); + let implement = folder.join("21-20261019T110000-0400~spec-implement.jsonl"); + let resume = folder.join("21-20261019T110000-0400~spec-implement-resume.jsonl"); + let repair = folder.join("21-20261019T110000-0400~spec-repair-2-resume.jsonl"); + let previous = folder.join("21-20261019T110000-0400~spec-repair-1.jsonl"); + let retained_reference = folder.join("retained-session.jsonl"); + for (index, session) in [&retained_reference, &implement, &resume, &previous, &repair] + .into_iter() + .enumerate() + { + fs::write(session, "").unwrap(); + let file = fs::File::open(session).unwrap(); + file.set_modified(std::time::UNIX_EPOCH + std::time::Duration::from_secs(index as u64)) + .unwrap(); + } + fs::write( + folder.join("21-20261019T110000-0400~other-repair-9.jsonl"), + "unrelated", + ) + .unwrap(); + let mut ticket = record(21); + ticket["parent_record"] = json!(path); + ticket["stages"] = json!([ + {"name": "implement", "began_at": "2026-10-19T11:00:00-04:00", "session_log": implement}, + {"name": "Repair", "began_at": "2026-10-19T11:30:00-04:00", "session_log": retained_reference} + ]); + ticket["session_log"] = json!(implement); + let child_path = write( + dir.path(), + "20-20261019T110000-0400~spec.command.issue-21.child.json", + &ticket, + ); + let snapshot = Snapshot::load(dir.path(), now()).unwrap(); + let selected = snapshot.resolve("ACME/WIDGETS#21").unwrap(); + assert_eq!(selected.owner.unwrap().record_path, path); + assert_eq!(selected.child_owner.unwrap().record.name, "acme/widgets#20"); + let Observation::Record(command) = selected.observation else { + panic!("record-backed Ticket") + }; + assert_eq!(command.record_path, child_path); + assert_eq!(command.stage.as_ref().unwrap().age_seconds, 1800); + assert_eq!( + selected.observation.sessions().unwrap(), + [&retained_reference, &implement, &resume, &previous, &repair] + .into_iter() + .cloned() + .collect::>() + ); + assert_eq!(selected.observation.session("implement").unwrap(), resume); + assert_eq!(selected.observation.session("repair").unwrap(), repair); + assert_eq!(selected.observation.session("repair-1").unwrap(), previous); + assert_eq!( + selected.stop_target("widgets#21").unwrap_err().to_string(), + "cannot stop widgets#21: child Run; stop Spec run acme/widgets#20 instead" + ); +} + +#[test] +fn historical_selection_keeps_unknown_facts_and_has_no_process_or_stop_authority() { + let dir = tempfile::tempdir().unwrap(); + let folder = dir.path().join("acme/widgets/commands/issue"); + fs::create_dir_all(&folder).unwrap(); + let session = folder.join("7-20261019T110000-0400~history-implement.jsonl"); + fs::write(&session, "not interpreted as a live event").unwrap(); + let snapshot = Snapshot::load(dir.path(), now()).unwrap(); + assert!(snapshot.list(false).is_empty()); + assert_eq!(snapshot.list(true).len(), 1); + let selected = snapshot.resolve("widgets#7").unwrap(); + assert!(selected.owner.is_none()); + assert!(selected.child_owner.is_none()); + assert!(selected.process_exited()); + assert_eq!(selected.observation.session("implement").unwrap(), session); + let details = serde_json::to_value(selected.observation.details().unwrap()).unwrap(); + assert_eq!(details["state"], "unknown"); + assert_eq!(details["stages"], json!([])); + assert_eq!(details["stage"], Value::Null); + assert_eq!(details["harness"], Value::Null); + assert_eq!(details["model"], Value::Null); + assert_eq!(details["source"], "session_log"); + assert_eq!( + selected.stop_target("widgets#7").unwrap_err().to_string(), + "cannot stop widgets#7: Command unknown" + ); +} + +#[test] +fn outcome_ends_display_but_not_the_captured_process_for_following() { + let dir = tempfile::tempdir().unwrap(); + let mut ended = record(7); + ended["outcome"] = "ready for review".into(); + ended["ended_at"] = "2026-10-19T11:30:00-04:00".into(); + let path = write( + dir.path(), + "7-20261019T110000-0400~ended.command.json", + &ended, + ); + let snapshot = Snapshot::load(dir.path(), now()).unwrap(); + let selected = snapshot.resolve("widgets#7").unwrap(); + assert_eq!(selected.owner.unwrap().state, "ended"); + assert!(!selected.process_exited()); + assert_eq!( + selected.stop_target("widgets#7").unwrap_err().to_string(), + "cannot stop widgets#7: Command ended: ready for review" + ); + // A later record cannot retarget this captured identity. + ended["process_start"] = "reused-process-identity".into(); + fs::write(&path, serde_json::to_vec(&ended).unwrap()).unwrap(); + assert!(!selected.process_exited()); + let refreshed = Snapshot::load(dir.path(), now()).unwrap(); + assert!(refreshed.resolve("widgets#7").unwrap().process_exited()); +} diff --git a/src/logs/status.rs b/src/logs/status.rs index be1d11b23..e6ee22ff8 100644 --- a/src/logs/status.rs +++ b/src/logs/status.rs @@ -1,161 +1,85 @@ -//! Read machine-wide Command records without entering a Launch directory. +//! Format machine-wide Command observations. -mod history; pub mod inspect; -use std::collections::{BTreeMap, BTreeSet, HashSet}; -use std::fs; -use std::path::{Path, PathBuf}; - -use anyhow::{Context, Result}; -use chrono::{DateTime, Local}; +use anyhow::Result; +use chrono::Local; use serde::Serialize; -use super::command_record::{CommandRecord, CommandTree}; -use super::effects::OnMachine; -use super::{Pass, activity}; +use super::command_record::CommandRecord; +use super::reading::{Command, Entry, IdlePass, Snapshot}; use crate::config::UserConfig; use crate::issue::IssueUrl; #[derive(Serialize)] -struct Status { +struct Status<'a> { version: u32, hostname: String, - commands: Vec, - idle_passes: Vec, -} - -#[derive(Serialize)] -struct IdlePass { - repository: String, - kind: String, - reason: String, -} - -#[derive(Serialize)] -#[serde(untagged)] -enum Entry { - Record(Box), - History(Box), -} - -impl Entry { - fn ordering( - &self, - all: bool, - ) -> ( - bool, - std::cmp::Reverse>>, - &str, - ) { - let (state, started, repo) = match self { - Self::Record(command) => ( - command.state, - command.record.started_at.as_str(), - command.record.repository.as_str(), - ), - Self::History(command) => ( - command.state, - command.started_at.as_str(), - command.repository.as_str(), - ), - }; - ( - !all && state != "live", - std::cmp::Reverse(DateTime::parse_from_rfc3339(started).ok()), - repo, - ) + commands: Vec<&'a Entry>, + idle_passes: &'a [IdlePass], +} + +fn print_command(command: &Command, indent: &str) { + let facts = &command.record; + let stage = stage_text(command); + let subject = match &facts.issue { + Some(issue) => format!( + "{}#{} {}", + facts.repository, + issue.number, + issue.title.as_deref().unwrap_or("(title unavailable)") + ), + None => format!("{} on {}", facts.name, facts.repository), + }; + println!( + "{indent}{} {subject} · {} · {} · {} · {stage}", + facts.kind, + facts.harness, + facts.model.as_deref().unwrap_or("default Model"), + facts.effort.as_deref().unwrap_or("default Effort") + ); + print_tickets(facts, &format!("{indent} ")); + for dispatched in &command.dispatched { + print_command(dispatched, &format!("{indent} ")); } -} - -#[derive(Serialize)] -struct Command { - #[serde(skip)] - record_path: PathBuf, - #[serde(flatten)] - record: CommandRecord, - state: &'static str, - stage: Option, - children: Vec, - source: &'static str, - command_log: PathBuf, - dispatched: Vec, -} - -#[derive(Serialize)] -struct CurrentStage { - name: String, - began_at: String, - age_seconds: i64, -} - -impl Command { - fn new(record: CommandRecord, record_path: PathBuf, now: DateTime) -> Self { - let state = if record.outcome.is_some() { - "ended" - } else if record.live(&mut OnMachine) { - "live" - } else { - "died" - }; - let stage = record.current_stage().map(|stage| CurrentStage { - name: stage.name.clone(), - began_at: stage.began_at.clone(), - age_seconds: DateTime::parse_from_rfc3339(&stage.began_at) - .map(|began| now.signed_duration_since(began).num_seconds().max(0)) - .unwrap_or(0), - }); - Self { - command_log: command_log(&record_path), - record_path, - record, - state, - stage, - children: Vec::new(), - source: "command_record", - dispatched: Vec::new(), - } + for child in &command.children { + print_command(child, &format!("{indent} ")); } +} - fn print(&self, indent: &str) { - let facts = &self.record; - let stage = stage_text(self); - let subject = match &facts.issue { - Some(issue) => format!( - "{}#{} {}", - facts.repository, - issue.number, - issue.title.as_deref().unwrap_or("(title unavailable)") - ), - None => format!("{} on {}", facts.name, facts.repository), - }; - println!( - "{indent}{} {subject} · {} · {} · {} · {stage}", - facts.kind, - facts.harness, - facts.model.as_deref().unwrap_or("default Model"), - facts.effort.as_deref().unwrap_or("default Effort") - ); - print_tickets(facts, &format!("{indent} ")); - for dispatched in &self.dispatched { - dispatched.print(&format!("{indent} ")); - } - for child in &self.children { - child.print(&format!("{indent} ")); - } - } +pub(super) fn print_history(command: &super::reading::HistoricalCommand) { + let issue = command + .issue + .as_ref() + .map(|issue| format!("#{}", issue.number)) + .unwrap_or_default(); + println!( + "{}{issue} {} · {} → {} · {}", + command.repository, + command.kind, + command.started_at, + command.ended_at.as_deref().unwrap_or("unknown end"), + command.outcome.as_deref().unwrap_or("unknown outcome") + ); } pub fn show(json: bool, all: bool) -> Result<()> { - let status = read(all)?; + let config = UserConfig::load()?; + let snapshot = Snapshot::load(&config.logs_dir, Local::now())?; + let status = Status { + version: 1, + hostname: crate::host::name().unwrap_or_else(|| "unknown".into()), + commands: snapshot.list(all), + idle_passes: &snapshot.idle_passes, + }; if json { println!("{}", serde_json::to_string_pretty(&status)?); } else { println!("{}", status.hostname); for entry in status.commands { match entry { - Entry::Record(command) => command.print(""), - Entry::History(command) => command.print(), + Entry::Record(command) => print_command(command, ""), + Entry::History(command) => print_history(command), } } for pass in status.idle_passes { @@ -165,176 +89,7 @@ pub fn show(json: bool, all: bool) -> Result<()> { Ok(()) } -fn read(all: bool) -> Result { - let config = UserConfig::load()?; - let now = Local::now(); - let mut records = BTreeMap::new(); - let mut commands = Vec::new(); - let mut idle_passes = Vec::new(); - for owner in directories(&config.logs_dir)? { - for repo in directories(&owner)? { - let mut covered = HashSet::new(); - idle_passes.extend(read_idle_passes(&repo)?); - for folder in directories(&repo.join("commands"))? { - for (path, tree) in read_records(&folder)? { - let record = &tree.command; - if record.parent_record.is_none() { - covered.insert(command_log(&path)); - } - for session in std::iter::once(record) - .chain(&tree.dispatched) - .flat_map(CommandRecord::session_references) - { - covered.insert(session.clone()); - } - records.insert(path, tree); - } - } - commands.extend( - history::read(&repo, &covered)? - .into_iter() - .filter(|command| all || recent(command.ended_at.as_deref(), now)) - .map(|command| Entry::History(Box::new(command))), - ); - } - } - let roots: Vec<_> = records - .iter() - .filter(|(_, tree)| { - let record = &tree.command; - record.parent_record.is_none() - && (all || record.outcome.is_none() || recent(record.ended_at.as_deref(), now)) - }) - .map(|(path, _)| path.clone()) - .collect(); - commands.extend( - roots - .into_iter() - .map(|path| Entry::Record(Box::new(tree(&path, &mut records, &now)))), - ); - commands.sort_by(|a, b| a.ordering(all).cmp(&b.ordering(all))); - idle_passes.sort_by(|a, b| (&a.repository, &a.kind).cmp(&(&b.repository, &b.kind))); - Ok(Status { - version: 1, - hostname: crate::host::name().unwrap_or_else(|| "unknown".into()), - commands, - idle_passes, - }) -} - -fn read_records(folder: &Path) -> Result> { - let mut records = BTreeMap::new(); - for entry in entries(folder)? { - let name = entry.file_name(); - let name = name.to_string_lossy(); - if !entry.file_type()?.is_file() - || (!name.ends_with(".command.json") && !name.ends_with(".child.json")) - { - continue; - } - let path = entry.path(); - let bytes = - fs::read(&path).with_context(|| format!("could not read {}", path.display()))?; - let Ok(mut tree) = serde_json::from_slice::(&bytes) else { - continue; - }; - let record = &mut tree.command; - if record.version != 1 { - continue; - } - // Records written before completion times existed were last - // replaced when their outcome was recorded. - if record.outcome.is_some() && record.ended_at.is_none() { - record.ended_at = entry - .metadata()? - .modified() - .ok() - .map(|time| DateTime::::from(time).to_rfc3339()); - } - records.insert(path, tree); - } - Ok(records) -} - -/// Stop uses the same exact-parent tree as status, including child-owned files. -pub(super) fn retained_resources(path: &Path) -> Result> { - let mut records = read_records(path.parent().context("Command record has no directory")?)?; - anyhow::ensure!(records.contains_key(path), "Command record is unavailable"); - let command = tree(path, &mut records, &Local::now()); - let mut resources = BTreeSet::new(); - fn collect(command: &Command, resources: &mut BTreeSet) { - if let Some(branch) = &command.record.issue_branch { - resources.insert(format!("Issue branch left: {branch}")); - } - if let Some(pr) = &command.record.pull_request { - resources.insert(format!("Pull request left: {pr}")); - } - for child in command.dispatched.iter().chain(&command.children) { - collect(child, resources); - } - } - collect(&command, &mut resources); - Ok(resources) -} - -fn command_log(path: &Path) -> PathBuf { - super::owner::record_log(path) -} - -/// Keep the filename spelling alongside its parsed time for naming and history. -fn command_stamp(path: &Path) -> Option<(&str, DateTime)> { - let decoded = super::owner::decode(path)?; - Some((decoded.identity.stamp, decoded.time())) -} - -fn tree( - path: &Path, - records: &mut BTreeMap, - now: &DateTime, -) -> Command { - let tree = records.remove(path).expect("a discovered record exists"); - let mut command = Command::new(tree.command, path.to_owned(), *now); - command.children = children(path, None, records, now); - command.dispatched = tree - .dispatched - .into_iter() - .map(|record| { - let mut dispatched = Command::new(record, path.to_owned(), *now); - let issue = dispatched - .record - .issue - .as_ref() - .map(|issue| issue.url.as_str()); - dispatched.children = children(path, issue, records, now); - dispatched - }) - .collect(); - command -} - -fn children( - path: &Path, - dispatch: Option<&str>, - records: &mut BTreeMap, - now: &DateTime, -) -> Vec { - let paths: Vec<_> = records - .iter() - .filter(|(_, child)| { - child.command.parent_record.as_deref() == Some(path) - && child.command.parent_dispatch.as_deref() == dispatch - }) - .map(|(path, _)| path.clone()) - .collect(); - let mut children: Vec<_> = paths - .into_iter() - .map(|path| tree(&path, records, now)) - .collect(); - children.sort_by_key(|child| child.record.issue.as_ref().map(|issue| issue.number)); - children -} - -fn stage_text(command: &Command) -> String { +pub(super) fn stage_text(command: &Command) -> String { if let Some(outcome) = &command.record.outcome { return outcome.clone(); } @@ -356,7 +111,7 @@ fn stage_text(command: &Command) -> String { ) } -fn print_tickets(facts: &CommandRecord, indent: &str) { +pub(super) fn print_tickets(facts: &CommandRecord, indent: &str) { for ticket in &facts.tickets { use super::command_record::TicketState; let state = match ticket.state { @@ -386,46 +141,6 @@ fn print_tickets(facts: &CommandRecord, indent: &str) { } } -fn recent(ended: Option<&str>, now: DateTime) -> bool { - ended - .and_then(|time| DateTime::parse_from_rfc3339(time).ok()) - .is_some_and(|ended| ended >= now - chrono::Duration::hours(24)) -} - -fn read_idle_passes(repo: &Path) -> Result> { - let path = repo.join(activity::FILE); - let activity = match fs::read_to_string(&path) { - Ok(activity) => activity, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(error) => { - return Err(error).with_context(|| format!("could not read {}", path.display())); - } - }; - let repository = format!( - "{}/{}", - repo.parent() - .unwrap() - .file_name() - .unwrap() - .to_string_lossy(), - repo.file_name().unwrap().to_string_lossy() - ); - let mut idle = Vec::new(); - for pass in [Pass::Architect, Pass::Pickup, Pass::Security] { - let kind = activity::Kind::Pass(pass); - if let Some(line) = activity::last_of(&activity, kind) - && let Some(reason) = line.strip_prefix(&format!("{kind} skipped: ")) - { - idle.push(IdlePass { - repository: repository.clone(), - kind: kind.to_string(), - reason: reason.into(), - }); - } - } - Ok(idle) -} - fn age(seconds: i64) -> String { if seconds < 60 { format!("{seconds} sec") @@ -435,21 +150,3 @@ fn age(seconds: i64) -> String { format!("{} hr {} min", seconds / 3600, seconds % 3600 / 60) } } - -fn entries(path: &Path) -> Result> { - match fs::read_dir(path) { - Ok(entries) => entries.collect::>().map_err(Into::into), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()), - Err(error) => Err(error).with_context(|| format!("could not list {}", path.display())), - } -} - -fn directories(path: &Path) -> Result> { - let mut paths = Vec::new(); - for entry in entries(path)? { - if entry.file_type()?.is_dir() { - paths.push(entry.path()); - } - } - Ok(paths) -} diff --git a/src/logs/status/inspect.rs b/src/logs/status/inspect.rs index 6049fd147..1afd0819d 100644 --- a/src/logs/status/inspect.rs +++ b/src/logs/status/inspect.rs @@ -1,279 +1,28 @@ -//! Resolve human Command names and read one Command's state and logs. +//! Render one Command and stream its selected log bytes. -use std::collections::BTreeSet; -use std::io::{self, Read, Seek, SeekFrom, Write}; -use std::path::{Path, PathBuf}; +use std::io::{self, Write}; -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result}; +use chrono::Local; use serde::Serialize; -use super::{Command, Entry}; -use crate::issue::IssueUrl; -use crate::logs::effects::Outside; -use crate::logs::owner; +use crate::config::UserConfig; +use crate::logs::reading::{Command, Observation, Snapshot}; -enum Name { - Issue { repository: String, number: u64 }, - Pass(String), -} - -impl Name { - fn parse(name: &str) -> Result { - if name.starts_with("https://") { - let issue = IssueUrl::parse(name)?; - return Ok(Self::Issue { - repository: issue.repo_slug(), - number: issue.number, - }); - } - if let Some((repository, number)) = name.split_once('#') { - let number = number - .parse::() - .with_context(|| format!("invalid Command name: {name}"))? - .get(); - if !repository.is_empty() { - return Ok(Self::Issue { - repository: repository.into(), - number, - }); - } - } - if ["architect-", "pickup-", "secure-"] - .iter() - .any(|prefix| name.starts_with(prefix)) - { - return Ok(Self::Pass(name.into())); - } - bail!("invalid Command name: {name}; use repo#number, an Issue URL or kind-stamp") - } - - fn matches_repo(&self, repo: &str) -> bool { - match self { - Self::Issue { repository, .. } => { - repo.eq_ignore_ascii_case(repository) - || (!repository.contains('/') - && repo - .rsplit('/') - .next() - .is_some_and(|name| name.eq_ignore_ascii_case(repository))) - } - Self::Pass(_) => true, - } - } -} - -#[derive(Clone, Copy, Serialize)] -#[serde(untagged)] -enum Selected<'a> { - Record(&'a Command), - History(&'a super::history::Command), -} - -impl<'a> Selected<'a> { - fn repository(self) -> &'a str { - match self { - Self::Record(command) => &command.record.repository, - Self::History(command) => &command.repository, - } - } - - fn matches(self, name: &Name) -> bool { - if !name.matches_repo(self.repository()) { - return false; - } - match name { - Name::Issue { number, .. } => { - let issue = match self { - Self::Record(command) => command.record.issue.as_ref(), - Self::History(command) => command.issue.as_ref(), - }; - issue.is_some_and(|issue| issue.number == *number) - } - Name::Pass(name) => match self { - Self::Record(command) => command.record.name == *name, - Self::History(command) => command.command_log.as_ref().is_some_and(|path| { - let Some(kind) = pass_prefix(&command.kind) else { - return false; - }; - super::command_stamp(path) - .is_some_and(|(stamp, _)| *name == format!("{kind}-{stamp}")) - }), - }, - } - } - - fn ordering( - self, - ) -> ( - bool, - std::cmp::Reverse>>, - ) { - let (state, started) = match self { - Self::Record(command) => (command.state, command.record.started_at.as_str()), - Self::History(command) => (command.state, command.started_at.as_str()), - }; - ( - state != "live", - std::cmp::Reverse(chrono::DateTime::parse_from_rfc3339(started).ok()), - ) - } - - fn command_log(self) -> Option<&'a Path> { - match self { - Self::Record(command) => Some(&command.command_log), - Self::History(command) => command.command_log.as_deref(), - } - } - - fn print(self) -> Result<()> { - match self { - Self::Record(command) => print_details(command, "")?, - Self::History(command) => command.print(), - } - if let Some(path) = self.command_log() { - println!("Command log: {}", path.display()); - } - if matches!(self, Self::History(_)) { - for path in self.sessions()? { - println!("Session log: {}", path.display()); - } - } - Ok(()) - } - - fn sessions(self) -> Result> { - let (mut paths, issue, kind) = match self { - Self::Record(command) => { - let mut paths = Vec::new(); - for path in command.record.session_references() { - if !paths.contains(path) { - paths.push(path.clone()); - } - } - ( - paths, - command.record.issue.as_ref(), - command.record.kind.as_str(), - ) - } - Self::History(command) => ( - command.session_logs.clone(), - command.issue.as_ref(), - command.kind.as_str(), - ), - }; - // A stage can cover several sessions (for example Architecture review - // and its Resume), so stage references alone are not a log inventory. - if let Some(log) = self.command_log() - && owner::decode(log).is_some() - { - let prefix = issue - .map(|issue| issue.number.to_string()) - .or_else(|| pass_prefix(kind).map(String::from)) - .context("cannot name Session logs for this Command kind")?; - for entry in super::entries(log.parent().context("Command log has no directory")?)? { - let path = entry.path(); - if entry.file_type()?.is_file() - && owner::decode(&path).is_some_and(|decoded| { - decoded.session_kind.is_some() && decoded.subject == prefix - }) - && owner::session_matches(&path, log) - && !paths.contains(&path) - { - paths.push(path); - } - } - } - let mut dated = paths - .into_iter() - .map(|path| { - let modified = std::fs::metadata(&path) - .and_then(|metadata| metadata.modified()) - .unwrap_or(std::time::SystemTime::UNIX_EPOCH); - (modified, path) - }) - .collect::>(); - dated.sort(); - Ok(dated.into_iter().map(|(_, path)| path).collect()) - } - - fn session(self, kind: &str) -> Result { - self.sessions()? - .into_iter() - .rev() - .find(|path| { - let Some(session) = owner::decode(path).and_then(|decoded| decoded.session_kind) - else { - return false; - }; - let original = session.strip_suffix("-resume").unwrap_or(session); - session == kind - || original == kind - || (kind == "repair" - && original - .strip_prefix("repair-") - .is_some_and(|number| number.parse::().is_ok())) - }) - .with_context(|| format!("no {kind} Session log for this Command")) +fn print(command: Observation<'_>) -> Result<()> { + match command { + Observation::Record(command) => print_details(command, "")?, + Observation::History(command) => super::print_history(command), } -} - -fn pass_prefix(kind: &str) -> Option<&'static str> { - match kind { - "Architect run" => Some("architect"), - "Pickup run" => Some("pickup"), - "Security run" => Some("secure"), - _ => None, + if let Some(path) = command.command_log() { + println!("Command log: {}", path.display()); } -} - -fn records<'a>(command: &'a Command, found: &mut Vec>) { - found.push(Selected::Record(command)); - for child in command.dispatched.iter().chain(&command.children) { - records(child, found); - } -} - -fn select<'a>(entries: &'a [Entry], input: &str) -> Result> { - let name = Name::parse(input)?; - let mut candidates = Vec::new(); - for entry in entries { - match entry { - Entry::Record(command) => records(command, &mut candidates), - Entry::History(command) => candidates.push(Selected::History(command)), + if matches!(command, Observation::History(_)) { + for path in command.sessions()? { + println!("Session log: {}", path.display()); } } - let repositories: BTreeSet<_> = candidates - .iter() - .filter(|command| name.matches_repo(command.repository())) - .map(|command| command.repository().to_ascii_lowercase()) - .collect(); - if matches!(&name, Name::Issue { repository, .. } if !repository.contains('/')) - && repositories.len() > 1 - { - bail!( - "ambiguous Command name {input}; matching repositories: {}", - repositories.into_iter().collect::>().join(", ") - ); - } - candidates.retain(|command| command.matches(&name)); - if matches!(name, Name::Pass(_)) { - let repositories: BTreeSet<_> = candidates - .iter() - .map(|command| command.repository()) - .collect(); - if repositories.len() > 1 { - bail!( - "ambiguous Command name {input}; matching repositories: {}", - repositories.into_iter().collect::>().join(", ") - ); - } - } - candidates.sort_by_key(|command| command.ordering()); - candidates - .into_iter() - .next() - .with_context(|| format!("no Command matches {input}")) + Ok(()) } fn print_details(command: &Command, indent: &str) -> Result<()> { @@ -292,10 +41,10 @@ fn print_details(command: &Command, indent: &str) -> Result<()> { for stage in &command.record.stages { println!("{indent} {} · {}", stage.name, stage.began_at); } - for path in Selected::Record(command).sessions()? { + for path in command.sessions()? { println!("{indent}Session log: {}", path.display()); } - if let Some(event) = latest_event(command) { + if let Some(event) = command.latest_event() { println!("{indent}{event}"); } super::print_tickets(&command.record, &format!("{indent} ")); @@ -306,154 +55,35 @@ fn print_details(command: &Command, indent: &str) -> Result<()> { } pub fn show(input: &str, json: bool) -> Result<()> { - let status = super::read(true)?; - let command = select(&status.commands, input)?; + let config = UserConfig::load()?; + let snapshot = Snapshot::load(&config.logs_dir, Local::now())?; + let command = snapshot.resolve(input)?.observation; if json { #[derive(Serialize)] struct Show<'a> { version: u32, hostname: String, - command: Detailed<'a>, + command: crate::logs::reading::Detailed<'a>, } println!( "{}", serde_json::to_string_pretty(&Show { version: 1, - hostname: status.hostname, - command: match command { - Selected::Record(command) => Detailed::Record(Box::new(Details::new(command)?)), - Selected::History(command) => { - let mut details = command.clone(); - details.session_logs = Selected::History(command).sessions()?; - Detailed::History(Box::new(details)) - } - } + hostname: crate::host::name().unwrap_or_else(|| "unknown".into()), + command: command.details()?, })? ); } else { - command.print()?; + print(command)?; } Ok(()) } -/// Resolve Stop through the same naming rules as show and logs. A dispatched -/// Run belongs to its owning Command, whose process and record are stopped. -pub(in crate::logs) fn stop_target(input: &str) -> Result { - let status = super::read(true)?; - match select(&status.commands, input)? { - Selected::Record(command) if command.record.parent_record.is_some() => { - let owner = status - .commands - .iter() - .find_map(|entry| { - let Entry::Record(root) = entry else { - return None; - }; - contains(root, command).then(|| { - root.dispatched - .iter() - .find(|dispatch| contains(dispatch, command)) - .unwrap_or(root) - }) - }) - .context("could not find the Command owning this child Run")?; - bail!( - "cannot stop {input}: child Run; stop {} {} instead", - owner.record.kind, - owner.record.name - ); - } - Selected::Record(command) if command.state == "live" => Ok(command.record_path.clone()), - Selected::Record(command) => bail!( - "cannot stop {input}: Command {}{}", - command.state, - command - .record - .outcome - .as_ref() - .map(|outcome| format!(": {outcome}")) - .unwrap_or_default() - ), - Selected::History(command) => bail!("cannot stop {input}: Command {}", command.state), - } -} - -fn latest_event(command: &Command) -> Option { - if command.state != "live" { - return None; - } - let path = command.record.session_log.as_ref()?; - let mut file = std::fs::File::open(path).ok()?; - // A show reads at most 64 KiB, independent of the transcript's size. - let start = file.metadata().ok()?.len().saturating_sub(64 * 1024); - file.seek(SeekFrom::Start(start)).ok()?; - let mut bytes = Vec::new(); - file.take(64 * 1024).read_to_end(&mut bytes).ok()?; - let tail = String::from_utf8_lossy(&bytes); - let mut lines = tail.lines(); - if start > 0 { - lines.next(); - } - let harness = crate::harness::Harness::named(&command.record.harness)?; - let mut stream = harness.adapter().interpretation(Path::new(""), ""); - let event = lines.flat_map(|line| stream.condense(line)).last()?; - let event = event - .strip_prefix("$ ") - .map_or_else(|| event.clone(), |command| format!("Bash: {command}")); - Some(event.lines().collect::>().join(" ")) -} - -#[derive(Serialize)] -#[serde(untagged)] -enum Detailed<'a> { - Record(Box>), - History(Box), -} - -/// The detailed view adds a complete Session inventory and a live event to -/// each Run while retaining the status reader's version 1 field names. -#[derive(Serialize)] -struct Details<'a> { - #[serde(flatten)] - record: &'a crate::logs::command_record::CommandRecord, - state: &'static str, - stage: &'a Option, - source: &'static str, - command_log: &'a Path, - session_logs: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - latest_event: Option, - children: Vec>, - dispatched: Vec>, -} - -impl<'a> Details<'a> { - fn new(command: &'a Command) -> Result { - Ok(Self { - record: &command.record, - state: command.state, - stage: &command.stage, - source: command.source, - command_log: &command.command_log, - session_logs: Selected::Record(command).sessions()?, - latest_event: latest_event(command), - children: command - .children - .iter() - .map(Self::new) - .collect::>()?, - dispatched: command - .dispatched - .iter() - .map(Self::new) - .collect::>()?, - }) - } -} - pub fn logs(input: &str, session: Option<&str>, follow: bool) -> Result<()> { - let status = super::read(true)?; - let command = select(&status.commands, input)?; + let config = UserConfig::load()?; + let snapshot = Snapshot::load(&config.logs_dir, Local::now())?; + let selection = snapshot.resolve(input)?; + let command = selection.observation; let path = match session { Some(kind) => command.session(kind)?, None => command @@ -463,19 +93,11 @@ pub fn logs(input: &str, session: Option<&str>, follow: bool) -> Result<()> { }; let mut file = std::fs::File::open(&path).with_context(|| format!("could not open {}", path.display()))?; - let owner = status.commands.iter().find_map(|entry| { - if let (Entry::Record(root), Selected::Record(selected)) = (entry, command) - && contains(root, selected) - { - return Some(root.as_ref()); - } - None - }); let mut stdout = io::stdout().lock(); loop { // Check before draining, so a completed Command's final bytes are // always copied even when it ended between two polls. - let ended = owner.is_none_or(ended); + let ended = selection.process_exited(); io::copy(&mut file, &mut stdout)?; stdout.flush()?; if !follow || ended { @@ -484,21 +106,3 @@ pub fn logs(input: &str, session: Option<&str>, follow: bool) -> Result<()> { std::thread::sleep(std::time::Duration::from_millis(100)); } } - -fn contains(root: &Command, selected: &Command) -> bool { - std::ptr::eq(root, selected) - || root - .dispatched - .iter() - .chain(&root.children) - .any(|child| contains(child, selected)) -} - -fn ended(command: &Command) -> bool { - let record = &command.record; - // The outcome is recorded before the final summary and notification are - // printed. Follow the captured process identity through that last output. - record.process_start.is_empty() - || super::OnMachine.process_start(record.process_id).as_deref() - != Some(&record.process_start) -} diff --git a/src/logs/stop.rs b/src/logs/stop.rs index 2b9d3730b..4e75f632f 100644 --- a/src/logs/stop.rs +++ b/src/logs/stop.rs @@ -9,6 +9,8 @@ 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 { @@ -59,7 +61,9 @@ fn require_live(record: &CommandRecord) -> Result<()> { } pub fn stop(input: &str, requeue: bool) -> Result<()> { - let path = super::status::inspect::stop_target(input)?; + 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 user = std::process::Command::new("id") @@ -138,8 +142,12 @@ fn print_ending(path: &Path, tree: &CommandTree, output: &mut impl Write) -> Res .as_deref() .unwrap_or("died before recording an outcome") )?; - for resource in super::status::retained_resources(path)? { - writeln!(output, "{resource}")?; + let work = reading::retained_work(path, chrono::Local::now())?; + for branch in work.branches { + writeln!(output, "Issue branch left: {branch}")?; + } + for pr in work.pull_requests { + writeln!(output, "Pull request left: {pr}")?; } Ok(()) } diff --git a/tests/show_logs.rs b/tests/show_logs.rs index 09c595b0a..35a184b6d 100644 --- a/tests/show_logs.rs +++ b/tests/show_logs.rs @@ -405,6 +405,64 @@ fn logs_follow_streams_appended_bytes_and_exits_when_the_command_ends() { ); } +#[test] +fn logs_follow_drains_final_bytes_after_an_outcome_or_selected_child_ends() { + for selected in ["widgets#7", "widgets#21"] { + let scenario = Scenario::new(); + scenario.agent_does(&format!( + "touch '{}'\nwhile ! test -f '{}'; do sleep 0.02; done\necho '{{\"type\":\"assistant\",\"message\":{{\"content\":[{{\"type\":\"tool_use\",\"name\":\"Bash\",\"input\":{{\"command\":\"echo final bytes after durable outcome\"}}}}]}}}}'\n{OPENS_PR}", + scenario.path("started").display(), + scenario.path("release").display() + )); + let held = scenario.run_until(&[&scenario.issue_url(7)], &[], "started"); + let folder = scenario.path("home/.thirdshift/logs/acme/widgets/commands/issue"); + let path = fs::read_dir(&folder) + .unwrap() + .map(|entry| entry.unwrap().path()) + .find(|path| path.to_string_lossy().ends_with(".command.json")) + .unwrap(); + let mut record: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + record["outcome"] = "ready for review".into(); + // The durable outcome can precede the outer process's last output. + fs::write(&path, serde_json::to_vec(&record).unwrap()).unwrap(); + let mut child = record.clone(); + child["name"] = "acme/widgets#21".into(); + child["issue"]["number"] = 21.into(); + child["issue"]["url"] = scenario.issue_url(21).into(); + child["parent_record"] = serde_json::json!(path); + fs::write( + path.with_extension("issue-21.child.json"), + serde_json::to_vec(&child).unwrap(), + ) + .unwrap(); + let status = scenario.run(&["status", "--json"]); + let status: Value = serde_json::from_str(&status.stdout).unwrap(); + let ended_for_display = status["commands"][0]["state"] == "ended"; + let output = scenario.path("followed.txt"); + let follower = scenario.read_to_file(&["logs", selected, "--follow"], &output); + let deadline = Instant::now() + Duration::from_secs(5); + while fs::read_to_string(&output).unwrap().is_empty() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + // Allow a follower that wrongly treats the outcome as process exit to + // finish its first poll before releasing the final output. + std::thread::sleep(Duration::from_millis(200)); + fs::write(scenario.path("release"), "").unwrap(); + let ended = held.finish(); + let followed = follower.finish(); + assert_eq!(ended.code, Some(0), "{}", ended.stderr); + assert_eq!(followed.code, Some(0), "{}", followed.stderr); + assert!(ended_for_display, "{status}"); + let followed = fs::read_to_string(output).unwrap(); + assert!( + followed.contains("final bytes after durable outcome"), + "{selected}: {followed}" + ); + let log = status["commands"][0]["command_log"].as_str().unwrap(); + assert_eq!(followed, fs::read_to_string(log).unwrap()); + } +} + #[test] fn logs_session_repair_selects_the_latest_numbered_repair() { let scenario = Scenario::new();