#[allow(unused_imports)]
use crate::sync_util::LockExt;
use rig::completion::ToolDefinition;
use rig::tool::Tool;
use serde::Deserialize;
use std::collections::HashMap;
use std::sync::Mutex;
use tokio::sync::mpsc;
use uuid::Uuid;
use crate::agent::agent_loop::tool::AbortSignal;
use crate::agent::tools::background::{BackgroundStore, TaskState};
use crate::agent::tools::{AskSender, PermCheck, ToolError, check_perm};
use crate::config::SubagentWriteIsolation;
use crate::provider::AnyModel;
use crate::sandbox::Sandbox;
#[cfg(feature = "git-worktree")]
type WriterWorktree = crate::extras::git_worktree::WorktreeInfo;
#[cfg(not(feature = "git-worktree"))]
#[allow(dead_code)]
#[derive(Clone)]
struct WriterWorktree {
branch: String,
worktree_path: std::path::PathBuf,
main_repo_path: std::path::PathBuf,
}
fn writer_worktree_enabled(
is_writer: bool,
isolation: SubagentWriteIsolation,
is_microvm: bool,
sandbox_confines_writes: bool,
) -> Result<bool, &'static str> {
if !is_writer || isolation == SubagentWriteIsolation::Serialize {
return Ok(false);
}
if is_microvm {
return match isolation {
SubagentWriteIsolation::Worktree => {
Err("worktree write isolation is unavailable in microVM sandbox mode")
}
SubagentWriteIsolation::Auto => Ok(false),
SubagentWriteIsolation::Serialize => Ok(false),
};
}
if !sandbox_confines_writes {
return match isolation {
SubagentWriteIsolation::Worktree => Err(
"worktree write isolation requires a confining sandbox; run dirge with a Linux bwrap sandbox, or the writer's shell could escape the worktree",
),
SubagentWriteIsolation::Auto => Ok(false),
SubagentWriteIsolation::Serialize => unreachable!(),
};
}
Ok(true)
}
fn current_repo_is_dirty() -> Result<bool, String> {
let repo = std::env::current_dir()
.map_err(|e| format!("failed to resolve current directory: {e}"))?
.canonicalize()
.map_err(|e| format!("failed to canonicalize repository: {e}"))?;
crate::extras::git_worktree::repo_is_dirty(&repo)
}
#[cfg(feature = "git-worktree")]
fn create_writer_worktree(
task_id: &str,
) -> Result<crate::extras::git_worktree::WorktreeInfo, String> {
let repo = std::env::current_dir()
.map_err(|e| format!("failed to resolve current directory: {e}"))?
.canonicalize()
.map_err(|e| format!("failed to canonicalize current repository: {e}"))?;
if crate::extras::git_worktree::repo_is_dirty(&repo)? {
return Err("current repository has uncommitted changes".to_string());
}
let parent = repo
.parent()
.ok_or_else(|| "current repository has no parent directory".to_string())?;
let name = format!("dirge-task-{task_id}");
crate::extras::git_worktree::create_at(&repo, parent, &name, &name)
}
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub enum SubagentChatEvent {
Spawn {
id: String,
prompt: String,
agent: Option<String>,
},
Complete { id: String, result: String },
Failed { id: String, error: String },
Token { id: String, text: String },
Reasoning { id: String, text: String },
ToolCall {
id: String,
tool_name: String,
args_summary: String,
},
ToolResult {
id: String,
tool_name: String,
output_summary: String,
},
Aborted { id: String },
}
pub const SUBAGENT_CHAT_CAP: usize = 1024;
pub type SubagentChatSender = mpsc::Sender<SubagentChatEvent>;
#[allow(dead_code)]
pub type SubagentChatReceiver = mpsc::Receiver<SubagentChatEvent>;
static SUBAGENT_CHAT_SINK: std::sync::OnceLock<SubagentChatSender> = std::sync::OnceLock::new();
pub fn set_subagent_chat_sink(sink: SubagentChatSender) {
if SUBAGENT_CHAT_SINK.set(sink).is_err() {
tracing::debug!("subagent chat sink already set; ignoring re-set");
}
}
pub fn subagent_chat_sink() -> Option<SubagentChatSender> {
SUBAGENT_CHAT_SINK.get().cloned()
}
static SUBAGENT_ABORT_REGISTRY: std::sync::OnceLock<Mutex<HashMap<String, AbortSignal>>> =
std::sync::OnceLock::new();
fn abort_registry() -> &'static Mutex<HashMap<String, AbortSignal>> {
SUBAGENT_ABORT_REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}
pub fn register_subagent_abort(id: &str, signal: AbortSignal) {
let mut map = abort_registry().lock_ignore_poison();
map.insert(id.to_string(), signal);
}
pub fn unregister_subagent_abort(id: &str) {
let mut map = abort_registry().lock_ignore_poison();
map.remove(id);
}
fn spawn_abort_watcher(
abort: AbortSignal,
handle: tokio::task::AbortHandle,
cancel_tx: tokio::sync::mpsc::Sender<()>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
loop {
if abort.is_cancelled() {
let _ = cancel_tx.try_send(()); handle.abort(); return;
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
})
}
struct SubagentCleanup {
id: String,
watcher: Option<tokio::task::JoinHandle<()>>,
store: Option<BackgroundStore>,
}
impl SubagentCleanup {
fn new(id: String, watcher: tokio::task::JoinHandle<()>) -> Self {
Self {
id,
watcher: Some(watcher),
store: None,
}
}
fn with_store(
id: String,
watcher: tokio::task::JoinHandle<()>,
store: BackgroundStore,
) -> Self {
Self {
id,
watcher: Some(watcher),
store: Some(store),
}
}
fn from_store(id: String, store: BackgroundStore) -> Self {
Self {
id,
watcher: None,
store: Some(store),
}
}
}
impl Drop for SubagentCleanup {
fn drop(&mut self) {
if let Some(watcher) = self.watcher.take() {
watcher.abort();
}
unregister_subagent_abort(&self.id);
if let Some(store) = &self.store {
store.notify_if_running(
&self.id,
TaskState::Failed("subagent exited without producing a result".into()),
);
}
}
}
const SUBAGENT_READONLY_BASE: &[&str] = &[
"read",
"read_minified",
"grep",
"find_files",
"glob",
"list_dir",
"repo_overview",
"websearch",
"webfetch",
];
const SUBAGENT_READWRITE_BASE: &[&str] = &[
"read",
"read_minified",
"grep",
"find_files",
"glob",
"list_dir",
"repo_overview",
"websearch",
"webfetch",
"write",
"edit",
"edit_lines",
"edit_minified",
"apply_patch",
"bash",
"bash_output",
"kill_shell",
];
const SUBAGENT_FORCED_EXCLUDES: &[&str] = &[
"task",
"task_status",
"memory",
"skill",
"spec",
"write_todo_list",
"session_search",
"issue",
"graph",
"question",
"plan_enter",
"plan_exit",
];
pub const SUBAGENT_DEFAULT_MAX_TURNS: usize = 25;
const SUBAGENT_DEFAULT_PREAMBLE: &str = "You are a subagent working on a specific subtask. Complete it thoroughly using the tools available to you. You cannot spawn further subagents.";
fn writer_preamble(
route_preamble: Option<&str>,
worktree: &std::path::Path,
branch: &str,
) -> String {
let mut preamble = route_preamble
.unwrap_or(SUBAGENT_DEFAULT_PREAMBLE)
.to_string();
preamble.push_str(&format!(
"\n\nYou are an isolated writer. Your root is {} on branch {}. Do not modify the parent checkout. Inspect git status, stage intended changes, make one descriptive commit, and report the commit hash, changed files, tests run, and unresolved issues.",
worktree.display(),
branch,
));
preamble
}
pub fn resolve_subagent_allow(
p: &crate::context::agent_defs::SubagentToolPolicy,
) -> Option<Vec<String>> {
use crate::context::agent_defs::SubagentToolTier;
use std::collections::BTreeSet;
let universe: &[&str] = match p.tier {
SubagentToolTier::Toolless => return None,
SubagentToolTier::Readonly => SUBAGENT_READONLY_BASE,
SubagentToolTier::ReadWrite => SUBAGENT_READWRITE_BASE,
};
let mut set: BTreeSet<String> = if p.allow.is_empty() {
universe.iter().map(|s| s.to_ascii_lowercase()).collect()
} else {
p.allow
.iter()
.map(|s| s.to_ascii_lowercase())
.filter(|a| universe.iter().any(|u| u.eq_ignore_ascii_case(a)))
.collect()
};
for d in &p.deny {
set.remove(&d.to_ascii_lowercase());
}
for x in SUBAGENT_FORCED_EXCLUDES {
set.remove(&x.to_ascii_lowercase());
}
Some(set.into_iter().collect())
}
pub fn resolve_subagent_max_turns(p: &crate::context::agent_defs::SubagentToolPolicy) -> usize {
p.max_turns.unwrap_or(SUBAGENT_DEFAULT_MAX_TURNS)
}
pub fn resolve_subagent_timeout(
p: &crate::context::agent_defs::SubagentToolPolicy,
) -> std::time::Duration {
const DEFAULT_SECS: u64 = 600;
const MIN_SECS: u64 = 30;
const MAX_SECS: u64 = 3600;
let raw = p.timeout_secs.unwrap_or(DEFAULT_SECS);
let secs = raw.clamp(MIN_SECS, MAX_SECS);
if secs != raw {
tracing::warn!(
timeout_secs = raw,
resolved_timeout_secs = secs,
"subagent timeout is outside the supported range; clamping"
);
}
std::time::Duration::from_secs(secs)
}
#[derive(Clone)]
pub struct SubagentRoute {
pub model: Option<AnyModel>,
pub preamble: Option<String>,
pub tool_allow: Option<Vec<String>>,
pub max_turns: usize,
pub timeout: std::time::Duration,
pub tier: crate::context::agent_defs::SubagentToolTier,
}
#[derive(Debug, Clone)]
pub struct CoordinatorProfile {
pub name: String,
pub description: Option<String>,
pub tier: crate::context::agent_defs::SubagentToolTier,
}
#[derive(Debug, Clone, Default)]
pub struct CoordinatorProfiles {
pub readonly: Vec<CoordinatorProfile>,
pub readwrite: Vec<CoordinatorProfile>,
}
pub fn resolve_coordinator_profiles(
agent_defs: &crate::context::agent_defs::AgentRegistry,
) -> CoordinatorProfiles {
let mut profiles = CoordinatorProfiles::default();
for definition in agent_defs.iter() {
let profile = CoordinatorProfile {
name: definition.name.clone(),
description: definition.description.clone(),
tier: definition.subagent.tier.clone(),
};
match &profile.tier {
crate::context::agent_defs::SubagentToolTier::Readonly => {
profiles.readonly.push(profile)
}
crate::context::agent_defs::SubagentToolTier::ReadWrite => {
profiles.readwrite.push(profile)
}
crate::context::agent_defs::SubagentToolTier::Toolless => {}
}
}
profiles
}
static SUBAGENT_ROUTES: std::sync::OnceLock<HashMap<String, SubagentRoute>> =
std::sync::OnceLock::new();
pub fn set_subagent_routes(routes: HashMap<String, SubagentRoute>) {
if SUBAGENT_ROUTES.set(routes).is_err() {
tracing::debug!("subagent routes already set; ignoring re-set");
}
}
fn subagent_route(name: &str) -> Option<SubagentRoute> {
SUBAGENT_ROUTES.get()?.get(name).cloned()
}
fn subagent_routes_available() -> bool {
SUBAGENT_ROUTES.get().is_some_and(|m| !m.is_empty())
}
fn subagent_route_names() -> Vec<String> {
let Some(map) = SUBAGENT_ROUTES.get() else {
return Vec::new();
};
let mut names: Vec<String> = map.keys().cloned().collect();
names.sort();
names
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum KillOutcome {
NotFound,
Ambiguous(Vec<String>),
Killed(String),
}
pub fn kill_subagent(id_prefix: &str) -> KillOutcome {
let trimmed = id_prefix.trim();
if trimmed.is_empty() {
return KillOutcome::NotFound;
}
let map = abort_registry().lock_ignore_poison();
if let Some(sig) = map.get(trimmed) {
sig.cancel();
return KillOutcome::Killed(trimmed.to_string());
}
let matches: Vec<String> = map
.keys()
.filter(|k| k.starts_with(trimmed))
.cloned()
.collect();
match matches.len() {
0 => KillOutcome::NotFound,
1 => {
let id = matches.into_iter().next().unwrap();
if let Some(sig) = map.get(&id) {
sig.cancel();
}
KillOutcome::Killed(id)
}
_ => KillOutcome::Ambiguous(matches),
}
}
#[allow(dead_code)]
pub fn registered_subagent_ids() -> Vec<String> {
let map = abort_registry().lock_ignore_poison();
map.keys().cloned().collect()
}
#[cfg(test)]
pub fn clear_abort_registry_for_test() {
let mut map = abort_registry().lock_ignore_poison();
map.clear();
}
fn one_line(s: &str, max: usize) -> String {
let collapsed: String = s.split_whitespace().collect::<Vec<_>>().join(" ");
if collapsed.chars().count() <= max {
collapsed
} else {
let truncated: String = collapsed.chars().take(max).collect();
format!("{truncated}…")
}
}
fn summarize_json_args(args: &serde_json::Value, max: usize) -> String {
let raw = match args {
serde_json::Value::Null => String::new(),
serde_json::Value::Object(map) => map
.iter()
.map(|(k, v)| format!("{k}={}", value_brief(v)))
.collect::<Vec<_>>()
.join(" "),
other => other.to_string(),
};
one_line(&raw, max)
}
fn value_brief(v: &serde_json::Value) -> String {
match v {
serde_json::Value::String(s) => s.clone(),
serde_json::Value::Null => "null".into(),
serde_json::Value::Bool(b) => b.to_string(),
serde_json::Value::Number(n) => n.to_string(),
serde_json::Value::Array(a) => format!("[{}]", a.len()),
serde_json::Value::Object(o) => format!("{{+{}}}", o.len()),
}
}
async fn drain_subagent_runner(
runner: crate::agent::runner::AgentRunner,
id: &str,
emit: impl Fn(SubagentChatEvent),
) -> Result<String, String> {
use crate::agent::runner::AbortRunnerOnDrop;
use crate::event::AgentEvent;
let crate::agent::runner::AgentRunner {
event_rx,
task,
cancel_tx,
..
} = runner;
let _guard = AbortRunnerOnDrop { task, cancel_tx };
let mut rx = event_rx;
let mut text = String::new();
while let Some(event) = rx.recv().await {
match event {
AgentEvent::Token(t) => {
text.push_str(&t);
}
AgentEvent::Reasoning(t) => emit(SubagentChatEvent::Reasoning {
id: id.to_string(),
text: t.to_string(),
}),
AgentEvent::ToolCall { name, args, .. } => emit(SubagentChatEvent::ToolCall {
id: id.to_string(),
tool_name: name.to_string(),
args_summary: summarize_json_args(&args, 120),
}),
AgentEvent::ToolResult { output, .. } => emit(SubagentChatEvent::ToolResult {
id: id.to_string(),
tool_name: String::new(), output_summary: one_line(&output, 120),
}),
AgentEvent::Done { response, .. } => {
let response = if response.is_empty() {
text
} else {
response.to_string()
};
emit(SubagentChatEvent::Token {
id: id.to_string(),
text: response.clone(),
});
emit(SubagentChatEvent::Complete {
id: id.to_string(),
result: response.clone(),
});
return Ok(response);
}
AgentEvent::Error(msg) => return Err(msg.to_string()),
AgentEvent::ContextOverflow { error, .. } => return Err(error.to_string()),
_ => {}
}
}
Err("subagent runner ended without Done".to_string())
}
pub struct TaskTool {
pub permission: Option<PermCheck>,
pub ask_tx: Option<AskSender>,
model: AnyModel,
bg_store: BackgroundStore,
sandbox: Sandbox,
write_isolation: SubagentWriteIsolation,
chat_sink: Option<SubagentChatSender>,
}
impl TaskTool {
pub fn new(
permission: Option<PermCheck>,
ask_tx: Option<AskSender>,
model: AnyModel,
bg_store: BackgroundStore,
sandbox: Sandbox,
write_isolation: SubagentWriteIsolation,
) -> Self {
Self {
permission,
ask_tx,
model,
bg_store,
sandbox,
write_isolation,
chat_sink: None,
}
}
#[allow(dead_code)]
pub fn with_chat_sink(mut self, sink: SubagentChatSender) -> Self {
self.chat_sink = Some(sink);
self
}
fn emit_chat(&self, event: SubagentChatEvent) {
if let Some(sink) = &self.chat_sink {
let _ = sink.try_send(event);
return;
}
if let Some(sink) = subagent_chat_sink() {
let _ = sink.try_send(event);
}
}
#[allow(clippy::too_many_arguments)]
async fn run_tooled(
&self,
route_model: Option<AnyModel>,
route_preamble: Option<String>,
allowed: Vec<String>,
max_turns: usize,
timeout: std::time::Duration,
background: bool,
is_writer: bool,
retry_of: Option<String>,
prompt: String,
agent_name: Option<String>,
) -> Result<String, ToolError> {
let agent = crate::provider::current_agent().ok_or_else(|| {
ToolError::Msg(
"tooled subagents require a live interactive agent; this path (headless/test) \
does not support them. Use a tool-less profile or omit `agent`."
.into(),
)
})?;
if background {
let running = self.bg_store.running_count();
let cap = BackgroundStore::max_concurrent();
if running >= cap {
return Err(ToolError::Msg(format!(
"background subagent cap reached ({}/{} in flight). Wait for one to finish \
(use task_status) or run inline (background=false). Capping prevents fan-out \
from burning the API budget.",
running, cap,
)));
}
let task_id = Uuid::new_v4().to_string();
let coordinator_dispatch = self.bg_store.coordinator_strategy().is_some();
let attempts_isolated_writer = coordinator_dispatch
&& writer_worktree_enabled(
is_writer,
self.write_isolation,
self.sandbox.is_microvm(),
self.sandbox.confines_writes(),
)
.map_err(|error| ToolError::Msg(error.to_string()))?;
if coordinator_dispatch {
self.bg_store
.insert_coordinator_dispatch(
task_id.clone(),
prompt.clone(),
is_writer,
attempts_isolated_writer,
retry_of.as_deref(),
)
.map_err(ToolError::Msg)?;
} else {
self.bg_store.insert_for_dispatch(task_id.clone());
}
let rooted_worktree: Option<(WriterWorktree, std::path::PathBuf, String)> =
if attempts_isolated_writer {
#[cfg(feature = "git-worktree")]
match create_writer_worktree(&task_id) {
Ok(info) => {
let main_git_dir = info.main_repo_path.join(".git");
let base_commit = match crate::extras::git_worktree::head_commit(
&info.main_repo_path,
) {
Ok(commit) => commit,
Err(error) => {
let _ = crate::extras::git_worktree::remove_worktree_if_clean(
&info,
);
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
};
if let Err(error) = self.bg_store.set_coordinator_dispatch_worktree(
&task_id,
info.branch.clone(),
info.worktree_path.display().to_string(),
) {
let _ =
crate::extras::git_worktree::remove_worktree_if_clean(&info);
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
self.bg_store
.register_writer_worktree(task_id.clone(), info.clone());
Some((info, main_git_dir, base_commit))
}
Err(_) if self.write_isolation == SubagentWriteIsolation::Auto => {
tracing::warn!(
task_id = %task_id,
"worktree writer isolation unavailable; falling back to serialized parent checkout"
);
if let Err(error) = self
.bg_store
.set_coordinator_dispatch_isolated(&task_id, false)
{
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
None
}
Err(error) => {
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
}
#[cfg(not(feature = "git-worktree"))]
{
if self.write_isolation == SubagentWriteIsolation::Auto {
tracing::warn!(
task_id = %task_id,
"git-worktree support is unavailable; falling back to serialized parent checkout"
);
if let Err(error) = self
.bg_store
.set_coordinator_dispatch_isolated(&task_id, false)
{
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
None
} else {
let error =
"worktree write isolation requires the git-worktree feature"
.to_string();
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
}
} else {
None
};
if is_writer && rooted_worktree.is_none() {
match current_repo_is_dirty() {
Ok(true) => {
let error = "cannot run a read-write subagent in a dirty parent \
checkout without worktree isolation — commit or \
stash your changes, or run under a Linux bwrap \
sandbox for isolated worktrees"
.to_string();
self.bg_store
.notify(&task_id, TaskState::Failed(error.clone()));
return Err(ToolError::Msg(error));
}
Ok(false) => { }
Err(_) => { }
}
}
self.bg_store.notify_started(&task_id);
self.emit_chat(SubagentChatEvent::Spawn {
id: task_id.clone(),
prompt: prompt.clone(),
agent: agent_name.clone(),
});
let abort = AbortSignal::new();
register_subagent_abort(&task_id, abort.clone());
let store = self.bg_store.clone();
let chat_sink = self.chat_sink.clone();
let tid_for_task = task_id.clone();
let preamble_for_task = route_preamble.clone();
let allowed_for_task = allowed.clone();
let model_for_task = route_model.clone();
let abort_for_task = abort.clone();
let permission_for_task = self.permission.clone();
let ask_tx_for_task = self.ask_tx.clone();
let sandbox_for_task = self.sandbox.clone();
let rooted_worktree_for_task = rooted_worktree.clone();
let route_timeout = timeout;
let store_for_task = store.clone();
let handle = tokio::spawn(async move {
let child_sid = format!("sub-{}", crate::agent::runner::uuid_v4_simple());
let system_prompt = if let Some((info, _, _)) = rooted_worktree_for_task.as_ref() {
writer_preamble(
preamble_for_task.as_deref(),
&info.worktree_path,
&info.branch,
)
} else {
preamble_for_task
.as_deref()
.unwrap_or(SUBAGENT_DEFAULT_PREAMBLE)
.to_string()
};
let runner = if let Some((info, main_git_dir, base_commit)) =
rooted_worktree_for_task.as_ref()
{
let worktree = info.worktree_path.clone();
let root = match crate::agent::tools::ToolRoot::new(&worktree) {
Ok(root) => root,
Err(error) => {
let state = TaskState::Failed(error.to_string());
#[cfg(feature = "git-worktree")]
{
let commits = crate::extras::git_worktree::worktree_commits_since(
info,
base_commit,
)
.unwrap_or_default();
let dirty = crate::extras::git_worktree::worktree_is_dirty(info)
.unwrap_or(true);
let retained = dirty || !commits.is_empty();
if !retained {
let _ =
crate::extras::git_worktree::remove_worktree_if_clean(info);
}
store_for_task.set_coordinator_dispatch_worktree_outcome(
&tid_for_task,
commits,
dirty,
retained,
);
store_for_task.unregister_writer_worktree(&tid_for_task);
}
store_for_task.notify(&tid_for_task, state);
return;
}
};
let tools = crate::agent::builder::build_rooted_writer_tools(
root,
permission_for_task,
ask_tx_for_task,
sandbox_for_task,
crate::sandbox::SandboxExecutionRoot {
worktree,
main_git_dir: main_git_dir.clone(),
},
)
.await;
agent.spawn_subagent_runner_with_tools(
prompt.clone(),
system_prompt,
tools,
&child_sid,
max_turns,
model_for_task.as_ref(),
)
} else {
agent.spawn_subagent_runner(
prompt.clone(),
system_prompt,
&allowed_for_task,
&child_sid,
max_turns,
model_for_task.as_ref(),
)
};
let abort_watcher = spawn_abort_watcher(
abort_for_task.clone(),
runner.task.abort_handle(),
runner.cancel_tx.clone(),
);
let _cleanup = SubagentCleanup::with_store(
tid_for_task.clone(),
abort_watcher,
store_for_task.clone(),
);
let chat_sink_drain = chat_sink.clone();
let emit = move |ev: SubagentChatEvent| {
if let Some(sink) = &chat_sink_drain {
let _ = sink.try_send(ev);
} else if let Some(sink) = subagent_chat_sink() {
let _ = sink.try_send(ev);
}
};
let drained = drain_subagent_runner(runner, &tid_for_task, emit);
let outer = tokio::time::timeout(route_timeout, drained).await;
let aborted = abort_for_task.is_cancelled();
let (state, chat_event) = match outer {
Ok(Ok(text)) => (TaskState::Completed(text), None),
Ok(Err(e)) => {
let aborted_msg = "aborted by user".to_string();
if aborted {
(
TaskState::Cancelled(aborted_msg.clone()),
Some(SubagentChatEvent::Aborted {
id: tid_for_task.clone(),
}),
)
} else {
(
TaskState::Failed(e.clone()),
Some(SubagentChatEvent::Failed {
id: tid_for_task.clone(),
error: e,
}),
)
}
}
Err(_) => {
let msg = format!("subagent timed out after {}s", route_timeout.as_secs());
(
TaskState::Failed(msg.clone()),
Some(SubagentChatEvent::Failed {
id: tid_for_task.clone(),
error: msg,
}),
)
}
};
if let Some(chat_event) = chat_event {
if let Some(sink) = chat_sink {
let _ = sink.try_send(chat_event);
} else if let Some(sink) = subagent_chat_sink() {
let _ = sink.try_send(chat_event);
}
}
#[cfg(feature = "git-worktree")]
if let Some((info, _, base_commit)) = rooted_worktree_for_task.as_ref() {
let commits =
crate::extras::git_worktree::worktree_commits_since(info, base_commit)
.unwrap_or_default();
let dirty =
crate::extras::git_worktree::worktree_is_dirty(info).unwrap_or(true);
let retained = dirty || !commits.is_empty();
if !retained {
let _ = crate::extras::git_worktree::remove_worktree_if_clean(info);
}
store_for_task.set_coordinator_dispatch_worktree_outcome(
&tid_for_task,
commits,
dirty,
retained,
);
store_for_task.unregister_writer_worktree(&tid_for_task);
}
store_for_task.notify(&tid_for_task, state);
});
store.attach_handle(&task_id, handle);
Ok(format!(
"background task started — task_id: {}\n\nThe subagent runs in the background. \
Completion will be delivered automatically as a <system-reminder> at the start \
of your next turn. Do NOT poll task_status or sleep waiting — continue with \
other work.",
task_id
))
} else {
let task_id = Uuid::new_v4().to_string();
self.emit_chat(SubagentChatEvent::Spawn {
id: task_id.clone(),
prompt: prompt.clone(),
agent: agent_name.clone(),
});
let abort = AbortSignal::new();
register_subagent_abort(&task_id, abort.clone());
let child_sid = format!("sub-{}", crate::agent::runner::uuid_v4_simple());
let system_prompt = route_preamble
.as_deref()
.unwrap_or(SUBAGENT_DEFAULT_PREAMBLE)
.to_string();
let runner = agent.spawn_subagent_runner(
prompt.clone(),
system_prompt,
&allowed,
&child_sid,
max_turns,
route_model.as_ref(),
);
let abort_watcher = spawn_abort_watcher(
abort.clone(),
runner.task.abort_handle(),
runner.cancel_tx.clone(),
);
let _cleanup = SubagentCleanup::new(task_id.clone(), abort_watcher);
let result = tokio::time::timeout(
timeout,
drain_subagent_runner(runner, &task_id, |ev| self.emit_chat(ev)),
)
.await;
let aborted = abort.is_cancelled();
match result {
Ok(Ok(text)) => {
let outcome =
crate::agent::tools::output_relay::relay_if_large("task", text, "");
Ok(outcome.text)
}
Ok(Err(e)) => {
if aborted {
self.emit_chat(SubagentChatEvent::Aborted { id: task_id });
Err(ToolError::Msg("Subagent aborted by user".to_string()))
} else {
self.emit_chat(SubagentChatEvent::Failed {
id: task_id,
error: e.clone(),
});
Err(ToolError::Msg(format!("Subagent error: {e}")))
}
}
Err(_) => {
let message = format!("Subagent timed out after {}s", timeout.as_secs());
self.emit_chat(SubagentChatEvent::Failed {
id: task_id,
error: message.clone(),
});
Err(ToolError::Msg(message))
}
}
}
}
}
#[derive(Deserialize)]
pub struct Args {
pub prompt: String,
#[serde(default)]
pub background: Option<bool>,
#[serde(default)]
pub agent: Option<String>,
#[serde(default)]
pub retry_of: Option<String>,
}
fn normalize_retry_of(retry_of: Option<String>) -> Option<String> {
retry_of
.map(|id| id.trim().to_string())
.filter(|id| !id.is_empty())
}
impl Tool for TaskTool {
const NAME: &'static str = "task";
type Error = ToolError;
type Args = Args;
type Output = String;
async fn definition(&self, _prompt: String) -> ToolDefinition {
let mut description = "Spawn a subagent to handle a specific subtask. The subagent runs as a one-shot query (no tools) and returns its result inline. Use for research, analysis, or planning subtasks that don't require file access. Set background=true to run asynchronously — completion is delivered to you automatically as a <system-reminder> at the start of your next turn. Do NOT poll task_status in a loop or sleep waiting; continue with other work."
.to_string();
let mut properties = serde_json::json!({
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": "Task description for the subagent"
},
"background": {
"type": "boolean",
"description": "Run asynchronously (default: false). When true, returns a task_id immediately. The result is delivered automatically as a <system-reminder> on your next turn — do NOT poll task_status."
},
"retry_of": {
"type": "string",
"description": "Coordinator-only: retry one failed background task once."
}
},
"required": ["prompt"]
});
let names = subagent_route_names();
if !names.is_empty() {
description.push_str(&format!(
" Optionally set agent=<name> to run the subagent under a defined profile (its own model + system prompt). Available profiles: {}.",
names.join(", ")
));
if let Some(props) = properties
.get_mut("properties")
.and_then(|p| p.as_object_mut())
{
props.insert(
"agent".to_string(),
serde_json::json!({
"type": "string",
"enum": names,
"description": "Agent profile to run this subagent under (model + system prompt). Omit for the default subagent."
}),
);
}
if SUBAGENT_ROUTES
.get()
.is_some_and(|m| m.values().any(|r| r.tool_allow.is_some()))
{
let has_write = SUBAGENT_ROUTES.get().is_some_and(|m| {
m.values().any(|r| {
r.tool_allow
.as_ref()
.is_some_and(|a| a.iter().any(|t| t == "write" || t == "edit"))
})
});
if has_write {
description.push_str(
" Some profiles enable a read-write tooled subagent (read, grep, glob, \
edit, write, bash — no recursion, no session-scoped tools) that can \
investigate AND edit the repo directly; pick such a profile for \
implementation subtasks. Others are read-only.",
);
} else {
description.push_str(
" Some profiles enable a read-only tooled subagent (read, grep, glob, \
list_dir, etc. — no mutation, no recursion) that can investigate the \
repo directly; pick such a profile for research/exploration subtasks.",
);
}
}
}
ToolDefinition {
name: "task".to_string(),
description,
parameters: properties,
}
}
async fn call(&self, args: Args) -> Result<String, ToolError> {
check_perm(&self.permission, &self.ask_tx, "task", &args.prompt).await?;
let (route_model, route_preamble, tool_allow, max_turns, timeout, tier) = match args
.agent
.as_deref()
{
None => (
None,
None,
None,
SUBAGENT_DEFAULT_MAX_TURNS,
std::time::Duration::from_secs(600),
crate::context::agent_defs::SubagentToolTier::Toolless,
),
Some(name) => {
if !subagent_routes_available() {
return Err(ToolError::Msg(format!(
"agent profile '{}' requested but no profiles are defined. Add .dirge/agents/<name>.md or a config.json \"agents\" entry, or omit `agent`.",
name
)));
}
match subagent_route(name) {
Some(r) => (
r.model,
r.preamble,
r.tool_allow,
r.max_turns,
r.timeout,
r.tier,
),
None => {
return Err(ToolError::Msg(format!(
"unknown agent profile '{}'. Available: {}.",
name,
subagent_route_names().join(", ")
)));
}
}
}
};
let background = args.background.unwrap_or(false);
let retry_of = normalize_retry_of(args.retry_of);
let is_writer = matches!(
tier,
crate::context::agent_defs::SubagentToolTier::ReadWrite
);
if retry_of.is_some() && !background {
return Err(ToolError::Msg("retry_of requires background=true".into()));
}
if self.bg_store.coordinator_strategy().is_some() && !background && is_writer {
return Err(ToolError::Msg(
"coordinator writer subagents require background=true so writer isolation can be enforced"
.into(),
));
}
if self.bg_store.coordinator_strategy()
== Some(crate::config::SubagentDispatchStrategy::Full)
&& !background
&& !matches!(tier, crate::context::agent_defs::SubagentToolTier::Toolless)
{
return Err(ToolError::Msg(
"full coordinator mode requires tier-routed subagents to use background=true"
.into(),
));
}
if let Some(allowed) = tool_allow {
return self
.run_tooled(
route_model,
route_preamble,
allowed,
max_turns,
timeout,
background,
is_writer,
retry_of,
args.prompt,
args.agent,
)
.await;
}
let model = route_model.unwrap_or_else(|| self.model.clone());
let background = args.background.unwrap_or(false);
if background {
let running = self.bg_store.running_count();
let cap = BackgroundStore::max_concurrent();
if running >= cap {
return Err(ToolError::Msg(format!(
"background subagent cap reached ({}/{} in flight). Wait for one to finish (use task_status) or run inline (background=false). Capping prevents fan-out from burning the API budget.",
running, cap,
)));
}
let task_id = Uuid::new_v4().to_string();
if self.bg_store.coordinator_strategy().is_some() {
self.bg_store
.insert_coordinator_dispatch(
task_id.clone(),
args.prompt.clone(),
is_writer,
false,
retry_of.as_deref(),
)
.map_err(ToolError::Msg)?;
} else {
self.bg_store.insert_for_dispatch(task_id.clone());
}
self.bg_store.notify_started(&task_id);
self.emit_chat(SubagentChatEvent::Spawn {
id: task_id.clone(),
prompt: args.prompt.clone(),
agent: args.agent.clone(),
});
let abort = AbortSignal::new();
register_subagent_abort(&task_id, abort.clone());
let prompt = args.prompt;
let store = self.bg_store.clone();
let tid = task_id.clone();
let chat_sink = self.chat_sink.clone();
let abort_for_task = abort.clone();
let preamble_for_task = route_preamble.clone();
let route_timeout = timeout;
let store_for_task = store.clone();
let tid_for_task = tid.clone();
let handle = tokio::spawn(async move {
let _cleanup =
SubagentCleanup::from_store(tid_for_task.clone(), store_for_task.clone());
let fut = model.btw_query_with(
format!(
"You are a subagent working on a specific subtask. Complete it thoroughly.\n\nTask: {}",
prompt
),
preamble_for_task.as_deref(),
);
let abort_check = abort_for_task.clone();
let raced = async {
tokio::pin!(fut);
loop {
tokio::select! {
r = &mut fut => break Ok::<_, ()>(r),
_ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
if abort_check.is_cancelled() {
break Err(());
}
}
}
}
};
let outer = tokio::time::timeout(route_timeout, raced).await;
let (state, chat_event) = match outer {
Ok(Ok(Ok(text))) => (
TaskState::Completed(text.clone()),
SubagentChatEvent::Token {
id: tid_for_task.clone(),
text: text.clone(),
},
),
Ok(Ok(Err(e))) => {
let msg = e.to_string();
(
TaskState::Failed(msg.clone()),
SubagentChatEvent::Failed {
id: tid_for_task.clone(),
error: msg,
},
)
}
Ok(Err(())) => {
let msg = "aborted by user".to_string();
(
TaskState::Cancelled(msg.clone()),
SubagentChatEvent::Aborted {
id: tid_for_task.clone(),
},
)
}
Err(_) => {
let msg = format!("subagent timed out after {}s", route_timeout.as_secs(),);
(
TaskState::Failed(msg.clone()),
SubagentChatEvent::Failed {
id: tid_for_task.clone(),
error: msg,
},
)
}
};
let final_event = match &chat_event {
SubagentChatEvent::Token { id, text } => {
if let Some(sink) = &chat_sink {
let _ = sink.try_send(chat_event.clone());
} else if let Some(sink) = subagent_chat_sink() {
let _ = sink.try_send(chat_event.clone());
}
SubagentChatEvent::Complete {
id: id.clone(),
result: text.clone(),
}
}
_ => chat_event.clone(),
};
if let Some(sink) = chat_sink {
let _ = sink.try_send(final_event);
} else if let Some(sink) = subagent_chat_sink() {
let _ = sink.try_send(final_event);
}
unregister_subagent_abort(&tid_for_task);
store_for_task.notify(&tid_for_task, state);
});
store.attach_handle(&tid, handle);
Ok(format!(
"background task started — task_id: {}\n\nThe subagent runs in the background. Completion will be delivered automatically as a <system-reminder> at the start of your next turn. Do NOT poll task_status or sleep waiting — continue with other work.",
task_id
))
} else {
let task_id = Uuid::new_v4().to_string();
self.emit_chat(SubagentChatEvent::Spawn {
id: task_id.clone(),
prompt: args.prompt.clone(),
agent: args.agent.clone(),
});
let abort = AbortSignal::new();
register_subagent_abort(&task_id, abort.clone());
let fut = model.btw_query_with(
format!(
"You are a subagent working on a specific subtask. Complete it thoroughly.\n\nTask: {}",
args.prompt
),
route_preamble.as_deref(),
);
let abort_check = abort.clone();
let raced = async {
tokio::pin!(fut);
loop {
tokio::select! {
r = &mut fut => break Ok::<_, ()>(r),
_ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
if abort_check.is_cancelled() {
break Err(());
}
}
}
}
};
let result = raced.await;
unregister_subagent_abort(&task_id);
match result {
Ok(Ok(text)) => {
self.emit_chat(SubagentChatEvent::Token {
id: task_id.clone(),
text: text.clone(),
});
self.emit_chat(SubagentChatEvent::Complete {
id: task_id,
result: text.clone(),
});
let outcome =
crate::agent::tools::output_relay::relay_if_large("task", text, "");
Ok(outcome.text)
}
Ok(Err(e)) => {
let msg = e.to_string();
self.emit_chat(SubagentChatEvent::Failed {
id: task_id,
error: msg.clone(),
});
Err(ToolError::Msg(format!("Subagent error: {}", msg)))
}
Err(()) => {
self.emit_chat(SubagentChatEvent::Aborted { id: task_id });
Err(ToolError::Msg("Subagent aborted by user".to_string()))
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::agent::tools::background::BackgroundStore;
use crate::provider::AnyModel;
use rig::client::CompletionClient;
use rig::providers::openrouter;
#[test]
fn blank_retry_id_is_treated_as_absent() {
assert_eq!(normalize_retry_of(Some("".to_string())), None);
assert_eq!(normalize_retry_of(Some(" ".to_string())), None);
assert_eq!(
normalize_retry_of(Some(" task-id ".to_string())),
Some("task-id".to_string())
);
}
fn mock_tool() -> TaskTool {
let client = openrouter::Client::builder()
.api_key("test-key")
.http_client(crate::provider::compressing_http::CompressingHttpClient::default())
.build()
.unwrap();
let model = client.completion_model("anthropic/claude-sonnet-4.5");
TaskTool::new(
None,
None,
AnyModel::OpenRouter(model),
BackgroundStore::new(),
Sandbox::new(crate::sandbox::SandboxMode::Off),
SubagentWriteIsolation::Serialize,
)
}
#[test]
fn isolated_writer_preamble_requires_commit_and_report() {
let preamble = writer_preamble(
None,
std::path::Path::new("/tmp/dirge-task-writer"),
"dirge-task-writer",
);
assert!(preamble.contains("stage intended changes"));
assert!(preamble.contains("one descriptive commit"));
assert!(preamble.contains("commit hash"));
assert!(preamble.contains("/tmp/dirge-task-writer"));
}
#[test]
fn writer_worktree_policy_respects_mode_and_sandbox() {
assert_eq!(
writer_worktree_enabled(false, SubagentWriteIsolation::Worktree, false, false),
Ok(false)
);
assert_eq!(
writer_worktree_enabled(true, SubagentWriteIsolation::Serialize, false, false),
Ok(false)
);
assert_eq!(
writer_worktree_enabled(true, SubagentWriteIsolation::Auto, false, true),
Ok(true)
);
assert_eq!(
writer_worktree_enabled(true, SubagentWriteIsolation::Auto, false, false),
Ok(false)
);
assert!(
writer_worktree_enabled(true, SubagentWriteIsolation::Worktree, false, false).is_err()
);
assert_eq!(
writer_worktree_enabled(true, SubagentWriteIsolation::Worktree, false, true),
Ok(true)
);
assert_eq!(
writer_worktree_enabled(true, SubagentWriteIsolation::Auto, true, true),
Ok(false)
);
assert!(
writer_worktree_enabled(true, SubagentWriteIsolation::Worktree, true, true).is_err()
);
}
#[test]
fn model_provider_name_follows_model_variant() {
use rig::client::CompletionClient;
let openai = rig::providers::openai::CompletionsClient::builder()
.http_client(crate::provider::compressing_http::CompressingHttpClient::default())
.api_key("test-key")
.build()
.unwrap()
.completion_model("gpt-test");
let anthropic = rig::providers::anthropic::Client::builder()
.api_key("test-key")
.http_client(crate::provider::compressing_http::CompressingHttpClient::default())
.build()
.unwrap()
.completion_model("claude-test");
assert_eq!(AnyModel::OpenAI(openai).provider_name(), "openai");
assert_eq!(AnyModel::Anthropic(anthropic).provider_name(), "anthropic");
}
#[tokio::test]
async fn cancel_all_cleans_tooled_subagent_registry_and_watcher() {
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
let _guard = registry_test_lock().await;
clear_abort_registry_for_test();
let id = "cancel-all-tooled";
register_subagent_abort(id, AbortSignal::new());
struct MarkDropped(Arc<AtomicBool>);
impl Drop for MarkDropped {
fn drop(&mut self) {
self.0.store(true, Ordering::SeqCst);
}
}
let watcher_dropped = Arc::new(AtomicBool::new(false));
let dropped = watcher_dropped.clone();
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
let outer = tokio::spawn(async move {
let watcher = tokio::spawn(async move {
let _mark = MarkDropped(dropped);
std::future::pending::<()>().await;
});
let _cleanup = SubagentCleanup::new(id.to_string(), watcher);
let _ = ready_tx.send(());
std::future::pending::<()>().await;
});
ready_rx.await.unwrap();
let store = BackgroundStore::new();
store.insert(id.to_string());
store.attach_handle(id, outer);
store.cancel_all();
tokio::time::timeout(std::time::Duration::from_secs(1), async {
while registered_subagent_ids()
.iter()
.any(|task_id| task_id == id)
|| !watcher_dropped.load(Ordering::SeqCst)
{
tokio::task::yield_now().await;
}
})
.await
.expect("cancel_all must drop cleanup and abort watcher");
}
#[tokio::test]
async fn abort_watcher_can_be_stopped_after_runner_completion() {
let runner_task = tokio::spawn(std::future::pending::<()>());
let (cancel_tx, _cancel_rx) = tokio::sync::mpsc::channel(1);
let watcher =
spawn_abort_watcher(AbortSignal::new(), runner_task.abort_handle(), cancel_tx);
watcher.abort();
let result = tokio::time::timeout(std::time::Duration::from_secs(1), watcher)
.await
.expect("aborted watcher must terminate");
assert!(result.unwrap_err().is_cancelled());
runner_task.abort();
}
#[tokio::test]
async fn definition_steers_agent_away_from_polling() {
let tool = mock_tool();
let def = tool.definition(String::new()).await;
let desc = def.description.to_lowercase();
assert!(
desc.contains("system-reminder") || desc.contains("automatically"),
"task description must reference automatic notification: {}",
def.description
);
assert!(
desc.contains("do not poll") || desc.contains("not poll"),
"task description must explicitly discourage polling: {}",
def.description
);
}
#[test]
fn subagent_path_is_stateless_no_session_search_leakage() {
let _expected_fields = ["permission", "ask_tx", "model", "bg_store", "chat_sink"];
let _tool: TaskTool = mock_tool();
let provider_src = include_str!("../../provider/dispatch.rs");
let btw_idx = provider_src
.find("pub async fn btw_query_with")
.expect("btw_query_with must exist in provider/dispatch.rs");
let btw_end = provider_src[btw_idx..]
.find("\n }\n")
.map(|i| btw_idx + i)
.unwrap_or(provider_src.len());
let btw_body = &provider_src[btw_idx..btw_end];
assert!(
!btw_body.contains(".tool("),
"btw_query_with must not attach tools to the subagent — that would \
require auditing session_id propagation per dirge-mifq. \
Source snippet:\n{btw_body}"
);
assert!(
!btw_body.contains(".tools("),
"btw_query_with must not attach tools to the subagent — that would \
require auditing session_id propagation per dirge-mifq."
);
let leaky = [
"session_search",
"memory",
"skill",
"issue",
"graph",
"write_todo_list",
"spec",
"task",
"task_status",
"question",
"plan_enter",
"plan_exit",
];
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
for tier in [SubagentToolTier::Readonly, SubagentToolTier::ReadWrite] {
let policy = SubagentToolPolicy {
tier: tier.clone(),
allow: leaky.iter().map(|s| s.to_string()).collect(), deny: Vec::new(),
max_turns: None,
timeout_secs: None,
};
let resolved = resolve_subagent_allow(&policy)
.unwrap_or_else(|| panic!("{tier:?} should yield a tool set"));
for l in leaky {
assert!(
!resolved.iter().any(|t| t == l),
"{tier:?} subagent allow-list must exclude {l:?} (dirge-mifq); got {resolved:?}"
);
}
}
}
#[test]
fn resolve_readonly_base_is_expected_universe() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::Readonly,
..Default::default()
};
let mut got = resolve_subagent_allow(&p).unwrap();
got.sort();
let mut want: Vec<String> = SUBAGENT_READONLY_BASE
.iter()
.map(|s| s.to_string())
.collect();
want.sort();
assert_eq!(got, want, "readonly base must match the verified universe");
}
#[test]
fn non_empty_allow_narrows_readonly_base() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::Readonly,
allow: vec!["read".into(), "grep".into()],
..Default::default()
};
assert_eq!(
resolve_subagent_allow(&p).unwrap(),
vec!["grep".to_string(), "read".to_string()]
);
}
#[test]
fn resolve_toolless_returns_none() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::Toolless,
..Default::default()
};
assert_eq!(resolve_subagent_allow(&p), None);
}
#[test]
fn resolve_readwrite_includes_writes_and_excludes_leaky() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::ReadWrite,
..Default::default()
};
let got = resolve_subagent_allow(&p).expect("readwrite yields a tool set");
for w in ["write", "edit", "edit_lines", "apply_patch", "bash"] {
assert!(
got.iter().any(|t| t == w),
"readwrite must include {w}: {got:?}"
);
}
assert!(got.iter().any(|t| t == "read"));
assert!(got.iter().any(|t| t == "grep"));
for l in [
"session_search",
"memory",
"skill",
"issue",
"graph",
"write_todo_list",
"spec",
"task",
"task_status",
"question",
] {
assert!(
!got.iter().any(|t| t == l),
"readwrite must still exclude {l:?} (dirge-mifq): {got:?}"
);
}
}
#[test]
fn deny_narrows_readwrite_base() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::ReadWrite,
deny: vec!["bash".into(), "edit".into()],
..Default::default()
};
let got = resolve_subagent_allow(&p).unwrap();
assert!(!got.iter().any(|t| t == "bash"));
assert!(!got.iter().any(|t| t == "edit"));
assert!(got.iter().any(|t| t == "write"));
assert!(got.iter().any(|t| t == "apply_patch"));
}
#[test]
fn deny_narrows_readonly_base() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::Readonly,
deny: vec!["webfetch".into(), "grep".into()],
..Default::default()
};
let got = resolve_subagent_allow(&p).unwrap();
assert!(!got.iter().any(|t| t == "webfetch"));
assert!(!got.iter().any(|t| t == "grep"));
assert!(got.iter().any(|t| t == "read"));
assert!(got.iter().any(|t| t == "glob"));
}
#[test]
fn allow_cannot_escalate_past_readonly() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let p = SubagentToolPolicy {
tier: SubagentToolTier::Readonly,
allow: vec!["bash".into(), "edit".into(), "write".into()],
..Default::default()
};
let got = resolve_subagent_allow(&p).unwrap();
assert!(
got.is_empty(),
"invalid-only allow must narrow to no tools: {got:?}"
);
}
#[test]
fn resolve_max_turns_honors_override_and_default() {
use crate::context::agent_defs::{SubagentToolPolicy, SubagentToolTier};
let none = SubagentToolPolicy {
tier: SubagentToolTier::Readonly,
..Default::default()
};
assert_eq!(
resolve_subagent_max_turns(&none),
SUBAGENT_DEFAULT_MAX_TURNS
);
let over = SubagentToolPolicy {
tier: SubagentToolTier::Readonly,
max_turns: Some(7),
..Default::default()
};
assert_eq!(resolve_subagent_max_turns(&over), 7);
}
#[test]
fn one_line_and_summarize_helpers_truncate() {
assert_eq!(one_line("hello world", 100), "hello world");
let long = "a ".repeat(200);
assert!(one_line(&long, 50).chars().count() <= 51); let args = serde_json::json!({"path": "/tmp/x", "n": 3});
let got = summarize_json_args(&args, 120);
let tokens: Vec<&str> = got.split(' ').collect();
assert!(tokens.contains(&"path=/tmp/x"), "got: {got}");
assert!(tokens.contains(&"n=3"), "got: {got}");
}
#[tokio::test]
async fn drain_buffers_tokens_and_emits_response_once() {
use crate::agent::runner::AgentRunner;
use crate::event::{AgentEvent, ToolContent};
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc;
fn replay(events: Vec<AgentEvent>) -> AgentRunner {
let (tx, event_rx) = mpsc::channel(events.len().max(1));
for e in events {
tx.try_send(e).expect("test channel sized to fit events");
}
drop(tx); let (interject_tx, _) = mpsc::channel(1);
let (cancel_tx, _) = mpsc::channel(1);
let task = tokio::spawn(async {});
AgentRunner {
event_rx,
task,
interject_tx,
cancel_tx,
}
}
let runner = replay(vec![
AgentEvent::Token("hello ".into()),
AgentEvent::Token("world".into()),
AgentEvent::Reasoning("thinking".into()),
AgentEvent::ToolCall {
id: "c1".into(),
name: "read".into(),
args: serde_json::json!({"path": "/tmp/x"}),
},
AgentEvent::ToolResult {
id: "c1".into(),
output: "file contents".into(),
kind: ToolContent::Text,
},
AgentEvent::Done {
response: "hello world".into(),
tokens: 5,
cost: 0.0,
},
]);
let collected: Arc<Mutex<Vec<SubagentChatEvent>>> = Arc::new(Mutex::new(Vec::new()));
let sink = collected.clone();
let text = drain_subagent_runner(runner, "sub-1", move |ev| {
sink.lock().unwrap().push(ev);
})
.await
.expect("drain returns the Done response text");
assert_eq!(text, "hello world");
let events = collected.lock().unwrap();
use SubagentChatEvent::*;
assert!(matches!(
&events[0],
Reasoning { id, text } if id == "sub-1" && text == "thinking"
));
assert!(matches!(
&events[1],
ToolCall { id, tool_name, args_summary }
if id == "sub-1" && tool_name == "read" && args_summary.contains("path=/tmp/x")
));
assert!(matches!(
&events[2],
ToolResult { id, output_summary, .. }
if id == "sub-1" && output_summary.contains("file contents")
));
assert!(matches!(
&events[3],
Token { id, text } if id == "sub-1" && text == "hello world"
));
assert!(matches!(
&events[4],
Complete { id, result } if id == "sub-1" && result == "hello world"
));
assert_eq!(
events.len(),
5,
"success must emit one buffered response then Complete: {events:?}"
);
}
#[tokio::test]
async fn subagent_routing_advertises_and_validates() {
let mut routes = HashMap::new();
routes.insert(
"reviewer".to_string(),
SubagentRoute {
model: None,
preamble: Some("You are a reviewer.".to_string()),
tool_allow: None,
max_turns: SUBAGENT_DEFAULT_MAX_TURNS,
timeout: std::time::Duration::from_secs(600),
tier: crate::context::agent_defs::SubagentToolTier::Toolless,
},
);
set_subagent_routes(routes);
assert!(subagent_routes_available());
assert_eq!(subagent_route_names(), vec!["reviewer".to_string()]);
assert!(subagent_route("reviewer").is_some());
assert!(subagent_route("ghost").is_none());
let tool = mock_tool();
let def = tool.definition(String::new()).await;
assert!(
def.description.contains("Available profiles: reviewer"),
"definition must list installed profiles: {}",
def.description
);
let props = def
.parameters
.get("properties")
.and_then(|v| v.as_object())
.expect("properties present");
let agent_prop = props.get("agent").expect("agent property advertised");
assert_eq!(agent_prop["enum"][0], "reviewer");
}
#[tokio::test]
async fn definition_advertises_background_field() {
let tool = mock_tool();
let def = tool.definition(String::new()).await;
let props = def
.parameters
.get("properties")
.and_then(|v| v.as_object())
.expect("properties present");
assert!(props.contains_key("background"));
let bg_desc = props["background"]["description"]
.as_str()
.unwrap()
.to_lowercase();
assert!(bg_desc.contains("automatically") || bg_desc.contains("system-reminder"));
assert!(bg_desc.contains("do not poll") || bg_desc.contains("not poll"));
}
#[test]
fn task_short_output_returned_verbatim() {
let short = "subagent: 42 is the answer.\n".to_string();
let outcome = crate::agent::tools::output_relay::relay_if_large("task", short.clone(), "");
assert!(
outcome.relayed_to.is_none(),
"short output must not trigger the disk relay",
);
assert_eq!(
outcome.text, short,
"short subagent output must round-trip unchanged to the parent",
);
}
#[test]
fn task_large_output_relayed_to_disk_with_summary() {
let huge: String = "subagent line\n".repeat(5_000);
let original_len = huge.len();
let outcome = crate::agent::tools::output_relay::relay_if_large("task", huge, "");
let path = outcome
.relayed_to
.as_ref()
.expect("large output must trigger the disk relay");
assert!(path.exists(), "relayed file must exist at {path:?}");
let written = std::fs::read_to_string(path).expect("read relayed file");
assert_eq!(
written.len(),
original_len,
"the FULL original payload must be on disk (not the truncated head)",
);
let summary = &outcome.text;
assert!(
summary.len() < original_len,
"summary should be much smaller than the original payload",
);
assert!(
summary.contains("`read`"),
"summary must mention the `read` tool so the agent can recover the full payload: {summary}",
);
assert!(
summary.contains("transient") || summary.contains(".dirge"),
"summary must reference the transient path: {summary}",
);
let _ = std::fs::remove_file(path);
}
async fn registry_test_lock() -> tokio::sync::MutexGuard<'static, ()> {
static LOCK: std::sync::OnceLock<tokio::sync::Mutex<()>> = std::sync::OnceLock::new();
LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
.lock()
.await
}
#[tokio::test]
async fn kill_unknown_id_no_op() {
let _guard = registry_test_lock().await;
clear_abort_registry_for_test();
assert_eq!(kill_subagent("abc"), KillOutcome::NotFound);
assert_eq!(kill_subagent(""), KillOutcome::NotFound);
let sig = AbortSignal::new();
register_subagent_abort("aaa-1111", sig.clone());
assert_eq!(kill_subagent("zzz"), KillOutcome::NotFound);
assert!(
!sig.is_cancelled(),
"unmatched kill must NOT cancel the surviving subagent",
);
unregister_subagent_abort("aaa-1111");
}
#[tokio::test]
async fn kill_resolves_by_prefix_unique_match() {
let _guard = registry_test_lock().await;
clear_abort_registry_for_test();
let sig_a = AbortSignal::new();
let sig_b = AbortSignal::new();
register_subagent_abort("aa11-deadbeef", sig_a.clone());
register_subagent_abort("bb22-cafef00d", sig_b.clone());
match kill_subagent("aa") {
KillOutcome::Killed(id) => assert_eq!(id, "aa11-deadbeef"),
other => panic!("expected Killed; got {:?}", other),
}
assert!(sig_a.is_cancelled(), "matched signal must be cancelled");
assert!(!sig_b.is_cancelled(), "unmatched signal must survive");
let sig_a2 = AbortSignal::new();
register_subagent_abort("aa99-othertask", sig_a2.clone());
match kill_subagent("aa") {
KillOutcome::Ambiguous(ids) => {
assert_eq!(ids.len(), 2);
assert!(ids.iter().any(|i| i == "aa11-deadbeef"));
assert!(ids.iter().any(|i| i == "aa99-othertask"));
}
other => panic!("expected Ambiguous; got {:?}", other),
}
assert!(
!sig_a2.is_cancelled(),
"ambiguous kill must NOT cancel any signal",
);
clear_abort_registry_for_test();
let s1 = AbortSignal::new();
let s2 = AbortSignal::new();
register_subagent_abort("abc", s1.clone());
register_subagent_abort("abcdef", s2.clone());
match kill_subagent("abc") {
KillOutcome::Killed(id) => assert_eq!(id, "abc"),
other => panic!("expected exact-match Killed; got {:?}", other),
}
assert!(s1.is_cancelled());
assert!(!s2.is_cancelled());
clear_abort_registry_for_test();
}
#[tokio::test]
async fn subagent_complete_after_kill_returns_aborted_result() {
let _guard = registry_test_lock().await;
clear_abort_registry_for_test();
let tid = "t-abort-1";
let abort = AbortSignal::new();
register_subagent_abort(tid, abort.clone());
let abort_check = abort.clone();
let fut = async {
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
Ok::<String, anyhow::Error>("never-arrives".to_string())
};
let raced = async {
tokio::pin!(fut);
loop {
tokio::select! {
r = &mut fut => break Ok::<_, ()>(r),
_ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {
if abort_check.is_cancelled() {
break Err(());
}
}
}
}
};
let killer = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(75)).await;
assert!(matches!(kill_subagent("t-abort"), KillOutcome::Killed(_)));
});
let result = tokio::time::timeout(std::time::Duration::from_secs(2), raced)
.await
.expect("racer must exit before the 2s test timeout");
killer.await.unwrap();
match result {
Err(()) => { }
Ok(other) => panic!("expected abort; got Ok({:?})", other),
}
unregister_subagent_abort(tid);
clear_abort_registry_for_test();
}
#[test]
fn subagent_chat_channel_is_bounded_and_drops_on_overflow() {
let (tx, _rx) = mpsc::channel::<SubagentChatEvent>(SUBAGENT_CHAT_CAP);
for i in 0..SUBAGENT_CHAT_CAP {
tx.try_send(SubagentChatEvent::Token {
id: "x".into(),
text: format!("{i}"),
})
.expect("sends within capacity succeed");
}
let overflow = tx.try_send(SubagentChatEvent::Token {
id: "x".into(),
text: "overflow".into(),
});
assert!(
overflow.is_err(),
"channel must be bounded — an over-capacity try_send drops"
);
}
#[test]
fn subagent_token_event_routes_to_chat_slot() {
let (tx, mut rx) = mpsc::channel::<SubagentChatEvent>(SUBAGENT_CHAT_CAP);
tx.try_send(SubagentChatEvent::Token {
id: "a1".into(),
text: "hello world".into(),
})
.unwrap();
tx.try_send(SubagentChatEvent::Reasoning {
id: "a1".into(),
text: "thinking".into(),
})
.unwrap();
tx.try_send(SubagentChatEvent::Aborted { id: "a1".into() })
.unwrap();
match rx.try_recv().unwrap() {
SubagentChatEvent::Token { id, text } => {
assert_eq!(id, "a1");
assert_eq!(text, "hello world");
}
other => panic!("expected Token; got {:?}", other),
}
match rx.try_recv().unwrap() {
SubagentChatEvent::Reasoning { id, text } => {
assert_eq!(id, "a1");
assert_eq!(text, "thinking");
}
other => panic!("expected Reasoning; got {:?}", other),
}
match rx.try_recv().unwrap() {
SubagentChatEvent::Aborted { id } => assert_eq!(id, "a1"),
other => panic!("expected Aborted; got {:?}", other),
}
}
#[test]
fn subagent_tool_call_event_routes_to_chat_slot() {
let (tx, mut rx) = mpsc::channel::<SubagentChatEvent>(SUBAGENT_CHAT_CAP);
tx.try_send(SubagentChatEvent::ToolCall {
id: "a1".into(),
tool_name: "read".into(),
args_summary: "path=/tmp/x".into(),
})
.unwrap();
tx.try_send(SubagentChatEvent::ToolResult {
id: "a1".into(),
tool_name: "read".into(),
output_summary: "12 lines".into(),
})
.unwrap();
match rx.try_recv().unwrap() {
SubagentChatEvent::ToolCall {
id,
tool_name,
args_summary,
} => {
assert_eq!(id, "a1");
assert_eq!(tool_name, "read");
assert_eq!(args_summary, "path=/tmp/x");
}
other => panic!("expected ToolCall; got {:?}", other),
}
match rx.try_recv().unwrap() {
SubagentChatEvent::ToolResult {
id,
tool_name,
output_summary,
} => {
assert_eq!(id, "a1");
assert_eq!(tool_name, "read");
assert_eq!(output_summary, "12 lines");
}
other => panic!("expected ToolResult; got {:?}", other),
}
}
}