Merge pull request #19 from Dark-Alex-17/feat/background-jobs

feat: background jobs (job__* tools) with push notifications

Cargo.lock

@@ -1703,6 +1703,7 @@ dependencies = [
  "inquire",
  "is-terminal",
  "json-patch",
+ "libc",
  "log",
  "log4rs",
  "nu-ansi-term",

Cargo.toml

@@ -139,6 +139,9 @@ arboard = { version = "3.3.0", default-features = false, features = [
 [target.'cfg(not(any(target_os = "linux", target_os = "android", target_os = "emscripten")))'.dependencies]
 arboard = { version = "3.3.0", default-features = false }
 
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"
+
 [dev-dependencies]
 pretty_assertions = "1.4.0"
 rmcp = { version = "3.1.2", features = ["server"] }

README.md

@@ -51,6 +51,7 @@ Coming from [AIChat](https://github.com/sigoden/aichat)? Follow the [migration g
 * [Skills](https://github.com/Dark-Alex-17/coyote/wiki/Skills): Modular knowledge or capability packs the LLM can load and unload mid-conversation. Multiple skills compose; instructions stack, tools and MCPs union.
 * [Agents](https://github.com/Dark-Alex-17/coyote/wiki/Agents): Leverage AI agents to perform complex tasks and workflows, including sub-agent spawning, teammate messaging, and user interaction tools.
     * [Graph Agents](https://github.com/Dark-Alex-17/coyote/wiki/Graph-Agents): Define an agent as a declarative, YAML-driven workflow. A directed graph of typed nodes (LLM calls, scripts, approvals, user input, RAG retrieval, sub-agent spawns).
+* [Background Jobs](https://github.com/Dark-Alex-17/coyote/wiki/Background-Jobs): Run long tool calls (builds, test suites, slow MCP calls) in the background with the `job__*` tools while the model keeps working, and completion arrives as a push notification.
 * [Todo System](https://github.com/Dark-Alex-17/coyote/wiki/TODO-System): Built-in task tracking for improved LLM reliability with smaller models.
 * [Environment Variables](https://github.com/Dark-Alex-17/coyote/wiki/Environment-Variables): Override and customize your Coyote configuration at runtime with environment variables.
 * [Client Configurations](https://github.com/Dark-Alex-17/coyote/wiki/Clients): Configuration instructions for various LLM providers.

assets/agents/architect/config.yaml

@@ -261,7 +261,7 @@ instructions: |
 
   3. **Wait for Sisyphus.** Do not poll `agent__collect` on a running agent — do non-overlapping work
      (e.g. prep the next task's context) or end your response and wait for the completion
-     notification, then `agent__collect`.
+     notification (a `system_notifications` entry on your next tool result), then `agent__collect`.
 
   4. **Verify against the plan (divergence check).** When Sisyphus returns, do NOT trust its
      self-report — get an INDEPENDENT conformance verdict:

assets/agents/sisyphus/config.yaml

@@ -233,7 +233,7 @@ instructions: |
 
   1. Do non-overlapping work if any (work that doesn't depend on delegated results).
   2. If none → **end your response.** Do not call `agent__collect` immediately.
-  3. The system notifies you on completion.
+  3. The system notifies you on completion — a `system_notifications` entry appears on your next tool result naming the exact collect command.
   4. On notification, call `agent__collect` to retrieve results.
 
   ### Anti-duplication rule (BLOCKING)

assets/functions/tools/execute_command.sh

@@ -20,5 +20,12 @@ main() {
     trap "rm -f '$script'" EXIT
     # shellcheck disable=SC2154
     printf '%s\n' "$argc_command" > "$script"
-    bash -e -o pipefail "$script" >> "$LLM_OUTPUT"
+    # No -e: the command gets standard interactive-shell semantics — the last
+    # statement decides the exit code, so trailing guards like `; exit 0` work
+    # and an intermediate non-zero status (grep with no matches, a failing
+    # test run being inspected) cannot abort the script mid-way. pipefail is
+    # kept so a failing pipeline stage still surfaces in the exit code. 2>&1:
+    # the harness only returns $LLM_OUTPUT on success, so without it stderr
+    # (git push, cargo progress, curl -v) vanishes from successful calls.
+    bash -o pipefail "$script" >> "$LLM_OUTPUT" 2>&1
 }

config.agent.example.yaml

@@ -41,6 +41,8 @@ can_spawn_agents: false          # Enable the agent to spawn child agents
                                  # Graph agents (graph.yaml) ignore this; they declare spawn targets in agent nodes.
 max_concurrent_agents: 4         # Maximum number of agents that can run simultaneously
 max_agent_depth: 3               # Maximum nesting depth for sub-agents (prevents runaway spawning)
+max_concurrent_jobs: 5           # Max background jobs (`job__*` tools) running at once for this agent
+                                 # (overrides the global setting; 0 disables background jobs for this agent)
 inject_spawn_instructions: true  # Inject the default agent spawning instructions into the agent's system prompt
 summarization_model: null        # Model to use for summarizing sub-agent output (e.g. 'openai:gpt-4o-mini'); defaults to current model
 summarization_threshold: 4000    # Character threshold above which sub-agent output is summarized before returning to parent

config.example.yaml

@@ -204,6 +204,7 @@ summary_context_prompt: >        # The text prompt used for including the summar
   'This is a summary of the chat history as a recap: '
 compression_keep_last: 0         # Number of most-recent messages to keep visible after compression (0 = compress all messages)
 max_tool_result_chars: null      # Cap on tool result characters forwarded to the model per call (null = no cap)
+max_concurrent_jobs: 5           # Max background jobs (`job__*` tools) running at once per context (default: 5; 0 disables background jobs entirely)
 
 # ---- Memory ----
 # See the [Memory documentation](https://github.com/Dark-Alex-17/coyote/wiki/Memory) for more information.

graph.example.yaml

@@ -36,6 +36,14 @@ top_p: null                        # Default sampling top-p for `llm` nodes
 reasoning_effort: null             # Default reasoning effort for `llm` nodes that don't override it.
                                    # Only valid when the model declares reasoning_levels.
 
+max_concurrent_jobs: 5             # Max background jobs (`job__*` tools) running at once across the
+                                   # whole graph run: every `llm` node (including parallel branches)
+                                   # draws from this one pool, so the budget is graph-wide — there is
+                                   # no per-node override. Jobs themselves are node-local: the node
+                                   # that starts a job must collect or cancel it before it ends, and
+                                   # anything left running at node exit is cancelled. Overrides the
+                                   # global setting; 0 disables background jobs for this graph agent.
+
 global_tools:                      # Tool universe an `llm` node's `tools:` whitelist draws from
   - web_search_coyote.sh
   - fetch_url_via_curl.sh

src/acp/server.rs

@@ -1,7 +1,7 @@
 use super::types::{METHOD_NOT_FOUND, PARSE_ERROR, Request, Response};
 use crate::client::call_chat_completions_streaming;
 use crate::config::{Input, RenderMode, RequestContext};
-use crate::function::supervisor::{GuardrailAction, check_pending_agents_guardrail};
+use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail};
 use crate::utils;
 use crate::utils::AbortSignal;
 use anyhow::Result;
@@ -211,7 +211,7 @@ async fn run_prompt_turn(
             input = input.merge_tool_results(output, tool_results);
             continue;
         }
-        match check_pending_agents_guardrail(ctx) {
+        match check_pending_tasks_guardrail(ctx) {
             GuardrailAction::Inject(prompt) => {
                 input = Input::from_str(ctx, &prompt, None)?;
             }

src/config/agent.rs

@@ -3,15 +3,19 @@ use super::*;
 use crate::{
     client::Model,
     config::memory,
-    function::{Functions, run_llm_function},
+    function::{
+        Functions,
+        jobs::{DEFAULT_MAX_CONCURRENT_JOBS, JOB_FUNCTION_PREFIX},
+        run_llm_function,
+    },
     graph, rag,
 };
 
 use super::rag_cache::RagKey;
 use crate::config::paths;
 use crate::config::prompts::{
-    DEFAULT_SPAWN_INSTRUCTIONS, DEFAULT_TEAMMATE_INSTRUCTIONS, DEFAULT_TODO_INSTRUCTIONS,
-    DEFAULT_USER_INTERACTION_INSTRUCTIONS,
+    DEFAULT_JOB_INSTRUCTIONS, DEFAULT_SPAWN_INSTRUCTIONS, DEFAULT_TEAMMATE_INSTRUCTIONS,
+    DEFAULT_TODO_INSTRUCTIONS, DEFAULT_USER_INTERACTION_INSTRUCTIONS,
 };
 use crate::graph::types::RagNode;
 use crate::graph::{Graph, GraphParser, NodeType};
@@ -225,6 +229,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();
 
@@ -440,6 +454,18 @@ impl Agent {
             output.push_str(DEFAULT_SPAWN_INSTRUCTIONS);
         }
 
+        if self
+            .functions
+            .declarations()
+            .iter()
+            .any(|f| f.name.starts_with(JOB_FUNCTION_PREFIX))
+        {
+            if !output.ends_with('\n') {
+                output.push('\n');
+            }
+            output.push_str(DEFAULT_JOB_INSTRUCTIONS);
+        }
+
         output.push_str(DEFAULT_TEAMMATE_INSTRUCTIONS);
         output.push_str(DEFAULT_USER_INTERACTION_INSTRUCTIONS);
 
@@ -561,6 +587,10 @@ impl Agent {
         self.config.max_tool_result_chars
     }
 
+    pub fn max_concurrent_jobs(&self) -> Option<usize> {
+        self.config.max_concurrent_jobs
+    }
+
     pub fn compression_keep_last(&self) -> Option<usize> {
         self.config.compression_keep_last
     }
@@ -735,6 +765,8 @@ pub struct AgentConfig {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_tool_result_chars: Option<usize>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_concurrent_jobs: Option<usize>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub compression_keep_last: Option<usize>,
     #[serde(default)]
     pub description: String,
@@ -822,6 +854,7 @@ impl AgentConfig {
             variables: graph.variables.clone(),
             can_spawn_agents: graph.has_agent_node(),
             max_concurrent_agents: default_max_concurrent_agents(),
+            max_concurrent_jobs: graph.max_concurrent_jobs,
             max_agent_depth: default_max_agent_depth(),
             escalation_timeout: default_escalation_timeout(),
             ..AgentConfig::default()
@@ -1296,6 +1329,7 @@ variables:
             model: claude:claude-sonnet-4-6
             temperature: 0.3
             top_p: 0.8
+            max_concurrent_jobs: 2
             global_tools:
               - fetch_pdf.sh
             mcp_servers:
@@ -1318,6 +1352,7 @@ variables:
         assert_eq!(config.model_id.as_deref(), Some("claude:claude-sonnet-4-6"));
         assert_eq!(config.temperature, Some(0.3));
         assert_eq!(config.top_p, Some(0.8));
+        assert_eq!(config.max_concurrent_jobs, Some(2));
         assert_eq!(config.global_tools, vec!["fetch_pdf.sh"]);
         assert_eq!(config.mcp_servers, vec!["pubmed-search"]);
         assert_eq!(config.conversation_starters, vec!["Start here"]);
@@ -1481,4 +1516,51 @@ nodes: {}
         assert_eq!(config.top_k, Some(7));
         assert_eq!(config.embedding_model.as_deref(), Some("some:model"));
     }
+
+    #[test]
+    fn interpolated_instructions_without_job_declarations_is_byte_identical_across_job_settings() {
+        let agent = |max_concurrent_jobs| {
+            Agent::test_new(AgentConfig {
+                instructions: "hi".to_string(),
+                max_concurrent_jobs,
+                ..AgentConfig::default()
+            })
+        };
+
+        let baseline = agent(None).interpolated_instructions();
+        assert!(
+            !baseline.contains(DEFAULT_JOB_INSTRUCTIONS),
+            "no job guidance may be injected without job__ declarations"
+        );
+        assert_eq!(baseline, agent(Some(0)).interpolated_instructions());
+        assert_eq!(baseline, agent(Some(7)).interpolated_instructions());
+
+        let mut with_unrelated = agent(None);
+        with_unrelated.functions.append_todo_functions();
+        assert_eq!(
+            baseline,
+            with_unrelated.interpolated_instructions(),
+            "job guidance injection must key strictly on the job__ prefix"
+        );
+    }
+
+    #[test]
+    fn interpolated_instructions_with_job_declarations_appends_job_guidance() {
+        let config = AgentConfig {
+            instructions: "hi".to_string(),
+            ..AgentConfig::default()
+        };
+        let baseline = Agent::test_new(config.clone()).interpolated_instructions();
+
+        let mut agent = Agent::test_new(config);
+        agent.functions.append_job_functions();
+        let output = agent.interpolated_instructions();
+
+        assert!(output.contains(DEFAULT_JOB_INSTRUCTIONS));
+        let expected = format!(
+            "hi\n{DEFAULT_JOB_INSTRUCTIONS}{}",
+            baseline.strip_prefix("hi").unwrap()
+        );
+        assert_eq!(output, expected);
+    }
 }

src/config/app_config.rs

@@ -68,6 +68,7 @@ pub struct AppConfig {
     pub summarization_prompt: Option<String>,
     pub summary_context_prompt: Option<String>,
     pub max_tool_result_chars: Option<usize>,
+    pub max_concurrent_jobs: Option<usize>,
 
     pub memory: Option<bool>,
     pub memory_cap_with_tools: Option<usize>,
@@ -153,6 +154,7 @@ impl Default for AppConfig {
             summarization_prompt: None,
             summary_context_prompt: None,
             max_tool_result_chars: None,
+            max_concurrent_jobs: None,
 
             memory: None,
             memory_cap_with_tools: None,
@@ -239,6 +241,7 @@ impl AppConfig {
             summarization_prompt: config.summarization_prompt,
             summary_context_prompt: config.summary_context_prompt,
             max_tool_result_chars: config.max_tool_result_chars,
+            max_concurrent_jobs: config.max_concurrent_jobs,
 
             memory: config.memory,
             memory_cap_with_tools: config.memory_cap_with_tools,
@@ -574,6 +577,9 @@ impl AppConfig {
         {
             self.compression_threshold = v;
         }
+        if let Some(v) = super::read_env_value::<usize>(&get_env_name("max_concurrent_jobs")) {
+            self.max_concurrent_jobs = v;
+        }
         if let Some(v) = super::read_env_value::<String>(&get_env_name("summarization_prompt")) {
             self.summarization_prompt = v;
         }
@@ -838,16 +844,59 @@ mod tests {
 
         unsafe {
             match prev {
-                Some(v) => std::env::set_var(&env_name, v),
-                None => std::env::remove_var(&env_name),
+                Some(v) => env::set_var(&env_name, v),
+                None => env::remove_var(&env_name),
+            }
+        }
+    }
+
+    #[test]
+    fn from_config_copies_max_concurrent_jobs() {
+        let cfg = Config {
+            model_id: "test-model".to_string(),
+            max_concurrent_jobs: Some(3),
+            clients: vec![ClientConfig::default()],
+            ..Config::default()
+        };
+
+        let app = AppConfig::from_config(cfg).unwrap();
+
+        assert_eq!(app.max_concurrent_jobs, Some(3));
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn load_envs_overrides_max_concurrent_jobs() {
+        let env_name = get_env_name("max_concurrent_jobs");
+        let prev = env::var_os(&env_name);
+
+        let mut app = AppConfig::default();
+
+        unsafe { env::set_var(&env_name, "7") };
+        app.load_envs();
+        assert_eq!(app.max_concurrent_jobs, Some(7));
+
+        unsafe { env::set_var(&env_name, "0") };
+        app.load_envs();
+        assert_eq!(app.max_concurrent_jobs, Some(0));
+
+        unsafe { env::remove_var(&env_name) };
+        app.max_concurrent_jobs = Some(2);
+        app.load_envs();
+        assert_eq!(app.max_concurrent_jobs, Some(2));
+
+        unsafe {
+            match prev {
+                Some(v) => env::set_var(&env_name, v),
+                None => env::remove_var(&env_name),
             }
         }
     }
 
     #[test]
     fn editor_returns_configured_value() {
-        let configured = cached_editor()
-            .unwrap_or_else(|| std::env::current_exe().unwrap().display().to_string());
+        let configured =
+            cached_editor().unwrap_or_else(|| env::current_exe().unwrap().display().to_string());
         let app = AppConfig {
             editor: Some(configured.clone()),
             ..AppConfig::default()
@@ -864,9 +913,9 @@ mod tests {
             return;
         }
 
-        let expected = std::env::current_exe().unwrap().display().to_string();
+        let expected = env::current_exe().unwrap().display().to_string();
         unsafe {
-            std::env::set_var("VISUAL", &expected);
+            env::set_var("VISUAL", &expected);
         }
 
         let app = AppConfig::default();
@@ -934,7 +983,7 @@ mod tests {
         let app = AppConfig::from_config(cfg).unwrap();
 
         let ua = app.user_agent.as_deref().unwrap();
-        assert!(ua != "auto", "user_agent should have been resolved");
+        assert_ne!(ua, "auto", "user_agent should have been resolved");
         assert!(ua.contains('/'), "user_agent should be '<name>/<version>'");
     }
 

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;
@@ -73,6 +74,10 @@ impl AppState {
             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
         } else {

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,13 @@ impl Input {
         self.data_urls.clone()
     }
 
+    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
     }
@@ -593,6 +605,8 @@ mod tests {
     use super::*;
     use crate::config::request_context::RequestContext;
     use crate::config::{AppState, WorkingMode};
+    use crate::function::ToolCall;
+    use serde_json::json;
     use std::fs;
     use std::sync::Arc;
     use std::time::SystemTime;
@@ -973,4 +987,70 @@ mod tests {
         ));
         assert!(result.is_err());
     }
+
+    fn tool_result(id: &str, output: &str) -> ToolResult {
+        ToolResult::new(
+            ToolCall::new("t".into(), json!({}), Some(id.to_string())),
+            json!(output),
+        )
+    }
+
+    #[test]
+    fn merge_tool_results_first_merge_creates_container() {
+        let ctx = create_test_ctx();
+        let input = Input::from_str(&ctx, "test", None).unwrap();
+
+        let input =
+            input.merge_tool_results("assistant text".into(), vec![tool_result("id-1", "ok")]);
+
+        let tool_calls = input.tool_calls().as_ref().unwrap();
+        assert_eq!(tool_calls.text, "assistant text");
+        assert!(!tool_calls.sequence);
+        assert_eq!(tool_calls.tool_results.len(), 1);
+        assert!(tool_calls.tool_results[0].text.is_none());
+    }
+
+    #[test]
+    fn merge_tool_results_second_merge_marks_sequence_and_tags_text() {
+        let ctx = create_test_ctx();
+        let input = Input::from_str(&ctx, "test", None)
+            .unwrap()
+            .merge_tool_results("assistant text".into(), vec![tool_result("id-1", "ok")]);
+
+        let input =
+            input.merge_tool_results("second text".into(), vec![tool_result("id-2", "ok2")]);
+
+        let tool_calls = input.tool_calls().as_ref().unwrap();
+        assert!(tool_calls.sequence);
+        assert_eq!(tool_calls.tool_results.len(), 2);
+        assert_eq!(tool_calls.text, "assistant text");
+        assert!(tool_calls.tool_results[0].text.is_none());
+        assert_eq!(
+            tool_calls.tool_results[1].text,
+            Some("second text".to_string())
+        );
+    }
+
+    #[test]
+    fn build_messages_wraps_tool_results_in_single_assistant_message() {
+        let ctx = create_test_ctx();
+        let input = Input::from_str(&ctx, "test", None)
+            .unwrap()
+            .merge_tool_results("assistant text".into(), vec![tool_result("id-1", "ok")])
+            .merge_tool_results("second text".into(), vec![tool_result("id-2", "ok2")]);
+
+        let messages = input.build_messages().unwrap();
+
+        let tool_call_messages: Vec<_> = messages
+            .iter()
+            .filter(|m| matches!(m.content, MessageContent::ToolCalls(_)))
+            .collect();
+        assert_eq!(tool_call_messages.len(), 1);
+        let message = tool_call_messages[0];
+        assert!(matches!(message.role, MessageRole::Assistant));
+        let MessageContent::ToolCalls(tool_calls) = &message.content else {
+            unreachable!();
+        };
+        assert_eq!(tool_calls.tool_results.len(), 2);
+    }
 }

src/config/mod.rs

@@ -42,7 +42,10 @@ pub use self::macro_policy::{
     MacroAllowlistLevel, MacroPolicy, MacroSource, MacroState, RESERVED_MACRO_NAMES, ResolvedMacro,
 };
 #[allow(unused_imports)]
-pub use self::request_context::{RenderMode, RequestContext, should_inject_skill_instructions};
+pub use self::request_context::{
+    RenderMode, RequestContext, effective_max_concurrent_jobs, jobs_enabled,
+    should_inject_skill_instructions,
+};
 pub use self::role::{
     CODE_ROLE, CREATE_TITLE_ROLE, EXPLAIN_SHELL_ROLE, Role, RoleLike, SHELL_ROLE,
 };
@@ -56,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::{
@@ -264,6 +268,7 @@ pub struct Config {
     pub summarization_prompt: Option<String>,
     pub summary_context_prompt: Option<String>,
     pub max_tool_result_chars: Option<usize>,
+    pub max_concurrent_jobs: Option<usize>,
 
     pub memory: Option<bool>,
     pub memory_cap_with_tools: Option<usize>,
@@ -346,6 +351,7 @@ impl Default for Config {
             summarization_prompt: None,
             summary_context_prompt: None,
             max_tool_result_chars: None,
+            max_concurrent_jobs: None,
 
             memory: None,
             memory_cap_with_tools: None,

src/config/prompts.rs

@@ -82,7 +82,7 @@ pub(in crate::config) const DEFAULT_SPAWN_INSTRUCTIONS: &str = indoc! {"
     | Tool | Purpose |
     |------|----------|
     | `agent__spawn` | Spawn a subagent in the background. Returns an `id` immediately. |
-    | `agent__check` | Non-blocking check: is the agent done yet? Returns PENDING or result. |
+    | `agent__check` | Non-blocking status probe: running or finished. Never returns/consumes the result — use `agent__collect`. |
     | `agent__collect` | Blocking wait: wait for an agent to finish, return its output. |
     | `agent__list_available` | List all agent types you can spawn (name + description). Use this to discover specialists before calling `agent__spawn`. |
     | `agent__list_running` | List all subagents YOU have spawned, with their status. |
@@ -112,9 +112,10 @@ pub(in crate::config) const DEFAULT_SPAWN_INSTRUCTIONS: &str = indoc! {"
 
     ### CRITICAL: Never end your turn with pending agents
 
-    Spawned agents do NOT report back on their own. They run in the background until you
-    actively reclaim them with `agent__collect` (to get their output) or `agent__cancel`
-    (to discard them). If you spawn agents and then emit a final message without reclaiming
+    Spawned agents do NOT deliver their results on their own. When one finishes, a
+    `system_notifications` entry appears on your next tool result naming the exact collect
+    command — but the output is only retrieved when you actively reclaim it with `agent__collect`
+    (or discard it with `agent__cancel`). If you spawn agents and then emit a final message without reclaiming
     them, the system will detect the unreclaimed agents and reject the turn-end, injecting
     a reminder forcing you to handle them. After several such reminders, the system will
     auto-cancel them and warn you that work was lost.
@@ -190,6 +191,23 @@ pub(in crate::config) const DEFAULT_SPAWN_INSTRUCTIONS: &str = indoc! {"
     4. **Respond promptly**; the child agent is blocked and waiting (5-minute timeout).
 "};
 
+pub(in crate::config) const DEFAULT_JOB_INSTRUCTIONS: &str = indoc! {"
+    ## Background Jobs
+
+    For long-running tool calls (builds, test suites, slow commands), call `job__start` and keep
+    working instead of blocking — completion arrives as a `system_notifications` entry on your
+    next tool result. Check progress with `job__check` (sparingly), block on the result with
+    `job__collect` (only when you have nothing else to do), cancel with `job__cancel`, and list
+    jobs with `job__list`. Collected results over 50,000 chars are tail-capped; collecting is
+    consume-once, so when you need the complete output pass `full_result: true` (or have the
+    command write to a file). Collect or cancel every job you started before ending your turn. In
+    graph LLM nodes, jobs are node-local: collect or cancel every job you start before the node
+    ends — an uncollected job burns node iterations via the guardrail, and anything still
+    running when the node exits is cancelled with its result discarded. Jobs run against a
+    snapshot of the current config/environment and do not survive coyote exiting.
+"
+};
+
 pub(in crate::config) const DEFAULT_TEAMMATE_INSTRUCTIONS: &str = indoc! {"
     ## Teammate Messaging
 

src/graph/agent.rs

@@ -2,7 +2,7 @@ use super::state::StateManager;
 use super::structured;
 use super::types::AgentNode;
 use crate::config::RequestContext;
-use crate::function::supervisor::run_agent_for_graph;
+use crate::function::agents::run_agent_for_graph;
 use anyhow::{Context, Result};
 use serde_json::Value;
 use std::time::Duration;

src/graph/executor.rs

@@ -563,8 +563,14 @@ mod tests {
 mod integration_tests {
     use super::*;
     use crate::config::{AppState, WorkingMode};
+    #[cfg(unix)]
+    use crate::function::jobs::RingBuf;
+    #[cfg(unix)]
+    use crate::supervisor::{JobHandle, JobResult, JobState, JobStatus, Supervisor, notification};
     use crate::utils::{create_abort_signal, temp_file};
     use std::fs;
+    #[cfg(unix)]
+    use std::mem;
 
     fn cmd_available(name: &str) -> bool {
         which::which(name).is_ok()
@@ -856,4 +862,83 @@ nodes:
         );
         assert!(err.contains("sleeper"), "error should name frontier: {err}");
     }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn background_job_survives_graph_node_execution() {
+        if !cmd_available("bash") {
+            eprintln!("skipping: bash not available");
+            return;
+        }
+        let ws = TestWorkspace::new();
+        ws.write_script("noop.sh", "#!/bin/bash\necho '{}'\n");
+
+        let yaml = r#"
+name: background_job_survival_test
+start: noop
+nodes:
+  noop:
+    type: script
+    script: noop.sh
+    state_updates: {}
+    next: done
+  done:
+    type: end
+    output: "done"
+"#;
+        let graph: Graph = serde_yaml::from_str(yaml).unwrap();
+
+        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,
+            })
+        });
+        mem::forget(rt);
+        let handle = JobHandle {
+            id: "job_bg".to_string(),
+            tool: "execute_command".to_string(),
+            started_at: Instant::now(),
+            join_handle,
+            abort_signal: create_abort_signal(),
+            state: Arc::new(parking_lot::Mutex::new(JobState {
+                status: JobStatus::Completed,
+                pgid: None,
+            })),
+            output_buf: Arc::new(parking_lot::Mutex::new(RingBuf::default())),
+            no_change_checks: 0,
+            last_check_state: None,
+        };
+        let mut sup = Supervisor::new(0, 3).with_max_concurrent_jobs(4);
+        sup.register(handle).unwrap();
+
+        let mut ctx = make_ctx();
+        ctx.supervisor = Some(Arc::new(parking_lot::RwLock::new(sup)));
+        ctx.notification_queue.push(notification::job_notification(
+            "job_bg",
+            "execute_command",
+            true,
+        ));
+
+        let abort = create_abort_signal();
+        let result = GraphExecutor::new(graph, &ws.dir)
+            .execute(&mut ctx, abort)
+            .await
+            .unwrap_or_else(|e| panic!("executor failed: {e:#}"));
+        assert_eq!(result, "done");
+
+        assert!(
+            ctx.supervisor.as_ref().unwrap().read().has_job("job_bg"),
+            "graph execution must not touch registered job handles"
+        );
+        let events = ctx.notification_queue.drain();
+        assert_eq!(events.len(), 1, "queued notification must survive the run");
+        assert_eq!(events[0].id, "job_bg");
+        assert_eq!(events[0].event, "job_completed");
+    }
 }

src/graph/llm.rs

@@ -6,8 +6,9 @@ use crate::config::prompts::DEFAULT_SKILL_INSTRUCTIONS;
 use crate::config::{
     Input, RequestContext, Role, RoleLike, SkillPolicy, should_inject_skill_instructions,
 };
+use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail};
+use crate::function::jobs::reap_jobs;
 use crate::function::skill::skill_function_declarations;
-use crate::function::supervisor::{GuardrailAction, check_pending_agents_guardrail};
 use crate::utils::create_abort_signal;
 use anyhow::{Context, Error, Result, anyhow, bail};
 use log::warn;
@@ -173,6 +174,9 @@ async fn run(
 
     let saved_role = parent_ctx.role.clone();
     parent_ctx.role = Some(composed_role);
+    // Jobs are node-local: everything job__start registers while this node
+    // runs is recorded here and reaped on every exit path below.
+    let saved_job_scope = parent_ctx.node_job_scope.replace(Vec::new());
     let result = match node.timeout {
         Some(secs) => match timeout(
             Duration::from_secs(secs),
@@ -186,6 +190,9 @@ async fn run(
         None => run_with_retries(node, &prompt, parent_ctx).await,
     };
     parent_ctx.role = saved_role;
+    let node_jobs =
+        std::mem::replace(&mut parent_ctx.node_job_scope, saved_job_scope).unwrap_or_default();
+    reap_jobs(parent_ctx.supervisor.as_ref(), &node_jobs).await;
     restore_agent_skill_policy(parent_ctx, saved_agent_skill_state);
     result
 }
@@ -268,7 +275,7 @@ async fn run_chat_loop(node: &LlmNode, prompt: &str, ctx: &mut RequestContext) -
         }
 
         if tool_results.is_empty() {
-            match check_pending_agents_guardrail(ctx) {
+            match check_pending_tasks_guardrail(ctx) {
                 GuardrailAction::NoAction => return Ok(accumulated),
                 GuardrailAction::ForceTerminate(ids) => {
                     warn!(

src/graph/types.rs

@@ -28,6 +28,9 @@ pub struct Graph {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub reasoning_effort: Option<String>,
 
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_concurrent_jobs: Option<usize>,
+
     #[serde(default)]
     pub global_tools: Vec<String>,
 
@@ -895,6 +898,7 @@ nodes:
         assert!(graph.model.is_none());
         assert!(graph.temperature.is_none());
         assert!(graph.top_p.is_none());
+        assert!(graph.max_concurrent_jobs.is_none());
         assert!(graph.global_tools.is_empty());
         assert!(graph.mcp_servers.is_empty());
         assert!(graph.conversation_starters.is_empty());

src/graph/validator.rs

@@ -998,6 +998,7 @@ mod tests {
             temperature: None,
             top_p: None,
             reasoning_effort: None,
+            max_concurrent_jobs: None,
             global_tools: Vec::new(),
             mcp_servers: Vec::new(),
             skills_enabled: None,

src/main.rs

@@ -29,7 +29,7 @@ use crate::config::{
     install_builtins, list_agents, load_env_file, macro_execute, sync_models,
 };
 use crate::config::{memory, paths};
-use crate::function::supervisor::{GuardrailAction, check_pending_agents_guardrail};
+use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail};
 use crate::mcp::McpServersConfig;
 use crate::render::{prompt_theme, render_error};
 use crate::repl::Repl;
@@ -595,7 +595,7 @@ async fn start_directive(
         )
         .await?;
     } else {
-        match check_pending_agents_guardrail(ctx) {
+        match check_pending_tasks_guardrail(ctx) {
             GuardrailAction::Inject(prompt) => {
                 let guardrail_input = Input::from_str(ctx, &prompt, None)?;
                 return start_directive(ctx, guardrail_input, code_mode, abort_signal).await;

src/repl/mod.rs

@@ -16,7 +16,7 @@ use crate::config::{
     StateFlags, flatten_prompt_messages, macro_execute, resolve_prompt_args, sanitize_display_text,
 };
 use crate::config::{AssetCategory, paths};
-use crate::function::supervisor::{GuardrailAction, check_pending_agents_guardrail};
+use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail};
 use crate::render::render_error;
 use crate::utils::{
     AbortSignal, SHELL, abortable_run_with_spinner, create_abort_signal, dimmed_text,
@@ -602,7 +602,7 @@ pub async fn run_repl_command(
     abort_signal: AbortSignal,
     mut line: &str,
 ) -> Result<bool> {
-    ctx.pending_agents_guardrail_count = 0;
+    ctx.pending_tasks_guardrail_count = 0;
     if let Ok(Some(captures)) = MULTILINE_RE.captures(line)
         && let Some(text_match) = captures.get(1)
     {
@@ -1475,7 +1475,7 @@ async fn ask(
         )
         .await
     } else {
-        match check_pending_agents_guardrail(ctx) {
+        match check_pending_tasks_guardrail(ctx) {
             GuardrailAction::Inject(prompt) => {
                 let guardrail_input = Input::from_str(ctx, &prompt, None)?;
                 return ask(ctx, abort_signal, guardrail_input, false).await;

src/supervisor/mod.rs

@@ -1,17 +1,21 @@
 pub mod escalation;
 pub mod mailbox;
+pub mod notification;
 pub mod taskqueue;
 
+use crate::function::jobs::RingBuf;
 use crate::utils::AbortSignal;
 use fmt::{Debug, Formatter};
 use mailbox::Inbox;
-use parking_lot::RwLock;
+use parking_lot::{Mutex, RwLock};
 use taskqueue::TaskQueue;
 
 use anyhow::{Result, bail};
+use serde_json::Value;
 use std::collections::HashMap;
 use std::fmt;
 use std::sync::Arc;
+use std::time::Instant;
 use tokio::task::JoinHandle;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -37,11 +41,85 @@ pub struct AgentHandle {
     pub child_supervisor: Option<Arc<RwLock<Supervisor>>>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum JobStatus {
+    Running,
+    Completed,
+    Failed,
+}
+
+pub struct JobState {
+    pub status: JobStatus,
+    pub pgid: Option<i32>,
+}
+
+pub struct JobResult {
+    pub output: Value,
+    pub exit_code: Option<i32>,
+    pub output_bytes_captured: u64,
+}
+
+pub struct JobHandle {
+    pub id: String,
+    pub tool: String,
+    pub started_at: Instant,
+    pub join_handle: JoinHandle<Result<JobResult>>,
+    pub abort_signal: AbortSignal,
+    pub state: Arc<Mutex<JobState>>,
+    pub output_buf: Arc<Mutex<RingBuf>>,
+    pub no_change_checks: u32,
+    pub last_check_state: Option<(JobStatus, u64)>,
+}
+
+impl JobHandle {
+    // pgid == child pid under process_group(0); after wait() reaps the child
+    // the pid can be recycled, so never kill unless pgid is still set.
+    fn kill_process_group(&self) {
+        #[cfg(unix)]
+        if let Some(pgid) = self.state.lock().pgid {
+            unsafe {
+                libc::killpg(pgid, libc::SIGTERM);
+            }
+        }
+    }
+}
+
+impl Drop for JobHandle {
+    fn drop(&mut self) {
+        self.kill_process_group();
+        self.join_handle.abort();
+    }
+}
+
+pub enum TaskHandle {
+    Agent(AgentHandle),
+    Job(JobHandle),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TaskKind {
+    Agent,
+    Job,
+}
+
+impl From<AgentHandle> for TaskHandle {
+    fn from(handle: AgentHandle) -> Self {
+        Self::Agent(handle)
+    }
+}
+
+impl From<JobHandle> for TaskHandle {
+    fn from(handle: JobHandle) -> Self {
+        Self::Job(handle)
+    }
+}
+
 pub struct Supervisor {
-    handles: HashMap<String, AgentHandle>,
+    handles: HashMap<String, TaskHandle>,
     task_queue: TaskQueue,
     max_concurrent: usize,
     max_depth: usize,
+    max_concurrent_jobs: usize,
 }
 
 impl Supervisor {
@@ -51,17 +129,64 @@ impl Supervisor {
             task_queue: TaskQueue::new(),
             max_concurrent,
             max_depth,
+            max_concurrent_jobs: 0,
         }
     }
 
+    pub fn with_max_concurrent_jobs(mut self, max_concurrent_jobs: usize) -> Self {
+        self.max_concurrent_jobs = max_concurrent_jobs;
+        self
+    }
+
+    fn agent(&self, id: &str) -> Option<&AgentHandle> {
+        match self.handles.get(id) {
+            Some(TaskHandle::Agent(handle)) => Some(handle),
+            _ => None,
+        }
+    }
+
+    fn agents(&self) -> impl Iterator<Item = &AgentHandle> {
+        self.handles.values().filter_map(|handle| match handle {
+            TaskHandle::Agent(handle) => Some(handle),
+            TaskHandle::Job(_) => None,
+        })
+    }
+
+    pub fn job(&self, id: &str) -> Option<&JobHandle> {
+        match self.handles.get(id) {
+            Some(TaskHandle::Job(handle)) => Some(handle),
+            _ => None,
+        }
+    }
+
+    pub fn job_mut(&mut self, id: &str) -> Option<&mut JobHandle> {
+        match self.handles.get_mut(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.handles.len()
+        self.agents().count()
     }
 
     pub fn effective_active_count(&self) -> usize {
+        self.agents()
+            .filter(|h| !h.join_handle.is_finished())
+            .count()
+    }
+
+    pub fn active_job_count(&self) -> usize {
         self.handles
             .values()
-            .filter(|h| !h.join_handle.is_finished())
+            .filter(|h| matches!(h, TaskHandle::Job(job) if !job.join_handle.is_finished()))
             .count()
     }
 
@@ -73,6 +198,10 @@ impl Supervisor {
         self.max_depth
     }
 
+    pub fn max_concurrent_jobs(&self) -> usize {
+        self.max_concurrent_jobs
+    }
+
     pub fn task_queue(&self) -> &TaskQueue {
         &self.task_queue
     }
@@ -81,7 +210,9 @@ impl Supervisor {
         &mut self.task_queue
     }
 
-    pub fn register(&mut self, handle: AgentHandle) -> Result<()> {
+    pub fn register(&mut self, handle: impl Into<TaskHandle>) -> Result<()> {
+        match handle.into() {
+            TaskHandle::Agent(handle) => {
                 if self.effective_active_count() >= self.max_concurrent {
                     bail!(
                         "Cannot spawn agent: at capacity ({}/{})",
@@ -96,53 +227,120 @@ impl Supervisor {
                         self.max_depth
                     );
                 }
-        self.handles.insert(handle.id.clone(), handle);
+                self.handles
+                    .insert(handle.id.clone(), TaskHandle::Agent(handle));
+            }
+            TaskHandle::Job(handle) => {
+                if self.active_job_count() >= self.max_concurrent_jobs {
+                    bail!(
+                        "Cannot start job: at capacity ({}/{})",
+                        self.active_job_count(),
+                        self.max_concurrent_jobs
+                    );
+                }
+                self.handles
+                    .insert(handle.id.clone(), TaskHandle::Job(handle));
+            }
+        }
         Ok(())
     }
 
     pub fn is_finished(&self, id: &str) -> Option<bool> {
-        self.handles.get(id).map(|h| h.join_handle.is_finished())
+        self.agent(id).map(|h| h.join_handle.is_finished())
     }
 
     pub fn take(&mut self, id: &str) -> Option<AgentHandle> {
-        self.handles.remove(id)
+        self.agent(id)?;
+        match self.handles.remove(id) {
+            Some(TaskHandle::Agent(handle)) => Some(handle),
+            _ => None,
+        }
+    }
+
+    pub fn take_job(&mut self, id: &str) -> Option<JobHandle> {
+        if !self.has_job(id) {
+            return None;
+        }
+        match self.handles.remove(id) {
+            Some(TaskHandle::Job(handle)) => Some(handle),
+            _ => None,
+        }
+    }
+
+    pub fn has_job(&self, id: &str) -> bool {
+        matches!(self.handles.get(id), Some(TaskHandle::Job(_)))
+    }
+
+    pub fn has_agent(&self, id: &str) -> bool {
+        self.agent(id).is_some()
     }
 
     pub fn inbox(&self, id: &str) -> Option<&Arc<Inbox>> {
-        self.handles.get(id).map(|h| &h.inbox)
+        self.agent(id).map(|h| &h.inbox)
     }
 
     pub fn abort_signal_for(&self, id: &str) -> Option<AbortSignal> {
-        self.handles.get(id).map(|h| h.abort_signal.clone())
+        self.agent(id).map(|h| h.abort_signal.clone())
     }
 
     pub fn list_agents(&self) -> Vec<(&str, &str)> {
+        self.agents()
+            .map(|h| (h.id.as_str(), h.agent_name.as_str()))
+            .collect()
+    }
+
+    pub fn list_tasks(&self) -> Vec<(&str, TaskKind, bool)> {
         self.handles
             .values()
-            .map(|h| (h.id.as_str(), h.agent_name.as_str()))
+            .map(|handle| match handle {
+                TaskHandle::Agent(agent) => (
+                    agent.id.as_str(),
+                    TaskKind::Agent,
+                    agent.join_handle.is_finished(),
+                ),
+                TaskHandle::Job(job) => (
+                    job.id.as_str(),
+                    TaskKind::Job,
+                    job.join_handle.is_finished(),
+                ),
+            })
             .collect()
     }
 
     pub fn cancel_all(&self) {
         for handle in self.handles.values() {
-            handle.abort_signal.set_ctrlc();
+            match handle {
+                TaskHandle::Agent(agent) => agent.abort_signal.set_ctrlc(),
+                TaskHandle::Job(job) => {
+                    job.abort_signal.set_ctrlc();
+                    job.kill_process_group();
+                }
+            }
         }
     }
 
     pub fn cancel_recursive(&self) {
         for handle in self.handles.values() {
-            handle.abort_signal.set_ctrlc();
-            if let Some(child_sup) = handle.child_supervisor.as_ref() {
+            match handle {
+                TaskHandle::Agent(agent) => {
+                    agent.abort_signal.set_ctrlc();
+                    if let Some(child_sup) = agent.child_supervisor.as_ref() {
                         child_sup.read().cancel_recursive();
                     }
                 }
+                TaskHandle::Job(job) => {
+                    job.abort_signal.set_ctrlc();
+                    job.kill_process_group();
+                }
+            }
+        }
     }
 }
 
 impl Debug for Supervisor {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         f.debug_struct("Supervisor")
-            .field("active_agents", &self.handles.len())
+            .field("active_agents", &self.active_count())
             .field("max_concurrent", &self.max_concurrent)
             .field("max_depth", &self.max_depth)
             .finish()
@@ -154,6 +352,7 @@ mod tests {
     use super::*;
     use crate::utils::create_abort_signal;
     use anyhow::Error;
+    use std::mem;
     use tokio::runtime::Builder;
 
     fn make_handle(id: &str, agent_name: &str, depth: usize) -> AgentHandle {
@@ -177,6 +376,34 @@ mod tests {
         }
     }
 
+    fn make_job(id: &str, abort_signal: AbortSignal) -> JobHandle {
+        // Keep the runtime alive so the spawned task is never polled and the
+        // job counts as running for capacity checks.
+        let rt = 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,
+            })
+        });
+        mem::forget(rt);
+        JobHandle {
+            id: id.to_string(),
+            tool: "execute_command".to_string(),
+            started_at: Instant::now(),
+            join_handle,
+            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,
+            last_check_state: None,
+        }
+    }
+
     #[test]
     fn supervisor_new_empty() {
         let sup = Supervisor::new(4, 3);
@@ -294,4 +521,111 @@ mod tests {
             AgentExitStatus::Failed("x".into())
         );
     }
+
+    #[test]
+    fn cancel_recursive_aborts_nested_supervisors() {
+        let child_sig = create_abort_signal();
+        let mut child_handle = make_handle("c1", "worker", 2);
+        child_handle.abort_signal = child_sig.clone();
+        let mut child_sup = Supervisor::new(4, 3);
+        child_sup.register(child_handle).unwrap();
+
+        let parent_sig = create_abort_signal();
+        let mut parent_handle = make_handle("a1", "explore", 1);
+        parent_handle.abort_signal = parent_sig.clone();
+        parent_handle.child_supervisor = Some(Arc::new(RwLock::new(child_sup)));
+        let mut sup = Supervisor::new(4, 3);
+        sup.register(parent_handle).unwrap();
+
+        sup.cancel_recursive();
+
+        assert!(parent_sig.aborted());
+        assert!(child_sig.aborted());
+    }
+
+    #[test]
+    fn job_registration_rejects_when_job_capacity_zero() {
+        let mut sup = Supervisor::new(4, 3);
+
+        let result = sup.register(make_job("j1", create_abort_signal()));
+
+        assert!(result.is_err());
+        assert!(result.unwrap_err().to_string().contains("at capacity"));
+    }
+
+    #[test]
+    fn job_registration_rejects_at_job_capacity() {
+        let mut sup = Supervisor::new(4, 3).with_max_concurrent_jobs(1);
+        sup.register(make_job("j1", create_abort_signal())).unwrap();
+
+        let result = sup.register(make_job("j2", create_abort_signal()));
+
+        assert!(result.is_err());
+        assert!(result.unwrap_err().to_string().contains("at capacity"));
+    }
+
+    #[test]
+    fn job_capacity_is_independent_of_agent_capacity() {
+        let mut sup = Supervisor::new(1, 3).with_max_concurrent_jobs(1);
+
+        sup.register(make_job("j1", create_abort_signal())).unwrap();
+        sup.register(make_handle("a1", "explore", 1)).unwrap();
+
+        assert_eq!(sup.active_job_count(), 1);
+        assert_eq!(sup.active_count(), 1);
+        assert_eq!(sup.max_concurrent_jobs(), 1);
+    }
+
+    #[test]
+    fn agent_accessors_ignore_jobs() {
+        let mut sup = Supervisor::new(4, 3).with_max_concurrent_jobs(2);
+
+        sup.register(make_job("j1", create_abort_signal())).unwrap();
+
+        assert_eq!(sup.active_count(), 0);
+        assert_eq!(sup.effective_active_count(), 0);
+        assert!(sup.list_agents().is_empty());
+        assert_eq!(sup.is_finished("j1"), None);
+        assert!(sup.inbox("j1").is_none());
+        assert!(sup.abort_signal_for("j1").is_none());
+        assert!(sup.take("j1").is_none());
+        assert!(sup.has_job("j1"));
+        assert!(!sup.has_agent("j1"));
+        assert_eq!(sup.active_job_count(), 1);
+    }
+
+    #[test]
+    fn take_job_removes_job_but_not_agents() {
+        let mut sup = Supervisor::new(4, 3).with_max_concurrent_jobs(2);
+
+        sup.register(make_job("j1", create_abort_signal())).unwrap();
+        sup.register(make_handle("a1", "explore", 1)).unwrap();
+
+        assert!(sup.take_job("a1").is_none());
+        assert!(sup.has_agent("a1"));
+        assert!(sup.take_job("j1").is_some());
+        assert_eq!(sup.active_job_count(), 0);
+    }
+
+    #[test]
+    fn cancel_recursive_aborts_jobs() {
+        let sig = create_abort_signal();
+        let mut sup = Supervisor::new(4, 3).with_max_concurrent_jobs(1);
+        sup.register(make_job("j1", sig.clone())).unwrap();
+
+        sup.cancel_recursive();
+
+        assert!(sig.aborted());
+    }
+
+    #[test]
+    fn cancel_all_aborts_jobs() {
+        let sig = create_abort_signal();
+        let mut sup = Supervisor::new(4, 3).with_max_concurrent_jobs(1);
+        sup.register(make_job("j1", sig.clone())).unwrap();
+
+        sup.cancel_all();
+
+        assert!(sig.aborted());
+    }
 }

src/supervisor/notification.rs

@@ -0,0 +1,168 @@
+use fmt::{Debug, Formatter};
+use serde_json::{Value, json};
+use std::fmt;
+
+/// One background-task completion event, delivered to the context that
+/// started the task by merging a `system_notifications` entry onto the last
+/// tool result of a batch.
+#[derive(Clone)]
+pub struct SystemNotification {
+    pub event: &'static str,
+    pub id: String,
+    pub tool_or_agent: String,
+    pub status: &'static str,
+    pub next_action: String,
+}
+
+impl SystemNotification {
+    pub fn to_value(&self) -> Value {
+        json!({
+            "event": self.event,
+            "id": self.id,
+            "tool_or_agent": self.tool_or_agent,
+            "status": self.status,
+            "next_action": self.next_action,
+        })
+    }
+}
+
+pub fn job_notification(id: &str, tool: &str, success: bool) -> SystemNotification {
+    SystemNotification {
+        event: if success {
+            "job_completed"
+        } else {
+            "job_failed"
+        },
+        id: id.to_string(),
+        tool_or_agent: tool.to_string(),
+        status: if success { "success" } else { "failed" },
+        next_action: format!("job__collect --id {id} for output"),
+    }
+}
+
+pub fn agent_notification(id: &str, agent_name: &str, success: bool) -> SystemNotification {
+    SystemNotification {
+        event: if success {
+            "agent_completed"
+        } else {
+            "agent_failed"
+        },
+        id: id.to_string(),
+        tool_or_agent: agent_name.to_string(),
+        status: if success { "success" } else { "failed" },
+        next_action: format!("agent__collect --id {id} for output"),
+    }
+}
+
+/// Completion events for background work started by ONE context. Unlike the
+/// escalation queue (shared, root-owned), every context owns a fresh queue:
+/// a queue shared between parent and child would race their drains and
+/// deliver one context's events into the other's transcript.
+pub struct NotificationQueue {
+    pending: parking_lot::Mutex<Vec<SystemNotification>>,
+}
+
+impl NotificationQueue {
+    pub fn new() -> Self {
+        Self {
+            pending: parking_lot::Mutex::new(Vec::new()),
+        }
+    }
+
+    pub fn push(&self, notification: SystemNotification) {
+        self.pending.lock().push(notification);
+    }
+
+    pub fn drain(&self) -> Vec<SystemNotification> {
+        std::mem::take(&mut *self.pending.lock())
+    }
+}
+
+impl Default for NotificationQueue {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Debug for NotificationQueue {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        let count = self.pending.lock().len();
+        f.debug_struct("NotificationQueue")
+            .field("pending_count", &count)
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn job_notification_success_shape() {
+        let event = job_notification("job_a1b2", "execute_command", true);
+        assert_eq!(
+            event.to_value(),
+            json!({
+                "event": "job_completed",
+                "id": "job_a1b2",
+                "tool_or_agent": "execute_command",
+                "status": "success",
+                "next_action": "job__collect --id job_a1b2 for output",
+            })
+        );
+    }
+
+    #[test]
+    fn job_notification_failure_shape() {
+        let event = job_notification("job_a1b2", "execute_command", false);
+        assert_eq!(event.event, "job_failed");
+        assert_eq!(event.status, "failed");
+        assert_eq!(event.next_action, "job__collect --id job_a1b2 for output");
+    }
+
+    #[test]
+    fn agent_notification_success_shape() {
+        let event = agent_notification("agent_explore_a1b2", "explore", true);
+        assert_eq!(
+            event.to_value(),
+            json!({
+                "event": "agent_completed",
+                "id": "agent_explore_a1b2",
+                "tool_or_agent": "explore",
+                "status": "success",
+                "next_action": "agent__collect --id agent_explore_a1b2 for output",
+            })
+        );
+    }
+
+    #[test]
+    fn agent_notification_failure_shape() {
+        let event = agent_notification("agent_explore_a1b2", "explore", false);
+        assert_eq!(event.event, "agent_failed");
+        assert_eq!(event.status, "failed");
+        assert_eq!(
+            event.next_action,
+            "agent__collect --id agent_explore_a1b2 for output"
+        );
+    }
+
+    #[test]
+    fn drain_empties_queue_and_preserves_order() {
+        let queue = NotificationQueue::new();
+        queue.push(job_notification("job_1", "execute_command", true));
+        queue.push(job_notification("job_2", "execute_command", false));
+
+        let drained = queue.drain();
+
+        assert_eq!(drained.len(), 2);
+        assert_eq!(drained[0].id, "job_1");
+        assert_eq!(drained[1].id, "job_2");
+        assert!(queue.drain().is_empty());
+    }
+
+    #[test]
+    fn drain_on_empty_queue_is_a_noop() {
+        let queue = NotificationQueue::default();
+        assert!(queue.drain().is_empty());
+    }
+}