use super::{
activity::{SubagentActivitySink, 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::{
AgentOutputSink, AgentRunOutput, AgentRunRequest, AgentSession, ContextBudgetError,
ContextBudgetPhase,
},
cancellation::AgentCancellation,
config::AutoCompactionLimit,
hooks::{HookPolicyError, HookRuntime},
output::{
ActivityEvent, ActivityId, ActivityKind, ActivityStatus, OutputEvent, redact_sensitive_text,
},
providers::{Provider, ProviderSelection},
sessions::{Session, SessionEventKind, SessionManager},
tools::ToolRuntime,
};
use serde::Serialize;
use serde_json::{Value, json};
use std::{
collections::BTreeSet,
path::{Path, PathBuf},
sync::{Arc, atomic::AtomicU64},
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_PROVIDER_RETRY_MAX_ATTEMPTS: u64 = 1;
pub(super) const SUBAGENT_PROVIDER_RETRY_BACKOFF: Duration = Duration::from_secs(2);
const SUBAGENT_PROVIDER_RETRY_POLL_INTERVAL: Duration = Duration::from_millis(100);
pub(super) const SUBAGENT_RESULT_ERROR_CHAR_LIMIT: usize = 8_000;
pub(super) const SUBAGENT_TRUNCATION_MARKER: &str = "\n[subagent result truncated: original_chars=";
const SNAPSHOT_EVENT_LIMIT: usize = 200;
pub(super) const SNAPSHOT_BYTE_LIMIT: usize = 256 * 1024;
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 = batch_id.child(&id);
finisher.try_start(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, None),
});
let child_run = match child_agent_for_task(&task, &cwd, config) {
Ok(run) => {
finisher.try_enrich(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, Some(&run)),
});
run
}
Err(error) => {
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: error.to_string(),
finisher: &finisher,
session: None,
partial_output: None,
usage: None,
});
}
};
let session = match config.sessions_root.as_ref() {
Some(root) => match SessionManager::new(root.join("subagents"))
.create()
.and_then(crate::sessions::Session::activate)
{
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,
partial_output: None,
usage: None,
});
}
},
None => None,
};
if let (Some(root), Some(parent_id), Some(child)) =
(&config.sessions_root, &config.parent_session_id, &session)
{
let parent_root = if config.depth == 0 {
root.clone()
} else {
root.join("subagents")
};
let linked = SessionManager::new(parent_root)
.open_existing(parent_id.clone())
.and_then(|parent| {
crate::sessions::record_session_event(
Some(&parent),
&config.parent_cwd,
SessionEventKind::SubagentSession,
json!({"session_id": child.id()}),
)
});
if let Err(error) = linked {
return finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error: format!("subagent session linkage failed: {error}"),
finisher: &finisher,
session: session.as_ref(),
partial_output: None,
usage: 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 profile_disabled_tools = task
.identity
.as_deref()
.and_then(|identity| config.profiles.get(identity))
.map(|profile| &profile.disabled_tools)
.filter(|disabled_tools| !disabled_tools.is_empty());
let cloned_tools = match profile_disabled_tools {
Some(disabled_tools) => config
.parent_tools
.clone_for_cwd_with_subagent_depth_and_additional_disabled_tools(
&cwd,
config.depth + 1,
disabled_tools,
),
None => config
.parent_tools
.clone_for_cwd_with_subagent_depth(&cwd, config.depth + 1),
};
let mut tools = match cloned_tools {
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(),
partial_output: None,
usage: None,
});
}
};
let child_agent = child_run.agent.clone();
let request_agent = if tools.subagents_schema_enabled()
&& let Some(subagent_profiles_prompt) = &config.subagent_profiles_prompt
{
child_agent.with_appended_system_prompt(subagent_profiles_prompt)
} else {
child_agent.clone()
};
let child_compaction = child_run.compaction.clone();
let child_config = SubagentRunConfig {
parent_agent: child_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(),
parent_session_id: None,
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,
compaction: child_compaction.clone(),
};
tools = tools.with_subagents(move |arguments, context| {
let mut config = child_config.clone();
config.parent_session_id = context.hook_context.session_id;
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(),
partial_output: None,
usage: None,
});
}
},
None => None,
};
let request_sequence = Arc::new(AtomicU64::new(0));
let mut child_sink = SubagentActivitySink {
parent_id: task_activity_id.clone(),
activity_sender: config.activity_sender.clone(),
assistant_id: task_activity_id.child("assistant"),
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(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: super::activity::UsageAccumulator::default(),
active_compaction: None,
};
let max_schema_retries = config.schema_validation_max_retries.min(5);
let mut attempt = 0_u64;
let mut provider_retry_attempt = 0_u64;
let mut auto_compaction_state = ChildAutoCompactionState::default();
let mut prompt = prompt;
let mut result = loop {
let run_result = run_child_turn_with_auto_compaction(ChildTurnRun {
agent: &request_agent,
original_task: &task,
provider: child_run.provider.as_ref(),
prompt: &prompt,
tools: &tools,
hooks: child_hooks.as_ref(),
session: session.as_ref(),
cwd: &cwd,
sink: &mut child_sink,
cancellation: &cancellation,
semantic_progress_timeout: config.semantic_progress_timeout,
agent_id: task.identity.clone().or_else(|| task.agent.clone()),
request_sequence: Arc::clone(&request_sequence),
compaction: child_compaction.as_ref(),
auto_compaction_state: &mut auto_compaction_state,
});
match run_result {
Ok(mut output) => {
output.total_tokens = child_sink.usage_by_request.totals.total();
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);
break completed_subagent_result(
id,
task,
cwd,
session.as_ref(),
output,
Some(structured_output),
child_sink.usage_metrics(),
);
}
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 mut result = failed_result_with_session_and_usage(
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),
child_sink.usage_metrics(),
);
result.total_tokens = child_sink.usage_by_request.totals.total();
finisher.finish(config, task_activity_id, ActivityStatus::Failed);
break result;
}
}
}
child_sink.finish_assistant(ActivityStatus::Success);
finisher.finish(config, task_activity_id, ActivityStatus::Success);
break completed_subagent_result(
id,
task,
cwd,
session.as_ref(),
output,
None,
child_sink.usage_metrics(),
);
}
Err(error) => {
let (error, partial_output) = match error.downcast::<ChildCompactionFailure>() {
Ok(error) => error.into_parts(),
Err(error) => (error, None),
};
let retryable = crate::providers::error::retryable_provider_error(&error);
let can_retry = can_retry_subagent_provider(
provider_retry_attempt,
&cancellation,
retryable,
child_sink.has_filesystem_changes(),
auto_compaction_state.checkpoint_committed,
);
if can_retry && wait_for_provider_retry(&cancellation) {
provider_retry_attempt = provider_retry_attempt.saturating_add(1);
continue;
}
let finish_status = if cancellation.is_canceled() {
ActivityStatus::Canceled
} else {
ActivityStatus::Failed
};
let _ = child_sink.finish_compaction(finish_status);
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(),
);
let error = if retryable
&& provider_retry_attempt >= SUBAGENT_PROVIDER_RETRY_MAX_ATTEMPTS
{
format!(
"[provider retry exhausted after {} attempts] {error}",
provider_retry_attempt.saturating_add(1)
)
} else {
error
};
let mut result = finish_failed_subagent(FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error,
finisher: &finisher,
session: session.as_ref(),
partial_output,
usage: child_sink.usage_metrics(),
});
result.total_tokens = child_sink.usage_by_request.totals.total();
break result;
}
}
};
result.changed_files = child_sink.changed_files.into_iter().collect();
result
}
#[derive(Default)]
struct ChildAutoCompactionState {
count: usize,
checkpoint_committed: bool,
}
#[derive(Debug)]
struct BoundedPartialOutput {
text: String,
truncated: bool,
}
#[derive(Debug)]
struct ChildCompactionFailure {
error: anyhow::Error,
partial_output: Option<BoundedPartialOutput>,
}
impl ChildCompactionFailure {
fn new(error: anyhow::Error, partial_output: &mut String) -> Self {
let mut text = std::mem::take(partial_output);
let truncated = truncate_string_field(&mut text, SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT);
let partial_output = (!text.is_empty()).then_some(BoundedPartialOutput { text, truncated });
Self {
error,
partial_output,
}
}
fn into_parts(self) -> (anyhow::Error, Option<BoundedPartialOutput>) {
(self.error, self.partial_output)
}
}
impl std::fmt::Display for ChildCompactionFailure {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.error.fmt(formatter)
}
}
impl std::error::Error for ChildCompactionFailure {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(self.error.as_ref())
}
}
struct ChildTurnRun<'a> {
agent: &'a AgentSession,
provider: &'a dyn Provider,
original_task: &'a SubagentTask,
prompt: &'a str,
tools: &'a ToolRuntime,
hooks: Option<&'a HookRuntime>,
session: Option<&'a Session>,
cwd: &'a Path,
sink: &'a mut SubagentActivitySink,
cancellation: &'a AgentCancellation,
semantic_progress_timeout: Duration,
agent_id: Option<String>,
compaction: Option<&'a super::config::SubagentCompactionConfig>,
auto_compaction_state: &'a mut ChildAutoCompactionState,
request_sequence: Arc<AtomicU64>,
}
fn run_child_turn_with_auto_compaction(
mut run: ChildTurnRun<'_>,
) -> anyhow::Result<AgentRunOutput> {
let initial_prompt = run.prompt.to_string();
let Some(session) = run.session else {
return run_child_agent_once(
&mut run,
&initial_prompt,
crate::output::UserPromptOrigin::User,
None,
);
};
let Some(compaction) = run.compaction else {
return run_child_agent_once(
&mut run,
&initial_prompt,
crate::output::UserPromptOrigin::User,
None,
);
};
let auto = &compaction.settings.compaction.auto;
let Some(auto_policy) = (auto.is_enabled() && run.agent.context_enabled())
.then(|| crate::agent::runner::auto_compaction_policy(auto, run.agent.context_max_tokens()))
.flatten()
else {
return run_child_agent_once(
&mut run,
&initial_prompt,
crate::output::UserPromptOrigin::User,
None,
);
};
let compaction_limit: AutoCompactionLimit = auto.compaction_limit();
let mut prompt = initial_prompt;
let mut prompt_origin = crate::output::UserPromptOrigin::User;
let mut combined = AgentRunOutput::default();
let mut combined_segments = 0_usize;
run.cancellation.check()?;
let static_tokens = run
.agent
.project_prompt_input_tokens_without_session(&prompt, Some(run.tools))?;
let current_tokens =
run.agent
.project_prompt_input_tokens(&prompt, session, Some(run.tools))?;
let hard_threshold = run.agent.context_threshold_tokens();
if static_tokens <= hard_threshold
&& current_tokens > hard_threshold
&& compaction_limit.allows(run.auto_compaction_state.count)
{
let (summary, rotation_warning) = compact_child_session_with_partial_output(
&mut run,
session,
compaction,
current_tokens,
format!("preflight hard budget ({hard_threshold} tokens)"),
&mut combined.text,
)?;
run.auto_compaction_state.count = run.auto_compaction_state.count.saturating_add(1);
let projected_tokens =
run.agent
.project_prompt_input_tokens(&prompt, session, Some(run.tools))?;
run.sink.output_event(OutputEvent::CompactionCompleted {
current_tokens: projected_tokens,
max_tokens: run.agent.context_max_tokens(),
summary,
})?;
if let Some(warning) = rotation_warning {
run.sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: warning,
})?;
}
run.auto_compaction_state.checkpoint_committed = true;
run.agent
.ensure_prompt_context_fits(&prompt, session, Some(run.tools))?;
}
loop {
run.cancellation.check()?;
let policy = (compaction_limit.allows(run.auto_compaction_state.count))
.then_some(auto_policy.clone());
let output_result = run_child_agent_once(&mut run, &prompt, prompt_origin, policy);
let mut continuation_overflow = None;
let output = match output_result {
Ok(output) => output,
Err(error)
if compaction_limit.allows(run.auto_compaction_state.count)
&& error
.downcast_ref::<ContextBudgetError>()
.is_some_and(|budget| {
budget.phase() == ContextBudgetPhase::Continuation
}) =>
{
let budget = error
.downcast::<ContextBudgetError>()
.expect("context budget error checked above");
let threshold = budget.threshold_tokens();
let threshold_display = if auto_policy.0 == threshold {
auto_policy.1.clone()
} else {
format!("continuation hard budget ({threshold} tokens)")
};
continuation_overflow = Some((
budget.estimated_tokens(),
threshold_display,
budget.to_string(),
));
budget.into_partial_output().unwrap_or_default()
}
Err(error) => return Err(error),
};
let persistence_degraded = output.persistence_degraded;
let recovery_blocked = output.auto_compaction_blocked_by_recovery;
merge_agent_output(&mut combined, output, combined_segments == 0);
combined_segments = combined_segments.saturating_add(1);
if persistence_degraded || recovery_blocked {
if let Some((_, _, message)) = continuation_overflow {
return Err(anyhow::anyhow!(message));
}
return Ok(combined);
}
if !compaction_limit.allows(run.auto_compaction_state.count) {
return Ok(combined);
}
let continuation_prompt = "continue";
let current_tokens =
run.agent
.project_prompt_input_tokens(continuation_prompt, session, Some(run.tools))?;
if continuation_overflow.is_none()
&& !auto.triggered(current_tokens as u64, run.agent.context_max_tokens() as u64)
{
return Ok(combined);
}
let (trigger_tokens, trigger_threshold) = continuation_overflow
.map(|(tokens, threshold, _)| (tokens, threshold))
.unwrap_or_else(|| (current_tokens, auto_policy.1.clone()));
let (summary, rotation_warning) = compact_child_session_with_partial_output(
&mut run,
session,
compaction,
trigger_tokens,
trigger_threshold,
&mut combined.text,
)?;
run.auto_compaction_state.count = run.auto_compaction_state.count.saturating_add(1);
let projected_tokens =
run.agent
.project_prompt_input_tokens(continuation_prompt, session, Some(run.tools))?;
run.sink.output_event(OutputEvent::CompactionCompleted {
current_tokens: projected_tokens,
max_tokens: run.agent.context_max_tokens(),
summary,
})?;
if let Some(warning) = rotation_warning {
run.sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: warning,
})?;
}
run.auto_compaction_state.checkpoint_committed = true;
let compacted_tokens =
run.agent
.ensure_prompt_context_fits(continuation_prompt, session, Some(run.tools))?;
if auto.triggered(
compacted_tokens as u64,
run.agent.context_max_tokens() as u64,
) {
anyhow::bail!(
"automatic subagent compaction completed, but the child context remains at or above threshold; no child continuation was sent"
);
}
run.cancellation.check()?;
prompt = continuation_prompt.to_string();
prompt_origin = crate::output::UserPromptOrigin::AutomaticCompaction;
}
}
fn run_child_agent_once(
run: &mut ChildTurnRun<'_>,
prompt: &str,
prompt_origin: crate::output::UserPromptOrigin,
continuation_auto_compaction_policy: Option<(usize, String)>,
) -> anyhow::Result<AgentRunOutput> {
run.agent
.run_print_with_tools_streaming_output_cancellable_untracked(
run.provider,
AgentRunRequest {
prompt,
prompt_origin,
effective_prompt: None,
tools: Some(run.tools),
hooks: run.hooks,
session: run.session,
cwd: run.cwd,
output_sink: Some(&mut *run.sink),
cancellation: run.cancellation.clone(),
session_title_job: None,
semantic_progress_timeout: Some(run.semantic_progress_timeout),
invocation_mode: crate::output::InvocationMode::Subagent,
agent_id: run.agent_id.clone(),
herdr_reporter: None,
initial_instructions: &[],
ttsr: crate::config::TtsrSettings::default(),
continuation_auto_compaction_policy,
},
None,
false,
Arc::clone(&run.request_sequence),
)
}
const SUBAGENT_COMPACTION_SCOPE_PREFIX: &str = "The following single JSON document extends to the end of this instruction. The document contains the authoritative scope for this compaction:\n";
const SUBAGENT_COMPACTION_SCOPE_DIRECTIVE: &str = "The original task payload is authoritative; summarize only that task; exclude unrelated work or tasks.";
#[derive(Serialize)]
struct SubagentCompactionTaskPayload<'a> {
intent: &'a str,
agent: Option<&'a str>,
identity: Option<&'a str>,
context: Option<&'a str>,
cwd: Option<&'a Path>,
}
impl<'a> From<&'a SubagentTask> for SubagentCompactionTaskPayload<'a> {
fn from(task: &'a SubagentTask) -> Self {
Self {
intent: &task.intent,
agent: task.agent.as_deref(),
identity: task.identity.as_deref(),
context: task.context.as_deref(),
cwd: task.cwd.as_deref(),
}
}
}
#[derive(Serialize)]
struct SubagentCompactionScope<'a> {
directive: &'static str,
original_task: SubagentCompactionTaskPayload<'a>,
}
pub(super) fn subagent_compaction_instructions(task: &SubagentTask) -> anyhow::Result<String> {
let scope = SubagentCompactionScope {
directive: SUBAGENT_COMPACTION_SCOPE_DIRECTIVE,
original_task: SubagentCompactionTaskPayload::from(task),
};
let serialized_scope = serde_json::to_string(&scope).map_err(|error| {
anyhow::anyhow!("failed to serialize original subagent task payload: {error}")
})?;
Ok(format!(
"{SUBAGENT_COMPACTION_SCOPE_PREFIX}{serialized_scope}"
))
}
fn compact_child_session(
run: &mut ChildTurnRun<'_>,
session: &Session,
compaction: &super::config::SubagentCompactionConfig,
current_tokens: usize,
threshold: String,
) -> anyhow::Result<(String, Option<String>)> {
run.cancellation.check()?;
let additional_instructions = subagent_compaction_instructions(run.original_task)?;
run.sink.output_event(OutputEvent::CompactionTriggered {
current_tokens,
max_tokens: run.agent.context_max_tokens(),
threshold,
})?;
run.sink.output_event(OutputEvent::CompactionStarted)?;
let sequence = crate::agent::next_request_sequence(&run.request_sequence);
let mut usage = crate::agent::provider_stream::CompactionUsage::default();
match crate::compaction::compact_session_observed(
crate::compaction::CompactSessionJob {
active_config: compaction.active_config.clone(),
settings: compaction.settings.clone(),
session: session.clone(),
cwd: run.cwd.to_path_buf(),
cancellation: run.cancellation.clone(),
custom_instructions: None,
additional_instructions: Some(additional_instructions),
},
&mut |provider, event| usage.observe(provider, event, sequence, &mut Some(run.sink)),
) {
Ok(Some(result)) => Ok((result.summary, result.rotation_warning)),
Ok(None) => {
let message = format!(
"automatic subagent compaction produced no usable summary; {}; partial child output is preserved in the failed child result; no child continuation was sent",
child_session_recovery_context(session),
);
Err(anyhow::anyhow!(
crate::agent::runner::bounded_auto_compaction_diagnostic(message)
))
}
Err(error) if crate::cancellation::is_run_canceled(&error) => Err(error),
Err(error) => {
if error
.downcast_ref::<crate::sessions::CompactionRotationError>()
.is_some()
{
run.auto_compaction_state.count = run.auto_compaction_state.count.saturating_add(1);
run.auto_compaction_state.checkpoint_committed = true;
}
let message = format!(
"automatic subagent compaction failed; {}; partial child output is preserved in the failed child result; no child continuation was sent: {}",
child_session_recovery_context(session),
crate::agent::runner::bounded_auto_compaction_diagnostic(error),
);
Err(anyhow::anyhow!(
crate::agent::runner::bounded_auto_compaction_diagnostic(message)
))
}
}
}
fn compact_child_session_with_partial_output(
run: &mut ChildTurnRun<'_>,
session: &Session,
compaction: &super::config::SubagentCompactionConfig,
current_tokens: usize,
threshold: String,
partial_output: &mut String,
) -> anyhow::Result<(String, Option<String>)> {
match compact_child_session(run, session, compaction, current_tokens, threshold) {
Ok(result) => Ok(result),
Err(error) if crate::cancellation::is_run_canceled(&error) => Err(error),
Err(error) => Err(ChildCompactionFailure::new(error, partial_output).into()),
}
}
fn child_session_recovery_context(session: &Session) -> String {
format!(
"child session id={} path={}",
redact_sensitive_text(session.id()),
redact_sensitive_text(&session.path().display().to_string()),
)
}
pub(super) fn can_retry_subagent_provider(
retry_attempt: u64,
cancellation: &AgentCancellation,
retryable: bool,
has_filesystem_changes: bool,
compaction_checkpoint_committed: bool,
) -> bool {
retry_attempt < SUBAGENT_PROVIDER_RETRY_MAX_ATTEMPTS
&& !cancellation.is_canceled()
&& retryable
&& !has_filesystem_changes
&& !compaction_checkpoint_committed
}
fn merge_agent_output(combined: &mut AgentRunOutput, output: AgentRunOutput, first_segment: bool) {
if !combined.text.is_empty() && !output.text.is_empty() {
combined.text.push('\n');
}
combined.text.push_str(&output.text);
combined.usage = output.usage;
combined.total_tokens = if first_segment {
output.total_tokens
} else {
match (combined.total_tokens, output.total_tokens) {
(Some(total), Some(next)) => Some(total.saturating_add(next)),
_ => None,
}
};
combined.tool_results.extend(output.tool_results);
combined.persistence_degraded |= output.persistence_degraded;
combined.recovered_incomplete_stream |= output.recovered_incomplete_stream;
combined.auto_compaction_blocked_by_recovery |= output.auto_compaction_blocked_by_recovery;
}
fn wait_for_provider_retry(cancellation: &AgentCancellation) -> bool {
let deadline = std::time::Instant::now() + SUBAGENT_PROVIDER_RETRY_BACKOFF;
loop {
if cancellation.is_canceled() {
return false;
}
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
if remaining.is_zero() {
return true;
}
std::thread::sleep(remaining.min(SUBAGENT_PROVIDER_RETRY_POLL_INTERVAL));
}
}
pub(super) struct ChildAgentRun {
pub(super) agent: AgentSession,
pub(super) provider: Arc<dyn Provider>,
pub(super) compaction: Option<super::config::SubagentCompactionConfig>,
}
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),
compaction: config.compaction.clone(),
});
};
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 role_prompt =
crate::primary_agents::render_your_role_prompt(&profile.name, &profile.prompt);
let agent = config
.parent_agent
.with_appended_system_prompt(&role_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);
let mut compaction = config.compaction.clone();
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(&selection, cwd)?;
provider = Arc::clone(&resolved.provider);
let settings = resolved.settings.clone();
let compaction_settings = config
.compaction
.as_ref()
.map(|compaction| compaction.settings.clone())
.unwrap_or_else(|| settings.clone());
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_context_budget(resolved.context_budget)
.with_text_verbosity(settings.text_verbosity_for(&selection.provider));
compaction = Some(super::config::SubagentCompactionConfig::new(
resolved.active_config,
compaction_settings,
));
} else if let Some(reasoning) = profile.reasoning {
agent = agent.with_reasoning_override(reasoning);
}
Ok(ChildAgentRun {
agent,
provider,
compaction,
})
}
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>,
usage: Option<crate::output::NormalizedUsageAggregate>,
) -> 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,
usage,
changed_files: Vec::new(),
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>,
partial_output: Option<BoundedPartialOutput>,
usage: Option<crate::output::NormalizedUsageAggregate>,
}
pub(super) fn finish_failed_subagent(input: FailedSubagentInput<'_>) -> SubagentTaskResult {
let FailedSubagentInput {
config,
task_activity_id,
id,
task,
cwd,
error,
finisher,
session,
partial_output,
usage,
} = input;
let partial_output = partial_output.filter(|output| !output.text.trim().is_empty());
let output_truncated = partial_output
.as_ref()
.is_some_and(|output| output.truncated);
let output = partial_output
.map(|output| output.text)
.or_else(|| failed_subagent_session_snapshot(session));
let mut result = failed_result_with_session_and_output(
id,
task,
cwd,
session.map(|session| session.id().to_string()),
session.map(|session| session.path().to_path_buf()),
error,
output,
);
result.usage = usage;
result.output_truncated |= output_truncated;
finisher.finish(config, 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_usage(id, task, cwd, session_id, session_path, error, None)
}
pub(super) fn failed_result_with_session_and_usage(
id: String,
task: SubagentTask,
cwd: PathBuf,
session_id: Option<String>,
session_path: Option<PathBuf>,
error: String,
usage: Option<crate::output::NormalizedUsageAggregate>,
) -> SubagentTaskResult {
failed_result_with_session_and_output_and_usage(
id,
task,
cwd,
session_id,
session_path,
error,
None,
usage,
)
}
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 {
failed_result_with_session_and_output_and_usage(
id,
task,
cwd,
session_id,
session_path,
error,
output,
None,
)
}
#[allow(clippy::too_many_arguments)]
fn failed_result_with_session_and_output_and_usage(
id: String,
task: SubagentTask,
cwd: PathBuf,
session_id: Option<String>,
session_path: Option<PathBuf>,
error: String,
output: Option<String>,
usage: Option<crate::output::NormalizedUsageAggregate>,
) -> SubagentTaskResult {
let mut output = output.unwrap_or_default();
let output_was_truncated =
truncate_string_field(&mut output, SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT);
SubagentTaskResult {
id,
status: SubagentStatus::Failed,
intent: task.intent,
agent: task.agent,
identity: task.identity,
cwd,
session_id,
session_path,
total_tokens: None,
usage,
changed_files: Vec::new(),
output,
structured_output: None,
output_truncated: output_was_truncated,
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> {
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));
}
Some(SessionEventKind::TurnStatus) => {
if let Some(payload) = event.turn_status_payload()
&& matches!(
payload.status,
crate::sessions::TurnStatus::Cancelled
| crate::sessions::TurnStatus::Failed
| crate::sessions::TurnStatus::CompactionRequired
)
&& let Some(text) = payload.assistant_text
{
authoritative_assistant_output = Some(text);
}
}
_ => {}
}
}
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 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
}