refactor(supervisor): dispatch work items in-process instead of re-invoking agentflare work

crates/agentflare-jobs/src/executor.rs

@@ -0,0 +1,21 @@
+/// 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>;
+}

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};

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<JobInfo> {
     // `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::<crate::types::AgentJob>(&payload_json)
-        .map(|job| (job.command, job.args))
+    let (command, args, in_process) = serde_json::from_str::<crate::types::AgentJob>(&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<JobInfo> {
             stdout_total_bytes: stdout_bytes as u64,
             stderr_total_bytes: stderr_bytes as u64,
         }),
+        in_process,
     })
 }
 

crates/agentflare-jobs/src/types.rs

@@ -31,6 +31,15 @@ pub struct AgentJob {
     pub agent_type: Option<String>,
     pub prompt: Option<String>,
     pub metadata: HashMap<String, String>,
+    /// 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<String>) -> Self {
         self.args.push(arg.into());
         self
@@ -114,4 +132,5 @@ pub struct JobInfo {
     pub started_at: Option<i64>,
     pub finished_at: Option<i64>,
     pub output: Option<JobOutput>,
+    pub in_process: bool,
 }

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<Queue>,
     handles: Vec<JoinHandle<()>>,
     running: Arc<AtomicBool>,
+    executor: Option<Arc<dyn InProcessExecutor>>,
 }
 
 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<dyn InProcessExecutor>) -> 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<dyn InProcessExecutor>>) {
     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,97 @@ 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<dyn InProcessExecutor>>,
+) {
+    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::<Result<(), String>>();
+    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}");
+            }
+        }
+    }
+}