feat: claude and openai native web search via web_search_coyote

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/code-reviewer/config.yaml

@@ -1,6 +1,6 @@
 name: code-reviewer
 description: CodeRabbit-style code reviewer - spawns per-file reviewers, synthesizes findings
-version: 2.2.0
+version: 2.3.0
 
 auto_continue: true
 max_auto_continues: 20
@@ -76,6 +76,7 @@ instructions: |
   ## MUST DO
   - Load `code-review` and `ai-slop-remover` skills before reading any code
   - Load `transactional-integrity` as well if this file's diff touches state-changing code (DB writes, transactions, queue/webhook/job handlers, retries, external side effects)
+  - Load `logging-discipline` as well if this file's diff touches boundaries, error paths, background jobs, or state transitions
   - Apply all loaded skill checklists to the diff
   - Use targeted fs_read with offset/limit; max 5 file reads
   - End with REVIEW_COMPLETE

assets/agents/coder/graph.yaml

@@ -17,6 +17,7 @@ enabled_skills:
   - code-review
   - comment-discipline
   - diagnosing-bugs
+  - logging-discipline
   - git-master
   - frontend-ui-ux
   - verification-gates
@@ -171,6 +172,7 @@ nodes:
       - code-review
       - comment-discipline
       - diagnosing-bugs
+      - logging-discipline
       - git-master
       - frontend-ui-ux
       - verification-gates
@@ -221,7 +223,12 @@ nodes:
          `comment-discipline` (self-documenting / api-documented /
          comment-heavy) and write comments to match. When the signal is
          weak, write NO comment.
-      4. Follow the same patterns exactly. Do not invent new ones.
+      4. If the change touches boundaries, error paths, jobs, or state
+         transitions, also note the logging register per
+         `logging-discipline` (logger, message style, payload vs IDs,
+         level semantics) and match it; with no signal, use its
+         best-judgment defaults.
+      5. Follow the same patterns exactly. Do not invent new ones.
 
       ## Fix loop
 

assets/agents/file-reviewer/config.yaml

@@ -1,12 +1,13 @@
 name: file-reviewer
 description: Reviews a single file's diff for bugs, style issues, and cross-cutting concerns
-version: 2.1.0
+version: 2.2.0
 
 skills_enabled: true
 enabled_skills:
   - code-review
   - ai-slop-remover
   - transactional-integrity
+  - logging-discipline
 
 variables:
   - name: project_dir
@@ -32,6 +33,8 @@ instructions: |
 
   Additionally load `transactional-integrity` when the diff touches state-changing code — database writes, transaction blocks, queue/webhook/job handlers, retry logic, or calls to external state-holding systems. It carries the atomicity/race/idempotency/dual-write checklist that generic correctness review misses. Skip it for pure reads, UI, and stateless computation.
 
+  Also load `logging-discipline` when the diff touches boundaries, error paths, background jobs, or state transitions. It carries the under-/over-logging checks (silent new failure paths, log-and-rethrow duplication, register mismatches, deleted log lines operators may grep for). Skip it for diffs with no operational surface.
+
   Apply every loaded checklist in every review. Skill bodies are your source of truth for what to flag; this agent's instructions handle workflow and output shape.
 
   ## Your Mission

assets/agents/sisyphus/config.yaml

@@ -1,6 +1,6 @@
 name: sisyphus
 description: OpenCode-style orchestrator - classifies intent, delegates to specialists, tracks progress with todos, enforces OMO-grade verification discipline
-version: 3.7.0
+version: 3.9.0
 
 agent_session: temp
 auto_continue: true
@@ -30,6 +30,8 @@ enabled_skills:
   - comment-discipline
   - diagnosing-bugs
   - grilling
+  - logging-discipline
+  - observability-review
   - git-master
   - frontend-ui-ux
   - delegation-protocol
@@ -45,6 +47,9 @@ variables:
   - name: project_dir
     description: Project directory to work in
     default: '.'
+  - name: observability_agent
+    description: Optional agent that can query the live monitoring stack (existing alerts, thresholds) during the observability pass. Empty disables the live lookup; repo-derived inventory still runs.
+    default: ''
   - name: auto_confirm
     description: Auto-confirm command execution
     default: '1'
