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 {