feat(supervisor): push agent completion notifications to the spawning context
The spawned-agent task now pushes an agent_completed/agent_failed event into the spawning context's notification queue before returning, so a parent that keeps working learns mid-turn that a child finished instead of discovering it only at the turn-end guardrail. Cancelled or already-collected agents are suppressed by the existing drain-time registration filter. This delivery applies regardless of whether background jobs are enabled.
This commit is contained in:
+63
-1
@@ -2553,7 +2553,7 @@ mod tests {
|
||||
};
|
||||
use crate::config::{Agent, AgentConfig, AppConfig, AppState, WorkingMode};
|
||||
use crate::supervisor::escalation::{EscalationQueue, EscalationRequest};
|
||||
use crate::supervisor::notification::job_notification;
|
||||
use crate::supervisor::notification::{agent_notification, job_notification};
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use rmcp::model::{CallToolResult, ContentBlock};
|
||||
@@ -2779,6 +2779,68 @@ mod tests {
|
||||
assert!(drain_live_notifications(&ctx).is_empty());
|
||||
}
|
||||
|
||||
fn ctx_with_registered_agent(id: &str) -> RequestContext {
|
||||
let rt = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.unwrap();
|
||||
let agent_id = id.to_string();
|
||||
let join_handle = rt.spawn(async move {
|
||||
Ok(crate::supervisor::AgentResult {
|
||||
id: agent_id,
|
||||
agent_name: "explore".into(),
|
||||
output: String::new(),
|
||||
exit_status: crate::supervisor::AgentExitStatus::Completed,
|
||||
})
|
||||
});
|
||||
std::mem::forget(rt);
|
||||
let handle = crate::supervisor::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(),
|
||||
join_handle,
|
||||
child_supervisor: None,
|
||||
};
|
||||
let mut sup = crate::supervisor::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)));
|
||||
ctx
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drain_live_notifications_keeps_registered_agent_events() {
|
||||
let ctx = ctx_with_registered_agent("agent_explore_1");
|
||||
ctx.notification_queue
|
||||
.push(agent_notification("agent_explore_1", "explore", true));
|
||||
|
||||
let live = drain_live_notifications(&ctx);
|
||||
|
||||
assert_eq!(live.len(), 1);
|
||||
assert_eq!(live[0]["event"], "agent_completed");
|
||||
assert_eq!(
|
||||
live[0]["next_action"],
|
||||
"agent__collect --id agent_explore_1 for output"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drain_live_notifications_drops_collected_agent_events() {
|
||||
let ctx = ctx_with_registered_agent("agent_explore_1");
|
||||
ctx.supervisor
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.write()
|
||||
.take("agent_explore_1")
|
||||
.unwrap();
|
||||
ctx.notification_queue
|
||||
.push(agent_notification("agent_explore_1", "explore", true));
|
||||
|
||||
assert!(drain_live_notifications(&ctx).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn eval_tool_calls_merges_notifications_at_depth_without_escalations() {
|
||||
let mut ctx = ctx_with_registered_job("job_n1");
|
||||
|
||||
@@ -5,6 +5,7 @@ use crate::config::{
|
||||
jobs_enabled, list_agents_with_descriptions,
|
||||
};
|
||||
use crate::supervisor::mailbox::{Envelope, EnvelopePayload, Inbox};
|
||||
use crate::supervisor::notification::agent_notification;
|
||||
use crate::supervisor::{AgentExitStatus, AgentHandle, AgentResult, Supervisor, TaskKind};
|
||||
use crate::utils::{AbortSignal, create_abort_signal, wait_abort_signal};
|
||||
|
||||
@@ -866,25 +867,33 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
|
||||
let spawn_agent_id = agent_id.clone();
|
||||
let spawn_agent_name = agent_name.clone();
|
||||
let spawn_abort = child_abort.clone();
|
||||
let spawn_notifications = Arc::clone(&ctx.notification_queue);
|
||||
let child_supervisor = child_ctx.supervisor.clone();
|
||||
|
||||
let join_handle = tokio::spawn(async move {
|
||||
let result = run_child_agent(child_ctx, input, spawn_abort).await;
|
||||
|
||||
match result {
|
||||
Ok(output) => Ok(AgentResult {
|
||||
let agent_result = match result {
|
||||
Ok(output) => AgentResult {
|
||||
id: spawn_agent_id,
|
||||
agent_name: spawn_agent_name,
|
||||
output,
|
||||
exit_status: AgentExitStatus::Completed,
|
||||
}),
|
||||
Err(e) => Ok(AgentResult {
|
||||
},
|
||||
Err(e) => AgentResult {
|
||||
id: spawn_agent_id,
|
||||
agent_name: spawn_agent_name,
|
||||
output: String::new(),
|
||||
exit_status: AgentExitStatus::Failed(e.to_string()),
|
||||
}),
|
||||
}
|
||||
},
|
||||
};
|
||||
let success = agent_result.exit_status == AgentExitStatus::Completed;
|
||||
spawn_notifications.push(agent_notification(
|
||||
&agent_result.id,
|
||||
&agent_result.agent_name,
|
||||
success,
|
||||
));
|
||||
Ok(agent_result)
|
||||
});
|
||||
|
||||
let handle = AgentHandle {
|
||||
|
||||
Reference in New Issue
Block a user