@@ -228,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)
@@ -386,11 +391,25 @@ instructions: |
 
   Like `adversary`, re-running `security-reviewer` once after a fix is expected — a FAIL verdict is a hard gate, and confirming the fix closed the attack path is the point. Run all applicable reviewers (`code-reviewer`, `adversary`, `security-reviewer`) — they cover disjoint failure modes; one passing says nothing about the others.
 
+  ### Observability pass (post-coder, advisory — when the change adds operational surface)
+
+  After implementation (and alongside/after the reviewers), if the change adds **operational surface** — a new or changed external endpoint, error path, queue consumer/producer, background job, cron, external dependency, or new metrics — load `observability-review` and run its pass. If none of these apply, skip with a one-line note.
+
+  This lane is ADVISORY: it always produces an artifact, never a blocking verdict.
+
+  1. Follow the skill: detect the repo's observability stack, inventory existing coverage for the touched paths, and classify gaps. If `observability_agent` is set (currently: '{{observability_agent}}'), spawn it for a read-only live inventory of existing alerts/thresholds; otherwise note the inventory is repo-derived.
+  2. **Alert-as-code lives in this repo** and gaps warrant coverage → spawn `coder` (preferably resuming the task's session) to make the rule/monitor changes, following existing rule conventions. These are ordinary code changes — the usual review gates apply to them.
+  3. **Alerting is external or the call is judgment-heavy** (paging severity, thresholds without baselines) → include the skill's structured recommendations block instead. Never touch external alerting systems.
+  4. Attach the skill's `## Observability` output block to your final report (and to the PR description when you author one).
+
+  Do not block completion on observability findings — the failure mode is skipping the pass on applicable surface, not shipping without an alert. Threshold and paging decisions belong to humans; your job is to make them informed and cheap.
+
   ## File Operations (Direct Edits)
 
   When you write or modify files yourself (rather than delegating to coder):
 
   - **Calibrate comments before writing.** Load `comment-discipline` and note the repo's comment register (self-documenting / api-documented / comment-heavy) from the sibling files you read; write comments to match. When the signal is weak, write NO comment.
+  - **Calibrate logging before writing.** When the change touches boundaries, error paths, jobs, or state transitions, load `logging-discipline` and note the repo's logging register (logger, message style, payload vs IDs, level semantics) from the same sibling reads; match it. No discernible convention → its best-judgment defaults. Never leave a new error path silently swallowed, and never delete existing log lines as drive-by cleanup.
   - **For editing an existing file**, prefer `fs_patch`. It's a surgical edit that preserves unchanged content. Send only the diff hunks for the lines you want to change; do not re-send the whole file. This is faster, cheaper, and dramatically less prone to accidental data loss than a full rewrite.
   - **For writing a NEW file or doing a COMPLETE rewrite**, use `fs_write`. Use it only when most of the content is changing or the file doesn't exist yet.
   - **NEVER write files via `execute_command`.** Do not use:

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
 }

assets/functions/tools/fs_grep.sh

