Dark-Alex-17/coyote
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

refactor: Modified the naming of several generalized supervisor values

Browse Source
This commit is contained in:
Dark-Alex-17
2026-08-26 12:43:51 -06:00
parent 5a9f8c42b9
commit 4e50b4ff4a
16 changed files with 253 additions and 1355 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/acp/server.rs
+2 -2
View File
@@ -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::supervisor::{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
-2
View File
@@ -454,8 +454,6 @@ impl Agent {
output.push_str(DEFAULT_SPAWN_INSTRUCTIONS);
}
// Job declarations are appended at init iff jobs are enabled for this
// agent, so their presence doubles as the jobs_enabled predicate.
if self
.functions
.declarations()
src/config/app_config.rs
+13 -13
View File
@@ -844,8 +844,8 @@ 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),
}
}
}
@@ -868,35 +868,35 @@ mod tests {
#[serial_test::serial]
fn load_envs_overrides_max_concurrent_jobs() {
let env_name = get_env_name("max_concurrent_jobs");
let prev = std::env::var_os(&env_name);
let prev = env::var_os(&env_name);
let mut app = AppConfig::default();
unsafe { std::env::set_var(&env_name, "7") };
unsafe { env::set_var(&env_name, "7") };
app.load_envs();
assert_eq!(app.max_concurrent_jobs, Some(7));
unsafe { std::env::set_var(&env_name, "0") };
unsafe { env::set_var(&env_name, "0") };
app.load_envs();
assert_eq!(app.max_concurrent_jobs, Some(0));
unsafe { std::env::remove_var(&env_name) };
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) => 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 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()
@@ -913,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();
@@ -983,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
View File
@@ -73,6 +73,7 @@ impl AppState {
if !mcp_registry.is_empty() && config.mcp_server_support {
functions.append_mcp_meta_functions(mcp_registry.server_features());
}
if jobs_enabled(None, &config) {
functions.append_job_functions();
}
src/config/input.rs
-1
View File
@@ -163,7 +163,6 @@ impl Input {
self.data_urls.clone()
}
/// Names of the function declarations this request will send to the model.
pub fn declared_function_names(&self) -> HashSet<String> {
self.functions
.as_ref()
src/config/request_context.rs
+32 -28
View File
@@ -334,7 +334,7 @@ pub struct RequestContext {
pub notification_queue: Arc<NotificationQueue>,
pub current_depth: usize,
pub auto_continue_count: usize,
pub pending_agents_guardrail_count: u32,
pub pending_tasks_guardrail_count: u32,
pub todo_list: TodoList,
pub skill_registry: SkillRegistry,
pub last_continuation_response: Option<String>,
@@ -369,7 +369,7 @@ impl RequestContext {
notification_queue: Arc::new(NotificationQueue::new()),
current_depth: 0,
auto_continue_count: 0,
pending_agents_guardrail_count: 0,
pending_tasks_guardrail_count: 0,
todo_list: TodoList::default(),
skill_registry: SkillRegistry::default(),
last_continuation_response: None,
@@ -430,7 +430,7 @@ impl RequestContext {
notification_queue: Arc::new(NotificationQueue::new()),
current_depth: 0,
auto_continue_count: 0,
pending_agents_guardrail_count: 0,
pending_tasks_guardrail_count: 0,
todo_list: TodoList::default(),
skill_registry: SkillRegistry::default(),
last_continuation_response: None,
@@ -478,7 +478,7 @@ impl RequestContext {
notification_queue: self.notification_queue.clone(),
current_depth: self.current_depth,
auto_continue_count: 0,
pending_agents_guardrail_count: 0,
pending_tasks_guardrail_count: 0,
todo_list: self.todo_list.clone(),
skill_registry: self.skill_registry.clone(),
last_continuation_response: None,
@@ -524,7 +524,7 @@ impl RequestContext {
notification_queue: Arc::new(NotificationQueue::new()),
current_depth,
auto_continue_count: 0,
pending_agents_guardrail_count: 0,
pending_tasks_guardrail_count: 0,
todo_list: TodoList::default(),
skill_registry: SkillRegistry::default(),
last_continuation_response: None,
@@ -923,6 +923,12 @@ impl RequestContext {
pub fn before_chat_completion(&mut self, input: &Input) -> Result<()> {
// `job__start` validates against exactly what was declared to the
// model for THIS request; refresh it every time.
//
// This is necessary to prevent the model from invoking functions it
// otherwise wouldn't have access to by going through the free `tool`
// argument of `job__start`. If a function is disabled, the model
// shouldn't be able to invoke it at all in any way. This prevents
// that backdoor.
self.declared_function_names = input.declared_function_names();
self.last_message = Some(LastMessage::new(input.clone(), String::new()));
Ok(())
@@ -4082,11 +4088,6 @@ impl RequestContext {
Ok(())
}
#[allow(dead_code)]
pub fn jobs_enabled(&self) -> bool {
jobs_enabled(self.agent.as_ref(), &self.app.config)
}
pub async fn use_agent(
&mut self,
app: &AppConfig,
@@ -4179,7 +4180,7 @@ impl RequestContext {
let jobs_enabled = jobs_enabled(Some(&agent), app);
let should_init_supervisor = agent.can_spawn_agents() || jobs_enabled;
let max_concurrent = if agent.can_spawn_agents() {
let max_concurrent_agents = if agent.can_spawn_agents() {
agent.max_concurrent_agents()
} else {
0
@@ -4188,7 +4189,8 @@ impl RequestContext {
let max_jobs = effective_max_concurrent_jobs(Some(&agent), app);
let supervisor = should_init_supervisor.then(|| {
Arc::new(RwLock::new(
Supervisor::new(max_concurrent, max_depth).with_max_concurrent_jobs(max_jobs),
Supervisor::new(max_concurrent_agents, max_depth)
.with_max_concurrent_jobs(max_jobs),
))
});
@@ -4254,7 +4256,7 @@ impl RequestContext {
self.notification_queue = Arc::new(NotificationQueue::new());
self.current_depth = 0;
self.auto_continue_count = 0;
self.pending_agents_guardrail_count = 0;
self.pending_tasks_guardrail_count = 0;
self.todo_list = TodoList::default();
self.rag.take();
// Cleared alongside `rag` so the pair never disagrees: an agent RAG is
@@ -4750,18 +4752,22 @@ mod tests {
use super::*;
use crate::config::AppState;
use crate::config::agent::AgentConfig;
use crate::function::jobs::RingBuf;
use crate::function::{ToolCall, skill};
use crate::mcp::{McpServer, McpServerFeatures, McpServersConfig, McpTransportType};
use crate::supervisor::{
AgentExitStatus, AgentHandle, AgentResult, JobHandle, JobResult, JobState, JobStatus,
};
use crate::utils;
use crate::utils::get_env_name;
use crate::vault::Vault;
use rmcp::model::PromptArgument;
use serde_json::json;
use serial_test::serial;
use std::env;
use std::fs::{create_dir_all, remove_dir_all, write};
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use std::{env, mem};
struct TestConfigDirGuard {
key: String,
@@ -5339,14 +5345,14 @@ mod tests {
.unwrap()
.block_on(async {
let join_handle = tokio::spawn(async {
Ok(crate::supervisor::AgentResult {
Ok(AgentResult {
id: "a1".into(),
agent_name: "explore".into(),
output: String::new(),
exit_status: crate::supervisor::AgentExitStatus::Completed,
exit_status: AgentExitStatus::Completed,
})
});
let handle = crate::supervisor::AgentHandle {
let handle = AgentHandle {
id: "a1".to_string(),
agent_name: "explore".to_string(),
depth: 1,
@@ -7696,7 +7702,7 @@ mod tests {
}
}
fn make_running_job(abort_signal: utils::AbortSignal) -> crate::supervisor::JobHandle {
fn make_running_job(abort_signal: utils::AbortSignal) -> JobHandle {
// Leak the runtime so the spawned task is never polled and the job
// stays running for the duration of the test.
let rt = tokio::runtime::Builder::new_current_thread()
@@ -7704,26 +7710,24 @@ mod tests {
.build()
.unwrap();
let join_handle = rt.spawn(async {
Ok(crate::supervisor::JobResult {
Ok(JobResult {
output: serde_json::Value::Null,
exit_code: Some(0),
output_bytes_captured: 0,
})
});
std::mem::forget(rt);
crate::supervisor::JobHandle {
mem::forget(rt);
JobHandle {
id: "j1".to_string(),
tool: "execute_command".to_string(),
started_at: std::time::Instant::now(),
started_at: Instant::now(),
join_handle,
abort_signal,
state: Arc::new(parking_lot::Mutex::new(crate::supervisor::JobState {
status: crate::supervisor::JobStatus::Running,
state: Arc::new(parking_lot::Mutex::new(JobState {
status: JobStatus::Running,
pgid: None,
})),
output_buf: Arc::new(parking_lot::Mutex::new(
crate::function::jobs::RingBuf::default(),
)),
output_buf: Arc::new(parking_lot::Mutex::new(RingBuf::default())),
no_change_checks: 0,
last_check_state: None,
}
src/function/jobs.rs
+56 -8
View File
@@ -70,17 +70,20 @@ impl RingBuf {
if self.capacity == 0 {
return;
}
let src = if bytes.len() > self.capacity {
&bytes[bytes.len() - self.capacity..]
} else {
bytes
};
for &byte in src {
if self.buf.len() < self.capacity {
self.buf.push(byte);
} else {
self.buf[self.write_pos] = byte;
}
self.write_pos = (self.write_pos + 1) % self.capacity;
}
}
@@ -93,6 +96,7 @@ impl RingBuf {
if self.buf.len() < self.capacity {
return self.buf.clone();
}
let mut out = Vec::with_capacity(self.capacity);
out.extend_from_slice(&self.buf[self.write_pos..]);
out.extend_from_slice(&self.buf[..self.write_pos]);
@@ -367,7 +371,8 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
return Ok(json!({
"status": "error",
"message": format!(
"'{tool}' is not enabled in this context — job__start can only background tools declared to you in this request. Use the exact name of a tool from your current catalog."
"'{tool}' is not enabled in this context — job__start can only background tools declared to you in this \
request. Use the exact name of a tool from your current catalog."
),
}));
}
@@ -464,7 +469,9 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
} else {
JobStatus::Failed
};
drop(job_state);
task_notifications.push(job_notification(&notify_id, &notify_tool, success));
result
})
@@ -495,7 +502,8 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
"status": "ok",
"job_id": job_id,
"tool": tool,
"message": "Running in background. Check with job__check, block with job__collect, cancel with job__cancel. You will receive a system_notifications entry on completion. Jobs do not survive coyote exiting.",
"message": "Running in background. Check with job__check, block with job__collect, cancel with job__cancel. \
You will receive a system_notifications entry on completion. Jobs do not survive coyote exiting.",
}))
}
@@ -520,12 +528,14 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result<Value> {
(buf.tail(), buf.total_written())
};
let check_state = (status, total_written);
if job.last_check_state == Some(check_state) {
job.no_change_checks += 1;
} else {
job.no_change_checks = 0;
job.last_check_state = Some(check_state);
}
let tail_truncated = (tail.len() as u64) < total_written;
let mut result = json!({
"status": job_status_str(status),
@@ -536,10 +546,12 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result<Value> {
"output_bytes_captured": total_written,
"tail_truncated": tail_truncated,
});
if matches!(status, JobStatus::Running) {
result["message"] = json!(
"Job is still running. Call job__collect to block for the result, or do other work — you will be notified on completion."
);
if job.no_change_checks >= 3 {
result["hint"] = json!(
"No change across repeated checks — still running; call job__collect to block, or do other work — a system notification will fire on completion."
@@ -606,11 +618,14 @@ async fn handle_collect(ctx: &RequestContext, args: &Value) -> Result<Value> {
let sup = supervisor.read();
sup.job(id).is_none_or(|job| job.join_handle.is_finished())
};
if is_finished {
break;
}
time::sleep(Duration::from_millis(50)).await;
}
break;
}
@@ -642,6 +657,7 @@ async fn handle_collect(ctx: &RequestContext, args: &Value) -> Result<Value> {
if let Some(pgid) = handle.state.lock().pgid {
unsafe { libc::killpg(pgid, libc::SIGKILL) };
}
match time::timeout(JOB_KILL_GRACE, &mut handle.join_handle).await {
Ok(joined) => joined,
Err(_) => {
@@ -694,6 +710,7 @@ async fn handle_collect(ctx: &RequestContext, args: &Value) -> Result<Value> {
"output_tail": output_tail,
"output_bytes_captured": job_result.output_bytes_captured,
});
if let Some(exit_code) = job_result.exit_code {
response["exit_code"] = json!(exit_code);
}
@@ -968,6 +985,7 @@ async fn run_process_job(
"Tool call '{}' timed out after {}s and was killed (set COYOTE_TOOL_TIMEOUT to adjust; 0 = unlimited)",
snapshot.display_name, snapshot.timeout_secs
);
return Ok(JobResult {
output: json!({"tool_call_error": message}),
exit_code: None,
@@ -1011,6 +1029,7 @@ async fn run_process_job(
{
error_json["output"] = json!(contents);
}
return Ok(JobResult {
output: error_json,
exit_code,
@@ -1075,6 +1094,7 @@ async fn run_mcp_job(
}
};
let output = render_tool_result(serde_json::to_value(raw)?, &server)?;
Ok(JobResult {
output,
exit_code: None,
@@ -1105,6 +1125,7 @@ fn cap_result(output: Value, tail_lines: Option<usize>) -> (Value, bool) {
text = capped;
truncated = true;
}
if truncated {
(json!(text), true)
} else {
@@ -1117,11 +1138,13 @@ fn tail_chars(text: &str, max_chars: usize) -> Option<String> {
if total <= max_chars {
return None;
}
let cut = text
.char_indices()
.nth(total - max_chars)
.map(|(i, _)| i)
.unwrap_or(0);
Some(format!(
"[truncated: kept last {max_chars} of {total} chars]\n{}",
&text[cut..]
@@ -1133,11 +1156,12 @@ mod tests {
use super::*;
use crate::config::{AppConfig, AppState, WorkingMode};
use crate::function::supervisor::{
GuardrailAction, check_pending_agents_guardrail, handle_supervisor_tool,
GuardrailAction, check_pending_tasks_guardrail, handle_supervisor_tool,
};
use crate::supervisor::mailbox::Inbox;
use crate::supervisor::{AgentExitStatus, AgentHandle, AgentResult};
use std::future::Future;
use std::mem;
fn default_app_state() -> Arc<AppState> {
Arc::new(AppState::test_default())
@@ -1175,7 +1199,7 @@ mod tests {
output_bytes_captured: 0,
})
});
std::mem::forget(rt);
mem::forget(rt);
JobHandle {
id: id.to_string(),
tool: "execute_command".to_string(),
@@ -1219,8 +1243,10 @@ mod tests {
#[test]
fn ring_buf_returns_contents_below_capacity() {
let mut buf = RingBuf::new(8);
buf.push(b"abc");
buf.push(b"de");
assert_eq!(buf.tail(), b"abcde");
assert_eq!(buf.total_written(), 5);
}
@@ -1228,7 +1254,9 @@ mod tests {
#[test]
fn ring_buf_exact_fit_keeps_everything() {
let mut buf = RingBuf::new(5);
buf.push(b"abcde");
assert_eq!(buf.tail(), b"abcde");
assert_eq!(buf.total_written(), 5);
}
@@ -1236,8 +1264,10 @@ mod tests {
#[test]
fn ring_buf_wrap_around_keeps_newest_bytes() {
let mut buf = RingBuf::new(5);
buf.push(b"abcde");
buf.push(b"fg");
assert_eq!(buf.tail(), b"cdefg");
assert_eq!(buf.total_written(), 7);
}
@@ -1245,7 +1275,9 @@ mod tests {
#[test]
fn ring_buf_oversize_push_keeps_last_capacity_bytes() {
let mut buf = RingBuf::new(4);
buf.push(b"abcdefghij");
assert_eq!(buf.tail(), b"ghij");
assert_eq!(buf.total_written(), 10);
}
@@ -1254,7 +1286,9 @@ mod tests {
fn ring_buf_default_capacity_is_64_kib() {
let mut buf = RingBuf::default();
let payload = vec![b'x'; 64 * 1024 + 1];
buf.push(&payload);
assert_eq!(buf.tail().len(), 64 * 1024);
assert_eq!(buf.total_written(), 64 * 1024 + 1);
}
@@ -1280,7 +1314,7 @@ mod tests {
exit_status: AgentExitStatus::Completed,
})
});
std::mem::forget(rt);
mem::forget(rt);
let handle = AgentHandle {
id: "a1".to_string(),
agent_name: "explore".to_string(),
@@ -1304,6 +1338,7 @@ mod tests {
.into_iter()
.map(|d| d.name)
.collect();
assert_eq!(
names,
vec![
@@ -1320,6 +1355,7 @@ mod tests {
fn whitelist_rejects_state_mutating_tools() {
for tool in ["memory__write", "todo__add", "skill__load", "rag__query"] {
let rejection = whitelist_rejection(tool).unwrap();
let message = rejection["message"].as_str().unwrap();
assert!(
message.contains("mutates agent/session state"),
@@ -1335,6 +1371,7 @@ mod tests {
.as_str()
.unwrap()
.to_string();
assert!(
message.contains("already asynchronous"),
"unexpected message for {tool}: {message}"
@@ -1359,6 +1396,7 @@ mod tests {
.as_str()
.unwrap()
.to_string();
assert!(
message.contains("sub-second"),
"unexpected message for {tool}: {message}"
@@ -1373,6 +1411,7 @@ mod tests {
.as_str()
.unwrap()
.to_string();
assert!(
message.contains("is fast"),
"unexpected message for {tool}: {message}"
@@ -1550,7 +1589,9 @@ mod tests {
#[test]
fn handle_check_unknown_id_teaches_job_list() {
let ctx = ctx_with_job_supervisor(4);
let result = handle_check(&ctx, &json!({"id": "job_x"})).unwrap();
assert_eq!(result["status"], "error");
assert!(
result["message"]
@@ -1708,7 +1749,9 @@ mod tests {
#[test]
fn job_handlers_miss_without_supervisor() {
let ctx = plain_ctx();
let result = handle_check(&ctx, &json!({"id": "job_x"})).unwrap();
assert_eq!(result["status"], "error");
assert!(
result["message"]
@@ -2049,7 +2092,9 @@ mod tests {
#[test]
fn handle_list_without_supervisor_reports_empty() {
let ctx = plain_ctx();
let result = handle_list(&ctx).unwrap();
assert_eq!(result["active_jobs"], 0);
assert_eq!(result["max_concurrent_jobs"], 5);
assert_eq!(result["jobs"].as_array().unwrap().len(), 0);
@@ -2058,6 +2103,7 @@ mod tests {
#[test]
fn cap_result_normalizes_null_to_done() {
let (value, truncated) = cap_result(Value::Null, None);
assert_eq!(value, json!("DONE"));
assert!(!truncated);
}
@@ -2065,6 +2111,7 @@ mod tests {
#[test]
fn cap_result_preserves_small_values() {
let (value, truncated) = cap_result(json!({"a": 1}), None);
assert_eq!(value, json!({"a": 1}));
assert!(!truncated);
}
@@ -2085,6 +2132,7 @@ mod tests {
#[test]
fn tail_chars_floors_to_char_boundary() {
let capped = tail_chars("aébc", 2).unwrap();
assert!(capped.ends_with("bc"));
assert!(capped.starts_with("[truncated: kept last 2 of 4 chars]"));
assert!(tail_chars("abc", 3).is_none());
@@ -2321,18 +2369,18 @@ mod tests {
.register(make_running_job("j1"))
.unwrap();
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => {
assert!(prompt.contains("j1"));
assert!(prompt.contains("job__collect"));
}
_ => panic!("expected Inject for a running job"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 1);
assert_eq!(ctx.pending_tasks_guardrail_count, 1);
let empty_ctx = &mut ctx_with_job_supervisor(5);
assert!(matches!(
check_pending_agents_guardrail(empty_ctx),
check_pending_tasks_guardrail(empty_ctx),
GuardrailAction::NoAction
));
}
src/function/mod.rs
+48 -22
View File
@@ -378,6 +378,7 @@ fn drain_live_notifications(ctx: &RequestContext) -> Vec<Value> {
if events.is_empty() {
return vec![];
}
let Some(supervisor) = ctx.supervisor.as_ref() else {
return vec![];
};
@@ -399,6 +400,7 @@ fn merge_system_channel(last: &mut ToolResult, escalations: Vec<Value>, notifica
if escalations.is_empty() && notifications.is_empty() {
return;
}
let escalation_instruction = "Child agents are BLOCKED waiting for your reply. \
Call agent__reply_escalation for each pending escalation to unblock them.";
let notification_instruction =
@@ -416,6 +418,7 @@ fn merge_system_channel(last: &mut ToolResult, escalations: Vec<Value>, notifica
}
}
};
if !escalations.is_empty() {
map.insert("pending_escalations".into(), json!(escalations));
map.insert(
@@ -423,6 +426,7 @@ fn merge_system_channel(last: &mut ToolResult, escalations: Vec<Value>, notifica
json!(escalation_instruction),
);
}
if !notifications.is_empty() {
map.insert("system_notifications".into(), json!(notifications));
map.insert(
@@ -468,6 +472,7 @@ impl ToolResult {
while !s.is_char_boundary(cut) {
cut -= 1;
}
let prefix = &s[..cut];
self.output = json!(format!(
"[truncated: tool output exceeded {max_chars} chars]\n{prefix}"
@@ -2446,6 +2451,7 @@ impl ToolCallTracker {
if is_loop_tracker_exempt(&new_call.name) {
return None;
}
if self.last_calls.len() < self.max_repeats {
return None;
}
@@ -2515,6 +2521,7 @@ impl ToolCallTracker {
if is_loop_tracker_exempt(&call.name) {
return;
}
if self.last_calls.len() >= self.chain_len * self.max_repeats {
self.last_calls.pop_front();
}
@@ -2572,14 +2579,20 @@ mod tests {
};
use crate::config::{Agent, AgentConfig, AppConfig, AppState, WorkingMode};
use crate::supervisor::escalation::{EscalationQueue, EscalationRequest};
use crate::supervisor::mailbox::Inbox;
use crate::supervisor::notification::{agent_notification, job_notification};
use crate::supervisor::{
AgentExitStatus, AgentHandle, AgentResult, JobHandle, JobResult, JobState, JobStatus,
Supervisor,
};
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use jobs::RingBuf;
use rmcp::model::{CallToolResult, ContentBlock};
use serde_json::json;
use serial_test::serial;
use std::process;
use std::sync::Arc;
use std::{mem, process};
fn call(name: &str, id: Option<&str>) -> ToolCall {
ToolCall::new(name.to_string(), json!({}), id.map(|s| s.to_string()))
@@ -2663,28 +2676,28 @@ mod tests {
.build()
.unwrap();
let join_handle = rt.spawn(async {
Ok(crate::supervisor::JobResult {
Ok(JobResult {
output: Value::Null,
exit_code: Some(0),
output_bytes_captured: 0,
})
});
std::mem::forget(rt);
let handle = crate::supervisor::JobHandle {
mem::forget(rt);
let handle = JobHandle {
id: id.to_string(),
tool: "execute_command".to_string(),
started_at: std::time::Instant::now(),
started_at: Instant::now(),
join_handle,
abort_signal: crate::utils::create_abort_signal(),
state: Arc::new(parking_lot::Mutex::new(crate::supervisor::JobState {
status: crate::supervisor::JobStatus::Completed,
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(jobs::RingBuf::default())),
output_buf: Arc::new(parking_lot::Mutex::new(RingBuf::default())),
no_change_checks: 0,
last_check_state: None,
};
let mut sup = crate::supervisor::Supervisor::new(0, 3).with_max_concurrent_jobs(4);
let mut sup = Supervisor::new(0, 3).with_max_concurrent_jobs(4);
sup.register(handle).unwrap();
let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
ctx.supervisor = Some(Arc::new(parking_lot::RwLock::new(sup)));
@@ -2724,7 +2737,9 @@ mod tests {
#[test]
fn merge_system_channel_adds_notifications_without_escalation_keys() {
let mut result = ToolResult::new(call("t", Some("id-1")), json!({"status": "ok"}));
merge_system_channel(&mut result, vec![], vec![json!({"id": "job_1"})]);
assert_eq!(result.output["status"], "ok");
assert_eq!(
result.output["system_notifications"],
@@ -2743,11 +2758,13 @@ mod tests {
#[test]
fn merge_system_channel_wraps_non_object_once_with_both_channels() {
let mut result = ToolResult::new(call("t", Some("id-1")), json!("DONE"));
merge_system_channel(
&mut result,
vec![json!({"escalation_id": "esc_1"})],
vec![json!({"id": "job_1"})],
);
assert_eq!(result.output["output"], json!("DONE"));
assert_eq!(
result.output["pending_escalations"][0]["escalation_id"],
@@ -2806,24 +2823,24 @@ mod tests {
.unwrap();
let agent_id = id.to_string();
let join_handle = rt.spawn(async move {
Ok(crate::supervisor::AgentResult {
Ok(AgentResult {
id: agent_id,
agent_name: "explore".into(),
output: String::new(),
exit_status: crate::supervisor::AgentExitStatus::Completed,
exit_status: AgentExitStatus::Completed,
})
});
std::mem::forget(rt);
let handle = crate::supervisor::AgentHandle {
mem::forget(rt);
let handle = AgentHandle {
id: id.to_string(),
agent_name: "explore".to_string(),
depth: 1,
inbox: Arc::new(crate::supervisor::mailbox::Inbox::new()),
abort_signal: crate::utils::create_abort_signal(),
inbox: Arc::new(Inbox::new()),
abort_signal: create_abort_signal(),
join_handle,
child_supervisor: None,
};
let mut sup = crate::supervisor::Supervisor::new(4, 3);
let mut sup = Supervisor::new(4, 3);
sup.register(handle).unwrap();
let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
ctx.supervisor = Some(Arc::new(parking_lot::RwLock::new(sup)));
@@ -2958,7 +2975,7 @@ mod tests {
let job_id = started["job_id"].as_str().unwrap().to_string();
let supervisor = ctx.supervisor.clone().unwrap();
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
while tokio::time::Instant::now() < deadline {
let finished = supervisor
.read()
@@ -2967,7 +2984,7 @@ mod tests {
if finished {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
tokio::time::sleep(Duration::from_millis(10)).await;
}
let calls = vec![call("unknown_tool", Some("id-2"))];
@@ -3172,9 +3189,8 @@ mod tests {
#[test]
fn loop_tracker_exempt_list_is_exactly_the_polling_tools() {
let actual: std::collections::HashSet<&str> =
LOOP_TRACKER_EXEMPT_TOOLS.iter().copied().collect();
let expected: std::collections::HashSet<&str> = [
let actual: HashSet<&str> = LOOP_TRACKER_EXEMPT_TOOLS.iter().copied().collect();
let expected: HashSet<&str> = [
"job__check",
"job__list",
"agent__check",
@@ -3210,10 +3226,12 @@ mod tests {
fn tracker_exempt_interleave_does_not_mask_real_loop() {
let mut tracker = ToolCallTracker::default();
let x = call_with_args("execute_command", json!({"command": "ls"}));
tracker.record_call(call_with_args("job__check", json!({"id": "j1"})));
tracker.record_call(x.clone());
tracker.record_call(call_with_args("job__check", json!({"id": "j1"})));
tracker.record_call(x.clone());
assert!(tracker.check_loop(&x).is_some());
}
@@ -3221,8 +3239,10 @@ mod tests {
fn tracker_non_exempt_behavior_unchanged() {
let mut tracker = ToolCallTracker::default();
let c = call_with_args("fs_cat", json!({"path": "a.txt"}));
tracker.record_call(c.clone());
tracker.record_call(c.clone());
assert!(tracker.check_loop(&c).is_some());
}
@@ -3300,7 +3320,9 @@ mod tests {
#[test]
fn functions_append_job_adds_declarations() {
let mut f = Functions::default();
f.append_job_functions();
assert!(f.contains("job__start"));
assert!(f.contains("job__check"));
assert!(f.contains("job__collect"));
@@ -3313,7 +3335,9 @@ mod tests {
let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
ctx.tool_scope.functions.append_job_functions();
let calls = vec![call("job__list", Some("id-1"))];
let results = run_async(eval_tool_calls(&mut ctx, calls)).unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].output["active_jobs"], 0);
assert_eq!(results[0].output["jobs"], json!([]));
@@ -3323,7 +3347,9 @@ mod tests {
fn eval_soft_fails_job_calls_when_jobs_not_declared() {
let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
let calls = vec![call("job__start", Some("id-1"))];
let results = run_async(eval_tool_calls(&mut ctx, calls)).unwrap();
let err = results[0].output["tool_call_error"].as_str().unwrap();
assert!(err.contains("Unexpected call"));
}
src/function/supervisor.rs
+68 -54
View File
@@ -25,7 +25,7 @@ use uuid::Uuid;
pub const SUPERVISOR_FUNCTION_PREFIX: &str = "agent__";
pub const PENDING_AGENTS_GUARDRAIL_MAX: u32 = 3;
pub const PENDING_TASKS_GUARDRAIL_MAX: u32 = 3;
fn agent_permitted(whitelist: Option<&[String]>, target: &str) -> bool {
match whitelist {
@@ -73,11 +73,12 @@ pub fn pending_tasks(ctx: &RequestContext) -> Vec<PendingTask> {
finished,
})
.collect();
tasks.sort_by(|a, b| a.id.cmp(&b.id));
tasks
}
pub fn build_pending_agents_guardrail_prompt(tasks: &[PendingTask]) -> String {
pub fn build_pending_tasks_guardrail_prompt(tasks: &[PendingTask]) -> String {
let running: Vec<&PendingTask> = tasks.iter().filter(|t| !t.finished).collect();
let finished: Vec<&PendingTask> = tasks.iter().filter(|t| t.finished).collect();
@@ -105,6 +106,7 @@ pub fn build_pending_agents_guardrail_prompt(tasks: &[PendingTask]) -> String {
count = running.len()
));
}
if !finished.is_empty() {
let cmd_list = finished
.iter()
@@ -124,6 +126,7 @@ pub fn build_pending_agents_guardrail_prompt(tasks: &[PendingTask]) -> String {
count = finished.len()
));
}
format!(
"[SYSTEM GUARDRAIL] You attempted to end your turn with {count} unreclaimed background \
task(s).\n\n{body}",
@@ -132,14 +135,14 @@ pub fn build_pending_agents_guardrail_prompt(tasks: &[PendingTask]) -> String {
)
}
pub fn check_pending_agents_guardrail(ctx: &mut RequestContext) -> GuardrailAction {
pub fn check_pending_tasks_guardrail(ctx: &mut RequestContext) -> GuardrailAction {
let pending = pending_tasks(ctx);
if pending.is_empty() {
ctx.pending_agents_guardrail_count = 0;
ctx.pending_tasks_guardrail_count = 0;
return GuardrailAction::NoAction;
}
if ctx.pending_agents_guardrail_count >= PENDING_AGENTS_GUARDRAIL_MAX {
if ctx.pending_tasks_guardrail_count >= PENDING_TASKS_GUARDRAIL_MAX {
if let Some(sup) = ctx.supervisor.as_ref().cloned() {
sup.read().cancel_recursive();
let finished: Vec<&PendingTask> = pending.iter().filter(|t| t.finished).collect();
@@ -162,13 +165,13 @@ pub fn check_pending_agents_guardrail(ctx: &mut RequestContext) -> GuardrailActi
}
}
}
ctx.pending_agents_guardrail_count = 0;
ctx.pending_tasks_guardrail_count = 0;
return GuardrailAction::ForceTerminate(pending.into_iter().map(|t| t.id).collect());
}
ctx.pending_agents_guardrail_count += 1;
let mut prompt = build_pending_agents_guardrail_prompt(&pending);
ctx.pending_tasks_guardrail_count += 1;
let mut prompt = build_pending_tasks_guardrail_prompt(&pending);
if let Some(queue) = ctx.root_escalation_queue()
&& queue.has_pending()
{
@@ -184,7 +187,8 @@ pub fn check_pending_agents_guardrail(ctx: &mut RequestContext) -> GuardrailActi
pub fn escalation_function_declarations() -> Vec<FunctionDeclaration> {
vec![FunctionDeclaration {
name: format!("{SUPERVISOR_FUNCTION_PREFIX}reply_escalation"),
description: "Reply to a pending escalation from a child agent. The child is blocked waiting for this reply. Use this after seeing pending_escalations notifications.".to_string(),
description: "Reply to a pending escalation from a child agent. The child is blocked waiting for this reply. \
Use this after seeing pending_escalations notifications.".to_string(),
parameters: JsonSchema {
type_value: Some("object".to_string()),
properties: Some(IndexMap::from([
@@ -200,7 +204,8 @@ pub fn escalation_function_declarations() -> Vec<FunctionDeclaration> {
"reply".to_string(),
JsonSchema {
type_value: Some("string".to_string()),
description: Some("Your answer to the child agent's question. For ask/confirm questions, use the exact option text. For input questions, provide the text response.".into()),
description: Some("Your answer to the child agent's question. For ask/confirm questions, use \
the exact option text. For input questions, provide the text response.".into()),
..Default::default()
},
),
@@ -256,7 +261,8 @@ pub fn supervisor_function_declarations() -> Vec<FunctionDeclaration> {
},
FunctionDeclaration {
name: format!("{SUPERVISOR_FUNCTION_PREFIX}check"),
description: "Non-blocking status probe: reports whether a spawned agent is still running or finished. NEVER returns or consumes the result — when finished, call agent__collect to retrieve it.".to_string(),
description: "Non-blocking status probe: reports whether a spawned agent is still running or finished. \
NEVER returns or consumes the result — when finished, call agent__collect to retrieve it.".to_string(),
parameters: JsonSchema {
type_value: Some("object".to_string()),
properties: Some(IndexMap::from([(
@@ -558,10 +564,10 @@ pub fn run_child_agent(
}
if tool_results.is_empty() {
match check_pending_agents_guardrail(&mut child_ctx) {
match check_pending_tasks_guardrail(&mut child_ctx) {
GuardrailAction::NoAction => break,
GuardrailAction::ForceTerminate(ids) => {
log::warn!(
warn!(
"Pending-agent guardrail force-cancelled {} agent(s) after max reminders: {:?}",
ids.len(),
ids
@@ -642,7 +648,7 @@ pub async fn run_agent_for_graph(
let session = agent.agent_session().map(|v| v.to_string());
let child_jobs_enabled = jobs_enabled(Some(&agent), app_config.as_ref());
let should_init_supervisor = agent.can_spawn_agents() || child_jobs_enabled;
let agent_max_concurrent = if agent.can_spawn_agents() {
let agent_max_concurrent_subagents = if agent.can_spawn_agents() {
agent.max_concurrent_agents()
} else {
0
@@ -661,7 +667,7 @@ pub async fn run_agent_for_graph(
child_ctx.agent = Some(agent);
if should_init_supervisor {
child_ctx.supervisor = Some(Arc::new(RwLock::new(
Supervisor::new(agent_max_concurrent, agent_max_depth)
Supervisor::new(agent_max_concurrent_subagents, agent_max_depth)
.with_max_concurrent_jobs(agent_max_jobs),
)));
}
@@ -827,7 +833,7 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
let session = agent.agent_session().map(|v| v.to_string());
let child_jobs_enabled = jobs_enabled(Some(&agent), app_config.as_ref());
let should_init_supervisor = agent.can_spawn_agents() || child_jobs_enabled;
let max_concurrent = if agent.can_spawn_agents() {
let max_concurrent_agents = if agent.can_spawn_agents() {
agent.max_concurrent_agents()
} else {
0
@@ -845,7 +851,7 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
child_ctx.agent = Some(agent);
if should_init_supervisor {
child_ctx.supervisor = Some(Arc::new(RwLock::new(
Supervisor::new(max_concurrent, max_depth).with_max_concurrent_jobs(max_jobs),
Supervisor::new(max_concurrent_agents, max_depth).with_max_concurrent_jobs(max_jobs),
)));
}
@@ -893,6 +899,7 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
&agent_result.agent_name,
success,
));
Ok(agent_result)
});
@@ -975,6 +982,7 @@ async fn handle_check(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
if is_job_task(ctx.supervisor.as_ref(), id) {
return Ok(job_id_teaching_error(id));
}
Ok(json!({
"status": "error",
"message": format!("No agent found with id '{id}'")
@@ -1001,6 +1009,7 @@ async fn handle_collect(ctx: &mut RequestContext, args: &Value) -> Result<Value>
if id.starts_with("job_") || sup.has_job(id) {
return Ok(job_id_teaching_error(id));
}
return Ok(json!({
"status": "error",
"message": format!("Agent '{id}' not found. Use agent__check to verify it exists and is finished.")
@@ -1068,7 +1077,7 @@ async fn handle_collect(ctx: &mut RequestContext, args: &Value) -> Result<Value>
.map_err(|e| anyhow!("Agent failed: {e}"))?;
let output = summarize_output(ctx, &result.agent_name, &result.output).await?;
ctx.pending_agents_guardrail_count = 0;
ctx.pending_tasks_guardrail_count = 0;
Ok(json!({
"status": "completed",
@@ -1169,7 +1178,7 @@ async fn handle_cancel(ctx: &mut RequestContext, args: &Value) -> Result<Value>
let cleanup = tokio::time::timeout(Duration::from_secs(5), handle.join_handle).await;
ctx.pending_agents_guardrail_count = 0;
ctx.pending_tasks_guardrail_count = 0;
let message = match cleanup {
Ok(_) => format!("Cancelled agent '{agent_name}' and waited for cleanup."),
@@ -1187,6 +1196,7 @@ async fn handle_cancel(ctx: &mut RequestContext, args: &Value) -> Result<Value>
if is_job_task(ctx.supervisor.as_ref(), id) {
return Ok(job_id_teaching_error(id));
}
Ok(json!({
"status": "error",
"message": format!("No agent found with id '{id}'"),
@@ -1244,6 +1254,7 @@ fn handle_send_message(ctx: &mut RequestContext, args: &Value) -> Result<Value>
{
return Ok(job_id_teaching_error(id));
}
Ok(json!({
"status": "error",
"message": format!("No agent found with id '{id}'. Agent may not exist or may have already completed."),
@@ -1591,6 +1602,7 @@ mod tests {
use parking_lot::Mutex;
use serde_json::json;
use serial_test::serial;
use std::mem;
fn default_app_state() -> Arc<AppState> {
Arc::new(AppState::test_default())
@@ -1622,7 +1634,7 @@ mod tests {
output_bytes_captured: 0,
})
});
std::mem::forget(rt);
mem::forget(rt);
JobHandle {
id: id.to_string(),
tool: "execute_command".to_string(),
@@ -1681,7 +1693,7 @@ mod tests {
exit_status: AgentExitStatus::Completed,
})
});
std::mem::forget(rt);
mem::forget(rt);
let handle = AgentHandle {
id: id.to_string(),
@@ -2314,7 +2326,7 @@ mod tests {
reply_tx: tx,
});
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => {
assert!(prompt.contains("agent__reply_escalation"));
assert!(prompt.contains("esc_9"));
@@ -2358,7 +2370,7 @@ mod tests {
.register(handle)
.unwrap();
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => {
assert!(!prompt.contains("agent__reply_escalation"));
}
@@ -2371,7 +2383,7 @@ mod tests {
fn handle_collect_finished_agent_returns_output_and_consumes_handle() {
let mut ctx = ctx_with_supervisor(4, 3);
register_fake_agent(&mut ctx, "a1", "explore");
ctx.pending_agents_guardrail_count = 2;
ctx.pending_tasks_guardrail_count = 2;
let result = run_async(handle_collect(&mut ctx, &json!({"id": "a1"}))).unwrap();
@@ -2380,7 +2392,7 @@ mod tests {
assert_eq!(result["agent"], "explore");
assert_eq!(result["exit_status"], "Completed");
assert_eq!(result["output"], "fake output");
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
assert_eq!(
ctx.supervisor.as_ref().unwrap().read().is_finished("a1"),
None
@@ -2424,7 +2436,9 @@ mod tests {
#[test]
fn handle_collect_unknown_agent_errors() {
let mut ctx = ctx_with_supervisor(4, 3);
let result = run_async(handle_collect(&mut ctx, &json!({"id": "missing"}))).unwrap();
assert_eq!(result["status"], "error");
assert!(result["message"].as_str().unwrap().contains("not found"));
}
@@ -2473,13 +2487,13 @@ mod tests {
#[test]
fn guardrail_no_supervisor_is_no_action_and_resets_counter() {
let mut ctx = RequestContext::new(default_app_state(), WorkingMode::Cmd);
ctx.pending_agents_guardrail_count = 2;
ctx.pending_tasks_guardrail_count = 2;
assert!(matches!(
check_pending_agents_guardrail(&mut ctx),
check_pending_tasks_guardrail(&mut ctx),
GuardrailAction::NoAction
));
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
}
/// A finished-but-uncollected agent counts as pending: the turn-end
@@ -2490,9 +2504,9 @@ mod tests {
let mut ctx = ctx_with_supervisor(4, 3);
register_fake_agent(&mut ctx, "a1", "explore");
wait_until_finished(&ctx, "a1");
ctx.pending_agents_guardrail_count = 2;
ctx.pending_tasks_guardrail_count = 2;
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => {
assert!(prompt.contains("a1"));
assert!(prompt.contains("agent__collect --id a1"));
@@ -2500,7 +2514,7 @@ mod tests {
}
_ => panic!("expected Inject action"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 3);
assert_eq!(ctx.pending_tasks_guardrail_count, 3);
assert_eq!(
ctx.supervisor.as_ref().unwrap().read().is_finished("a1"),
Some(true)
@@ -2512,15 +2526,15 @@ mod tests {
let mut ctx = ctx_with_supervisor(4, 3);
register_fake_agent(&mut ctx, "a1", "explore");
wait_until_finished(&ctx, "a1");
ctx.pending_agents_guardrail_count = PENDING_AGENTS_GUARDRAIL_MAX;
ctx.pending_tasks_guardrail_count = PENDING_TASKS_GUARDRAIL_MAX;
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::ForceTerminate(ids) => {
assert_eq!(ids, vec!["a1".to_string()]);
}
_ => panic!("expected ForceTerminate action"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
assert_eq!(
ctx.supervisor.as_ref().unwrap().read().is_finished("a1"),
None
@@ -2540,7 +2554,7 @@ mod tests {
register_fake_agent(&mut ctx, "a1", "explore");
wait_until_finished(&ctx, "a1");
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => {
assert!(prompt.contains("Still running"));
assert!(prompt.contains("slow (agent)"));
@@ -2567,7 +2581,7 @@ mod tests {
},
];
let prompt = build_pending_agents_guardrail_prompt(&tasks);
let prompt = build_pending_tasks_guardrail_prompt(&tasks);
assert!(prompt.contains("job_1 (job)"));
assert!(prompt.contains("job__cancel"));
@@ -2596,15 +2610,15 @@ mod tests {
rt.block_on(async {
let mut ctx = ctx_with_supervisor(4, 3);
let abort = register_running_agent(&mut ctx, "slow", "test");
ctx.pending_agents_guardrail_count = PENDING_AGENTS_GUARDRAIL_MAX;
ctx.pending_tasks_guardrail_count = PENDING_TASKS_GUARDRAIL_MAX;
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::ForceTerminate(ids) => {
assert_eq!(ids, vec!["slow".to_string()]);
}
_ => panic!("expected ForceTerminate action"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
assert!(abort.aborted());
});
}
@@ -2619,16 +2633,16 @@ mod tests {
rt.block_on(async {
let mut ctx = ctx_with_supervisor(4, 3);
let _abort = register_running_agent(&mut ctx, "slow", "test");
ctx.pending_agents_guardrail_count = 1;
ctx.pending_tasks_guardrail_count = 1;
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => {
assert!(prompt.contains("slow"));
assert!(prompt.contains("agent__collect"));
}
_ => panic!("expected Inject action"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 2);
assert_eq!(ctx.pending_tasks_guardrail_count, 2);
});
}
@@ -2636,12 +2650,12 @@ mod tests {
fn handle_cancel_resets_guardrail_counter() {
let mut ctx = ctx_with_supervisor(4, 3);
register_fake_agent(&mut ctx, "a1", "explore");
ctx.pending_agents_guardrail_count = 2;
ctx.pending_tasks_guardrail_count = 2;
let result = run_async(handle_cancel(&mut ctx, &json!({"id": "a1"}))).unwrap();
assert_eq!(result["status"], "ok");
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
}
#[test]
@@ -2798,7 +2812,7 @@ mod tests {
.write()
.register(handle)
.unwrap();
ctx.pending_agents_guardrail_count = 2;
ctx.pending_tasks_guardrail_count = 2;
let result = handle_cancel(&mut ctx, &json!({"id": "a1"})).await.unwrap();
@@ -2811,7 +2825,7 @@ mod tests {
ctx.supervisor.as_ref().unwrap().read().is_finished("a1"),
None
);
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
});
}
@@ -2951,21 +2965,21 @@ mod tests {
.register(handle)
.unwrap();
for expected_count in 1..=PENDING_AGENTS_GUARDRAIL_MAX {
match check_pending_agents_guardrail(&mut ctx) {
for expected_count in 1..=PENDING_TASKS_GUARDRAIL_MAX {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::Inject(prompt) => assert!(prompt.contains("job_1")),
_ => panic!("expected Inject below max"),
}
assert_eq!(ctx.pending_agents_guardrail_count, expected_count);
assert_eq!(ctx.pending_tasks_guardrail_count, expected_count);
}
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::ForceTerminate(ids) => {
assert_eq!(ids, vec!["job_1".to_string()]);
}
_ => panic!("expected ForceTerminate at max"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
assert!(abort.aborted());
});
}
@@ -2982,15 +2996,15 @@ mod tests {
);
std::thread::sleep(Duration::from_millis(10));
}
ctx.pending_agents_guardrail_count = PENDING_AGENTS_GUARDRAIL_MAX;
ctx.pending_tasks_guardrail_count = PENDING_TASKS_GUARDRAIL_MAX;
match check_pending_agents_guardrail(&mut ctx) {
match check_pending_tasks_guardrail(&mut ctx) {
GuardrailAction::ForceTerminate(ids) => {
assert_eq!(ids, vec!["job_1".to_string()]);
}
_ => panic!("expected ForceTerminate action"),
}
assert_eq!(ctx.pending_agents_guardrail_count, 0);
assert_eq!(ctx.pending_tasks_guardrail_count, 0);
assert!(!ctx.supervisor.as_ref().unwrap().read().has_job("job_1"));
}
}
src/graph/executor.rs
+15 -16
View File
@@ -563,8 +563,10 @@ mod tests {
mod integration_tests {
use super::*;
use crate::config::{AppState, WorkingMode};
use crate::function::jobs::RingBuf;
use crate::supervisor::{JobHandle, JobResult, JobState, JobStatus, Supervisor, notification};
use crate::utils::{create_abort_signal, temp_file};
use std::fs;
use std::{fs, mem};
fn cmd_available(name: &str) -> bool {
which::which(name).is_ok()
@@ -887,40 +889,37 @@ nodes:
.build()
.unwrap();
let join_handle = rt.spawn(async {
Ok(crate::supervisor::JobResult {
Ok(JobResult {
output: Value::Null,
exit_code: Some(0),
output_bytes_captured: 0,
})
});
std::mem::forget(rt);
let handle = crate::supervisor::JobHandle {
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(crate::supervisor::JobState {
status: crate::supervisor::JobStatus::Completed,
state: Arc::new(parking_lot::Mutex::new(JobState {
status: JobStatus::Completed,
pgid: None,
})),
output_buf: Arc::new(parking_lot::Mutex::new(
crate::function::jobs::RingBuf::default(),
)),
output_buf: Arc::new(parking_lot::Mutex::new(RingBuf::default())),
no_change_checks: 0,
last_check_state: None,
};
let mut sup = crate::supervisor::Supervisor::new(0, 3).with_max_concurrent_jobs(4);
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(crate::supervisor::notification::job_notification(
"job_bg",
"execute_command",
true,
));
ctx.notification_queue.push(notification::job_notification(
"job_bg",
"execute_command",
true,
));
let abort = create_abort_signal();
let result = GraphExecutor::new(graph, &ws.dir)
src/graph/llm.rs
+2 -2
View File
@@ -7,7 +7,7 @@ use crate::config::{
Input, RequestContext, Role, RoleLike, SkillPolicy, should_inject_skill_instructions,
};
use crate::function::skill::skill_function_declarations;
use crate::function::supervisor::{GuardrailAction, check_pending_agents_guardrail};
use crate::function::supervisor::{GuardrailAction, check_pending_tasks_guardrail};
use crate::utils::create_abort_signal;
use anyhow::{Context, Error, Result, anyhow, bail};
use log::warn;
@@ -268,7 +268,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/main.rs
+2 -2
View File
@@ -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::supervisor::{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
+3 -3
View File
@@ -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::supervisor::{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
+10 -1
View File
@@ -352,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 {
@@ -386,7 +387,7 @@ mod tests {
output_bytes_captured: 0,
})
});
std::mem::forget(rt);
mem::forget(rt);
JobHandle {
id: id.to_string(),
tool: "execute_command".to_string(),
@@ -545,7 +546,9 @@ mod tests {
#[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"));
}
@@ -554,7 +557,9 @@ mod tests {
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"));
}
@@ -562,8 +567,10 @@ mod tests {
#[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);
@@ -572,6 +579,7 @@ mod tests {
#[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);
@@ -589,6 +597,7 @@ mod tests {
#[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();