From 3972b175e52d28160ff09f74147692092cfeb501 Mon Sep 17 00:00:00 2001 From: shiva Date: Sat, 8 Aug 2026 13:08:00 +0530 Subject: [PATCH 1/2] refactor(supervisor): dispatch work items in-process instead of re-invoking `agentflare work` Flattens the daemon's 3-layer dispatch (daemon -> agentflare-jobs::Supervisor -wrapped `agentflare work` OS subprocess -> agent CLI subprocess) to 2 layers: the daemon now runs a work item's claim/resolve-agent/build-prompt/done logic in-process, spawning a real subprocess only for the one genuinely open-ended part -- the agent CLI itself (still via agent_launch::run_captured, with its own hard-cap/idle-timeout kill from item #20). Extracted `run_work`'s body into `execute_work` (src/cli/work.rs), which now takes a `&mut dyn Write` for progress output instead of println!, so the same logic works identically for a human running `agentflare work` and for `WorkItemExecutor` running in-process. Progress lands in that job's own `{id}.stdout` log file -- the same path/name Supervisor already used for subprocess jobs -- so the dashboard's existing live log tail (`/api/jobs/:id /stream`) keeps working unchanged. agentflare-jobs gained an `InProcessExecutor` trait (defined there so the crate never depends upward on the binary that implements it) and an `AgentJob::in_process` flag; `WorkerPool` branches on it per job, keeping the existing subprocess path (used by ad-hoc `POST /api/jobs` submissions) fully intact and unchanged. An in-process job's executor call runs on a short-lived thread with `job.timeout_secs` as a watchdog, so a stuck claim/worktree/done step -- which loses the OS-level SIGKILL backstop a subprocess had -- can't wedge a pooled worker thread forever; the stuck thread itself is abandoned (no safe way to force-kill a thread in Rust) rather than truly killed, a real, deliberate trade-off called out where it's implemented. Two correctness issues surfaced by moving multiple work items into worker threads sharing one process, both fixed narrowly rather than serializing dispatch back down to one job at a time: - claim/comment/done authorship (`claims::owner_id()`) read the process-global `AGENTFLARE_AGENT` env var, which two worker threads dispatching different agents would race on. Added a thread-local override (`claims::with_owner_override`) that `owner_id()` checks first -- each worker thread's identity stays independent, no shared mutable state. `cli/handoff.rs`'s sender resolution duplicated the same env read separately; unified it through `claims::owner_id()` so it picks up the override too instead of needing its own fix. - the agent CLI subprocess itself relied on inheriting `AGENTFLARE_AGENT` from its parent's ambient env for `flare-git-shim`'s bypass classification when the agent shells out to git -- correct for a subprocess-per-dispatch caller, wrong for a daemon whose own ambient env can't be right for every concurrently-running job. `agent_launch::run_headless` now sets it explicitly on the spawned Command instead of relying on inheritance. `dispatch_item` (src/supervisor.rs) now enqueues an in-process job with `[item_id, agent]` args instead of `agentflare work --agent `; the outer job timeout (aligned to work's hard cap by the previous commit) still applies, now as the in-process watchdog rather than a subprocess kill. --- crates/agentflare-jobs/src/executor.rs | 16 ++ crates/agentflare-jobs/src/lib.rs | 2 + crates/agentflare-jobs/src/queue.rs | 6 +- crates/agentflare-jobs/src/types.rs | 19 ++ crates/agentflare-jobs/src/worker.rs | 112 ++++++++++- .../agentflare-jobs/tests/in_process_test.rs | 176 ++++++++++++++++++ src/agent_launch.rs | 9 + src/claims.rs | 98 ++++++++++ src/cli/handoff.rs | 16 +- src/cli/mod.rs | 2 +- src/cli/work.rs | 93 +++++++-- src/dashboard/server.rs | 3 +- src/supervisor.rs | 49 +++-- 13 files changed, 550 insertions(+), 51 deletions(-) create mode 100644 crates/agentflare-jobs/src/executor.rs create mode 100644 crates/agentflare-jobs/tests/in_process_test.rs diff --git a/crates/agentflare-jobs/src/executor.rs b/crates/agentflare-jobs/src/executor.rs new file mode 100644 index 00000000..75e3be40 --- /dev/null +++ b/crates/agentflare-jobs/src/executor.rs @@ -0,0 +1,16 @@ +/// Lets a worker run certain jobs by calling application code directly, +/// in-process, instead of spawning a fresh OS subprocess for `job.command` +/// (see `AgentJob::in_process`/`in_process()`). Defined in this crate, +/// implemented by the caller: this crate's `Queue`/`Supervisor`/`WorkerPool` +/// stay generic and never depend upward on whatever binary embeds them — the +/// binary implements this trait for its own job kind(s) and hands an +/// instance to `WorkerPool::with_executor`. +pub trait InProcessExecutor: Send + Sync { + /// Runs one job's `args`. `log` should receive the same kind of + /// progress output a subprocess would have written to stdout — a worker + /// writes it to that job's own `{id}.stdout` log file, the same file + /// name/location `Supervisor::spawn` uses for subprocess jobs, so + /// existing log-tailing (e.g. the dashboard's `/api/jobs/:id/stream`) + /// keeps working unchanged for in-process jobs too. + fn execute(&self, job_id: &str, args: &[String], log: &mut dyn std::io::Write) -> Result<(), String>; +} diff --git a/crates/agentflare-jobs/src/lib.rs b/crates/agentflare-jobs/src/lib.rs index 929b2922..6c1083ad 100644 --- a/crates/agentflare-jobs/src/lib.rs +++ b/crates/agentflare-jobs/src/lib.rs @@ -1,8 +1,10 @@ +pub mod executor; pub mod queue; pub mod supervisor; pub mod types; pub mod worker; +pub use executor::InProcessExecutor; pub use queue::Queue; pub use supervisor::Supervisor; pub use types::{AgentJob, JobInfo, JobOutput, JobState}; diff --git a/crates/agentflare-jobs/src/queue.rs b/crates/agentflare-jobs/src/queue.rs index 96fabdf5..289a6fa4 100644 --- a/crates/agentflare-jobs/src/queue.rs +++ b/crates/agentflare-jobs/src/queue.rs @@ -138,6 +138,7 @@ impl Queue { started_at: None, finished_at: None, output: None, + in_process: job.in_process, }) } @@ -328,8 +329,8 @@ fn map_job_row(r: &rusqlite::Row<'_>) -> rusqlite::Result { // `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)) + let (command, args, in_process) = serde_json::from_str::(&payload_json) + .map(|job| (job.command, job.args, job.in_process)) .unwrap_or_default(); Ok(JobInfo { id: r.get(0)?, @@ -356,6 +357,7 @@ fn map_job_row(r: &rusqlite::Row<'_>) -> rusqlite::Result { stdout_total_bytes: stdout_bytes as u64, stderr_total_bytes: stderr_bytes as u64, }), + in_process, }) } diff --git a/crates/agentflare-jobs/src/types.rs b/crates/agentflare-jobs/src/types.rs index a057a49a..42af7eaa 100644 --- a/crates/agentflare-jobs/src/types.rs +++ b/crates/agentflare-jobs/src/types.rs @@ -31,6 +31,15 @@ pub struct AgentJob { pub agent_type: Option, pub prompt: Option, pub metadata: HashMap, + /// When true, a worker runs this job by calling the registered + /// `InProcessExecutor` with `args` instead of spawning `command` as an + /// OS subprocess — `command`/`env`/`cwd`/`kill_after_secs` are then + /// unused (subprocess-only) but kept for display/back-compat. Defaults + /// to false via `#[serde(default)]` so every job persisted before this + /// field existed, and every plain `POST /api/jobs` submission (which + /// never sets it), keeps spawning a real subprocess exactly as before. + #[serde(default)] + pub in_process: bool, } impl AgentJob { @@ -47,9 +56,18 @@ impl AgentJob { agent_type: None, prompt: None, metadata: HashMap::new(), + in_process: false, } } + /// Marks this job to run via the daemon's registered `InProcessExecutor` + /// instead of spawning `command` as a subprocess — see the `in_process` + /// field's doc comment. + pub fn in_process(mut self) -> Self { + self.in_process = true; + self + } + pub fn arg(mut self, arg: impl Into) -> Self { self.args.push(arg.into()); self @@ -114,4 +132,5 @@ pub struct JobInfo { pub started_at: Option, pub finished_at: Option, pub output: Option, + pub in_process: bool, } diff --git a/crates/agentflare-jobs/src/worker.rs b/crates/agentflare-jobs/src/worker.rs index 1a81ea1e..40fb6d06 100644 --- a/crates/agentflare-jobs/src/worker.rs +++ b/crates/agentflare-jobs/src/worker.rs @@ -1,5 +1,7 @@ +use crate::executor::InProcessExecutor; use crate::queue::Queue; use crate::supervisor::Supervisor; +use crate::types::JobOutput; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::JoinHandle; @@ -9,6 +11,7 @@ pub struct WorkerPool { queue: Arc, handles: Vec>, running: Arc, + executor: Option>, } impl WorkerPool { @@ -17,16 +20,29 @@ impl WorkerPool { queue: Arc::new(queue), handles: vec![], running: Arc::new(AtomicBool::new(false)), + executor: None, } } + /// Registers the executor `start`'s workers use for jobs marked + /// `AgentJob::in_process` — see `InProcessExecutor`'s doc comment. A job + /// marked `in_process` with no executor registered fails immediately + /// (see `worker_loop`) rather than falling back to spawning `command` as + /// a subprocess, so a caller that forgets to register one gets a loud, + /// per-job failure instead of a silent behavior change. + pub fn with_executor(mut self, executor: Arc) -> Self { + self.executor = Some(executor); + self + } + pub fn start(&mut self, num_workers: usize) { self.running.store(true, Ordering::SeqCst); for _ in 0..num_workers { let queue = self.queue.clone(); let running = self.running.clone(); + let executor = self.executor.clone(); self.handles.push(std::thread::spawn(move || { - worker_loop(&queue, &running); + worker_loop(&queue, &running, executor.as_ref()); })); } } @@ -58,9 +74,12 @@ impl WorkerPool { } } -fn worker_loop(queue: &Queue, running: &AtomicBool) { +fn worker_loop(queue: &Queue, running: &AtomicBool, executor: Option<&Arc>) { while running.load(Ordering::SeqCst) { match queue.dequeue() { + Ok(Some((id, job))) if job.in_process => { + run_in_process(queue, &id, &job, executor); + } Ok(Some((id, job))) => { let mut sup = Supervisor::new( id.clone(), @@ -98,3 +117,92 @@ fn worker_loop(queue: &Queue, running: &AtomicBool) { } } } + +/// Runs one `in_process` job via `executor`, writing its progress to the +/// same `{id}.stdout` log-file path `Supervisor::spawn` uses for subprocess +/// jobs (see `InProcessExecutor`'s doc comment), then records the outcome +/// through `queue.complete`/`queue.fail` exactly as the subprocess path does +/// — so persisted state, retries, and the dashboard's job API/SSE all work +/// identically regardless of which way a given job actually ran. +/// +/// The executor call itself runs on a fresh, short-lived thread rather than +/// this (long-lived, pooled) one, with `job.timeout_secs` as a watchdog: an +/// in-process job has no OS-level SIGKILL backstop the way a subprocess does +/// (a stuck claim/worktree/done step can't be force-killed), so without this +/// a hang there would wedge one of the pool's worker threads forever. On +/// timeout the job is marked failed and this worker moves on to the next +/// queued job; the stuck thread itself is abandoned (there is no safe way to +/// force-kill a thread in Rust) rather than actually terminated — a real, +/// deliberate trade-off against the OS-level guarantee a subprocess gets, +/// not a bug. The one genuinely open-ended part of a work item -- the agent +/// CLI itself -- is still a real subprocess under `agent_launch::run_captured` +/// with its own hard-cap/idle-timeout kill, unaffected by any of this. +fn run_in_process( + queue: &Queue, + id: &str, + job: &crate::types::AgentJob, + executor: Option<&Arc>, +) { + let Some(executor) = executor else { + if let Err(e) = queue.fail(id, "job is marked in_process but no InProcessExecutor is registered on this WorkerPool") { + eprintln!("agentflare-jobs: failed to record failure for {id}: {e}"); + } + return; + }; + let _ = std::fs::create_dir_all(queue.log_dir()); + let stdout_path = queue.log_dir().join(format!("{id}.stdout")); + let stderr_path = queue.log_dir().join(format!("{id}.stderr")); + let mut log_file = match std::fs::File::create(&stdout_path) { + Ok(f) => f, + Err(e) => { + if let Err(qe) = queue.fail(id, &format!("failed to open job log file: {e}")) { + eprintln!("agentflare-jobs: failed to record failure for {id}: {qe}"); + } + return; + } + }; + + let (tx, rx) = std::sync::mpsc::channel::>(); + let executor = executor.clone(); + let job_id = id.to_string(); + let args = job.args.clone(); + std::thread::spawn(move || { + let result = executor.execute(&job_id, &args, &mut log_file); + let _ = tx.send(result); + }); + + let outcome = rx.recv_timeout(Duration::from_secs(job.timeout_secs.max(1))); + match outcome { + Ok(Ok(())) => { + let stdout_total_bytes = std::fs::metadata(&stdout_path).map(|m| m.len()).unwrap_or(0); + let output = JobOutput { + exit_code: Some(0), + timed_out: false, + stdout_path, + stderr_path, + stdout_total_bytes, + stderr_total_bytes: 0, + }; + if let Err(e) = queue.complete(id, &output, true) { + eprintln!("agentflare-jobs: failed to complete job {id}: {e}"); + } + } + Ok(Err(msg)) => { + if let Err(e) = queue.fail(id, &msg) { + eprintln!("agentflare-jobs: failed to record failure for {id}: {e}"); + } + } + Err(_) => { + let msg = format!( + "in-process job exceeded its {}s timeout and was abandoned \ + (a coordination step may be stuck — the agent CLI subprocess \ + itself has its own separate hard-cap/idle-timeout and is not \ + what this timeout measures)", + job.timeout_secs + ); + if let Err(e) = queue.fail(id, &msg) { + eprintln!("agentflare-jobs: failed to record failure for {id}: {e}"); + } + } + } +} diff --git a/crates/agentflare-jobs/tests/in_process_test.rs b/crates/agentflare-jobs/tests/in_process_test.rs new file mode 100644 index 00000000..74475920 --- /dev/null +++ b/crates/agentflare-jobs/tests/in_process_test.rs @@ -0,0 +1,176 @@ +//! Item #19: jobs marked `AgentJob::in_process` run via a registered +//! `InProcessExecutor` instead of spawning `command` as a subprocess. These +//! prove the executor dispatch, log-file capture (dashboard tail parity), +//! failure-message propagation, and the stuck-job watchdog all work — +//! separate from `worker_test.rs`'s existing subprocess-path coverage, which +//! stays unchanged and passing to prove that path is untouched. + +use agentflare_jobs::{AgentJob, InProcessExecutor, JobInfo, JobState, Queue, WorkerPool}; +use std::sync::Arc; + +fn test_queue() -> Queue { + let dir = tempfile::tempdir().unwrap(); + Queue::open_memory(dir.path().join("logs")).unwrap() +} + +fn wait_for_terminal(q: &Queue, id: &str, attempts: usize) -> JobInfo { + for _ in 0..attempts { + let i = q.get(id).unwrap(); + if matches!( + i.state, + JobState::Exited | JobState::Failed | JobState::Killed + ) { + return i; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + panic!("job {id} did not reach a terminal state in time"); +} + +struct EchoExecutor; +impl InProcessExecutor for EchoExecutor { + fn execute( + &self, + job_id: &str, + args: &[String], + log: &mut dyn std::io::Write, + ) -> Result<(), String> { + let _ = writeln!(log, "running job {job_id} with args {args:?}"); + Ok(()) + } +} + +#[test] +fn in_process_job_completes_via_the_registered_executor_and_writes_its_log() { + let q = test_queue(); + let mut pool = WorkerPool::new(q.clone()).with_executor(Arc::new(EchoExecutor)); + pool.start(1); + + let info = q + .enqueue( + &AgentJob::new("label-only") + .args(["a".to_string(), "b".to_string()]) + .in_process(), + ) + .unwrap(); + assert!(info.in_process, "enqueue should echo the in_process flag back"); + + let final_info = wait_for_terminal(&q, &info.id, 200); + pool.shutdown(); + + assert_eq!(final_info.state, JobState::Exited); + assert!(final_info.in_process); + let output = final_info.output.expect("a completed job has output"); + assert_eq!(output.exit_code, Some(0)); + assert!(!output.timed_out); + // The dashboard's live log tail reads exactly this path for subprocess + // jobs (Supervisor names its files the same way) — in-process jobs must + // land their progress output at the identical path for that to keep + // working unchanged. + let log = std::fs::read_to_string(&output.stdout_path).unwrap(); + assert!(log.contains(&format!("running job {} with args", info.id))); +} + +struct FailingExecutor; +impl InProcessExecutor for FailingExecutor { + fn execute( + &self, + _job_id: &str, + _args: &[String], + _log: &mut dyn std::io::Write, + ) -> Result<(), String> { + Err("deliberate failure".to_string()) + } +} + +#[test] +fn in_process_job_failure_is_recorded_with_the_executors_own_message() { + let q = test_queue(); + let mut pool = WorkerPool::new(q.clone()).with_executor(Arc::new(FailingExecutor)); + pool.start(1); + + let info = q + .enqueue(&AgentJob::new("label-only").in_process().max_retries(0)) + .unwrap(); + + let final_info = wait_for_terminal(&q, &info.id, 200); + pool.shutdown(); + + assert_eq!(final_info.state, JobState::Failed); + assert_eq!(final_info.error.as_deref(), Some("deliberate failure")); +} + +#[test] +fn in_process_job_fails_fast_when_no_executor_is_registered() { + let q = test_queue(); + let mut pool = WorkerPool::new(q.clone()); // no .with_executor + pool.start(1); + + let info = q + .enqueue(&AgentJob::new("label-only").in_process().max_retries(0)) + .unwrap(); + + let final_info = wait_for_terminal(&q, &info.id, 200); + pool.shutdown(); + + assert_eq!(final_info.state, JobState::Failed); + assert!( + final_info + .error + .as_deref() + .unwrap_or_default() + .contains("no InProcessExecutor"), + "got: {:?}", + final_info.error + ); +} + +struct SlowExecutor; +impl InProcessExecutor for SlowExecutor { + fn execute( + &self, + _job_id: &str, + _args: &[String], + _log: &mut dyn std::io::Write, + ) -> Result<(), String> { + std::thread::sleep(std::time::Duration::from_secs(5)); + Ok(()) + } +} + +// An in-process job has no OS-level SIGKILL backstop the way a subprocess +// does, so a stuck coordination step (the trade-off item #19 explicitly +// weighs) must not wedge the worker pool forever -- `job.timeout_secs` acts +// as a watchdog that abandons the stuck executor call and lets the pool move +// on, at the cost of the stuck thread itself leaking rather than actually +// being killed (there is no safe way to force-kill a thread in Rust). +#[test] +fn in_process_job_that_hangs_is_abandoned_at_its_timeout_instead_of_wedging_the_worker() { + let q = test_queue(); + let mut pool = WorkerPool::new(q.clone()).with_executor(Arc::new(SlowExecutor)); + pool.start(1); + + let start = std::time::Instant::now(); + let info = q + .enqueue(&AgentJob::new("label-only").in_process().timeout(1).max_retries(0)) + .unwrap(); + + let final_info = wait_for_terminal(&q, &info.id, 400); + let elapsed = start.elapsed(); + pool.shutdown(); + + assert_eq!(final_info.state, JobState::Failed); + assert!( + final_info + .error + .as_deref() + .unwrap_or_default() + .contains("abandoned"), + "got: {:?}", + final_info.error + ); + assert!( + elapsed < std::time::Duration::from_secs(3), + "should fail at the 1s watchdog, not wait out the 5s executor call, took {elapsed:?}" + ); +} diff --git a/src/agent_launch.rs b/src/agent_launch.rs index 96bacd00..5d3904d8 100644 --- a/src/agent_launch.rs +++ b/src/agent_launch.rs @@ -352,6 +352,15 @@ pub fn run_headless( // See the matching strip in `run_launch_env` above (item #139) — same // rationale applies to headless child processes. cmd.env_remove("CARGO_TARGET_DIR"); + // Explicit, not inherited from this process's own ambient env: when the + // agent shells out to `git`, the `flare-git-shim` on its PATH classifies + // bypass eligibility by `AGENTFLARE_AGENT` (see `flare-git-core::classify`). + // A subprocess-per-`agentflare work`-invocation caller already has this + // set correctly in its own ambient env and this is a no-op for it, but a + // caller running multiple work items as threads inside one long-lived + // process (item #19's in-process dispatch) has no single ambient value + // that's correct for all of them — only an explicit per-spawn env var is. + cmd.env("AGENTFLARE_AGENT", spec.id.as_str()); match run_captured(cmd, hard_cap, idle_timeout) { Ok(c) if c.success => HeadlessOutcome::Ok(c.stdout), Ok(c) if c.timed_out => { diff --git a/src/claims.rs b/src/claims.rs index 42ab49aa..3a061583 100644 --- a/src/claims.rs +++ b/src/claims.rs @@ -248,6 +248,46 @@ pub fn scope_clear_warning( // --- identity / config resolution (impure; thin wrappers over env + git) --- +std::thread_local! { + // Per-thread override for `owner_id()`. `AGENTFLARE_AGENT`/`AGENTFLARE_SESSION` + // are process-global, which is fine for a fresh-process-per-command CLI + // but wrong once multiple work items run as threads inside one long-lived + // daemon process (see `with_owner_override`'s doc comment) — two worker + // threads racing on `std::env::set_var` could attribute a claim/comment + // to the wrong agent. Thread-local sidesteps that: each worker thread's + // override is independent, no shared mutable state, no locking needed. + static OWNER_OVERRIDE: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; +} + +/// Runs `f` with `owner_id()` returning `owner` instead of resolving it from +/// env — for in-process job execution (`agentflare_jobs::InProcessExecutor`), +/// where every worker thread shares the daemon's one process env/pid and so +/// can't rely on `AGENTFLARE_AGENT`/`AGENTFLARE_SESSION`/pid the way a fresh +/// `agentflare work` subprocess naturally can. `owner` should already be the +/// full `:` pair (e.g. `claude-code:`) — the job's +/// own queue id makes a good instance discriminator, playing the same role +/// the subprocess's own unique pid plays today. +pub fn with_owner_override(owner: impl Into, f: impl FnOnce() -> R) -> R { + struct ClearOnDrop; + impl Drop for ClearOnDrop { + fn drop(&mut self) { + OWNER_OVERRIDE.with(|cell| *cell.borrow_mut() = None); + } + } + OWNER_OVERRIDE.with(|cell| *cell.borrow_mut() = Some(owner.into())); + let _clear = ClearOnDrop; + f() +} + +/// Whether the calling thread is currently inside a [`with_owner_override`] +/// scope — lets callers that would otherwise mutate the process-global +/// `AGENTFLARE_AGENT` env var (safe only when there's a single process per +/// identity) skip that mutation once identity is already established +/// per-thread instead. +pub fn has_owner_override() -> bool { + OWNER_OVERRIDE.with(|cell| cell.borrow().is_some()) +} + /// `:` — same agent chain as handoff, plus an instance /// discriminator so two parallel sessions of one agent are distinct owners. /// @@ -258,6 +298,9 @@ pub fn scope_clear_warning( /// separate `agentflare claim` invocations (acquire in one, release in /// another); otherwise each command is a distinct owner. pub fn owner_id() -> String { + if let Some(owner) = OWNER_OVERRIDE.with(|cell| cell.borrow().clone()) { + return owner; + } let agent = std::env::var("AGENTFLARE_AGENT") .ok() .filter(|s| !s.is_empty()) @@ -553,4 +596,59 @@ mod tests { assert!(repo_key_from_url("git@gitlab.com:o/r.git").is_none()); assert!(repo_key_from_url("ssh://git@gitlab.com/o/r.git").is_none()); } + + #[test] + fn with_owner_override_makes_owner_id_return_the_given_value() { + let seen = with_owner_override("claude-code:job-123", owner_id); + assert_eq!(seen, "claude-code:job-123"); + } + + #[test] + fn owner_id_falls_back_to_env_once_the_override_scope_ends() { + // Without an override, owner_id() reads env/pid as usual — just + // asserting the override doesn't leak past its own scope (a stale + // leftover would misattribute every subsequent claim/comment on this + // thread to the wrong agent). + let overridden = with_owner_override("claude-code:job-123", owner_id); + let after = owner_id(); + assert_eq!(overridden, "claude-code:job-123"); + assert_ne!(after, "claude-code:job-123"); + } + + #[test] + fn has_owner_override_reflects_whether_a_scope_is_active() { + assert!(!has_owner_override()); + with_owner_override("codex:job-9", || { + assert!(has_owner_override()); + }); + assert!(!has_owner_override()); + } + + // The whole point of a thread-local override (vs. the process-global + // `AGENTFLARE_AGENT` env var it replaces for in-process job dispatch) is + // that concurrent worker threads each running a different job's agent + // never see each other's identity — proves that directly rather than + // just trusting thread_local!'s documented semantics. + #[test] + fn concurrent_overrides_on_different_threads_never_see_each_others_value() { + let handles: Vec<_> = (0..8) + .map(|i| { + std::thread::spawn(move || { + let owner = format!("agent-{i}:job-{i}"); + with_owner_override(owner.clone(), || { + // Yield repeatedly so other threads' set/clear cycles + // have every chance to interleave with this one if + // the override were shared instead of thread-local. + for _ in 0..50 { + assert_eq!(owner_id(), owner); + std::thread::yield_now(); + } + }); + }) + }) + .collect(); + for h in handles { + h.join().unwrap(); + } + } } diff --git a/src/cli/handoff.rs b/src/cli/handoff.rs index cb8d4f0b..f975433a 100644 --- a/src/cli/handoff.rs +++ b/src/cli/handoff.rs @@ -79,15 +79,17 @@ impl HandoffArgs { .unwrap_or_default(); format!("t{nanos}") }); + // Routed through `claims::owner_id()` (which strips the `:instance` + // suffix via `agent_of`) rather than reading `AGENTFLARE_AGENT` + // directly: same fallback chain (env -> detected agent -> "cli") for + // every existing caller, but it also picks up a thread-local + // identity override for free when this runs as in-process dispatched + // work inside the daemon (see `claims::with_owner_override`) — + // env-var-based identity doesn't work there since worker threads + // share one process env. let sender = self .sender - .or_else(|| { - std::env::var("AGENTFLARE_AGENT") - .ok() - .filter(|s| !s.is_empty()) - }) - .or_else(agent_detector::agent_name) - .unwrap_or_else(|| "cli".into()); + .unwrap_or_else(|| crate::claims::agent_of(&crate::claims::owner_id()).to_string()); let store: agentflare_artifacts::ArtifactStore = match self.dir.clone() { Some(d) => agentflare_artifacts::ArtifactStore::new(d), diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 4a7cc962..324942d7 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -27,7 +27,7 @@ mod uninstall; mod update; mod vault; mod vent; -mod work; +pub(crate) mod work; use clap::{Parser, Subcommand}; use std::sync::LazyLock; diff --git a/src/cli/work.rs b/src/cli/work.rs index 8d093b9b..72e8df67 100644 --- a/src/cli/work.rs +++ b/src/cli/work.rs @@ -5,6 +5,14 @@ use agent_registry::{self, autonomous_args, headless_args}; use clap::Args; use std::time::Duration; +/// `agentflare work --timeout`'s default, and what the in-process +/// [`WorkItemExecutor`] uses for daemon-dispatched work items (which don't +/// go through CLI arg parsing, so can't pick up clap's `default_value_t`) — +/// named so the two can't silently drift apart. +pub const DEFAULT_TIMEOUT_SECS: u64 = 21_600; +/// `agentflare work --idle-timeout`'s default; see [`DEFAULT_TIMEOUT_SECS`]. +pub const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 300; + /// Claim a work item, run an agent on it in an isolated worktree, and /// report the result (comment + PR, or error) back onto the item. #[derive(Args)] @@ -19,7 +27,7 @@ pub struct WorkArgs { /// Absolute hard-cap timeout in seconds, regardless of activity /// (default 21600 = 6h). A backstop against a runaway process, not the /// primary signal for whether to keep a job alive — see --idle-timeout. - #[arg(long, default_value_t = 21600)] + #[arg(long, default_value_t = DEFAULT_TIMEOUT_SECS)] pub timeout: u64, /// Kill the agent if it produces no new stdout/stderr output for this /// many seconds (default 300 = 5 min). This is the primary liveness @@ -27,7 +35,7 @@ pub struct WorkArgs { /// --timeout even if that takes hours; a genuinely stuck task is caught /// quickly instead of running out the full --timeout with nothing /// happening. - #[arg(long, default_value_t = 300)] + #[arg(long, default_value_t = DEFAULT_IDLE_TIMEOUT_SECS)] pub idle_timeout: u64, /// Max agent turns before forced stop (Claude Code only). #[arg(long)] @@ -274,11 +282,19 @@ fn notify(recipient: &str, body: &str, item_id: &str) { impl WorkArgs { pub fn run(self) { - std::process::exit(run_work(self)); + std::process::exit(execute_work(self, &mut std::io::stdout())); } } -fn run_work(args: WorkArgs) -> i32 { +/// Claims `args.target`, runs the resolved agent on it, and reports the +/// outcome back onto the item — the whole body of `agentflare work`. +/// Progress lines that used to go straight to stdout now go through `log` +/// instead, so this same logic can run in-process inside the daemon +/// (`WorkItemExecutor`, called from `agentflare_jobs::WorkerPool`) with its +/// progress captured into that job's own log file — the exact same file the +/// dashboard already tails for subprocess-dispatched jobs — rather than only +/// working when there's a real subprocess's stdout to capture. +pub(crate) fn execute_work(args: WorkArgs, log: &mut dyn std::io::Write) -> i32 { let mcp = AgentflareMcp::default(); let timeout = Duration::from_secs(args.timeout); let idle_timeout = Duration::from_secs(args.idle_timeout); @@ -306,10 +322,21 @@ fn run_work(args: WorkArgs) -> i32 { // sniffing, so it wins outright here, same for a human typing it // directly or the supervisor dispatching this exact command. // - // SAFETY: set once, synchronously, before any worker threads exist - // in this process (this is the first thing `run_work` does). - unsafe { - std::env::set_var("AGENTFLARE_AGENT", resolved.as_str()); + // Skipped when a per-thread owner override is already active + // (`WorkItemExecutor`'s in-process path, see `claims::owner_id()`): + // that already makes `owner_id()` resolve correctly, and mutating + // this process-global env var from a worker thread would race + // against every other worker thread doing the same for a different + // job — see `claims::with_owner_override`'s doc comment. + // + // SAFETY: when reached, this is a single-process-per-command CLI + // invocation (no owner override active) — set once, synchronously, + // before any worker threads exist in this process (this is the + // first thing `execute_work` does). + if !crate::claims::has_owner_override() { + unsafe { + std::env::set_var("AGENTFLARE_AGENT", resolved.as_str()); + } } } @@ -339,7 +366,7 @@ fn run_work(args: WorkArgs) -> i32 { .unwrap_or(&args.target) .to_string(); let item_id = item_id.as_str(); - println!("claimed: {item_id}"); + let _ = writeln!(log, "claimed: {item_id}"); // --- Worktree --- let worktree_path = claim["worktree_path"] @@ -351,7 +378,7 @@ fn run_work(args: WorkArgs) -> i32 { crate::ui::error(msg); return 1; }; - println!("worktree: {}", wpath.display()); + let _ = writeln!(log, "worktree: {}", wpath.display()); // --- Fetch item + prior discussion + labels --- let fetched = mcp.with_backend_db(|conn| { @@ -414,7 +441,7 @@ fn run_work(args: WorkArgs) -> i32 { crate::ui::error(&msg); return 1; } - println!("agent: {} ({route_reason})", agent_enum.as_str()); + let _ = writeln!(log, "agent: {} ({route_reason})", agent_enum.as_str()); let prompt = build_prompt(&item_detail, &comments); @@ -485,9 +512,9 @@ fn run_work(args: WorkArgs) -> i32 { notify(recipient, &comment_body, item_id); } - println!("done: {item_id}"); + let _ = writeln!(log, "done: {item_id}"); if let Some(url) = &pr_url { - println!("pr: {url}"); + let _ = writeln!(log, "pr: {url}"); } 0 } @@ -495,11 +522,51 @@ fn run_work(args: WorkArgs) -> i32 { let msg = failure_message(&other); release_and_comment(&mcp, item_id, &msg, args.notify.as_deref()); crate::ui::error(&msg); + let _ = writeln!(log, "failed: {msg}"); 1 } } } +/// Runs an in-process work-item dispatch job for `agentflare_jobs::WorkerPool` +/// (see `dispatch_item` in `src/supervisor.rs`, which enqueues jobs this +/// executes) instead of the daemon spawning a fresh `agentflare work` +/// subprocess per item. `args` is `[item_id, agent]` — see `dispatch_item` +/// for how it's built. +pub struct WorkItemExecutor; + +impl agentflare_jobs::InProcessExecutor for WorkItemExecutor { + fn execute(&self, job_id: &str, args: &[String], log: &mut dyn std::io::Write) -> Result<(), String> { + let (Some(item_id), Some(agent)) = (args.first(), args.get(1)) else { + return Err(format!( + "malformed in-process work job: expected [item_id, agent], got {args:?}" + )); + }; + let work_args = WorkArgs { + target: item_id.clone(), + agent: Some(agent.clone()), + timeout: DEFAULT_TIMEOUT_SECS, + idle_timeout: DEFAULT_IDLE_TIMEOUT_SECS, + max_turns: None, + max_cost_usd: None, + notify: None, + }; + // `:` — the job's own queue id is a natural instance + // discriminator, playing the role a subprocess's unique pid plays + // for `claims::owner_id()` in the CLI path (see its doc comment). + let owner = format!("{agent}:{job_id}"); + let exit_code = + crate::claims::with_owner_override(owner, || execute_work(work_args, log)); + if exit_code == 0 { + Ok(()) + } else { + Err(format!( + "agentflare work exited with code {exit_code} — see the job log for details" + )) + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/dashboard/server.rs b/src/dashboard/server.rs index 35fbf8f3..7b5656a5 100644 --- a/src/dashboard/server.rs +++ b/src/dashboard/server.rs @@ -605,7 +605,8 @@ pub async fn run(host: &str, port: u16, open: bool, yes_expose: bool) { // 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()); + let mut worker_pool = agentflare_jobs::WorkerPool::new(queue.clone()) + .with_executor(std::sync::Arc::new(crate::cli::work::WorkItemExecutor)); worker_pool.start(2); spawn_job_cleanup(queue.clone()); spawn_supervisor_discovery( diff --git a/src/supervisor.rs b/src/supervisor.rs index 15cff6a5..e0079823 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -1,5 +1,6 @@ //! Background discovery loop: finds items labeled `ready-for-work` and -//! dispatches an `agentflare work` job for each one whose assignee is a +//! dispatches an in-process work job (`WorkItemExecutor`, running the same +//! logic as `agentflare work`) for each one whose assignee is a //! confirmed-autonomous agent (skips the rest with a comment). use crate::mcp_server::AgentflareMcp; @@ -10,18 +11,17 @@ const DISPATCHED_LABEL: &str = "dispatched"; const NEEDS_MANUAL_LABEL: &str = "needs-manual-dispatch"; const NEEDS_HUMAN_GATE_LABEL: &str = "needs-human-gate"; -/// `agentflare work`'s own `--timeout` is its hard-cap safety net, not the -/// primary judge of whether it's still making progress -- that's -/// `--idle-timeout`, which lets a job run for hours as long as it keeps -/// producing output (see item #20). It defaults to 21600s (6h); this is -/// that budget plus margin for the claim/worktree setup and done/push/PR -/// steps around it, so the outer job timeout never cuts off a run before -/// work's own inner timeout would. Before item #20 this outer timeout was -/// 2100s (aligned to work's old 1800s fixed timeout) -- left unaligned -/// after work's default grew, it would have silently reintroduced the same -/// "killed a legitimately-progressing job" bug for every job actually -/// dispatched by the daemon, since this is the timeout that governs them, -/// not work's own. +/// Since item #19, work items run in-process via `WorkItemExecutor` rather +/// than as a spawned `agentflare work` subprocess, so this is no longer an +/// outer subprocess wall-clock kill -- it's the watchdog `run_in_process` +/// (agentflare-jobs' `worker.rs`) uses to abandon a stuck job (see its doc +/// comment) rather than let a hung coordination step wedge a worker thread +/// forever. `WorkArgs::DEFAULT_TIMEOUT_SECS` (21600s = 6h) is `agentflare +/// work`'s own hard-cap safety net -- not the primary judge of progress, +/// that's `--idle-timeout` (item #20) -- so this stays that budget plus +/// margin for the claim/worktree/done steps around it, exactly as when it +/// wrapped a real subprocess: it must never fire before the work being +/// watched would have stopped on its own. const WORK_JOB_TIMEOUT_SECS: u64 = 21_900; /// Returns the matching `Agent` only if `agent_registry::autonomous_args` @@ -183,17 +183,16 @@ fn dispatch_item( label_id_by_name: &std::collections::HashMap, ready_id: &str, ) -> bool { - let command = std::env::current_exe() - .map(|p| p.to_string_lossy().to_string()) - .unwrap_or_else(|_| "agentflare".to_string()); - let job = agentflare_jobs::AgentJob::new(command) - .args([ - "work".to_string(), - item.id.clone(), - "--agent".to_string(), - agent.as_str().to_string(), - ]) - .timeout(WORK_JOB_TIMEOUT_SECS); + // Runs in-process via `WorkItemExecutor` (registered on the daemon's + // `WorkerPool`, see `dashboard/server.rs::run`) instead of spawning a + // fresh `agentflare work` subprocess — item #19. `command` is a display + // label only (shown in the dashboard's job list); nothing spawns it. + // `args` is `[item_id, agent]`, exactly what `WorkItemExecutor::execute` + // expects. + let job = agentflare_jobs::AgentJob::new("agentflare-work") + .args([item.id.clone(), agent.as_str().to_string()]) + .timeout(WORK_JOB_TIMEOUT_SECS) + .in_process(); let Ok(info) = queue.enqueue(&job) else { return false; }; @@ -328,7 +327,7 @@ mod tests { let jobs = queue.list(None).unwrap(); assert_eq!(jobs.len(), 1); - assert!(jobs[0].args.contains(&"work".to_string())); + assert!(jobs[0].in_process, "work-item jobs must run in-process (item #19)"); assert!(jobs[0].args.contains(&item_id)); assert!(jobs[0].args.contains(&"claude-code".to_string())); From 1ecd03ffefea2305fee72d56650accaaf02a5d6e Mon Sep 17 00:00:00 2001 From: shiva Date: Sat, 8 Aug 2026 13:12:39 +0530 Subject: [PATCH 2/2] style: apply cargo fmt to item #19's changes --- crates/agentflare-jobs/src/executor.rs | 7 ++++++- crates/agentflare-jobs/src/worker.rs | 9 +++++++-- crates/agentflare-jobs/tests/in_process_test.rs | 12 ++++++++++-- src/cli/work.rs | 10 +++++++--- src/supervisor.rs | 5 ++++- 5 files changed, 34 insertions(+), 9 deletions(-) diff --git a/crates/agentflare-jobs/src/executor.rs b/crates/agentflare-jobs/src/executor.rs index 75e3be40..c27b6016 100644 --- a/crates/agentflare-jobs/src/executor.rs +++ b/crates/agentflare-jobs/src/executor.rs @@ -12,5 +12,10 @@ pub trait InProcessExecutor: Send + Sync { /// name/location `Supervisor::spawn` uses for subprocess jobs, so /// existing log-tailing (e.g. the dashboard's `/api/jobs/:id/stream`) /// keeps working unchanged for in-process jobs too. - fn execute(&self, job_id: &str, args: &[String], log: &mut dyn std::io::Write) -> Result<(), String>; + fn execute( + &self, + job_id: &str, + args: &[String], + log: &mut dyn std::io::Write, + ) -> Result<(), String>; } diff --git a/crates/agentflare-jobs/src/worker.rs b/crates/agentflare-jobs/src/worker.rs index 40fb6d06..720ff583 100644 --- a/crates/agentflare-jobs/src/worker.rs +++ b/crates/agentflare-jobs/src/worker.rs @@ -144,7 +144,10 @@ fn run_in_process( executor: Option<&Arc>, ) { let Some(executor) = executor else { - if let Err(e) = queue.fail(id, "job is marked in_process but no InProcessExecutor is registered on this WorkerPool") { + if let Err(e) = queue.fail( + id, + "job is marked in_process but no InProcessExecutor is registered on this WorkerPool", + ) { eprintln!("agentflare-jobs: failed to record failure for {id}: {e}"); } return; @@ -174,7 +177,9 @@ fn run_in_process( let outcome = rx.recv_timeout(Duration::from_secs(job.timeout_secs.max(1))); match outcome { Ok(Ok(())) => { - let stdout_total_bytes = std::fs::metadata(&stdout_path).map(|m| m.len()).unwrap_or(0); + let stdout_total_bytes = std::fs::metadata(&stdout_path) + .map(|m| m.len()) + .unwrap_or(0); let output = JobOutput { exit_code: Some(0), timed_out: false, diff --git a/crates/agentflare-jobs/tests/in_process_test.rs b/crates/agentflare-jobs/tests/in_process_test.rs index 74475920..72a36e53 100644 --- a/crates/agentflare-jobs/tests/in_process_test.rs +++ b/crates/agentflare-jobs/tests/in_process_test.rs @@ -53,7 +53,10 @@ fn in_process_job_completes_via_the_registered_executor_and_writes_its_log() { .in_process(), ) .unwrap(); - assert!(info.in_process, "enqueue should echo the in_process flag back"); + assert!( + info.in_process, + "enqueue should echo the in_process flag back" + ); let final_info = wait_for_terminal(&q, &info.id, 200); pool.shutdown(); @@ -152,7 +155,12 @@ fn in_process_job_that_hangs_is_abandoned_at_its_timeout_instead_of_wedging_the_ let start = std::time::Instant::now(); let info = q - .enqueue(&AgentJob::new("label-only").in_process().timeout(1).max_retries(0)) + .enqueue( + &AgentJob::new("label-only") + .in_process() + .timeout(1) + .max_retries(0), + ) .unwrap(); let final_info = wait_for_terminal(&q, &info.id, 400); diff --git a/src/cli/work.rs b/src/cli/work.rs index 72e8df67..ee073c58 100644 --- a/src/cli/work.rs +++ b/src/cli/work.rs @@ -536,7 +536,12 @@ pub(crate) fn execute_work(args: WorkArgs, log: &mut dyn std::io::Write) -> i32 pub struct WorkItemExecutor; impl agentflare_jobs::InProcessExecutor for WorkItemExecutor { - fn execute(&self, job_id: &str, args: &[String], log: &mut dyn std::io::Write) -> Result<(), String> { + fn execute( + &self, + job_id: &str, + args: &[String], + log: &mut dyn std::io::Write, + ) -> Result<(), String> { let (Some(item_id), Some(agent)) = (args.first(), args.get(1)) else { return Err(format!( "malformed in-process work job: expected [item_id, agent], got {args:?}" @@ -555,8 +560,7 @@ impl agentflare_jobs::InProcessExecutor for WorkItemExecutor { // discriminator, playing the role a subprocess's unique pid plays // for `claims::owner_id()` in the CLI path (see its doc comment). let owner = format!("{agent}:{job_id}"); - let exit_code = - crate::claims::with_owner_override(owner, || execute_work(work_args, log)); + let exit_code = crate::claims::with_owner_override(owner, || execute_work(work_args, log)); if exit_code == 0 { Ok(()) } else { diff --git a/src/supervisor.rs b/src/supervisor.rs index e0079823..f4ac8672 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -327,7 +327,10 @@ mod tests { let jobs = queue.list(None).unwrap(); assert_eq!(jobs.len(), 1); - assert!(jobs[0].in_process, "work-item jobs must run in-process (item #19)"); + assert!( + jobs[0].in_process, + "work-item jobs must run in-process (item #19)" + ); assert!(jobs[0].args.contains(&item_id)); assert!(jobs[0].args.contains(&"claude-code".to_string()));