feat: add background job runner, job__* handlers, and start gates

Detached tokio::process runner with a frozen JobEnvSnapshot (env-derived
bin dirs, vault-interpolated agent envs, COYOTE_TOOL_TIMEOUT resolved at
start), process_group(0) with pgid-guarded SIGTERM/SIGKILL escalation,
capture-only ring-buffer telemetry, and LLM_OUTPUT read after wait().
MCP jobs snapshot a single-entry McpRuntime holding only the validated
server and render through the same free fn as the foreground path.

job__start enforces its gates synchronously before any spawn:
jobs_enabled, the backgroundable whitelist with directionality teaching
errors, the per-request declared-names stash captured in
before_chat_completion, then capacity (lazy supervisor get-or-init in
plain sessions). job__check/list read the shared JobState cell without
consuming; job__collect blocks with the escalation early-out and applies
a tail-biased char-boundary cap plus optional tail_lines; job__cancel
kills the group with a 5s grace.

Job declarations are injected iff jobs are enabled at agent init, the
plain-session function-init sites, and the exit_agent rebuild; job__ is
carved out of enabled_tools filtering and excluded from
concrete_tool_names so REPL toggles cannot grant or revoke it.

src/config/agent.rs

@@ -3,7 +3,7 @@ use super::*;
 use crate::{
     client::Model,
     config::memory,
-    function::{Functions, run_llm_function},
+    function::{Functions, jobs::DEFAULT_MAX_CONCURRENT_JOBS, run_llm_function},
     graph, rag,
 };
 
@@ -225,6 +225,16 @@ impl Agent {
             functions.append_supervisor_functions();
         }
 
+        if app.function_calling_support
+            && agent_config
+                .max_concurrent_jobs
+                .or(app.max_concurrent_jobs)
+                .unwrap_or(DEFAULT_MAX_CONCURRENT_JOBS)
+                > 0
+        {
+            functions.append_job_functions();
+        }
+
         functions.append_teammate_functions();
         functions.append_user_interaction_functions();
 

src/config/app_state.rs

@@ -1,6 +1,7 @@
 use super::mcp_factory::{McpFactory, McpServerKey};
 use super::rag_cache::RagCache;
 use crate::config::AppConfig;
+use crate::config::jobs_enabled;
 use crate::function::Functions;
 use crate::mcp::{McpRegistry, McpServersConfig};
 use crate::utils::AbortSignal;
