diff --git a/Cargo.lock b/Cargo.lock index 7f95a04a..c69ca8f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -74,6 +74,7 @@ dependencies = [ "agentflare-flare-code", "agentflare-flare-output", "agentflare-gateway-registry", + "agentflare-jobs", "agentflare-skill-registry", "agentflare-store", "axum", diff --git a/Cargo.toml b/Cargo.toml index dccba76d..af69053e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -84,6 +84,7 @@ flare-git-core = { path = "crates/flare-git-core" } flare-proxy = { path = "crates/flare-proxy" } flare-vault = { path = "crates/flare-vault" } skill = { version = "0.8", default-features = false, features = ["network"] } +agentflare-jobs = { path = "crates/agentflare-jobs" } [target.'cfg(unix)'.dependencies] libc = "0.2" @@ -110,7 +111,7 @@ built = { version = "0.8", features = ["chrono"] } insta = { version = "1", features = ["json"] } tempfile = "3" pretty_assertions = "1" -reqwest = { version = "0.12", default-features = false, features = ["stream"] } +reqwest = { version = "0.12", default-features = false, features = ["stream", "json"] } [lints.rust] unsafe_code = "warn" diff --git a/crates/agentflare-jobs/src/queue.rs b/crates/agentflare-jobs/src/queue.rs index 765f3866..96fabdf5 100644 --- a/crates/agentflare-jobs/src/queue.rs +++ b/crates/agentflare-jobs/src/queue.rs @@ -1,12 +1,23 @@ use crate::types::{JobInfo, JobOutput, JobState}; -use parking_lot::Mutex; +use parking_lot::{Condvar, Mutex}; use rusqlite::params; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; +#[derive(Clone)] pub struct Queue { conn: Arc>, log_dir: PathBuf, + // Wakes idle workers the moment a job becomes available (fresh enqueue, + // or a retry going back to 'queued'), instead of them finding out only + // on their next poll. The bool is a pending-signal flag, not app state: + // `notify_all()` alone only wakes threads already parked in `wait_for`, + // so a wake that lands between a worker's dequeue-check and its call to + // `wait_for_work` would otherwise be lost until the fallback timeout — + // the flag makes the signal durable across that race by having + // `wait_for_work` check it (under the same lock) before ever blocking. + notify: Arc<(Mutex, Condvar)>, } #[derive(Debug, thiserror::Error)] @@ -24,25 +35,31 @@ pub enum Error { } pub fn migrations() -> rusqlite_migration::Migrations<'static> { - rusqlite_migration::Migrations::new(vec![rusqlite_migration::M::up( - "CREATE TABLE IF NOT EXISTS agent_jobs ( - id TEXT PRIMARY KEY NOT NULL, - state TEXT NOT NULL DEFAULT 'queued', - payload TEXT NOT NULL, - retries INTEGER NOT NULL DEFAULT 0, - max_retries INTEGER NOT NULL DEFAULT 3, - error TEXT, - stdout_log_path TEXT, - stderr_log_path TEXT, - exit_code INTEGER, - timed_out INTEGER NOT NULL DEFAULT 0, - created_at INTEGER NOT NULL, - started_at INTEGER, - finished_at INTEGER - ); - CREATE INDEX IF NOT EXISTS idx_agent_jobs_state ON agent_jobs(state); - CREATE INDEX IF NOT EXISTS idx_agent_jobs_created ON agent_jobs(created_at);", - )]) + rusqlite_migration::Migrations::new(vec![ + rusqlite_migration::M::up( + "CREATE TABLE IF NOT EXISTS agent_jobs ( + id TEXT PRIMARY KEY NOT NULL, + state TEXT NOT NULL DEFAULT 'queued', + payload TEXT NOT NULL, + retries INTEGER NOT NULL DEFAULT 0, + max_retries INTEGER NOT NULL DEFAULT 3, + error TEXT, + stdout_log_path TEXT, + stderr_log_path TEXT, + exit_code INTEGER, + timed_out INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + started_at INTEGER, + finished_at INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_agent_jobs_state ON agent_jobs(state); + CREATE INDEX IF NOT EXISTS idx_agent_jobs_created ON agent_jobs(created_at);", + ), + rusqlite_migration::M::up( + "ALTER TABLE agent_jobs ADD COLUMN stdout_bytes INTEGER NOT NULL DEFAULT 0; + ALTER TABLE agent_jobs ADD COLUMN stderr_bytes INTEGER NOT NULL DEFAULT 0;", + ), + ]) } impl Queue { @@ -51,6 +68,7 @@ impl Queue { Ok(Self { conn: Arc::new(Mutex::new(conn)), log_dir, + notify: Arc::new((Mutex::new(false), Condvar::new())), }) } @@ -59,6 +77,7 @@ impl Queue { Ok(Self { conn: Arc::new(Mutex::new(conn)), log_dir, + notify: Arc::new((Mutex::new(false), Condvar::new())), }) } @@ -66,6 +85,35 @@ impl Queue { &self.log_dir } + /// Blocks the calling (worker) thread until a job becomes available or + /// `timeout` elapses, whichever comes first. Checks the pending-signal + /// flag before blocking, so a `wake_workers` call that landed just + /// before this one (e.g. between this worker's dequeue-check and this + /// call) is still observed immediately instead of being lost until + /// `timeout` — normal pickup is near-instant via `notify`, not the + /// timeout, in either ordering. + pub fn wait_for_work(&self, timeout: Duration) { + let (lock, cvar) = &*self.notify; + let mut signaled = lock.lock(); + if !*signaled { + cvar.wait_for(&mut signaled, timeout); + } + *signaled = false; + } + + /// Wakes every thread parked in `wait_for_work` (or the next one to call + /// it, per the flag above), whether or not there's actually a job for + /// them — they just re-check `dequeue` either way. Called after + /// enqueueing/re-queueing a job, and by `WorkerPool::shutdown` so + /// workers notice a stop request immediately rather than up to + /// `timeout` later. + pub fn wake_workers(&self) { + let (lock, cvar) = &*self.notify; + let mut signaled = lock.lock(); + *signaled = true; + cvar.notify_all(); + } + pub fn enqueue(&self, job: &crate::types::AgentJob) -> Result { let id = db_kit::ids::new_id(); let now = db_kit::ids::now(); @@ -76,8 +124,12 @@ impl Queue { VALUES (?1, 'queued', ?2, ?3, ?4)", params![id, payload, job.max_retries, now], )?; + drop(conn); + self.wake_workers(); Ok(JobInfo { id, + command: job.command.clone(), + args: job.args.clone(), state: JobState::Queued, retries: 0, max_retries: job.max_retries, @@ -127,14 +179,18 @@ impl Queue { timed_out = ?3, stdout_log_path = ?4, stderr_log_path = ?5, - finished_at = ?6 - WHERE id = ?7", + stdout_bytes = ?6, + stderr_bytes = ?7, + finished_at = ?8 + WHERE id = ?9", params![ state, output.exit_code, output.timed_out as i32, output.stdout_path.to_string_lossy().as_ref(), output.stderr_path.to_string_lossy().as_ref(), + output.stdout_total_bytes as i64, + output.stderr_total_bytes as i64, now, id ], @@ -150,7 +206,8 @@ impl Queue { params![id], |r| Ok((r.get(0)?, r.get(1)?)), )?; - if retries < max_retries { + let retried = retries < max_retries; + if retried { conn.execute( "UPDATE agent_jobs SET state = 'queued', retries = retries + 1, error = ?1, started_at = NULL @@ -165,6 +222,12 @@ impl Queue { params![error, now, id], )?; } + drop(conn); + // A retry goes back to 'queued' — wake workers so it's picked up + // promptly instead of waiting out the fallback poll interval. + if retried { + self.wake_workers(); + } Ok(()) } @@ -176,7 +239,8 @@ impl Queue { }; let sql = format!( "SELECT id, state, retries, max_retries, error, created_at, started_at, - finished_at, exit_code, timed_out, stdout_log_path, stderr_log_path + finished_at, exit_code, timed_out, stdout_log_path, stderr_log_path, + stdout_bytes, stderr_bytes, payload FROM agent_jobs {where_clause} ORDER BY created_at DESC LIMIT 100" @@ -197,7 +261,8 @@ impl Queue { let conn = self.conn.lock(); let row = conn.query_row( "SELECT id, state, retries, max_retries, error, created_at, started_at, - finished_at, exit_code, timed_out, stdout_log_path, stderr_log_path + finished_at, exit_code, timed_out, stdout_log_path, stderr_log_path, + stdout_bytes, stderr_bytes, payload FROM agent_jobs WHERE id = ?1", params![id], map_job_row, @@ -219,13 +284,34 @@ impl Queue { Ok(()) } + /// Deletes finished jobs older than `older_than_secs`, including their + /// stdout/stderr log files on disk — those aren't tracked anywhere else, + /// so leaving them behind here would mean disk usage grows forever even + /// as the DB rows are reclaimed. pub fn cleanup(&self, older_than_secs: i64) -> Result { let cutoff = db_kit::ids::now() - older_than_secs; let conn = self.conn.lock(); + let log_paths: Vec<(Option, Option)> = { + let mut stmt = conn.prepare( + "SELECT stdout_log_path, stderr_log_path FROM agent_jobs + WHERE finished_at IS NOT NULL AND finished_at < ?1", + )?; + stmt.query_map(params![cutoff], |r| Ok((r.get(0)?, r.get(1)?)))? + .collect::, _>>()? + }; let deleted = conn.execute( "DELETE FROM agent_jobs WHERE finished_at IS NOT NULL AND finished_at < ?1", params![cutoff], )?; + drop(conn); + for (stdout_path, stderr_path) in log_paths { + if let Some(p) = stdout_path { + let _ = std::fs::remove_file(p); + } + if let Some(p) = stderr_path { + let _ = std::fs::remove_file(p); + } + } Ok(deleted as u64) } } @@ -236,8 +322,19 @@ fn map_job_row(r: &rusqlite::Row<'_>) -> rusqlite::Result { let timed_out: bool = r.get::<_, i32>(9)? != 0; let stdout_path: Option = r.get(10)?; let stderr_path: Option = r.get(11)?; + let stdout_bytes: i64 = r.get(12)?; + let stderr_bytes: i64 = r.get(13)?; + let payload_json: String = r.get(14)?; + // `payload` is our own `serde_json::to_string(&AgentJob)` from `enqueue`, + // so a parse failure here would mean on-disk corruption, not bad input — + // fall back to an empty command/args rather than failing the whole read. + let (command, args) = serde_json::from_str::(&payload_json) + .map(|job| (job.command, job.args)) + .unwrap_or_default(); Ok(JobInfo { id: r.get(0)?, + command, + args, state: match state_str.as_str() { "running" => JobState::Running, "exited" => JobState::Exited, @@ -256,8 +353,8 @@ fn map_job_row(r: &rusqlite::Row<'_>) -> rusqlite::Result { timed_out, stdout_path: stdout_path.map(Into::into).unwrap_or_default(), stderr_path: stderr_path.map(Into::into).unwrap_or_default(), - stdout_total_bytes: 0, - stderr_total_bytes: 0, + stdout_total_bytes: stdout_bytes as u64, + stderr_total_bytes: stderr_bytes as u64, }), }) } diff --git a/crates/agentflare-jobs/src/supervisor.rs b/crates/agentflare-jobs/src/supervisor.rs index 8032b591..7df7b199 100644 --- a/crates/agentflare-jobs/src/supervisor.rs +++ b/crates/agentflare-jobs/src/supervisor.rs @@ -18,7 +18,14 @@ pub struct Supervisor { } impl Supervisor { + /// `id` names the log files (`{id}.stdout`/`{id}.stderr`) — callers + /// running this under `agentflare-jobs::Queue` must pass the job's own + /// queue id, not a fresh one, so a running job's log path is derivable + /// from its id alone (`queue.log_dir().join(format!("{id}.stdout"))`) + /// without waiting for the job to finish and report it back. + #[allow(clippy::too_many_arguments)] pub fn new( + id: String, command: String, args: Vec, env: Vec<(String, String)>, @@ -27,7 +34,6 @@ impl Supervisor { kill_after_secs: u64, log_dir: PathBuf, ) -> Self { - let id = db_kit::ids::new_id(); let stdout_path = log_dir.join(format!("{id}.stdout")); let stderr_path = log_dir.join(format!("{id}.stderr")); Self { diff --git a/crates/agentflare-jobs/src/types.rs b/crates/agentflare-jobs/src/types.rs index 1ca4f6c6..a057a49a 100644 --- a/crates/agentflare-jobs/src/types.rs +++ b/crates/agentflare-jobs/src/types.rs @@ -104,6 +104,8 @@ pub struct JobOutput { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct JobInfo { pub id: String, + pub command: String, + pub args: Vec, pub state: JobState, pub retries: u32, pub max_retries: u32, diff --git a/crates/agentflare-jobs/src/worker.rs b/crates/agentflare-jobs/src/worker.rs index a716a6e7..097848a3 100644 --- a/crates/agentflare-jobs/src/worker.rs +++ b/crates/agentflare-jobs/src/worker.rs @@ -33,6 +33,9 @@ impl WorkerPool { pub fn shutdown(&mut self) { self.running.store(false, Ordering::SeqCst); + // Workers may be parked in `wait_for_work` for up to its timeout — + // wake them immediately so shutdown doesn't wait that out. + self.queue.wake_workers(); let handles = std::mem::take(&mut self.handles); for h in handles { let _ = h.join(); @@ -45,6 +48,7 @@ fn worker_loop(queue: &Queue, running: &AtomicBool) { match queue.dequeue() { Ok(Some((id, job))) => { let mut sup = Supervisor::new( + id.clone(), job.command.clone(), job.args.clone(), job.env.clone(), @@ -68,7 +72,9 @@ fn worker_loop(queue: &Queue, running: &AtomicBool) { } } Ok(None) => { - std::thread::sleep(Duration::from_millis(200)); + // Woken immediately by `wake_workers` (enqueue/retry/shutdown); + // the timeout is only a safety net against a missed wakeup. + queue.wait_for_work(Duration::from_secs(1)); } Err(e) => { eprintln!("agentflare-jobs: dequeue error: {e}"); diff --git a/crates/agentflare-jobs/tests/queue_test.rs b/crates/agentflare-jobs/tests/queue_test.rs index 5e4e45fb..0c9e78f4 100644 --- a/crates/agentflare-jobs/tests/queue_test.rs +++ b/crates/agentflare-jobs/tests/queue_test.rs @@ -1,8 +1,10 @@ use agentflare_jobs::{AgentJob, JobState, Queue}; fn test_queue() -> Queue { - let dir = tempfile::tempdir().unwrap(); - Queue::open_memory(dir.path().join("logs")).unwrap() + // `.keep()` so the dir outlives this function — otherwise the returned + // `Queue`'s `log_dir` would point at an already-deleted path. + let dir = tempfile::tempdir().unwrap().keep(); + Queue::open_memory(dir.join("logs")).unwrap() } fn true_cmd() -> (&'static str, Vec<&'static str>) { @@ -47,6 +49,7 @@ fn complete_sets_exited() { q.enqueue(&AgentJob::new(cmd).args(args)).unwrap(); let (id, job) = q.dequeue().unwrap().unwrap(); let sup = agentflare_jobs::Supervisor::new( + id.clone(), job.command.clone(), job.args.clone(), vec![], @@ -65,6 +68,52 @@ fn complete_sets_exited() { assert_eq!(info.output.as_ref().unwrap().exit_code, Some(0)); } +#[test] +fn complete_persists_stdout_and_stderr_byte_counts() { + let q = test_queue(); + let (cmd, args) = if cfg!(windows) { + ("cmd", vec!["/c", "echo hello & echo world 1>&2"]) + } else { + ("sh", vec!["-c", "echo hello; echo world 1>&2"]) + }; + q.enqueue(&AgentJob::new(cmd).args(args)).unwrap(); + let (id, job) = q.dequeue().unwrap().unwrap(); + let mut supervisor = agentflare_jobs::Supervisor::new( + id.clone(), + job.command.clone(), + job.args.clone(), + vec![], + None, + 10, + 2, + q.log_dir().to_path_buf(), + ); + let (output, _) = supervisor.spawn().unwrap(); + assert!(output.stdout_total_bytes > 0); + assert!(output.stderr_total_bytes > 0); + q.complete(&id, &output, true).unwrap(); + + // Regression: `get`/`list` used to always report 0 for these two fields + // regardless of what `Supervisor` actually captured, because the + // columns didn't exist in the schema and `complete` silently dropped + // them on write. + let info = q.get(&id).unwrap(); + let persisted = info.output.as_ref().unwrap(); + assert_eq!(persisted.stdout_total_bytes, output.stdout_total_bytes); + assert_eq!(persisted.stderr_total_bytes, output.stderr_total_bytes); + + let listed = q.list(None).unwrap(); + let listed_output = listed + .iter() + .find(|j| j.id == id) + .unwrap() + .output + .as_ref() + .unwrap(); + assert_eq!(listed_output.stdout_total_bytes, output.stdout_total_bytes); + assert_eq!(listed_output.stderr_total_bytes, output.stderr_total_bytes); +} + #[test] fn fail_retries_then_permanent() { let q = test_queue(); @@ -136,3 +185,35 @@ fn cleanup_removes_old_jobs() { let count = q.cleanup(1_000_000_000).unwrap(); assert_eq!(count, 0); } + +#[test] +fn cleanup_removes_old_jobs_and_their_log_files() { + let q = test_queue(); + let (cmd, args) = true_cmd(); + q.enqueue(&AgentJob::new(cmd).args(args)).unwrap(); + let (id, job) = q.dequeue().unwrap().unwrap(); + let mut supervisor = agentflare_jobs::Supervisor::new( + id.clone(), + job.command.clone(), + job.args.clone(), + vec![], + None, + 10, + 2, + q.log_dir().to_path_buf(), + ); + let (output, _) = supervisor.spawn().unwrap(); + let stdout_path = output.stdout_path.clone(); + let stderr_path = output.stderr_path.clone(); + q.complete(&id, &output, true).unwrap(); + assert!(stdout_path.exists()); + assert!(stderr_path.exists()); + + // -1 => cutoff is 1s in the future, so "finished before cutoff" matches + // everything already finished, deterministically, without a real sleep. + let count = q.cleanup(-1).unwrap(); + assert_eq!(count, 1); + assert!(q.get(&id).is_err(), "job row should be gone"); + assert!(!stdout_path.exists(), "stdout log should be removed"); + assert!(!stderr_path.exists(), "stderr log should be removed"); +} diff --git a/crates/agentflare-jobs/tests/supervisor_test.rs b/crates/agentflare-jobs/tests/supervisor_test.rs new file mode 100644 index 00000000..28a15f03 --- /dev/null +++ b/crates/agentflare-jobs/tests/supervisor_test.rs @@ -0,0 +1,153 @@ +use agentflare_jobs::{JobState, Supervisor}; +use std::fs; +use tempfile::TempDir; + +fn sup( + command: &str, + args: &[&str], + timeout_secs: u64, + kill_after_secs: u64, +) -> (TempDir, Supervisor) { + let log_tmp = tempfile::tempdir().unwrap(); + let supervisor = Supervisor::new( + "test-job".to_string(), + command.to_string(), + args.iter().map(|s| s.to_string()).collect(), + vec![], + None, + timeout_secs, + kill_after_secs, + log_tmp.path().to_path_buf(), + ); + (log_tmp, supervisor) +} + +#[test] +fn exit_success_reports_zero_code() { + let (cmd, args) = if cfg!(windows) { + ("cmd", vec!["/c", "exit 0"]) + } else { + ("true", vec![]) + }; + let (_log_tmp, mut supervisor) = sup(cmd, &args, 10, 2); + let (output, state) = supervisor.spawn().unwrap(); + + assert_eq!(state, JobState::Exited); + assert_eq!(output.exit_code, Some(0)); + assert!(!output.timed_out); +} + +#[test] +fn exit_failure_reports_nonzero_code() { + let (cmd, args) = if cfg!(windows) { + ("cmd", vec!["/c", "exit 7"]) + } else { + ("sh", vec!["-c", "exit 7"]) + }; + let (_log_tmp, mut supervisor) = sup(cmd, &args, 10, 2); + let (output, state) = supervisor.spawn().unwrap(); + + assert_eq!(state, JobState::Exited); + assert_eq!(output.exit_code, Some(7)); + assert!(!output.timed_out); +} + +#[test] +fn timeout_kills_long_running_process() { + let (cmd, args) = if cfg!(windows) { + // `timeout /t` refuses to run with redirected stdin ("INPUT + // REDIRECTION IS NOT SUPPORTED") and exits instantly instead of + // sleeping — `ping` against loopback is the standard + // redirection-safe stand-in for a long delay on Windows. + ("cmd", vec!["/c", "ping -n 31 127.0.0.1 >nul"]) + } else { + ("sleep", vec!["30"]) + }; + let (_log_tmp, mut supervisor) = sup(cmd, &args, 1, 1); + + let start = std::time::Instant::now(); + let (output, state) = supervisor.spawn().unwrap(); + let elapsed = start.elapsed(); + + assert_eq!(state, JobState::Killed); + assert!(output.timed_out); + // Bounded by timeout + kill_after, not the full 30s sleep. + assert!( + elapsed.as_secs() < 15, + "expected kill well before natural exit, took {elapsed:?}" + ); +} + +#[test] +fn env_vars_are_propagated_to_child() { + let (cmd, args) = if cfg!(windows) { + ("cmd", vec!["/c", "echo %SUPERVISOR_TEST_VAR%"]) + } else { + ("sh", vec!["-c", "echo $SUPERVISOR_TEST_VAR"]) + }; + let log_tmp = tempfile::tempdir().unwrap(); + let mut supervisor = Supervisor::new( + "test-job".to_string(), + cmd.to_string(), + args.iter().map(|s| s.to_string()).collect(), + vec![("SUPERVISOR_TEST_VAR".to_string(), "hello-env".to_string())], + None, + 10, + 2, + log_tmp.path().to_path_buf(), + ); + let (output, state) = supervisor.spawn().unwrap(); + + assert_eq!(state, JobState::Exited); + let stdout = fs::read_to_string(&output.stdout_path).unwrap(); + assert!(stdout.contains("hello-env"), "stdout was: {stdout:?}"); +} + +#[test] +fn cwd_is_applied_to_child() { + let dir = tempfile::tempdir().unwrap(); + let dir_path = dir.path().to_path_buf(); + let (cmd, args) = if cfg!(windows) { + ("cmd", vec!["/c", "cd"]) + } else { + ("pwd", vec![]) + }; + let log_tmp = tempfile::tempdir().unwrap(); + let mut supervisor = Supervisor::new( + "test-job".to_string(), + cmd.to_string(), + args.iter().map(|s| s.to_string()).collect(), + vec![], + Some(dir_path.clone()), + 10, + 2, + log_tmp.path().to_path_buf(), + ); + let (output, state) = supervisor.spawn().unwrap(); + + assert_eq!(state, JobState::Exited); + let stdout = fs::read_to_string(&output.stdout_path).unwrap(); + let canonical_dir = fs::canonicalize(&dir_path).unwrap(); + let canonical_stdout = fs::canonicalize(stdout.trim()).unwrap(); + assert_eq!(canonical_stdout, canonical_dir); +} + +#[test] +fn stdout_and_stderr_are_captured_separately() { + let (cmd, args) = if cfg!(windows) { + ("cmd", vec!["/c", "echo out-line & echo err-line 1>&2"]) + } else { + ("sh", vec!["-c", "echo out-line; echo err-line 1>&2"]) + }; + let (_log_tmp, mut supervisor) = sup(cmd, &args, 10, 2); + let (output, state) = supervisor.spawn().unwrap(); + + assert_eq!(state, JobState::Exited); + let stdout = fs::read_to_string(&output.stdout_path).unwrap(); + let stderr = fs::read_to_string(&output.stderr_path).unwrap(); + assert!(stdout.contains("out-line"), "stdout was: {stdout:?}"); + assert!(!stdout.contains("err-line"), "stdout was: {stdout:?}"); + assert!(stderr.contains("err-line"), "stderr was: {stderr:?}"); + assert!(output.stdout_total_bytes > 0); + assert!(output.stderr_total_bytes > 0); +} diff --git a/crates/agentflare-jobs/tests/worker_test.rs b/crates/agentflare-jobs/tests/worker_test.rs new file mode 100644 index 00000000..a331047e --- /dev/null +++ b/crates/agentflare-jobs/tests/worker_test.rs @@ -0,0 +1,94 @@ +use agentflare_jobs::{AgentJob, JobState, Queue, WorkerPool}; + +fn test_queue() -> Queue { + let dir = tempfile::tempdir().unwrap(); + Queue::open_memory(dir.path().join("logs")).unwrap() +} + +fn true_cmd() -> (&'static str, Vec<&'static str>) { + if cfg!(windows) { + ("cmd", vec!["/c", "exit 0"]) + } else { + ("true", vec![]) + } +} + +#[test] +fn worker_pool_picks_up_a_queued_job_and_completes_it() { + let q = test_queue(); + let mut pool = WorkerPool::new(q.clone()); + pool.start(1); + + let (cmd, args) = true_cmd(); + let info = q.enqueue(&AgentJob::new(cmd).args(args)).unwrap(); + + let mut final_info = None; + for _ in 0..200 { + let i = q.get(&info.id).unwrap(); + if matches!( + i.state, + JobState::Exited | JobState::Failed | JobState::Killed + ) { + final_info = Some(i); + break; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + pool.shutdown(); + + let final_info = final_info.expect("job should have finished within 2s"); + assert_eq!(final_info.state, JobState::Exited); +} + +#[test] +fn worker_pool_picks_up_job_promptly_via_notify_not_just_the_fallback_poll() { + let q = test_queue(); + let mut pool = WorkerPool::new(q.clone()); + pool.start(1); + + let (cmd, args) = true_cmd(); + let start = std::time::Instant::now(); + let info = q.enqueue(&AgentJob::new(cmd).args(args)).unwrap(); + + let mut final_info = None; + for _ in 0..100 { + let i = q.get(&info.id).unwrap(); + if matches!( + i.state, + JobState::Exited | JobState::Failed | JobState::Killed + ) { + final_info = Some(i); + break; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + let elapsed = start.elapsed(); + pool.shutdown(); + + assert!(final_info.is_some(), "job should have finished"); + // The fallback poll timeout is 1s; a working notify wakes the worker + // essentially immediately, so a generous 300ms bound still clearly + // distinguishes "notified" from "waited out the fallback timeout". + assert!( + elapsed.as_millis() < 300, + "expected notify-driven pickup well under the 1s fallback poll, took {elapsed:?}" + ); +} + +#[test] +fn shutdown_returns_promptly_even_when_workers_are_idle() { + let q = test_queue(); + let mut pool = WorkerPool::new(q); + pool.start(2); + + let start = std::time::Instant::now(); + pool.shutdown(); + let elapsed = start.elapsed(); + + // Idle workers are parked in wait_for_work (1s timeout); shutdown must + // wake them via wake_workers rather than waiting out that timeout. + assert!( + elapsed.as_millis() < 300, + "expected shutdown to wake idle workers immediately, took {elapsed:?}" + ); +} diff --git a/dashboard/web/board.html b/dashboard/web/board.html index 4d043248..d208203b 100644 --- a/dashboard/web/board.html +++ b/dashboard/web/board.html @@ -53,6 +53,7 @@ List Webhooks Claims + Jobs Cost diff --git a/dashboard/web/claims.html b/dashboard/web/claims.html index a4a12826..9073830c 100644 --- a/dashboard/web/claims.html +++ b/dashboard/web/claims.html @@ -53,6 +53,7 @@ List Webhooks Claims + Jobs Cost diff --git a/dashboard/web/cost.html b/dashboard/web/cost.html index 49ba1a62..e31444e7 100644 --- a/dashboard/web/cost.html +++ b/dashboard/web/cost.html @@ -53,6 +53,7 @@ List Webhooks Claims + Jobs Cost diff --git a/dashboard/web/item.html b/dashboard/web/item.html index eeff1555..4d489148 100644 --- a/dashboard/web/item.html +++ b/dashboard/web/item.html @@ -53,6 +53,7 @@ List Webhooks Claims + Jobs Cost diff --git a/dashboard/web/jobs.html b/dashboard/web/jobs.html new file mode 100644 index 00000000..f878b47e --- /dev/null +++ b/dashboard/web/jobs.html @@ -0,0 +1,289 @@ + + + + + + + agentflare dashboard — jobs + + + + + +
+ + +
+
+

Jobs

+

Job

+ + + + + +
+
+ + + + +
+ + + + + + + + + + + + + + +
IDCommandStateCreatedStartedFinished
+
+ + + +
+
+
+ + + + + + diff --git a/dashboard/web/list.html b/dashboard/web/list.html index 9ee29b3a..52257237 100644 --- a/dashboard/web/list.html +++ b/dashboard/web/list.html @@ -53,6 +53,7 @@ List Webhooks Claims + Jobs Cost diff --git a/dashboard/web/shell.html b/dashboard/web/shell.html index d0e16ebc..ff6770ef 100644 --- a/dashboard/web/shell.html +++ b/dashboard/web/shell.html @@ -61,6 +61,7 @@ List Webhooks Claims + Jobs Cost diff --git a/dashboard/web/webhooks.html b/dashboard/web/webhooks.html index 10f2cf9f..7d2929f5 100644 --- a/dashboard/web/webhooks.html +++ b/dashboard/web/webhooks.html @@ -53,6 +53,7 @@ List Webhooks Claims + Jobs Cost diff --git a/src/cli/serve.rs b/src/cli/serve.rs index 8c51150d..b00e7c44 100644 --- a/src/cli/serve.rs +++ b/src/cli/serve.rs @@ -16,10 +16,39 @@ pub struct ServeArgs { /// Required whenever --host is not 127.0.0.1/localhost/::1. #[arg(long)] pub yes_expose: bool, + /// Internal: marks this invocation as the daemon-managed foreground + /// process (spawned by `agentflare daemon start` or the installed + /// systemd/launchd unit). Not meant for direct use; the singleton check + /// and pid registration below apply either way. + #[arg(long = "_foreground-daemon", hide = true)] + pub foreground_daemon: bool, } impl ServeArgs { pub fn run(self) { + // Scoped so the lock is released before `dashboard::serve` (which + // never returns in normal operation) rather than held for the + // server's whole lifetime. + { + let guard = crate::daemon::acquire_singleton_lock(); + if let Err(ref e) = guard { + eprintln!( + "warning: failed to acquire daemon singleton lock ({e}); proceeding without race protection against a concurrent `serve` invocation." + ); + } + if let Some(pid) = crate::daemon::is_daemon_running() { + eprintln!("agentflare dashboard is already running (pid {pid})."); + eprintln!( + "stop it first with `agentflare daemon stop`, or use the running instance instead of starting another." + ); + std::process::exit(1); + } + if let Err(e) = crate::daemon::write_pid_file() { + eprintln!( + "warning: failed to record daemon pid ({e}); `agentflare daemon status`/`stop` won't see this instance." + ); + } + } crate::dashboard::serve(&self.host, self.port, self.open, self.yes_expose); } } diff --git a/src/daemon.rs b/src/daemon.rs index bdd38ad6..81811df8 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -16,6 +16,12 @@ pub fn daemon_start_lock_path() -> PathBuf { .unwrap_or_else(|| std::env::temp_dir().join("agentflare-daemon.start.lock")) } +fn daemon_singleton_lock_path() -> PathBuf { + dirs::runtime_dir() + .map(|d| d.join("agentflare").join("daemon.singleton.lock")) + .unwrap_or_else(|| std::env::temp_dir().join("agentflare-daemon.singleton.lock")) +} + pub fn cleanup_daemon_files() { let pid_path = daemon_pid_path(); let _ = std::fs::remove_file(&pid_path); @@ -37,7 +43,25 @@ impl Drop for LockGuard { } pub fn acquire_start_lock() -> Result { - let lock_path = daemon_start_lock_path(); + acquire_lock(daemon_start_lock_path(), Duration::from_secs(5)) +} + +/// Serializes the is_daemon_running-check + write_pid_file critical section +/// in `ServeArgs::run()`, so two `agentflare serve` invocations racing each +/// other (e.g. on different `--port`s, started within moments of each other) +/// can't both pass the check before either has written its pid file. +/// +/// Deliberately a *different* lock file from `daemon_start_lock_path()`: +/// `start_daemon()` holds that lock for its whole ~5s spawn-and-poll window, +/// and the process it spawns calls back into this same check via +/// `serve --_foreground-daemon` — sharing one lock would deadlock the two +/// (parent waiting on the child's pid file while holding the lock the child +/// needs to write it). +pub fn acquire_singleton_lock() -> Result { + acquire_lock(daemon_singleton_lock_path(), Duration::from_secs(2)) +} + +fn acquire_lock(lock_path: PathBuf, timeout: Duration) -> Result { if let Some(parent) = lock_path.parent() { std::fs::create_dir_all(parent).map_err(|e| format!("create lock dir {parent:?}: {e}"))?; } @@ -47,7 +71,7 @@ pub fn acquire_start_lock() -> Result { .truncate(false) .open(&lock_path) .map_err(|e| format!("open lock file: {e}"))?; - let deadline = std::time::Instant::now() + Duration::from_secs(5); + let deadline = std::time::Instant::now() + timeout; loop { if file.try_lock_exclusive().is_ok() { return Ok(LockGuard { @@ -56,9 +80,9 @@ pub fn acquire_start_lock() -> Result { }); } if std::time::Instant::now() >= deadline { - return Err("could not acquire daemon start lock within 5s".to_string()); + return Err(format!("could not acquire lock within {timeout:?}")); } - std::thread::sleep(Duration::from_millis(100)); + std::thread::sleep(Duration::from_millis(50)); } } @@ -77,6 +101,19 @@ fn read_pid_from_file() -> Option { content.trim().parse::().ok() } +/// Records the calling process's own pid as the running daemon, so a later +/// `is_daemon_running`/`daemon status`/`daemon stop` (possibly from a +/// different process) can find it. Called by the `serve` foreground process +/// itself once it starts, not just by `start_daemon`'s spawner. +pub fn write_pid_file() -> Result<(), String> { + let path = daemon_pid_path(); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|e| format!("create pid dir {parent:?}: {e}"))?; + } + std::fs::write(&path, std::process::id().to_string()) + .map_err(|e| format!("write pid file: {e}")) +} + pub fn start_daemon() -> Result { let _guard = acquire_start_lock()?; @@ -85,7 +122,13 @@ pub fn start_daemon() -> Result { } let binary = std::env::current_exe().map_err(|e| format!("current_exe: {e}"))?; - let _pid = process::spawn_detached(&binary.to_string_lossy(), &["--_foreground-daemon"])?; + // Must match the `ExecStart`/`ProgramArguments` invocation the installed + // systemd/launchd units use (see `daemon_autostart.rs`) — both spawn + // `serve --_foreground-daemon`, not just the bare flag. + let _pid = process::spawn_detached( + &binary.to_string_lossy(), + &["serve", "--_foreground-daemon"], + )?; for _ in 0..20 { std::thread::sleep(Duration::from_millis(250)); diff --git a/src/dashboard/server.rs b/src/dashboard/server.rs index 9b03c150..f99be454 100644 --- a/src/dashboard/server.rs +++ b/src/dashboard/server.rs @@ -1,13 +1,16 @@ +use agentflare_jobs::{AgentJob, JobState, Queue}; use axum::response::sse::{Event, KeepAlive, Sse}; use axum::{ - Router, - extract::Query, + Json, Router, + extract::{Path, Query, State}, http::{StatusCode, Uri, header}, response::{IntoResponse, Response}, routing::get, }; use rust_embed::RustEmbed; use serde::Deserialize; +use std::io::Read; +use std::path::PathBuf; #[derive(RustEmbed)] #[folder = "dashboard/web/"] @@ -133,6 +136,35 @@ async fn cost_handler(Query(q): Query) -> Response { const TICK: std::time::Duration = std::time::Duration::from_secs(3); const COST_REFRESH: std::time::Duration = std::time::Duration::from_secs(30); +/// How often to sweep finished jobs (and their stdout/stderr log files) out +/// of the queue, and how old a finished job must be before it's eligible. +/// Without this, `agent_jobs` rows and job-logs/ files accumulate forever — +/// nothing else ever calls `Queue::cleanup`. +const JOB_CLEANUP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3600); +const JOB_RETENTION_SECS: i64 = 7 * 24 * 3600; + +/// Runs for the lifetime of the process (like `snapshot_broadcaster`'s +/// producer task): wakes on `JOB_CLEANUP_INTERVAL`, deletes finished jobs +/// older than `JOB_RETENTION_SECS`. The blocking SQLite + filesystem work +/// happens off the async worker threads via `spawn_blocking`. +fn spawn_job_cleanup(queue: agentflare_jobs::Queue) { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(JOB_CLEANUP_INTERVAL); + loop { + ticker.tick().await; + let queue = queue.clone(); + let result = + tokio::task::spawn_blocking(move || queue.cleanup(JOB_RETENTION_SECS)).await; + match result { + Ok(Ok(0)) => {} + Ok(Ok(deleted)) => eprintln!("agentflare-jobs: cleaned up {deleted} old job(s)"), + Ok(Err(e)) => eprintln!("agentflare-jobs: cleanup failed: {e}"), + Err(e) => eprintln!("agentflare-jobs: cleanup task panicked: {e}"), + } + } + }); +} + /// Single shared broadcast of the live `{ claims, cost_today }` snapshot. Every /// `/events` client subscribes to this one channel, so there is no per-client /// work — the producer below runs at most one refresh cycle per `TICK`, @@ -225,7 +257,259 @@ fn mime_for(p: &str) -> &'static str { } } -pub fn router() -> Router { +/// Body for `POST /api/jobs`. Only `command` is required; everything else +/// falls back to `AgentJob::new`'s defaults (300s timeout, 3 retries). +#[derive(Deserialize)] +struct SubmitJobRequest { + command: String, + #[serde(default)] + args: Vec, + #[serde(default)] + env: Vec<(String, String)>, + cwd: Option, + timeout_secs: Option, +} + +async fn submit_job_handler( + State(queue): State, + Json(req): Json, +) -> Response { + if req.command.trim().is_empty() { + return (StatusCode::BAD_REQUEST, "command must not be empty").into_response(); + } + let mut job = AgentJob::new(req.command).args(req.args); + for (k, v) in req.env { + job = job.env(k, v); + } + if let Some(cwd) = req.cwd { + job = job.cwd(cwd); + } + if let Some(secs) = req.timeout_secs { + job = job.timeout(secs); + } + match queue.enqueue(&job) { + Ok(info) => (StatusCode::CREATED, Json(info)).into_response(), + Err(e) => ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("failed to enqueue job: {e}"), + ) + .into_response(), + } +} + +#[derive(Deserialize)] +struct JobsQuery { + id: Option, + state: Option, +} + +/// `GET /api/jobs?id=` fetches one job; `GET /api/jobs[?state=]` +/// lists the most recent 100, optionally filtered by state. +/// +/// `queue.get`/`queue.list` hit SQLite synchronously, so both go through +/// `spawn_blocking` rather than running inline on the async request path — +/// matching the SSE handlers below, which do the same for identical calls. +async fn jobs_handler(State(queue): State, Query(q): Query) -> Response { + if let Some(id) = q.id { + let got = tokio::task::spawn_blocking({ + let queue = queue.clone(); + move || queue.get(&id) + }) + .await; + return match got { + Ok(Ok(info)) => Json(info).into_response(), + Ok(Err(_)) => (StatusCode::NOT_FOUND, "job not found").into_response(), + Err(e) => ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("job lookup task failed: {e}"), + ) + .into_response(), + }; + } + let state_filter = match q.state.as_deref() { + None => None, + Some("queued") => Some(JobState::Queued), + Some("running") => Some(JobState::Running), + Some("exited") => Some(JobState::Exited), + Some("killed") => Some(JobState::Killed), + Some("failed") => Some(JobState::Failed), + Some(other) => { + return ( + StatusCode::BAD_REQUEST, + format!("invalid `state` value {other:?}"), + ) + .into_response(); + } + }; + let listed = tokio::task::spawn_blocking(move || queue.list(state_filter)).await; + match listed { + Ok(Ok(jobs)) => Json(jobs).into_response(), + Ok(Err(e)) => ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("failed to list jobs: {e}"), + ) + .into_response(), + Err(e) => ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("job list task failed: {e}"), + ) + .into_response(), + } +} + +/// SSE stream of the live job list. Each subscriber independently polls +/// `queue.list(None)` on the same `TICK` cadence as the existing event +/// stream — cheap enough (SQLite LIMIT 100) that a shared broadcaster is +/// overkill for typical dashboard usage (1-2 tabs). +async fn jobs_events_handler( + State(queue): State, +) -> Sse>> { + let (tx, rx) = + tokio::sync::mpsc::unbounded_channel::>(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(TICK); + loop { + ticker.tick().await; + let listed = tokio::task::spawn_blocking({ + let queue = queue.clone(); + move || queue.list(None) + }) + .await; + let list = match listed { + Ok(Ok(list)) => list, + Ok(Err(e)) => { + eprintln!("[dashboard/server] jobs_events: list failed: {e}"); + vec![] + } + Err(e) => { + eprintln!("[dashboard/server] jobs_events: task failed: {e}"); + vec![] + } + }; + let json = serde_json::to_string(&list).unwrap_or_else(|_| "[]".to_string()); + if tx.send(Ok(Event::default().data(json))).is_err() { + break; + } + } + }); + Sse::new(tokio_stream::wrappers::UnboundedReceiverStream::new(rx)) + .keep_alive(KeepAlive::default()) +} + +/// `GET /api/jobs/:id/stream` — SSE stream that tails the job's stdout log +/// file incrementally. +/// +/// The stdout path is derived directly from `queue.log_dir()` and the job's +/// own id (`Supervisor` names its log files `{id}.stdout`/`{id}.stderr` for +/// exactly this reason) rather than read from `JobInfo.output`, which is +/// only populated once the job reaches a terminal state — waiting on it +/// would mean this stream never emits anything until the job is already +/// finished, defeating the point of a *live* tail. +/// +/// While the job is queued or running, we poll every 300ms: check the +/// current job state (`queue.get(id)`), read any new bytes from the stdout +/// file (which may not exist yet if the job is still queued), and push them +/// as SSE `data:` events. Once the job reaches a terminal state +/// (exited/failed/killed), any remaining buffered content is flushed, a +/// final `event: done\ndata:` frame is sent, and the stream closes. +async fn jobs_stream_handler(State(queue): State, Path(id): Path) -> Response { + // 404 on unknown id before spawning anything. + let exists = tokio::task::spawn_blocking({ + let queue = queue.clone(); + let id = id.clone(); + move || queue.get(&id).is_ok() + }) + .await + .unwrap_or(false); + if !exists { + return (StatusCode::NOT_FOUND, "job not found").into_response(); + } + let stdout_path = queue.log_dir().join(format!("{id}.stdout")); + let (tx, rx) = + tokio::sync::mpsc::unbounded_channel::>(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_millis(300)); + let mut offset: u64 = 0; + let mut file: Option = None; + loop { + interval.tick().await; + let info = match tokio::task::spawn_blocking({ + let queue = queue.clone(); + let id = id.clone(); + move || queue.get(&id) + }) + .await + { + Ok(Ok(info)) => info, + _ => { + let _ = tx.send(Ok(Event::default().data("").event("done"))); + break; + } + }; + let terminal = info.state.is_terminal(); + if file.is_none() { + if stdout_path.exists() { + match std::fs::File::open(&stdout_path) { + Ok(f) => file = Some(f), + Err(_) => { + if terminal { + let _ = tx.send(Ok(Event::default().data("").event("done"))); + } + break; + } + } + } + if file.is_none() { + if terminal { + let _ = tx.send(Ok(Event::default().data("").event("done"))); + break; + } + continue; + } + } + if let Some(ref mut f) = file { + use std::io::Seek as _; + let _ = f.seek(std::io::SeekFrom::Start(offset)); + let mut buf = Vec::new(); + match f.read_to_end(&mut buf) { + Ok(n) if n > 0 => { + offset += n as u64; + // Lossy, not strict: `buf` is a raw byte window that + // can split a multi-byte UTF-8 char across polls, and + // `offset` has already moved past it — a strict parse + // would silently and permanently drop those bytes + // from the live view instead of showing a stray + // replacement char at the boundary. + let text = String::from_utf8_lossy(&buf).into_owned(); + if tx.send(Ok(Event::default().data(text))).is_err() { + break; + } + } + Ok(_) => {} + Err(_) => {} + } + } + if terminal { + let _ = tx.send(Ok(Event::default().data("").event("done"))); + break; + } + } + }); + Sse::new(tokio_stream::wrappers::UnboundedReceiverStream::new(rx)) + .keep_alive(KeepAlive::default()) + .into_response() +} + +/// Submit/fetch/stream endpoints as their own state-scoped sub-router. +fn jobs_router(queue: Queue) -> Router { + Router::new() + .route("/api/jobs", get(jobs_handler).post(submit_job_handler)) + .route("/api/jobs/events", get(jobs_events_handler)) + .route("/api/jobs/{id}/stream", get(jobs_stream_handler)) + .with_state(queue) +} + +pub fn router(queue: Queue) -> Router { Router::new() .route("/api/claims", get(claims_handler)) .route("/api/pm/workspaces", get(pm_workspaces_handler)) @@ -237,6 +521,7 @@ pub fn router() -> Router { .route("/api/webhooks", get(webhooks_handler)) .route("/api/cost", get(cost_handler)) .route("/events", get(events_handler)) + .merge(jobs_router(queue)) .nest("/artifacts", super::artifacts::router()) .merge(flare_proxy::router()) .fallback(static_handler) @@ -254,6 +539,25 @@ pub async fn run(host: &str, port: u16, open: bool, yes_expose: bool) { eprintln!("pass --yes-expose to bind anyway (trusted networks only)."); std::process::exit(1); } + // `agent_jobs` is relational state, so it lives in the single + // source-of-truth `agentflare.db` (see `db.rs`'s header comment) rather + // than a new file — this just opens its own `Connection` to that same + // path and runs its own (non-overlapping) migration for that one table. + // Job stdout/stderr logs are rebuildable artifacts, not DB state, so + // those still get their own directory. + let queue = Queue::open( + &crate::db::agentflare_db_path(), + crate::state::state_dir().join("job-logs"), + ) + .expect("failed to open job queue"); + // Kept alive for the rest of `run()` (which never returns in normal + // operation) so its worker threads keep polling the queue; there is no + // graceful in-process shutdown path today (see `daemon::stop_daemon`, + // which relies on SIGTERM/SIGKILL), so neither does this. + let mut worker_pool = agentflare_jobs::WorkerPool::new(queue.clone()); + worker_pool.start(2); + spawn_job_cleanup(queue.clone()); + let listener = tokio::net::TcpListener::bind((host, port)) .await .expect("failed to bind dashboard server"); @@ -266,7 +570,7 @@ pub async fn run(host: &str, port: u16, open: bool, yes_expose: bool) { if open { crate::dashboard::open_browser(&url); } - axum::serve(listener, router()) + axum::serve(listener, router(queue)) .await .expect("dashboard server error"); } @@ -275,6 +579,15 @@ pub async fn run(host: &str, port: u16, open: bool, yes_expose: bool) { mod tests { use super::*; + fn test_queue() -> Queue { + // `.keep()` so the dir outlives this function — otherwise the + // returned `Queue`'s `log_dir` would point at an already-deleted + // path (harmless for tests that never touch logs, but a footgun for + // ones that do). + let dir = tempfile::tempdir().unwrap().keep(); + Queue::open_memory(dir.join("logs")).unwrap() + } + #[tokio::test] async fn claims_endpoint_returns_json_array() { let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) @@ -282,7 +595,7 @@ mod tests { .unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { - axum::serve(listener, router()).await.unwrap(); + axum::serve(listener, router(test_queue())).await.unwrap(); }); let body = reqwest::get(format!("http://{addr}/api/claims")) .await @@ -293,6 +606,80 @@ mod tests { assert!(body.starts_with('['), "expected JSON array, got: {body}"); } + #[tokio::test] + async fn jobs_endpoint_round_trips_submit_and_fetch() { + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, router(test_queue())).await.unwrap(); + }); + + let (cmd, args): (&str, Vec<&str>) = if cfg!(windows) { + ("cmd", vec!["/c", "exit 0"]) + } else { + ("true", vec![]) + }; + let client = reqwest::Client::new(); + let submit_resp = client + .post(format!("http://{addr}/api/jobs")) + .json(&serde_json::json!({ "command": cmd, "args": args })) + .send() + .await + .unwrap(); + assert_eq!(submit_resp.status(), StatusCode::CREATED); + let submitted: serde_json::Value = submit_resp.json().await.unwrap(); + assert_eq!(submitted["state"], "queued"); + assert_eq!(submitted["command"], cmd); + let id = submitted["id"].as_str().unwrap().to_string(); + + let fetch_resp = reqwest::get(format!("http://{addr}/api/jobs?id={id}")) + .await + .unwrap(); + assert_eq!(fetch_resp.status(), StatusCode::OK); + let fetched: serde_json::Value = fetch_resp.json().await.unwrap(); + assert_eq!(fetched["id"], id); + assert_eq!( + fetched["command"], cmd, + "list/get should surface the job's command, not just its id" + ); + } + + #[tokio::test] + async fn jobs_endpoint_rejects_empty_command() { + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, router(test_queue())).await.unwrap(); + }); + let client = reqwest::Client::new(); + let resp = client + .post(format!("http://{addr}/api/jobs")) + .json(&serde_json::json!({ "command": "" })) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn jobs_endpoint_404s_unknown_id() { + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, router(test_queue())).await.unwrap(); + }); + let resp = reqwest::get(format!("http://{addr}/api/jobs?id=does-not-exist")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + } + #[test] fn is_local_bind_recognizes_loopback_only() { assert!(is_local_bind("127.0.0.1")); @@ -371,6 +758,202 @@ mod tests { assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } + #[tokio::test] + async fn jobs_events_endpoint_streams_job_list() { + use tokio_stream::StreamExt as _; + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let queue = test_queue(); + // Submit a job so the list isn't empty. + let job = agentflare_jobs::AgentJob::new("true"); + queue.enqueue(&job).unwrap(); + tokio::spawn(async move { + axum::serve(listener, router(queue)).await.unwrap(); + }); + let resp = reqwest::get(format!("http://{addr}/api/jobs/events")) + .await + .unwrap(); + let mut stream = resp.bytes_stream(); + let first = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next()) + .await + .expect("first SSE frame within 10s") + .expect("stream item") + .unwrap(); + let text = String::from_utf8(first.to_vec()).unwrap(); + let data_line = text + .lines() + .find_map(|l| l.strip_prefix("data: ")) + .expect("SSE data line"); + let v: serde_json::Value = serde_json::from_str(data_line).unwrap(); + assert!(v.is_array(), "expected JSON array, got: {v}"); + assert!( + !v.as_array().unwrap().is_empty(), + "expected at least one job" + ); + } + + #[tokio::test] + async fn jobs_stream_endpoint_404s_unknown_id() { + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, router(test_queue())).await.unwrap(); + }); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let resp = reqwest::get(format!("http://{addr}/api/jobs/does-not-exist/stream")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn jobs_stream_endpoint_streams_stdout_of_finished_job() { + use tokio_stream::StreamExt as _; + let dir = tempfile::tempdir().unwrap(); + let queue = Queue::open_memory(dir.path().join("logs")).unwrap(); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let job = agentflare_jobs::AgentJob::new("echo").arg("hello from job"); + let info = queue.enqueue(&job).unwrap(); + let id = info.id.clone(); + let log_dir = queue.log_dir().to_path_buf(); + std::fs::create_dir_all(&log_dir).unwrap(); + let stdout_path = log_dir.join(format!("{id}.stdout")); + std::fs::write(&stdout_path, b"hello from job\n").unwrap(); + let output = agentflare_jobs::JobOutput { + exit_code: Some(0), + timed_out: false, + stdout_path: stdout_path.clone(), + stderr_path: log_dir.join(format!("{id}.stderr")), + stdout_total_bytes: 16, + stderr_total_bytes: 0, + }; + queue.complete(&id, &output, true).unwrap(); + tokio::spawn(async move { + axum::serve(listener, router(queue)).await.unwrap(); + }); + // Give the server a moment to start. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let resp = reqwest::get(format!("http://{addr}/api/jobs/{id}/stream")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let mut stream = resp.bytes_stream(); + // Collect all SSE frames with a timeout. + let mut all_text = String::new(); + loop { + match tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()).await { + Ok(Some(Ok(chunk))) => { + all_text.push_str(&String::from_utf8_lossy(&chunk)); + if all_text.contains("event: done") { + break; + } + } + Ok(Some(Err(_))) => break, + Ok(None) => break, + Err(_) => break, + } + } + assert!( + all_text.contains("hello from job"), + "expected stdout content in stream, got: {all_text}" + ); + } + + #[tokio::test] + async fn jobs_stream_endpoint_tails_output_of_a_still_running_job() { + // Regression: the stream used to key off `JobInfo.output`, which is + // only populated once the job is already terminal — so it emitted + // nothing at all until the job finished, then dumped everything in + // one lump. This drives a real job through a real `WorkerPool` and + // asserts the first chunk arrives (and the job is still `running`) + // well before the job's own sleep would let it finish. + use tokio_stream::StreamExt as _; + let dir = tempfile::tempdir().unwrap(); + let queue = Queue::open_memory(dir.path().join("logs")).unwrap(); + let mut pool = agentflare_jobs::WorkerPool::new(queue.clone()); + pool.start(1); + + let (cmd, args): (&str, Vec<&str>) = if cfg!(windows) { + // `timeout /t` refuses to run with redirected stdin ("INPUT + // REDIRECTION IS NOT SUPPORTED") and exits instantly instead of + // sleeping — `ping` against loopback is the standard + // redirection-safe stand-in for a ~2s delay on Windows. + ( + "cmd", + vec!["/c", "echo tick1 & ping -n 3 127.0.0.1 >nul & echo tick2"], + ) + } else { + ("sh", vec!["-c", "echo tick1; sleep 2; echo tick2"]) + }; + let info = queue + .enqueue(&agentflare_jobs::AgentJob::new(cmd).args(args)) + .unwrap(); + let id = info.id.clone(); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let queue_for_router = queue.clone(); + tokio::spawn(async move { + axum::serve(listener, router(queue_for_router)) + .await + .unwrap(); + }); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let resp = reqwest::get(format!("http://{addr}/api/jobs/{id}/stream")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let mut stream = resp.bytes_stream(); + + let mut all_text = String::new(); + let mut state_at_first_tick: Option = None; + loop { + match tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()).await { + Ok(Some(Ok(chunk))) => { + all_text.push_str(&String::from_utf8_lossy(&chunk)); + if state_at_first_tick.is_none() && all_text.contains("tick1") { + let fetched: serde_json::Value = + reqwest::get(format!("http://{addr}/api/jobs?id={id}")) + .await + .unwrap() + .json() + .await + .unwrap(); + state_at_first_tick = + Some(fetched["state"].as_str().unwrap_or("").to_string()); + } + if all_text.contains("event: done") { + break; + } + } + Ok(Some(Err(_))) => break, + Ok(None) => break, + Err(_) => break, + } + } + pool.shutdown(); + + assert_eq!( + state_at_first_tick.as_deref(), + Some("running"), + "expected the job to still be running when the first chunk arrived, got: {state_at_first_tick:?}" + ); + assert!( + all_text.contains("tick1") && all_text.contains("tick2"), + "expected both ticks in stream, got: {all_text}" + ); + } + #[tokio::test] async fn events_endpoint_streams_claims_and_cost_snapshot() { use tokio_stream::StreamExt as _; @@ -379,7 +962,7 @@ mod tests { .unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { - axum::serve(listener, router()).await.unwrap(); + axum::serve(listener, router(test_queue())).await.unwrap(); }); let resp = reqwest::get(format!("http://{addr}/events")).await.unwrap(); let mut stream = resp.bytes_stream(); @@ -401,8 +984,31 @@ mod tests { ); } + // The PATH_LOCK guard below is intentionally held across `.await` + // points: it must stay held for this whole test so no other test can + // reset AGENTFLARE_HOME_OVERRIDE out from under it (a real race, not + // just a lint nit — see the comment at the lock acquisition). Safe here + // because `#[tokio::test]` gives each test its own current-thread + // runtime on its own OS thread, so no other task ever contends for this + // thread while the guard is held. + #[allow(clippy::await_holding_lock)] #[tokio::test] async fn run_serves_normally_on_local_bind_without_yes_expose() { + // `run()` now opens a real, writable job queue under + // `state::state_dir()` (`~/.agentflare` by default). Route it at a + // throwaway home for the duration of this test — same lock + + // env-var mechanism `paths::test_support::with_temp_home` uses, but + // inlined because that helper's guard is sync-only and would reset + // the override before the `.await`s below ever run. + let _guard = agent_registry::detect::PATH_LOCK + .lock() + .unwrap_or_else(|e| e.into_inner()); + let home_dir = tempfile::tempdir().unwrap(); + unsafe { + // SAFETY: PATH_LOCK serializes all env mutation in this test binary. + std::env::set_var("AGENTFLARE_HOME_OVERRIDE", home_dir.path()); + } + // Probe an ephemeral port, then hand that exact port to `run()` — // `run()` takes a fixed port rather than returning the bound address, // so this is the only way to know where to connect afterward. Only @@ -427,6 +1033,12 @@ mod tests { } tokio::time::sleep(std::time::Duration::from_millis(100)).await; } + + unsafe { + // SAFETY: still under PATH_LOCK. + std::env::remove_var("AGENTFLARE_HOME_OVERRIDE"); + } + assert!( started, "expected dashboard to serve on a local bind without --yes-expose"