use super::*;
pub(super) const DANGLING_TOOL_INTENT_DIAGNOSTIC: &str = "Provider said it would invoke tools but emitted no tool call; auto-continuing once (reason: dangling_tool_intent).";
#[derive(Clone, Copy)]
pub(super) enum AutoContinuePersistence {
Ephemeral,
Prompt,
PromptAndDiagnostic,
}
pub(super) 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>,
expand_prompt: impl FnOnce(&str) -> String,
) -> anyhow::Result<bool> {
let (Some(steering), Some(batch)) = (steering, batch) else {
return Ok(false);
};
let provider_text = expand_prompt(&batch.text);
let persisted = steering.persist_and_acknowledge(&batch, || {
session_persistence
.record_required(
SessionEventKind::UserInput,
json!({"text": provider_text, "origin": "steering"}),
)
.map_err(|error| {
anyhow::anyhow!(
"failed to persist steering input before provider continuation: {error}"
)
})
})?;
if !persisted {
return Ok(false);
}
if let Some(sink) = output_sink.as_deref_mut() {
sink.steering_prompts_acknowledged(&batch.original_prompts);
sink.output_event(OutputEvent::SteeringInjected {
text: provider_text.clone(),
})?;
}
turn_state.append_provider_context_items([ProviderConversationItem::Message(
ChatMessage::user(provider_text),
)]);
Ok(true)
}
#[derive(Debug)]
pub(crate) struct RequiredUserInputPersistenceError {
pub(super) message: String,
}
impl RequiredUserInputPersistenceError {
pub(super) 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 {}
impl AgentSession {
pub(super) 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(())
}
pub(super) fn record_provider_stream_failure(
&self,
failure: &ProviderStreamFailure,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) {
let status = if failure.cancelled {
TurnStatus::Cancelled
} else {
TurnStatus::Failed
};
let error = (!failure.cancelled).then_some(&failure.error);
let _ = state.session_persistence.record_terminal_status_with_error(
status,
&state.output.text,
error,
&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);
}
}
pub(super) fn record_terminal_failure(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
error: &anyhow::Error,
) -> anyhow::Result<()> {
let status = if is_run_canceled(error) {
TurnStatus::Cancelled
} else {
TurnStatus::Failed
};
state.session_persistence.record_terminal_status_with_error(
status,
&state.output.text,
(!is_run_canceled(error)).then_some(error),
&mut run.output_sink,
)
}
pub(super) fn record_cancelled_terminal_status(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
state.session_persistence.record_terminal_status(
TurnStatus::Cancelled,
&state.output.text,
&mut run.output_sink,
)
}
pub(super) fn auto_continue_after_incomplete_stream(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
self.auto_continue_turn(
state,
run,
"Continue",
"incomplete_provider_stream",
"Provider stream ended before completion after partial response; auto-continuing once.",
AutoContinuePersistence::Ephemeral,
)
}
pub(super) fn auto_continue_after_reasoning_only_turn(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
self.auto_continue_turn(
state,
run,
"Continue",
"reasoning_only_no_output",
"Provider returned reasoning with no text or tool calls; auto-continuing once.",
AutoContinuePersistence::Prompt,
)
}
pub(super) fn auto_continue_after_dangling_tool_intent(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
) -> anyhow::Result<()> {
self.auto_continue_turn(
state,
run,
DANGLING_TOOL_INTENT_PROMPT,
DANGLING_TOOL_INTENT_REASON,
DANGLING_TOOL_INTENT_DIAGNOSTIC,
AutoContinuePersistence::PromptAndDiagnostic,
)?;
state.turn_state.finish_dangling_tool_intent_recovery();
Ok(())
}
pub(super) fn auto_continue_turn(
&self,
state: &mut PrintTurnState<'_>,
run: &mut AgentRunRequest<'_, '_>,
prompt: &str,
reason: &str,
diagnostic: &str,
persistence: AutoContinuePersistence,
) -> anyhow::Result<()> {
let persist_prompt = !matches!(persistence, AutoContinuePersistence::Ephemeral);
let persist_diagnostic =
matches!(persistence, AutoContinuePersistence::PromptAndDiagnostic);
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(prompt);
if persist_prompt {
state.session_persistence.record_required(
SessionEventKind::UserInput,
json!({
"text": prompt,
"auto_recovery": true,
"reason": reason
}),
)?;
}
if persist_diagnostic {
state.session_persistence.try_record(
SessionEventKind::Diagnostic,
json!({
"level": "info",
"message": diagnostic,
"auto_recovery": true,
"reason": reason
}),
&mut run.output_sink,
)?;
}
if let Some(sink) = run.output_sink.as_deref_mut() {
let event = if persist_prompt {
OutputEvent::UserPrompt {
text: prompt.to_string(),
}
} else {
OutputEvent::AutomaticUserPrompt {
text: prompt.to_string(),
}
};
sink.output_event(event)?;
sink.output_event(OutputEvent::Diagnostic {
level: "info".to_string(),
message: diagnostic.to_string(),
})?;
}
Ok(())
}
pub(super) 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(())
}
}