From caabf41b65f46ec5d7dea07cbacf1609bb703206 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Tue, 25 Aug 2026 17:52:19 -0600 Subject: [PATCH] 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. --- src/function/mod.rs | 64 +++++++++++++++++++++++++++++++++- src/function/supervisor.rs | 21 +++++++---- src/supervisor/notification.rs | 40 +++++++++++++++++++++ 3 files changed, 118 insertions(+), 7 deletions(-) diff --git a/src/function/mod.rs b/src/function/mod.rs index 4555dca..5b1b614 100644 --- a/src/function/mod.rs +++ b/src/function/mod.rs @@ -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"); diff --git a/src/function/supervisor.rs b/src/function/supervisor.rs index 2e766e2..fee6e3e 100644 --- a/src/function/supervisor.rs +++ b/src/function/supervisor.rs @@ -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 { 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 { diff --git a/src/supervisor/notification.rs b/src/supervisor/notification.rs index 7f9d830..4f9261a 100644 --- a/src/supervisor/notification.rs +++ b/src/supervisor/notification.rs @@ -40,6 +40,20 @@ pub fn job_notification(id: &str, tool: &str, success: bool) -> SystemNotificati } } +pub fn agent_notification(id: &str, agent_name: &str, success: bool) -> SystemNotification { + SystemNotification { + event: if success { + "agent_completed" + } else { + "agent_failed" + }, + id: id.to_string(), + tool_or_agent: agent_name.to_string(), + status: if success { "success" } else { "failed" }, + next_action: format!("agent__collect --id {id} for output"), + } +} + /// Completion events for background work started by ONE context. Unlike the /// escalation queue (shared, root-owned), every context owns a fresh queue: /// a queue shared between parent and child would race their drains and @@ -106,6 +120,32 @@ mod tests { assert_eq!(event.next_action, "job__collect --id job_a1b2 for output"); } + #[test] + fn agent_notification_success_shape() { + let event = agent_notification("agent_explore_a1b2", "explore", true); + assert_eq!( + event.to_value(), + json!({ + "event": "agent_completed", + "id": "agent_explore_a1b2", + "tool_or_agent": "explore", + "status": "success", + "next_action": "agent__collect --id agent_explore_a1b2 for output", + }) + ); + } + + #[test] + fn agent_notification_failure_shape() { + let event = agent_notification("agent_explore_a1b2", "explore", false); + assert_eq!(event.event, "agent_failed"); + assert_eq!(event.status, "failed"); + assert_eq!( + event.next_action, + "agent__collect --id agent_explore_a1b2 for output" + ); + } + #[test] fn drain_empties_queue_and_preserves_order() { let queue = NotificationQueue::new();