use super::{
activity::{SubagentActivitySink, emit_activity, subagent_task_activity_metadata},
config::SubagentRunConfig,
dispatch_subagents,
dto::{SubagentStatus, SubagentTask, SubagentTaskResult, SubagentsOutput},
output_schema::{self, SchemaValidationError, SubagentOutputSchemaRef},
profiles,
scheduler::{SchedulerTaskReporter, TaskActivityFinisher},
};
use crate::{
agent::cancellation::AgentCancellation,
agent::{AgentRunOutput, AgentRunRequest, AgentSession},
hooks::HookPolicyError,
output::{ActivityEvent, ActivityId, ActivityKind, ActivityStatus, redact_sensitive_text},
providers::{Provider, ProviderSelection},
sessions::{SessionEventKind, SessionManager},
tools::ToolResult,
};
use serde_json::{Value, json};
use std::{
collections::BTreeSet,
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
#[cfg(test)]
pub(super) const SUBAGENT_PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT: Duration =
Duration::from_millis(40);
pub(super) const SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT: usize = 24_000;
pub(super) const SUBAGENT_RESULT_ERROR_CHAR_LIMIT: usize = 8_000;
pub(super) const SUBAGENT_TRUNCATION_MARKER: &str = "\n[subagent result truncated: original_chars=";
pub(super) struct SubagentRunInput<'a> {
pub(super) id: String,
pub(super) task: SubagentTask,
pub(super) cwd: PathBuf,
pub(super) config: &'a SubagentRunConfig,
pub(super) cancellation: AgentCancellation,
pub(super) batch_id: &'a ActivityId,
pub(super) reporter: Option<SchedulerTaskReporter>,
pub(super) finisher: TaskActivityFinisher,
}
pub(super) fn run_one_subagent(input: SubagentRunInput<'_>) -> SubagentTaskResult {
let SubagentRunInput {
id,
task,
cwd,
config,
cancellation,
batch_id,
reporter,
finisher,
} = input;
let task_activity_id = ActivityId::new(format!("{}/{}", batch_id.as_str(), id));
emit_activity(
config,
ActivityEvent::Started {
id: task_activity_id.clone(),
parent_id: Some(batch_id.clone()),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Running,
metadata: subagent_task_activity_metadata(&id, &task, config.depth + 1),
},
);
let child_run = match child_agent_for_task(&task, &cwd, config) {
Ok(run) => run,
Err(error) => {
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: error.to_string(),
finisher: &finisher,
session: None,
});
}
};
let session = match config.sessions_root.as_ref() {
Some(root) => match SessionManager::new(root.join("subagents")).create() {
Ok(session) => Some(session),
Err(error) => {
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: format!("subagent session creation failed: {error}"),
finisher: &finisher,
session: None,
});
}
},
None => None,
};
if let (Some(reporter), Some(session)) = (&reporter, &session) {
reporter.session(session.id().to_string(), session.path().to_path_buf());
reporter.progress();
}
let output_schema = resolve_output_schema_for_task(&task, config);
let prompt = subagent_prompt(&id, &task, output_schema.as_ref());
let mut tools = match config
.parent_tools
.clone_for_cwd_with_subagent_depth(&cwd, config.depth + 1)
{
Ok(tools) => tools,
Err(error) => {
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: error.to_string(),
finisher: &finisher,
session: session.as_ref(),
});
}
};
let request_agent = if tools.subagents_schema_enabled()
&& let Some(subagent_profiles_prompt) = &config.subagent_profiles_prompt
{
child_run
.agent
.with_appended_system_prompt(subagent_profiles_prompt)
} else {
child_run.agent.clone()
};
let child_config = SubagentRunConfig {
parent_agent: child_run.agent.clone(),
provider: Arc::clone(&child_run.provider),
provider_override: config.provider_override.clone(),
parent_tools: tools.clone(),
parent_cwd: cwd.clone(),
cancellation: cancellation.clone(),
profiles: config.profiles.clone(),
subagent_profiles_prompt: if tools.subagents_schema_enabled() {
config.subagent_profiles_prompt.clone()
} else {
None
},
sessions_root: config.sessions_root.clone(),
depth: config.depth + 1,
parent_activity_id: Some(task_activity_id.clone()),
activity_sender: config.activity_sender.clone(),
inherited_hooks: config.inherited_hooks.clone(),
semantic_progress_timeout: config.semantic_progress_timeout,
schema_validation_max_retries: config.schema_validation_max_retries,
};
tools = tools.with_subagents(move |arguments, context| {
let mut config = child_config.clone();
config.parent_activity_id = context.parent_activity_id;
config.activity_sender = context.activity_sender;
dispatch_subagents(arguments, config)
});
let child_hooks = match config.inherited_hooks.as_ref() {
Some(hooks) => match hooks.clone_for_cwd(&cwd) {
Ok(hooks) => Some(hooks),
Err(error) => {
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: format!("subagent inherited hook setup failed: {error}"),
finisher: &finisher,
session: session.as_ref(),
});
}
},
None => None,
};
let mut child_sink = SubagentActivitySink {
parent_id: task_activity_id.clone(),
activity_sender: config.activity_sender.clone(),
assistant_id: ActivityId::new(format!("{}/assistant", task_activity_id.as_str())),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: reporter
.map(|reporter| Arc::new(move || reporter.progress()) as Arc<dyn Fn() + Send + Sync>),
cancellation: cancellation.clone(),
};
let max_schema_retries = config.schema_validation_max_retries.min(5);
let mut attempt = 0_u64;
let mut prompt = prompt;
loop {
let run_result = request_agent.run_print_with_tools_streaming_output_cancellable(
child_run.provider.as_ref(),
AgentRunRequest {
prompt: &prompt,
prompt_origin: crate::output::UserPromptOrigin::User,
effective_prompt: None,
tools: Some(&tools),
hooks: child_hooks.as_ref(),
session: session.as_ref(),
cwd: &cwd,
output_sink: Some(&mut child_sink),
cancellation: cancellation.clone(),
session_title_job: None,
semantic_progress_timeout: Some(config.semantic_progress_timeout),
invocation_mode: crate::output::InvocationMode::Subagent,
agent_id: task.identity.clone().or_else(|| task.agent.clone()),
herdr_reporter: None,
initial_instructions: &[],
ttsr: crate::config::TtsrSettings::default(),
continuation_auto_compaction_policy: None,
},
);
match run_result {
Ok(output) => {
if let Some(schema) = &output_schema {
match validate_schema_bound_output(schema, &output.text) {
Ok(structured_output) => {
child_sink.finish_assistant(ActivityStatus::Success);
finisher.finish(config, task_activity_id, ActivityStatus::Success);
return completed_subagent_result(
id,
task,
cwd,
session.as_ref(),
output,
Some(structured_output),
);
}
Err(errors) if attempt < max_schema_retries => {
attempt = attempt.saturating_add(1);
prompt = schema_validation_feedback_prompt(
attempt,
max_schema_retries,
&errors,
);
continue;
}
Err(errors) => {
child_sink.finish_assistant(ActivityStatus::Failed);
let result = failed_result_with_session(
id,
task,
cwd,
session.as_ref().map(|session| session.id().to_string()),
session.as_ref().map(|session| session.path().to_path_buf()),
schema_validation_exhausted_error(&errors),
);
finisher.finish(config, task_activity_id, ActivityStatus::Failed);
return result;
}
}
}
child_sink.finish_assistant(ActivityStatus::Success);
finisher.finish(config, task_activity_id, ActivityStatus::Success);
return completed_subagent_result(id, task, cwd, session.as_ref(), output, None);
}
Err(error) => {
let finish_status = if cancellation.is_canceled() {
ActivityStatus::Canceled
} else {
ActivityStatus::Failed
};
child_sink.finish_assistant(finish_status);
let error = error.downcast_ref::<HookPolicyError>().map_or_else(
|| error.to_string(),
|hook_error| hook_error.parent_safe_summary().to_string(),
);
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: error.to_string(),
finisher: &finisher,
session: session.as_ref(),
});
}
}
}
}
pub(super) struct ChildAgentRun {
pub(super) agent: AgentSession,
pub(super) provider: Arc<dyn Provider>,
}
pub(super) fn child_agent_for_task(
task: &SubagentTask,
cwd: &Path,
config: &SubagentRunConfig,
) -> anyhow::Result<ChildAgentRun> {
let Some(identity) = task.identity.as_deref() else {
return Ok(ChildAgentRun {
agent: config.parent_agent.clone(),
provider: Arc::clone(&config.provider),
});
};
let identity = profiles::validate_subagent_identity_id(identity)?;
let profile = config.profiles.get(&identity).ok_or_else(|| {
anyhow::anyhow!(
"subagent identity '{identity}' is unavailable: no matching profile found under $MC_HOME/subagents"
)
})?;
let agent = config
.parent_agent
.with_appended_system_prompt(&profile.prompt);
child_agent_with_profile_overrides(agent, profile, cwd, config)
}
pub(super) fn child_agent_with_profile_overrides(
mut agent: AgentSession,
profile: &profiles::SubagentProfile,
cwd: &Path,
config: &SubagentRunConfig,
) -> anyhow::Result<ChildAgentRun> {
let mut provider = Arc::clone(&config.provider);
if let Some(model) = &profile.model {
let override_context = config.provider_override.as_ref().ok_or_else(|| {
anyhow::anyhow!(
"subagent provider/model overrides are unavailable in this execution context"
)
})?;
let selection = ProviderSelection {
provider: model.provider.clone(),
model: model.model.clone(),
};
let resolved = (override_context.resolve_provider)(&selection, cwd)?;
provider = resolved.provider;
let settings = crate::config::read_settings(&override_context.paths)?;
let thinking_levels =
subagent_model_thinking_levels(&override_context.paths, &selection, resolved.scope);
agent = agent
.with_provider_model_reasoning(
selection.provider.clone(),
selection.model,
thinking_levels,
profile.reasoning,
)
.with_text_verbosity(settings.text_verbosity_for(&selection.provider));
} else if let Some(reasoning) = profile.reasoning {
agent = agent.with_reasoning_override(reasoning);
}
Ok(ChildAgentRun { agent, provider })
}
pub(super) fn subagent_model_thinking_levels(
paths: &crate::config::McPaths,
selection: &ProviderSelection,
scope: crate::thinking::ThinkingCapabilityScope,
) -> Vec<crate::thinking::ThinkingLevel> {
let cached = crate::model_catalog::cached_model_thinking_metadata(
paths,
&selection.provider,
&selection.model,
);
crate::thinking::available_thinking_levels(
&selection.provider,
&selection.model,
cached.as_ref(),
scope,
)
}
pub(super) fn subagent_prompt(
id: &str,
task: &SubagentTask,
output_schema: Option<&Value>,
) -> String {
let mut prompt = format!("Subagent {id} task intent:\n{}\n", task.intent);
if let Some(agent) = &task.agent {
prompt.push_str(&format!("\nAgent label/persona: {agent}\n"));
}
if let Some(context) = &task.context {
prompt.push_str("\nTask context:\n");
prompt.push_str(context);
prompt.push('\n');
}
if let Some(schema) = output_schema {
let schema_text = serde_json::to_string(schema).unwrap_or_else(|_| "{}".to_string());
prompt.push_str("\nOutput schema contract:\nReturn only a raw JSON object matching this JSON Schema. Do not include markdown, code fences, or prose before or after the JSON object.\n");
prompt.push_str(&schema_text);
prompt.push('\n');
}
prompt
}
fn resolve_output_schema_for_task(
task: &SubagentTask,
config: &SubagentRunConfig,
) -> Option<Value> {
let identity = task.identity.as_deref()?;
let schema_ref = config
.profiles
.get(identity)
.and_then(|profile| profile.output_schema.as_ref())
.cloned()
.or_else(|| {
output_schema::default_phase_for_identity(identity)
.map(SubagentOutputSchemaRef::Builtin)
})?;
output_schema::schema_for_ref(&schema_ref)
}
fn validate_schema_bound_output(
schema: &Value,
output_text: &str,
) -> Result<Value, Vec<SchemaValidationError>> {
let trimmed = output_text.trim();
let instance = serde_json::from_str::<Value>(trimmed).map_err(|error| {
vec![SchemaValidationError {
instance_path: String::new(),
schema_path: String::new(),
message: format!("output must be one JSON object: {error}"),
}]
})?;
if !instance.is_object() {
return Err(vec![SchemaValidationError {
instance_path: String::new(),
schema_path: String::new(),
message: "output must be one JSON object".to_string(),
}]);
}
output_schema::validate_instance(schema, &instance)?;
Ok(instance)
}
fn schema_validation_feedback_prompt(
attempt: u64,
max_retries: u64,
errors: &[SchemaValidationError],
) -> String {
json!({
"schema_validation_error": true,
"attempt": attempt,
"max_retries": max_retries,
"errors": redact_schema_validation_errors(errors),
"instruction": "Return only a JSON object matching the required output_schema. Do not include markdown or prose."
})
.to_string()
}
fn schema_validation_exhausted_error(errors: &[SchemaValidationError]) -> String {
json!({
"schema_validation_error": true,
"exhausted": true,
"errors": redact_schema_validation_errors(errors),
"instruction": "Subagent did not return JSON matching the required output_schema before retry exhaustion."
})
.to_string()
}
fn redact_schema_validation_errors(errors: &[SchemaValidationError]) -> Vec<SchemaValidationError> {
errors
.iter()
.map(|error| SchemaValidationError {
instance_path: redact_sensitive_text(&error.instance_path),
schema_path: redact_sensitive_text(&error.schema_path),
message: redact_sensitive_text(&error.message),
})
.collect()
}
fn completed_subagent_result(
id: String,
task: SubagentTask,
cwd: PathBuf,
session: Option<&crate::sessions::Session>,
output: AgentRunOutput,
structured_output: Option<Value>,
) -> SubagentTaskResult {
let mut output_text = output.text;
let output_truncated =
truncate_string_field(&mut output_text, SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT);
SubagentTaskResult {
id,
status: SubagentStatus::Completed,
intent: task.intent,
agent: task.agent,
identity: task.identity,
cwd,
session_id: session.map(|session| session.id().to_string()),
session_path: session.map(|session| session.path().to_path_buf()),
total_tokens: output.total_tokens,
changed_files: changed_files_from_tool_results(&output.tool_results),
output: output_text,
structured_output,
output_truncated,
error: None,
}
}
pub(super) struct FailedSubagentInput<'a> {
config: &'a SubagentRunConfig,
task_activity_id: ActivityId,
id: String,
task: SubagentTask,
cwd: PathBuf,
error: String,
finisher: &'a TaskActivityFinisher,
session: Option<&'a crate::sessions::Session>,
}
pub(super) fn finish_failed_subagent(input: FailedSubagentInput<'_>) -> SubagentTaskResult {
let result = failed_result_with_session_and_output(
input.id,
input.task,
input.cwd,
input.session.map(|session| session.id().to_string()),
input.session.map(|session| session.path().to_path_buf()),
input.error,
failed_subagent_session_snapshot(input.session),
);
input
.finisher
.finish(input.config, input.task_activity_id, ActivityStatus::Failed);
result
}
#[cfg(test)]
pub(super) fn failed_result(
id: String,
task: SubagentTask,
cwd: PathBuf,
error: String,
) -> SubagentTaskResult {
failed_result_with_session(id, task, cwd, None, None, error)
}
pub(super) fn failed_result_with_session(
id: String,
task: SubagentTask,
cwd: PathBuf,
session_id: Option<String>,
session_path: Option<PathBuf>,
error: String,
) -> SubagentTaskResult {
failed_result_with_session_and_output(id, task, cwd, session_id, session_path, error, None)
}
pub(super) fn failed_result_with_session_and_output(
id: String,
task: SubagentTask,
cwd: PathBuf,
session_id: Option<String>,
session_path: Option<PathBuf>,
error: String,
output: Option<String>,
) -> SubagentTaskResult {
SubagentTaskResult {
id,
status: SubagentStatus::Failed,
intent: task.intent,
agent: task.agent,
identity: task.identity,
cwd,
session_id,
session_path,
total_tokens: None,
changed_files: Vec::new(),
output: output.unwrap_or_default(),
structured_output: None,
output_truncated: false,
error: Some(redact_sensitive_text(&error)),
}
}
pub(super) fn format_duration(duration: Duration) -> String {
format!("{:.3}s", duration.as_secs_f64())
}
pub(super) fn failed_subagent_session_snapshot(
session: Option<&crate::sessions::Session>,
) -> Option<String> {
const SNAPSHOT_EVENT_LIMIT: usize = 200;
const SNAPSHOT_BYTE_LIMIT: usize = 256 * 1024;
let session = session?;
let events = session
.read_recent_events_tolerant(SNAPSHOT_EVENT_LIMIT, SNAPSHOT_BYTE_LIMIT)
.ok()?
.events;
let mut assistant_chunks = String::new();
let mut authoritative_assistant_output = None;
let mut recovery = Vec::new();
for event in events {
match event.kind() {
Some(SessionEventKind::AssistantChunk) => {
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
assistant_chunks.push_str(text);
}
}
Some(SessionEventKind::AssistantOutput) => {
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
authoritative_assistant_output = Some(text.to_string());
}
}
Some(SessionEventKind::ReasoningSummary) => {
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
recovery.push(format!("reasoning summary: {text}"));
}
}
Some(SessionEventKind::AbortRecovery) => {
recovery.push(format!("abort recovery: {}", event.payload));
}
_ => {}
}
}
let assistant = authoritative_assistant_output
.filter(|text| !text.trim().is_empty())
.unwrap_or(assistant_chunks);
let mut snapshot = String::new();
if !assistant.trim().is_empty() {
snapshot.push_str("Partial subagent output before failure:\n");
snapshot.push_str(assistant.trim());
snapshot.push('\n');
}
for item in recovery {
if !snapshot.is_empty() {
snapshot.push('\n');
}
snapshot.push_str(&item);
}
if snapshot.trim().is_empty() {
None
} else {
truncate_string_field(&mut snapshot, SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT);
Some(snapshot)
}
}
pub(super) fn changed_files_from_tool_results(tool_results: &[ToolResult]) -> Vec<PathBuf> {
tool_results
.iter()
.filter(|result| {
result.success && matches!(result.tool_name.as_str(), "write" | "hash_edit")
})
.filter_map(|result| result.metadata.get("path").and_then(Value::as_str))
.map(PathBuf::from)
.collect::<BTreeSet<_>>()
.into_iter()
.collect()
}
pub(super) fn truncate_subagents_output(output: &mut SubagentsOutput) {
for result in &mut output.results {
let output_was_truncated =
truncate_string_field(&mut result.output, SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT);
let error_was_truncated = result
.error
.as_mut()
.is_some_and(|error| truncate_string_field(error, SUBAGENT_RESULT_ERROR_CHAR_LIMIT));
result.output_truncated |= output_was_truncated || error_was_truncated;
}
}
pub(super) fn truncate_string_field(value: &mut String, char_limit: usize) -> bool {
let char_count = value.chars().count();
if char_count <= char_limit {
return false;
}
if char_limit == 0 {
value.clear();
return true;
}
let marker = format!("{SUBAGENT_TRUNCATION_MARKER}{char_count} limit={char_limit}]");
let marker_len = marker.chars().count();
if marker_len >= char_limit {
*value = marker.chars().take(char_limit).collect();
return true;
}
let keep = char_limit - marker_len;
let mut truncated = value.chars().take(keep).collect::<String>();
truncated.push_str(&marker);
*value = truncated;
true
}