From cb025b7fffcc0e1e70ee5af69ad9b4c0c4dcaa84 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Tue, 25 Aug 2026 16:54:44 -0600 Subject: [PATCH] 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 | 12 +- src/config/app_state.rs | 4 + src/config/input.rs | 15 +- src/config/mod.rs | 3 +- src/config/request_context.rs | 188 +++- src/function/jobs.rs | 1700 ++++++++++++++++++++++++++++++++- src/function/mod.rs | 45 + src/supervisor/mod.rs | 21 +- 8 files changed, 1965 insertions(+), 23 deletions(-) diff --git a/src/config/agent.rs b/src/config/agent.rs index acddb70..c139b71 100644 --- a/src/config/agent.rs +++ b/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(); diff --git a/src/config/app_state.rs b/src/config/app_state.rs index 7a6cb0e..d4c9079 100644 --- a/src/config/app_state.rs +++ b/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 diff --git a/src/config/input.rs b/src/config/input.rs index 5d35064..85a6a68 100644 --- a/src/config/input.rs +++ b/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 { + self.functions + .as_ref() + .map(|functions| functions.iter().map(|f| f.name.clone()).collect()) + .unwrap_or_default() + } + pub fn tool_calls(&self) -> &Option { &self.tool_calls } diff --git a/src/config/mod.rs b/src/config/mod.rs index 6f070b4..e89caee 100644 --- a/src/config/mod.rs +++ b/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::{ diff --git a/src/config/request_context.rs b/src/config/request_context.rs index 3dc350c..00be93c 100644 --- a/src/config/request_context.rs +++ b/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, - rag_query::RAG_FUNCTION_PREFIX, skill::SKILL_FUNCTION_PREFIX, - supervisor::SUPERVISOR_FUNCTION_PREFIX, todo::TODO_FUNCTION_PREFIX, + FunctionDeclaration, Functions, ToolCallTracker, ToolResult, + jobs::{DEFAULT_MAX_CONCURRENT_JOBS, JOB_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, + pub supervisor: Option>>, pub parent_supervisor: Option>>, pub self_agent_id: Option, @@ -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(); diff --git a/src/function/jobs.rs b/src/function/jobs.rs index 05ac697..14c66a1 100644 --- a/src/function/jobs.rs +++ b/src/function/jobs.rs @@ -1,9 +1,44 @@ -use crate::supervisor::Supervisor; +use super::memory::MEMORY_FUNCTION_PREFIX; +use super::rag_query::RAG_FUNCTION_PREFIX; +use super::skill::SKILL_FUNCTION_PREFIX; +use super::supervisor::SUPERVISOR_FUNCTION_PREFIX; +use super::todo::TODO_FUNCTION_PREFIX; +use super::user_interaction::USER_FUNCTION_PREFIX; +use super::{FunctionDeclaration, JsonSchema, PATH_SEP, mcp_error_display, render_tool_result}; +use crate::config::{ + McpRuntime, RequestContext, effective_max_concurrent_jobs, jobs_enabled, paths, +}; +use crate::graph; +use crate::mcp::{ + MCP_DESCRIBE_META_FUNCTION_NAME_PREFIX, MCP_INVOKE_META_FUNCTION_NAME_PREFIX, + MCP_PROMPT_META_FUNCTION_NAME_PREFIX, MCP_READ_META_FUNCTION_NAME_PREFIX, + MCP_SEARCH_META_FUNCTION_NAME_PREFIX, +}; +use crate::supervisor::{JobHandle, JobResult, JobState, JobStatus, Supervisor}; +use crate::utils::{create_abort_signal, muted_warning_text, temp_file, wait_abort_signal}; -use parking_lot::RwLock; +use anyhow::{Context, Result, anyhow, bail}; +use indexmap::IndexMap; +use parking_lot::{Mutex, RwLock}; +use serde_json::{Value, json}; +use std::collections::HashMap; +use std::path::PathBuf; +use std::process::Stdio; use std::sync::Arc; +use std::time::{Duration, Instant}; +use std::{env, fs}; +use tokio::io::AsyncReadExt; +use tokio::time; +use uuid::Uuid; + +pub const JOB_FUNCTION_PREFIX: &str = "job__"; + +pub const DEFAULT_MAX_CONCURRENT_JOBS: usize = 5; + +const JOB_RESULT_TAIL_CAP_CHARS: usize = 50_000; + +const JOB_KILL_GRACE: Duration = Duration::from_secs(5); -#[allow(dead_code)] pub fn is_agent_task(supervisor: Option<&Arc>>, id: &str) -> bool { id.starts_with("agent_") || id.starts_with("graph_agent_") @@ -17,7 +52,6 @@ pub struct RingBuf { total_written: u64, } -#[allow(dead_code)] impl RingBuf { pub fn new(capacity: usize) -> Self { Self { @@ -69,9 +103,1019 @@ impl Default for RingBuf { } } +/// Everything a detached process job needs, frozen at `job__start`: config, +/// env, and PATH changes made afterwards do not affect a running job. +pub struct JobEnvSnapshot { + cmd_name: String, + display_name: String, + cmd_args: Vec, + envs: HashMap, + output_file: PathBuf, + timeout_secs: u64, +} + +/// The complete context an MCP job task owns. The runtime holds ONLY the +/// validated server's handle, so the detached task cannot reach any other +/// server even by bug. +pub struct JobCtx { + mcp_runtime: McpRuntime, + current_depth: usize, +} + +pub fn job_function_declarations() -> Vec { + vec![ + FunctionDeclaration { + name: format!("{JOB_FUNCTION_PREFIX}start"), + description: "Run a tool call as a background job and return immediately with a job id, so you can \ + keep working while it runs. `arguments` is the same object the tool takes when called \ + directly. Backgroundable tools: external command tools (e.g. execute_command) and \ + `mcp_invoke_*` calls; built-in `agent__`/`job__`/`user__`/`todo__`/`memory__`/`skill__` \ + tools cannot be backgrounded. The job runs against a snapshot of the current config and \ + environment; later changes do not affect it. Process jobs honor COYOTE_TOOL_TIMEOUT; MCP \ + jobs have NO timeout — cancel a hung one with `job__cancel`. Jobs do not survive coyote \ + exiting.".to_string(), + parameters: JsonSchema { + type_value: Some("object".to_string()), + properties: Some(IndexMap::from([ + ( + "tool".to_string(), + JsonSchema { + type_value: Some("string".to_string()), + description: Some("Name of the tool to run in the background, exactly as it appears in your tool catalog".into()), + ..Default::default() + }, + ), + ( + "arguments".to_string(), + JsonSchema { + type_value: Some("object".to_string()), + description: Some("The arguments object the tool takes when called directly".into()), + ..Default::default() + }, + ), + ])), + required: Some(vec!["tool".to_string()]), + ..Default::default() + }, + agent: false, + }, + FunctionDeclaration { + name: format!("{JOB_FUNCTION_PREFIX}check"), + description: "Non-blocking status probe for a background job. Returns status, elapsed time, and a tail \ + of the output captured so far; it NEVER consumes the result — use `job__collect` for \ + that. Call sparingly: if repeated checks show no change, do other work instead — you \ + will be notified when the job completes.".to_string(), + parameters: JsonSchema { + type_value: Some("object".to_string()), + properties: Some(IndexMap::from([( + "id".to_string(), + JsonSchema { + type_value: Some("string".to_string()), + description: Some("The job ID returned by job__start".into()), + ..Default::default() + }, + )])), + required: Some(vec!["id".to_string()]), + ..Default::default() + }, + agent: false, + }, + FunctionDeclaration { + name: format!("{JOB_FUNCTION_PREFIX}collect"), + description: "Block until the named background job finishes, then return its result and remove the \ + job. The result keeps the LAST 50,000 chars by default (failures land at the tail of \ + build logs); pass `tail_lines` to keep only the last N lines instead.".to_string(), + parameters: JsonSchema { + type_value: Some("object".to_string()), + properties: Some(IndexMap::from([ + ( + "id".to_string(), + JsonSchema { + type_value: Some("string".to_string()), + description: Some("The job ID returned by job__start".into()), + ..Default::default() + }, + ), + ( + "tail_lines".to_string(), + JsonSchema { + type_value: Some("number".to_string()), + description: Some("Keep only the last N lines of the result".into()), + ..Default::default() + }, + ), + ])), + required: Some(vec!["id".to_string()]), + ..Default::default() + }, + agent: false, + }, + FunctionDeclaration { + name: format!("{JOB_FUNCTION_PREFIX}cancel"), + description: "Cancel a background job: kills its process (SIGTERM, then SIGKILL after a 5s grace) and \ + discards the handle. Returns any partial output captured so far. The id cannot be used \ + afterwards.".to_string(), + parameters: JsonSchema { + type_value: Some("object".to_string()), + properties: Some(IndexMap::from([( + "id".to_string(), + JsonSchema { + type_value: Some("string".to_string()), + description: Some("The job ID returned by job__start".into()), + ..Default::default() + }, + )])), + required: Some(vec!["id".to_string()]), + ..Default::default() + }, + agent: false, + }, + FunctionDeclaration { + name: format!("{JOB_FUNCTION_PREFIX}list"), + description: "List all background jobs you have started that are still registered, with status, \ + elapsed time, and bytes of output captured.".to_string(), + parameters: JsonSchema { + type_value: Some("object".to_string()), + properties: Some(IndexMap::new()), + ..Default::default() + }, + agent: false, + }, + ] +} + +pub async fn handle_job_tool( + ctx: &mut RequestContext, + cmd_name: &str, + args: &Value, +) -> Result { + let action = cmd_name + .strip_prefix(JOB_FUNCTION_PREFIX) + .unwrap_or(cmd_name); + + match action { + "start" => handle_start(ctx, args).await, + "check" => handle_check(ctx, args), + "collect" => handle_collect(ctx, args).await, + "cancel" => handle_cancel(ctx, args).await, + "list" => handle_list(ctx), + _ => bail!("Unknown job action: {action}"), + } +} + +fn job_status_str(status: JobStatus) -> &'static str { + match status { + JobStatus::Running => "running", + JobStatus::Completed => "completed", + JobStatus::Failed => "failed", + } +} + +fn job_miss_error(supervisor: Option<&Arc>>, id: &str) -> Value { + if is_agent_task(supervisor, id) { + json!({ + "status": "error", + "message": format!( + "'{id}' is a spawned agent, not a background job — use agent__check / agent__collect / agent__cancel" + ), + }) + } else { + json!({ + "status": "error", + "message": format!( + "No job '{id}' is registered — it may have already been collected or cancelled. job__list shows active jobs." + ), + }) + } +} + +fn whitelist_rejection(tool: &str) -> Option { + let non_invoke_mcp_prefixes = [ + MCP_SEARCH_META_FUNCTION_NAME_PREFIX, + MCP_DESCRIBE_META_FUNCTION_NAME_PREFIX, + MCP_READ_META_FUNCTION_NAME_PREFIX, + MCP_PROMPT_META_FUNCTION_NAME_PREFIX, + ]; + let reason = if tool.starts_with(SUPERVISOR_FUNCTION_PREFIX) + || tool.starts_with(JOB_FUNCTION_PREFIX) + { + Some(format!( + "'{tool}' is already asynchronous — call it directly. Agents may start jobs, but jobs never start agents or other jobs." + )) + } else if tool.starts_with(USER_FUNCTION_PREFIX) { + Some(format!( + "'{tool}' is interactive and must run in-turn — a background job cannot touch the terminal. Call it directly." + )) + } else if tool.starts_with(TODO_FUNCTION_PREFIX) + || tool.starts_with(MEMORY_FUNCTION_PREFIX) + || tool.starts_with(SKILL_FUNCTION_PREFIX) + || tool.starts_with(RAG_FUNCTION_PREFIX) + { + Some(format!( + "'{tool}' mutates agent/session state and must run in-turn. Call it directly." + )) + } else if non_invoke_mcp_prefixes + .iter() + .any(|prefix| tool.starts_with(prefix)) + { + Some(format!( + "'{tool}' is a sub-second call; invoke it directly." + )) + } else { + None + }; + + reason.map(|why| { + json!({ + "status": "error", + "message": format!( + "{why} Backgroundable tools: external command tools (e.g. execute_command) and mcp_invoke_* calls." + ), + }) + }) +} + +async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result { + if !jobs_enabled(ctx.agent.as_ref(), &ctx.app.config) { + return Ok(json!({ + "status": "error", + "message": "Background jobs are disabled in this context (max_concurrent_jobs is 0).", + })); + } + + let tool = args + .get("tool") + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()) + .ok_or_else(|| anyhow!("'tool' is required"))? + .to_string(); + let arguments = args.get("arguments").cloned().unwrap_or_else(|| json!({})); + + if let Some(rejection) = whitelist_rejection(&tool) { + return Ok(rejection); + } + + if !ctx.declared_function_names.contains(&tool) { + return Ok(json!({ + "status": "error", + "message": format!( + "'{tool}' is not enabled in this context — job__start can only background tools declared to you in this request. Use the exact name of a tool from your current catalog." + ), + })); + } + + let supervisor = match ctx.supervisor.as_ref() { + Some(sup) => Arc::clone(sup), + None => { + let max_jobs = effective_max_concurrent_jobs(ctx.agent.as_ref(), &ctx.app.config); + let sup = Arc::new(RwLock::new( + Supervisor::new(0, 0).with_max_concurrent_jobs(max_jobs), + )); + ctx.supervisor = Some(Arc::clone(&sup)); + sup + } + }; + + { + let sup = supervisor.read(); + if sup.active_job_count() >= sup.max_concurrent_jobs() { + return Ok(json!({ + "status": "error", + "message": format!( + "At capacity: {}/{} jobs running. Collect or cancel one first.", + sup.active_job_count(), + sup.max_concurrent_jobs() + ), + })); + } + } + + let short_uuid = &Uuid::new_v4().to_string()[..8]; + let job_id = format!("job_{short_uuid}"); + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + + let join_handle = if tool.starts_with(MCP_INVOKE_META_FUNCTION_NAME_PREFIX) { + let server = tool.replace(&format!("{MCP_INVOKE_META_FUNCTION_NAME_PREFIX}_"), ""); + let Some(server_handle) = ctx.tool_scope.mcp_runtime.get(&server) else { + return Ok(json!({ + "status": "error", + "message": format!("MCP server '{server}' is not connected in this context."), + })); + }; + let inner_tool = arguments + .get("tool") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("Missing 'tool' in arguments"))? + .to_string(); + let inner_args = arguments + .get("arguments") + .cloned() + .unwrap_or_else(|| json!({})); + let mut mcp_runtime = McpRuntime::new(); + mcp_runtime.insert(server.clone(), Arc::clone(server_handle)); + let job_ctx = JobCtx { + mcp_runtime, + current_depth: ctx.current_depth, + }; + let task_state = Arc::clone(&state); + tokio::spawn(async move { + let result = run_mcp_job(job_ctx, server, inner_tool, inner_args).await; + task_state.lock().status = match &result { + Ok(_) => JobStatus::Completed, + Err(_) => JobStatus::Failed, + }; + result + }) + } else { + let snapshot = build_env_snapshot(ctx, &tool, &arguments)?; + let task_state = Arc::clone(&state); + let task_buf = Arc::clone(&output_buf); + tokio::spawn(async move { + let result = run_process_job(snapshot, Arc::clone(&task_state), task_buf).await; + let mut job_state = task_state.lock(); + job_state.pgid = None; + job_state.status = match &result { + Ok(job_result) if job_result.exit_code == Some(0) => JobStatus::Completed, + _ => JobStatus::Failed, + }; + drop(job_state); + result + }) + }; + + let handle = JobHandle { + id: job_id.clone(), + tool: tool.clone(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state, + output_buf, + no_change_checks: 0, + }; + + // On a capacity race the handle is dropped here, which kills the process + // group and aborts the task. + if let Err(e) = supervisor.write().register(handle) { + return Ok(json!({ + "status": "error", + "message": format!("{e}"), + })); + } + + Ok(json!({ + "status": "ok", + "job_id": job_id, + "tool": tool, + "message": "Running in background. Check with job__check, block with job__collect, cancel with job__cancel. You will receive a system_notifications entry on completion. Jobs do not survive coyote exiting.", + })) +} + +fn handle_check(ctx: &RequestContext, args: &Value) -> Result { + let id = args + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("'id' is required"))?; + + let Some(supervisor) = ctx.supervisor.as_ref() else { + return Ok(job_miss_error(None, id)); + }; + let sup = supervisor.read(); + let Some(job) = sup.job(id) else { + drop(sup); + return Ok(job_miss_error(ctx.supervisor.as_ref(), id)); + }; + + let status = job.state.lock().status; + let (tail, total_written) = { + let buf = job.output_buf.lock(); + (buf.tail(), buf.total_written()) + }; + let tail_truncated = (tail.len() as u64) < total_written; + let mut result = json!({ + "status": job_status_str(status), + "id": id, + "tool": job.tool, + "elapsed_secs": job.started_at.elapsed().as_secs(), + "output_tail": String::from_utf8_lossy(&tail).to_string(), + "output_bytes_captured": total_written, + "tail_truncated": tail_truncated, + }); + if matches!(status, JobStatus::Running) { + result["message"] = json!( + "Job is still running. Call job__collect to block for the result, or do other work — you will be notified on completion." + ); + } else { + result["message"] = json!(format!( + "Job finished — retrieve the result with job__collect --id {id}" + )); + } + + Ok(result) +} + +async fn handle_collect(ctx: &RequestContext, args: &Value) -> Result { + let id = args + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("'id' is required"))?; + let tail_lines = args + .get("tail_lines") + .and_then(Value::as_u64) + .map(|n| n as usize); + + let Some(supervisor) = ctx.supervisor.as_ref().cloned() else { + return Ok(job_miss_error(None, id)); + }; + + let target_abort = { + let sup = supervisor.read(); + let Some(job) = sup.job(id) else { + drop(sup); + return Ok(job_miss_error(ctx.supervisor.as_ref(), id)); + }; + job.abort_signal.clone() + }; + + loop { + let is_finished = { + let sup = supervisor.read(); + sup.job(id).is_none_or(|job| job.join_handle.is_finished()) + }; + + if is_finished { + break; + } + + if let Some(queue) = ctx.root_escalation_queue() + && queue.has_pending() + { + let summary = queue.pending_summary(); + return Ok(json!({ + "status": "pending", + "id": id, + "message": format!("Job '{id}' is still running, but child agents have pending escalations that need your reply. Reply via agent__reply_escalation, then call job__collect again."), + "pending_escalations": summary, + })); + } + + if target_abort.aborted() { + let deadline = time::Instant::now() + Duration::from_secs(2); + while time::Instant::now() < deadline { + let is_finished = { + let sup = supervisor.read(); + sup.job(id).is_none_or(|job| job.join_handle.is_finished()) + }; + if is_finished { + break; + } + time::sleep(Duration::from_millis(50)).await; + } + break; + } + + tokio::select! { + _ = time::sleep(Duration::from_millis(200)) => {} + _ = wait_abort_signal(&target_abort) => {} + } + } + + let handle = { + let mut sup = supervisor.write(); + sup.take_job(id) + }; + + let Some(mut handle) = handle else { + return Ok(json!({ + "status": "error", + "message": format!("Job '{id}' completed but could not be collected. It may have been collected by another call."), + })); + }; + + let joined = (&mut handle.join_handle).await; + let tool = handle.tool.clone(); + let elapsed_secs = handle.started_at.elapsed().as_secs(); + let status = handle.state.lock().status; + let (tail, total_written) = { + let buf = handle.output_buf.lock(); + (buf.tail(), buf.total_written()) + }; + let output_tail = String::from_utf8_lossy(&tail).to_string(); + + let job_result = match joined { + Err(join_err) => { + return Ok(json!({ + "status": "failed", + "id": id, + "tool": tool, + "error": format!("Job task panicked: {join_err}"), + "output_tail": output_tail, + "output_bytes_captured": total_written, + })); + } + Ok(Err(e)) => { + return Ok(json!({ + "status": "failed", + "id": id, + "tool": tool, + "error": format!("{e}"), + "output_tail": output_tail, + "output_bytes_captured": total_written, + })); + } + Ok(Ok(job_result)) => job_result, + }; + + let (result_value, result_truncated) = cap_result(job_result.output, tail_lines); + let mut response = json!({ + "status": job_status_str(status), + "id": id, + "tool": tool, + "elapsed_secs": elapsed_secs, + "result": result_value, + "output_tail": output_tail, + "output_bytes_captured": job_result.output_bytes_captured, + }); + if let Some(exit_code) = job_result.exit_code { + response["exit_code"] = json!(exit_code); + } + if result_truncated { + response["result_truncated"] = json!(true); + } + + Ok(response) +} + +async fn handle_cancel(ctx: &RequestContext, args: &Value) -> Result { + let id = args + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("'id' is required"))?; + + let Some(supervisor) = ctx.supervisor.as_ref() else { + return Ok(job_miss_error(None, id)); + }; + + let handle = { + let mut sup = supervisor.write(); + sup.take_job(id) + }; + + let Some(mut handle) = handle else { + return Ok(job_miss_error(ctx.supervisor.as_ref(), id)); + }; + + handle.abort_signal.set_ctrlc(); + kill_job_with_grace(&mut handle).await; + + let tool = handle.tool.clone(); + let (tail, total_written) = { + let buf = handle.output_buf.lock(); + (buf.tail(), buf.total_written()) + }; + + Ok(json!({ + "status": "cancelled", + "id": id, + "tool": tool, + "output_tail": String::from_utf8_lossy(&tail).to_string(), + "output_bytes_captured": total_written, + })) +} + +/// SIGTERM the process group, give it a grace period, then SIGKILL. Every +/// group kill is gated on `state.pgid` still being set: the job task clears +/// it right after `wait()` reaps the child, and killing after the reap could +/// signal an innocent recycled pid. MCP jobs (no pgid) fall through to a +/// plain task abort. +async fn kill_job_with_grace(handle: &mut JobHandle) { + #[cfg(unix)] + { + let pgid = handle.state.lock().pgid; + if let Some(pgid) = pgid { + unsafe { libc::killpg(pgid, libc::SIGTERM) }; + if time::timeout(JOB_KILL_GRACE, &mut handle.join_handle) + .await + .is_ok() + { + return; + } + if handle.state.lock().pgid.is_some() { + unsafe { libc::killpg(pgid, libc::SIGKILL) }; + } + } + } + handle.join_handle.abort(); + let _ = time::timeout(JOB_KILL_GRACE, &mut handle.join_handle).await; +} + +fn handle_list(ctx: &RequestContext) -> Result { + let Some(supervisor) = ctx.supervisor.as_ref() else { + return Ok(json!({ + "active_jobs": 0, + "max_concurrent_jobs": effective_max_concurrent_jobs(ctx.agent.as_ref(), &ctx.app.config), + "jobs": [], + })); + }; + let sup = supervisor.read(); + + let jobs: Vec = sup + .jobs() + .map(|job| { + let status = job.state.lock().status; + json!({ + "id": job.id, + "tool": job.tool, + "status": job_status_str(status), + "elapsed_secs": job.started_at.elapsed().as_secs(), + "output_bytes_captured": job.output_buf.lock().total_written(), + }) + }) + .collect(); + + Ok(json!({ + "active_jobs": sup.active_job_count(), + "max_concurrent_jobs": sup.max_concurrent_jobs(), + "jobs": jobs, + })) +} + +/// Mirrors the foreground `extract_call_config` + `run_llm_function` env +/// assembly, resolved eagerly so the detached task owns everything it needs. +fn build_env_snapshot( + ctx: &RequestContext, + tool: &str, + arguments: &Value, +) -> Result { + let agent = ctx.agent.as_ref(); + let (cmd_name, mut cmd_args, mut envs) = match agent { + Some(agent) => match agent.functions().find(tool) { + Some(declaration) if declaration.agent => ( + format!("{}-{tool}", agent.name()), + vec![tool.to_string()], + agent.variable_envs(), + ), + Some(_) => (tool.to_string(), vec![], agent.variable_envs()), + None => (tool.to_string(), vec![], HashMap::new()), + }, + None => (tool.to_string(), vec![], HashMap::new()), + }; + + let mut bin_dirs: Vec = vec![]; + if let Some(agent) = agent { + let dir = paths::agent_bin_dir(agent.name()); + if dir.exists() { + bin_dirs.push(dir); + } + if graph::agent_has_graph(agent.name()) { + envs.insert("AUTO_CONFIRM".into(), "true".into()); + } + } else { + bin_dirs.push(paths::functions_bin_dir()); + } + let current_path = env::var("PATH").context("No PATH environment variable")?; + let prepend_path = bin_dirs + .iter() + .map(|v| format!("{}{PATH_SEP}", v.display())) + .collect::>() + .join(""); + envs.insert("PATH".into(), format!("{prepend_path}{current_path}")); + + let output_file = temp_file("-job-", ""); + envs.insert("LLM_OUTPUT".into(), output_file.display().to_string()); + envs.insert("CLICOLOR_FORCE".into(), "1".into()); + envs.insert("FORCE_COLOR".into(), "1".into()); + + cmd_args.push(arguments.to_string()); + + #[cfg(windows)] + let cmd_name = super::polyfill_cmd_name(&cmd_name, &bin_dirs); + + #[cfg(windows)] + let cmd_args = { + let mut args = cmd_args; + if let Some(json_data) = args.pop() { + let tool_data_file = temp_file("-tool-data-", ".json"); + fs::write(&tool_data_file, &json_data)?; + envs.insert( + "LLM_TOOL_DATA_FILE".into(), + tool_data_file.display().to_string(), + ); + } + args + }; + + let timeout_secs = env::var("COYOTE_TOOL_TIMEOUT") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(1800); + + Ok(JobEnvSnapshot { + cmd_name, + display_name: tool.to_string(), + cmd_args, + envs, + output_file, + timeout_secs, + }) +} + +async fn pump_into_ring(mut reader: impl AsyncReadExt + Unpin, output_buf: Arc>) { + let mut chunk = [0u8; 1024]; + loop { + match reader.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => output_buf.lock().push(&chunk[..n]), + } + } +} + +async fn run_process_job( + snapshot: JobEnvSnapshot, + state: Arc>, + output_buf: Arc>, +) -> Result { + let mut command = tokio::process::Command::new(&snapshot.cmd_name); + command + .args(&snapshot.cmd_args) + .envs(&snapshot.envs) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + #[cfg(unix)] + command.process_group(0); + + let mut child = command + .spawn() + .map_err(|err| anyhow!("Unable to run {}, {err}", snapshot.display_name))?; + + #[cfg(unix)] + if let Some(pid) = child.id() { + state.lock().pgid = Some(pid as i32); + } + + let stdout = child + .stdout + .take() + .ok_or_else(|| anyhow!("Failed to capture stdout"))?; + let stderr = child + .stderr + .take() + .ok_or_else(|| anyhow!("Failed to capture stderr"))?; + let stdout_pump = tokio::spawn(pump_into_ring(stdout, Arc::clone(&output_buf))); + let stderr_pump = tokio::spawn(pump_into_ring(stderr, Arc::clone(&output_buf))); + + let wait_result = if snapshot.timeout_secs > 0 { + match time::timeout(Duration::from_secs(snapshot.timeout_secs), child.wait()).await { + Ok(wait_result) => wait_result, + Err(_) => { + kill_expired_job(&mut child, &state).await; + state.lock().pgid = None; + let _ = stdout_pump.await; + let _ = stderr_pump.await; + let output_bytes_captured = output_buf.lock().total_written(); + let message = format!( + "Tool call '{}' timed out after {}s and was killed (set COYOTE_TOOL_TIMEOUT to adjust; 0 = unlimited)", + snapshot.display_name, snapshot.timeout_secs + ); + return Ok(JobResult { + output: json!({"tool_call_error": message}), + exit_code: None, + output_bytes_captured, + }); + } + } + } else { + child.wait().await + }; + let status = + wait_result.map_err(|err| anyhow!("Unable to run {}, {err}", snapshot.display_name))?; + // pid-reuse guard: the child is reaped, so a later group kill against + // this pgid could hit an innocent recycled pid. + state.lock().pgid = None; + let _ = stdout_pump.await; + let _ = stderr_pump.await; + let output_bytes_captured = output_buf.lock().total_written(); + + let exit_code = status.code(); + if exit_code.unwrap_or_default() != 0 { + let message = format!( + "Tool call '{}' exited with code {}", + snapshot.display_name, + exit_code.unwrap_or_default() + ); + let mut error_json = json!({"tool_call_error": message}); + if let Ok(contents) = fs::read_to_string(&snapshot.output_file) + && !contents.trim().is_empty() + { + error_json["output"] = json!(contents); + } + return Ok(JobResult { + output: error_json, + exit_code, + output_bytes_captured, + }); + } + + let mut output = Value::Null; + if snapshot.output_file.exists() { + let contents = fs::read_to_string(&snapshot.output_file) + .context("Failed to retrieve tool call output")?; + if !contents.is_empty() { + output = serde_json::from_str(&contents) + .ok() + .unwrap_or_else(|| json!({"output": contents})); + } + } + + Ok(JobResult { + output, + exit_code, + output_bytes_captured, + }) +} + +async fn kill_expired_job(child: &mut tokio::process::Child, state: &Arc>) { + #[cfg(unix)] + { + let pgid = state.lock().pgid; + if let Some(pgid) = pgid { + unsafe { libc::killpg(pgid, libc::SIGTERM) }; + if time::timeout(JOB_KILL_GRACE, child.wait()).await.is_err() { + unsafe { libc::killpg(pgid, libc::SIGKILL) }; + let _ = child.wait().await; + } + return; + } + } + #[cfg(not(unix))] + let _ = state; + let _ = child.start_kill(); + let _ = child.wait().await; +} + +async fn run_mcp_job( + job_ctx: JobCtx, + server: String, + tool: String, + arguments: Value, +) -> Result { + let raw = match job_ctx.mcp_runtime.invoke(&server, &tool, arguments).await { + Ok(raw) => raw, + Err(e) => { + if job_ctx.current_depth == 0 { + let error_msg = format!("MCP job invocation failed: {e}"); + eprintln!("{}", muted_warning_text(&mcp_error_display(&error_msg))); + } + return Err(e); + } + }; + let output = render_tool_result(serde_json::to_value(raw)?, &server)?; + Ok(JobResult { + output, + exit_code: None, + output_bytes_captured: 0, + }) +} + +/// Tail-biased result capping owned by the collect handler: keeps the LAST +/// `tail_lines`/50,000 chars (build failures land at the tail), always cutting +/// on a char boundary. +fn cap_result(output: Value, tail_lines: Option) -> (Value, bool) { + if output.is_null() { + return (json!("DONE"), false); + } + let mut text = match &output { + Value::String(s) => s.clone(), + other => other.to_string(), + }; + let mut truncated = false; + if let Some(n) = tail_lines { + let lines: Vec<&str> = text.lines().collect(); + if lines.len() > n { + text = lines[lines.len() - n..].join("\n"); + truncated = true; + } + } + if let Some(capped) = tail_chars(&text, JOB_RESULT_TAIL_CAP_CHARS) { + text = capped; + truncated = true; + } + if truncated { + (json!(text), true) + } else { + (output, false) + } +} + +fn tail_chars(text: &str, max_chars: usize) -> Option { + let total = text.chars().count(); + if total <= max_chars { + return None; + } + let cut = text + .char_indices() + .nth(total - max_chars) + .map(|(i, _)| i) + .unwrap_or(0); + Some(format!( + "[truncated: kept last {max_chars} of {total} chars]\n{}", + &text[cut..] + )) +} + #[cfg(test)] mod tests { use super::*; + use crate::config::{AppConfig, AppState, WorkingMode}; + use crate::function::supervisor::{ + GuardrailAction, check_pending_agents_guardrail, handle_supervisor_tool, + }; + use crate::supervisor::mailbox::Inbox; + use crate::supervisor::{AgentExitStatus, AgentHandle, AgentResult}; + use std::future::Future; + + fn default_app_state() -> Arc { + Arc::new(AppState::test_default()) + } + + fn app_state_with_config(update: impl FnOnce(&mut AppConfig)) -> Arc { + let mut state = AppState::test_default(); + let mut config = (*state.config).clone(); + update(&mut config); + state.config = Arc::new(config); + Arc::new(state) + } + + fn plain_ctx() -> RequestContext { + RequestContext::new(default_app_state(), WorkingMode::Cmd) + } + + fn ctx_with_job_supervisor(max_jobs: usize) -> RequestContext { + let mut ctx = plain_ctx(); + ctx.supervisor = Some(Arc::new(RwLock::new( + Supervisor::new(0, 3).with_max_concurrent_jobs(max_jobs), + ))); + ctx + } + + fn make_running_job(id: &str) -> JobHandle { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let join_handle = rt.spawn(async { + Ok(JobResult { + output: Value::Null, + exit_code: Some(0), + output_bytes_captured: 0, + }) + }); + std::mem::forget(rt); + JobHandle { + id: id.to_string(), + tool: "execute_command".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })), + output_buf: Arc::new(Mutex::new(RingBuf::default())), + no_change_checks: 0, + } + } + + fn run_async(f: F) -> F::Output { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(f) + } + + #[cfg(unix)] + fn test_snapshot(cmd: &str, args: &[&str], timeout_secs: u64) -> JobEnvSnapshot { + let output_file = temp_file("-job-test-", ""); + let mut envs = HashMap::new(); + envs.insert("PATH".to_string(), env::var("PATH").unwrap()); + envs.insert("LLM_OUTPUT".to_string(), output_file.display().to_string()); + JobEnvSnapshot { + cmd_name: cmd.to_string(), + display_name: cmd.to_string(), + cmd_args: args.iter().map(|s| s.to_string()).collect(), + envs, + output_file, + timeout_secs, + } + } #[test] fn ring_buf_returns_contents_below_capacity() { @@ -125,10 +1169,6 @@ mod tests { #[test] fn is_agent_task_matches_registered_agents() { - use crate::supervisor::mailbox::Inbox; - use crate::supervisor::{AgentExitStatus, AgentHandle, AgentResult}; - use crate::utils::create_abort_signal; - let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -158,4 +1198,648 @@ mod tests { assert!(is_agent_task(Some(&sup), "a1")); assert!(!is_agent_task(Some(&sup), "missing")); } + + #[test] + fn job_function_declarations_cover_all_five_actions() { + let names: Vec = job_function_declarations() + .into_iter() + .map(|d| d.name) + .collect(); + assert_eq!( + names, + vec![ + "job__start", + "job__check", + "job__collect", + "job__cancel", + "job__list" + ] + ); + } + + #[test] + fn whitelist_rejects_state_mutating_tools() { + for tool in ["memory__write", "todo__add", "skill__load", "rag__query"] { + let rejection = whitelist_rejection(tool).unwrap(); + let message = rejection["message"].as_str().unwrap(); + assert!( + message.contains("mutates agent/session state"), + "unexpected message for {tool}: {message}" + ); + } + } + + #[test] + fn whitelist_rejects_async_and_interactive_tools() { + for tool in ["agent__spawn", "job__check"] { + let message = whitelist_rejection(tool).unwrap()["message"] + .as_str() + .unwrap() + .to_string(); + assert!( + message.contains("already asynchronous"), + "unexpected message for {tool}: {message}" + ); + } + let message = whitelist_rejection("user__select").unwrap()["message"] + .as_str() + .unwrap() + .to_string(); + assert!(message.contains("interactive")); + } + + #[test] + fn whitelist_rejects_fast_mcp_meta_tools() { + for tool in [ + "mcp_search_github", + "mcp_describe_github", + "mcp_read_github", + "mcp_prompt_github", + ] { + let message = whitelist_rejection(tool).unwrap()["message"] + .as_str() + .unwrap() + .to_string(); + assert!( + message.contains("sub-second"), + "unexpected message for {tool}: {message}" + ); + } + } + + #[test] + fn whitelist_allows_external_and_mcp_invoke_tools() { + assert!(whitelist_rejection("execute_command").is_none()); + assert!(whitelist_rejection("mcp_invoke_github").is_none()); + } + + #[test] + fn handle_start_rejects_non_whitelisted_tool_without_spawn() { + let mut ctx = plain_ctx(); + ctx.declared_function_names.insert("memory__write".into()); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "memory__write", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("mutates agent/session state") + ); + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } + + #[test] + fn handle_start_rejects_undeclared_tool_without_spawn() { + let mut ctx = plain_ctx(); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "execute_command", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("not enabled in this context") + ); + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } + + #[test] + fn handle_start_rejects_when_jobs_disabled() { + let app_state = app_state_with_config(|config| config.max_concurrent_jobs = Some(0)); + let mut ctx = RequestContext::new(app_state, WorkingMode::Cmd); + ctx.declared_function_names.insert("execute_command".into()); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "execute_command", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!(result["message"].as_str().unwrap().contains("disabled")); + assert!(ctx.supervisor.is_none()); + } + + #[test] + fn handle_start_rejects_at_capacity_without_spawn() { + let mut ctx = ctx_with_job_supervisor(1); + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(make_running_job("j1")) + .unwrap(); + ctx.declared_function_names.insert("execute_command".into()); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "execute_command", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("At capacity: 1/1") + ); + assert_eq!( + ctx.supervisor.as_ref().unwrap().read().active_job_count(), + 1 + ); + } + + #[test] + fn handle_start_rejects_unconnected_mcp_server() { + let mut ctx = plain_ctx(); + ctx.declared_function_names + .insert("mcp_invoke_github".into()); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "mcp_invoke_github", "arguments": {"tool": "search"}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("not connected") + ); + assert_eq!( + ctx.supervisor.as_ref().unwrap().read().active_job_count(), + 0 + ); + } + + #[cfg(unix)] + #[test] + fn handle_start_lazy_inits_supervisor_and_collect_returns_result() { + run_async(async { + let mut ctx = plain_ctx(); + ctx.declared_function_names.insert("echo".into()); + + let started = handle_start(&mut ctx, &json!({"tool": "echo", "arguments": {}})) + .await + .unwrap(); + + assert_eq!(started["status"], "ok"); + let job_id = started["job_id"].as_str().unwrap().to_string(); + assert!(job_id.starts_with("job_")); + assert!( + ctx.supervisor.is_some(), + "plain sessions lazily init a supervisor" + ); + + let collected = handle_collect(&ctx, &json!({"id": job_id})).await.unwrap(); + + assert_eq!(collected["status"], "completed"); + assert_eq!(collected["result"], "DONE"); + assert_eq!(collected["exit_code"], 0); + assert!(collected["output_tail"].as_str().unwrap().contains("{}")); + assert!(!ctx.supervisor.as_ref().unwrap().read().has_job(&job_id)); + }); + } + + #[test] + fn handle_check_unknown_id_teaches_job_list() { + let ctx = ctx_with_job_supervisor(4); + let result = handle_check(&ctx, &json!({"id": "job_x"})).unwrap(); + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("No job 'job_x' is registered") + ); + } + + #[test] + fn job_handlers_teach_cross_kind_for_agent_ids() { + let ctx = ctx_with_job_supervisor(4); + for result in [ + handle_check(&ctx, &json!({"id": "agent_explore_1"})).unwrap(), + run_async(handle_collect(&ctx, &json!({"id": "agent_explore_1"}))).unwrap(), + run_async(handle_cancel(&ctx, &json!({"id": "agent_explore_1"}))).unwrap(), + ] { + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("is a spawned agent, not a background job") + ); + } + } + + #[test] + fn job_handlers_miss_without_supervisor() { + let ctx = plain_ctx(); + let result = handle_check(&ctx, &json!({"id": "job_x"})).unwrap(); + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("No job 'job_x'") + ); + } + + #[test] + fn handle_check_reports_running_job_tail() { + let ctx = ctx_with_job_supervisor(4); + let job = make_running_job("j1"); + job.output_buf.lock().push(b"hello"); + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(job) + .unwrap(); + + let result = handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + + assert_eq!(result["status"], "running"); + assert_eq!(result["tool"], "execute_command"); + assert_eq!(result["output_tail"], "hello"); + assert_eq!(result["output_bytes_captured"], 5); + assert_eq!(result["tail_truncated"], false); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("still running") + ); + } + + #[test] + fn handle_check_finished_job_points_at_collect_without_consuming() { + let ctx = ctx_with_job_supervisor(4); + let job = make_running_job("j1"); + job.state.lock().status = JobStatus::Completed; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(job) + .unwrap(); + + let result = handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + + assert_eq!(result["status"], "completed"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("job__collect --id j1") + ); + assert!(ctx.supervisor.as_ref().unwrap().read().has_job("j1")); + } + + #[test] + fn handle_collect_applies_tail_lines() { + run_async(async { + let ctx = ctx_with_job_supervisor(4); + let join_handle = tokio::spawn(async { + Ok(JobResult { + output: json!("l1\nl2\nl3"), + exit_code: Some(0), + output_bytes_captured: 0, + }) + }); + let handle = JobHandle { + id: "j1".to_string(), + tool: "execute_command".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::new(Mutex::new(JobState { + status: JobStatus::Completed, + pgid: None, + })), + output_buf: Arc::new(Mutex::new(RingBuf::default())), + no_change_checks: 0, + }; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(handle) + .unwrap(); + + let result = handle_collect(&ctx, &json!({"id": "j1", "tail_lines": 2})) + .await + .unwrap(); + + assert_eq!(result["status"], "completed"); + assert_eq!(result["result"], "l2\nl3"); + assert_eq!(result["result_truncated"], true); + }); + } + + #[test] + fn handle_collect_maps_panic_to_failed_with_ring_content() { + run_async(async { + let ctx = ctx_with_job_supervisor(4); + let join_handle: tokio::task::JoinHandle> = + tokio::spawn(async { panic!("boom") }); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + output_buf.lock().push(b"partial"); + let handle = JobHandle { + id: "j1".to_string(), + tool: "execute_command".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::new(Mutex::new(JobState { + status: JobStatus::Failed, + pgid: None, + })), + output_buf, + no_change_checks: 0, + }; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(handle) + .unwrap(); + + let result = handle_collect(&ctx, &json!({"id": "j1"})).await.unwrap(); + + assert_eq!(result["status"], "failed"); + assert!(result["error"].as_str().unwrap().contains("panicked")); + assert_eq!(result["output_tail"], "partial"); + }); + } + + #[cfg(unix)] + #[test] + fn handle_cancel_kills_running_process_job() { + run_async(async { + let ctx = ctx_with_job_supervisor(4); + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + let snapshot = test_snapshot("sleep", &["30"], 0); + let task_state = Arc::clone(&state); + let task_buf = Arc::clone(&output_buf); + let join_handle = + tokio::spawn(async move { run_process_job(snapshot, task_state, task_buf).await }); + + let deadline = time::Instant::now() + Duration::from_secs(2); + while state.lock().pgid.is_none() && time::Instant::now() < deadline { + time::sleep(Duration::from_millis(10)).await; + } + assert!(state.lock().pgid.is_some(), "runner must record the pgid"); + + let handle = JobHandle { + id: "j1".to_string(), + tool: "sleep".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::clone(&state), + output_buf, + no_change_checks: 0, + }; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(handle) + .unwrap(); + + let result = handle_cancel(&ctx, &json!({"id": "j1"})).await.unwrap(); + + assert_eq!(result["status"], "cancelled"); + assert_eq!(result["tool"], "sleep"); + assert!(!ctx.supervisor.as_ref().unwrap().read().has_job("j1")); + }); + } + + #[test] + fn handle_list_reports_jobs_and_capacity() { + let ctx = ctx_with_job_supervisor(4); + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(make_running_job("j1")) + .unwrap(); + + let result = handle_list(&ctx).unwrap(); + + assert_eq!(result["active_jobs"], 1); + assert_eq!(result["max_concurrent_jobs"], 4); + assert_eq!(result["jobs"][0]["id"], "j1"); + assert_eq!(result["jobs"][0]["status"], "running"); + } + + #[test] + fn handle_list_without_supervisor_reports_empty() { + let ctx = plain_ctx(); + let result = handle_list(&ctx).unwrap(); + assert_eq!(result["active_jobs"], 0); + assert_eq!(result["max_concurrent_jobs"], 5); + assert_eq!(result["jobs"].as_array().unwrap().len(), 0); + } + + #[test] + fn cap_result_normalizes_null_to_done() { + let (value, truncated) = cap_result(Value::Null, None); + assert_eq!(value, json!("DONE")); + assert!(!truncated); + } + + #[test] + fn cap_result_preserves_small_values() { + let (value, truncated) = cap_result(json!({"a": 1}), None); + assert_eq!(value, json!({"a": 1})); + assert!(!truncated); + } + + #[test] + fn cap_result_keeps_last_chars_on_char_boundary() { + let text = "é".repeat(JOB_RESULT_TAIL_CAP_CHARS + 10); + let (value, truncated) = cap_result(json!(text), None); + assert!(truncated); + let capped = value.as_str().unwrap(); + assert!(capped.starts_with(&format!( + "[truncated: kept last {} of {} chars]", + JOB_RESULT_TAIL_CAP_CHARS, + JOB_RESULT_TAIL_CAP_CHARS + 10 + ))); + } + + #[test] + fn tail_chars_floors_to_char_boundary() { + let capped = tail_chars("aébc", 2).unwrap(); + assert!(capped.ends_with("bc")); + assert!(capped.starts_with("[truncated: kept last 2 of 4 chars]")); + assert!(tail_chars("abc", 3).is_none()); + } + + #[cfg(unix)] + #[test] + fn run_process_job_times_out_and_clears_pgid() { + run_async(async { + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + let snapshot = test_snapshot("sleep", &["30"], 1); + + let result = run_process_job(snapshot, Arc::clone(&state), output_buf) + .await + .unwrap(); + + assert_eq!(result.exit_code, None); + assert!( + result.output["tool_call_error"] + .as_str() + .unwrap() + .contains("timed out after 1s") + ); + assert!( + state.lock().pgid.is_none(), + "pid-reuse guard must clear pgid" + ); + }); + } + + #[cfg(unix)] + #[test] + fn run_process_job_reads_llm_output_and_captures_ring() { + run_async(async { + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + let snapshot = test_snapshot( + "sh", + &["-c", "printf hi > \"$LLM_OUTPUT\"; echo captured"], + 0, + ); + + let result = run_process_job(snapshot, state, Arc::clone(&output_buf)) + .await + .unwrap(); + + assert_eq!(result.exit_code, Some(0)); + assert_eq!(result.output, json!({"output": "hi"})); + let tail = String::from_utf8_lossy(&output_buf.lock().tail()).to_string(); + assert!(tail.contains("captured")); + assert_eq!(result.output_bytes_captured, 9); + }); + } + + #[cfg(unix)] + #[test] + fn run_process_job_reports_nonzero_exit_with_partial_output() { + run_async(async { + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + let snapshot = + test_snapshot("sh", &["-c", "printf partial > \"$LLM_OUTPUT\"; exit 3"], 0); + + let result = run_process_job(snapshot, state, output_buf).await.unwrap(); + + assert_eq!(result.exit_code, Some(3)); + assert!( + result.output["tool_call_error"] + .as_str() + .unwrap() + .contains("exited with code 3") + ); + assert_eq!(result.output["output"], "partial"); + }); + } + + #[test] + fn jobs_only_supervisor_rejects_agent_spawn_at_capacity_zero() { + let mut ctx = ctx_with_job_supervisor(5); + + let result = run_async(handle_supervisor_tool( + &mut ctx, + "agent__spawn", + &json!({"agent": "explore", "prompt": "x"}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("At capacity: 0/0") + ); + } + + #[test] + fn jobs_only_supervisor_guardrail_takes_no_action() { + let mut ctx = ctx_with_job_supervisor(5); + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(make_running_job("j1")) + .unwrap(); + + assert!(matches!( + check_pending_agents_guardrail(&mut ctx), + GuardrailAction::NoAction + )); + } + + #[test] + fn jobs_only_supervisor_agent_surfaces_stay_functional() { + let mut ctx = ctx_with_job_supervisor(5); + + let listed = run_async(handle_supervisor_tool( + &mut ctx, + "agent__list_running", + &json!({}), + )) + .unwrap(); + assert_eq!(listed["active_count"], 0); + assert_eq!(listed["max_concurrent"], 0); + + let created = run_async(handle_supervisor_tool( + &mut ctx, + "agent__task_create", + &json!({"subject": "research"}), + )) + .unwrap(); + assert_eq!(created["status"], "ok"); + + let tasks = run_async(handle_supervisor_tool( + &mut ctx, + "agent__task_list", + &json!({}), + )) + .unwrap(); + assert_eq!(tasks["tasks"].as_array().unwrap().len(), 1); + } } diff --git a/src/function/mod.rs b/src/function/mod.rs index 579c430..78ee969 100644 --- a/src/function/mod.rs +++ b/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(); diff --git a/src/supervisor/mod.rs b/src/supervisor/mod.rs index dfa072d..ac143b5 100644 --- a/src/supervisor/mod.rs +++ b/src/supervisor/mod.rs @@ -40,7 +40,6 @@ pub struct AgentHandle { pub child_supervisor: Option>>, } -#[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, } -#[allow(dead_code)] pub struct JobResult { pub output: Value, pub exit_code: Option, @@ -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>, pub abort_signal: AbortSignal, pub state: Arc>, - #[allow(dead_code)] pub output_buf: Arc>, #[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 { + 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 }