diff --git a/crates/agentflare-jobs/src/executor.rs b/crates/agentflare-jobs/src/executor.rs
new file mode 100644
index 00000000..c27b6016
--- /dev/null
+++ b/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>;
+}
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<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,
     })
 }
 
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<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,
 }
diff --git a/crates/agentflare-jobs/src/worker.rs b/crates/agentflare-jobs/src/worker.rs
index 1a81ea1e..720ff583 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<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}");
+            }
+        }
+    }
+}
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..72a36e53
--- /dev/null
+++ b/crates/agentflare-jobs/tests/in_process_test.rs
@@ -0,0 +1,184 @@
+//! 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<Option<String>> = 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 `<agent>:<instance>` pair (e.g. `claude-code:<job-id>`) — 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<R>(owner: impl Into<String>, 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())
+}
+
 /// `<agent>:<instance>` — 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..ee073c58 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,55 @@ 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,
+        };
+        // `<agent>:<job-id>` — 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 553145d1..4892a90e 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,28 +183,18 @@ fn dispatch_item(
     label_id_by_name: &std::collections::HashMap<String, String>,
     ready_id: &str,
 ) -> bool {
-    // `current_exe()` can return `Ok` with a path that no longer exists: on
-    // Linux, once the running binary's file is replaced (cargo install,
-    // package upgrade, `agentflare update`) it resolves via /proc/self/exe
-    // to the old, now-deleted inode -- Ok(path), but that path won't exec.
-    // A long-running daemon that resolves this once at dispatch time then
-    // fails every dispatch until someone notices and restarts it, so check
-    // the path is still real and fall back to a bare "agentflare" (resolved
-    // via PATH at spawn time, same fallback already used for the Err case)
-    // rather than trust a stale exe path.
-    let command = std::env::current_exe()
-        .ok()
-        .filter(|p| p.exists())
-        .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, so
+    // master's `current_exe()`-staleness fix (see git history) is moot here:
+    // there's no exe path to resolve at all once dispatch never spawns one.
+    // `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;
     };
@@ -339,7 +329,10 @@ 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()));