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                    // Apply subagent spawn policy: inherit credential source AND
128                    // unconditionally force cache TTL to 5m. Subagents are short-lived
129                    // one-shots — paying the 1h write premium (~2× input price) on them
130                    // is unrecoverable waste (~$0.23 per 10-spawn fan-out). (#110)
131                    super::apply_subagent_runtime_policy(&mut runtime, &crate::config::load_config());
132                    runtime.set_system_prompt(system_prompt);
133                    runtime.set_model(model);
134                    runtime.set_tools(crate::ToolRegistry::without_subagent());
135
136                    let cancel = crate::CancellationToken::new();
137                    let cancel_inner = cancel.clone();
138
139                    tokio::spawn(async move {
140                        let _ = shutdown_rx.await;
141                        cancel_inner.cancel();
142                    });
143
144                    let mut stream = runtime.run_stream(task, cancel).await;
145
146                let mut final_text = String::new();
147                let mut tool_count = 0u32;
148                let mut tool_log: Vec<String> = Vec::new();
149                let mut total_input_tokens = 0u64;
150                let mut total_output_tokens = 0u64;
151                let mut total_cache_read = 0u64;
152                let mut total_cache_creation = 0u64;
153                // TTL split: None only if no turn ever reported one; otherwise summed.
154                let mut total_cache_5m: Option<u64> = None;
155                let mut total_cache_1h: Option<u64> = None;
156
157                let timeout_fut = tokio::time::sleep(Duration::from_secs(timeout_secs));
158                tokio::pin!(timeout_fut);
159
160                loop {
161                    tokio::select! {
162                        event = stream.next() => {
163                            let Some(event) = event else { break };
164                            match event {
165                                crate::StreamEvent::Llm(LlmEvent::Thinking(_)) => {
166                                    if let Some(ref tx) = tx_events_inner {
167                                        let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
168                                            subagent_id,
169                                            agent_name: label_inner.clone(),
170                                            status: "💭 thinking...".to_string(),
171                                        }));
172                                    }
173                                }
174                                crate::StreamEvent::Llm(LlmEvent::Text(text)) => {
175                                    final_text.push_str(&text);
176                                }
177                                crate::StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name: name, .. }) => {
178                                    tool_count += 1;
179                                    if let Some(ref tx) = tx_events_inner {
180                                        let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
181                                            subagent_id,
182                                            agent_name: label_inner.clone(),
183                                            status: format!("⚙ {} (tool #{})", name, tool_count),
184                                        }));
185                                    }
186                                }
187                                crate::StreamEvent::Llm(LlmEvent::ToolUse { tool_name, input, .. }) => {
188                                    let input_str = input.to_string();
189                                    let input_preview: String = input_str.chars().take(200).collect();
190                                    tool_log.push(format!("[tool_use]: {} — {}", tool_name, input_preview));
191                                    // Build a rich status from the tool input
192                                    let detail = match tool_name.as_str() {
193                                        "bash" => {
194                                            let cmd = input["command"].as_str().unwrap_or("");
195                                            let preview: String = cmd.chars().take(60).collect();
196                                            format!("$ {}", preview)
197                                        }
198                                        "read" => {
199                                            let path = input["path"].as_str().unwrap_or("?");
200                                            let short = path.rsplit('/').next().unwrap_or(path);
201                                            format!("reading {}", short)
202                                        }
203                                        "write" => {
204                                            let path = input["path"].as_str().unwrap_or("?");
205                                            let short = path.rsplit('/').next().unwrap_or(path);
206                                            format!("writing {}", short)
207                                        }
208                                        "edit" => {
209                                            let path = input["path"].as_str().unwrap_or("?");
210                                            let short = path.rsplit('/').next().unwrap_or(path);
211                                            format!("editing {}", short)
212                                        }
213                                        "grep" => {
214                                            let pat = input["pattern"].as_str().unwrap_or("?");
215                                            let preview: String = pat.chars().take(30).collect();
216                                            format!("grep /{}/", preview)
217                                        }
218                                        "find" => {
219                                            let pat = input["pattern"].as_str().unwrap_or("?");
220                                            format!("find {}", pat)
221                                        }
222                                        "ls" => {
223                                            let path = input["path"].as_str().unwrap_or(".");
224                                            let short = path.rsplit('/').next().unwrap_or(path);
225                                            format!("ls {}", short)
226                                        }
227                                        "subagent" => {
228                                            let name = input["agent"].as_str()
229                                                .or_else(|| input["system_prompt"].as_str().map(|s| if s.len() > 20 { "inline" } else { s }))
230                                                .unwrap_or("?");
231                                            format!("spawning {}", name)
232                                        }
233                                        other => {
234                                            // MCP or unknown tools — show tool name + first param
235                                            let short_name = if other.starts_with("ext__") {
236                                                other.splitn(3, "__").last().unwrap_or(other)
237                                            } else {
238                                                other
239                                            };
240                                            short_name.to_string()
241                                        }
242                                    };
243                                    if let Some(ref tx) = tx_events_inner {
244                                        let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentUpdate {
245                                            subagent_id,
246                                            agent_name: label_inner.clone(),
247                                            status: detail,
248                                        }));
249                                    }
250                                }
251                                crate::StreamEvent::Llm(LlmEvent::ToolResult { result, .. }) => {
252                                    let preview: String = result.chars().take(300).collect();
253                                    tool_log.push(format!("[tool_result]: {}", preview));
254                                }
255                                crate::StreamEvent::Session(SessionEvent::Usage {
256                                    input_tokens, output_tokens,
257                                    cache_read_input_tokens, cache_creation_input_tokens,
258                                    cache_creation_5m, cache_creation_1h,
259                                    model: _,
260                                }) => {
261                                    total_input_tokens += input_tokens;
262                                    total_output_tokens += output_tokens;
263                                    total_cache_read += cache_read_input_tokens;
264                                    total_cache_creation += cache_creation_input_tokens;
265                                    crate::core::rpc_dispatch::merge_split(&mut total_cache_5m, cache_creation_5m);
266                                    crate::core::rpc_dispatch::merge_split(&mut total_cache_1h, cache_creation_1h);
267                                }
268                                crate::StreamEvent::Session(SessionEvent::Error(e)) => {
269                                    return Err(e);
270                                }
271                                crate::StreamEvent::Session(SessionEvent::Done) => break,
272                                _ => {}
273                            }
274                        }
275                        _ = &mut timeout_fut => {
276                            // Return partial work instead of just an error
277                            let mut partial = format!("[TIMED OUT after {}s — partial results below]\n\n", timeout_secs);
278                            if !tool_log.is_empty() {
279                                partial.push_str(&tool_log.join("\n"));
280                                partial.push('\n');
281                            }
282                            if !final_text.is_empty() {
283                                partial.push_str("\n[partial response]:\n");
284                                partial.push_str(&final_text);
285                            }
286                            return Ok(SubagentResult {
287                                text: partial,
288                                model: model_inner,
289                                input_tokens: total_input_tokens,
290                                output_tokens: total_output_tokens,
291                                cache_read: total_cache_read,
292                                cache_creation: total_cache_creation,
293                                cache_creation_5m: total_cache_5m,
294                                cache_creation_1h: total_cache_1h,
295                                tool_count,
296                            });
297                        }
298                    }
299                }
300
301                Ok(SubagentResult {
302                    text: final_text,
303                    model: model_inner,
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                let _ = result_tx.send(result);
315            }));
316
317            if let Err(panic_info) = result {
318                let msg = if let Some(s) = panic_info.downcast_ref::<&str>() {
319                    s.to_string()
320                } else if let Some(s) = panic_info.downcast_ref::<String>() {
321                    s.clone()
322                } else {
323                    "unknown panic".to_string()
324                };
325                tracing::error!("Subagent thread panicked: {}", msg);
326                // result_tx is consumed inside the closure, so we can't send here —
327                // the oneshot receiver will see a RecvError, handled below.
328            }
329        });
330
331        let result = result_rx.await;
332        let elapsed = start_time.elapsed().as_secs_f64();
333
334        drop(shutdown_tx);
335
336        let log_dir = crate::config::base_dir().join("logs").join("subagents");
337        let _ = tokio::fs::create_dir_all(&log_dir).await;
338        let timestamp = chrono::Local::now().format("%Y%m%d-%H%M%S");
339
340        match result {
341            Ok(Ok(sa_result)) => {
342                let preview: String = sa_result.text.chars().take(120).collect();
343
344                if let Some(ref tx) = ctx.channels.tx_events {
345                    let _ = tx.send(crate::StreamEvent::Session(SessionEvent::Usage {
346                        input_tokens: sa_result.input_tokens,
347                        output_tokens: sa_result.output_tokens,
348                        cache_read_input_tokens: sa_result.cache_read,
349                        cache_creation_input_tokens: sa_result.cache_creation,
350                        // TTL split aggregated across the subagent's turns —
351                        // forwarded so the parent bills 1h writes at 2.0x, not 1.25x.
352                        cache_creation_5m: sa_result.cache_creation_5m,
353                        cache_creation_1h: sa_result.cache_creation_1h,
354                        model: Some(sa_result.model),
355                    }));
356                    let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
357                        subagent_id,
358                        agent_name: label.clone(),
359                        result_preview: preview,
360                        duration_secs: elapsed,
361                    }));
362                }
363
364                let log_content = format!(
365                    "# Subagent: {}\nDate: {}\nModel: {}\nTask: {}\nDuration: {:.1}s\nTokens: {}in/{}out ({}cr/{}cw)\nTools used: {}\n\n## Result\n\n{}\n",
366                    label, timestamp, params["model"].as_str().unwrap_or("sonnet"),
367                    task_preview, elapsed,
368                    sa_result.input_tokens, sa_result.output_tokens,
369                    sa_result.cache_read, sa_result.cache_creation,
370                    sa_result.tool_count, sa_result.text,
371                );
372                let log_path = log_dir.join(format!("{}-{}.md", timestamp, label));
373                let _ = tokio::fs::write(&log_path, &log_content).await;
374
375                Ok(format!("[subagent:{}] {}", label, sa_result.text))
376            }
377            Ok(Err(e)) => {
378                if let Some(ref tx) = ctx.channels.tx_events {
379                    let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
380                        subagent_id,
381                        agent_name: label.clone(),
382                        result_preview: format!("ERROR: {}", e),
383                        duration_secs: elapsed,
384                    }));
385                }
386                let log_path = log_dir.join(format!("{}-{}-error.md", timestamp, label));
387                let _ = tokio::fs::write(&log_path, format!("# Subagent ERROR: {}\nTask: {}\nError: {}\n", label, task_preview, e)).await;
388                Ok(format!("[subagent:{} ERROR] {}", label, e))
389            }
390            Err(_) => {
391                if let Some(ref tx) = ctx.channels.tx_events {
392                    let _ = tx.send(crate::StreamEvent::Agent(AgentEvent::SubagentDone {
393                        subagent_id,
394                        agent_name: label.clone(),
395                        result_preview: "Task panicked or dropped".to_string(),
396                        duration_secs: elapsed,
397                    }));
398                }
399                Ok(format!("[subagent:{} ERROR] Subagent task panicked or was dropped", label))
400            }
401        }
402    }
403}
404
405
406#[cfg(test)]
407mod tests {
408    use super::*;
409    use crate::tools::test_helpers::create_tool_context;
410    use crate::tools::Tool;
411    use serde_json::json;
412
413    #[test]
414    fn test_subagent_tool_schema() {
415        let tool = SubagentTool;
416        assert_eq!(tool.name(), "subagent");
417        assert!(!tool.description().is_empty());
418
419        let params = tool.parameters();
420        assert_eq!(params["type"], "object");
421        assert!(params["properties"].is_object());
422        assert!(params["required"].is_array());
423    }
424
425    #[tokio::test]
426    async fn test_subagent_blank_agent_uses_system_prompt() {
427        let tool = SubagentTool;
428        let ctx = create_tool_context();
429        let params = json!({
430            "agent": "",
431            "system_prompt": "You are a concise test subagent. Reply with only: ok",
432            "task": "Say ok",
433            "model": "claude-sonnet-4-6",
434            "timeout": 1
435        });
436
437        let result = tool.execute(params, ctx).await;
438        assert!(result.is_ok(), "blank agent should not be resolved as ~/.synaps-cli/agents/.md: {result:?}");
439    }
440}
agent_engine/tools/subagent/oneshot.rs

agent_engine/tools/subagent/
oneshot.rs
