mobius/middleware/
subagents.rs

1//! Durable asynchronous child-agent middleware.
2
3use std::collections::BTreeMap;
4use std::collections::BTreeSet;
5use std::sync::Arc;
6
7use serde::Deserialize;
8use serde::Serialize;
9use serde_json::Value;
10
11use super::ActiveCommandContext;
12use super::Middleware;
13use super::MiddlewareCommandContext;
14use super::MiddlewareCommandOutput;
15use super::ModelContext;
16use super::PromptSection;
17use super::RuntimeContext;
18use super::SessionStartContext;
19use super::SessionStartSource;
20use super::SubmissionResult;
21use super::manifest::{MiddlewareManifest, MiddlewareSettingChoices, MiddlewareSettingManifest};
22use super::tools::Catalog;
23use super::tools::labeled_tool_heading;
24use super::tools::render_tool_event;
25use crate::BoxFuture;
26use crate::Error;
27use crate::Result;
28use crate::agent::{Agent, AgentRole};
29use crate::backend::checkpoint::Checkpoint;
30use crate::backend::checkpoint::CheckpointStore;
31use crate::backend::model::internal_user_message;
32use crate::protocol::EventMsg;
33use crate::protocol::FrontendBlock;
34use crate::protocol::FrontendBlockUpdate;
35use crate::protocol::FrontendCommand;
36use crate::protocol::FrontendContribution;
37use crate::protocol::FrontendEvent;
38use crate::protocol::FrontendPreviewUpdate;
39use crate::protocol::MessageAuthor;
40use crate::protocol::Op;
41use crate::protocol::internal_message_kind;
42use crate::protocol::message_metadata;
43
44use self::runtime::Shared;
45
46mod runtime;
47mod tools;
48
49use self::tools::{InterruptAgent, ListAgents, SendMessage, SpawnAgent, WaitAgent, fork_context};
50#[cfg(test)]
51use self::tools::{cleanup_error, supervise, wait_parameters, wait_timeout};
52
53const MAX_TASK_NAME_BYTES: usize = 64;
54const IDENTITY_KEY: &str = "subagents.identity";
55const SPAWN_CONTEXT_KEY: &str = "subagents.spawn_context";
56mod text {
57    pub const COMMAND_DESCRIPTION: &str = "open a subagent thread";
58    pub const DEFAULTS_MAX_AGENTS: i64 = 101;
59    pub const DEFAULTS_MAX_CONCURRENCY: i64 = 8;
60    pub const DEFAULTS_MAX_DEPTH: i64 = 4;
61    pub const DEFAULTS_WAIT_MS: i64 = 30000;
62    pub const MANIFEST_DESCRIPTION: &str = "Delegate independent work to durable child agents";
63    pub const MANIFEST_LABEL: &str = "Subagents";
64    pub const PROMPT_DEFAULT: &str = "Complete the task and report concisely to your parent.";
65    pub const PROMPT_ROOT: &str = "Delegate independent work to subagents when it can run in parallel. Spawn with fresh context by default; include recent turns only when the task requires them, and full history only when essential. They share your workspace; continue your own work while they run, and wait only when you need their results.";
66    pub const RENDER_AGENT: &str = "Agent";
67    pub const RENDER_AGENTS: &str = "Agents";
68    pub const RENDER_EMPTY: &str = "no subagents";
69    pub const RENDER_INTERRUPT: &str = "Interrupt";
70    pub const RENDER_MESSAGE: &str = "Message";
71    pub const RENDER_OPEN: &str = "Open subagent";
72    pub const RENDER_WAIT: &str = "Wait";
73    pub const SETTING_MAX_AGENTS_DESCRIPTION: &str = "Maximum retained agents, including the root";
74    pub const SETTING_MAX_AGENTS_LABEL: &str = "Maximum agents";
75    pub const SETTING_MAX_AGENTS_STEP: i64 = 1;
76    pub const SETTING_MAX_CONCURRENCY_DESCRIPTION: &str =
77        "Maximum active agents, including the root";
78    pub const SETTING_MAX_CONCURRENCY_LABEL: &str = "Maximum concurrency";
79    pub const SETTING_MAX_CONCURRENCY_STEP: i64 = 1;
80    pub const SETTING_MAX_DEPTH_DESCRIPTION: &str = "Maximum child-agent nesting depth";
81    pub const SETTING_MAX_DEPTH_LABEL: &str = "Maximum depth";
82    pub const SETTING_MAX_DEPTH_STEP: i64 = 1;
83    pub const SETTING_MODEL_ROUTE_DESCRIPTION: &str =
84        "Model route used by child agents when a spawn does not select one";
85    pub const SETTING_MODEL_ROUTE_LABEL: &str = "Default model";
86    pub const SETTING_MODEL_ROUTE_UNSET_LABEL: &str = "Inherit parent";
87    pub const TOOL_INTERRUPT_AGENT_DESCRIPTION: &str = "Interrupt a subagent in this chat's task tree and return its prior status. Cannot target /root.";
88    pub const TOOL_LIST_AGENTS_DESCRIPTION: &str =
89        "List this chat's subagents and their canonical task paths; does not list peer Bots.";
90    pub const TOOL_PARAMETER_TARGET_DESCRIPTION: &str = "Exact canonical task path from spawn_agent or list_agents, such as /root/reviewer; not a Bot handle or Bot ID.";
91    pub const TOOL_SEND_MESSAGE_DESCRIPTION: &str = "Send collaboration context to an agent in this chat's subagent task tree; completed or interrupted children are started again for the message. A child may message its parent at the parent's canonical path, including /root. The root agent is never restarted.";
92    pub const TOOL_SPAWN_AGENT_DESCRIPTION: &str = "Start an async child for independent work in this chat's subagent task tree; return its canonical task path.";
93    pub const TOOL_SPAWN_AGENT_PARAMETER_FORK_TURNS_DESCRIPTION: &str = "`none` for a fresh child (default), a positive integer for required recent turns, or `all` only when full history is essential.";
94    pub const TOOL_SPAWN_AGENT_PARAMETER_MODEL_DESCRIPTION: &str =
95        "Registered route; defaults to the child route, then parent.";
96    pub const TOOL_SPAWN_AGENT_PARAMETER_REASONING_EFFORT_DESCRIPTION: &str =
97        "Reasoning effort for the selected model; defaults to middleware configuration.";
98    pub const TOOL_SPAWN_AGENT_PARAMETER_TASK_NAME_DESCRIPTION: &str =
99        "1-64 lowercase letters, digits, or underscores.";
100    pub const TOOL_WAIT_AGENT_DESCRIPTION: &str =
101        "Wait for an update from this chat's subagent task tree.";
102}
103const MIN_WAIT_MS: u64 = 10_000;
104const MAX_WAIT_MS: u64 = 120_000;
105const MAX_CONFIGURED_DEPTH: u8 = 16;
106const MAX_CONFIGURED_CONCURRENCY: usize = 64;
107const MAX_CONFIGURED_AGENTS: usize = 256;
108const _: () = {
109    assert!(text::DEFAULTS_WAIT_MS >= MIN_WAIT_MS as i64);
110    assert!(text::DEFAULTS_WAIT_MS <= MAX_WAIT_MS as i64);
111    assert!(text::DEFAULTS_MAX_DEPTH >= 1);
112    assert!(text::DEFAULTS_MAX_DEPTH <= MAX_CONFIGURED_DEPTH as i64);
113    assert!(text::DEFAULTS_MAX_CONCURRENCY >= 2);
114    assert!(text::DEFAULTS_MAX_CONCURRENCY <= MAX_CONFIGURED_CONCURRENCY as i64);
115    assert!(text::DEFAULTS_MAX_AGENTS >= text::DEFAULTS_MAX_CONCURRENCY);
116    assert!(text::DEFAULTS_MAX_AGENTS <= MAX_CONFIGURED_AGENTS as i64);
117    assert!(text::SETTING_MAX_DEPTH_STEP > 0);
118    assert!(text::SETTING_MAX_CONCURRENCY_STEP > 0);
119    assert!(text::SETTING_MAX_AGENTS_STEP > 0);
120};
121const DEFAULT_WAIT_MS: u64 = text::DEFAULTS_WAIT_MS as u64;
122/// Default maximum child-agent nesting depth.
123pub const DEFAULT_MAX_DEPTH: u8 = text::DEFAULTS_MAX_DEPTH as u8;
124/// Default number of concurrently active agents, including the root.
125pub const DEFAULT_MAX_CONCURRENCY: usize = text::DEFAULTS_MAX_CONCURRENCY as usize;
126/// Default number of retained agents, including the root.
127pub const DEFAULT_MAX_AGENTS: usize = text::DEFAULTS_MAX_AGENTS as usize;
128const SETTINGS: &[MiddlewareSettingManifest] = &[
129    MiddlewareSettingManifest::Select {
130        id: "model_route",
131        label: text::SETTING_MODEL_ROUTE_LABEL,
132        description: text::SETTING_MODEL_ROUTE_DESCRIPTION,
133        choices: MiddlewareSettingChoices::ModelRoutes,
134        unset_label: Some(text::SETTING_MODEL_ROUTE_UNSET_LABEL),
135        default: None,
136        max_bytes: 4 * 1024,
137        composer: false,
138    },
139    MiddlewareSettingManifest::Integer {
140        id: "max_depth",
141        label: text::SETTING_MAX_DEPTH_LABEL,
142        description: text::SETTING_MAX_DEPTH_DESCRIPTION,
143        min: 1,
144        max: Some(MAX_CONFIGURED_DEPTH as i64),
145        step: text::SETTING_MAX_DEPTH_STEP,
146        default: DEFAULT_MAX_DEPTH as i64,
147    },
148    MiddlewareSettingManifest::Integer {
149        id: "max_concurrency",
150        label: text::SETTING_MAX_CONCURRENCY_LABEL,
151        description: text::SETTING_MAX_CONCURRENCY_DESCRIPTION,
152        min: 2,
153        max: Some(MAX_CONFIGURED_CONCURRENCY as i64),
154        step: text::SETTING_MAX_CONCURRENCY_STEP,
155        default: DEFAULT_MAX_CONCURRENCY as i64,
156    },
157    MiddlewareSettingManifest::Integer {
158        id: "max_agents",
159        label: text::SETTING_MAX_AGENTS_LABEL,
160        description: text::SETTING_MAX_AGENTS_DESCRIPTION,
161        min: 2,
162        max: Some(MAX_CONFIGURED_AGENTS as i64),
163        step: text::SETTING_MAX_AGENTS_STEP,
164        default: DEFAULT_MAX_AGENTS as i64,
165    },
166];
167
168/// Configuration and presentation metadata for child-agent collaboration.
169pub const MANIFEST: MiddlewareManifest = MiddlewareManifest {
170    id: "subagents",
171    label: text::MANIFEST_LABEL,
172    description: text::MANIFEST_DESCRIPTION,
173    required: false,
174    default_enabled: true,
175    settings: SETTINGS,
176};
177
178/// Child-agent parameters owned by the subagent capability.
179#[derive(Clone)]
180pub struct SubagentLaunch {
181    pub session_id: String,
182    pub model: String,
183    pub reasoning_effort: Option<String>,
184    pub metadata: BTreeMap<String, Value>,
185    pub role: AgentRole,
186}
187
188/// Creates one child agent for this capability.
189pub type SubagentLauncher =
190    Arc<dyn Fn(SubagentLaunch) -> BoxFuture<'static, Result<Agent>> + Send + Sync>;
191
192#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
193enum ForkTurns {
194    #[default]
195    None,
196    All,
197    Last(usize),
198}
199
200impl ForkTurns {
201    fn label(self) -> String {
202        match self {
203            Self::None => "No context".into(),
204            Self::All => "Full context".into(),
205            Self::Last(1) => "Last 1 turn".into(),
206            Self::Last(turns) => format!("Last {turns} turns"),
207        }
208    }
209}
210
211#[derive(Clone, Deserialize)]
212#[serde(deny_unknown_fields)]
213struct AgentIdentity {
214    root_session_id: String,
215    agent_path: String,
216    depth: u8,
217}
218
219#[derive(Debug, Serialize, Deserialize)]
220#[serde(deny_unknown_fields)]
221struct PreviewCursor {
222    path: String,
223    before_sequence: u64,
224}
225
226impl AgentIdentity {
227    fn read(session_id: &str, metadata: &BTreeMap<String, Value>) -> Result<Self> {
228        let Some(value) = metadata.get(IDENTITY_KEY) else {
229            return Ok(Self {
230                root_session_id: session_id.into(),
231                agent_path: "/root".into(),
232                depth: 0,
233            });
234        };
235        Ok(serde_json::from_value(value.clone())?)
236    }
237
238    fn metadata(&self, mut metadata: BTreeMap<String, Value>) -> BTreeMap<String, Value> {
239        metadata.insert(
240            IDENTITY_KEY.into(),
241            serde_json::json!({
242                "root_session_id": self.root_session_id,
243                "agent_path": self.agent_path,
244                "depth": self.depth,
245            }),
246        );
247        metadata
248    }
249}
250
251#[derive(Clone)]
252struct AgentScope {
253    checkpoints: Arc<dyn CheckpointStore>,
254    launch_agent: SubagentLauncher,
255    session_id: String,
256    root_session_id: String,
257    agent_path: String,
258    depth: u8,
259    model: String,
260    metadata: BTreeMap<String, Value>,
261}
262
263impl AgentScope {
264    fn new(runtime: &RuntimeContext, launch_agent: SubagentLauncher) -> Result<Self> {
265        let identity = AgentIdentity::read(&runtime.session_id, &runtime.metadata)?;
266        Ok(Self {
267            checkpoints: Arc::clone(&runtime.checkpoints),
268            launch_agent,
269            session_id: runtime.session_id.clone(),
270            root_session_id: identity.root_session_id,
271            agent_path: identity.agent_path,
272            depth: identity.depth,
273            model: runtime.model_route.clone(),
274            metadata: runtime.metadata.clone(),
275        })
276    }
277
278    async fn fork(
279        &self,
280        session_id: String,
281        agent_path: String,
282        model: String,
283        reasoning_effort: Option<String>,
284        turns: ForkTurns,
285        parent_turn_id: String,
286    ) -> Result<Agent> {
287        let parent = self
288            .checkpoints
289            .load(&self.session_id)
290            .await?
291            .ok_or_else(|| Error::Checkpoint("parent checkpoint is missing".into()))?;
292        let parent_sequence = parent.sequence;
293        let pending = parent
294            .pending_tools
295            .iter()
296            .map(|call| call.call_id.clone())
297            .collect::<BTreeSet<_>>();
298        let context = parent
299            .context
300            .into_iter()
301            .filter(|item| {
302                item.get("type").and_then(Value::as_str) != Some("function_call")
303                    || item
304                        .get("call_id")
305                        .and_then(Value::as_str)
306                        .is_none_or(|call_id| !pending.contains(call_id))
307            })
308            .collect::<Vec<_>>();
309        let mut checkpoint = Checkpoint::empty(&session_id);
310        checkpoint.catalog_visible = false;
311        checkpoint.context = fork_context(&context, turns);
312        checkpoint.session_context = parent.session_context;
313        let mut metadata = AgentIdentity {
314            root_session_id: self.root_session_id.clone(),
315            agent_path: agent_path.clone(),
316            depth: self.depth + 1,
317        }
318        .metadata(self.metadata.clone());
319        metadata.insert(SPAWN_CONTEXT_KEY.into(), Value::String(turns.label()));
320        checkpoint.metadata.clone_from(&metadata);
321        self.checkpoints
322            .fork(&self.session_id, parent_sequence, &checkpoint)
323            .await?;
324        (self.launch_agent)(SubagentLaunch {
325            role: AgentRole::Subagent {
326                parent_session_id: self.session_id.clone(),
327                parent_turn_id,
328            },
329            session_id,
330            model,
331            reasoning_effort,
332            metadata,
333        })
334        .await
335    }
336
337    async fn resume(
338        &self,
339        session_id: String,
340        agent_path: String,
341        depth: u8,
342        model: String,
343        parent_turn_id: String,
344    ) -> Result<Agent> {
345        let checkpoint = self.checkpoints.load(&session_id).await?.ok_or_else(|| {
346            Error::Checkpoint(format!("checkpoint for `{agent_path}` is missing"))
347        })?;
348        (self.launch_agent)(SubagentLaunch {
349            role: AgentRole::Subagent {
350                parent_session_id: self.session_id.clone(),
351                parent_turn_id,
352            },
353            session_id,
354            model,
355            reasoning_effort: None,
356            metadata: AgentIdentity {
357                root_session_id: self.root_session_id.clone(),
358                agent_path,
359                depth,
360            }
361            .metadata(checkpoint.metadata),
362        })
363        .await
364    }
365}
366
367/// Contributes asynchronous collaboration tools.
368pub struct Subagents {
369    max_depth: u8,
370    launch_agent: SubagentLauncher,
371    default_model: Option<String>,
372    default_reasoning: Option<String>,
373    prompt: String,
374    shared: Arc<Shared>,
375}
376
377impl Subagents {
378    /// Creates a child-agent capability with hard depth, concurrency, and agent limits.
379    ///
380    /// `max_concurrency` counts active agents and `max_agents` counts retained agents;
381    /// both include the root.
382    pub fn new(
383        max_depth: u8,
384        max_concurrency: usize,
385        max_agents: usize,
386        launch_agent: SubagentLauncher,
387    ) -> Result<Self> {
388        if max_depth == 0 || max_depth > MAX_CONFIGURED_DEPTH {
389            return Err(Error::Config(format!(
390                "subagent max depth must be between 1 and {MAX_CONFIGURED_DEPTH}"
391            )));
392        }
393        if max_concurrency > MAX_CONFIGURED_CONCURRENCY {
394            return Err(Error::Config(format!(
395                "subagent max concurrency cannot exceed {MAX_CONFIGURED_CONCURRENCY}"
396            )));
397        }
398        if max_agents > MAX_CONFIGURED_AGENTS {
399            return Err(Error::Config(format!(
400                "subagent max agents cannot exceed {MAX_CONFIGURED_AGENTS}"
401            )));
402        }
403        Ok(Self {
404            max_depth,
405            launch_agent,
406            default_model: None,
407            default_reasoning: None,
408            prompt: text::PROMPT_DEFAULT.into(),
409            shared: Arc::new(Shared::new(max_concurrency, max_agents)?),
410        })
411    }
412
413    /// Reports whether this root session has a pending or running child agent.
414    pub async fn has_active_children(&self, root_session_id: &str) -> Result<bool> {
415        self.shared.has_active_children(root_session_id).await
416    }
417
418    /// Selects a registered provider/model route for children by default.
419    #[must_use]
420    pub fn default_model(mut self, model: impl Into<String>) -> Self {
421        self.default_model = Some(model.into());
422        self
423    }
424
425    /// Selects a reasoning effort for children by default.
426    pub fn default_reasoning(mut self, reasoning: impl Into<String>) -> Result<Self> {
427        let reasoning = reasoning.into();
428        if reasoning.trim().is_empty() {
429            return Err(Error::Config(
430                "subagent reasoning effort cannot be empty".into(),
431            ));
432        }
433        self.default_reasoning = Some(reasoning);
434        Ok(self)
435    }
436
437    /// Overrides the instruction given to child agents.
438    pub fn prompt(mut self, prompt: impl Into<String>) -> Result<Self> {
439        let prompt = prompt.into();
440        if prompt.trim().is_empty() {
441            return Err(Error::Config("subagent prompt cannot be empty".into()));
442        }
443        self.prompt = prompt;
444        Ok(self)
445    }
446
447    fn section(&self, identity: &AgentIdentity) -> PromptSection {
448        let body = if identity.depth == 0 {
449            text::PROMPT_ROOT.into()
450        } else {
451            format!(
452                "You are `{}`, a child agent.\n{}",
453                identity.agent_path,
454                self.prompt.trim()
455            )
456        };
457        PromptSection::new(body)
458    }
459
460    async fn read_command(
461        &self,
462        session_id: &str,
463        metadata: &BTreeMap<String, Value>,
464        arguments: &str,
465    ) -> Result<MiddlewareCommandOutput> {
466        let path = arguments.trim();
467        if path.starts_with('{') {
468            return self.read_preview_page(session_id, metadata, path).await;
469        }
470        let identity = AgentIdentity::read(session_id, metadata)?;
471        if !path.is_empty() {
472            return self
473                .preview_page(&identity.root_session_id, path, None)
474                .await;
475        }
476        let options = self
477            .shared
478            .resume_options(&identity.root_session_id)
479            .await?;
480        if options.is_empty() {
481            return Ok(MiddlewareCommandOutput::events(vec![
482                FrontendEvent::Picker {
483                    title: format!("{} · {}", text::RENDER_OPEN, text::RENDER_EMPTY),
484                    options,
485                },
486            ]));
487        }
488        Ok(MiddlewareCommandOutput::events(vec![
489            FrontendEvent::Picker {
490                title: text::RENDER_OPEN.into(),
491                options,
492            },
493        ]))
494    }
495
496    async fn read_preview_page(
497        &self,
498        session_id: &str,
499        metadata: &BTreeMap<String, Value>,
500        arguments: &str,
501    ) -> Result<MiddlewareCommandOutput> {
502        let identity = AgentIdentity::read(session_id, metadata)?;
503        let cursor: PreviewCursor = serde_json::from_str(arguments)
504            .map_err(|_| Error::Tool("invalid subagent preview cursor".into()))?;
505        if cursor.path.trim() != cursor.path
506            || cursor.path.is_empty()
507            || cursor.before_sequence == 0
508        {
509            return Err(Error::Tool("invalid subagent preview cursor".into()));
510        }
511        self.preview_page(
512            &identity.root_session_id,
513            &cursor.path,
514            Some(cursor.before_sequence),
515        )
516        .await
517    }
518
519    async fn preview_page(
520        &self,
521        root_session_id: &str,
522        path: &str,
523        before_sequence: Option<u64>,
524    ) -> Result<MiddlewareCommandOutput> {
525        let page = self
526            .shared
527            .preview(root_session_id, path, before_sequence)
528            .await?;
529        let next = page
530            .next
531            .map(|before_sequence| -> Result<Op> {
532                Ok(Op::CapabilityCommand {
533                    capability: MANIFEST.id.into(),
534                    command: "subagents".into(),
535                    arguments: serde_json::to_string(&PreviewCursor {
536                        path: path.into(),
537                        before_sequence,
538                    })?,
539                    input: None,
540                    target: None,
541                })
542            })
543            .transpose()?;
544        Ok(MiddlewareCommandOutput::events(vec![
545            FrontendEvent::Preview {
546                id: path.into(),
547                title: path.rsplit('/').next().unwrap_or(path).into(),
548                subtitle: page.subtitle,
549                page_id: page.page_id,
550                update: if before_sequence.is_some() {
551                    FrontendPreviewUpdate::Prepend
552                } else {
553                    FrontendPreviewUpdate::Replace
554                },
555                events: page.events,
556                next,
557            },
558        ]))
559    }
560}
561
562impl Middleware for Subagents {
563    fn name(&self) -> &'static str {
564        MANIFEST.id
565    }
566
567    fn session_start<'a>(
568        &'a self,
569        context: &'a mut SessionStartContext<'_>,
570    ) -> BoxFuture<'a, Result<()>> {
571        if context.source() == SessionStartSource::Compact {
572            return Box::pin(async { Ok(()) });
573        }
574        Box::pin(self.shared.session_start((*context.runtime).clone()))
575    }
576
577    fn register(&self, catalog: &mut Catalog, runtime: &RuntimeContext) -> Result<()> {
578        let scope = Arc::new(AgentScope::new(runtime, Arc::clone(&self.launch_agent))?);
579        if scope.depth < self.max_depth {
580            catalog.register(Arc::new(SpawnAgent {
581                default_model: self.default_model.clone(),
582                default_reasoning: self.default_reasoning.clone(),
583                shared: Arc::clone(&self.shared),
584                scope: Arc::clone(&scope),
585            }))?;
586        }
587        catalog.register(Arc::new(SendMessage {
588            shared: Arc::clone(&self.shared),
589            scope: Arc::clone(&scope),
590        }))?;
591        catalog.register(Arc::new(ListAgents {
592            shared: Arc::clone(&self.shared),
593            scope: Arc::clone(&scope),
594        }))?;
595        catalog.register(Arc::new(InterruptAgent {
596            shared: Arc::clone(&self.shared),
597            scope: Arc::clone(&scope),
598        }))?;
599        catalog.register(Arc::new(WaitAgent {
600            shared: Arc::clone(&self.shared),
601            scope,
602        }))
603    }
604
605    fn prompt_section(&self, runtime: &RuntimeContext) -> Result<Option<PromptSection>> {
606        let identity = AgentIdentity::read(&runtime.session_id, &runtime.metadata)?;
607        Ok(Some(self.section(&identity)))
608    }
609
610    fn frontend(&self) -> FrontendContribution {
611        FrontendContribution {
612            capability: self.name().into(),
613            accepts_file_attachments: false,
614            count: None,
615            commands: vec![FrontendCommand {
616                name: "subagents".into(),
617                arguments: String::new(),
618                description: text::COMMAND_DESCRIPTION.into(),
619                requires_idle: false,
620            }],
621            widgets: Vec::new(),
622            references: Vec::new(),
623        }
624    }
625
626    fn render(&self, event: &EventMsg, _session_id: &str) -> Option<FrontendBlock> {
627        let mut block = render_tool_event(
628            event,
629            |name| {
630                matches!(
631                    name,
632                    "spawn_agent"
633                        | "send_message"
634                        | "list_agents"
635                        | "interrupt_agent"
636                        | "wait_agent"
637                )
638            },
639            |name, arguments| match name {
640                _ if matches!(event, EventMsg::ToolCallEnd(_)) => name.into(),
641                "spawn_agent" => labeled_tool_heading(text::RENDER_AGENT, "task_name", arguments),
642                "send_message" => labeled_tool_heading(text::RENDER_MESSAGE, "target", arguments),
643                "list_agents" => {
644                    labeled_tool_heading(text::RENDER_AGENTS, "path_prefix", arguments)
645                }
646                "interrupt_agent" => {
647                    labeled_tool_heading(text::RENDER_INTERRUPT, "target", arguments)
648                }
649                "wait_agent" => labeled_tool_heading(text::RENDER_WAIT, "timeout_ms", arguments),
650                _ => name.to_string().into(),
651            },
652        )?;
653        if let EventMsg::ToolCallBegin(call) = event
654            && call.name == "send_message"
655            && let Some(message) = call.arguments.get("text").and_then(Value::as_str)
656        {
657            FrontendBlockUpdate::Append.apply(&mut block.text, message);
658        }
659        Some(block)
660    }
661
662    fn command<'a>(
663        &'a self,
664        context: MiddlewareCommandContext<'a>,
665    ) -> BoxFuture<'a, Result<MiddlewareCommandOutput>> {
666        Box::pin(async move {
667            match context.command {
668                "subagents" => {
669                    self.read_command(
670                        context.session_id,
671                        &context.checkpoint.metadata,
672                        context.arguments,
673                    )
674                    .await
675                }
676                command => Err(Error::Unknown(format!("subagents command `{command}`"))),
677            }
678        })
679    }
680
681    fn active_command<'a>(
682        &'a self,
683        context: &'a mut ActiveCommandContext<'_>,
684    ) -> BoxFuture<'a, Result<Option<SubmissionResult>>> {
685        Box::pin(async move {
686            let output = match context.command {
687                "subagents" => {
688                    self.read_command(context.session_id, context.metadata, context.arguments)
689                        .await
690                }
691                _ => return Ok(None),
692            };
693            match output {
694                Ok(output) => {
695                    context
696                        .events
697                        .extend(output.events.into_iter().map(EventMsg::Frontend));
698                    Ok(Some(SubmissionResult::Handled))
699                }
700                Err(error) => Ok(Some(SubmissionResult::Rejected(error.to_string()))),
701            }
702        })
703    }
704
705    fn pre_model<'a>(&'a self, context: &'a mut ModelContext<'_>) -> BoxFuture<'a, Result<()>> {
706        Box::pin(async move {
707            let identity = AgentIdentity::read(context.session_id, context.metadata)?;
708            let acknowledged = context
709                .input()
710                .iter()
711                .filter_map(internal_message_kind)
712                .filter_map(|kind| kind.strip_prefix("subagent_update:"))
713                .map(str::to_owned)
714                .collect();
715            let delivered_message_ids = context
716                .input()
717                .iter()
718                .filter_map(message_metadata)
719                .filter_map(|message| match message.author {
720                    MessageAuthor::Peer { message_id, .. } => Some(message_id),
721                    MessageAuthor::User => None,
722                })
723                .collect();
724            let updates = self
725                .shared
726                .receive_updates(
727                    &identity.root_session_id,
728                    &identity.agent_path,
729                    &acknowledged,
730                )
731                .await?;
732            for update in updates {
733                context.push_input(internal_user_message(
734                    &update.internal_kind(),
735                    &update.render(&delivered_message_ids),
736                ))?;
737            }
738            Ok(())
739        })
740    }
741
742    fn session_end<'a>(&'a self, runtime: &'a RuntimeContext) -> BoxFuture<'a, Result<()>> {
743        Box::pin(async move {
744            let identity = AgentIdentity::read(&runtime.session_id, &runtime.metadata)?;
745            if matches!(runtime.role, AgentRole::Main) && identity.depth == 0 {
746                self.shared.remove_root(&identity.root_session_id).await?;
747            } else {
748                self.shared
749                    .remove_sender(&identity.root_session_id, &identity.agent_path)
750                    .await;
751            }
752            Ok(())
753        })
754    }
755}
756
757#[cfg(test)]
758mod tests;
mobius/middleware/subagents.rs

mobius/middleware/
subagents.rs
