From 8be942194f374f90722374a63fe727887c769a61 Mon Sep 17 00:00:00 2001 From: sebi Date: Fri, 26 Jun 2026 19:40:26 -0500 Subject: [PATCH 1/6] feat(jobs): JobId newtype with slug-from-cmd + local HH:MM MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New src/jobs/id.rs: `JobId(String)` matching `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$` — e.g. `cargo-build-23:30`, replacing opaque counter ids with conversation-friendly ones. - slug(): first 3 cmd tokens, lowercased, non-alnum→`-`, collapsed, trimmed, capped at 24; falls back to `job` when nothing survives. - generate(cmd, &AtomicU64, exists): clean base id; appends `-{seq}` only on collision. Takes an `exists` predicate (vs. bare signature) so collision detection lives in this module and stays decoupled from the store's id map — required for the collision unit test. - chrono (clock, no default features) for `Local` HH:MM. - 10 unit tests: slug rules, format regex shape, collision suffix, Display/AsRef, transparent serialize. - #[allow(dead_code)] until the next change wires JobId into JobStore. Co-Authored-By: Claude --- Cargo.lock | 1 + Cargo.toml | 3 + src/jobs/id.rs | 189 ++++++++++++++++++++++++++++++++++++++++++++++++ src/jobs/mod.rs | 1 + 4 files changed, 194 insertions(+) create mode 100644 src/jobs/id.rs diff --git a/Cargo.lock b/Cargo.lock index 8875f2f..32b1e29 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -721,6 +721,7 @@ dependencies = [ "anyhow", "axum", "base64", + "chrono", "clap", "rand 0.8.6", "rmcp", diff --git a/Cargo.toml b/Cargo.toml index b7478e5..dcf12a0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,9 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" schemars = "1" +# time — local HH:MM stamped into human-readable job ids +chrono = { version = "0.4", default-features = false, features = ["clock"] } + # cli + config clap = { version = "4", features = ["derive"] } toml = "0.8" diff --git a/src/jobs/id.rs b/src/jobs/id.rs new file mode 100644 index 0000000..40e81c4 --- /dev/null +++ b/src/jobs/id.rs @@ -0,0 +1,189 @@ +//! Human-readable job ids: a slug from the command plus the local `HH:MM` it +//! started, e.g. `cargo-build-23:30` — far easier to refer to in conversation +//! than an opaque counter. A monotonic sequence suffix disambiguates the rare +//! case of two identical-slug commands starting within the same minute. + +// Staged: `JobStore` constructs `JobId` in the follow-up change that wires it +// through the engine; this module's tests exercise it now, so allow dead code +// until then to keep `clippy -D warnings` green. +#![allow(dead_code)] + +use std::fmt; +use std::sync::atomic::{AtomicU64, Ordering}; + +use chrono::Local; + +/// Max slug length before the `-HH:MM` suffix. Short enough to type, long enough +/// to carry the gist of the command. +const MAX_SLUG: usize = 24; + +/// A human-readable job identifier: `-` with an optional `-` +/// collision suffix. Always matches `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$`. +#[derive(Debug, Clone, Hash, Eq, PartialEq, serde::Serialize)] +pub struct JobId(String); + +impl JobId { + /// Build an id from `cmd` and the current local time. `exists` reports + /// whether a candidate is already taken; only on a clash do we append a + /// `-` drawn from `seq`, so clean ids stay clean. The predicate keeps + /// this decoupled from however the caller tracks live ids. + pub fn generate(cmd: &str, seq: &AtomicU64, exists: impl Fn(&str) -> bool) -> Self { + let base = format!("{}-{}", slug(cmd), Local::now().format("%H:%M")); + if !exists(&base) { + return Self(base); + } + // Same slug, same minute: a monotonic suffix keeps the id unique without + // sacrificing readability. + let n = seq.fetch_add(1, Ordering::Relaxed); + Self(format!("{base}-{n}")) + } +} + +impl fmt::Display for JobId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl AsRef for JobId { + fn as_ref(&self) -> &str { + &self.0 + } +} + +/// Turn the first few command tokens into a lowercase `[a-z0-9-]` slug: collapse +/// runs of non-alphanumerics to single dashes, trim them off both ends, and cap +/// the length. Falls back to `job` when nothing alphanumeric survives. +fn slug(cmd: &str) -> String { + let head = cmd.split_whitespace().take(3).collect::>().join(" "); + let mut out = String::with_capacity(head.len()); + let mut prev_dash = false; + for ch in head.chars() { + if ch.is_ascii_alphanumeric() { + out.push(ch.to_ascii_lowercase()); + prev_dash = false; + } else if !prev_dash { + out.push('-'); + prev_dash = true; + } + } + // Cap first, then trim: the cut can land mid-run and leave a trailing dash. + let capped: String = out.trim_matches('-').chars().take(MAX_SLUG).collect(); + let trimmed = capped.trim_end_matches('-'); + if trimmed.is_empty() { + "job".to_string() + } else { + trimmed.to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Mirror the documented `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$` shape without + /// pulling in a regex dependency. The id has exactly one `:` (slug and + /// suffix never contain one), so split on it to isolate the time. + fn valid_format(s: &str) -> bool { + let two_digits = |x: &str| x.len() == 2 && x.bytes().all(|b| b.is_ascii_digit()); + let Some((left, right)) = s.split_once(':') else { + return false; + }; + let Some((prefix, hh)) = left.rsplit_once('-') else { + return false; + }; + let (mm, suffix_ok) = match right.split_once('-') { + Some((mm, sfx)) => ( + mm, + !sfx.is_empty() && sfx.bytes().all(|b| b.is_ascii_digit()), + ), + None => (right, true), + }; + two_digits(hh) + && two_digits(mm) + && suffix_ok + && !prefix.is_empty() + && prefix + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') + } + + #[test] + fn slug_lowercases_and_replaces_non_alnum() { + assert_eq!(slug("ECHO Hello/World"), "echo-hello-world"); + } + + #[test] + fn slug_takes_first_three_tokens_and_collapses_dashes() { + // 4th token dropped; the space-then-`-` run collapses to one dash. + assert_eq!(slug("git commit -m message extra"), "git-commit-m"); + } + + #[test] + fn slug_caps_length_at_24() { + let s = slug("abcdefghijklmnopqrstuvwxyz0123456789"); + assert_eq!(s.len(), 24); + assert_eq!(s, "abcdefghijklmnopqrstuvwx"); + } + + #[test] + fn slug_trims_dash_left_by_the_cap() { + // 23 chars then a separator: the cut lands on char 24, leaving a dash. + let input = format!("{} b", "a".repeat(23)); + assert_eq!(slug(&input), "a".repeat(23)); + } + + #[test] + fn slug_falls_back_when_nothing_alphanumeric() { + assert_eq!(slug(""), "job"); + assert_eq!(slug("!@#$ %^&*"), "job"); + } + + #[test] + fn generated_id_matches_documented_format() { + let seq = AtomicU64::new(1); + let id = JobId::generate("cargo build --release", &seq, |_| false); + assert!(valid_format(id.as_ref()), "bad format: {}", id.as_ref()); + } + + #[test] + fn no_collision_leaves_id_clean_and_seq_untouched() { + let seq = AtomicU64::new(1); + let id = JobId::generate("echo hi", &seq, |_| false); + let s = id.as_ref(); + assert!(valid_format(s)); + let (_, right) = s.split_once(':').expect("id carries a time"); + assert!(!right.contains('-'), "unexpected collision suffix: {s}"); + assert_eq!( + seq.load(Ordering::Relaxed), + 1, + "seq advanced without a clash" + ); + } + + #[test] + fn collision_appends_sequence_suffix() { + let seq = AtomicU64::new(1); + // `exists` always true => the base is "taken", forcing the suffix path. + let id = JobId::generate("echo hi", &seq, |_| true); + let s = id.as_ref(); + assert!(valid_format(s)); + assert!(s.ends_with("-1"), "expected `-1` seq suffix, got {s}"); + assert_eq!(seq.load(Ordering::Relaxed), 2, "clash must advance seq"); + } + + #[test] + fn display_and_as_ref_agree() { + let seq = AtomicU64::new(1); + let id = JobId::generate("ls", &seq, |_| false); + assert_eq!(id.to_string(), id.as_ref()); + } + + #[test] + fn serializes_as_a_plain_string() { + let seq = AtomicU64::new(1); + let id = JobId::generate("ls", &seq, |_| false); + let json = serde_json::to_string(&id).expect("serialize"); + assert_eq!(json, format!("\"{}\"", id.as_ref())); + } +} diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index 9fede31..ce145a0 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -14,6 +14,7 @@ use std::{ use tokio::sync::{Mutex, watch}; +mod id; mod reaper; use reaper::{kill_job, spawn_reaper}; From 9901ea70d8f159172cdacbaa292df45f74825427 Mon Sep 17 00:00:00 2001 From: sebi Date: Fri, 26 Jun 2026 19:50:48 -0500 Subject: [PATCH 2/6] feat(jobs): wire JobId through the engine - run() mints `slug-HH:MM` ids via id::generate under the jobs lock, so concurrent same-minute commands reserve unique ids without clobbering. - RunResult::Backgrounded, JobSummary, the jobs map, and the reaper all key on JobId; poll/kill take &JobId, wrapped at the tools edge from client input. - JobId gains Borrow (O(1) candidate probe), From/<&str> (lookup wrapping), Ord (list sort); drop the staging dead_code allow. Co-Authored-By: Claude --- src/jobs/id.rs | 33 ++++++++++++++++++++++------ src/jobs/mod.rs | 55 +++++++++++++++++++++++++++------------------- src/jobs/reaper.rs | 10 ++++----- src/tools/mod.rs | 4 +++- 4 files changed, 67 insertions(+), 35 deletions(-) diff --git a/src/jobs/id.rs b/src/jobs/id.rs index 40e81c4..298d427 100644 --- a/src/jobs/id.rs +++ b/src/jobs/id.rs @@ -3,11 +3,7 @@ //! than an opaque counter. A monotonic sequence suffix disambiguates the rare //! case of two identical-slug commands starting within the same minute. -// Staged: `JobStore` constructs `JobId` in the follow-up change that wires it -// through the engine; this module's tests exercise it now, so allow dead code -// until then to keep `clippy -D warnings` green. -#![allow(dead_code)] - +use std::borrow::Borrow; use std::fmt; use std::sync::atomic::{AtomicU64, Ordering}; @@ -18,8 +14,9 @@ use chrono::Local; const MAX_SLUG: usize = 24; /// A human-readable job identifier: `-` with an optional `-` -/// collision suffix. Always matches `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$`. -#[derive(Debug, Clone, Hash, Eq, PartialEq, serde::Serialize)] +/// collision suffix. A generated id matches `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$`; +/// ids wrapped from client input for lookup are not revalidated. +#[derive(Debug, Clone, Hash, Eq, PartialEq, PartialOrd, Ord, serde::Serialize)] pub struct JobId(String); impl JobId { @@ -51,6 +48,28 @@ impl AsRef for JobId { } } +/// Lets a `HashMap` be probed with a `&str` key — used when generating +/// an id to test a candidate against the live ids. +impl Borrow for JobId { + fn borrow(&self) -> &str { + &self.0 + } +} + +/// Wrap a client-supplied id for lookup. Not validated — an unknown or malformed +/// id simply matches no job; only `generate` mints well-formed ids. +impl From for JobId { + fn from(s: String) -> Self { + Self(s) + } +} + +impl From<&str> for JobId { + fn from(s: &str) -> Self { + Self(s.to_owned()) + } +} + /// Turn the first few command tokens into a lowercase `[a-z0-9-]` slug: collapse /// runs of non-alphanumerics to single dashes, trim them off both ends, and cap /// the length. Falls back to `job` when nothing alphanumeric survives. diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index ce145a0..ff8ebc2 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -5,10 +5,7 @@ use std::{ collections::HashMap, path::PathBuf, process::Stdio, - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, + sync::{Arc, atomic::AtomicU64}, time::Duration, }; @@ -17,6 +14,7 @@ use tokio::sync::{Mutex, watch}; mod id; mod reaper; +pub use id::JobId; use reaper::{kill_job, spawn_reaper}; #[derive(Debug, Clone, serde::Serialize)] @@ -50,7 +48,7 @@ pub enum RunResult { /// Finished within the inline window — output is ready now. Inline { state: JobState, page: Page }, /// Still running — poll this id. - Backgrounded { id: String }, + Backgrounded { id: JobId }, } /// One page of log lines plus a cursor to fetch the next page. @@ -64,7 +62,7 @@ pub struct Page { #[derive(Debug, serde::Serialize)] pub struct JobSummary { - pub id: String, + pub id: JobId, pub cmd: String, pub state: JobState, } @@ -74,7 +72,7 @@ pub struct JobStore { dir: PathBuf, inline_timeout: Duration, seq: Arc, - jobs: Arc>>>, + jobs: Arc>>>, } const DEFAULT_PAGE: usize = 200; @@ -101,7 +99,13 @@ impl JobStore { timeout_secs: Option, background: bool, ) -> std::io::Result { - let id = format!("j{}", self.seq.fetch_add(1, Ordering::Relaxed)); + // Hold the jobs lock across id generation and insertion so the id is + // reserved atomically: two identical commands launched in the same minute + // can't both see their `slug-HH:MM` as free and clobber each other's + // entry. Nothing awaits while the guard is held (file create + spawn are + // synchronous), so the critical section stays short. + let mut jobs = self.jobs.lock().await; + let id = JobId::generate(&cmd, &self.seq, |candidate| jobs.contains_key(candidate)); let log_path = self.dir.join(format!("{id}.log")); // ponytail: stdout+stderr merged into one log (terminal-style). Split into @@ -154,7 +158,8 @@ impl JobStore { done: rx.clone(), started: tokio::time::Instant::now(), }); - self.jobs.lock().await.insert(id.clone(), job); + jobs.insert(id.clone(), job); + drop(jobs); // `bg: true` — don't wait, hand back the id straight away. if background { @@ -181,7 +186,7 @@ impl JobStore { /// Status + one page of a job's log. pub async fn poll( &self, - id: &str, + id: &JobId, cursor: usize, limit: Option, ) -> Option<(JobState, Page)> { @@ -193,7 +198,7 @@ impl JobStore { pub async fn list(&self) -> Vec { // Snapshot while holding the map lock, then drop it before any .await. - let snapshot: Vec<(String, Arc)> = { + let snapshot: Vec<(JobId, Arc)> = { let jobs = self.jobs.lock().await; jobs.iter() .map(|(id, job)| (id.clone(), Arc::clone(job))) @@ -213,7 +218,7 @@ impl JobStore { /// Kill a running job by signalling its whole process group. Returns `false` /// when the id is unknown or the job already finished — nothing to signal. - pub async fn kill(&self, id: &str) -> bool { + pub async fn kill(&self, id: &JobId) -> bool { let Some(job) = self.jobs.lock().await.get(id).cloned() else { return false; }; @@ -354,7 +359,11 @@ mod tests { #[tokio::test] async fn kill_unknown_id_returns_false() { - assert!(!store(Duration::from_secs(5)).kill("nope").await); + assert!( + !store(Duration::from_secs(5)) + .kill(&JobId::from("nope")) + .await + ); } #[tokio::test] @@ -366,7 +375,10 @@ mod tests { .await .unwrap(); assert!(matches!(r, RunResult::Inline { .. })); - assert!(!store.kill("j1").await); + // Inline doesn't surface the id; fetch the finished job's id from the + // listing. Killing an already-exited job must return false. + let id = store.list().await.first().expect("job tracked").id.clone(); + assert!(!store.kill(&id).await); } #[cfg(unix)] @@ -462,13 +474,14 @@ mod tests { done: rx, started: tokio::time::Instant::now() - Duration::from_secs(1), }); - store.jobs.lock().await.insert("jfake".into(), job); + let id = JobId::from("jfake"); + store.jobs.lock().await.insert(id.clone(), job); // The backdated job is stale; kill fails => must not be evicted. reaper::reap_once(&store.jobs, Duration::ZERO).await; assert!( - store.poll("jfake", 0, None).await.is_some(), + store.poll(&id, 0, None).await.is_some(), "running job whose kill failed must stay tracked" ); assert!(log_path.exists(), "its log must not be deleted"); @@ -529,15 +542,13 @@ mod tests { #[tokio::test] async fn poll_paginates() { let store = store(Duration::from_secs(5)); - let r = store + store .run("seq 1 10".into(), None, None, false) .await .unwrap(); - // seq finishes inline; re-poll the job id to exercise pagination. - let id = match r { - RunResult::Inline { .. } => "j1".to_string(), - RunResult::Backgrounded { id } => id, - }; + // seq finishes inline; fetch its id from the listing to re-poll and + // exercise pagination. + let id = store.list().await.first().expect("job tracked").id.clone(); let (_s, page) = store.poll(&id, 0, Some(3)).await.unwrap(); assert_eq!(page.lines.len(), 3); assert_eq!(page.next_cursor, 3); diff --git a/src/jobs/reaper.rs b/src/jobs/reaper.rs index 62bf51b..f09f5ae 100644 --- a/src/jobs/reaper.rs +++ b/src/jobs/reaper.rs @@ -7,7 +7,7 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::{Mutex, watch}; -use super::{Job, JobState, ProcessGroupId}; +use super::{Job, JobId, JobState, ProcessGroupId}; /// Jobs (and their logs) older than this are reaped hourly. const RETENTION: Duration = Duration::from_secs(24 * 3600); @@ -68,7 +68,7 @@ async fn exited_within(mut done: watch::Receiver, grace: Duration) -> bool /// Hourly: drop jobs (and their log files) older than `RETENTION` so history /// doesn't grow without bound. ponytail: time-based only; a busy box could still /// hold ≤24h of jobs in memory — add a count cap if that ever bites. -pub(super) fn spawn_reaper(jobs: Arc>>>) { +pub(super) fn spawn_reaper(jobs: Arc>>>) { tokio::spawn(async move { let mut tick = tokio::time::interval(Duration::from_secs(3600)); loop { @@ -80,10 +80,10 @@ pub(super) fn spawn_reaper(jobs: Arc>>>) { /// One reaping pass: evict every job older than `retention`. A still-`Running` /// job is killed first, so eviction never orphans its process group. -pub(super) async fn reap_once(jobs: &Mutex>>, retention: Duration) { +pub(super) async fn reap_once(jobs: &Mutex>>, retention: Duration) { let now = tokio::time::Instant::now(); let map = jobs.lock().await; - let stale: Vec<(String, Arc)> = map + let stale: Vec<(JobId, Arc)> = map .iter() .filter(|(_, j)| now.duration_since(j.started) > retention) .map(|(id, j)| (id.clone(), j.clone())) @@ -93,7 +93,7 @@ pub(super) async fn reap_once(jobs: &Mutex>>, retention // Kill first so a still-running group is never orphaned by eviction. Only // evict jobs that finished or whose group we actually signalled; a running // job whose kill failed stays tracked (pollable/killable) for a later pass. - let mut removable: Vec<(String, Arc)> = Vec::new(); + let mut removable: Vec<(JobId, Arc)> = Vec::new(); for (id, job) in &stale { if kill_job(job).await || !matches!(*job.state.lock().await, JobState::Running) { removable.push((id.clone(), job.clone())); diff --git a/src/tools/mod.rs b/src/tools/mod.rs index eca9ba8..4ae93f0 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -11,7 +11,7 @@ use rmcp::{ // `RequestId` above is rmcp's extractor for the per-call JSON-RPC request id. use tracing::Instrument; -use crate::jobs::{JobState, JobStore, Page, RunResult}; +use crate::jobs::{JobId, JobState, JobStore, Page, RunResult}; mod files; @@ -151,6 +151,7 @@ impl Tools { let Some(id) = args.id else { return Ok(err("poll requires `id`")); }; + let id = JobId::from(id); match self .jobs .poll(&id, args.cursor.unwrap_or(0), args.limit) @@ -168,6 +169,7 @@ impl Tools { let Some(id) = args.id else { return Ok(err("kill requires `id`")); }; + let id = JobId::from(id); if self.jobs.kill(&id).await { Ok(ok(format!("killed {id}"))) } else { From a1a9944728714dc9db17b7184cf61ef954bfe53f Mon Sep 17 00:00:00 2001 From: sebi Date: Fri, 26 Jun 2026 19:52:27 -0500 Subject: [PATCH 3/6] refactor(jobs): extract Page + read_page into jobs/log.rs Move Page struct, DEFAULT_PAGE const, read_page fn, and their test into a dedicated src/jobs/log.rs; re-export Page from mod.rs. Co-Authored-By: Claude --- src/jobs/log.rs | 59 +++++++++++++++++++++++++++++++++++++++++++++++++ src/jobs/mod.rs | 56 +++------------------------------------------- 2 files changed, 62 insertions(+), 53 deletions(-) create mode 100644 src/jobs/log.rs diff --git a/src/jobs/log.rs b/src/jobs/log.rs new file mode 100644 index 0000000..4500512 --- /dev/null +++ b/src/jobs/log.rs @@ -0,0 +1,59 @@ +//! Paginated log reading for per-job output files. + +/// One page of log lines plus a cursor to fetch the next page. +#[derive(Debug, serde::Serialize)] +pub struct Page { + pub lines: Vec, + pub next_cursor: usize, + pub total_lines: usize, + pub has_more: bool, +} + +pub const DEFAULT_PAGE: usize = 200; + +/// Read lines `[cursor, cursor+limit)` from a log file. Re-reads the whole file +/// each call — fine for typical logs; seek by byte offset if they get huge. +pub async fn read_page(path: &std::path::Path, cursor: usize, limit: usize) -> Page { + // Binary-safe: a command that writes non-UTF-8 bytes (e.g. compiled output, + // escape sequences) must not produce a silently-empty log page. + let bytes = tokio::fs::read(path).await.unwrap_or_default(); + let content = String::from_utf8_lossy(&bytes); + let all: Vec<&str> = content.lines().collect(); + let total = all.len(); + let end = (cursor + limit).min(total); + let lines = all + .get(cursor..end) + .unwrap_or(&[]) + .iter() + .map(|s| s.to_string()) + .collect(); + Page { + lines, + next_cursor: end, + total_lines: total, + has_more: end < total, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn read_page_handles_binary_log_without_empty_output() { + // Simulate a command that writes non-UTF-8 bytes to its log. + let dir = tempfile::tempdir().unwrap(); + let log = dir.path().join("bin.log"); + // Valid UTF-8 prefix + invalid bytes + valid suffix. + tokio::fs::write(&log, b"line1\nline2\xff\xfeline3\n") + .await + .unwrap(); + let page = read_page(&log, 0, 100).await; + // Must return lines, not an empty page. + assert!( + !page.lines.is_empty(), + "binary log must not produce empty page" + ); + assert!(page.lines[0].contains("line1")); + } +} diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index ff8ebc2..1c35efc 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -12,9 +12,12 @@ use std::{ use tokio::sync::{Mutex, watch}; mod id; +mod log; mod reaper; pub use id::JobId; +pub use log::Page; +use log::{DEFAULT_PAGE, read_page}; use reaper::{kill_job, spawn_reaper}; #[derive(Debug, Clone, serde::Serialize)] @@ -51,15 +54,6 @@ pub enum RunResult { Backgrounded { id: JobId }, } -/// One page of log lines plus a cursor to fetch the next page. -#[derive(Debug, serde::Serialize)] -pub struct Page { - pub lines: Vec, - pub next_cursor: usize, - pub total_lines: usize, - pub has_more: bool, -} - #[derive(Debug, serde::Serialize)] pub struct JobSummary { pub id: JobId, @@ -75,8 +69,6 @@ pub struct JobStore { jobs: Arc>>>, } -const DEFAULT_PAGE: usize = 200; - impl JobStore { pub fn new(dir: PathBuf, inline_timeout: Duration) -> std::io::Result { std::fs::create_dir_all(&dir)?; @@ -226,30 +218,6 @@ impl JobStore { } } -/// Read lines `[cursor, cursor+limit)` from a log file. Re-reads the whole file -/// each call — fine for typical logs; seek by byte offset if they get huge. -async fn read_page(path: &std::path::Path, cursor: usize, limit: usize) -> Page { - // Binary-safe: a command that writes non-UTF-8 bytes (e.g. compiled output, - // escape sequences) must not produce a silently-empty log page. - let bytes = tokio::fs::read(path).await.unwrap_or_default(); - let content = String::from_utf8_lossy(&bytes); - let all: Vec<&str> = content.lines().collect(); - let total = all.len(); - let end = (cursor + limit).min(total); - let lines = all - .get(cursor..end) - .unwrap_or(&[]) - .iter() - .map(|s| s.to_string()) - .collect(); - Page { - lines, - next_cursor: end, - total_lines: total, - has_more: end < total, - } -} - #[cfg(test)] mod tests { use super::*; @@ -555,22 +523,4 @@ mod tests { assert!(page.has_more); assert_eq!(page.total_lines, 10); } - - #[tokio::test] - async fn read_page_handles_binary_log_without_empty_output() { - // Simulate a command that writes non-UTF-8 bytes to its log. - let dir = tempfile::tempdir().unwrap(); - let log = dir.path().join("bin.log"); - // Valid UTF-8 prefix + invalid bytes + valid suffix. - tokio::fs::write(&log, b"line1\nline2\xff\xfeline3\n") - .await - .unwrap(); - let page = read_page(&log, 0, 100).await; - // Must return lines, not an empty page. - assert!( - !page.lines.is_empty(), - "binary log must not produce empty page" - ); - assert!(page.lines[0].contains("line1")); - } } From 7444d55f4fa59405a77c9a8b343a13594fd58c15 Mon Sep 17 00:00:00 2001 From: sebi Date: Fri, 26 Jun 2026 19:53:42 -0500 Subject: [PATCH 4/6] docs: replace job id examples j1..j7 with human-readable format (cargo-build-23:30) Updated docs to reflect the new human-readable job id format (command slug + local hour): - docs/usage.md: replaced j1,j2,...,j7 examples with cargo-build-23:30 - docs/architecture.md: updated module map to show jobs/id.rs, jobs/log.rs, jobs/reaper.rs separately (was a single jobs.rs); noted id format gives context without UUIDs - CLAUDE.md: added module map entries for jobs/id.rs and jobs/log.rs Co-Authored-By: Claude Haiku 4.5 --- CLAUDE.md | 2 ++ docs/architecture.md | 7 +++++-- docs/usage.md | 20 ++++++++++---------- 3 files changed, 17 insertions(+), 12 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 9d5d205..720b604 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -54,6 +54,8 @@ Keep this accurate — it's the navigation aid. | `src/auth.rs` | HTTP Basic auth middleware | | `src/oauth/` | minimal OAuth 2.1 server: discovery metadata, dynamic client registration, authorize + token with PKCE, bearer validation | | `src/jobs/mod.rs` | job engine: run a command, return inline if fast (<2s) else a job id (or immediately when `bg`); output streams to a per-job log file, polled paginated | +| `src/jobs/id.rs` | JobId newtype: human-readable ids from command slugs + local hour (e.g., `cargo-build-23:30`) | +| `src/jobs/log.rs` | job log pagination: read per-job log files by page (cursor + limit) | | `src/jobs/reaper.rs` | hourly reaper drops jobs >24h old (killing any still-`Running` group first); process-group kill helpers (TERM→KILL escalation), shared with `job(action="kill")` | | `src/tools/mod.rs` | MCP tool surface (`#[tool_router]`/`#[tool]` from rmcp): 3 tools (`bash`/`job`/`file`) dispatching on `action`. Thin adapters over jobs + files | | `src/tools/files.rs` | file operations (`tokio::fs`; `ls`/`find`/`grep` shelled out) | diff --git a/docs/architecture.md b/docs/architecture.md index 8cee573..497a54e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -22,7 +22,10 @@ TLS is deliberately **not** in the binary (keeps deps minimal). A reverse proxy | `config.rs` | Runtime config from env + file. Fails fast at boot if credentials are missing. | | `auth.rs` | The auth middleware — **bearer-only** guard on `/mcp`; rejects everything that is not a valid OAuth 2.1 access token. | | `oauth` | A minimal **OAuth 2.1 authorization server** (per the MCP authorization spec) so GUI clients can log in. | -| `jobs.rs` | The job engine: run a command, return inline-or-background (incl. immediate `bg`), paginated logs, and an hourly reaper that drops jobs >24h old. | +| `jobs/mod.rs` | The job engine: run a command, return inline-or-background (incl. immediate `bg`), paginated logs, and an hourly reaper that drops jobs >24h old. | +| `jobs/id.rs` | JobId newtype: human-readable ids from command names + local hour (e.g., `cargo-build-23:30`). | +| `jobs/log.rs` | Job log pagination: read log files by page (cursor + limit). | +| `jobs/reaper.rs` | Hourly reaper evicts jobs >24h old; process-group kill helpers (TERM→KILL escalation). | | `tools/mod.rs` | The MCP tool surface — **three tools** (`bash`/`job`/`file`) dispatching on an `action` param; thin adapters over `jobs` and `files`. | | `tools/files.rs` | File operations (read/write/append/delete/list/grep/move). | @@ -56,7 +59,7 @@ When a command starts (`JobStore::run`): - **Finished in time** → `RunResult::Inline` — status + first page of the log, returned now. - **Still running (or `bg`)** → `RunResult::Backgrounded { id }` — the agent gets a job id to poll. -Jobs live in an in-memory map keyed by id (`j1`, `j2`, …); the log files persist on disk under the job dir. An **hourly reaper** drops any job whose age exceeds 24h — evicting it from the map and deleting its log file — so the map and the job dir stay bounded without manual cleanup. +Jobs live in an in-memory map keyed by human-readable id (e.g., `cargo-build-23:30`); the log files persist on disk under the job dir. An **hourly reaper** drops any job whose age exceeds 24h — evicting it from the map and deleting its log file — so the map and the job dir stay bounded without manual cleanup. Job ids are derived from the command name (slugified) + local hour, giving context without UUIDs. ``` JobState = Running | Exited { code } | Failed { error } diff --git a/docs/usage.md b/docs/usage.md index 0efb1b9..6e151ce 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -9,24 +9,24 @@ Three tools — `bash`, `job`, `file` — each heavily parametrized. Their param | Outcome | What you get back | |---|---| | Finishes within the window | Status + the first page of output, **inline** | -| Still running after the window | A **job id** (`j1`, `j2`, …) — monitor it with `job(action="poll")` | +| Still running after the window | A **job id** (e.g., `cargo-build-23:30`) — monitor it with `job(action="poll")` | This is what makes a 20-minute build safe: it backgrounds automatically, and you stream the log a page at a time instead of dumping it all into context. ``` bash("cargo build --release") -→ "job j7 still running after the inline window. Poll it with job(action=\"poll\", id=\"j7\")." +→ "job cargo-build-23:30 still running after the inline window. Poll it with job(action=\"poll\", id=\"cargo-build-23:30\")." -job(action="poll", id="j7") +job(action="poll", id="cargo-build-23:30") → {"status":"running"} Compiling mcp-ssh v0.1.0 ...200 lines... [lines ..200 of 540; next_cursor=200] -job(action="poll", id="j7", cursor=200) +job(action="poll", id="cargo-build-23:30", cursor=200) → ...next page... -job(action="poll", id="j7", cursor=540) +job(action="poll", id="cargo-build-23:30", cursor=540) → {"status":"exited","code":0} Finished `release` profile [optimized] target(s) ``` @@ -80,11 +80,11 @@ Manage jobs created by `bash`. `action` is one of `poll`, `list`, `kill`. `action="poll"` returns the job's status plus **one page** of merged stdout+stderr, with `next_cursor`/`has_more` so you walk long logs forward without flooding context. `action="list"` returns all jobs with their status (`running` / `exited` / `failed`). `action="kill"` kills a running job. ``` -job(action="poll", id="j7") # first page -job(action="poll", id="j7", cursor=200) # next page -job(action="poll", id="j7", cursor=400, limit=500) -job(action="list") # all jobs + status -job(action="kill", id="j7") # kill a running job +job(action="poll", id="cargo-build-23:30") # first page +job(action="poll", id="cargo-build-23:30", cursor=200) # next page +job(action="poll", id="cargo-build-23:30", cursor=400, limit=500) +job(action="list") # all jobs + status +job(action="kill", id="cargo-build-23:30") # kill a running job ``` ## 📁 Files From ff2b58ab5e66b30ba42a1e99f896fce6b7b7de45 Mon Sep 17 00:00:00 2001 From: sebi Date: Fri, 26 Jun 2026 20:16:47 -0500 Subject: [PATCH 5/6] =?UTF-8?q?fix(jobs):=20address=20PR=20#9=20review=20?= =?UTF-8?q?=E2=80=94=20secret-safe=20ids,=20typed=20log=20errors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - security: derive JobId from a neutral "job" label, never command text; a secret in leading tokens (mysql -psecret, PGPASSWORD=…) no longer leaks into ids, reaper logs, or log filenames - jobs/log: read_page returns Result (thiserror) so a missing/unreadable log surfaces as an error instead of an empty page; poll/run/tools wired through, run converts at its io::Result boundary - jobs/log: saturating_add for the page end to avoid cursor+limit overflow - docs: sync architecture diagram to jobs/mod.rs + tools/files.rs; fix usage poll-cursor progression (200→400→exit); tag fenced block (MD040); precise id description (local time, neutral prefix) in CLAUDE/architecture Co-Authored-By: Claude --- CLAUDE.md | 2 +- docs/architecture.md | 6 ++--- docs/usage.md | 23 ++++++++++---------- src/jobs/id.rs | 39 ++++++++++++++++++++------------- src/jobs/log.rs | 52 +++++++++++++++++++++++++++++++++++++------- src/jobs/mod.rs | 49 ++++++++++++++++++++++++----------------- src/tools/mod.rs | 5 +++-- 7 files changed, 116 insertions(+), 60 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 720b604..80fd308 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -54,7 +54,7 @@ Keep this accurate — it's the navigation aid. | `src/auth.rs` | HTTP Basic auth middleware | | `src/oauth/` | minimal OAuth 2.1 server: discovery metadata, dynamic client registration, authorize + token with PKCE, bearer validation | | `src/jobs/mod.rs` | job engine: run a command, return inline if fast (<2s) else a job id (or immediately when `bg`); output streams to a per-job log file, polled paginated | -| `src/jobs/id.rs` | JobId newtype: human-readable ids from command slugs + local hour (e.g., `cargo-build-23:30`) | +| `src/jobs/id.rs` | JobId newtype: human-readable ids — neutral `job` prefix + local `HH:MM` (e.g., `job-23:30`); free of command text so secrets can't leak into an id, log line, or filename | | `src/jobs/log.rs` | job log pagination: read per-job log files by page (cursor + limit) | | `src/jobs/reaper.rs` | hourly reaper drops jobs >24h old (killing any still-`Running` group first); process-group kill helpers (TERM→KILL escalation), shared with `job(action="kill")` | | `src/tools/mod.rs` | MCP tool surface (`#[tool_router]`/`#[tool]` from rmcp): 3 tools (`bash`/`job`/`file`) dispatching on `action`. Thin adapters over jobs + files | diff --git a/docs/architecture.md b/docs/architecture.md index 497a54e..f4a454b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -23,7 +23,7 @@ TLS is deliberately **not** in the binary (keeps deps minimal). A reverse proxy | `auth.rs` | The auth middleware — **bearer-only** guard on `/mcp`; rejects everything that is not a valid OAuth 2.1 access token. | | `oauth` | A minimal **OAuth 2.1 authorization server** (per the MCP authorization spec) so GUI clients can log in. | | `jobs/mod.rs` | The job engine: run a command, return inline-or-background (incl. immediate `bg`), paginated logs, and an hourly reaper that drops jobs >24h old. | -| `jobs/id.rs` | JobId newtype: human-readable ids from command names + local hour (e.g., `cargo-build-23:30`). | +| `jobs/id.rs` | JobId newtype: human-readable ids — neutral `job` prefix + local `HH:MM` (e.g., `job-23:30`); free of command text so secrets can't leak. | | `jobs/log.rs` | Job log pagination: read log files by page (cursor + limit). | | `jobs/reaper.rs` | Hourly reaper evicts jobs >24h old; process-group kill helpers (TERM→KILL escalation). | | `tools/mod.rs` | The MCP tool surface — **three tools** (`bash`/`job`/`file`) dispatching on an `action` param; thin adapters over `jobs` and `files`. | @@ -41,7 +41,7 @@ client ──HTTPS──▶ reverse proxy ──HTTP──▶ axum (127.0.0.1:13 Tools (tool_router) bash · job · file (action) ╱ ╲ - jobs.rs files.rs + jobs/mod.rs tools/files.rs (bash, job) (file ops) ``` @@ -59,7 +59,7 @@ When a command starts (`JobStore::run`): - **Finished in time** → `RunResult::Inline` — status + first page of the log, returned now. - **Still running (or `bg`)** → `RunResult::Backgrounded { id }` — the agent gets a job id to poll. -Jobs live in an in-memory map keyed by human-readable id (e.g., `cargo-build-23:30`); the log files persist on disk under the job dir. An **hourly reaper** drops any job whose age exceeds 24h — evicting it from the map and deleting its log file — so the map and the job dir stay bounded without manual cleanup. Job ids are derived from the command name (slugified) + local hour, giving context without UUIDs. +Jobs live in an in-memory map keyed by human-readable id (e.g., `job-23:30`); the log files persist on disk under the job dir. An **hourly reaper** drops any job whose age exceeds 24h — evicting it from the map and deleting its log file — so the map and the job dir stay bounded without manual cleanup. Job ids are a neutral `job` prefix + local `HH:MM` — deliberately free of command text so a secret on the command line can't leak into an id, log line, or filename; the full command stays available via `job(action="list")`. ``` JobState = Running | Exited { code } | Failed { error } diff --git a/docs/usage.md b/docs/usage.md index 6e151ce..aaa9564 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -9,24 +9,25 @@ Three tools — `bash`, `job`, `file` — each heavily parametrized. Their param | Outcome | What you get back | |---|---| | Finishes within the window | Status + the first page of output, **inline** | -| Still running after the window | A **job id** (e.g., `cargo-build-23:30`) — monitor it with `job(action="poll")` | +| Still running after the window | A **job id** (e.g., `job-23:30`) — monitor it with `job(action="poll")` | This is what makes a 20-minute build safe: it backgrounds automatically, and you stream the log a page at a time instead of dumping it all into context. -``` +```text bash("cargo build --release") -→ "job cargo-build-23:30 still running after the inline window. Poll it with job(action=\"poll\", id=\"cargo-build-23:30\")." +→ "job job-23:30 still running after the inline window. Poll it with job(action=\"poll\", id=\"job-23:30\")." -job(action="poll", id="cargo-build-23:30") +job(action="poll", id="job-23:30") → {"status":"running"} Compiling mcp-ssh v0.1.0 ...200 lines... [lines ..200 of 540; next_cursor=200] -job(action="poll", id="cargo-build-23:30", cursor=200) -→ ...next page... +job(action="poll", id="job-23:30", cursor=200) +→ ...next 200 lines... + [lines ..400 of 540; next_cursor=400] -job(action="poll", id="cargo-build-23:30", cursor=540) +job(action="poll", id="job-23:30", cursor=400) → {"status":"exited","code":0} Finished `release` profile [optimized] target(s) ``` @@ -80,11 +81,11 @@ Manage jobs created by `bash`. `action` is one of `poll`, `list`, `kill`. `action="poll"` returns the job's status plus **one page** of merged stdout+stderr, with `next_cursor`/`has_more` so you walk long logs forward without flooding context. `action="list"` returns all jobs with their status (`running` / `exited` / `failed`). `action="kill"` kills a running job. ``` -job(action="poll", id="cargo-build-23:30") # first page -job(action="poll", id="cargo-build-23:30", cursor=200) # next page -job(action="poll", id="cargo-build-23:30", cursor=400, limit=500) +job(action="poll", id="job-23:30") # first page +job(action="poll", id="job-23:30", cursor=200) # next page +job(action="poll", id="job-23:30", cursor=400, limit=500) job(action="list") # all jobs + status -job(action="kill", id="cargo-build-23:30") # kill a running job +job(action="kill", id="job-23:30") # kill a running job ``` ## 📁 Files diff --git a/src/jobs/id.rs b/src/jobs/id.rs index 298d427..90384dc 100644 --- a/src/jobs/id.rs +++ b/src/jobs/id.rs @@ -1,7 +1,9 @@ -//! Human-readable job ids: a slug from the command plus the local `HH:MM` it -//! started, e.g. `cargo-build-23:30` — far easier to refer to in conversation -//! than an opaque counter. A monotonic sequence suffix disambiguates the rare -//! case of two identical-slug commands starting within the same minute. +//! Human-readable job ids: a neutral `job` label plus the local `HH:MM` it +//! started, e.g. `job-23:30` — easier to refer to in conversation than an opaque +//! counter. The id is deliberately free of command text: a command can carry a +//! secret in its leading tokens, and the id surfaces in replies, `job(list)`, +//! reaper logs, and the log filename, so slugging the command would leak it. +//! A monotonic sequence suffix disambiguates jobs starting within the same minute. use std::borrow::Borrow; use std::fmt; @@ -20,12 +22,14 @@ const MAX_SLUG: usize = 24; pub struct JobId(String); impl JobId { - /// Build an id from `cmd` and the current local time. `exists` reports - /// whether a candidate is already taken; only on a clash do we append a - /// `-` drawn from `seq`, so clean ids stay clean. The predicate keeps - /// this decoupled from however the caller tracks live ids. - pub fn generate(cmd: &str, seq: &AtomicU64, exists: impl Fn(&str) -> bool) -> Self { - let base = format!("{}-{}", slug(cmd), Local::now().format("%H:%M")); + /// Build an id from a neutral `label` and the current local time. Callers + /// pass a fixed label (e.g. `"job"`), never raw command text, so a secret on + /// the command line can't leak into the id. `exists` reports whether a + /// candidate is already taken; only on a clash do we append a `-` drawn + /// from `seq`, so clean ids stay clean. The predicate keeps this decoupled + /// from however the caller tracks live ids. + pub fn generate(label: &str, seq: &AtomicU64, exists: impl Fn(&str) -> bool) -> Self { + let base = format!("{}-{}", slug(label), Local::now().format("%H:%M")); if !exists(&base) { return Self(base); } @@ -70,11 +74,16 @@ impl From<&str> for JobId { } } -/// Turn the first few command tokens into a lowercase `[a-z0-9-]` slug: collapse -/// runs of non-alphanumerics to single dashes, trim them off both ends, and cap -/// the length. Falls back to `job` when nothing alphanumeric survives. -fn slug(cmd: &str) -> String { - let head = cmd.split_whitespace().take(3).collect::>().join(" "); +/// Sanitize a label into a lowercase `[a-z0-9-]` slug so any caller-supplied +/// string yields a well-formed id: take the first few tokens, collapse runs of +/// non-alphanumerics to single dashes, trim them off both ends, and cap the +/// length. Falls back to `job` when nothing alphanumeric survives. +fn slug(label: &str) -> String { + let head = label + .split_whitespace() + .take(3) + .collect::>() + .join(" "); let mut out = String::with_capacity(head.len()); let mut prev_dash = false; for ch in head.chars() { diff --git a/src/jobs/log.rs b/src/jobs/log.rs index 4500512..1bce429 100644 --- a/src/jobs/log.rs +++ b/src/jobs/log.rs @@ -11,28 +11,56 @@ pub struct Page { pub const DEFAULT_PAGE: usize = 200; +/// A job log could not be read. Kept distinct from a legitimately empty log so +/// `job(action="poll")` can surface "log unavailable" instead of silently +/// returning an empty page (a missing or unreadable log must not look identical +/// to one that simply has no output yet). +#[derive(Debug, thiserror::Error)] +pub enum JobLogError { + #[error("reading job log: {0}")] + Read(#[from] std::io::Error), +} + +/// Convert back to `io::Error` at the `JobStore::run` boundary, which already +/// reports in `std::io::Result`. The typed error matters for `poll`, which +/// surfaces it to the caller rather than collapsing it into an empty page. +impl From for std::io::Error { + fn from(e: JobLogError) -> Self { + match e { + JobLogError::Read(io) => io, + } + } +} + /// Read lines `[cursor, cursor+limit)` from a log file. Re-reads the whole file /// each call — fine for typical logs; seek by byte offset if they get huge. -pub async fn read_page(path: &std::path::Path, cursor: usize, limit: usize) -> Page { - // Binary-safe: a command that writes non-UTF-8 bytes (e.g. compiled output, - // escape sequences) must not produce a silently-empty log page. - let bytes = tokio::fs::read(path).await.unwrap_or_default(); +pub async fn read_page( + path: &std::path::Path, + cursor: usize, + limit: usize, +) -> Result { + // A failed read (missing/unreadable log) propagates as a typed error. Only + // non-UTF-8 *content* is rendered lossily: a command that writes raw bytes + // (compiled output, escape sequences) must not produce a silently-empty page. + let bytes = tokio::fs::read(path).await?; let content = String::from_utf8_lossy(&bytes); let all: Vec<&str> = content.lines().collect(); let total = all.len(); - let end = (cursor + limit).min(total); + // cursor + limit come from tool input; saturate so an oversized request can't + // overflow into a bogus window. + let end = cursor.saturating_add(limit).min(total); let lines = all .get(cursor..end) .unwrap_or(&[]) .iter() .map(|s| s.to_string()) .collect(); - Page { + Ok(Page { lines, next_cursor: end, total_lines: total, has_more: end < total, - } + }) } #[cfg(test)] @@ -48,7 +76,7 @@ mod tests { tokio::fs::write(&log, b"line1\nline2\xff\xfeline3\n") .await .unwrap(); - let page = read_page(&log, 0, 100).await; + let page = read_page(&log, 0, 100).await.unwrap(); // Must return lines, not an empty page. assert!( !page.lines.is_empty(), @@ -56,4 +84,12 @@ mod tests { ); assert!(page.lines[0].contains("line1")); } + + #[tokio::test] + async fn read_page_errors_on_missing_log() { + // A missing log is an error, not a silently-empty page. + let dir = tempfile::tempdir().unwrap(); + let missing = dir.path().join("nope.log"); + assert!(read_page(&missing, 0, 100).await.is_err()); + } } diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index 1c35efc..d1e3e34 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -16,8 +16,8 @@ mod log; mod reaper; pub use id::JobId; -pub use log::Page; use log::{DEFAULT_PAGE, read_page}; +pub use log::{JobLogError, Page}; use reaper::{kill_job, spawn_reaper}; #[derive(Debug, Clone, serde::Serialize)] @@ -92,12 +92,17 @@ impl JobStore { background: bool, ) -> std::io::Result { // Hold the jobs lock across id generation and insertion so the id is - // reserved atomically: two identical commands launched in the same minute - // can't both see their `slug-HH:MM` as free and clobber each other's - // entry. Nothing awaits while the guard is held (file create + spawn are - // synchronous), so the critical section stays short. + // reserved atomically: two jobs launched in the same minute can't both + // see their `job-HH:MM` as free and clobber each other's entry. Nothing + // awaits while the guard is held (file create + spawn are synchronous), + // so the critical section stays short. + // + // The id is built from a fixed `"job"` label, never `cmd`: a command can + // carry a secret in its leading tokens (`mysql -psecret`, `PGPASSWORD=…`), + // and the id ends up in `bash`'s reply, `job(list)`, reaper logs, and the + // log filename — so deriving it from `cmd` would leak that secret. let mut jobs = self.jobs.lock().await; - let id = JobId::generate(&cmd, &self.seq, |candidate| jobs.contains_key(candidate)); + let id = JobId::generate("job", &self.seq, |candidate| jobs.contains_key(candidate)); let log_path = self.dir.join(format!("{id}.log")); // ponytail: stdout+stderr merged into one log (terminal-style). Split into @@ -170,22 +175,26 @@ impl JobStore { JobState::Running => Ok(RunResult::Backgrounded { id }), finished => Ok(RunResult::Inline { state: finished, - page: read_page(&log_path, 0, DEFAULT_PAGE).await, + page: read_page(&log_path, 0, DEFAULT_PAGE).await?, }), } } - /// Status + one page of a job's log. + /// Status + one page of a job's log. `Ok(None)` means the id is unknown; + /// `Err` means the log existed but could not be read — surfaced to the caller + /// rather than collapsed into an empty page. pub async fn poll( &self, id: &JobId, cursor: usize, limit: Option, - ) -> Option<(JobState, Page)> { - let job = self.jobs.lock().await.get(id).cloned()?; + ) -> Result, JobLogError> { + let Some(job) = self.jobs.lock().await.get(id).cloned() else { + return Ok(None); + }; let state = job.state.lock().await.clone(); - let page = read_page(&job.log_path, cursor, limit.unwrap_or(DEFAULT_PAGE)).await; - Some((state, page)) + let page = read_page(&job.log_path, cursor, limit.unwrap_or(DEFAULT_PAGE)).await?; + Ok(Some((state, page))) } pub async fn list(&self) -> Vec { @@ -264,7 +273,7 @@ mod tests { RunResult::Inline { .. } => panic!("slow command should background"), }; for _ in 0..50 { - let (state, page) = store.poll(&id, 0, None).await.unwrap(); + let (state, page) = store.poll(&id, 0, None).await.unwrap().unwrap(); if matches!(state, JobState::Exited { .. }) { assert!(page.lines.iter().any(|l| l.contains("done"))); return; @@ -303,7 +312,7 @@ mod tests { // Pull the descendant's pid out of the log. let mut child_pid = None; for _ in 0..50 { - let (_s, page) = store.poll(&id, 0, None).await.unwrap(); + let (_s, page) = store.poll(&id, 0, None).await.unwrap().unwrap(); if let Some(line) = page.lines.iter().find_map(|l| l.strip_prefix("pid:")) { child_pid = Some(line.trim().to_string()); break; @@ -373,7 +382,7 @@ mod tests { // TERM is ignored; the post-grace KILL must still bring it down. for _ in 0..100 { - let (state, _) = store.poll(&id, 0, None).await.unwrap(); + let (state, _) = store.poll(&id, 0, None).await.unwrap().unwrap(); if matches!(state, JobState::Exited { .. }) { return; } @@ -397,7 +406,7 @@ mod tests { let mut child_pid = None; for _ in 0..50 { - let (_s, page) = store.poll(&id, 0, None).await.unwrap(); + let (_s, page) = store.poll(&id, 0, None).await.unwrap().unwrap(); if let Some(line) = page.lines.iter().find_map(|l| l.strip_prefix("pid:")) { child_pid = Some(line.trim().to_string()); break; @@ -412,7 +421,7 @@ mod tests { // Evicted from the map (poll can't find it)... assert!( - store.poll(&id, 0, None).await.is_none(), + store.poll(&id, 0, None).await.unwrap().is_none(), "job should be evicted" ); // ...and its process group reaped, not orphaned. @@ -449,7 +458,7 @@ mod tests { reaper::reap_once(&store.jobs, Duration::ZERO).await; assert!( - store.poll(&id, 0, None).await.is_some(), + store.poll(&id, 0, None).await.unwrap().is_some(), "running job whose kill failed must stay tracked" ); assert!(log_path.exists(), "its log must not be deleted"); @@ -476,7 +485,7 @@ mod tests { // After kill the state must transition away from Running. for _ in 0..50 { - let (state, _) = store.poll(&id, 0, None).await.unwrap(); + let (state, _) = store.poll(&id, 0, None).await.unwrap().unwrap(); if !matches!(state, JobState::Running) { return; } @@ -517,7 +526,7 @@ mod tests { // seq finishes inline; fetch its id from the listing to re-poll and // exercise pagination. let id = store.list().await.first().expect("job tracked").id.clone(); - let (_s, page) = store.poll(&id, 0, Some(3)).await.unwrap(); + let (_s, page) = store.poll(&id, 0, Some(3)).await.unwrap().unwrap(); assert_eq!(page.lines.len(), 3); assert_eq!(page.next_cursor, 3); assert!(page.has_more); diff --git a/src/tools/mod.rs b/src/tools/mod.rs index 4ae93f0..e8d63a9 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -157,8 +157,9 @@ impl Tools { .poll(&id, args.cursor.unwrap_or(0), args.limit) .await { - Some((state, page)) => Ok(ok(render(&state, &page))), - None => Ok(err(format!("no such job: {id}"))), + Ok(Some((state, page))) => Ok(ok(render(&state, &page))), + Ok(None) => Ok(err(format!("no such job: {id}"))), + Err(e) => Ok(err(e.to_string())), } } JobAction::List => { From 7e1397bf470e1de35110f05bf35bf48c6d4de74a Mon Sep 17 00:00:00 2001 From: sebi Date: Fri, 26 Jun 2026 20:30:12 -0500 Subject: [PATCH 6/6] fix(jobs): close command-text leak surfaces, harden pagination MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Omit `cmd` from JobSummary/Job — job(list) no longer returns raw commands (could carry PGPASSWORD=…/bearer tokens) in MCP responses - Hard-code the `job` prefix in JobId::generate; drop the label param and slug() so command text can never reach an id - Clamp read_page limit to >=1 so limit=0 can't return a non-advancing page (client paging on next_cursor would spin) - Map reaper eviction races in poll (NotFound + id gone) to Ok(None) instead of surfacing a filesystem error - Update architecture.md: list exposes only id + status Co-Authored-By: Claude --- docs/architecture.md | 2 +- src/jobs/id.rs | 125 ++++++++++--------------------------------- src/jobs/log.rs | 17 ++++++ src/jobs/mod.rs | 64 ++++++++++++++++------ 4 files changed, 94 insertions(+), 114 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index f4a454b..9299ae3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -59,7 +59,7 @@ When a command starts (`JobStore::run`): - **Finished in time** → `RunResult::Inline` — status + first page of the log, returned now. - **Still running (or `bg`)** → `RunResult::Backgrounded { id }` — the agent gets a job id to poll. -Jobs live in an in-memory map keyed by human-readable id (e.g., `job-23:30`); the log files persist on disk under the job dir. An **hourly reaper** drops any job whose age exceeds 24h — evicting it from the map and deleting its log file — so the map and the job dir stay bounded without manual cleanup. Job ids are a neutral `job` prefix + local `HH:MM` — deliberately free of command text so a secret on the command line can't leak into an id, log line, or filename; the full command stays available via `job(action="list")`. +Jobs live in an in-memory map keyed by human-readable id (e.g., `job-23:30`); the log files persist on disk under the job dir. An **hourly reaper** drops any job whose age exceeds 24h — evicting it from the map and deleting its log file — so the map and the job dir stay bounded without manual cleanup. Job ids are a neutral `job` prefix + local `HH:MM` — deliberately free of command text so a secret on the command line can't leak into an id, log line, or filename. For the same reason `job(action="list")` exposes only non-sensitive metadata — the id and status — never the command text (which could carry a pasted secret). ``` JobState = Running | Exited { code } | Failed { error } diff --git a/src/jobs/id.rs b/src/jobs/id.rs index 90384dc..025aaf8 100644 --- a/src/jobs/id.rs +++ b/src/jobs/id.rs @@ -1,9 +1,10 @@ -//! Human-readable job ids: a neutral `job` label plus the local `HH:MM` it +//! Human-readable job ids: a neutral `job` prefix plus the local `HH:MM` it //! started, e.g. `job-23:30` — easier to refer to in conversation than an opaque -//! counter. The id is deliberately free of command text: a command can carry a -//! secret in its leading tokens, and the id surfaces in replies, `job(list)`, -//! reaper logs, and the log filename, so slugging the command would leak it. -//! A monotonic sequence suffix disambiguates jobs starting within the same minute. +//! counter. The prefix is hard-coded, never derived from the command: a command +//! can carry a secret in its leading tokens, and the id surfaces in replies, +//! `job(list)`, reaper logs, and the log filename, so the constructor refuses to +//! accept command text at all. A monotonic sequence suffix disambiguates jobs +//! starting within the same minute. use std::borrow::Borrow; use std::fmt; @@ -11,30 +12,26 @@ use std::sync::atomic::{AtomicU64, Ordering}; use chrono::Local; -/// Max slug length before the `-HH:MM` suffix. Short enough to type, long enough -/// to carry the gist of the command. -const MAX_SLUG: usize = 24; - -/// A human-readable job identifier: `-` with an optional `-` -/// collision suffix. A generated id matches `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$`; +/// A human-readable job identifier: `job-` with an optional `-` +/// collision suffix. A generated id matches `^job-\d{2}:\d{2}(-\d+)?$`; /// ids wrapped from client input for lookup are not revalidated. #[derive(Debug, Clone, Hash, Eq, PartialEq, PartialOrd, Ord, serde::Serialize)] pub struct JobId(String); impl JobId { - /// Build an id from a neutral `label` and the current local time. Callers - /// pass a fixed label (e.g. `"job"`), never raw command text, so a secret on - /// the command line can't leak into the id. `exists` reports whether a - /// candidate is already taken; only on a clash do we append a `-` drawn - /// from `seq`, so clean ids stay clean. The predicate keeps this decoupled - /// from however the caller tracks live ids. - pub fn generate(label: &str, seq: &AtomicU64, exists: impl Fn(&str) -> bool) -> Self { - let base = format!("{}-{}", slug(label), Local::now().format("%H:%M")); + /// Build an id from the hard-coded `job` prefix and the current local time. + /// The prefix is fixed here, never caller-supplied, so command text — which + /// can carry a secret in its leading tokens — can't leak into the id. `exists` + /// reports whether a candidate is already taken; only on a clash do we append + /// a `-` drawn from `seq`, so clean ids stay clean. The predicate keeps + /// this decoupled from however the caller tracks live ids. + pub fn generate(seq: &AtomicU64, exists: impl Fn(&str) -> bool) -> Self { + let base = format!("job-{}", Local::now().format("%H:%M")); if !exists(&base) { return Self(base); } - // Same slug, same minute: a monotonic suffix keeps the id unique without - // sacrificing readability. + // Same minute: a monotonic suffix keeps the id unique without sacrificing + // readability. let n = seq.fetch_add(1, Ordering::Relaxed); Self(format!("{base}-{n}")) } @@ -74,44 +71,13 @@ impl From<&str> for JobId { } } -/// Sanitize a label into a lowercase `[a-z0-9-]` slug so any caller-supplied -/// string yields a well-formed id: take the first few tokens, collapse runs of -/// non-alphanumerics to single dashes, trim them off both ends, and cap the -/// length. Falls back to `job` when nothing alphanumeric survives. -fn slug(label: &str) -> String { - let head = label - .split_whitespace() - .take(3) - .collect::>() - .join(" "); - let mut out = String::with_capacity(head.len()); - let mut prev_dash = false; - for ch in head.chars() { - if ch.is_ascii_alphanumeric() { - out.push(ch.to_ascii_lowercase()); - prev_dash = false; - } else if !prev_dash { - out.push('-'); - prev_dash = true; - } - } - // Cap first, then trim: the cut can land mid-run and leave a trailing dash. - let capped: String = out.trim_matches('-').chars().take(MAX_SLUG).collect(); - let trimmed = capped.trim_end_matches('-'); - if trimmed.is_empty() { - "job".to_string() - } else { - trimmed.to_string() - } -} - #[cfg(test)] mod tests { use super::*; - /// Mirror the documented `^[a-z0-9-]+-\d{2}:\d{2}(-\d+)?$` shape without - /// pulling in a regex dependency. The id has exactly one `:` (slug and - /// suffix never contain one), so split on it to isolate the time. + /// Mirror the documented `^job-\d{2}:\d{2}(-\d+)?$` shape without pulling in a + /// regex dependency. The id has exactly one `:` (prefix and suffix never + /// contain one), so split on it to isolate the time. fn valid_format(s: &str) -> bool { let two_digits = |x: &str| x.len() == 2 && x.bytes().all(|b| b.is_ascii_digit()); let Some((left, right)) = s.split_once(':') else { @@ -127,57 +93,20 @@ mod tests { ), None => (right, true), }; - two_digits(hh) - && two_digits(mm) - && suffix_ok - && !prefix.is_empty() - && prefix - .bytes() - .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') - } - - #[test] - fn slug_lowercases_and_replaces_non_alnum() { - assert_eq!(slug("ECHO Hello/World"), "echo-hello-world"); - } - - #[test] - fn slug_takes_first_three_tokens_and_collapses_dashes() { - // 4th token dropped; the space-then-`-` run collapses to one dash. - assert_eq!(slug("git commit -m message extra"), "git-commit-m"); - } - - #[test] - fn slug_caps_length_at_24() { - let s = slug("abcdefghijklmnopqrstuvwxyz0123456789"); - assert_eq!(s.len(), 24); - assert_eq!(s, "abcdefghijklmnopqrstuvwx"); - } - - #[test] - fn slug_trims_dash_left_by_the_cap() { - // 23 chars then a separator: the cut lands on char 24, leaving a dash. - let input = format!("{} b", "a".repeat(23)); - assert_eq!(slug(&input), "a".repeat(23)); - } - - #[test] - fn slug_falls_back_when_nothing_alphanumeric() { - assert_eq!(slug(""), "job"); - assert_eq!(slug("!@#$ %^&*"), "job"); + two_digits(hh) && two_digits(mm) && suffix_ok && prefix == "job" } #[test] fn generated_id_matches_documented_format() { let seq = AtomicU64::new(1); - let id = JobId::generate("cargo build --release", &seq, |_| false); + let id = JobId::generate(&seq, |_| false); assert!(valid_format(id.as_ref()), "bad format: {}", id.as_ref()); } #[test] fn no_collision_leaves_id_clean_and_seq_untouched() { let seq = AtomicU64::new(1); - let id = JobId::generate("echo hi", &seq, |_| false); + let id = JobId::generate(&seq, |_| false); let s = id.as_ref(); assert!(valid_format(s)); let (_, right) = s.split_once(':').expect("id carries a time"); @@ -193,7 +122,7 @@ mod tests { fn collision_appends_sequence_suffix() { let seq = AtomicU64::new(1); // `exists` always true => the base is "taken", forcing the suffix path. - let id = JobId::generate("echo hi", &seq, |_| true); + let id = JobId::generate(&seq, |_| true); let s = id.as_ref(); assert!(valid_format(s)); assert!(s.ends_with("-1"), "expected `-1` seq suffix, got {s}"); @@ -203,14 +132,14 @@ mod tests { #[test] fn display_and_as_ref_agree() { let seq = AtomicU64::new(1); - let id = JobId::generate("ls", &seq, |_| false); + let id = JobId::generate(&seq, |_| false); assert_eq!(id.to_string(), id.as_ref()); } #[test] fn serializes_as_a_plain_string() { let seq = AtomicU64::new(1); - let id = JobId::generate("ls", &seq, |_| false); + let id = JobId::generate(&seq, |_| false); let json = serde_json::to_string(&id).expect("serialize"); assert_eq!(json, format!("\"{}\"", id.as_ref())); } diff --git a/src/jobs/log.rs b/src/jobs/log.rs index 1bce429..bd395b4 100644 --- a/src/jobs/log.rs +++ b/src/jobs/log.rs @@ -39,6 +39,10 @@ pub async fn read_page( cursor: usize, limit: usize, ) -> Result { + // A zero limit would yield an empty page whose `next_cursor == cursor` with + // `has_more` still true, so a client advancing on `next_cursor` would spin + // forever on the same window. Clamp to at least one line. + let limit = limit.max(1); // A failed read (missing/unreadable log) propagates as a typed error. Only // non-UTF-8 *content* is rendered lossily: a command that writes raw bytes // (compiled output, escape sequences) must not produce a silently-empty page. @@ -85,6 +89,19 @@ mod tests { assert!(page.lines[0].contains("line1")); } + #[tokio::test] + async fn read_page_clamps_zero_limit_to_avoid_infinite_loop() { + // limit=0 must not produce an empty, non-advancing page: a client paging + // on next_cursor would otherwise loop forever on the same window. + let dir = tempfile::tempdir().unwrap(); + let log = dir.path().join("multi.log"); + tokio::fs::write(&log, b"a\nb\nc\n").await.unwrap(); + let page = read_page(&log, 0, 0).await.unwrap(); + assert_eq!(page.lines.len(), 1, "zero limit must clamp to one line"); + assert!(page.next_cursor > 0, "cursor must advance past the start"); + assert!(page.has_more, "more lines remain after the clamped page"); + } + #[tokio::test] async fn read_page_errors_on_missing_log() { // A missing log is an error, not a silently-empty page. diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index d1e3e34..944d878 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -34,7 +34,6 @@ pub enum JobState { struct ProcessGroupId(u32); struct Job { - cmd: String, log_path: PathBuf, /// Process group to signal on kill. The child leads its own group, so this /// equals its pid (see `run`). `None` only if the OS withheld a pid. @@ -54,10 +53,13 @@ pub enum RunResult { Backgrounded { id: JobId }, } +/// Public, per-job metadata for `job(action="list")`. Deliberately omits the +/// command text: a command can carry a secret in its leading tokens +/// (`PGPASSWORD=…`, bearer headers, pasted tokens), and this is returned in +/// ordinary MCP responses — so the listing exposes only the id and status. #[derive(Debug, serde::Serialize)] pub struct JobSummary { pub id: JobId, - pub cmd: String, pub state: JobState, } @@ -97,12 +99,12 @@ impl JobStore { // awaits while the guard is held (file create + spawn are synchronous), // so the critical section stays short. // - // The id is built from a fixed `"job"` label, never `cmd`: a command can - // carry a secret in its leading tokens (`mysql -psecret`, `PGPASSWORD=…`), - // and the id ends up in `bash`'s reply, `job(list)`, reaper logs, and the - // log filename — so deriving it from `cmd` would leak that secret. + // The id uses a hard-coded `job` prefix, never `cmd`: a command can carry + // a secret in its leading tokens (`mysql -psecret`, `PGPASSWORD=…`), and + // the id ends up in `bash`'s reply, `job(list)`, reaper logs, and the log + // filename — so deriving it from `cmd` would leak that secret. let mut jobs = self.jobs.lock().await; - let id = JobId::generate("job", &self.seq, |candidate| jobs.contains_key(candidate)); + let id = JobId::generate(&self.seq, |candidate| jobs.contains_key(candidate)); let log_path = self.dir.join(format!("{id}.log")); // ponytail: stdout+stderr merged into one log (terminal-style). Split into @@ -148,7 +150,6 @@ impl JobStore { } let job = Arc::new(Job { - cmd, log_path: log_path.clone(), pgid, state: state.clone(), @@ -193,7 +194,20 @@ impl JobStore { return Ok(None); }; let state = job.state.lock().await.clone(); - let page = read_page(&job.log_path, cursor, limit.unwrap_or(DEFAULT_PAGE)).await?; + // A reaper pass can evict this id and delete its log between the clone + // above and this read. If the log is gone *and* the id is no longer + // tracked, that's the stable "unknown job" result, not a read fault — + // re-check membership so an eviction race doesn't surface as an error. + let page = match read_page(&job.log_path, cursor, limit.unwrap_or(DEFAULT_PAGE)).await { + Ok(page) => page, + Err(JobLogError::Read(error)) if error.kind() == std::io::ErrorKind::NotFound => { + if self.jobs.lock().await.contains_key(id) { + return Err(JobLogError::Read(error)); + } + return Ok(None); + } + Err(error) => return Err(error), + }; Ok(Some((state, page))) } @@ -209,7 +223,6 @@ impl JobStore { for (id, job) in snapshot { out.push(JobSummary { id, - cmd: job.cmd.clone(), state: job.state.lock().await.clone(), }); } @@ -444,7 +457,6 @@ mod tests { let log_path = store.dir.join("jfake.log"); tokio::fs::write(&log_path, "running\n").await.unwrap(); let job = Arc::new(Job { - cmd: "unkillable".into(), log_path: log_path.clone(), pgid: None, state: Arc::new(Mutex::new(JobState::Running)), @@ -497,7 +509,7 @@ mod tests { #[tokio::test] async fn list_reports_all_jobs() { let store = store(Duration::from_secs(5)); - // Two distinct commands — one inline, one explicitly backgrounded. + // Two commands — one inline, one explicitly backgrounded — both tracked. store .run("echo alpha".into(), None, None, false) .await @@ -509,9 +521,6 @@ mod tests { let jobs = store.list().await; assert_eq!(jobs.len(), 2, "expected two jobs, got {}", jobs.len()); - let cmds: Vec<&str> = jobs.iter().map(|j| j.cmd.as_str()).collect(); - assert!(cmds.contains(&"echo alpha"), "missing 'echo alpha'"); - assert!(cmds.contains(&"echo beta"), "missing 'echo beta'"); // IDs must be sorted so the list is deterministic. assert!(jobs[0].id < jobs[1].id, "list should be sorted by id"); } @@ -532,4 +541,29 @@ mod tests { assert!(page.has_more); assert_eq!(page.total_lines, 10); } + + #[tokio::test] + async fn poll_surfaces_read_error_while_job_is_tracked() { + let store = store(Duration::from_secs(5)); + // A tracked job whose log has vanished is a genuine read fault, not an + // eviction race: poll must surface the error, never collapse it to + // Ok(None) (the NotFound -> Ok(None) mapping applies only once the id is + // gone from the map). + let (_tx, rx) = watch::channel(false); + let log_path = store.dir.join("gone.log"); // never created + let job = Arc::new(Job { + log_path, + pgid: None, + state: Arc::new(Mutex::new(JobState::Running)), + done: rx, + started: tokio::time::Instant::now(), + }); + let id = JobId::from("gone"); + store.jobs.lock().await.insert(id.clone(), job); + + assert!( + store.poll(&id, 0, None).await.is_err(), + "missing log for a still-tracked job must surface as an error" + ); + } }