@@ -25,7 +25,11 @@ main() {
         exit 1
     fi
 
-    local grep_args=(-nH --color=never)
+    # --binary-files=text: GNU grep's binary heuristic false-positives on valid
+    # UTF-8 source files >=128KiB when a multibyte character straddles an
+    # internal read-buffer boundary, silently returning zero matches. This tool
+    # only searches text, so always force text mode.
+    local grep_args=(-nH --color=never --binary-files=text)
 
     if [[ -d "$search_path" ]]; then
         # Use -r (not -R) so symlinks to directories are NOT followed - this avoids

assets/functions/tools/web_search_coyote.sh

@@ -15,6 +15,11 @@ set -e
 #   - vertexai:gemini-*
 #   - perplexity:*
 #   - ernie:*
+#   - claude:*                       (Anthropic native web_search server tool)
+#   - openai:gpt-4o-search-preview   (and -mini-; requires an api-key openai
+#                                     client — the codex OAuth path uses the
+#                                     Responses API where this parameter
+#                                     does not exist)
 # @env LLM_OUTPUT=/dev/stdout The output path
 
 # shellcheck disable=SC2154
@@ -30,6 +35,13 @@ main() {
 }'
     elif [[ "$client" == "ernie" ]]; then
         export COYOTE_PATCH_ERNIE_CHAT_COMPLETIONS='{".*":{"body":{"web_search":{"enable":true}}}}'
+    elif [[ "$client" == "claude" ]]; then
+        export COYOTE_PATCH_CLAUDE_CHAT_COMPLETIONS='{".*":{"body":{"tools":[{"type":"web_search_20250305","name":"web_search","max_uses":5}]}}}'
+    elif [[ "$client" == "openai" ]]; then
+        # Chat Completions native search exists only on the search-preview
+        # models; the regex scopes the patch so other OpenAI models run
+        # unpatched instead of erroring on an unsupported parameter.
+        export COYOTE_PATCH_OPENAI_CHAT_COMPLETIONS='{"gpt-4o.*search-preview.*":{"body":{"web_search_options":{}}}}'
     fi
 
     coyote -m "$WEB_SEARCH_MODEL" "$argc_query" >> "$LLM_OUTPUT"

assets/skills/logging-discipline/SKILL.md

@@ -0,0 +1,69 @@
+---
+description: Calibrate log output to the repository's existing logging conventions before writing code, and review diffs for under- and over-logging. Detects the repo's logging register from sibling files - logger/framework, message style (capitalization, length, tense), payload vs ID-only context, level semantics, error-path convention - and matches it; falls back to stated best-judgment defaults when no convention exists. Load when writing code that touches boundaries, error paths, jobs, or state transitions, or when reviewing such a diff. Complements security-review (which owns secrets/PII in logs) and incident-prior-art (which treats deleted log lines as operational leads).
+---
+You are writing or reviewing code that logs — or that should. LLMs fail in both directions: narrating every step (noise operators must grep past) and swallowing error paths silently (invisible failures at 3am). "Correct" is repo-relative: detect the register, match it; where no register exists, apply the best-judgment defaults below.
+
+## Step 0: Check for a declared policy first
+
+Check the workspace instructions already in your context (`COYOTE.md`/`AGENTS.md`, a logging section in `CONTRIBUTING.md`) for a stated logging convention. A declaration beats detection — obey it and skip Step 1.
+
+## Step 1: Detect the register (during reads you already do)
+
+Pattern-matching discipline already has you reading 2-3 sibling files before writing. While reading, note how THEY log:
+
+1. **Logger and shape** — which logging library/facade, and is output structured (key-value fields) or printf-style interpolated strings? Never introduce a second logging mechanism alongside an established one.
+2. **Message style** — capitalization (lowercase `"failed to connect"` vs sentence-case `"Failed to connect"`), punctuation (trailing periods or not), length (terse fragments vs full sentences), tense/mood ("connecting" / "connected" / "connect failed"). Match all of it — mixed message styles make logs harder to grep.
+3. **Context convention** — what rides along with the message: full payloads, or IDs only? Which fields are customary (request/correlation ID, entity IDs, durations)? Attached as structured fields or interpolated into the string? If the repo logs IDs-only, do NOT log payloads — that's both a style break and a data-exposure risk.
+4. **Level semantics in practice** — what does this repo actually use `error`/`warn`/`info`/`debug` for? Match observed usage over textbook definitions.
+5. **Error-path convention** — do errors get logged where they occur and then propagated, or propagated silently and logged once at the top? Match it; this determines where YOUR log lines go.
+
+Sample from the same language and layer you're editing — a chatty CLI layer and a quiet library core can coexist in one repo; the nearest siblings win.
+
+## Step 2: When to log (and when not)
+
+Warranted — a reader on-call should be able to see:
+
+- **Boundaries**: calls to external systems (network, DB, queues) — at minimum their failures, with enough context to identify the failing operation.
+- **Error paths**: every error is either logged or propagated to something that logs it — never silently swallowed, and **never both** (see invariants).
+- **Lifecycle**: job/worker/process start, finish, and abnormal exit; consumed/produced messages where the repo's register does so.
+- **State transitions an operator would care about** (order of magnitude: status changes, retries exhausted, fallbacks engaged).
+
+Unwarranted:
+
+- **Narration** — logging what the next line of code plainly does ("entering function", "about to save"). The comment-discipline rule, applied to logs.
+- **Hot paths** — per-item logging inside loops or per-request debug logging in high-volume paths; aggregate or sample instead.
+- **Log-and-rethrow** — logging an error AND re-raising it to a caller that logs again produces duplicate stacks that make incidents harder to read, not easier.
+- **Payloads the register doesn't log** — and never full payloads containing credentials or personal data regardless of register (security-review owns that judgment; don't create the finding).
+
+## Best-judgment defaults (weak or no signal)
+
+A greenfield file, a repo with no discernible convention, or contradictory siblings — use these and note the choice:
+
+- Structured logging if the ecosystem's standard library or dominant framework supports it; otherwise the language's idiomatic default.
+- Terse, lowercase, no trailing period, present-tense messages ("failed to fetch invoice"), stable wording (log messages are grepped and alerted on — treat them as identifiers, not prose).
+- IDs and small scalar fields, never payloads.
+- `error` = someone may need to act, `warn` = degraded but coping, `info` = lifecycle, `debug` = development detail.
+- When genuinely unsure whether a line earns its keep: boundaries and error paths yes, everything else no.
+
+## Review-side checks (for diffs)
+
+- **Underdone**: a new external call, error path, or background job with zero failure visibility — no log, no metric, no propagation to a logging caller. Cite the path and what an operator would be blind to.
+- **Overdone**: narration logs, log-and-rethrow duplication, hot-loop logging, payload logging in an IDs-only repo. Cite the line and the register evidence.
+- **Register mismatch**: new log lines that break the detected message style or use a different logger/mechanism than the siblings.
+- **Deleted or reworded log lines**: operators and alerts grep for exact strings; flag deletions/rewordings of lines that look triage-relevant so the change is conscious, not accidental (incident-prior-art treats these as leads — same instinct at review time).
+- Severity calibration: silent new failure paths are 🟡 findings; style/register mismatches are 🟢/💡.
+
+## Invariants (register-independent)
+
+1. **No error silently swallowed.** An empty catch/ignored error with no log, no metric, and no propagation is a finding in every repo.
+2. **No double-logging of one error** along a single propagation path — one log per failure, at the level the repo's convention chooses.
+3. **No secrets or personal data in logs**, ever, regardless of how payload-happy the register is.
+4. **Never delete existing log lines as drive-by "cleanup"** — that's out-of-scope churn AND an operational hazard; if a line must go, say so explicitly in the change description.
+
+## Anti-patterns
+
+- Importing your favorite logging style into a repo that has one.
+- Logging every function entry/exit because "more visibility is better" — noise is the enemy of visibility.
+- Flagging a quiet pure-computation module for "missing logs" — the trigger surface is boundaries, error paths, jobs, and state transitions; inert code needs none.
+- Rewording existing log messages to be "cleaner" — you just broke someone's saved Loki/CloudWatch query.
+- Treating textbook level definitions as authoritative over the repo's observed usage.

