diff --git a/CLAUDE.md b/CLAUDE.md index 9d5d205..80fd308 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 — 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 | | `src/tools/files.rs` | file operations (`tokio::fs`; `ls`/`find`/`grep` shelled out) | 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/docs/architecture.md b/docs/architecture.md index 8cee573..9299ae3 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 — 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`. | | `tools/files.rs` | File operations (read/write/append/delete/list/grep/move). | @@ -38,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) ``` @@ -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., `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/docs/usage.md b/docs/usage.md index 0efb1b9..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** (`j1`, `j2`, …) — 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 j7 still running after the inline window. Poll it with job(action=\"poll\", id=\"j7\")." +→ "job job-23:30 still running after the inline window. Poll it with job(action=\"poll\", id=\"job-23:30\")." -job(action="poll", id="j7") +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="j7", 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="j7", 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="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="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="job-23:30") # kill a running job ``` ## 📁 Files diff --git a/src/jobs/id.rs b/src/jobs/id.rs new file mode 100644 index 0000000..025aaf8 --- /dev/null +++ b/src/jobs/id.rs @@ -0,0 +1,146 @@ +//! 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 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; +use std::sync::atomic::{AtomicU64, Ordering}; + +use chrono::Local; + +/// 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 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 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 + } +} + +/// 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()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// 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 { + 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 == "job" + } + + #[test] + fn generated_id_matches_documented_format() { + let seq = AtomicU64::new(1); + 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(&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(&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(&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(&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 new file mode 100644 index 0000000..bd395b4 --- /dev/null +++ b/src/jobs/log.rs @@ -0,0 +1,112 @@ +//! 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; + +/// 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, +) -> 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. + 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(); + // 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(); + Ok(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.unwrap(); + // 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")); + } + + #[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. + 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 9fede31..944d878 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -5,17 +5,19 @@ use std::{ collections::HashMap, path::PathBuf, process::Stdio, - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, + sync::{Arc, atomic::AtomicU64}, time::Duration, }; use tokio::sync::{Mutex, watch}; +mod id; +mod log; mod reaper; +pub use id::JobId; +use log::{DEFAULT_PAGE, read_page}; +pub use log::{JobLogError, Page}; use reaper::{kill_job, spawn_reaper}; #[derive(Debug, Clone, serde::Serialize)] @@ -32,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. @@ -49,22 +50,16 @@ 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 }, -} - -/// 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, + 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: String, - pub cmd: String, + pub id: JobId, pub state: JobState, } @@ -73,11 +68,9 @@ pub struct JobStore { dir: PathBuf, inline_timeout: Duration, seq: Arc, - jobs: Arc>>>, + 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)?; @@ -100,7 +93,18 @@ 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 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 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(&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 @@ -146,14 +150,14 @@ impl JobStore { } let job = Arc::new(Job { - cmd, log_path: log_path.clone(), pgid, state: state.clone(), 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 { @@ -172,27 +176,44 @@ 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: &str, + 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)) + // 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))) } 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))) @@ -202,7 +223,6 @@ impl JobStore { for (id, job) in snapshot { out.push(JobSummary { id, - cmd: job.cmd.clone(), state: job.state.lock().await.clone(), }); } @@ -212,7 +232,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; }; @@ -220,30 +240,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::*; @@ -290,7 +286,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; @@ -329,7 +325,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; @@ -353,7 +349,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] @@ -365,7 +365,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)] @@ -392,7 +395,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; } @@ -416,7 +419,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; @@ -431,7 +434,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. @@ -454,20 +457,20 @@ 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)), 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.unwrap().is_some(), "running job whose kill failed must stay tracked" ); assert!(log_path.exists(), "its log must not be deleted"); @@ -494,7 +497,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; } @@ -506,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 @@ -518,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"); } @@ -528,16 +528,14 @@ 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, - }; - let (_s, page) = store.poll(&id, 0, Some(3)).await.unwrap(); + // 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().unwrap(); assert_eq!(page.lines.len(), 3); assert_eq!(page.next_cursor, 3); assert!(page.has_more); @@ -545,20 +543,27 @@ mod tests { } #[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. + 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!( - !page.lines.is_empty(), - "binary log must not produce empty page" + store.poll(&id, 0, None).await.is_err(), + "missing log for a still-tracked job must surface as an error" ); - assert!(page.lines[0].contains("line1")); } } 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..e8d63a9 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,13 +151,15 @@ 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) .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 => { @@ -168,6 +170,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 {