1use acp_thread::{AcpThread, AgentConnection, UserMessageId};
2use action_log::ActionLog;
3use agent_client_protocol as acp;
4use anyhow::{Result, anyhow};
5use collections::{BTreeMap, HashSet};
6use futures::{FutureExt, channel::mpsc};
7use gpui::{App, AppContext, AsyncApp, Entity, SharedString, Task, WeakEntity};
8use language_model::LanguageModelToolUseId;
9use project::Project;
10use schemars::JsonSchema;
11use serde::{Deserialize, Serialize};
12use smol::stream::StreamExt;
13use std::any::Any;
14use std::path::Path;
15use std::rc::Rc;
16use std::sync::Arc;
17use std::time::Duration;
18use util::ResultExt;
19use watch;
20
21use crate::{
22 AgentTool, AnyAgentTool, MAX_PARALLEL_SUBAGENTS, MAX_SUBAGENT_DEPTH, SubagentContext, Thread,
23 ThreadEvent, ToolCallAuthorization, ToolCallEventStream,
24};
25
26/// When a subagent's remaining context window falls below this fraction (25%),
27/// the "context running out" prompt is sent to encourage the subagent to wrap up.
28const CONTEXT_LOW_THRESHOLD: f32 = 0.25;
29
30/// Spawns a subagent with its own context window to perform a delegated task.
31///
32/// Use this tool when you want to do any of the following:
33/// - Perform an investigation where all you need to know is the outcome, not the research that led to that outcome.
34/// - Complete a self-contained task where you need to know if it succeeded or failed (and how), but none of its intermediate output.
35/// - Run multiple tasks in parallel that would take significantly longer to run sequentially.
36///
37/// You control what the subagent does by providing:
38/// 1. A task prompt describing what the subagent should do
39/// 2. A summary prompt that tells the subagent how to summarize its work when done
40/// 3. A "context running out" prompt for when the subagent is low on tokens
41///
42/// Each subagent has access to the same tools you do. You can optionally restrict
43/// which tools each subagent can use.
44///
45/// Note:
46/// - Maximum 8 subagents can run in parallel
47/// - Subagents cannot use tools you don't have access to
48/// - If spawning multiple subagents that might write to the filesystem, provide
49/// guidance on how to avoid conflicts (e.g. assign each to different directories)
50/// - Instruct subagents to be concise in their summaries to conserve your context
51#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
52pub struct SubagentToolInput {
53 /// Short label displayed in the UI while the subagent runs (e.g., "Researching alternatives")
54 pub label: String,
55
56 /// The initial prompt that tells the subagent what task to perform.
57 /// Be specific about what you want the subagent to accomplish.
58 pub task_prompt: String,
59
60 /// The prompt sent to the subagent when it completes its task, asking it
61 /// to summarize what it did and return results. This summary becomes the
62 /// tool result you receive.
63 ///
64 /// Example: "Summarize what you found, listing the top 3 alternatives with pros/cons."
65 pub summary_prompt: String,
66
67 /// The prompt sent if the subagent is running low on context (25% remaining).
68 /// Should instruct it to stop and summarize progress so far, plus what's left undone.
69 ///
70 /// Example: "Context is running low. Stop and summarize your progress so far,
71 /// and list what remains to be investigated."
72 pub context_low_prompt: String,
73
74 /// Optional: Maximum runtime in milliseconds. If exceeded, the subagent is
75 /// asked to summarize and return. No timeout by default.
76 #[serde(default)]
77 pub timeout_ms: Option<u64>,
78
79 /// Optional: List of tool names the subagent is allowed to use.
80 /// If not provided, the subagent can use all tools available to the parent.
81 /// Tools listed here must be a subset of the parent's available tools.
82 #[serde(default)]
83 pub allowed_tools: Option<Vec<String>>,
84}
85
86/// Tool that spawns a subagent thread to work on a task.
87pub struct SubagentTool {
88 parent_thread: WeakEntity<Thread>,
89 current_depth: u8,
90}
91
92impl SubagentTool {
93 pub fn new(parent_thread: WeakEntity<Thread>, current_depth: u8) -> Self {
94 Self {
95 parent_thread,
96 current_depth,
97 }
98 }
99
100 pub fn validate_allowed_tools(
101 &self,
102 allowed_tools: &Option<Vec<String>>,
103 cx: &App,
104 ) -> Result<()> {
105 let Some(allowed_tools) = allowed_tools else {
106 return Ok(());
107 };
108
109 let invalid_tools: Vec<_> = self.parent_thread.read_with(cx, |thread, _cx| {
110 allowed_tools
111 .iter()
112 .filter(|tool| !thread.tools.contains_key(tool.as_str()))
113 .map(|s| format!("'{s}'"))
114 .collect()
115 })?;
116
117 if !invalid_tools.is_empty() {
118 return Err(anyhow!(
119 "The following tools do not exist: {}",
120 invalid_tools.join(", ")
121 ));
122 }
123
124 Ok(())
125 }
126}
127
128impl AgentTool for SubagentTool {
129 type Input = SubagentToolInput;
130 type Output = String;
131
132 fn name() -> &'static str {
133 acp_thread::SUBAGENT_TOOL_NAME
134 }
135
136 fn kind() -> acp::ToolKind {
137 acp::ToolKind::Other
138 }
139
140 fn initial_title(
141 &self,
142 input: Result<Self::Input, serde_json::Value>,
143 _cx: &mut App,
144 ) -> SharedString {
145 input
146 .map(|i| i.label.into())
147 .unwrap_or_else(|_| "Subagent".into())
148 }
149
150 fn run(
151 self: Arc<Self>,
152 input: Self::Input,
153 event_stream: ToolCallEventStream,
154 cx: &mut App,
155 ) -> Task<Result<String>> {
156 if self.current_depth >= MAX_SUBAGENT_DEPTH {
157 return Task::ready(Err(anyhow!(
158 "Maximum subagent depth ({}) reached",
159 MAX_SUBAGENT_DEPTH
160 )));
161 }
162
163 if let Err(e) = self.validate_allowed_tools(&input.allowed_tools, cx) {
164 return Task::ready(Err(e));
165 }
166
167 let Some(parent_thread_entity) = self.parent_thread.upgrade() else {
168 return Task::ready(Err(anyhow!(
169 "Parent thread no longer exists (subagent depth={})",
170 self.current_depth + 1
171 )));
172 };
173 let parent_thread = parent_thread_entity.read(cx);
174
175 let running_count = parent_thread.running_subagent_count();
176 if running_count >= MAX_PARALLEL_SUBAGENTS {
177 return Task::ready(Err(anyhow!(
178 "Maximum parallel subagents ({}) reached. Wait for existing subagents to complete.",
179 MAX_PARALLEL_SUBAGENTS
180 )));
181 }
182
183 let parent_model = parent_thread.model().cloned();
184 let Some(model) = parent_model else {
185 return Task::ready(Err(anyhow!("No model configured")));
186 };
187
188 let parent_thread_id = parent_thread.id().clone();
189 let project = parent_thread.project.clone();
190 let project_context = parent_thread.project_context().clone();
191 let context_server_registry = parent_thread.context_server_registry.clone();
192 let templates = parent_thread.templates.clone();
193 let parent_tools = parent_thread.tools.clone();
194 let current_depth = self.current_depth;
195 let parent_thread_weak = self.parent_thread.clone();
196
197 cx.spawn(async move |cx| {
198 let subagent_context = SubagentContext {
199 parent_thread_id: parent_thread_id.clone(),
200 tool_use_id: LanguageModelToolUseId::from(uuid::Uuid::new_v4().to_string()),
201 depth: current_depth + 1,
202 summary_prompt: input.summary_prompt.clone(),
203 context_low_prompt: input.context_low_prompt.clone(),
204 };
205
206 // Determine which tools this subagent gets
207 let subagent_tools: BTreeMap<SharedString, Arc<dyn AnyAgentTool>> =
208 if let Some(ref allowed) = input.allowed_tools {
209 let allowed_set: HashSet<&str> = allowed.iter().map(|s| s.as_str()).collect();
210 parent_tools
211 .iter()
212 .filter(|(name, _)| allowed_set.contains(name.as_ref()))
213 .map(|(name, tool)| (name.clone(), tool.clone()))
214 .collect()
215 } else {
216 parent_tools.clone()
217 };
218
219 let subagent_thread: Entity<Thread> = cx.new(|cx| {
220 Thread::new_subagent(
221 project.clone(),
222 project_context.clone(),
223 context_server_registry.clone(),
224 templates.clone(),
225 model.clone(),
226 subagent_context,
227 subagent_tools,
228 cx,
229 )
230 });
231
232 let subagent_weak = subagent_thread.downgrade();
233
234 let acp_thread: Entity<AcpThread> = cx.new(|cx| {
235 let session_id = subagent_thread.read(cx).id().clone();
236 let action_log: Entity<ActionLog> = cx.new(|_| ActionLog::new(project.clone()));
237 let connection: Rc<dyn AgentConnection> = Rc::new(SubagentDisplayConnection);
238 AcpThread::new(
239 &input.label,
240 connection,
241 project.clone(),
242 action_log,
243 session_id,
244 watch::Receiver::constant(acp::PromptCapabilities::new()),
245 cx,
246 )
247 });
248
249 event_stream.update_subagent_thread(acp_thread.clone());
250
251 let mut user_stop_rx: watch::Receiver<bool> =
252 acp_thread.update(cx, |thread, _| thread.user_stop_receiver());
253
254 if let Some(parent) = parent_thread_weak.upgrade() {
255 parent.update(cx, |thread, _cx| {
256 thread.register_running_subagent(subagent_weak.clone());
257 });
258 }
259
260 // Helper to wait for user stop signal on the subagent card
261 let wait_for_user_stop = async {
262 loop {
263 if *user_stop_rx.borrow() {
264 return;
265 }
266 if user_stop_rx.changed().await.is_err() {
267 std::future::pending::<()>().await;
268 }
269 }
270 };
271
272 // Run the subagent, handling cancellation from both:
273 // 1. Parent turn cancellation (event_stream.cancelled_by_user)
274 // 2. Direct user stop on subagent card (user_stop_rx)
275 let result = futures::select! {
276 result = run_subagent(
277 &subagent_thread,
278 &acp_thread,
279 input.task_prompt,
280 input.timeout_ms,
281 cx,
282 ).fuse() => result,
283 _ = event_stream.cancelled_by_user().fuse() => {
284 let _ = subagent_thread.update(cx, |thread, cx| {
285 thread.cancel(cx).detach();
286 });
287 Err(anyhow!("Subagent cancelled by user"))
288 }
289 _ = wait_for_user_stop.fuse() => {
290 let _ = subagent_thread.update(cx, |thread, cx| {
291 thread.cancel(cx).detach();
292 });
293 Err(anyhow!("Subagent stopped by user"))
294 }
295 };
296
297 if let Some(parent) = parent_thread_weak.upgrade() {
298 let _ = parent.update(cx, |thread, _cx| {
299 thread.unregister_running_subagent(&subagent_weak);
300 });
301 }
302
303 result
304 })
305 }
306}
307
308async fn run_subagent(
309 subagent_thread: &Entity<Thread>,
310 acp_thread: &Entity<AcpThread>,
311 task_prompt: String,
312 timeout_ms: Option<u64>,
313 cx: &mut AsyncApp,
314) -> Result<String> {
315 let mut events_rx =
316 subagent_thread.update(cx, |thread, cx| thread.submit_user_message(task_prompt, cx))?;
317
318 let acp_thread_weak = acp_thread.downgrade();
319
320 let timed_out = if let Some(timeout) = timeout_ms {
321 forward_events_with_timeout(
322 &mut events_rx,
323 &acp_thread_weak,
324 Duration::from_millis(timeout),
325 cx,
326 )
327 .await
328 } else {
329 forward_events_until_stop(&mut events_rx, &acp_thread_weak, cx).await;
330 false
331 };
332
333 let should_interrupt =
334 timed_out || check_context_low(subagent_thread, CONTEXT_LOW_THRESHOLD, cx);
335
336 if should_interrupt {
337 let mut summary_rx =
338 subagent_thread.update(cx, |thread, cx| thread.interrupt_for_summary(cx))?;
339 forward_events_until_stop(&mut summary_rx, &acp_thread_weak, cx).await;
340 } else {
341 let mut summary_rx =
342 subagent_thread.update(cx, |thread, cx| thread.request_final_summary(cx))?;
343 forward_events_until_stop(&mut summary_rx, &acp_thread_weak, cx).await;
344 }
345
346 Ok(extract_last_message(subagent_thread, cx))
347}
348
349async fn forward_events_until_stop(
350 events_rx: &mut mpsc::UnboundedReceiver<Result<ThreadEvent>>,
351 acp_thread: &WeakEntity<AcpThread>,
352 cx: &mut AsyncApp,
353) {
354 while let Some(event) = events_rx.next().await {
355 match event {
356 Ok(ThreadEvent::Stop(_)) => break,
357 Ok(event) => {
358 forward_event_to_acp_thread(event, acp_thread, cx);
359 }
360 Err(_) => break,
361 }
362 }
363}
364
365async fn forward_events_with_timeout(
366 events_rx: &mut mpsc::UnboundedReceiver<Result<ThreadEvent>>,
367 acp_thread: &WeakEntity<AcpThread>,
368 timeout: Duration,
369 cx: &mut AsyncApp,
370) -> bool {
371 use futures::future::{self, Either};
372
373 let deadline = std::time::Instant::now() + timeout;
374
375 loop {
376 let remaining = deadline.saturating_duration_since(std::time::Instant::now());
377 if remaining.is_zero() {
378 return true;
379 }
380
381 let timeout_future = cx.background_executor().timer(remaining);
382 let event_future = events_rx.next();
383
384 match future::select(event_future, timeout_future).await {
385 Either::Left((event, _)) => match event {
386 Some(Ok(ThreadEvent::Stop(_))) => return false,
387 Some(Ok(event)) => {
388 forward_event_to_acp_thread(event, acp_thread, cx);
389 }
390 Some(Err(_)) => return false,
391 None => return false,
392 },
393 Either::Right((_, _)) => return true,
394 }
395 }
396}
397
398fn forward_event_to_acp_thread(
399 event: ThreadEvent,
400 acp_thread: &WeakEntity<AcpThread>,
401 cx: &mut AsyncApp,
402) {
403 match event {
404 ThreadEvent::UserMessage(message) => {
405 acp_thread
406 .update(cx, |thread, cx| {
407 for content in message.content {
408 thread.push_user_content_block(
409 Some(message.id.clone()),
410 content.into(),
411 cx,
412 );
413 }
414 })
415 .log_err();
416 }
417 ThreadEvent::AgentText(text) => {
418 acp_thread
419 .update(cx, |thread, cx| {
420 thread.push_assistant_content_block(text.into(), false, cx)
421 })
422 .log_err();
423 }
424 ThreadEvent::AgentThinking(text) => {
425 acp_thread
426 .update(cx, |thread, cx| {
427 thread.push_assistant_content_block(text.into(), true, cx)
428 })
429 .log_err();
430 }
431 ThreadEvent::ToolCallAuthorization(ToolCallAuthorization {
432 tool_call,
433 options,
434 response,
435 ..
436 }) => {
437 let outcome_task = acp_thread.update(cx, |thread, cx| {
438 thread.request_tool_call_authorization(tool_call, options, true, cx)
439 });
440 if let Ok(Ok(task)) = outcome_task {
441 cx.background_spawn(async move {
442 if let acp::RequestPermissionOutcome::Selected(
443 acp::SelectedPermissionOutcome { option_id, .. },
444 ) = task.await
445 {
446 response.send(option_id).ok();
447 }
448 })
449 .detach();
450 }
451 }
452 ThreadEvent::ToolCall(tool_call) => {
453 acp_thread
454 .update(cx, |thread, cx| thread.upsert_tool_call(tool_call, cx))
455 .log_err();
456 }
457 ThreadEvent::ToolCallUpdate(update) => {
458 acp_thread
459 .update(cx, |thread, cx| thread.update_tool_call(update, cx))
460 .log_err();
461 }
462 ThreadEvent::Retry(status) => {
463 acp_thread
464 .update(cx, |thread, cx| thread.update_retry_status(status, cx))
465 .log_err();
466 }
467 ThreadEvent::Stop(_) => {}
468 }
469}
470
471fn check_context_low(thread: &Entity<Thread>, threshold: f32, cx: &mut AsyncApp) -> bool {
472 thread.read_with(cx, |thread, _| {
473 if let Some(usage) = thread.latest_token_usage() {
474 let remaining_ratio = 1.0 - (usage.used_tokens as f32 / usage.max_tokens as f32);
475 remaining_ratio <= threshold
476 } else {
477 false
478 }
479 })
480}
481
482fn extract_last_message(thread: &Entity<Thread>, cx: &mut AsyncApp) -> String {
483 thread.read_with(cx, |thread, _| {
484 thread
485 .last_message()
486 .map(|m| m.to_markdown())
487 .unwrap_or_else(|| "No response from subagent".to_string())
488 })
489}
490
491#[cfg(test)]
492mod tests {
493 use super::*;
494 use language_model::LanguageModelToolSchemaFormat;
495
496 #[test]
497 fn test_subagent_tool_input_json_schema_is_valid() {
498 let schema = SubagentTool::input_schema(LanguageModelToolSchemaFormat::JsonSchema);
499 let schema_json = serde_json::to_value(&schema).expect("schema should serialize to JSON");
500
501 assert!(
502 schema_json.get("properties").is_some(),
503 "schema should have properties"
504 );
505 let properties = schema_json.get("properties").unwrap();
506
507 assert!(properties.get("label").is_some(), "should have label field");
508 assert!(
509 properties.get("task_prompt").is_some(),
510 "should have task_prompt field"
511 );
512 assert!(
513 properties.get("summary_prompt").is_some(),
514 "should have summary_prompt field"
515 );
516 assert!(
517 properties.get("context_low_prompt").is_some(),
518 "should have context_low_prompt field"
519 );
520 assert!(
521 properties.get("timeout_ms").is_some(),
522 "should have timeout_ms field"
523 );
524 assert!(
525 properties.get("allowed_tools").is_some(),
526 "should have allowed_tools field"
527 );
528 }
529
530 #[test]
531 fn test_subagent_tool_name() {
532 assert_eq!(SubagentTool::name(), "subagent");
533 }
534
535 #[test]
536 fn test_subagent_tool_kind() {
537 assert_eq!(SubagentTool::kind(), acp::ToolKind::Other);
538 }
539}
540
541struct SubagentDisplayConnection;
542
543impl AgentConnection for SubagentDisplayConnection {
544 fn telemetry_id(&self) -> SharedString {
545 "subagent".into()
546 }
547
548 fn auth_methods(&self) -> &[acp::AuthMethod] {
549 &[]
550 }
551
552 fn new_thread(
553 self: Rc<Self>,
554 _project: Entity<Project>,
555 _cwd: &Path,
556 _cx: &mut App,
557 ) -> Task<Result<Entity<AcpThread>>> {
558 unimplemented!("SubagentDisplayConnection does not support new_thread")
559 }
560
561 fn authenticate(&self, _method_id: acp::AuthMethodId, _cx: &mut App) -> Task<Result<()>> {
562 unimplemented!("SubagentDisplayConnection does not support authenticate")
563 }
564
565 fn prompt(
566 &self,
567 _id: Option<UserMessageId>,
568 _params: acp::PromptRequest,
569 _cx: &mut App,
570 ) -> Task<Result<acp::PromptResponse>> {
571 unimplemented!("SubagentDisplayConnection does not support prompt")
572 }
573
574 fn cancel(&self, _session_id: &acp::SessionId, _cx: &mut App) {}
575
576 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
577 self
578 }
579}