assets/skills/observability-review/SKILL.md

@@ -0,0 +1,72 @@
+---
+description: Post-implementation observability analysis - decide what monitoring, metrics, and alerts the just-implemented code needs, account for what already exists, and either produce concrete alert-as-code changes (routed through the normal implementation pipeline) or a structured recommendations block for the final report / PR description. Advisory by design - it always produces its artifact, never a blocking verdict. Load after implementing changes that add operational surface - new external endpoints, error paths, queues/jobs/crons, or notable state machines. Grants read-only filesystem access for stack detection and coverage inventory.
+enabled_tools: fs_read, fs_grep, fs_glob, fs_cat, fs_ls
+---
+Code was just implemented; you are deciding how anyone will know when it breaks. Logging (see `logging-discipline`) makes failures *inspectable*; this pass makes them *noticed* — metrics, alerts, dashboards. The output is always an artifact (code changes or a recommendations block), never a verdict: observability judgments (thresholds, paging severity) are ultimately human calls, so this lane informs and proposes rather than blocks.
+
+## When this pass applies
+
+The change adds **operational surface**: a new or changed external endpoint, a new error path or failure mode, a new queue consumer/producer, background job, or cron, a new external dependency, a notable state machine, or new metrics. If none of these — pure refactor, UI polish, docs, tests — skip with a one-line note. An observability pass on inert code is budget spent producing nothing.
+
+## Step 1: Detect the observability stack
+
+Establish what this repo HAS before proposing anything:
+
+1. **Metrics emission** — grep for the instrumentation the codebase already uses (a metrics client, OpenTelemetry, statsd-style calls, framework middleware). Note the naming convention of existing metrics.
+2. **Alert-as-code** — look for alert/monitor definitions living in the repo: rule files (e.g. Prometheus-style `*.rules.y*ml`), monitor/alert resources in infrastructure-as-code, `alerts/`/`monitoring/` directories, dashboard-as-code. THIS determines your output mode (Step 3).
+3. **Existing coverage inventory** — for the paths the change touches, find what already watches them: grep rule files and dashboards for the relevant metric names, service names, and log strings. An alert that already covers the new failure mode means UPDATE or NOTHING, not a duplicate.
+4. **Live-lookup hook (optional)** — if the caller configured an agent that can query the live monitoring stack, spawn it to verify the inventory ("what alerts currently cover <service/path>? current thresholds?") instead of trusting repo greps alone. The spawn prompt is that agent's whole context: name the services, metrics, and symptoms to look up, and state that it is read-only reconnaissance. If no such agent is configured, note that the inventory is repo-derived.
+
+## Step 2: Gap analysis
+
+For each new failure mode / operational surface in the change, walk the chain:
+
+1. **Is there a signal?** Does anything (metric, log line, built-in framework metric) even record this failing? No signal → no alert can exist; the first recommendation is the signal itself.
+2. **Is there detection on the signal?** An existing alert/monitor that would fire? Check semantics, not just existence — an endpoint-level 5xx alert may already cover your new handler; a queue-depth alert may NOT cover your new consumer's silent skip path.
+3. **Is the detection actionable?** Would it fire with enough context to triage (labels, runbook link), at the right urgency?
+
+Classify each gap: **covered** (existing signal + alert suffice), **update** (existing alert needs a label/threshold/scope change), **new** (nothing watches this), or **accepted-blind** (deliberately unwatched — say why, e.g. dev-only tooling).
+
+## Step 3: Produce the artifact (write vs recommend)
+
+The repo's alert-as-code situation decides:
+
+- **Alert-as-code lives in this repo** and the gap warrants coverage → produce the concrete rule/monitor changes (new rules, updated thresholds/labels/scopes) as ordinary code changes, following the existing rule files' conventions exactly. Route them through the caller's NORMAL implementation pipeline — same review gates as any code. An unreviewed alert is a false-page generator.
+- **Alerting lives outside the repo** (a UI-managed system, another team's repo), or the decision is judgment-heavy (paging severity, threshold without baseline data) → produce a structured **recommendations block** for the final report / PR description instead. Never attempt to modify external systems.
+
+Mixed outcomes are normal: write the mechanical rule update, recommend the judgment-heavy new pager.
+
+## Output format
+
+Always end with this block (it is the artifact the caller attaches to the report/PR):
+
+```
+## Observability
+
+Surface analyzed: <the operational surface this change adds, one line>
+Stack: <metrics lib / alert-as-code location or "external-only" / live inventory used: yes|no>
+
+Covered:
+- <failure mode> — covered by <existing alert/metric, path or name>
+
+Changes made (via the implementation pipeline):
+- <rule file:change> — <what and why>   (or "none")
+
+Recommendations (for humans to action):
+- <proposed alert> — signal: <metric/log>, condition: <threshold + rationale or "needs baseline data - start with X and tune">, urgency: <page|ticket>, runbook note: <one line>
+- <proposed metric/dashboard addition> — <why anyone would look at it>
+
+Accepted blind spots:
+- <what is deliberately unwatched and why>   (or "none")
+```
+
+For inapplicable changes the whole block collapses to: `## Observability` / `Not applicable: <one line>`.
+
+## Anti-patterns
+
+- **Alert spam.** Every alert costs attention forever. Page-worthy = a human must act NOW; everything else is a ticket or a dashboard. When in doubt, recommend ticket-urgency and say so.
+- **Invented thresholds.** A threshold with no baseline is a guess; either ground it in observed data (existing dashboards, load expectations stated in the change) or mark it explicitly as "start here, tune after N days".
+- **Duplicating existing coverage** because you only grepped for one spelling of the metric — inventory first, propose second.
+- **Metrics nobody will chart.** Each proposed metric names who would look at it and when. "Might be useful" is not a consumer.
+- **Blocking on this pass.** It is advisory: produce the artifact, attach it, move on. The only failure mode is skipping the pass on a change that added operational surface.
+- **Touching external alerting systems.** Recommendations only; live systems belong to humans and their change control.

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());
+    }
+}