use super::{ProviderRunOptions, bounded_auto_compaction_diagnostic, reborrow_output_sink};
use crate::{
agent::steering::AgentSteering,
agent::{AgentRunOutput, AgentRunRequest, AgentSession},
cancellation::is_run_canceled,
config::AutoCompactionLimit,
output::{OutputEvent, UserPromptOrigin},
};
use anyhow::Result;
use std::sync::{Arc, atomic::AtomicU64};
use super::preparation::PreparedRun;
pub(super) fn run(
prepared: PreparedRun<'_>,
instructions: &[crate::instructions::InstructionFile],
options: &mut ProviderRunOptions<'_, '_>,
steering: Option<AgentSteering>,
) -> Result<AgentRunOutput> {
let PreparedRun {
active_config,
settings,
context_budget,
cancellation,
parent_agent_for_provider,
provider,
hooks,
tools,
mut title_job,
auto,
auto_eligible,
auto_policy,
herdr_reporter,
} = prepared;
let config = active_config.as_ref();
let mut prompt = options.prompt.to_string();
let mut prompt_origin = UserPromptOrigin::User;
let mut effective_prompt: Option<String> = None;
let mut combined = AgentRunOutput::default();
let mut auto_compaction_count = 0_usize;
let compaction_limit: AutoCompactionLimit = auto.compaction_limit();
let mut preflight_compaction_used = false;
let request_sequence = Arc::new(AtomicU64::new(0));
let mut projected_candidate: Option<(String, String)> = None;
if auto_eligible {
cancellation.check()?;
let session = options
.session
.expect("auto-compaction eligibility requires session");
let effective_current_prompt = AgentSession::effective_prompt_for_projection(
&prompt,
Some(&tools),
options.invocation_mode,
);
let hard_threshold = context_budget.threshold_tokens();
let static_tokens = parent_agent_for_provider
.project_prompt_input_tokens_without_session(&effective_current_prompt, Some(&tools))?;
effective_prompt = Some(effective_current_prompt.clone());
if static_tokens <= hard_threshold && compaction_limit.allows(auto_compaction_count) {
let current_tokens = parent_agent_for_provider.project_prompt_input_tokens(
&effective_current_prompt,
session,
Some(&tools),
)?;
if current_tokens > hard_threshold {
preflight_compaction_used = true;
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionTriggered {
current_tokens,
max_tokens: parent_agent_for_provider.context_max_tokens(),
threshold: format!("preflight hard budget ({hard_threshold} tokens)"),
})?;
sink.output_event(OutputEvent::CompactionStarted)?;
}
let compaction = compact_accounted(
crate::compaction::CompactSessionJob {
active_config: config.clone(),
settings: settings.clone(),
session: session.clone(),
cwd: options.cwd.to_path_buf(),
cancellation: cancellation.clone(),
custom_instructions: None,
additional_instructions: None,
},
&mut combined,
&request_sequence,
&mut options.output_sink,
);
let summary = match compaction {
Ok(Some(result)) => {
super::emit_compaction_fast_observation(
&mut options.output_sink,
&result,
&request_sequence,
)?;
if let Some(warning) = result.rotation_warning.as_ref()
&& let Some(sink) = options.output_sink.as_deref_mut()
{
sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: warning.clone(),
})?;
}
result.summary
}
Ok(None) => {
let message = bounded_auto_compaction_diagnostic(
"automatic preflight compaction produced no usable summary; old primary remains authoritative; no provider request was sent",
);
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionFailed {
message: message.clone(),
canceled: false,
})?;
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
Err(error) if is_run_canceled(&error) => {
let message = "automatic preflight compaction canceled; old primary remains authoritative; no provider request was sent".to_string();
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionFailed {
message: message.clone(),
canceled: true,
})?;
sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: message.clone(),
})?;
}
return Err(error);
}
Err(error) => {
let authority = error
.downcast_ref::<crate::sessions::CompactionRotationError>()
.map(|_| "")
.unwrap_or(" old primary remains authoritative;");
let message = bounded_auto_compaction_diagnostic(format!(
"automatic preflight compaction failed;{authority} no provider request was sent: {error}"
));
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionFailed {
message: message.clone(),
canceled: false,
})?;
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
};
let projected_tokens = parent_agent_for_provider.project_prompt_input_tokens(
&effective_current_prompt,
session,
Some(&tools),
)?;
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionCompleted {
current_tokens: projected_tokens,
max_tokens: parent_agent_for_provider.context_max_tokens(),
summary: summary.clone(),
})?;
}
let _compacted_tokens = match parent_agent_for_provider.ensure_prompt_context_fits(
&effective_current_prompt,
session,
Some(&tools),
) {
Ok(tokens) => tokens,
Err(error) => {
let message = bounded_auto_compaction_diagnostic(format!(
"automatic preflight compaction completed, but current prompt still exceeds hard context budget; new checkpoint remains authoritative; no provider request was sent: {error}"
));
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
};
auto_compaction_count = auto_compaction_count.saturating_add(1);
}
}
}
loop {
let request = AgentRunRequest {
prompt: &prompt,
prompt_origin,
effective_prompt: effective_prompt.as_deref(),
tools: Some(&tools),
hooks: Some(&hooks),
session: options.session,
cwd: options.cwd,
output_sink: reborrow_output_sink(&mut options.output_sink),
cancellation: cancellation.clone(),
session_title_job: title_job.take(),
semantic_progress_timeout: Some(settings.provider_stream.semantic_progress_timeout()),
invocation_mode: options.invocation_mode,
agent_id: options
.selected_primary_agent
.as_ref()
.map(|profile| profile.id.clone()),
initial_instructions: instructions,
ttsr: settings.ttsr.clone(),
herdr_reporter: herdr_reporter.clone(),
continuation_auto_compaction_policy: (auto_eligible
&& compaction_limit.allows(auto_compaction_count))
.then(|| auto_policy.clone())
.flatten(),
};
let output_result = match steering.as_ref() {
Some(steering) if auto_eligible => parent_agent_for_provider
.run_print_with_tools_streaming_output_cancellable_untracked(
provider.as_ref(),
request,
Some(steering.clone()),
true,
Arc::clone(&request_sequence),
),
Some(steering) => parent_agent_for_provider
.run_print_with_tools_streaming_output_cancellable_untracked(
provider.as_ref(),
request,
Some(steering.clone()),
false,
Arc::clone(&request_sequence),
),
None => parent_agent_for_provider
.run_print_with_tools_streaming_output_cancellable_untracked(
provider.as_ref(),
request,
None,
false,
Arc::clone(&request_sequence),
),
};
let mut continuation_context_overflow = None;
let output = match output_result {
Ok(output) => output,
Err(error)
if auto_eligible
&& compaction_limit.allows(auto_compaction_count)
&& error
.downcast_ref::<crate::agent::ContextBudgetError>()
.is_some_and(|budget| {
budget.phase() == crate::agent::ContextBudgetPhase::Continuation
}) =>
{
let budget = error
.downcast::<crate::agent::ContextBudgetError>()
.expect("context budget error checked above");
let threshold = budget.threshold_tokens();
let threshold_display = auto_policy
.as_ref()
.filter(|(soft_threshold, _)| *soft_threshold == threshold)
.map(|(_, display)| display.clone())
.unwrap_or_else(|| format!("continuation hard budget ({threshold} tokens)"));
continuation_context_overflow = Some((
budget.estimated_tokens(),
threshold_display,
budget.to_string(),
));
budget.into_partial_output().unwrap_or_default()
}
Err(error)
if preflight_compaction_used
&& error
.downcast_ref::<crate::agent::RequiredUserInputPersistenceError>()
.is_some() =>
{
let message = bounded_auto_compaction_diagnostic(format!(
"current user input persistence failed after preflight compaction; new checkpoint remains authoritative; no provider request was sent: {error}"
));
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
Err(error)
if auto_compaction_count > 0
&& error
.downcast_ref::<crate::agent::RequiredUserInputPersistenceError>()
.is_some() =>
{
let message = bounded_auto_compaction_diagnostic(format!(
"automatic continuation persistence failed; new checkpoint remains authoritative; no provider continuation was sent: {error}"
));
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
Err(error) => return Err(error),
};
let persistence_degraded = output.persistence_degraded;
let recovered_incomplete_stream = output.recovered_incomplete_stream;
let auto_compaction_blocked_by_recovery = output.auto_compaction_blocked_by_recovery;
if !combined.text.is_empty() && !output.text.is_empty() {
combined.text.push('\n');
}
combined.text.push_str(&output.text);
combined.usage = output.usage;
if !matches!(output.fast_outcome, crate::fast::FastOutcome::NotRequested) {
combined.fast_outcome = output.fast_outcome.clone();
}
combined.total_tokens = match (combined.total_tokens, output.total_tokens) {
(Some(total), Some(next)) => Some(total.saturating_add(next)),
(None, next) => next,
(total, None) => total,
};
combined.tool_results.extend(output.tool_results);
combined.persistence_degraded |= persistence_degraded;
combined.recovered_incomplete_stream |= recovered_incomplete_stream;
combined.auto_compaction_blocked_by_recovery |= auto_compaction_blocked_by_recovery;
if !auto_eligible || !compaction_limit.allows(auto_compaction_count) {
return Ok(combined);
}
if persistence_degraded || auto_compaction_blocked_by_recovery {
if let Some((_, _, message)) = continuation_context_overflow {
return Err(anyhow::anyhow!(message));
}
return Ok(combined);
}
cancellation.check()?;
let session = options
.session
.expect("auto-compaction eligibility requires session");
let observed_steering = steering.as_ref().and_then(AgentSteering::observe_collapsed);
let candidate = observed_steering
.as_ref()
.map(|batch| batch.text.as_str())
.unwrap_or("continue");
let effective_candidate = match projected_candidate.as_ref() {
Some((raw, effective)) if raw == candidate => effective.clone(),
_ => {
let effective = AgentSession::effective_prompt_for_projection(
candidate,
Some(&tools),
options.invocation_mode,
);
projected_candidate = Some((candidate.to_string(), effective.clone()));
effective
}
};
let current_tokens = parent_agent_for_provider.project_prompt_input_tokens(
&effective_candidate,
session,
Some(&tools),
)?;
let max_tokens = parent_agent_for_provider.context_max_tokens();
if continuation_context_overflow.is_none()
&& !auto.triggered(current_tokens as u64, max_tokens as u64)
{
if let Some(batch) = observed_steering {
prompt = batch.text;
effective_prompt = Some(effective_candidate);
prompt_origin = UserPromptOrigin::Steering;
continue;
}
return Ok(combined);
}
let (trigger_tokens, trigger_threshold) = continuation_context_overflow
.as_ref()
.map(|(estimated, threshold, _)| (*estimated, threshold.clone()))
.unwrap_or_else(|| {
(
current_tokens,
auto.threshold_display()
.unwrap_or_else(|| "invalid threshold".to_string()),
)
});
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionTriggered {
current_tokens: trigger_tokens,
max_tokens,
threshold: trigger_threshold,
})?;
sink.output_event(OutputEvent::CompactionStarted)?;
}
let compaction = compact_accounted(
crate::compaction::CompactSessionJob {
active_config: config.clone(),
settings: settings.clone(),
session: session.clone(),
cwd: options.cwd.to_path_buf(),
cancellation: cancellation.clone(),
custom_instructions: None,
additional_instructions: None,
},
&mut combined,
&request_sequence,
&mut options.output_sink,
);
let result = match compaction {
Ok(Some(result)) => result,
Ok(None) => {
let message = "automatic compaction produced no usable summary; old primary remains authoritative; no continuation was sent";
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionFailed {
message: message.to_string(),
canceled: false,
})?;
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.to_string(),
})?;
}
anyhow::bail!(message)
}
Err(error) => {
if is_run_canceled(&error) {
let message = "automatic compaction canceled; old primary remains authoritative; no continuation was sent";
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionFailed {
message: message.to_string(),
canceled: true,
})?;
sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: message.to_string(),
})?;
}
return Err(error);
}
let message = if let Some(rotation_error) =
error.downcast_ref::<crate::sessions::CompactionRotationError>()
{
bounded_auto_compaction_diagnostic(format!(
"automatic compaction failed; no continuation was sent: {rotation_error}"
))
} else {
bounded_auto_compaction_diagnostic(format!(
"automatic compaction failed; old primary remains authoritative; no continuation was sent: {error}"
))
};
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionFailed {
message: message.clone(),
canceled: false,
})?;
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
};
super::emit_compaction_fast_observation(
&mut options.output_sink,
&result,
&request_sequence,
)?;
auto_compaction_count = auto_compaction_count.saturating_add(1);
let observed_steering = steering.as_ref().and_then(AgentSteering::observe_collapsed);
if let Some(warning) = result.rotation_warning.as_ref()
&& let Some(sink) = options.output_sink.as_deref_mut()
{
sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: warning.clone(),
})?;
}
let candidate = observed_steering
.as_ref()
.map(|batch| batch.text.as_str())
.unwrap_or("continue");
let effective_candidate = match projected_candidate.as_ref() {
Some((raw, effective)) if raw == candidate => effective.clone(),
_ => {
let effective = AgentSession::effective_prompt_for_projection(
candidate,
Some(&tools),
options.invocation_mode,
);
projected_candidate = Some((candidate.to_string(), effective.clone()));
effective
}
};
let projected_tokens = parent_agent_for_provider.project_prompt_input_tokens(
&effective_candidate,
session,
Some(&tools),
)?;
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::CompactionCompleted {
current_tokens: projected_tokens,
max_tokens,
summary: result.summary.clone(),
})?;
}
let compacted_tokens = match parent_agent_for_provider.ensure_prompt_context_fits(
&effective_candidate,
session,
Some(&tools),
) {
Ok(tokens) => tokens,
Err(error) => {
let message = bounded_auto_compaction_diagnostic(format!(
"automatic compaction completed, but projected continuation exceeds normal context budget; new checkpoint remains authoritative; no provider continuation was sent: {error}"
));
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.clone(),
})?;
}
return Err(anyhow::anyhow!(message));
}
};
if auto.triggered(compacted_tokens as u64, max_tokens as u64) {
let message = "automatic compaction completed, but projected context remains at or above threshold; new checkpoint remains authoritative; no continuation was sent";
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: "error".to_string(),
message: message.to_string(),
})?;
}
anyhow::bail!(message)
}
if let Err(error) = cancellation.check() {
let message = "automatic continuation canceled; new checkpoint remains authoritative; no provider continuation was sent";
if let Some(sink) = options.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: "warning".to_string(),
message: message.to_string(),
})?;
}
return Err(error);
}
match observed_steering {
Some(batch) => {
prompt = batch.text;
effective_prompt = Some(effective_candidate);
prompt_origin = UserPromptOrigin::Steering;
}
None => {
prompt = "continue".to_string();
effective_prompt = None;
prompt_origin = UserPromptOrigin::AutomaticCompaction;
}
}
}
}
fn compact_accounted(
job: crate::compaction::CompactSessionJob,
output: &mut AgentRunOutput,
sequence: &AtomicU64,
sink: &mut Option<&mut dyn crate::agent::AgentOutputSink>,
) -> Result<Option<crate::compaction::CompactionResult>> {
let sequence = crate::agent::next_request_sequence(sequence);
let mut usage = crate::agent::provider_stream::CompactionUsage::default();
let result = crate::compaction::compact_session_observed(job, &mut |provider, event| {
usage.observe(provider, event, sequence, sink)
});
if let Some(total) = usage.total {
output.total_tokens = Some(output.total_tokens.unwrap_or(0).saturating_add(total));
}
result
}