agent_engine/tools/subagent/
start.rs

1//! SubagentStartTool — dispatch a reactive subagent and return a handle_id immediately.
2//!
3//! Unlike the one-shot `subagent` tool, this tool returns *before* the subagent
4//! finishes. The caller gets a `handle_id` they can poll via `subagent_status`,
5//! steer via `subagent_steer`, or block on via `subagent_collect`.
6
7use serde_json::{json, Value};
8use std::sync::atomic::Ordering;
9use std::sync::{Arc, RwLock};
10use std::time::Duration;
11use tokio::sync::{mpsc, oneshot};
12
13use crate::{Result, RuntimeError, LlmEvent, SessionEvent, AgentEvent};
14use super::super::{Tool, ToolContext, resolve_agent_prompt, NEXT_SUBAGENT_ID};
15use crate::runtime::subagent::{SubagentHandle, SubagentResult, SubagentStatus, SubagentState};
16
17pub struct SubagentStartTool;
18
19#[async_trait::async_trait]
20impl Tool for SubagentStartTool {
21    fn name(&self) -> &str { "subagent_start" }
22
23    fn description(&self) -> &str {
24        "Dispatch a reactive subagent and return immediately with a handle_id. \
25         The subagent runs in the background — use subagent_status to poll, \
26         subagent_steer to inject guidance mid-run, and subagent_collect to poll for the result (non-blocking — call \
27         repeatedly until done). Use this for parallel execution or when you \
28         want to continue working while the subagent runs. For simple sequential \
29         delegation, use subagent instead. Provide either an agent name (resolves \
30         from ~/.synaps-cli/agents/<name>.md) or a system_prompt string directly."
31    }
32
33    fn parameters(&self) -> Value {
34        json!({
35            "type": "object",
36            "properties": {
37                "agent": {
38                    "type": "string",
39                    "description": "Agent name — resolves to ~/.synaps-cli/agents/<name>.md. Mutually exclusive with system_prompt."
40                },
41                "system_prompt": {
42                    "type": "string",
43                    "description": "Inline system prompt for the subagent. Use when you don't have a named agent file."
44                },
45                "task": {
46                    "type": "string",
47                    "description": "The task/prompt to send to the subagent."
48                },
49                "model": {
50                    "type": "string",
51                    "description": "Model override (default: claude-sonnet-4-6). Use claude-opus-4-7 for complex tasks."
52                },
53                "timeout": {
54                    "type": "integer",
55                    "description": "Timeout in seconds (default: 300). Increase for long-running tasks."
56                }
57            },
58            "required": ["task"]
59        })
60    }
61
62    async fn execute(&self, params: Value, ctx: ToolContext) -> Result<String> {
63        // ── Parse params ───────────────────────────────────────────────────────
64        let task = params["task"].as_str()
65            .ok_or_else(|| RuntimeError::Tool("Missing 'task' parameter".to_string()))?
66            .to_string();
67
68        // Treat blank / whitespace / control-char strings as absent.
69        // Some model providers serialize "unset" as "" rather than omitting the field,
70        // and we must not try to resolve "" (or " ", or "\u{0}") as an agent name —
71        // doing so produces a hard "Agent '' not found" error that the model then
72        // retries forever with sentinel values instead of falling back to system_prompt.
73        let is_blank = |s: &String| s.chars().all(|c| c.is_whitespace() || c.is_control());
74        let agent_name = params["agent"]
75            .as_str()
76            .map(|s| s.to_string())
77            .filter(|s| !is_blank(s));
78        let inline_prompt = params["system_prompt"]
79            .as_str()
80            .map(|s| s.to_string())
81            .filter(|s| !is_blank(s));
82        let model_override = params["model"].as_str().map(|s| s.to_string());
83        let timeout_secs   = params["timeout"].as_u64().unwrap_or(ctx.limits.subagent_timeout);
84
85        let system_prompt = match (&agent_name, &inline_prompt) {
86            (Some(name), _) => resolve_agent_prompt(name).map_err(RuntimeError::Tool)?,
87            (None, Some(p)) => p.clone(),
88            (None, None) => {
89                return Err(RuntimeError::Tool(
90                    "Must provide either 'agent' (name) or 'system_prompt' (inline). Got neither.".to_string()
91                ));
92            }
93        };
94
95        let label = agent_name.as_deref().unwrap_or("inline").to_string();
96        let model = model_override.unwrap_or_else(|| crate::models::default_model().to_string());
97        let task_preview: String = task.chars().take(80).collect();
98        let task_full = task.clone();
99        let subagent_id = NEXT_SUBAGENT_ID.fetch_add(1, Ordering::Relaxed);
100        let handle_id = format!("sa_{}", subagent_id);
101
102        tracing::info!("subagent_start: dispatching '{}' (id={}) model={}", label, handle_id, model);
103
104        // ── Shared state ───────────────────────────────────────────────────────
105        let state = Arc::new(RwLock::new(SubagentState::new()));
106
107        // ── Channels ───────────────────────────────────────────────────────────
108        let (steer_tx, steer_rx) = mpsc::unbounded_channel::<String>();
109        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
110        let (result_tx, result_rx) = oneshot::channel::<SubagentResult>();
111
112        // ── Forward SubagentStart event to TUI ─────────────────────────────────
113        if let Some(ref tx) = ctx.channels.tx_events {
114            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentStart {
115                subagent_id,
116                agent_name: label.clone(),
117                task_preview: task_preview.clone(),
118            }));
119        }
120
121        // ── Clone state for the spawned thread ─────────────────────────────────
122        let state_t          = Arc::clone(&state);
123        let task_full_a      = task_full.clone();
124        let label_inner      = label.clone();
125        let model_inner      = model.clone();
126        let tx_events_inner  = ctx.channels.tx_events.clone();
127        let start_time       = std::time::Instant::now();
128
129        // ── Spawn subagent thread (mirrors subagent.rs) ────────────────────────
130        let system_prompt_for_handle = system_prompt.clone();
131        let thread_handle = std::thread::spawn(move || {
132            let panic_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
133                let rt = match tokio::runtime::Builder::new_current_thread()
134                    .enable_all()
135                    .build()
136                {
137                    Ok(rt) => rt,
138                    Err(e) => {
139                        state_t.write().unwrap().status =
140                            SubagentStatus::Failed(format!("tokio runtime: {}", e));
141                        return;
142                    }
143                };
144
145                // Clones for the async block — the outer closure still needs the originals.
146                let state_a        = Arc::clone(&state_t);
147                let label_a        = label_inner.clone();
148                let model_a        = model_inner.clone();
149                let tx_events_a    = tx_events_inner.clone();
150                let task_for_timeout = task_full_a.clone();
151                let task_for_complete = task_full_a;
152
153                let outcome: std::result::Result<SubagentResult, String> = rt.block_on(async move {
154                    use futures::StreamExt;
155
156                    let mut runtime = match crate::Runtime::new().await {
157                        Ok(r) => r,
158                        Err(e) => return Err(format!("Failed to create subagent runtime: {}", e)),
159                    };
160
161                    // Inherit the user's credential source (Remote/broker). (#158 A3)
162                    runtime.apply_auth_config(&crate::config::load_config());
163                    runtime.set_system_prompt(system_prompt);
164                    runtime.set_model(model_a.clone());
165                    let tools = if let Some(ext_mgr) = crate::runtime::openai::extension_manager_for_routing() {
166                        let mgr = ext_mgr.read().await;
167                        if let Some(shared) = mgr.tools_shared() {
168                            let extension_tools = shared.read().await;
169                            crate::ToolRegistry::without_subagent_with_extensions(&extension_tools)
170                        } else {
171                            crate::ToolRegistry::without_subagent()
172                        }
173                    } else {
174                        crate::ToolRegistry::without_subagent()
175                    };
176                    runtime.set_tools(tools);
177
178                    let cancel = crate::CancellationToken::new();
179                    let cancel_inner = cancel.clone();
180                    tokio::spawn(async move {
181                        let _ = shutdown_rx.await;
182                        cancel_inner.cancel();
183                    });
184
185                    let mut stream = runtime.run_stream_with_messages(vec![serde_json::json!({"role": "user", "content": task})], cancel, Some(steer_rx), None, false).await;
186
187                    let mut tool_count = 0u32;
188                    let mut total_input_tokens = 0u64;
189                    let mut total_output_tokens = 0u64;
190                    let mut total_cache_read = 0u64;
191                    let mut total_cache_creation = 0u64;
192                    // TTL split: None only if no turn ever reported one; otherwise summed.
193                    let mut total_cache_5m: Option<u64> = None;
194                    let mut total_cache_1h: Option<u64> = None;
195
196                    let timeout_fut = tokio::time::sleep(Duration::from_secs(timeout_secs));
197                    tokio::pin!(timeout_fut);
198
199                    loop {
200                        tokio::select! {
201                            event = stream.next() => {
202                                let Some(event) = event else { break };
203                                match event {
204                                    crate::StreamEvent::Llm(LlmEvent::Thinking(_)) => {
205                                        if let Some(ref tx) = tx_events_a {
206                                            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
207                                                subagent_id,
208                                                agent_name: label_a.clone(),
209                                                status: "💭 thinking...".to_string(),
210                                            }));
211                                        }
212                                    }
213                                    crate::StreamEvent::Llm(LlmEvent::Text(text)) => {
214                                        state_a.write().unwrap().partial_text.push_str(&text);
215                                    }
216                                    crate::StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name: name, .. }) => {
217                                        tool_count += 1;
218                                        if let Some(ref tx) = tx_events_a {
219                                            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
220                                                subagent_id,
221                                                agent_name: label_a.clone(),
222                                                status: format!("⚙ {} (tool #{})", name, tool_count),
223                                            }));
224                                        }
225                                    }
226                                    crate::StreamEvent::Llm(LlmEvent::ToolUse { tool_name, input, .. }) => {
227                                        let input_str = input.to_string();
228                                        let input_preview: String = input_str.chars().take(200).collect();
229                                        state_a.write().unwrap().tool_log
230                                            .push(format!("[tool_use]: {} — {}", tool_name, input_preview));
231                                        let detail = match tool_name.as_str() {
232                                            "bash" => {
233                                                let cmd = input["command"].as_str().unwrap_or("");
234                                                let preview: String = cmd.chars().take(60).collect();
235                                                format!("$ {}", preview)
236                                            }
237                                            "read"  => format!("reading {}", input["path"].as_str().unwrap_or("?").rsplit('/').next().unwrap_or("?")),
238                                            "write" => format!("writing {}", input["path"].as_str().unwrap_or("?").rsplit('/').next().unwrap_or("?")),
239                                            "edit"  => format!("editing {}", input["path"].as_str().unwrap_or("?").rsplit('/').next().unwrap_or("?")),
240                                            "grep"  => format!("grep /{}/", input["pattern"].as_str().unwrap_or("?").chars().take(30).collect::<String>()),
241                                            "find"  => format!("find {}", input["pattern"].as_str().unwrap_or("?")),
242                                            "ls"    => format!("ls {}", input["path"].as_str().unwrap_or(".").rsplit('/').next().unwrap_or(".")),
243                                            other   => {
244                                                if other.starts_with("ext__") {
245                                                    other.splitn(3, "__").last().unwrap_or(other).to_string()
246                                                } else {
247                                                    other.to_string()
248                                                }
249                                            }
250                                        };
251                                        if let Some(ref tx) = tx_events_a {
252                                            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
253                                                subagent_id,
254                                                agent_name: label_a.clone(),
255                                                status: detail,
256                                            }));
257                                        }
258                                    }
259                                    crate::StreamEvent::Llm(LlmEvent::ToolResult { result, .. }) => {
260                                        let preview: String = result.chars().take(300).collect();
261                                        state_a.write().unwrap().tool_log
262                                            .push(format!("[tool_result]: {}", preview));
263                                    }
264                                    crate::StreamEvent::Session(SessionEvent::Usage {
265                                        input_tokens, output_tokens,
266                                        cache_read_input_tokens, cache_creation_input_tokens,
267                                        cache_creation_5m, cache_creation_1h,
268                                        model: _,
269                                    }) => {
270                                        total_input_tokens    += input_tokens;
271                                        total_output_tokens   += output_tokens;
272                                        total_cache_read      += cache_read_input_tokens;
273                                        total_cache_creation  += cache_creation_input_tokens;
274                                        crate::core::rpc_dispatch::merge_split(&mut total_cache_5m, cache_creation_5m);
275                                        crate::core::rpc_dispatch::merge_split(&mut total_cache_1h, cache_creation_1h);
276                                    }
277                                    crate::StreamEvent::Session(SessionEvent::Error(e)) => return Err(e),
278                                    crate::StreamEvent::Session(SessionEvent::Done) => break,
279                                    _ => {}
280                                }
281                            }
282                            _ = &mut timeout_fut => {
283                                let (partial, log) = {
284                                    let mut s = state_a.write().unwrap();
285                                    s.status = SubagentStatus::TimedOut;
286                                    s.conversation_state = vec![
287                                        serde_json::json!({"role": "user", "content": task_for_timeout.clone()}),
288                                        serde_json::json!({"role": "assistant", "content": &s.partial_text}),
289                                    ];
290                                    (s.partial_text.clone(), s.tool_log.clone())
291                                };
292                                let mut text = format!("[TIMED OUT after {}s — partial results below]\n\n", timeout_secs);
293                                if !log.is_empty() {
294                                    text.push_str(&log.join("\n"));
295                                    text.push('\n');
296                                }
297                                if !partial.is_empty() {
298                                    text.push_str("\n[partial response]:\n");
299                                    text.push_str(&partial);
300                                }
301                                return Ok(SubagentResult {
302                                    text,
303                                    model: model_a.clone(),
304                                    input_tokens: total_input_tokens,
305                                    output_tokens: total_output_tokens,
306                                    cache_read: total_cache_read,
307                                    cache_creation: total_cache_creation,
308                                    cache_creation_5m: total_cache_5m,
309                                    cache_creation_1h: total_cache_1h,
310                                    tool_count,
311                                });
312                            }
313                        }
314                    }
315
316                    Ok(SubagentResult {
317                        text: state_a.write().unwrap().partial_text.clone(),
318                        model: model_a.clone(),
319                        input_tokens: total_input_tokens,
320                        output_tokens: total_output_tokens,
321                        cache_read: total_cache_read,
322                        cache_creation: total_cache_creation,
323                        cache_creation_5m: total_cache_5m,
324                        cache_creation_1h: total_cache_1h,
325                        tool_count,
326                    })
327                });
328
329                match outcome {
330                    Ok(sa_result) => {
331                        // Only overwrite Running → Completed (don't stomp TimedOut).
332                        {
333                            let mut s = state_t.write().unwrap();
334                            if matches!(s.status, SubagentStatus::Running) {
335                                s.status = SubagentStatus::Completed;
336                                s.conversation_state = vec![
337                                    serde_json::json!({"role": "user", "content": task_for_complete.clone()}),
338                                    serde_json::json!({"role": "assistant", "content": sa_result.text.clone()}),
339                                ];
340                            }
341                        }
342                        let elapsed = start_time.elapsed().as_secs_f64();
343                        let preview: String = sa_result.text.chars().take(120).collect();
344                        if let Some(ref tx) = tx_events_inner {
345                            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
346                                subagent_id,
347                                agent_name: label_inner.clone(),
348                                result_preview: preview,
349                                duration_secs: elapsed,
350                            }));
351                        }
352                        let _ = result_tx.send(sa_result);
353                    }
354                    Err(e) => {
355                        state_t.write().unwrap().status = SubagentStatus::Failed(e.clone());
356                        let elapsed = start_time.elapsed().as_secs_f64();
357                        if let Some(ref tx) = tx_events_inner {
358                            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
359                                subagent_id,
360                                agent_name: label_inner.clone(),
361                                result_preview: format!("ERROR: {}", e),
362                                duration_secs: elapsed,
363                            }));
364                        }
365                        // drop result_tx — collect() will surface the closed channel
366                    }
367                }
368            }));
369
370            if let Err(panic_info) = panic_result {
371                let msg = if let Some(s) = panic_info.downcast_ref::<&str>() {
372                    s.to_string()
373                } else if let Some(s) = panic_info.downcast_ref::<String>() {
374                    s.clone()
375                } else {
376                    "unknown panic".to_string()
377                };
378                tracing::error!("Subagent thread panicked: {}", msg);
379                state_t.write().unwrap().status = SubagentStatus::Failed(format!("panic: {}", msg));
380            }
381        });
382
383        // ── Build handle + register ────────────────────────────────────────────
384        let handle = SubagentHandle::new(
385            handle_id.clone(),
386            label.clone(),
387            task_preview,
388            model,
389            system_prompt_for_handle,
390            timeout_secs,
391            state,
392            Some(steer_tx),
393            Some(shutdown_tx),
394            Some(result_rx),
395        );
396
397        if let Some(registry) = &ctx.capabilities.subagent_registry {
398            let mut reg = registry.lock().unwrap();
399            reg.register(handle);
400            if let Some(h) = reg.get_mut(&handle_id) {
401                h.set_thread_handle(thread_handle);
402            }
403        } else {
404            return Err(RuntimeError::Tool(
405                "subagent_start requires a subagent_registry in ToolContext".to_string()
406            ));
407        }
408
409        Ok(json!({
410            "handle_id":  handle_id,
411            "agent_name": label,
412            "status":     "running"
413        }).to_string())
414    }
415}
416
417#[cfg(test)]
418mod tests {
419    use super::*;
420    use crate::tools::test_helpers::create_tool_context;
421    use crate::tools::{SubagentRegistry, Tool};
422    use serde_json::json;
423    use std::sync::{Arc, Mutex};
424
425    #[tokio::test]
426    async fn test_subagent_start_blank_agent_uses_system_prompt() {
427        let tool = SubagentStartTool;
428        let mut ctx = create_tool_context();
429        ctx.capabilities.subagent_registry = Some(Arc::new(Mutex::new(SubagentRegistry::new())));
430
431        let params = json!({
432            "agent": "",
433            "system_prompt": "You are a concise test subagent. Reply with only: ok",
434            "task": "Say ok",
435            "model": "claude-sonnet-4-6",
436            "timeout": 1
437        });
438
439        let result = tool.execute(params, ctx).await;
440        assert!(result.is_ok(), "blank agent should not be resolved as ~/.synaps-cli/agents/.md: {result:?}");
441        let body: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
442        assert_eq!(body["agent_name"], "inline");
443        assert!(body["handle_id"].as_str().unwrap_or_default().starts_with("sa_"));
444    }
445}
agent_engine/tools/subagent/start.rs

agent_engine/tools/subagent/
start.rs
