pub(crate) mod cancellation;
mod prompt;
pub(crate) mod prompt_injections;
mod provider_stream;
pub(crate) mod runner;
mod session_persistence;
pub(crate) mod steering;
mod subdir_instructions;
#[cfg(test)]
mod tests;
mod tool_continuation;
mod tool_lifecycle;
mod tool_output_compression;
pub(crate) mod ttsr;
mod turn_state;
mod util;
#[cfg(test)]
use self::util::AssistantOnlySink;
use self::{
prompt::{build_system_prompt, build_system_prompt_with_prompt_dir_and_subagents_and_disabled},
provider_stream::{
ProviderEventCollector, ProviderStreamFailure, ProviderStreamResult,
collect_provider_events,
},
session_persistence::{SessionPersistence, emit_replay_diagnostics},
steering::{AgentSteering, SteeringBatch},
subdir_instructions::SubdirInstructionState,
tool_lifecycle::{
ToolLifecycleRun, record_provider_context_item, run_message_phase_hooks, run_tool_lifecycle,
},
turn_state::AgentTurnState,
util::{AssistantChunkBatch, normalize_agent_identifier, safe_error_message},
};
use crate::{
agent::cancellation::{AgentCancellation, is_run_canceled},
config::TextVerbosity,
context::{
ContextBudget, ContextCache, ContextCacheEntry, ContextTokenCount,
build_conversation_replay, conversation_cache_material,
project_provider_conversation_items_tokens, project_provider_request_input_tokens,
},
herdr::{HerdrReporter, HerdrTurnReporter},
hex::lower_hex,
hooks::{HookPhase, HookPolicyError, HookRuntime},
instructions::InstructionFile,
output::{
ActivityEvent, ActivityId, ActivitySender, ContextUsageSource, HookContextMetadata,
InvocationMode, OutputEvent,
},
providers::{
ChatMessage, Provider, ProviderConversationItem, ProviderRequest, ToolCall, Usage,
},
sessions::{Session, SessionEventKind, SessionReadDiagnostic, try_record_session_event},
skills::SkillDiscovery,
tool_display::format_tool_block,
tools::{ToolResult, ToolRuntime},
};
use serde_json::json;
use sha2::{Digest, Sha256};
use std::{
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
fn inject_steering_update_at_continuation_boundary(
steering: Option<&AgentSteering>,
batch: Option<SteeringBatch>,
turn_state: &mut AgentTurnState,
session_persistence: &SessionPersistence<'_>,
output_sink: &mut Option<&mut dyn AgentOutputSink>,
) -> anyhow::Result<bool> {
let Some(batch) = batch else {
return Ok(false);
};
session_persistence
.record_required(
SessionEventKind::UserInput,
json!({"text": batch.text.clone()}),
)
.map_err(|error| {
anyhow::anyhow!(
"failed to persist steering input before provider continuation: {error}"
)
})?;
if let Some(steering) = steering {
steering.acknowledge(batch.count);
}
if let Some(sink) = output_sink.as_deref_mut() {
sink.output_event(OutputEvent::UserPrompt {
text: batch.text.clone(),
})?;
}
turn_state.append_provider_context_items([ProviderConversationItem::Message(
ChatMessage::user(batch.text),
)]);
Ok(true)
}
pub(crate) use self::prompt::load_compact_prompt;
#[cfg(test)]
pub type AssistantTextSink<'a> = dyn FnMut(&str) -> anyhow::Result<()> + 'a;
pub trait AgentOutputSink {
fn assistant_delta(&mut self, text: &str) -> anyhow::Result<()>;
fn activity_event(&mut self, _event: ActivityEvent) -> anyhow::Result<()> {
Ok(())
}
fn activity_sender(&self) -> Option<ActivitySender> {
None
}
fn current_parent_activity_id(&self) -> Option<ActivityId> {
None
}
fn tool_result(&mut self, call: &ToolCall, result: &ToolResult) -> anyhow::Result<()> {
let block = format_tool_block(call, result);
self.tool_block(&block)
}
fn output_event(&mut self, event: OutputEvent) -> anyhow::Result<()> {
match event {
OutputEvent::AssistantDelta { text } => self.assistant_delta(&text),
OutputEvent::ToolResult { call, result, .. } => self.tool_result(&call, &result),
OutputEvent::SubdirInstructionInjection { path, .. } => self.tool_block(&format!(
"Loaded subdirectory instructions: {}",
path.display()
)),
OutputEvent::SessionHeader { .. }
| OutputEvent::UserPrompt { .. }
| OutputEvent::AutomaticUserPrompt { .. }
| OutputEvent::CompactionTriggered { .. }
| OutputEvent::CompactionStarted
| OutputEvent::CompactionCompleted { .. }
| OutputEvent::BashCommand { .. }
| OutputEvent::ContextUsage { .. }
| OutputEvent::ThinkingSummaryDelta { .. }
| OutputEvent::ThinkingSummaryComplete { .. }
| OutputEvent::ThinkingSummaryCompleteIdentified { .. }
| OutputEvent::AssistantComplete { .. }
| OutputEvent::Diagnostic { .. }
| OutputEvent::HookDiagnostic { .. }
| OutputEvent::ProviderContextInjection { .. }
| OutputEvent::ToolStarted { .. } => Ok(()),
}
}
fn tool_block(&mut self, block: &str) -> anyhow::Result<()>;
}
pub(crate) struct AgentRunRequest<'a, 'sink> {
pub(crate) prompt: &'a str,
pub(crate) prompt_origin: crate::output::UserPromptOrigin,
pub(crate) effective_prompt: Option<&'a str>,
pub(crate) tools: Option<&'a ToolRuntime>,
pub(crate) hooks: Option<&'a HookRuntime>,
pub(crate) session: Option<&'a Session>,
pub(crate) cwd: &'a Path,
pub(crate) output_sink: Option<&'sink mut dyn AgentOutputSink>,
pub(crate) cancellation: AgentCancellation,
pub(crate) session_title_job: Option<crate::sessions::titles::SessionTitleJob>,
pub(crate) semantic_progress_timeout: Option<Duration>,
pub(crate) invocation_mode: InvocationMode,
pub(crate) agent_id: Option<String>,
pub(crate) initial_instructions: &'a [InstructionFile],
pub(crate) ttsr: crate::config::TtsrSettings,
pub(crate) herdr_reporter: Option<HerdrReporter>,
pub(crate) continuation_auto_compaction_policy: Option<(usize, String)>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentSession {
provider_id: String,
model: String,
system_prompt: String,
context_budget: ContextBudget,
thinking_level: crate::thinking::ThinkingLevel,
text_verbosity: Option<TextVerbosity>,
thinking_levels: Vec<crate::thinking::ThinkingLevel>,
send_default_reasoning_summary: bool,
context_cache_dir: Option<PathBuf>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct AgentSessionConfig {
provider_id: String,
model: String,
context_budget: ContextBudget,
thinking_level: crate::thinking::ThinkingLevel,
text_verbosity: Option<TextVerbosity>,
thinking_levels: Vec<crate::thinking::ThinkingLevel>,
send_default_reasoning_summary: bool,
context_cache_dir: Option<PathBuf>,
}
impl AgentSessionConfig {
pub(crate) fn new(provider_id: impl Into<String>, model: impl Into<String>) -> Self {
Self {
provider_id: normalize_agent_identifier(provider_id.into(), "provider"),
model: normalize_agent_identifier(model.into(), "model"),
context_budget: ContextBudget::default(),
thinking_level: crate::thinking::ThinkingLevel::Default,
text_verbosity: None,
thinking_levels: crate::thinking::default_thinking_levels(),
send_default_reasoning_summary: false,
context_cache_dir: None,
}
}
fn for_model(model: impl Into<String>) -> Self {
Self::new(crate::providers::OPENAI_CODEX_PROVIDER, model)
}
pub(crate) fn with_context_budget(mut self, budget: ContextBudget) -> Self {
self.context_budget = budget;
self
}
pub(crate) fn with_context_cache_dir(mut self, cache_dir: PathBuf) -> Self {
self.context_cache_dir = Some(cache_dir);
self
}
pub(crate) fn with_thinking_level(
mut self,
thinking_level: crate::thinking::ThinkingLevel,
) -> Self {
self.thinking_level = thinking_level;
self
}
pub(crate) fn with_text_verbosity(mut self, text_verbosity: Option<TextVerbosity>) -> Self {
self.text_verbosity = text_verbosity;
self
}
pub(crate) fn with_thinking_levels(
mut self,
thinking_levels: Vec<crate::thinking::ThinkingLevel>,
) -> Self {
self.thinking_levels = crate::thinking::normalize_thinking_levels(&thinking_levels);
self.thinking_level =
crate::thinking::resolve_thinking_level(&self.thinking_levels, self.thinking_level);
self
}
pub(crate) fn with_default_reasoning_summary(mut self, supported: bool) -> Self {
self.send_default_reasoning_summary = supported;
self
}
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct AgentRunOutput {
pub text: String,
pub usage: Option<Usage>,
pub total_tokens: Option<u64>,
pub tool_results: Vec<ToolResult>,
pub(crate) persistence_degraded: bool,
pub(crate) recovered_incomplete_stream: bool,
pub(crate) auto_compaction_blocked_by_recovery: bool,
}
#[derive(Debug)]
pub(crate) struct RequiredUserInputPersistenceError {
message: String,
}
impl RequiredUserInputPersistenceError {
fn new(error: impl std::fmt::Display) -> Self {
let message = crate::output::redact_sensitive_text(&error.to_string())
.chars()
.take(240)
.collect();
Self { message }
}
}
impl std::fmt::Display for RequiredUserInputPersistenceError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
formatter,
"failed to persist user input before provider run: {}",
self.message
)
}
}
impl std::error::Error for RequiredUserInputPersistenceError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ContextBudgetPhase {
Initial,
Continuation,
}
#[derive(Debug)]
pub(crate) struct ContextBudgetError {
phase: ContextBudgetPhase,
estimated_tokens: usize,
threshold_tokens: usize,
partial_output: Option<AgentRunOutput>,
message: String,
}
impl ContextBudgetError {
fn new(
phase: ContextBudgetPhase,
estimated_tokens: usize,
threshold_tokens: usize,
max_tokens: usize,
reserve_tokens: usize,
) -> Self {
Self {
phase,
estimated_tokens,
threshold_tokens,
partial_output: None,
message: format!(
"full session history request is estimated at {estimated_tokens} tokens, exceeding threshold {threshold_tokens} tokens (max_tokens={max_tokens}, reserve_tokens={reserve_tokens}); no history was omitted; run /compact, choose a larger compaction model, or start /new"
),
}
}
fn with_partial_output(mut self, output: AgentRunOutput) -> Self {
self.partial_output = Some(output);
self
}
pub(crate) fn phase(&self) -> ContextBudgetPhase {
self.phase
}
pub(crate) fn estimated_tokens(&self) -> usize {
self.estimated_tokens
}
pub(crate) fn threshold_tokens(&self) -> usize {
self.threshold_tokens
}
pub(crate) fn into_partial_output(self) -> Option<AgentRunOutput> {
self.partial_output
}
}
impl std::fmt::Display for ContextBudgetError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(formatter, "{}", self.message)
}
}
impl std::error::Error for ContextBudgetError {}
struct InitialConversation {
conversation: Vec<ProviderConversationItem>,
session_read_diagnostics: Vec<SessionReadDiagnostic>,
}
struct PreparedInitialRun<'run> {
provider_prompt: String,
initial_conversation: InitialConversation,
base_conversation: Arc<[ProviderConversationItem]>,
prompt_cache_key: Option<String>,
initial_request: ProviderRequest,
initial_context_projection: Option<ContextTokenCount>,
request_projection_cache: RequestTokenProjectionCache,
session_persistence: SessionPersistence<'run>,
}
struct PrintTurnState<'run> {
base_conversation: Arc<[ProviderConversationItem]>,
prompt_cache_key: Option<String>,
session_persistence: SessionPersistence<'run>,
subdir_instruction_state: SubdirInstructionState,
assistant_chunk_batch: AssistantChunkBatch<'run>,
turn_state: AgentTurnState,
output: AgentRunOutput,
request_projection_cache: RequestTokenProjectionCache,
request_sequence: u64,
hook_context: HookContextMetadata,
herdr_turn: HerdrTurnReporter,
title_guard: TitleGenerationGuard,
}
fn should_expand_prompt_context(invocation_mode: InvocationMode) -> bool {
matches!(
invocation_mode,
InvocationMode::Print | InvocationMode::Shell | InvocationMode::MissionControl
)
}
fn prompt_cache_key_for_session_id(session_id: &str) -> String {
let digest = Sha256::digest(session_id.as_bytes());
format!("magi-code-session-{}", lower_hex(digest))
.chars()
.take("magi-code-session-".len() + 32)
.collect()
}
fn prompt_cache_key_for_subagent(provider_id: &str, model: &str, agent_id: &str) -> String {
let material = format!("provider={provider_id}\nmodel={model}\nidentity={agent_id}");
let digest = Sha256::digest(material.as_bytes());
format!("magi-code-session-{}", lower_hex(digest))
.chars()
.take("magi-code-session-".len() + 32)
.collect()
}
fn prompt_cache_key_for_run(
run: &AgentRunRequest<'_, '_>,
provider_id: &str,
model: &str,
) -> Option<String> {
match run.invocation_mode {
InvocationMode::Subagent => run
.agent_id
.as_deref()
.map(|agent_id| prompt_cache_key_for_subagent(provider_id, model, agent_id))
.or_else(|| {
run.session
.map(|session| prompt_cache_key_for_session_id(session.id()))
}),
InvocationMode::Print | InvocationMode::Shell | InvocationMode::MissionControl => run
.session
.map(|session| prompt_cache_key_for_session_id(session.id())),
}
}
fn join_title_if_finished(title_handle: &mut Option<std::thread::JoinHandle<()>>) -> bool {
if title_handle
.as_ref()
.is_some_and(std::thread::JoinHandle::is_finished)
{
let _ = title_handle.take().expect("checked title handle").join();
true
} else {
false
}
}
fn finish_title_generation_for_mode(
title_handle: &mut Option<std::thread::JoinHandle<()>>,
invocation_mode: InvocationMode,
) {
match invocation_mode {
InvocationMode::Print => {
if !join_title_if_finished(title_handle) {
let _ = title_handle.take();
}
}
InvocationMode::Shell | InvocationMode::MissionControl | InvocationMode::Subagent => {
join_title_if_finished(title_handle);
}
}
}
struct TitleGenerationGuard {
handle: Option<std::thread::JoinHandle<()>>,
invocation_mode: InvocationMode,
cancellation: AgentCancellation,
}
impl TitleGenerationGuard {
fn new(
handle: Option<std::thread::JoinHandle<()>>,
invocation_mode: InvocationMode,
cancellation: AgentCancellation,
) -> Self {
Self {
handle,
invocation_mode,
cancellation,
}
}
fn finish(&mut self) {
finish_title_generation_for_mode(&mut self.handle, self.invocation_mode);
}
}
impl Drop for TitleGenerationGuard {
fn drop(&mut self) {
if !self.cancellation.is_canceled() {
self.finish();
}
}
}
struct RequestTokenProjectionCache {
provider_id: String,
model: String,
base_projection: ContextTokenCount,
projected_turn_items: usize,
projected_turn_tokens: usize,
calibrated_turn_items: Option<usize>,
calibrated_tokens: usize,
}
impl RequestTokenProjectionCache {
#[cfg(test)]
fn new(provider_id: &str, model: &str, base_items: &[ProviderConversationItem]) -> Self {
Self::new_with_base_projection(
provider_id,
model,
project_provider_conversation_items_tokens(provider_id, model, base_items),
)
}
fn new_for_request(provider_id: &str, request: &ProviderRequest) -> Self {
Self::new_with_base_projection(
provider_id,
&request.model,
project_provider_request_input_tokens(provider_id, request),
)
}
fn new_with_base_projection(
provider_id: &str,
model: &str,
base_projection: ContextTokenCount,
) -> Self {
Self {
provider_id: provider_id.to_string(),
model: model.to_string(),
base_projection,
projected_turn_items: 0,
projected_turn_tokens: 0,
calibrated_turn_items: None,
calibrated_tokens: 0,
}
}
}
impl RequestTokenProjectionCache {
fn initial_projection(&self) -> ContextTokenCount {
self.base_projection
}
fn project_turn_items(&mut self, turn_items: &[ProviderConversationItem]) -> ContextTokenCount {
if let Some(calibrated_turn_items) = self.calibrated_turn_items
&& calibrated_turn_items <= turn_items.len()
{
let delta_projection = project_provider_conversation_items_tokens(
&self.provider_id,
&self.model,
&turn_items[calibrated_turn_items..],
);
return ContextTokenCount {
tokens: self
.calibrated_tokens
.saturating_add(delta_projection.tokens),
source: ContextUsageSource::LastProviderUsage,
};
}
if turn_items.len() < self.projected_turn_items {
self.projected_turn_items = 0;
self.projected_turn_tokens = 0;
}
if self.projected_turn_items < turn_items.len() {
let new_projection = project_provider_conversation_items_tokens(
&self.provider_id,
&self.model,
&turn_items[self.projected_turn_items..],
);
self.projected_turn_tokens = self
.projected_turn_tokens
.saturating_add(new_projection.tokens);
self.projected_turn_items = turn_items.len();
}
ContextTokenCount {
tokens: self
.base_projection
.tokens
.saturating_add(self.projected_turn_tokens),
source: self.base_projection.source,
}
}
fn calibrate_anthropic_input_tokens(&mut self, turn_item_count: usize, input_tokens: usize) {
if !matches!(
self.provider_id.as_str(),
crate::providers::ANTHROPIC_PROVIDER | crate::providers::CLAUDE_CODE_PROVIDER
) {
return;
}
self.calibrated_turn_items = Some(turn_item_count);
self.calibrated_tokens = input_tokens;
self.projected_turn_items = turn_item_count;
self.projected_turn_tokens = input_tokens.saturating_sub(self.base_projection.tokens);
}
}
fn disabled_tool_names_for_request(
tools: Option<&ToolRuntime>,
) -> std::collections::HashSet<String> {
let Some(tools) = tools else {
return Default::default();
};
tools.disabled_tool_names().unwrap_or_else(|_| {
crate::tools::MVP_TOOL_CAPABILITIES
.iter()
.map(|tool| tool.canonical_name().to_string())
.chain(
tools
.dynamic_provider_tool_definitions()
.into_iter()
.filter_map(|definition| {
definition
.get("name")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
}),
)
.collect()
})
}
impl AgentSession {
#[cfg(test)]
pub fn new(
model: impl Into<String>,
instructions: &[InstructionFile],
skills: &SkillDiscovery,
) -> Self {
Self::try_new(model, instructions, skills)
.expect("bundled prompt templates must render successfully")
}
#[cfg(test)]
pub fn try_new(
model: impl Into<String>,
instructions: &[InstructionFile],
skills: &SkillDiscovery,
) -> anyhow::Result<Self> {
Self::new_with_prompt_dir(model, None, instructions, skills)
}
#[cfg(test)]
pub fn new_with_prompt_dir(
model: impl Into<String>,
prompt_dir: Option<&Path>,
instructions: &[InstructionFile],
skills: &SkillDiscovery,
) -> anyhow::Result<Self> {
Self::new_with_prompt_dir_and_subagents(model, prompt_dir, instructions, skills, None)
}
pub(crate) fn new_with_prompt_dir_and_subagents(
model: impl Into<String>,
prompt_dir: Option<&Path>,
instructions: &[InstructionFile],
skills: &SkillDiscovery,
subagents_section: Option<&str>,
) -> anyhow::Result<Self> {
Self::new_with_prompt_dir_and_subagents_and_disabled(
model,
prompt_dir,
instructions,
skills,
subagents_section,
&std::collections::HashSet::new(),
)
}
pub(crate) fn new_with_prompt_dir_and_subagents_and_disabled(
model: impl Into<String>,
prompt_dir: Option<&Path>,
instructions: &[InstructionFile],
skills: &SkillDiscovery,
subagents_section: Option<&str>,
disabled_tools: &std::collections::HashSet<String>,
) -> anyhow::Result<Self> {
let system_prompt = match (prompt_dir, subagents_section) {
(Some(prompt_dir), subagents_section) => {
build_system_prompt_with_prompt_dir_and_subagents_and_disabled(
Some(prompt_dir),
instructions,
skills,
subagents_section,
disabled_tools,
)?
}
(None, None) if disabled_tools.is_empty() => build_system_prompt(instructions, skills)?,
(None, subagents_section) => {
build_system_prompt_with_prompt_dir_and_subagents_and_disabled(
None,
instructions,
skills,
subagents_section,
disabled_tools,
)?
}
};
Ok(Self::from_system_prompt(
system_prompt,
AgentSessionConfig::for_model(model),
))
}
fn from_system_prompt(system_prompt: String, config: AgentSessionConfig) -> Self {
Self {
provider_id: config.provider_id,
model: config.model,
system_prompt,
context_budget: config.context_budget,
thinking_level: config.thinking_level,
text_verbosity: config.text_verbosity,
thinking_levels: config.thinking_levels,
send_default_reasoning_summary: config.send_default_reasoning_summary,
context_cache_dir: config.context_cache_dir,
}
}
pub(crate) fn with_config(mut self, config: AgentSessionConfig) -> Self {
self.model = config.model;
self.context_cache_dir = None;
let mut agent = self
.with_provider_id(config.provider_id)
.with_thinking_level(config.thinking_level)
.with_text_verbosity(config.text_verbosity)
.with_thinking_levels(config.thinking_levels)
.with_default_reasoning_summary(config.send_default_reasoning_summary)
.with_context_budget(config.context_budget);
if let Some(cache_dir) = config.context_cache_dir {
agent = agent.with_context_cache_dir(cache_dir);
}
agent
}
pub(crate) fn system_prompt(&self) -> &str {
&self.system_prompt
}
pub(crate) fn with_appended_system_prompt(&self, section: &str) -> Self {
let mut cloned = self.clone();
let section = section.trim();
if !section.is_empty() {
cloned.system_prompt.push_str("\n\n");
cloned.system_prompt.push_str(section);
}
cloned
}
pub fn with_provider_id(mut self, provider_id: impl Into<String>) -> Self {
self.provider_id = normalize_agent_identifier(provider_id.into(), "provider");
self
}
pub fn with_context_budget(mut self, budget: ContextBudget) -> Self {
self.context_budget = budget;
self
}
pub fn with_context_cache_dir(mut self, cache_dir: PathBuf) -> Self {
self.context_cache_dir = Some(cache_dir);
self
}
pub(crate) fn with_model(mut self, model: impl Into<String>) -> Self {
self.model = normalize_agent_identifier(model.into(), "model");
self
}
pub(crate) fn with_thinking_level(
mut self,
thinking_level: crate::thinking::ThinkingLevel,
) -> Self {
self.thinking_level = thinking_level;
self
}
pub(crate) fn with_text_verbosity(mut self, text_verbosity: Option<TextVerbosity>) -> Self {
self.text_verbosity = text_verbosity;
self
}
pub(crate) fn with_thinking_levels(
mut self,
thinking_levels: Vec<crate::thinking::ThinkingLevel>,
) -> Self {
self.thinking_levels = crate::thinking::normalize_thinking_levels(&thinking_levels);
self.thinking_level =
crate::thinking::resolve_thinking_level(&self.thinking_levels, self.thinking_level);
self
}
pub(crate) fn with_default_reasoning_summary(mut self, supported: bool) -> Self {
self.send_default_reasoning_summary = supported;
self
}
pub(crate) fn with_reasoning_override(
mut self,
thinking_level: crate::thinking::ThinkingLevel,
) -> Self {
self.thinking_level =
crate::thinking::resolve_thinking_level(&self.thinking_levels, thinking_level);
self
}
pub(crate) fn with_provider_model_reasoning(
self,
provider_id: impl Into<String>,
model: impl Into<String>,
thinking_levels: Vec<crate::thinking::ThinkingLevel>,
thinking_level: Option<crate::thinking::ThinkingLevel>,
) -> Self {
let selected_thinking = thinking_level.unwrap_or(self.thinking_level);
let send_default_reasoning_summary = thinking_levels.len() > 1;
let effective_thinking =
crate::thinking::resolve_thinking_level(&thinking_levels, selected_thinking);
self.with_provider_id(provider_id)
.with_model(model)
.with_thinking_levels(thinking_levels)
.with_default_reasoning_summary(send_default_reasoning_summary)
.with_thinking_level(effective_thinking)
}
#[cfg(test)]
pub fn run_print_with_tools<P: Provider + ?Sized>(
&self,
provider: &P,
prompt: &str,
tools: Option<&ToolRuntime>,
session: Option<&Session>,
cwd: &Path,
) -> anyhow::Result<AgentRunOutput> {
self.run_print_with_tools_streaming(provider, prompt, tools, session, cwd, None)
}
#[cfg(test)]
pub fn run_print_with_tools_streaming<P: Provider + ?Sized>(
&self,
provider: &P,
prompt: &str,
tools: Option<&ToolRuntime>,
session: Option<&Session>,
cwd: &Path,
on_text_delta: Option<&mut AssistantTextSink<'_>>,
) -> anyhow::Result<AgentRunOutput> {
match on_text_delta {
Some(sink) => {
let mut sink = AssistantOnlySink {
on_text_delta: sink,
};
self.run_print_with_tools_streaming_output(
provider,
prompt,
tools,
session,
cwd,
Some(&mut sink),
)
}
None => self
.run_print_with_tools_streaming_output(provider, prompt, tools, session, cwd, None),
}
}
pub(crate) fn run_print_with_tools_streaming_output_cancellable<P: Provider + ?Sized>(
&self,
provider: &P,
request: AgentRunRequest<'_, '_>,
) -> anyhow::Result<AgentRunOutput> {
self.run_print_with_tools_streaming_output_inner(provider, request, None, false)
}
pub(crate) fn run_print_with_tools_streaming_output_cancellable_with_steering<
P: Provider + ?Sized,
>(
&self,
provider: &P,
request: AgentRunRequest<'_, '_>,
steering: AgentSteering,
) -> anyhow::Result<AgentRunOutput> {
self.run_print_with_tools_streaming_output_inner(provider, request, Some(steering), false)
}
pub(crate) fn run_print_with_tools_streaming_output_cancellable_with_deferred_steering<
P: Provider + ?Sized,
>(
&self,
provider: &P,
request: AgentRunRequest<'_, '_>,
steering: AgentSteering,
) -> anyhow::Result<AgentRunOutput> {
self.run_print_with_tools_streaming_output_inner(provider, request, Some(steering), true)
}
#[cfg(test)]
pub fn run_print_with_tools_streaming_output<P: Provider + ?Sized>(
&self,
provider: &P,
prompt: &str,
tools: Option<&ToolRuntime>,
session: Option<&Session>,
cwd: &Path,
output_sink: Option<&mut dyn AgentOutputSink>,
) -> anyhow::Result<AgentRunOutput> {
self.run_print_with_tools_streaming_output_inner(
provider,
AgentRunRequest {
prompt,
prompt_origin: crate::output::UserPromptOrigin::User,
effective_prompt: None,
tools,
hooks: None,
session,
cwd,
output_sink,
cancellation: AgentCancellation::default(),
session_title_job: None,
invocation_mode: InvocationMode::Print,
semantic_progress_timeout: None,
agent_id: None,
initial_instructions: &[],
ttsr: crate::config::TtsrSettings::default(),
herdr_reporter: None,
continuation_auto_compaction_policy: None,
},
None,
false,
)
}
fn run_print_with_tools_streaming_output_inner<P: Provider + ?Sized>(
&self,
provider: &P,
mut run: AgentRunRequest<'_, '_>,
steering: Option<AgentSteering>,
defer_final_steering: bool,
) -> anyhow::Result<AgentRunOutput> {
let cancellation = run.cancellation.clone();
cancellation.check()?;
let mut ttsr_rules = ttsr::TtsrRuleSet::new(&run.ttsr, run.initial_instructions)?;
let mut prepared = self.prepare_initial_run(&run)?;
self.record_initial_user_input_and_cache(&mut prepared, &mut run)?;
if matches!(run.prompt_origin, crate::output::UserPromptOrigin::Steering)
&& let Some(steering) = steering.as_ref()
{
steering.acknowledge_collapsed(run.prompt);
}
let title_guard = self.start_title_generation_guard(&mut run, cancellation.clone());
let replay_diagnostics = prepared
.initial_conversation
.session_read_diagnostics
.clone();
let mut state = self.start_print_turn_state(prepared, &run, title_guard);
let mut ttsr_retry_count = 0_u8;
let mut auto_continue_used = false;
macro_rules! terminal_try {
($expression:expr) => {
match $expression {
Ok(value) => value,
Err(error) => {
let _ = self.record_terminal_failure(&mut state, &mut run, &error);
return Err(error);
}
}
};
}
terminal_try!(self.update_tool_checkpoint_context(&run));
terminal_try!(
self.emit_initial_user_prompt_and_replay_diagnostics(&mut run, &replay_diagnostics,)
);
loop {
terminal_try!(cancellation.check());
let (request, request_sequence, request_input_tokens, request_turn_item_count) =
match self.build_turn_request_and_emit_context_usage(&mut state, &mut run) {
Ok(request) => request,
Err(error) => {
state.output.persistence_degraded = state.session_persistence.is_degraded();
let is_recoverable_continuation_budget =
!state.output.persistence_degraded
&& run.continuation_auto_compaction_policy.is_some()
&& error.downcast_ref::<ContextBudgetError>().is_some_and(
|budget| budget.phase() == ContextBudgetPhase::Continuation,
);
if is_recoverable_continuation_budget {
let _ = state.session_persistence.record_terminal_status(
"compaction_required",
&state.output.text,
&mut run.output_sink,
);
} else {
let _ = self.record_terminal_failure(&mut state, &mut run, &error);
}
state.output.persistence_degraded = state.session_persistence.is_degraded();
return match error.downcast::<ContextBudgetError>() {
Ok(budget) if budget.phase() == ContextBudgetPhase::Continuation => {
Err(budget
.with_partial_output(std::mem::take(&mut state.output))
.into())
}
Ok(budget) => Err(budget.into()),
Err(error) => Err(error),
};
}
};
let (request_cancellation, request_cancel_handle) = cancellation.child_token();
let stream_result = match self.collect_provider_turn(
provider,
request,
&mut state,
&mut run,
&request_cancellation,
request_sequence,
request_input_tokens,
ttsr_rules.as_mut(),
&request_cancel_handle,
) {
Ok(stream_result) => {
ttsr_retry_count = 0;
stream_result
}
Err(failure) if failure.ttsr_match.is_some() && ttsr_retry_count < 3 => {
terminal_try!(cancellation.check());
ttsr_retry_count = ttsr_retry_count.saturating_add(1);
state.output.auto_compaction_blocked_by_recovery = true;
let _ = self.record_ttsr_injection_and_retry(
&failure,
&mut state,
&mut run,
request_sequence,
);
if let Some(ttsr_rules) = ttsr_rules.as_mut() {
ttsr_rules.reset_stream_buffer();
}
continue;
}
Err(failure) if failure.ttsr_match.is_some() => {
let _ = self.record_ttsr_retry_exhaustion(&failure, &mut state, &mut run);
self.record_provider_stream_failure(&failure, &mut state, &mut run);
return Err(failure.error);
}
Err(failure) if !auto_continue_used && failure.eligible_for_auto_continue() => {
terminal_try!(cancellation.check());
let _ = self.record_auto_continue_decision(
&failure,
"continuing",
None,
&mut state,
&mut run,
);
terminal_try!(self.auto_continue_after_incomplete_stream(&mut state, &mut run));
state.output.recovered_incomplete_stream = true;
state.output.auto_compaction_blocked_by_recovery = true;
auto_continue_used = true;
continue;
}
Err(failure) => {
let reason = if auto_continue_used {
Some("auto-continue already used for this turn")
} else {
failure.auto_continue_block_reason()
};
if failure.recovery_diagnostic_relevant() {
let _ = self.record_auto_continue_decision(
&failure, "blocked", reason, &mut state, &mut run,
);
}
self.record_provider_stream_failure(&failure, &mut state, &mut run);
return Err(failure.error);
}
};
if let Err(error) = self.run_reasoning_message_hooks(
&stream_result.reasoning_summaries,
&mut state,
&mut run,
&cancellation,
) {
let _ = self.record_terminal_failure(&mut state, &mut run, &error);
return Err(error);
}
if let Some(input_tokens) = stream_result.provider_input_tokens {
state
.request_projection_cache
.calibrate_anthropic_input_tokens(request_turn_item_count, input_tokens);
}
if stream_result.tool_calls.is_empty()
&& state.turn_state.assistant_segment().trim().is_empty()
&& !stream_result.reasoning_summaries.is_empty()
&& !auto_continue_used
{
terminal_try!(self.auto_continue_after_reasoning_only_turn(&mut state, &mut run));
state.output.auto_compaction_blocked_by_recovery = true;
auto_continue_used = true;
continue;
}
if stream_result.tool_calls.is_empty() {
terminal_try!(self.complete_assistant_turn_without_tools(
&mut state,
&mut run,
&cancellation,
));
terminal_try!(cancellation.check());
if !defer_final_steering {
let steering_batch =
steering.as_ref().and_then(AgentSteering::observe_collapsed);
if steering_batch.is_some() {
state.turn_state.finish_text_action_for_continuation();
if terminal_try!(inject_steering_update_at_continuation_boundary(
steering.as_ref(),
steering_batch,
&mut state.turn_state,
&state.session_persistence,
&mut run.output_sink,
)) {
continue;
}
}
}
state.herdr_turn.done();
break;
}
terminal_try!(self.run_tool_turn_and_maybe_emit_assistant_complete(
stream_result.tool_calls,
&mut state,
&mut run,
&cancellation,
));
if let Err(error) = cancellation.check() {
let _ = self.record_cancelled_terminal_status(&mut state, &mut run);
return Err(error);
}
let steering_batch = steering.as_ref().and_then(AgentSteering::observe_collapsed);
if !state.session_persistence.is_degraded()
&& !state.output.auto_compaction_blocked_by_recovery
&& let Some((soft_threshold, _display)) =
run.continuation_auto_compaction_policy.as_ref()
{
let mut projected_items = state.turn_state.request_items_slice().to_vec();
if let Some(batch) = steering_batch.as_ref() {
projected_items.push(ProviderConversationItem::Message(ChatMessage::user(
batch.text.clone(),
)));
}
let projected_request = self.request_from_shared_conversation(
Arc::clone(&state.base_conversation),
&projected_items,
run.semantic_progress_timeout,
state.prompt_cache_key.as_deref(),
run.tools,
);
let projection =
project_provider_request_input_tokens(&self.provider_id, &projected_request);
if projection.tokens >= *soft_threshold {
let _ = state.session_persistence.record_terminal_status(
"compaction_required",
&state.output.text,
&mut run.output_sink,
);
state.output.persistence_degraded = state.session_persistence.is_degraded();
return Err(ContextBudgetError::new(
ContextBudgetPhase::Continuation,
projection.tokens,
*soft_threshold,
self.context_budget.max_tokens,
self.context_budget.reserve_tokens,
)
.with_partial_output(std::mem::take(&mut state.output))
.into());
}
}
terminal_try!(inject_steering_update_at_continuation_boundary(
steering.as_ref(),
steering_batch,
&mut state.turn_state,
&state.session_persistence,
&mut run.output_sink,
));
}
state.title_guard.finish();
state.output.persistence_degraded = state.session_persistence.is_degraded();
Ok(state.output)
}
fn prepare_initial_run<'run, 'sink>(
&self,
run: &AgentRunRequest<'run, 'sink>,
) -> anyhow::Result<PreparedInitialRun<'run>> {
let effective_prompt = run
.effective_prompt
.map(str::to_string)
.or_else(|| self.effective_prompt_for_run(run));
let provider_prompt = effective_prompt
.as_deref()
.unwrap_or(run.prompt)
.to_string();
let initial_conversation =
self.build_initial_conversation(&provider_prompt, run.session)?;
let base_conversation =
Arc::<[ProviderConversationItem]>::from(initial_conversation.conversation.clone());
let prompt_cache_key = prompt_cache_key_for_run(run, &self.provider_id, &self.model);
let initial_request = self.request_from_conversation(
initial_conversation.conversation.clone(),
None,
prompt_cache_key.as_deref(),
run.tools,
);
let request_projection_cache =
RequestTokenProjectionCache::new_for_request(&self.provider_id, &initial_request);
let initial_context_projection = (self.context_budget.enabled
|| self.context_cache_dir.is_some())
.then(|| {
self.ensure_request_context_fits(
request_projection_cache.initial_projection(),
ContextBudgetPhase::Initial,
)
})
.transpose()?;
Ok(PreparedInitialRun {
provider_prompt,
initial_conversation,
base_conversation,
prompt_cache_key,
initial_request,
initial_context_projection,
request_projection_cache,
session_persistence: SessionPersistence::new(run.session, run.cwd),
})
}
fn effective_prompt_for_run(&self, run: &AgentRunRequest<'_, '_>) -> Option<String> {
Self::expand_effective_prompt(run.prompt, run.tools, run.invocation_mode)
}
pub(crate) fn effective_prompt_for_projection(
prompt: &str,
tools: Option<&ToolRuntime>,
invocation_mode: InvocationMode,
) -> String {
Self::expand_effective_prompt(prompt, tools, invocation_mode)
.unwrap_or_else(|| prompt.to_string())
}
fn expand_effective_prompt(
prompt: &str,
tools: Option<&ToolRuntime>,
invocation_mode: InvocationMode,
) -> Option<String> {
should_expand_prompt_context(invocation_mode).then(|| {
tools.and_then(|tools| {
crate::agent::prompt_injections::expand_prompt_with_context_injections(
prompt,
|definition| {
tools.dispatch(
"bash",
json!({"command": definition.command, "timeout": 30}),
)
},
)
})
})?
}
fn record_initial_user_input_and_cache(
&self,
prepared: &mut PreparedInitialRun<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
if let Some(initial_context_projection) = prepared.initial_context_projection
&& let Err(error) = self.record_context_cache(
&prepared.initial_request,
initial_context_projection.tokens,
run.session,
run.cwd,
)
{
prepared
.session_persistence
.warn_once(&mut run.output_sink, &error)?;
}
prepared
.session_persistence
.record_required(
SessionEventKind::UserInput,
json!({"text": prepared.provider_prompt, "origin": run.prompt_origin.label()}),
)
.map_err(RequiredUserInputPersistenceError::new)?;
Ok(())
}
fn update_tool_checkpoint_context(&self, run: &AgentRunRequest<'_, '_>) -> anyhow::Result<()> {
let Some(tools) = run.tools else {
return Ok(());
};
let Some(session) = run.session else {
tools.set_checkpoint_context(None)?;
return Ok(());
};
let Some(paths) = tools.checkpoint_paths() else {
tools.set_checkpoint_context(None)?;
return Ok(());
};
let user_turn = crate::sessions::session_user_input_count(session)? as u64;
tools.set_checkpoint_context(Some(crate::checkpoints::SnapshotContext {
store: crate::checkpoints::CheckpointStore::from_paths(&paths),
paths,
session_id: session.id().to_string(),
user_turn,
}))
}
fn start_title_generation_guard(
&self,
run: &mut AgentRunRequest<'_, '_>,
cancellation: AgentCancellation,
) -> TitleGenerationGuard {
let mut title_handle = None;
if let (Some(session), Some(mut job)) = (run.session, run.session_title_job.take())
&& let Ok(metadata) =
crate::sessions::titles::should_start_title_generation_metadata(session)
&& metadata.latest_title.is_none()
&& metadata.user_input_count == 1
{
if let Some(first_prompt) = metadata.first_user_input_text {
job.first_prompt = first_prompt;
}
title_handle = Some(crate::sessions::titles::spawn_background(job));
}
TitleGenerationGuard::new(title_handle, run.invocation_mode, cancellation)
}
fn emit_initial_user_prompt_and_replay_diagnostics(
&self,
run: &mut AgentRunRequest<'_, '_>,
diagnostics: &[SessionReadDiagnostic],
) -> anyhow::Result<()> {
if let Some(sink) = run.output_sink.as_deref_mut() {
let event = match run.prompt_origin {
crate::output::UserPromptOrigin::AutomaticCompaction => {
OutputEvent::AutomaticUserPrompt {
text: run.prompt.to_string(),
}
}
crate::output::UserPromptOrigin::User
| crate::output::UserPromptOrigin::Steering => OutputEvent::UserPrompt {
text: run.prompt.to_string(),
},
};
sink.output_event(event)?;
}
emit_replay_diagnostics(&mut run.output_sink, diagnostics)
}
fn build_hook_context(&self, run: &AgentRunRequest<'_, '_>) -> HookContextMetadata {
HookContextMetadata {
session_id: run.session.map(|session| session.id().to_string()),
session_path: run.session.map(|session| session.path().to_path_buf()),
provider_id: Some(self.provider_id.clone()),
model_id: Some(self.model.clone()),
agent_id: run.agent_id.clone(),
invocation_mode: run.invocation_mode,
turn_id: None,
message_id: None,
subagent: matches!(run.invocation_mode, InvocationMode::Subagent),
}
}
fn start_print_turn_state<'run>(
&self,
prepared: PreparedInitialRun<'run>,
run: &AgentRunRequest<'run, '_>,
title_guard: TitleGenerationGuard,
) -> PrintTurnState<'run> {
PrintTurnState {
base_conversation: prepared.base_conversation,
prompt_cache_key: prepared.prompt_cache_key,
session_persistence: prepared.session_persistence,
subdir_instruction_state: SubdirInstructionState::new(
run.initial_instructions,
run.session,
),
assistant_chunk_batch: AssistantChunkBatch::new(run.session, run.cwd),
turn_state: AgentTurnState::default(),
output: AgentRunOutput::default(),
request_projection_cache: prepared.request_projection_cache,
request_sequence: 0,
hook_context: self.build_hook_context(run),
herdr_turn: HerdrTurnReporter::start(
(!matches!(run.invocation_mode, InvocationMode::Subagent))
.then(|| run.herdr_reporter.clone())
.flatten(),
),
title_guard,
}
}
fn build_turn_request_and_emit_context_usage(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<(ProviderRequest, u64, usize, usize)> {
let request_turn_item_count = state.turn_state.request_items_slice().len();
let request = self.request_from_shared_conversation(
Arc::clone(&state.base_conversation),
state.turn_state.request_items_slice(),
run.semantic_progress_timeout,
state.prompt_cache_key.as_deref(),
run.tools,
);
let context_projection = self.ensure_request_context_fits(
state
.request_projection_cache
.project_turn_items(state.turn_state.request_items_slice()),
ContextBudgetPhase::Continuation,
)?;
let current_request_sequence = state.request_sequence;
state.request_sequence = state.request_sequence.saturating_add(1);
if let Some(sink) = run.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::ContextUsage {
current_tokens: context_projection.tokens,
max_tokens: self.context_budget.max_tokens,
reasoning_tokens: None,
source: context_projection.source,
request_sequence: current_request_sequence,
})?;
}
Ok((
request,
current_request_sequence,
context_projection.tokens,
request_turn_item_count,
))
}
#[allow(clippy::too_many_arguments)]
fn collect_provider_turn<P: Provider + ?Sized>(
&self,
provider: &P,
request: ProviderRequest,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
cancellation: &AgentCancellation,
request_sequence: u64,
request_input_tokens: usize,
ttsr: Option<&mut ttsr::TtsrRuleSet>,
ttsr_cancel_handle: &crate::agent::cancellation::AgentCancellationHandle,
) -> Result<ProviderStreamResult, ProviderStreamFailure> {
collect_provider_events(
provider,
request,
ProviderEventCollector {
cancellation,
output_sink: &mut run.output_sink,
output: &mut state.output,
turn_state: &mut state.turn_state,
assistant_chunk_batch: &mut state.assistant_chunk_batch,
session_persistence: &mut state.session_persistence,
context_budget_max_tokens: self.context_budget.max_tokens,
provider_id: &self.provider_id,
model: &self.model,
request_sequence,
request_input_tokens,
ttsr,
ttsr_cancel_handle: Some(ttsr_cancel_handle),
},
)
}
fn record_auto_continue_decision(
&self,
failure: &ProviderStreamFailure,
decision: &str,
reason: Option<&str>,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
let mut message = format!("provider stream recovery decision: {decision}");
if let Some(reason) = reason {
message.push_str("; reason: ");
message.push_str(reason);
}
if failure.unsafe_tool_call_progress {
message.push_str("; partial tool-call progress observed and raw arguments omitted");
}
state.session_persistence.try_record(
SessionEventKind::Diagnostic,
json!({"level":"warning", "message": message}),
&mut run.output_sink,
)?;
if let Some(sink) = run.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::Diagnostic {
level: if decision == "continuing" {
"info"
} else {
"warning"
}
.to_string(),
message,
})?;
}
Ok(())
}
fn record_ttsr_injection_and_retry(
&self,
failure: &ProviderStreamFailure,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
request_sequence: u64,
) -> anyhow::Result<()> {
let Some(ttsr_match) = failure.ttsr_match.as_ref() else {
return Ok(());
};
state.session_persistence.try_record(
SessionEventKind::TtsrInjection,
json!({
"schema_version": 1,
"turn_index": state.turn_state.iteration(),
"request_sequence": request_sequence,
"rule_source": ttsr_match.source.as_str(),
"rule_pattern": ttsr_match.pattern,
"matched_text_redacted": ttsr_match.matched_text_redacted,
"reminder": ttsr_match.reminder,
"aborted_turn": state.turn_state.iteration(),
}),
&mut run.output_sink,
)?;
state
.turn_state
.append_ttsr_system_reminder(ttsr_match.reminder.clone());
Ok(())
}
fn record_ttsr_retry_exhaustion(
&self,
failure: &ProviderStreamFailure,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
let Some(ttsr_match) = failure.ttsr_match.as_ref() else {
return Ok(());
};
state.session_persistence.try_record(
SessionEventKind::Diagnostic,
json!({
"level": "warning",
"message": format!("TTSR retry cap exhausted for rule_source={} matched_text={}", ttsr_match.source.as_str(), ttsr_match.matched_text_redacted)
}),
&mut run.output_sink,
)
}
fn record_provider_stream_failure(
&self,
failure: &ProviderStreamFailure,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) {
let status = if failure.cancelled {
"cancelled"
} else {
"failed"
};
let _ = state.session_persistence.record_terminal_status(
status,
&state.output.text,
&mut run.output_sink,
);
if let Some(payload) = failure.provider_stream_trace_payload() {
let _ = state
.session_persistence
.record_provider_stream_trace(payload, &mut run.output_sink);
}
if let Some(payload) = failure.recovery_payload() {
let _ = state
.session_persistence
.record_abort_recovery(payload, &mut run.output_sink);
}
}
fn record_terminal_failure(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
error: &anyhow::Error,
) -> anyhow::Result<()> {
let status = if is_run_canceled(error) {
"cancelled"
} else {
"failed"
};
state.session_persistence.record_terminal_status(
status,
&state.output.text,
&mut run.output_sink,
)
}
fn record_cancelled_terminal_status(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
state.session_persistence.record_terminal_status(
"cancelled",
&state.output.text,
&mut run.output_sink,
)
}
fn auto_continue_after_incomplete_stream(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
self.auto_continue_turn(
state,
run,
"incomplete_provider_stream",
"Provider stream ended before completion after partial response; auto-continuing once.",
false,
)
}
fn auto_continue_after_reasoning_only_turn(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
self.auto_continue_turn(
state,
run,
"reasoning_only_no_output",
"Provider returned reasoning with no text or tool calls; auto-continuing once.",
true,
)
}
fn auto_continue_turn(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
reason: &str,
diagnostic: &str,
persist_prompt: bool,
) -> anyhow::Result<()> {
if let Err(error) = state.assistant_chunk_batch.flush() {
state
.session_persistence
.warn_once(&mut run.output_sink, &error)?;
}
state.turn_state.append_auto_continue();
if persist_prompt {
state.session_persistence.record_required(
SessionEventKind::UserInput,
json!({
"text": "Continue",
"auto_recovery": true,
"reason": reason
}),
)?;
}
if let Some(sink) = run.output_sink.as_deref_mut() {
let event = if persist_prompt {
OutputEvent::UserPrompt {
text: "Continue".to_string(),
}
} else {
OutputEvent::AutomaticUserPrompt {
text: "Continue".to_string(),
}
};
sink.output_event(event)?;
sink.output_event(OutputEvent::Diagnostic {
level: "info".to_string(),
message: diagnostic.to_string(),
})?;
}
Ok(())
}
fn run_reasoning_message_hooks(
&self,
reasoning_summaries: &[String],
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
cancellation: &AgentCancellation,
) -> anyhow::Result<()> {
for reasoning_text in reasoning_summaries {
let message_id = format!("reasoning-turn-{}", state.turn_state.iteration());
let hook_outcome = run_message_phase_hooks(
run.hooks,
HookPhase::AfterReasoning,
&message_id,
reasoning_text,
&state.hook_context.for_tool_call(
format!("turn-{}", state.turn_state.iteration()),
&message_id,
),
cancellation,
&mut state.session_persistence,
&mut run.output_sink,
)?;
for item in &hook_outcome.context_items {
record_provider_context_item(
&mut state.session_persistence,
&mut run.output_sink,
item,
)?;
}
state
.turn_state
.append_provider_context_items(hook_outcome.context_items);
if let Some(diagnostic) = hook_outcome.failure {
return Err(HookPolicyError::new(diagnostic).into());
}
cancellation.check()?;
}
Ok(())
}
fn complete_assistant_turn_without_tools(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
cancellation: &AgentCancellation,
) -> anyhow::Result<()> {
let assistant_text = state.turn_state.assistant_segment().to_string();
if let Err(error) = cancellation.check() {
self.record_cancelled_terminal_status(state, run)?;
return Err(error);
}
let assistant_complete_error = if let Some(sink) = run.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::AssistantComplete {
text: assistant_text.clone(),
})
.err()
} else {
None
};
if let Err(error) = state.assistant_chunk_batch.flush() {
state
.session_persistence
.warn_once(&mut run.output_sink, &error)?;
}
if !assistant_text.is_empty() && assistant_text.trim().is_empty() {
state.output.text.clear();
} else {
state.session_persistence.try_record(
SessionEventKind::AssistantOutput,
json!({"text": assistant_text, "usage": state.output.usage}),
&mut run.output_sink,
)?;
}
if let Some(error) = assistant_complete_error {
return Err(error);
}
if !assistant_text.is_empty() {
let assistant_message_id = format!("assistant-turn-{}", state.turn_state.iteration());
let hook_outcome = run_message_phase_hooks(
run.hooks,
HookPhase::AfterAssistant,
&assistant_message_id,
&assistant_text,
&state.hook_context.for_tool_call(
format!("turn-{}", state.turn_state.iteration()),
&assistant_message_id,
),
cancellation,
&mut state.session_persistence,
&mut run.output_sink,
)?;
for item in &hook_outcome.context_items {
record_provider_context_item(
&mut state.session_persistence,
&mut run.output_sink,
item,
)?;
}
state
.turn_state
.append_provider_context_items(hook_outcome.context_items);
if let Some(diagnostic) = hook_outcome.failure {
return Err(HookPolicyError::new(diagnostic).into());
}
if let Err(error) = cancellation.check() {
self.record_cancelled_terminal_status(state, run)?;
return Err(error);
}
}
Ok(())
}
fn prepare_tool_turn_assistant_complete(
&self,
state: &mut PrintTurnState<'_>,
) -> Option<String> {
if state.turn_state.assistant_segment().trim().is_empty() {
return None;
}
state.turn_state.prepare_tool_turn();
Some(state.turn_state.assistant_segment().to_string())
}
fn run_tool_turn_and_maybe_emit_assistant_complete<'run, 'sink>(
&self,
tool_calls: Vec<ToolCall>,
state: &mut PrintTurnState<'run>,
run: &mut AgentRunRequest<'run, 'sink>,
cancellation: &AgentCancellation,
) -> anyhow::Result<()> {
let tool_turn_assistant_complete = self.prepare_tool_turn_assistant_complete(state);
let assistant_complete_error = if let Some(text) = tool_turn_assistant_complete
&& let Some(sink) = run.output_sink.as_deref_mut()
{
sink.output_event(OutputEvent::AssistantComplete { text })
.err()
} else {
None
};
if let Err(error) = state.assistant_chunk_batch.flush() {
state
.session_persistence
.warn_once(&mut run.output_sink, &error)?;
}
if let Some(error) = assistant_complete_error {
return Err(error);
}
run_tool_lifecycle(ToolLifecycleRun {
tool_calls,
tools: run.tools,
hooks: run.hooks,
herdr_reporter: state.herdr_turn.reporter(),
output_sink: &mut run.output_sink,
cancellation,
turn_state: &mut state.turn_state,
output: &mut state.output,
session_persistence: &mut state.session_persistence,
subdir_instruction_state: Some(&mut state.subdir_instruction_state),
hook_context: state.hook_context.clone(),
})?;
state.turn_state.finish_tool_iteration();
Ok(())
}
fn project_prompt_input_token_count(
&self,
prompt: &str,
session: Option<&Session>,
tools: Option<&ToolRuntime>,
) -> anyhow::Result<ContextTokenCount> {
let conversation = self.build_initial_conversation(prompt, session)?;
let request = self.request_from_conversation(conversation.conversation, None, None, tools);
Ok(project_provider_request_input_tokens(
&self.provider_id,
&request,
))
}
pub(crate) fn project_prompt_input_tokens(
&self,
prompt: &str,
session: &Session,
tools: Option<&ToolRuntime>,
) -> anyhow::Result<usize> {
Ok(self
.project_prompt_input_token_count(prompt, Some(session), tools)?
.tokens)
}
pub(crate) fn project_prompt_input_tokens_without_session(
&self,
prompt: &str,
tools: Option<&ToolRuntime>,
) -> anyhow::Result<usize> {
Ok(self
.project_prompt_input_token_count(prompt, None, tools)?
.tokens)
}
pub(crate) fn ensure_prompt_context_fits(
&self,
prompt: &str,
session: &Session,
tools: Option<&ToolRuntime>,
) -> anyhow::Result<usize> {
Ok(self
.ensure_request_context_fits(
self.project_prompt_input_token_count(prompt, Some(session), tools)?,
ContextBudgetPhase::Initial,
)?
.tokens)
}
pub(crate) fn context_max_tokens(&self) -> usize {
self.context_budget.max_tokens
}
fn build_initial_conversation(
&self,
prompt: &str,
session: Option<&Session>,
) -> anyhow::Result<InitialConversation> {
let replay = build_conversation_replay(session)?;
let _legacy_lossy_events = replay.legacy_lossy_events;
let mut conversation = Vec::with_capacity(replay.items.len() + 2);
conversation.push(ProviderConversationItem::Message(ChatMessage::system(
self.system_prompt.clone(),
)));
conversation.extend(replay.items);
conversation.push(ProviderConversationItem::Message(ChatMessage::user(
prompt.to_string(),
)));
Ok(InitialConversation {
conversation,
session_read_diagnostics: replay.session_read_diagnostics,
})
}
fn request_from_conversation(
&self,
conversation: Vec<ProviderConversationItem>,
semantic_progress_timeout: Option<Duration>,
prompt_cache_key: Option<&str>,
tools: Option<&ToolRuntime>,
) -> ProviderRequest {
let disabled_tool_names = disabled_tool_names_for_request(tools);
let subagents_tool_enabled = tools
.map(ToolRuntime::subagents_schema_enabled)
.unwrap_or(true);
let mut request = ProviderRequest::from_conversation(self.model.clone(), conversation)
.with_thinking_level(self.thinking_level)
.with_text_verbosity(self.text_verbosity)
.with_default_reasoning_summary(self.send_default_reasoning_summary)
.with_subagents_tool_enabled(subagents_tool_enabled)
.with_disabled_tool_names(disabled_tool_names.into_iter().collect())
.with_dynamic_tool_definitions(
tools
.map(ToolRuntime::dynamic_provider_tool_definitions)
.unwrap_or_default(),
);
if let Some(prompt_cache_key) = prompt_cache_key {
request = request.with_prompt_cache_key(prompt_cache_key);
}
match semantic_progress_timeout {
Some(timeout) => request.with_semantic_progress_timeout(timeout),
None => request,
}
}
fn request_from_shared_conversation(
&self,
base_conversation: Arc<[ProviderConversationItem]>,
turn_items: &[ProviderConversationItem],
semantic_progress_timeout: Option<Duration>,
prompt_cache_key: Option<&str>,
tools: Option<&ToolRuntime>,
) -> ProviderRequest {
let disabled_tool_names = disabled_tool_names_for_request(tools);
let subagents_tool_enabled = tools
.map(ToolRuntime::subagents_schema_enabled)
.unwrap_or(true);
let mut request = ProviderRequest::from_shared_conversation(
self.model.clone(),
base_conversation,
turn_items,
)
.with_thinking_level(self.thinking_level)
.with_text_verbosity(self.text_verbosity)
.with_default_reasoning_summary(self.send_default_reasoning_summary)
.with_subagents_tool_enabled(subagents_tool_enabled)
.with_disabled_tool_names(disabled_tool_names.into_iter().collect())
.with_dynamic_tool_definitions(
tools
.map(ToolRuntime::dynamic_provider_tool_definitions)
.unwrap_or_default(),
);
if let Some(prompt_cache_key) = prompt_cache_key {
request = request.with_prompt_cache_key(prompt_cache_key);
}
match semantic_progress_timeout {
Some(timeout) => request.with_semantic_progress_timeout(timeout),
None => request,
}
}
fn ensure_request_context_fits(
&self,
projection: ContextTokenCount,
phase: ContextBudgetPhase,
) -> anyhow::Result<ContextTokenCount> {
if !self.context_budget.enabled {
return Ok(projection);
}
let estimated_tokens = projection.tokens;
let threshold = self.context_budget.threshold_tokens();
if estimated_tokens <= threshold {
return Ok(projection);
}
Err(ContextBudgetError::new(
phase,
estimated_tokens,
threshold,
self.context_budget.max_tokens,
self.context_budget.reserve_tokens,
)
.into())
}
fn record_context_cache(
&self,
request: &ProviderRequest,
token_estimate: usize,
session: Option<&Session>,
cwd: &Path,
) -> anyhow::Result<()> {
let Some(cache_dir) = &self.context_cache_dir else {
return Ok(());
};
let cache = ContextCache::new(cache_dir.clone());
let input_material = conversation_cache_material(
&self.provider_id,
&self.model,
&self.system_prompt,
request.conversation_items().as_ref(),
);
let key = ContextCache::key_for_material(&input_material, &[]);
let (status, error) = match cache.read(&key) {
Ok(Some(_)) => ("hit", None),
Ok(None) => match cache.write(&ContextCacheEntry {
key: key.clone(),
token_estimate,
messages: request.messages(),
input_material,
}) {
Ok(()) => ("miss_write", None),
Err(error) => ("write_error", Some(safe_error_message(&error))),
},
Err(error) => ("read_error", Some(safe_error_message(&error))),
};
let mut payload = json!({"status": status, "key": key, "token_estimate": token_estimate});
if let Some(error) = error {
payload["error"] = json!(error);
}
try_record_session_event(session, cwd, SessionEventKind::ContextCache, payload).map(|_| ())
}
}