@@ -72,6 +73,9 @@ impl AppState {
         if !mcp_registry.is_empty() && config.mcp_server_support {
             functions.append_mcp_meta_functions(mcp_registry.server_features());
         }
+        if jobs_enabled(None, &config) {
+            functions.append_job_functions();
+        }
 
         let mcp_registry = if mcp_registry.is_empty() {
             None

src/config/input.rs

@@ -9,7 +9,12 @@ use crate::utils::{AbortSignal, base64_encode, is_loader_protocol, sha256};
 
 use anyhow::{Context, Result, bail};
 use indexmap::IndexSet;
-use std::{collections::HashMap, fs::File, io::Read, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    fs::File,
+    io::Read,
+    sync::Arc,
+};
 use unicode_width::{UnicodeWidthChar, UnicodeWidthStr};
 
 const IMAGE_EXTS: [&str; 5] = ["png", "jpeg", "jpg", "webp", "gif"];
@@ -158,6 +163,14 @@ impl Input {
         self.data_urls.clone()
     }
 
+    /// Names of the function declarations this request will send to the model.
+    pub fn declared_function_names(&self) -> HashSet<String> {
+        self.functions
+            .as_ref()
+            .map(|functions| functions.iter().map(|f| f.name.clone()).collect())
+            .unwrap_or_default()
+    }
+
     pub fn tool_calls(&self) -> &Option<MessageContentToolCalls> {
         &self.tool_calls
     }

src/config/mod.rs

@@ -59,7 +59,8 @@ pub use self::skill_registry::SkillRegistry;
 #[cfg(test)]
 pub(crate) use self::tool_scope::test_fixtures;
 pub use self::tool_scope::{
-    McpPromptCompletion, flatten_prompt_messages, resolve_prompt_args, sanitize_display_text,
+    McpPromptCompletion, McpRuntime, flatten_prompt_messages, resolve_prompt_args,
+    sanitize_display_text,
 };
 pub use self::update::run_self_update;
 use crate::client::{

src/config/request_context.rs

@@ -17,9 +17,13 @@ use super::{
 use super::{MessageContentToolCalls, prompts};
 use crate::client::{Model, ModelType, list_models};
 use crate::function::{
-    FunctionDeclaration, Functions, ToolCallTracker, ToolResult, memory::MEMORY_FUNCTION_PREFIX,
+    FunctionDeclaration, Functions, ToolCallTracker, ToolResult,
-    rag_query::RAG_FUNCTION_PREFIX, skill::SKILL_FUNCTION_PREFIX,
+    jobs::{DEFAULT_MAX_CONCURRENT_JOBS, JOB_FUNCTION_PREFIX},
-    supervisor::SUPERVISOR_FUNCTION_PREFIX, todo::TODO_FUNCTION_PREFIX,
+    memory::MEMORY_FUNCTION_PREFIX,
+    rag_query::RAG_FUNCTION_PREFIX,
+    skill::SKILL_FUNCTION_PREFIX,
+    supervisor::SUPERVISOR_FUNCTION_PREFIX,
+    todo::TODO_FUNCTION_PREFIX,
     user_interaction::USER_FUNCTION_PREFIX,
 };
 use crate::mcp::{
@@ -142,7 +146,7 @@ pub fn effective_max_concurrent_jobs(agent: Option<&Agent>, app: &AppConfig) ->
     agent
         .and_then(|a| a.max_concurrent_jobs())
         .or(app.max_concurrent_jobs)
-        .unwrap_or(5)
+        .unwrap_or(DEFAULT_MAX_CONCURRENT_JOBS)
 }
 
 pub fn jobs_enabled(agent: Option<&Agent>, app: &AppConfig) -> bool {
@@ -319,6 +323,8 @@ pub struct RequestContext {
 
     pub tool_scope: ToolScope,
 
+    pub declared_function_names: HashSet<String>,
+
     pub supervisor: Option<Arc<RwLock<Supervisor>>>,
     pub parent_supervisor: Option<Arc<RwLock<Supervisor>>>,
     pub self_agent_id: Option<String>,
@@ -352,6 +358,7 @@ impl RequestContext {
             agent: None,
             last_message: None,
             tool_scope: ToolScope::default(),
+            declared_function_names: Default::default(),
             supervisor: None,
             parent_supervisor: None,
             self_agent_id: None,
@@ -411,6 +418,7 @@ impl RequestContext {
                 mcp_runtime,
                 tool_tracker: ToolCallTracker::default(),
             },
+            declared_function_names: Default::default(),
             supervisor: None,
             parent_supervisor: None,
             self_agent_id: None,
@@ -457,6 +465,7 @@ impl RequestContext {
             agent: self.agent.clone(),
             last_message: self.last_message.clone(),
             tool_scope: self.tool_scope.clone(),
+            declared_function_names: self.declared_function_names.clone(),
             supervisor: self.supervisor.clone(),
             parent_supervisor: self.parent_supervisor.clone(),
             self_agent_id: self.self_agent_id.clone(),
@@ -501,6 +510,7 @@ impl RequestContext {
                 mcp_runtime: McpRuntime::default(),
                 tool_tracker: tool_call_tracker,
             },
+            declared_function_names: Default::default(),
             supervisor: None,
             parent_supervisor: parent.supervisor.clone(),
             self_agent_id: Some(self_agent_id),
@@ -905,6 +915,9 @@ impl RequestContext {
     }
 
     pub fn before_chat_completion(&mut self, input: &Input) -> Result<()> {
+        // The R11 gate in `job__start` validates against exactly what was
+        // declared to the model for THIS request; refresh it every time.
+        self.declared_function_names = input.declared_function_names();
         self.last_message = Some(LastMessage::new(input.clone(), String::new()));
         Ok(())
     }
@@ -1313,6 +1326,7 @@ impl RequestContext {
                     && !v.name.starts_with("memory__")
                     && !v.name.starts_with("skill__")
                     && !v.name.starts_with("rag__")
+                    && !v.name.starts_with("job__")
             })
             .map(|v| v.name.clone())
             .collect()
@@ -2130,7 +2144,8 @@ impl RequestContext {
                                 && v.name.starts_with(SKILL_FUNCTION_PREFIX))
                             || (self.auto_continue_config().enabled
                                 && v.name.starts_with(TODO_FUNCTION_PREFIX))
-                            || v.name.starts_with(RAG_FUNCTION_PREFIX))
+                            || v.name.starts_with(RAG_FUNCTION_PREFIX)
+                            || v.name.starts_with(JOB_FUNCTION_PREFIX))
                             && !existing.contains(&v.name)
                     })
                     .cloned()
@@ -2157,6 +2172,7 @@ impl RequestContext {
                             || v.name.starts_with(SUPERVISOR_FUNCTION_PREFIX)
                             || v.name.starts_with(MEMORY_FUNCTION_PREFIX)
                             || v.name.starts_with(RAG_FUNCTION_PREFIX)
+                            || v.name.starts_with(JOB_FUNCTION_PREFIX)
                     });
                 }
 
@@ -3869,6 +3885,9 @@ impl RequestContext {
         {
             functions.append_rag_query_functions();
         }
+        if self.agent.is_none() && jobs_enabled(None, app) {
+            functions.append_job_functions();
+        }
 
         let tool_tracker = self.tool_scope.tool_tracker.clone();
         self.tool_scope = ToolScope {
@@ -4206,6 +4225,9 @@ impl RequestContext {
         if self.working_mode.is_repl() {
             functions.append_user_interaction_functions();
         }
+        if jobs_enabled(None, app) {
+            functions.append_job_functions();
+        }
         let tool_tracker = self.tool_scope.tool_tracker.clone();
         self.tool_scope = ToolScope {
             functions,
@@ -5713,6 +5735,126 @@ mod tests {
         assert!(ctx.select_functions(&role).is_none());
     }
 
+    #[test]
+    fn select_functions_returns_job_functions_even_with_no_enabled_tools() {
+        let mut ctx = create_test_ctx();
+        ctx.tool_scope.functions.append_job_functions();
+
+        let fns = ctx.select_functions(&Role::default()).unwrap();
+        let names: Vec<&str> = fns.iter().map(|f| f.name.as_str()).collect();
+
+        assert_eq!(
+            names,
+            vec![
+                "job__start",
+                "job__check",
+                "job__collect",
+                "job__cancel",
+                "job__list"
+            ]
+        );
+    }
+
+    #[test]
+    fn select_functions_preserves_job_tools_under_role_filter() {
+        let mut ctx = create_test_ctx();
+        ctx.tool_scope.functions.append_job_functions();
+
+        let mut role = Role::new("r", "p");
+        role.set_enabled_tools(Some(vec!["foo".to_string()]));
+
+        let fns = ctx.select_functions(&role).unwrap();
+        assert!(
+            fns.iter().any(|f| f.name == "job__start"),
+            "job__ tools must survive a role tool filter"
+        );
+    }
+
+    #[test]
+    fn concrete_tool_names_excludes_job_functions() {
+        let mut ctx = create_test_ctx();
+        ctx.tool_scope.functions.append_job_functions();
+
+        assert!(ctx.concrete_tool_names().is_empty());
+    }
+
+    #[test]
+    fn before_chat_completion_refreshes_declared_function_names() {
+        let mut ctx = create_test_ctx();
+        ctx.tool_scope.functions.append_job_functions();
+        let input = Input::from_str(&ctx, "hello", None).unwrap();
+        ctx.before_chat_completion(&input).unwrap();
+
+        assert_eq!(ctx.declared_function_names.len(), 5);
+        assert!(ctx.declared_function_names.contains("job__start"));
+
+        ctx.tool_scope = ToolScope::default();
+        let input = Input::from_str(&ctx, "hello again", None).unwrap();
+        ctx.before_chat_completion(&input).unwrap();
+
+        assert!(
+            ctx.declared_function_names.is_empty(),
+            "stash must be refreshed on every request"
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn rebuild_tool_scope_gates_job_functions_on_jobs_enabled() {
+        let _guard = TestConfigDirGuard::new();
+        let app_state = app_state_with_mcp_config(false, &[]);
+        let mut ctx = RequestContext::new(app_state, WorkingMode::Cmd);
+        let app = ctx.app.config.clone();
+        let abort = utils::create_abort_signal();
+
+        run_async(ctx.rebuild_tool_scope(&app, None, abort.clone())).unwrap();
+        assert!(ctx.tool_scope.functions.contains("job__start"));
+
+        let jobs_off = AppConfig {
+            max_concurrent_jobs: Some(0),
+            ..(*app).clone()
+        };
+        run_async(ctx.rebuild_tool_scope(&jobs_off, None, abort.clone())).unwrap();
+        assert!(
+            !ctx.tool_scope
+                .functions
+                .declarations()
+                .iter()
+                .any(|f| f.name.starts_with("job__"))
+        );
+
+        let fc_off = AppConfig {
+            function_calling_support: false,
+            ..(*app).clone()
+        };
+        run_async(ctx.rebuild_tool_scope(&fc_off, None, abort)).unwrap();
+        assert!(
+            !ctx.tool_scope
+                .functions
+                .declarations()
+                .iter()
+                .any(|f| f.name.starts_with("job__"))
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn exit_agent_rebuild_retains_job_functions_when_enabled() {
+        let _guard = TestConfigDirGuard::new();
+        let mut ctx = create_test_ctx();
+        let app = ctx.app.config.clone();
+
+        ctx.exit_agent(&app).unwrap();
+        assert!(ctx.tool_scope.functions.contains("job__start"));
+
+        let jobs_off = AppConfig {
+            max_concurrent_jobs: Some(0),
+            ..(*app).clone()
+        };
+        ctx.exit_agent(&jobs_off).unwrap();
+        assert!(!ctx.tool_scope.functions.contains("job__start"));
+    }
+
     #[test]
     fn select_functions_all_enabled_tools_returns_all_non_mcp() {
         let mut ctx = create_test_ctx();
@@ -5879,6 +6021,42 @@ mod tests {
         );
     }
 
+    #[test]
+    #[serial]
+    fn select_functions_preserves_job_tools_under_agent_filter() {
+        let _guard = TestConfigDirGuard::new();
+        let mut ctx = create_test_ctx();
+        let app = ctx.app.config.clone();
+        let agent_name = format!(
+            "test_job_agent_{}",
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        );
+        let agent_dir = paths::agent_data_dir(&agent_name);
+        create_dir_all(&agent_dir).unwrap();
+        write(
+            agent_dir.join("config.yaml"),
+            format!("name: {agent_name}\ninstructions: hi\n"),
+        )
+        .unwrap();
+
+        let abort = utils::create_abort_signal();
+        run_async(ctx.use_agent(&app, &agent_name, None, abort)).unwrap();
+
+        let mut role = Role::new("r", "p");
+        role.set_enabled_tools(Some(vec!["foo".to_string()]));
+
+        let fns = ctx.select_functions(&role).unwrap();
+        let names: Vec<&str> = fns.iter().map(|f| f.name.as_str()).collect();
+        assert!(
+            names.contains(&"job__start"),
+            "job__ tools must survive an agent tool filter, got: {names:?}"
+        );
+        assert!(names.contains(&"job__collect"));
+    }
+
     #[test]
     fn fork_for_branch_clones_skill_registry() {
         let mut ctx = create_test_ctx();

src/function/mod.rs

@@ -28,6 +28,7 @@ use anyhow::{Context, Result, anyhow, bail};
 use futures_util::future;
 use indexmap::IndexMap;
 use indoc::formatdoc;
+use jobs::JOB_FUNCTION_PREFIX;
 use memory::MEMORY_FUNCTION_PREFIX;
 use rag_query::RAG_FUNCTION_PREFIX;
 use rust_embed::Embed;
@@ -625,6 +626,10 @@ impl Functions {
             .extend(supervisor::escalation_function_declarations());
     }
 
+    pub fn append_job_functions(&mut self) {
+        self.declarations.extend(jobs::job_function_declarations());
+    }
+
     pub fn append_teammate_functions(&mut self) {
         self.declarations
             .extend(supervisor::teammate_function_declarations());
@@ -1548,6 +1553,15 @@ impl ToolCall {
                         json!({"tool_call_error": error_msg})
                     })
             }
+            _ if cmd_name.starts_with(JOB_FUNCTION_PREFIX) => {
+                jobs::handle_job_tool(ctx, &cmd_name, &json_data)
+                    .await
+                    .unwrap_or_else(|e| {
+                        let error_msg = format!("Job tool failed: {e}");
+                        eprintln!("{}", muted_warning_text(&format!("⚠️ {error_msg} ⚠️")));
+                        json!({"tool_call_error": error_msg})
+                    })
+            }
             _ => match run_llm_function(cmd_name, cmd_args, envs, agent_name) {
                 Ok(Some(contents)) => serde_json::from_str(&contents)
                     .ok()
@@ -2858,6 +2872,37 @@ mod tests {
         assert!(f.contains("agent__reply_escalation"));
     }
 
+    #[test]
+    fn functions_append_job_adds_declarations() {
+        let mut f = Functions::default();
+        f.append_job_functions();
+        assert!(f.contains("job__start"));
+        assert!(f.contains("job__check"));
+        assert!(f.contains("job__collect"));
+        assert!(f.contains("job__cancel"));
+        assert!(f.contains("job__list"));
+    }
+
+    #[test]
+    fn eval_routes_declared_job_calls_to_job_handlers() {
+        let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
+        ctx.tool_scope.functions.append_job_functions();
+        let calls = vec![call("job__list", Some("id-1"))];
+        let results = run_async(eval_tool_calls(&mut ctx, calls)).unwrap();
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].output["active_jobs"], 0);
+        assert_eq!(results[0].output["jobs"], json!([]));
+    }
+
+    #[test]
+    fn eval_soft_fails_job_calls_when_jobs_not_declared() {
+        let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
+        let calls = vec![call("job__start", Some("id-1"))];
+        let results = run_async(eval_tool_calls(&mut ctx, calls)).unwrap();
+        let err = results[0].output["tool_call_error"].as_str().unwrap();
+        assert!(err.contains("Unexpected call"));
+    }
+
     #[test]
     fn functions_append_teammate_adds_declarations() {
         let mut f = Functions::default();

src/supervisor/mod.rs

@@ -40,7 +40,6 @@ pub struct AgentHandle {
     pub child_supervisor: Option<Arc<RwLock<Supervisor>>>,
 }
 
-#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum JobStatus {
     Running,
@@ -49,12 +48,10 @@ pub enum JobStatus {
 }
 
 pub struct JobState {
-    #[allow(dead_code)]
     pub status: JobStatus,
     pub pgid: Option<i32>,
 }
 
-#[allow(dead_code)]
 pub struct JobResult {
     pub output: Value,
     pub exit_code: Option<i32>,
@@ -63,14 +60,11 @@ pub struct JobResult {
 
 pub struct JobHandle {
     pub id: String,
-    #[allow(dead_code)]
     pub tool: String,
-    #[allow(dead_code)]
     pub started_at: Instant,
     pub join_handle: JoinHandle<Result<JobResult>>,
     pub abort_signal: AbortSignal,
     pub state: Arc<Mutex<JobState>>,
-    #[allow(dead_code)]
     pub output_buf: Arc<Mutex<RingBuf>>,
     #[allow(dead_code)]
     pub no_change_checks: u32,
@@ -157,6 +151,20 @@ impl Supervisor {
         })
     }
 
+    pub fn job(&self, id: &str) -> Option<&JobHandle> {
+        match self.handles.get(id) {
+            Some(TaskHandle::Job(handle)) => Some(handle),
+            _ => None,
+        }
+    }
+
+    pub fn jobs(&self) -> impl Iterator<Item = &JobHandle> {
+        self.handles.values().filter_map(|handle| match handle {
+            TaskHandle::Job(handle) => Some(handle),
+            TaskHandle::Agent(_) => None,
+        })
+    }
+
     pub fn active_count(&self) -> usize {
         self.agents().count()
     }
@@ -182,7 +190,6 @@ impl Supervisor {
         self.max_depth
     }
 
-    #[allow(dead_code)]
     pub fn max_concurrent_jobs(&self) -> usize {
         self.max_concurrent_jobs
     }