From a93ca70fb472f0309788db18871be3536c88fd9a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 27 Feb 2026 19:59:56 -0500 Subject: [PATCH] Emit subagent lifecycle events and forward child session events Add SubAgentSpawned/Completed/Failed/Closed/Event variants to AgentEvent with a deferred callback mechanism on SubAgentManager. Child session events are subscribed to at spawn time and forwarded as wrapped SubAgentEvent (skipping streaming noise). Wired in both the standalone CLI and the pipeline AgentBackend. Co-Authored-By: Claude Opus 4.6 --- crates/agent/src/cli.rs | 40 +++++++ crates/agent/src/lib.rs | 2 +- crates/agent/src/session.rs | 12 +- crates/agent/src/subagent.rs | 173 ++++++++++++++++++++++++++-- crates/agent/src/types.rs | 99 ++++++++++++++++ crates/attractor/src/cli/backend.rs | 8 +- crates/attractor/src/cli/mod.rs | 115 ++++++++++++++++++ 7 files changed, 439 insertions(+), 10 deletions(-) diff --git a/crates/agent/src/cli.rs b/crates/agent/src/cli.rs index a54fbf6b0..3f6a7855d 100644 --- a/crates/agent/src/cli.rs +++ b/crates/agent/src/cli.rs @@ -346,6 +346,7 @@ pub async fn run() -> anyhow::Result<()> { let manager = Arc::new(tokio::sync::Mutex::new( SubAgentManager::new(config.max_subagent_depth), )); + let manager_for_callback = manager.clone(); let factory_client = client.clone(); let factory_provider = cli.provider.clone(); let factory_model = model.to_string(); @@ -372,6 +373,9 @@ pub async fn run() -> anyhow::Result<()> { let mut session = Session::new(client, profile, env, config); + // Wire subagent event callback to parent session's emitter + manager_for_callback.lock().await.set_event_callback(session.event_callback()); + // SIGINT handler let cancel_token = session.cancel_token(); tokio::spawn(async move { @@ -415,6 +419,42 @@ pub async fn run() -> anyhow::Result<()> { reset = s.reset, ); } + AgentEvent::SubAgentSpawned { agent_id, depth, task, .. } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + let task_preview = if task.len() > 60 { &task[..60] } else { task }; + eprintln!( + " {dim}\u{25b6} subagent {short_id} spawned (depth={depth}) task={task_preview:?}{reset}", + dim = s.dim, reset = s.reset, + ); + } + AgentEvent::SubAgentCompleted { agent_id, depth, success, turns_used } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + eprintln!( + " {dim}\u{25a0} subagent {short_id} completed (depth={depth}, success={success}, turns={turns_used}){reset}", + dim = s.dim, reset = s.reset, + ); + } + AgentEvent::SubAgentFailed { agent_id, depth, error } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + eprintln!( + " {red}\u{2717} subagent {short_id} failed (depth={depth}): {error}{reset}", + red = s.red, reset = s.reset, + ); + } + AgentEvent::SubAgentClosed { agent_id, depth } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + eprintln!( + " {dim}\u{25a0} subagent {short_id} closed (depth={depth}){reset}", + dim = s.dim, reset = s.reset, + ); + } + AgentEvent::SubAgentEvent { agent_id, event: child_event, .. } if verbose => { + let short_id = &agent_id[..8.min(agent_id.len())]; + eprintln!( + " {dim}[subagent {short_id}] {child_event:?}{reset}", + dim = s.dim, reset = s.reset, + ); + } _ => {} } } diff --git a/crates/agent/src/lib.rs b/crates/agent/src/lib.rs index ad1a70bd9..a1acadc90 100644 --- a/crates/agent/src/lib.rs +++ b/crates/agent/src/lib.rs @@ -36,7 +36,7 @@ pub use skills::Skill; pub use profiles::{AnthropicProfile, EnvContext, GeminiProfile, OpenAiProfile}; pub use provider_profile::{ProfileCapabilities, ProviderProfile}; pub use session::Session; -pub use subagent::{SubAgent, SubAgentManager, SubAgentResult}; +pub use subagent::{SubAgent, SubAgentEventCallback, SubAgentManager, SubAgentResult}; pub use tool_registry::ToolRegistry; pub use tools::{ make_edit_file_tool, make_glob_tool, make_grep_tool, make_read_file_tool, make_shell_tool, diff --git a/crates/agent/src/session.rs b/crates/agent/src/session.rs index a72941cb0..d7c32f77a 100644 --- a/crates/agent/src/session.rs +++ b/crates/agent/src/session.rs @@ -191,11 +191,21 @@ impl Session { self.followup_queue.clone() } - #[must_use] + #[must_use] pub fn cancel_token(&self) -> CancellationToken { self.cancel_token.clone() } + /// Build a callback that forwards `AgentEvent`s through this session's emitter. + #[must_use] + pub fn event_callback(&self) -> crate::subagent::SubAgentEventCallback { + let emitter = self.event_emitter.clone(); + let session_id = self.id.clone(); + Arc::new(move |event| { + emitter.emit(session_id.clone(), event); + }) + } + pub fn close(&mut self) { if self.state != SessionState::Closed { self.state = SessionState::Closed; diff --git a/crates/agent/src/subagent.rs b/crates/agent/src/subagent.rs index b9197ee11..023c144d5 100644 --- a/crates/agent/src/subagent.rs +++ b/crates/agent/src/subagent.rs @@ -2,13 +2,14 @@ use crate::error::AgentError; use crate::session::Session; use crate::tool_registry::RegisteredTool; use crate::tools::required_str; -use crate::types::Turn; +use crate::types::{AgentEvent, Turn}; use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex}; use llm::types::ToolDefinition; use tokio_util::sync::CancellationToken; pub type SessionFactory = Arc Session + Send + Sync>; +pub type SubAgentEventCallback = Arc; #[derive(Debug, Clone)] pub struct SubAgentResult { @@ -21,19 +22,32 @@ pub struct SubAgent { task: Option>>, followup_queue: Arc>>, cancel_token: CancellationToken, + depth: usize, } pub struct SubAgentManager { agents: HashMap, max_depth: usize, + event_callback: Option, } impl SubAgentManager { - #[must_use] + #[must_use] pub fn new(max_depth: usize) -> Self { Self { agents: HashMap::new(), max_depth, + event_callback: None, + } + } + + pub fn set_event_callback(&mut self, cb: SubAgentEventCallback) { + self.event_callback = Some(cb); + } + + fn emit_event(&self, event: AgentEvent) { + if let Some(ref cb) = self.event_callback { + cb(event); } } @@ -54,8 +68,38 @@ impl SubAgentManager { let followup_queue = session.followup_queue_handle(); let cancel_token = session.cancel_token(); + // Subscribe to child session events and forward them via callback + if let Some(ref cb) = self.event_callback { + let mut rx = session.subscribe(); + let cb = cb.clone(); + let fwd_agent_id = agent_id.clone(); + let child_depth = depth + 1; + tokio::spawn(async move { + while let Ok(event) = rx.recv().await { + // Skip streaming / noise events + if matches!( + &event.event, + AgentEvent::TextDelta { .. } + | AgentEvent::ToolCallOutputDelta { .. } + | AgentEvent::AssistantTextStart + | AgentEvent::SessionStarted + | AgentEvent::SessionEnded + | AgentEvent::SkillExpanded { .. } + ) { + continue; + } + cb(AgentEvent::SubAgentEvent { + agent_id: fwd_agent_id.clone(), + depth: child_depth, + event: Box::new(event.event), + }); + } + }); + } + + let task_prompt_for_spawn = task_prompt.clone(); let task = tokio::spawn(async move { - session.process_input(&task_prompt).await?; + session.process_input(&task_prompt_for_spawn).await?; let turns = session.history().turns(); let last_text = turns.iter().rev().find_map(|t| match t { Turn::Assistant { content, .. } => Some(content.clone()), @@ -74,9 +118,16 @@ impl SubAgentManager { task: Some(task), followup_queue, cancel_token, + depth: depth + 1, }, ); + self.emit_event(AgentEvent::SubAgentSpawned { + agent_id: agent_id.clone(), + depth: depth + 1, + task: task_prompt, + }); + Ok(agent_id) } @@ -105,12 +156,36 @@ impl SubAgentManager { AgentError::InvalidState(format!("No agent found with id: {agent_id}")) })?; + let depth = agent.depth; + match agent.task.take() { Some(join_handle) => match join_handle.await { - Ok(result) => result, - Err(e) => Err(AgentError::InvalidState(format!( - "Agent task panicked: {e}" - ))), + Ok(Ok(result)) => { + self.emit_event(AgentEvent::SubAgentCompleted { + agent_id: agent_id.to_string(), + depth, + success: result.success, + turns_used: result.turns_used, + }); + Ok(result) + } + Ok(Err(e)) => { + self.emit_event(AgentEvent::SubAgentFailed { + agent_id: agent_id.to_string(), + depth, + error: e.to_string(), + }); + Err(e) + } + Err(e) => { + let error = format!("Agent task panicked: {e}"); + self.emit_event(AgentEvent::SubAgentFailed { + agent_id: agent_id.to_string(), + depth, + error: error.clone(), + }); + Err(AgentError::InvalidState(error)) + } }, None => Err(AgentError::InvalidState(format!( "Agent {agent_id} has no running task" @@ -132,6 +207,11 @@ impl SubAgentManager { join_handle.abort(); } + self.emit_event(AgentEvent::SubAgentClosed { + agent_id: agent_id.to_string(), + depth: agent.depth, + }); + Ok(()) } @@ -421,4 +501,83 @@ mod tests { .unwrap(); assert!(close_required.contains(&serde_json::json!("agent_id"))); } + + fn captured_events() -> (SubAgentEventCallback, Arc>>) { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_clone = events.clone(); + let cb: SubAgentEventCallback = Arc::new(move |event| { + events_clone.lock().unwrap().push(event); + }); + (cb, events) + } + + #[tokio::test] + async fn callback_captures_spawn_event() { + let (cb, events) = captured_events(); + let mut manager = SubAgentManager::new(3); + manager.set_event_callback(cb); + + let session = make_session(vec![text_response("Hello")]).await; + let _agent_id = manager.spawn(session, "test task".into(), 0).unwrap(); + + let captured = events.lock().unwrap(); + assert_eq!(captured.len(), 1); + assert!(matches!(&captured[0], AgentEvent::SubAgentSpawned { depth: 1, task, .. } if task == "test task")); + } + + #[tokio::test] + async fn callback_captures_wait_completed_event() { + let (cb, events) = captured_events(); + let mut manager = SubAgentManager::new(3); + manager.set_event_callback(cb); + + let session = make_session(vec![text_response("done")]).await; + let agent_id = manager.spawn(session, "task".into(), 0).unwrap(); + let _result = manager.wait(&agent_id).await.unwrap(); + + let captured = events.lock().unwrap(); + assert!(captured.iter().any(|e| matches!(e, AgentEvent::SubAgentCompleted { success: true, depth: 1, .. }))); + } + + #[tokio::test] + async fn callback_captures_close_event() { + let (cb, events) = captured_events(); + let mut manager = SubAgentManager::new(3); + manager.set_event_callback(cb); + + let session = make_session(vec![text_response("Hello")]).await; + let agent_id = manager.spawn(session, "task".into(), 1).unwrap(); + manager.close(&agent_id).unwrap(); + + let captured = events.lock().unwrap(); + assert!(captured.iter().any(|e| matches!(e, AgentEvent::SubAgentClosed { depth: 2, .. }))); + } + + #[tokio::test] + async fn callback_forwards_child_events() { + let (cb, events) = captured_events(); + let mut manager = SubAgentManager::new(3); + manager.set_event_callback(cb); + + let session = make_session(vec![text_response("Hello")]).await; + let agent_id = manager.spawn(session, "task".into(), 0).unwrap(); + + // Wait for agent to complete - child events arrive asynchronously + let _result = manager.wait(&agent_id).await.unwrap(); + + // Give the forwarding task a moment to process remaining events + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let captured = events.lock().unwrap(); + let forwarded_count = captured.iter().filter(|e| matches!(e, AgentEvent::SubAgentEvent { .. })).count(); + // Child session emits at least UserInput and AssistantMessage (filtered from SessionStarted/SessionEnded/etc) + assert!(forwarded_count > 0, "expected at least one forwarded child event, got {forwarded_count}"); + } + + #[test] + fn no_callback_does_not_panic() { + // Manager without callback should not panic on emit + let manager = SubAgentManager::new(3); + manager.emit_event(AgentEvent::SubAgentClosed { agent_id: "x".into(), depth: 0 }); + } } diff --git a/crates/agent/src/types.rs b/crates/agent/src/types.rs index ef18b7ca4..b0c43a32a 100644 --- a/crates/agent/src/types.rs +++ b/crates/agent/src/types.rs @@ -109,6 +109,31 @@ pub enum AgentEvent { delay_secs: f64, error: String, }, + SubAgentSpawned { + agent_id: String, + depth: usize, + task: String, + }, + SubAgentCompleted { + agent_id: String, + depth: usize, + success: bool, + turns_used: usize, + }, + SubAgentFailed { + agent_id: String, + depth: usize, + error: String, + }, + SubAgentClosed { + agent_id: String, + depth: usize, + }, + SubAgentEvent { + agent_id: String, + depth: usize, + event: Box, + }, } #[derive(Debug, Clone)] @@ -157,6 +182,80 @@ mod tests { assert!(matches!(event, AgentEvent::SkillExpanded { skill_name } if skill_name == "commit")); } + #[test] + fn subagent_spawned_constructible() { + let event = AgentEvent::SubAgentSpawned { + agent_id: "sa-1".into(), + depth: 1, + task: "list files".into(), + }; + assert!(matches!(event, AgentEvent::SubAgentSpawned { depth: 1, .. })); + } + + #[test] + fn subagent_completed_constructible() { + let event = AgentEvent::SubAgentCompleted { + agent_id: "sa-1".into(), + depth: 1, + success: true, + turns_used: 5, + }; + assert!(matches!(event, AgentEvent::SubAgentCompleted { success: true, turns_used: 5, .. })); + } + + #[test] + fn subagent_failed_constructible() { + let event = AgentEvent::SubAgentFailed { + agent_id: "sa-1".into(), + depth: 0, + error: "timeout".into(), + }; + assert!(matches!(event, AgentEvent::SubAgentFailed { depth: 0, .. })); + } + + #[test] + fn subagent_closed_constructible() { + let event = AgentEvent::SubAgentClosed { + agent_id: "sa-1".into(), + depth: 2, + }; + assert!(matches!(event, AgentEvent::SubAgentClosed { depth: 2, .. })); + } + + #[test] + fn subagent_event_wraps_child_event() { + let child = AgentEvent::ToolCallStarted { + tool_name: "read_file".into(), + tool_call_id: "tc-1".into(), + arguments: serde_json::json!({}), + }; + let event = AgentEvent::SubAgentEvent { + agent_id: "sa-1".into(), + depth: 1, + event: Box::new(child), + }; + assert!(matches!(event, AgentEvent::SubAgentEvent { depth: 1, .. })); + } + + #[test] + fn subagent_events_serde_round_trip() { + let events = vec![ + AgentEvent::SubAgentSpawned { agent_id: "sa-1".into(), depth: 0, task: "test".into() }, + AgentEvent::SubAgentCompleted { agent_id: "sa-1".into(), depth: 0, success: true, turns_used: 3 }, + AgentEvent::SubAgentFailed { agent_id: "sa-1".into(), depth: 0, error: "oops".into() }, + AgentEvent::SubAgentClosed { agent_id: "sa-1".into(), depth: 0 }, + AgentEvent::SubAgentEvent { + agent_id: "sa-1".into(), + depth: 1, + event: Box::new(AgentEvent::SessionStarted), + }, + ]; + let json = serde_json::to_string(&events).unwrap(); + let deserialized: Vec = serde_json::from_str(&json).unwrap(); + assert_eq!(deserialized.len(), 5); + assert!(matches!(&deserialized[4], AgentEvent::SubAgentEvent { event, .. } if matches!(event.as_ref(), AgentEvent::SessionStarted))); + } + #[test] fn agent_event_assistant_message() { let usage = Usage { diff --git a/crates/attractor/src/cli/backend.rs b/crates/attractor/src/cli/backend.rs index 6e5f68b40..7a1ab485e 100644 --- a/crates/attractor/src/cli/backend.rs +++ b/crates/attractor/src/cli/backend.rs @@ -65,6 +65,7 @@ impl AgentBackend { let manager = Arc::new(tokio::sync::Mutex::new( SubAgentManager::new(config.max_subagent_depth), )); + let manager_for_callback = manager.clone(); // Build factory that creates child sessions WITHOUT subagent tools let factory_client = client.clone(); @@ -91,7 +92,12 @@ impl AgentBackend { profile.register_subagent_tools(manager, factory, 0); let profile: Arc = Arc::from(profile); - Ok(Session::new(client, profile, Arc::clone(execution_env), config)) + let session = Session::new(client, profile, Arc::clone(execution_env), config); + + // Wire subagent event callback to parent session's emitter + manager_for_callback.lock().await.set_event_callback(session.event_callback()); + + Ok(session) } fn build_profile(&self) -> Box { diff --git a/crates/attractor/src/cli/mod.rs b/crates/attractor/src/cli/mod.rs index 6c0c7d984..58b62e576 100644 --- a/crates/attractor/src/cli/mod.rs +++ b/crates/attractor/src/cli/mod.rs @@ -401,6 +401,27 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String { let delay_ms = (*delay_secs * 1000.0) as u64; format!("[LLM_RETRY] stage={stage} provider={provider} model={model} attempt={attempt} delay={delay_ms}ms error=\"{error}\"") } + AgentEvent::SubAgentSpawned { agent_id, depth, task } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + let task_preview = if task.len() > 60 { &task[..60] } else { task }; + format!("[SUBAGENT_SPAWNED] stage={stage} agent_id={short_id} depth={depth} task=\"{task_preview}\"") + } + AgentEvent::SubAgentCompleted { agent_id, depth, success, turns_used } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + format!("[SUBAGENT_COMPLETED] stage={stage} agent_id={short_id} depth={depth} success={success} turns={turns_used}") + } + AgentEvent::SubAgentFailed { agent_id, depth, error } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + format!("[SUBAGENT_FAILED] stage={stage} agent_id={short_id} depth={depth} error=\"{error}\"") + } + AgentEvent::SubAgentClosed { agent_id, depth } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + format!("[SUBAGENT_CLOSED] stage={stage} agent_id={short_id} depth={depth}") + } + AgentEvent::SubAgentEvent { agent_id, depth, event } => { + let short_id = &agent_id[..8.min(agent_id.len())]; + format!("[SUBAGENT_EVENT] stage={stage} agent_id={short_id} depth={depth} event={event:?}") + } other => format!("[AGENT] stage={stage} event={other:?}"), } PipelineEvent::ParallelEarlyTermination { @@ -691,6 +712,22 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String { let delay_ms = (*delay_secs * 1000.0) as u64; format!("{d}── LLM_RETRY ────────────────────────────────{r}\n {d}stage:{r} {stage}\n {d}provider:{r} {provider}\n {d}model:{r} {model}\n {d}attempt:{r} {attempt}\n {d}delay_ms:{r} {delay_ms}\n {d}error:{r} {error}\n") } + AgentEvent::SubAgentSpawned { agent_id, depth, task } => { + let task_preview = if task.len() > 200 { &task[..200] } else { task.as_str() }; + format!("{d}── SUBAGENT_SPAWNED ─────────────────────────{r}\n {d}stage:{r} {stage}\n {d}agent_id:{r} {agent_id}\n {d}depth:{r} {depth}\n {d}task:{r} {task_preview}\n") + } + AgentEvent::SubAgentCompleted { agent_id, depth, success, turns_used } => { + format!("{d}── SUBAGENT_COMPLETED ───────────────────────{r}\n {d}stage:{r} {stage}\n {d}agent_id:{r} {agent_id}\n {d}depth:{r} {depth}\n {d}success:{r} {success}\n {d}turns_used:{r} {turns_used}\n") + } + AgentEvent::SubAgentFailed { agent_id, depth, error } => { + format!("{d}── SUBAGENT_FAILED ──────────────────────────{r}\n {d}stage:{r} {stage}\n {d}agent_id:{r} {agent_id}\n {d}depth:{r} {depth}\n {d}error:{r} {error}\n") + } + AgentEvent::SubAgentClosed { agent_id, depth } => { + format!("{d}── SUBAGENT_CLOSED ──────────────────────────{r}\n {d}stage:{r} {stage}\n {d}agent_id:{r} {agent_id}\n {d}depth:{r} {depth}\n") + } + AgentEvent::SubAgentEvent { agent_id, depth, event } => { + format!("{d}── SUBAGENT_EVENT ───────────────────────────{r}\n {d}stage:{r} {stage}\n {d}agent_id:{r} {agent_id}\n {d}depth:{r} {depth}\n {d}event:{r} {event:?}\n") + } other => format!("{d}── AGENT ────────────────────────────────────{r}\n {d}stage:{r} {stage}\n {d}event:{r} {other:?}\n"), } PipelineEvent::ParallelEarlyTermination { @@ -864,6 +901,84 @@ mod tests { assert!(s.contains("42")); } + #[test] + fn format_summary_subagent_spawned() { + let event = PipelineEvent::Agent { + stage: "code".into(), + event: AgentEvent::SubAgentSpawned { + agent_id: "abcdef12-3456-7890-abcd-ef1234567890".into(), + depth: 1, + task: "list files".into(), + }, + }; + let s = format_event_summary(&event, test_styles()); + assert!(s.contains("[SUBAGENT_SPAWNED]")); + assert!(s.contains("abcdef12")); + assert!(s.contains("depth=1")); + } + + #[test] + fn format_summary_subagent_completed() { + let event = PipelineEvent::Agent { + stage: "code".into(), + event: AgentEvent::SubAgentCompleted { + agent_id: "abcdef12-xxxx".into(), + depth: 1, + success: true, + turns_used: 5, + }, + }; + let s = format_event_summary(&event, test_styles()); + assert!(s.contains("[SUBAGENT_COMPLETED]")); + assert!(s.contains("success=true")); + assert!(s.contains("turns=5")); + } + + #[test] + fn format_detail_subagent_failed() { + let event = PipelineEvent::Agent { + stage: "code".into(), + event: AgentEvent::SubAgentFailed { + agent_id: "abcdef12-xxxx".into(), + depth: 2, + error: "timeout".into(), + }, + }; + let s = format_event_detail(&event, test_styles()); + assert!(s.contains("SUBAGENT_FAILED")); + assert!(s.contains("timeout")); + assert!(s.contains("depth")); + } + + #[test] + fn format_detail_subagent_closed() { + let event = PipelineEvent::Agent { + stage: "code".into(), + event: AgentEvent::SubAgentClosed { + agent_id: "abcdef12-xxxx".into(), + depth: 1, + }, + }; + let s = format_event_detail(&event, test_styles()); + assert!(s.contains("SUBAGENT_CLOSED")); + assert!(s.contains("abcdef12-xxxx")); + } + + #[test] + fn format_summary_subagent_event() { + let event = PipelineEvent::Agent { + stage: "code".into(), + event: AgentEvent::SubAgentEvent { + agent_id: "abcdef12-xxxx".into(), + depth: 1, + event: Box::new(AgentEvent::SessionStarted), + }, + }; + let s = format_event_summary(&event, test_styles()); + assert!(s.contains("[SUBAGENT_EVENT]")); + assert!(s.contains("abcdef12")); + } + #[test] fn format_detail_setup_command_completed() { let event = PipelineEvent::SetupCommandCompleted {