agent_engine/tools/subagent/
oneshot.rs

1use serde_json::{json, Value};
2use std::sync::atomic::Ordering;
3use std::time::Duration;
4use crate::{Result, RuntimeError, LlmEvent, SessionEvent, AgentEvent};
5use super::super::{Tool, ToolContext, resolve_agent_prompt, NEXT_SUBAGENT_ID};
6pub use crate::runtime::subagent::SubagentResult;
7
8pub struct SubagentTool;
9
10#[async_trait::async_trait]
11impl Tool for SubagentTool {
12    fn name(&self) -> &str { "subagent" }
13
14    fn description(&self) -> &str {
15        "Dispatch a one-shot subagent with a specific system prompt to perform a task. The subagent gets its own tool suite (bash, read, write, edit, grep, find, ls) and runs autonomously until done. Use this when you need the result before continuing. Blocks until done. For parallel work, use subagent_start instead. Provide either an agent name (resolves from ~/.synaps-cli/agents/<name>.md) or a system_prompt string directly."
16    }
17
18    fn parameters(&self) -> Value {
19        json!({
20            "type": "object",
21            "properties": {
22                "agent": {
23                    "type": "string",
24                    "description": "Agent name — resolves to ~/.synaps-cli/agents/<name>.md. Mutually exclusive with system_prompt."
25                },
26                "system_prompt": {
27                    "type": "string",
28                    "description": "Inline system prompt for the subagent. Use when you don't have a named agent file."
29                },
30                "task": {
31                    "type": "string",
32                    "description": "The task/prompt to send to the subagent."
33                },
34                "model": {
35                    "type": "string",
36                    "description": "Model override (default: claude-sonnet-4-6). Use claude-opus-4-7 for complex tasks."
37                },
38                "timeout": {
39                    "type": "integer",
40                    "description": "Timeout in seconds (default: 300). Increase for long-running tasks."
41                }
42            },
43            "required": ["task"]
44        })
45    }
46
47    async fn execute(&self, params: Value, ctx: ToolContext) -> Result<String> {
48        let task = params["task"].as_str()
49            .ok_or_else(|| RuntimeError::Tool("Missing 'task' parameter".to_string()))?
50            .to_string();
51
52        // Treat blank / whitespace / control-char strings as absent — see
53        // subagent_start.rs for full rationale. Models occasionally pass `agent: ""`
54        // (or "\u{0}", or " ") alongside a real `system_prompt`; without this filter
55        // we'd try to resolve an empty agent name and fail, instead of falling
56        // through to the inline prompt.
57        let is_blank = |s: &String| s.chars().all(|c| c.is_whitespace() || c.is_control());
58        let agent_name = params["agent"]
59            .as_str()
60            .map(|s| s.to_string())
61            .filter(|s| !is_blank(s));
62        let inline_prompt = params["system_prompt"]
63            .as_str()
64            .map(|s| s.to_string())
65            .filter(|s| !is_blank(s));
66        let model_override = params["model"].as_str().map(|s| s.to_string());
67        let timeout_secs = params["timeout"].as_u64().unwrap_or(ctx.limits.subagent_timeout);
68
69        let system_prompt = match (&agent_name, &inline_prompt) {
70            (Some(name), _) => {
71                resolve_agent_prompt(name)
72                    .map_err(RuntimeError::Tool)?
73            }
74            (None, Some(prompt)) => prompt.clone(),
75            (None, None) => {
76                return Err(RuntimeError::Tool(
77                    "Must provide either 'agent' (name) or 'system_prompt' (inline). Got neither.".to_string()
78                ));
79            }
80        };
81
82        let label = agent_name.as_deref().unwrap_or("inline").to_string();
83        let model = model_override.unwrap_or_else(|| crate::models::default_model().to_string());
84        let task_preview: String = task.chars().take(80).collect();
85        let subagent_id = NEXT_SUBAGENT_ID.fetch_add(1, Ordering::Relaxed);
86
87        tracing::info!("Dispatching subagent '{}' (id={}) with model {}", label, subagent_id, model);
88
89        if let Some(ref tx) = ctx.channels.tx_events {
90            let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentStart {
91                subagent_id,
92                agent_name: label.clone(),
93                task_preview: task_preview.clone(),
94            }));
95        }
96
97        let start_time = std::time::Instant::now();
98
99        let (result_tx, result_rx) = tokio::sync::oneshot::channel::<std::result::Result<SubagentResult, String>>();
100        let label_inner = label.clone();
101        let model_inner = model.clone();
102        let tx_events_inner = ctx.channels.tx_events.clone();
103
104        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
105
106        let _thread_handle = std::thread::spawn(move || {
107            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
108                let rt = match tokio::runtime::Builder::new_current_thread()
109                    .enable_all()
110                    .build()
111                {
112                    Ok(rt) => rt,
113                    Err(e) => {
114                        let _ = result_tx.send(Err(format!("Failed to create tokio runtime: {}", e)));
115                        return;
116                    }
117                };
118
119                let result = rt.block_on(async move {
120                    use futures::StreamExt;
121
122                    let mut runtime = match crate::Runtime::new().await {
123                        Ok(r) => r,
124                        Err(e) => return Err(format!("Failed to create subagent runtime: {}", e)),
125                    };
126
127                    // Inherit the user's credential source (Remote/broker) — a
128                    // fresh Runtime::new() defaults to Local. (#158 A3)
129                    runtime.apply_auth_config(&crate::config::load_config());
130                    runtime.set_system_prompt(system_prompt);
131                    runtime.set_model(model);
132                    runtime.set_tools(crate::ToolRegistry::without_subagent());
133
134                    let cancel = crate::CancellationToken::new();
135                    let cancel_inner = cancel.clone();
136
137                    tokio::spawn(async move {
138                        let _ = shutdown_rx.await;
139                        cancel_inner.cancel();
140                    });
141
142                    let mut stream = runtime.run_stream(task, cancel).await;
143
144                let mut final_text = String::new();
145                let mut tool_count = 0u32;
146                let mut tool_log: Vec<String> = Vec::new();
147                let mut total_input_tokens = 0u64;
148                let mut total_output_tokens = 0u64;
149                let mut total_cache_read = 0u64;
150                let mut total_cache_creation = 0u64;
151                // TTL split: None only if no turn ever reported one; otherwise summed.
152                let mut total_cache_5m: Option<u64> = None;
153                let mut total_cache_1h: Option<u64> = None;
154
155                let timeout_fut = tokio::time::sleep(Duration::from_secs(timeout_secs));
156                tokio::pin!(timeout_fut);
157
158                loop {
159                    tokio::select! {
160                        event = stream.next() => {
161                            let Some(event) = event else { break };
162                            match event {
163                                crate::StreamEvent::Llm(LlmEvent::Thinking(_)) => {
164                                    if let Some(ref tx) = tx_events_inner {
165                                        let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
166                                            subagent_id,
167                                            agent_name: label_inner.clone(),
168                                            status: "💭 thinking...".to_string(),
169                                        }));
170                                    }
171                                }
172                                crate::StreamEvent::Llm(LlmEvent::Text(text)) => {
173                                    final_text.push_str(&text);
174                                }
175                                crate::StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name: name, .. }) => {
176                                    tool_count += 1;
177                                    if let Some(ref tx) = tx_events_inner {
178                                        let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
179                                            subagent_id,
180                                            agent_name: label_inner.clone(),
181                                            status: format!("⚙ {} (tool #{})", name, tool_count),
182                                        }));
183                                    }
184                                }
185                                crate::StreamEvent::Llm(LlmEvent::ToolUse { tool_name, input, .. }) => {
186                                    let input_str = input.to_string();
187                                    let input_preview: String = input_str.chars().take(200).collect();
188                                    tool_log.push(format!("[tool_use]: {} — {}", tool_name, input_preview));
189                                    // Build a rich status from the tool input
190                                    let detail = match tool_name.as_str() {
191                                        "bash" => {
192                                            let cmd = input["command"].as_str().unwrap_or("");
193                                            let preview: String = cmd.chars().take(60).collect();
194                                            format!("$ {}", preview)
195                                        }
196                                        "read" => {
197                                            let path = input["path"].as_str().unwrap_or("?");
198                                            let short = path.rsplit('/').next().unwrap_or(path);
199                                            format!("reading {}", short)
200                                        }
201                                        "write" => {
202                                            let path = input["path"].as_str().unwrap_or("?");
203                                            let short = path.rsplit('/').next().unwrap_or(path);
204                                            format!("writing {}", short)
205                                        }
206                                        "edit" => {
207                                            let path = input["path"].as_str().unwrap_or("?");
208                                            let short = path.rsplit('/').next().unwrap_or(path);
209                                            format!("editing {}", short)
210                                        }
211                                        "grep" => {
212                                            let pat = input["pattern"].as_str().unwrap_or("?");
213                                            let preview: String = pat.chars().take(30).collect();
214                                            format!("grep /{}/", preview)
215                                        }
216                                        "find" => {
217                                            let pat = input["pattern"].as_str().unwrap_or("?");
218                                            format!("find {}", pat)
219                                        }
220                                        "ls" => {
221                                            let path = input["path"].as_str().unwrap_or(".");
222                                            let short = path.rsplit('/').next().unwrap_or(path);
223                                            format!("ls {}", short)
224                                        }
225                                        "subagent" => {
226                                            let name = input["agent"].as_str()
227                                                .or_else(|| input["system_prompt"].as_str().map(|s| if s.len() > 20 { "inline" } else { s }))
228                                                .unwrap_or("?");
229                                            format!("spawning {}", name)
230                                        }
231                                        other => {
232                                            // MCP or unknown tools — show tool name + first param
233                                            let short_name = if other.starts_with("ext__") {
234                                                other.splitn(3, "__").last().unwrap_or(other)
235                                            } else {
236                                                other
237                                            };
238                                            short_name.to_string()
239                                        }
240                                    };
241                                    if let Some(ref tx) = tx_events_inner {
242                                        let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
243                                            subagent_id,
244                                            agent_name: label_inner.clone(),
245                                            status: detail,
246                                        }));
247                                    }
248                                }
249                                crate::StreamEvent::Llm(LlmEvent::ToolResult { result, .. }) => {
250                                    let preview: String = result.chars().take(300).collect();
251                                    tool_log.push(format!("[tool_result]: {}", preview));
252                                }
253                                crate::StreamEvent::Session(SessionEvent::Usage {
254                                    input_tokens, output_tokens,
255                                    cache_read_input_tokens, cache_creation_input_tokens,
256                                    cache_creation_5m, cache_creation_1h,
257                                    model: _,
258                                }) => {
259                                    total_input_tokens += input_tokens;
260                                    total_output_tokens += output_tokens;
261                                    total_cache_read += cache_read_input_tokens;
262                                    total_cache_creation += cache_creation_input_tokens;
263                                    crate::core::rpc_dispatch::merge_split(&mut total_cache_5m, cache_creation_5m);
264                                    crate::core::rpc_dispatch::merge_split(&mut total_cache_1h, cache_creation_1h);
265                                }
266                                crate::StreamEvent::Session(SessionEvent::Error(e)) => {
267                                    return Err(e);
268                                }
269                                crate::StreamEvent::Session(SessionEvent::Done) => break,
270                                _ => {}
271                            }
272                        }
273                        _ = &mut timeout_fut => {
274                            // Return partial work instead of just an error
275                            let mut partial = format!("[TIMED OUT after {}s — partial results below]\n\n", timeout_secs);
276                            if !tool_log.is_empty() {
277                                partial.push_str(&tool_log.join("\n"));
278                                partial.push('\n');
279                            }
280                            if !final_text.is_empty() {
281                                partial.push_str("\n[partial response]:\n");
282                                partial.push_str(&final_text);
283                            }
284                            return Ok(SubagentResult {
285                                text: partial,
286                                model: model_inner,
287                                input_tokens: total_input_tokens,
288                                output_tokens: total_output_tokens,
289                                cache_read: total_cache_read,
290                                cache_creation: total_cache_creation,
291                                cache_creation_5m: total_cache_5m,
292                                cache_creation_1h: total_cache_1h,
293                                tool_count,
294                            });
295                        }
296                    }
297                }
298
299                Ok(SubagentResult {
300                    text: final_text,
301                    model: model_inner,
302                    input_tokens: total_input_tokens,
303                    output_tokens: total_output_tokens,
304                    cache_read: total_cache_read,
305                    cache_creation: total_cache_creation,
306                    cache_creation_5m: total_cache_5m,
307                    cache_creation_1h: total_cache_1h,
308                    tool_count,
309                })
310            });
311
312                let _ = result_tx.send(result);
313            }));
314
315            if let Err(panic_info) = result {
316                let msg = if let Some(s) = panic_info.downcast_ref::<&str>() {
317                    s.to_string()
318                } else if let Some(s) = panic_info.downcast_ref::<String>() {
319                    s.clone()
320                } else {
321                    "unknown panic".to_string()
322                };
323                tracing::error!("Subagent thread panicked: {}", msg);
324                // result_tx is consumed inside the closure, so we can't send here —
325                // the oneshot receiver will see a RecvError, handled below.
326            }
327        });
328
329        let result = result_rx.await;
330        let elapsed = start_time.elapsed().as_secs_f64();
331
332        drop(shutdown_tx);
333
334        let log_dir = crate::config::base_dir().join("logs").join("subagents");
335        let _ = tokio::fs::create_dir_all(&log_dir).await;
336        let timestamp = chrono::Local::now().format("%Y%m%d-%H%M%S");
337
338        match result {
339            Ok(Ok(sa_result)) => {
340                let preview: String = sa_result.text.chars().take(120).collect();
341
342                if let Some(ref tx) = ctx.channels.tx_events {
343                    let _ = tx.send(crate::StreamEvent::Session(SessionEvent::Usage {
344                        input_tokens: sa_result.input_tokens,
345                        output_tokens: sa_result.output_tokens,
346                        cache_read_input_tokens: sa_result.cache_read,
347                        cache_creation_input_tokens: sa_result.cache_creation,
348                        // TTL split aggregated across the subagent's turns —
349                        // forwarded so the parent bills 1h writes at 2.0x, not 1.25x.
350                        cache_creation_5m: sa_result.cache_creation_5m,
351                        cache_creation_1h: sa_result.cache_creation_1h,
352                        model: Some(sa_result.model),
353                    }));
354                    let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
355                        subagent_id,
356                        agent_name: label.clone(),
357                        result_preview: preview,
358                        duration_secs: elapsed,
359                    }));
360                }
361
362                let log_content = format!(
363                    "# Subagent: {}\nDate: {}\nModel: {}\nTask: {}\nDuration: {:.1}s\nTokens: {}in/{}out ({}cr/{}cw)\nTools used: {}\n\n## Result\n\n{}\n",
364                    label, timestamp, params["model"].as_str().unwrap_or("sonnet"),
365                    task_preview, elapsed,
366                    sa_result.input_tokens, sa_result.output_tokens,
367                    sa_result.cache_read, sa_result.cache_creation,
368                    sa_result.tool_count, sa_result.text,
369                );
370                let log_path = log_dir.join(format!("{}-{}.md", timestamp, label));
371                let _ = tokio::fs::write(&log_path, &log_content).await;
372
373                Ok(format!("[subagent:{}] {}", label, sa_result.text))
374            }
375            Ok(Err(e)) => {
376                if let Some(ref tx) = ctx.channels.tx_events {
377                    let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
378                        subagent_id,
379                        agent_name: label.clone(),
380                        result_preview: format!("ERROR: {}", e),
381                        duration_secs: elapsed,
382                    }));
383                }
384                let log_path = log_dir.join(format!("{}-{}-error.md", timestamp, label));
385                let _ = tokio::fs::write(&log_path, format!("# Subagent ERROR: {}\nTask: {}\nError: {}\n", label, task_preview, e)).await;
386                Ok(format!("[subagent:{} ERROR] {}", label, e))
387            }
388            Err(_) => {
389                if let Some(ref tx) = ctx.channels.tx_events {
390                    let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
391                        subagent_id,
392                        agent_name: label.clone(),
393                        result_preview: "Task panicked or dropped".to_string(),
394                        duration_secs: elapsed,
395                    }));
396                }
397                Ok(format!("[subagent:{} ERROR] Subagent task panicked or was dropped", label))
398            }
399        }
400    }
401}
402
403
404#[cfg(test)]
405mod tests {
406    use super::*;
407    use crate::tools::test_helpers::create_tool_context;
408    use crate::tools::Tool;
409    use serde_json::json;
410
411    #[test]
412    fn test_subagent_tool_schema() {
413        let tool = SubagentTool;
414        assert_eq!(tool.name(), "subagent");
415        assert!(!tool.description().is_empty());
416
417        let params = tool.parameters();
418        assert_eq!(params["type"], "object");
419        assert!(params["properties"].is_object());
420        assert!(params["required"].is_array());
421    }
422
423    #[tokio::test]
424    async fn test_subagent_blank_agent_uses_system_prompt() {
425        let tool = SubagentTool;
426        let ctx = create_tool_context();
427        let params = json!({
428            "agent": "",
429            "system_prompt": "You are a concise test subagent. Reply with only: ok",
430            "task": "Say ok",
431            "model": "claude-sonnet-4-6",
432            "timeout": 1
433        });
434
435        let result = tool.execute(params, ctx).await;
436        assert!(result.is_ok(), "blank agent should not be resolved as ~/.synaps-cli/agents/.md: {result:?}");
437    }
438}
agent_engine/tools/subagent/oneshot.rs

agent_engine/tools/subagent/
oneshot.rs
