use super::{
AgentOutputSink, AgentRunOutput, session_persistence::SessionPersistence,
turn_state::AgentTurnState, util::AssistantChunkBatch,
};
use crate::{
agent::ttsr::{TtsrInterrupted, TtsrMatch, TtsrRuleSet},
cancellation::{AgentCancellation, AgentCancellationHandle, AgentRunCanceled, is_run_canceled},
context::project_text_tokens,
output::{ContextUsageSource, OutputEvent, redact_sensitive_text, sanitize_display_text},
providers::{
ANTHROPIC_PROVIDER, Provider, ProviderEvent, ProviderRequest, ReasoningSummary, ToolCall,
Usage, UsageObservation,
error::{ProviderStreamTrace, incomplete_stream_error, provider_stream_trace_from_error},
},
sessions::SessionEventKind,
};
fn sanitize_provider_response_item(mut item: serde_json::Value) -> serde_json::Value {
fn remove_encrypted_content(value: &mut serde_json::Value) {
match value {
serde_json::Value::Object(object) => {
object.remove("encrypted_content");
for child in object.values_mut() {
remove_encrypted_content(child);
}
}
serde_json::Value::Array(items) => {
for item in items {
remove_encrypted_content(item);
}
}
_ => {}
}
}
remove_encrypted_content(&mut item);
item
}
use serde_json::json;
pub(super) struct ProviderEventCollector<'a, 'sink, 'run> {
pub(super) cancellation: &'a AgentCancellation,
pub(super) output_sink: &'a mut Option<&'sink mut dyn AgentOutputSink>,
pub(super) output: &'a mut AgentRunOutput,
pub(super) turn_state: &'a mut AgentTurnState,
pub(super) assistant_chunk_batch: &'a mut AssistantChunkBatch<'run>,
pub(super) session_persistence: &'a mut SessionPersistence<'run>,
pub(super) context_budget_max_tokens: usize,
pub(super) provider_id: &'a str,
pub(super) model: &'a str,
pub(super) request_sequence: u64,
pub(super) request_input_tokens: usize,
pub(super) ttsr: Option<&'a mut TtsrRuleSet>,
pub(super) ttsr_cancel_handle: Option<&'a AgentCancellationHandle>,
}
pub(super) struct ProviderStreamResult {
pub(super) tool_calls: Vec<ToolCall>,
pub(super) reasoning_summaries: Vec<String>,
pub(super) provider_input_tokens: Option<usize>,
}
pub(super) struct ProviderStreamFailure {
pub(super) error: anyhow::Error,
pub(super) partial_assistant_text: bool,
pub(super) observed_tool_call: bool,
pub(super) observed_function_response_item: bool,
pub(super) observed_reasoning_delta: bool,
pub(super) unsafe_tool_call_progress: bool,
pub(super) reasoning_preview: Option<String>,
pub(super) partial_tool_call_summary: Option<String>,
pub(super) provider_stream_trace: Option<Box<ProviderStreamTrace>>,
pub(super) cancelled: bool,
pub(super) ttsr_match: Option<Box<TtsrMatch>>,
}
impl ProviderStreamFailure {
pub(super) fn recovery_payload(&self) -> Option<serde_json::Value> {
if !self.observed_reasoning_delta && !self.unsafe_tool_call_progress {
return None;
}
let mut payload = json!({
"observed_reasoning_delta": self.observed_reasoning_delta,
"unsafe_tool_call_progress": self.unsafe_tool_call_progress,
});
if let Some(preview) = &self.reasoning_preview
&& !preview.trim().is_empty()
{
payload["reasoning_preview"] = json!(sanitize_recovery_preview(preview));
}
if let Some(summary) = &self.partial_tool_call_summary
&& !summary.trim().is_empty()
{
payload["partial_tool_call_summary"] = json!(summary);
}
Some(payload)
}
pub(super) fn provider_stream_trace_payload(&self) -> Option<serde_json::Value> {
self.provider_stream_trace
.as_deref()
.and_then(|trace| serde_json::to_value(trace).ok())
}
pub(super) fn eligible_for_auto_continue(&self) -> bool {
self.auto_continue_block_reason().is_none()
}
pub(super) fn recovery_diagnostic_relevant(&self) -> bool {
self.unsafe_tool_call_progress
|| self.provider_stream_trace.is_some()
|| incomplete_stream_error(&self.error)
}
pub(super) fn auto_continue_block_reason(&self) -> Option<&'static str> {
if self.cancelled {
return Some("run was cancelled");
}
if !incomplete_stream_error(&self.error) {
return Some("failure was not an incomplete provider stream");
}
if !self.partial_assistant_text && !self.unsafe_tool_call_progress {
return Some("no recoverable partial assistant or tool-call progress was available");
}
if self.observed_tool_call {
return Some(
"complete tool call was observed; automatic recovery must not rerun tools",
);
}
if self.observed_function_response_item {
return Some(
"provider function response item was observed; automatic recovery must not infer missing protocol state",
);
}
None
}
}
impl From<anyhow::Error> for ProviderStreamFailure {
fn from(error: anyhow::Error) -> Self {
let cancelled = is_run_canceled(&error);
let provider_stream_trace = provider_stream_trace_from_error(&error).map(Box::new);
let ttsr_match = ttsr_match_from_error(&error);
Self {
error,
partial_assistant_text: false,
observed_tool_call: false,
observed_function_response_item: false,
observed_reasoning_delta: false,
unsafe_tool_call_progress: false,
reasoning_preview: None,
partial_tool_call_summary: None,
provider_stream_trace,
cancelled,
ttsr_match,
}
}
}
fn normalized_usage(
provider_id: &str,
usage: &Usage,
cache_known: bool,
) -> crate::output::NormalizedUsageSnapshot {
crate::output::NormalizedUsageSnapshot {
effective_input: normalized_input_tokens(provider_id, usage),
output: usage.output,
cache_read: usage.cache_read,
cache_known,
}
}
fn normalized_input_tokens(provider_id: &str, usage: &Usage) -> u64 {
if provider_id == ANTHROPIC_PROVIDER {
usage
.input
.saturating_add(usage.cache_read)
.saturating_add(usage.cache_write)
} else {
usage.input
}
}
fn coverage_known(provider_id: &str, presence: crate::providers::UsagePresence) -> bool {
presence.input
&& presence.cache_read
&& (provider_id != ANTHROPIC_PROVIDER || presence.cache_write)
}
#[derive(Debug, Default)]
pub(crate) struct RequestTokenTotals {
completed: Option<u64>,
active: Option<(u64, u64)>,
}
impl RequestTokenTotals {
pub(crate) fn observe(&mut self, sequence: u64, total: u64) {
if let Some((previous, value)) = self.active {
if sequence < previous {
return;
}
if sequence > previous {
self.completed = Some(self.completed.unwrap_or(0).saturating_add(value));
}
}
self.active = Some((sequence, total));
}
pub(crate) fn total(&self) -> Option<u64> {
self.active
.map(|(_, value)| self.completed.unwrap_or(0).saturating_add(value))
.or(self.completed)
}
}
fn known_usage_total(
provider_id: &str,
usage: &Usage,
presence: crate::providers::UsagePresence,
) -> Option<u64> {
if presence.total {
Some(usage.total)
} else if presence.input && presence.output {
Some(derived_usage_total(provider_id, usage))
} else {
None
}
}
fn merge_usage(
latest: &mut Option<(Usage, crate::providers::UsagePresence)>,
usage: Usage,
presence: crate::providers::UsagePresence,
) -> (Usage, crate::providers::UsagePresence) {
let mut merged = latest.take().unwrap_or_default();
if presence.input {
merged.0.input = usage.input;
}
if presence.output {
merged.0.output = usage.output;
}
if presence.cache_read {
merged.0.cache_read = usage.cache_read;
}
if presence.cache_write {
merged.0.cache_write = usage.cache_write;
}
if presence.total {
merged.0.total = usage.total;
}
if presence.reasoning {
merged.0.reasoning_tokens = usage.reasoning_tokens;
}
merged.1.input |= presence.input;
merged.1.output |= presence.output;
merged.1.cache_read |= presence.cache_read;
merged.1.cache_write |= presence.cache_write;
merged.1.total |= presence.total;
merged.1.reasoning |= presence.reasoning;
*latest = Some(merged.clone());
merged
}
fn usage_observation(event: &ProviderEvent) -> Option<UsageObservation> {
match event {
ProviderEvent::UsageObserved(observation) => Some(observation.clone()),
ProviderEvent::Usage(usage) => Some(UsageObservation {
usage: usage.clone(),
presence: crate::providers::UsagePresence {
input: true,
output: true,
cache_read: true,
cache_write: true,
total: true,
reasoning: usage.reasoning_tokens.is_some(),
},
}),
ProviderEvent::UsagePartial(usage) => Some(UsageObservation {
usage: usage.clone(),
presence: crate::providers::UsagePresence {
input: usage.input != 0,
output: usage.output != 0,
cache_read: usage.cache_read != 0,
cache_write: usage.cache_write != 0,
total: usage.total != 0,
reasoning: usage.reasoning_tokens.is_some(),
},
}),
_ => None,
}
}
#[derive(Default)]
pub(crate) struct CompactionUsage {
latest: Option<(Usage, crate::providers::UsagePresence)>,
pub(crate) total: Option<u64>,
}
impl CompactionUsage {
pub(crate) fn observe(
&mut self,
provider: &str,
observation: crate::compaction::CompactionObservation<'_>,
sequence: u64,
sink: &mut Option<&mut dyn AgentOutputSink>,
) -> anyhow::Result<()> {
let crate::compaction::CompactionObservation::ProviderEvent(event) = observation else {
if let Some(sink) = sink.as_deref_mut() {
sink.output_event(OutputEvent::UsageSnapshot {
usage: Default::default(),
request_sequence: sequence,
final_usage: false,
})?;
}
return Ok(());
};
if let Some(observation) = usage_observation(event) {
merge_usage(&mut self.latest, observation.usage, observation.presence);
} else if !matches!(event, ProviderEvent::Done) {
return Ok(());
}
let Some((usage, presence)) = &self.latest else {
return Ok(());
};
self.total = known_usage_total(provider, usage, *presence);
if let Some(sink) = sink.as_deref_mut() {
if let Some(total) = self.total {
sink.request_token_total(sequence, total)?;
}
if presence.input {
sink.output_event(OutputEvent::UsageSnapshot {
usage: normalized_usage(provider, usage, coverage_known(provider, *presence)),
request_sequence: sequence,
final_usage: matches!(event, ProviderEvent::Done) && presence.output,
})?;
}
}
Ok(())
}
}
#[allow(clippy::too_many_arguments)]
fn emit_usage_observation(
provider_id: &str,
request_sequence: u64,
context_budget_max_tokens: usize,
latest: &mut Option<(Usage, crate::providers::UsagePresence)>,
usage: Usage,
presence: crate::providers::UsagePresence,
provider_input_tokens: &mut Option<usize>,
output_sink: &mut Option<&mut dyn AgentOutputSink>,
) -> anyhow::Result<()> {
let merged = merge_usage(latest, usage, presence);
if let Some(total) = known_usage_total(provider_id, &merged.0, merged.1)
&& let Some(sink) = output_sink.as_deref_mut()
{
sink.request_token_total(request_sequence, total)?;
}
if !merged.1.input {
*latest = Some(merged);
return Ok(());
}
let input_tokens = provider_usage_input_tokens(provider_id, &merged.0);
*provider_input_tokens = Some(input_tokens);
if let Some(sink) = output_sink.as_deref_mut() {
sink.output_event(OutputEvent::ContextUsage {
current_tokens: input_tokens,
max_tokens: context_budget_max_tokens,
reasoning_tokens: merged
.0
.reasoning_tokens
.and_then(|v| usize::try_from(v).ok()),
source: ContextUsageSource::ProviderPartial,
request_sequence,
})?;
sink.output_event(OutputEvent::UsageSnapshot {
usage: normalized_usage(
provider_id,
&merged.0,
coverage_known(provider_id, merged.1),
),
request_sequence,
final_usage: false,
})?;
}
*latest = Some(merged);
Ok(())
}
pub(super) fn collect_provider_events<P: Provider + ?Sized>(
provider: &P,
request: ProviderRequest,
mut collector: ProviderEventCollector<'_, '_, '_>,
) -> Result<ProviderStreamResult, ProviderStreamFailure> {
let mut tool_calls = Vec::new();
let mut reasoning_summaries = Vec::new();
let mut reasoning_completion_sequence = 0u64;
let mut observed_function_response_item = false;
let mut observed_reasoning_delta = false;
let mut response_identities = Vec::new();
let mut reasoning_preview = String::new();
let mut provider_input_tokens = None;
let mut returned_service_tier = None;
let mut latest_usage: Option<(Usage, crate::providers::UsagePresence)> = None;
let mut projection_tracker = ContextProjectionTracker::new(
collector.provider_id,
collector.model,
collector.request_sequence,
collector.request_input_tokens,
collector.context_budget_max_tokens,
);
let stream_result =
provider.stream_cancellable(request, collector.cancellation, &mut |event| {
if collector.cancellation.is_canceled()
&& !matches!(event, ProviderEvent::ResponseIdentity(_))
{
return Err(AgentRunCanceled.into());
}
match event {
ProviderEvent::TextDelta(delta) => {
let pending_separator = collector.turn_state.has_pending_segment_separator();
let should_emit_separator =
pending_separator && !collector.output.text.is_empty();
if let Some(ttsr) = collector.ttsr.as_deref_mut() {
let candidate = if should_emit_separator {
format!("\n\n{delta}")
} else {
delta.clone()
};
if let Some(ttsr_match) = ttsr.check_text(&candidate) {
if let Some(handle) = collector.ttsr_cancel_handle {
handle.cancel();
}
if let Err(persistence_error) = collector.assistant_chunk_batch.flush()
{
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
return Err(TtsrInterrupted { ttsr_match }.into());
}
}
if should_emit_separator {
if let Some(sink) = collector.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::AssistantDelta {
text: "\n\n".to_string(),
})?;
}
collector.output.text.push_str("\n\n");
collector.turn_state.commit_pending_segment_separator();
}
let mut emit_delta = |text: &str,
record_turn_state: bool|
-> anyhow::Result<()> {
if let Some(sink) = collector.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::AssistantDelta {
text: text.to_string(),
})?;
}
collector.output.text.push_str(text);
if record_turn_state {
collector.turn_state.push_assistant_delta(text);
}
collector.assistant_chunk_batch.push(text);
if let Some(event) = projection_tracker.update_after_delta(text)
&& let Some(sink) = collector.output_sink.as_deref_mut()
&& let Err(error) = sink.output_event(event)
{
if let Err(persistence_error) = collector.assistant_chunk_batch.flush()
{
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
return Err(error);
}
if let Err(error) = collector.assistant_chunk_batch.flush_if_full() {
collector
.session_persistence
.warn_once(collector.output_sink, &error)?;
}
Ok(())
};
let delta_result = emit_delta(&delta, true);
if pending_separator && !should_emit_separator && delta_result.is_ok() {
collector.turn_state.commit_pending_segment_separator();
}
delta_result?;
}
event @ (ProviderEvent::Usage(_)
| ProviderEvent::UsagePartial(_)
| ProviderEvent::UsageObserved(_)) => {
let UsageObservation { usage, presence } =
usage_observation(&event).expect("usage event");
emit_usage_observation(
collector.provider_id,
collector.request_sequence,
collector.context_budget_max_tokens,
&mut latest_usage,
usage,
presence,
&mut provider_input_tokens,
collector.output_sink,
)?;
}
ProviderEvent::ReasoningSummaryDelta(text) => {
observed_reasoning_delta = true;
push_recovery_preview(&mut reasoning_preview, &text);
if let Err(error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &error)?;
}
if let Some(sink) = collector.output_sink.as_deref_mut()
&& let Err(error) = sink
.output_event(OutputEvent::ThinkingSummaryDelta { text: text.clone() })
{
return Err(error);
}
}
ProviderEvent::ReasoningSummaryComplete(text) => {
reasoning_completion_sequence = reasoning_completion_sequence.saturating_add(1);
let item_id = Some(format!(
"legacy-{}-{reasoning_completion_sequence}",
collector.request_sequence
));
let turn_id = Some(collector.request_sequence.to_string());
handle_reasoning_summary_completion(
&mut collector,
&mut reasoning_summaries,
&mut observed_reasoning_delta,
&mut reasoning_preview,
text,
Some((item_id, turn_id)),
false,
)?;
}
ProviderEvent::ReasoningSummaryCompleteIdentified(ReasoningSummary {
text,
item_id,
turn_id,
provider_summary,
}) => {
let item_id = item_id.or_else(|| {
reasoning_completion_sequence =
reasoning_completion_sequence.saturating_add(1);
Some(format!(
"legacy-{}-{reasoning_completion_sequence}",
collector.request_sequence
))
});
let turn_id = turn_id.or_else(|| Some(collector.request_sequence.to_string()));
handle_reasoning_summary_completion(
&mut collector,
&mut reasoning_summaries,
&mut observed_reasoning_delta,
&mut reasoning_preview,
text,
Some((item_id, turn_id)),
provider_summary,
)?;
}
ProviderEvent::ToolCall(call) => {
if let Some(event) = projection_tracker.force_event_if_changed()
&& let Some(sink) = collector.output_sink.as_deref_mut()
&& let Err(error) = sink.output_event(event)
{
if let Err(persistence_error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
return Err(error);
}
if let Some(ttsr) = collector.ttsr.as_deref_mut() {
let candidate =
json!({"name": call.name, "arguments": call.arguments}).to_string();
if let Some(ttsr_match) = ttsr.check_standalone(&candidate) {
if let Some(handle) = collector.ttsr_cancel_handle {
handle.cancel();
}
if let Err(persistence_error) = collector.assistant_chunk_batch.flush()
{
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
return Err(TtsrInterrupted { ttsr_match }.into());
}
}
tool_calls.push(call);
}
ProviderEvent::ResponseItem(item) => {
if item.get("type").and_then(serde_json::Value::as_str) == Some("function_call")
{
observed_function_response_item = true;
}
collector.turn_state.push_response_item(item.clone());
if let Err(error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &error)?;
}
collector.session_persistence.try_record(
SessionEventKind::ProviderResponseItem,
json!({"item": sanitize_provider_response_item(item.clone())}),
collector.output_sink,
)?;
}
ProviderEvent::ResponseIdentity(identity) => {
collector.session_persistence.try_record(
SessionEventKind::ProviderStreamTrace,
json!({"response_identity": identity}),
collector.output_sink,
)?;
response_identities.push(identity);
}
ProviderEvent::ServiceTier(tier) => {
returned_service_tier = Some(tier);
}
ProviderEvent::Done => {
if let Some((mut usage, presence)) = latest_usage.clone() {
if !presence.total {
usage.total = derived_usage_total(collector.provider_id, &usage);
}
if presence.input {
let input_tokens =
provider_usage_input_tokens(collector.provider_id, &usage);
provider_input_tokens = Some(input_tokens);
if let Some(sink) = collector.output_sink.as_deref_mut() {
sink.output_event(OutputEvent::ContextUsage {
current_tokens: input_tokens,
max_tokens: collector.context_budget_max_tokens,
reasoning_tokens: usage
.reasoning_tokens
.and_then(|v| usize::try_from(v).ok()),
source: ContextUsageSource::ProviderExact,
request_sequence: collector.request_sequence,
})?;
sink.output_event(OutputEvent::UsageSnapshot {
usage: normalized_usage(
collector.provider_id,
&usage,
coverage_known(collector.provider_id, presence),
),
request_sequence: collector.request_sequence,
final_usage: presence.output,
})?;
}
}
}
if let Some(event) = projection_tracker.force_event_if_changed()
&& let Some(sink) = collector.output_sink.as_deref_mut()
&& let Err(error) = sink.output_event(event)
{
if let Err(persistence_error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
return Err(error);
}
}
}
collector.cancellation.check()
});
if let Some((mut usage, presence)) = latest_usage {
if let Some(total) = known_usage_total(collector.provider_id, &usage, presence) {
usage.total = total;
accumulate_total_tokens(collector.output, &usage);
}
collector.output.usage = Some(usage);
}
if let Err(error) = stream_result {
if let Err(persistence_error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
let cancelled = is_run_canceled(&error);
let unsafe_tool_call_progress = unsafe_tool_call_progress_from_error(&error);
let provider_stream_trace = provider_stream_trace_from_error(&error).map(Box::new);
let ttsr_match = ttsr_match_from_error(&error);
return Err(ProviderStreamFailure {
error,
partial_assistant_text: !collector.output.text.trim().is_empty(),
observed_tool_call: !tool_calls.is_empty(),
observed_function_response_item,
observed_reasoning_delta,
unsafe_tool_call_progress,
reasoning_preview: (!reasoning_preview.trim().is_empty()).then_some(reasoning_preview),
partial_tool_call_summary: unsafe_tool_call_progress.then(|| {
"partial provider tool-call progress observed; raw arguments omitted; complete-tool-call recovery not attempted without a finished parser event".to_string()
}),
provider_stream_trace,
cancelled,
ttsr_match,
});
}
if let Err(error) = collector.cancellation.check() {
if let Err(persistence_error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &persistence_error)?;
}
let unsafe_tool_call_progress = false;
return Err(ProviderStreamFailure {
error,
partial_assistant_text: !collector.output.text.trim().is_empty(),
observed_tool_call: !tool_calls.is_empty(),
observed_function_response_item,
observed_reasoning_delta,
unsafe_tool_call_progress,
reasoning_preview: (!reasoning_preview.trim().is_empty()).then_some(reasoning_preview),
partial_tool_call_summary: None,
provider_stream_trace: None,
cancelled: true,
ttsr_match: None,
});
}
let fast_outcome = crate::fast::fast_outcome(
provider.requested_service_tier(),
returned_service_tier.as_deref(),
);
if let Some(requested_service_tier) = provider.requested_service_tier()
&& let Some(sink) = collector.output_sink.as_deref_mut()
{
sink.output_event(OutputEvent::FastObservation {
provider_id: collector.provider_id.to_string(),
model: collector.model.to_string(),
requested_service_tier: requested_service_tier.to_string(),
outcome: fast_outcome.clone(),
request_sequence: collector.request_sequence,
run_order: None,
})?;
}
collector.output.fast_outcome = fast_outcome;
emit_response_identity_diagnostic(
&response_identities,
collector.session_persistence,
collector.output_sink,
)?;
Ok(ProviderStreamResult {
tool_calls,
reasoning_summaries,
provider_input_tokens,
})
}
#[cfg(test)]
mod response_identity_diagnostic_tests {
use super::*;
use crate::providers::error::ResponseAttemptIdentity;
fn identity(
attempt: usize,
requested_model: &str,
provider_response_model: Option<&str>,
outcome: &str,
) -> ResponseAttemptIdentity {
ResponseAttemptIdentity {
schema_version: 1,
provider: "openai-compatible".to_string(),
attempt,
requested_model: requested_model.to_string(),
provider_response_model: provider_response_model.map(str::to_string),
request_id: None,
outcome: outcome.to_string(),
}
}
#[test]
fn normal_first_attempt_is_silent() {
assert!(
response_identity_diagnostic_event(&[identity(
1,
"requested",
Some("requested"),
"success"
)])
.is_none()
);
assert!(
response_identity_diagnostic_event(&[identity(1, "requested", None, "success")])
.is_none()
);
}
#[test]
fn retry_recovery_is_informational() {
let event = response_identity_diagnostic_event(&[
identity(1, "requested", None, "error"),
identity(2, "requested", Some("requested"), "success"),
])
.expect("retry recovery should be visible");
assert_eq!(
event,
OutputEvent::Diagnostic {
level: "info".to_string(),
message: "provider request recovered after 2 attempts".to_string(),
}
);
}
#[test]
fn model_metadata_difference_is_warning_without_substitution_claim() {
let event = response_identity_diagnostic_event(&[identity(
1,
"requested",
Some("reported"),
"success",
)])
.expect("different model metadata should be visible");
assert_eq!(
event,
OutputEvent::Diagnostic {
level: "warning".to_string(),
message: "provider-reported model metadata differed from request metadata; requested model 'requested', reported model 'reported'; observability signal only, not proof of model substitution".to_string(),
}
);
}
#[test]
fn model_metadata_difference_sanitizes_provider_control_sequences() {
let event = response_identity_diagnostic_event(&[identity(
1,
"requested\nmodel",
Some("reported\x1b]52;c;clipboard\x07\nmodel"),
"success",
)])
.expect("different model metadata should be visible");
let OutputEvent::Diagnostic { message, .. } = event else {
unreachable!("response identity diagnostics are diagnostic events");
};
assert!(message.contains("requested model 'requested model'"));
assert!(message.contains("reported model 'reported model'"));
assert!(!message.contains('\x1b'));
assert!(!message.contains('\n'));
assert!(!message.contains("clipboard"));
}
#[test]
fn unsuccessful_terminal_identity_does_not_duplicate_provider_error() {
assert!(
response_identity_diagnostic_event(&[identity(
1,
"requested",
Some("reported"),
"error"
)])
.is_none()
);
}
}
fn emit_response_identity_diagnostic(
identities: &[crate::providers::error::ResponseAttemptIdentity],
session_persistence: &mut SessionPersistence<'_>,
output_sink: &mut Option<&mut dyn AgentOutputSink>,
) -> anyhow::Result<()> {
let Some(event) = response_identity_diagnostic_event(identities) else {
return Ok(());
};
let OutputEvent::Diagnostic { level, message } = &event else {
unreachable!("response identity diagnostics are diagnostic events");
};
session_persistence.try_record(
SessionEventKind::Diagnostic,
json!({
"category": "provider_response_identity",
"level": level,
"message": message,
}),
output_sink,
)?;
if let Some(sink) = output_sink.as_deref_mut() {
sink.output_event(event)?;
}
Ok(())
}
fn response_identity_diagnostic_event(
identities: &[crate::providers::error::ResponseAttemptIdentity],
) -> Option<OutputEvent> {
let last = identities.last()?;
if last.outcome != "success" {
return None;
}
let mismatch = identities.iter().find_map(|identity| {
identity
.provider_response_model
.as_deref()
.filter(|model| !model.trim().is_empty() && *model != identity.requested_model)
.map(|model| {
(
sanitize_response_identity_display_value(&identity.requested_model),
sanitize_response_identity_display_value(model),
)
})
});
let recovered = last.attempt > 1;
if !recovered && mismatch.is_none() {
return None;
}
let level = if mismatch.is_some() {
"warning"
} else {
"info"
};
let mut message = if recovered {
format!("provider request recovered after {} attempts", last.attempt)
} else {
String::new()
};
if let Some((requested, reported)) = mismatch {
if !message.is_empty() {
message.push_str("; ");
}
message.push_str(&format!(
"provider-reported model metadata differed from request metadata; requested model '{requested}', reported model '{reported}'; observability signal only, not proof of model substitution"
));
}
Some(OutputEvent::Diagnostic {
level: level.to_string(),
message,
})
}
fn sanitize_response_identity_display_value(value: &str) -> String {
sanitize_display_text(value)
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
}
struct ContextProjectionTracker<'a> {
provider_id: &'a str,
model: &'a str,
request_sequence: u64,
request_input_tokens: usize,
max_tokens: usize,
has_output: bool,
pending_delta_text: String,
projected_output_tokens: usize,
projection_source: ContextUsageSource,
last_emitted_tokens: usize,
}
impl<'a> ContextProjectionTracker<'a> {
const FLUSH_BYTES: usize = 256;
const EMIT_TOKEN_DELTA: usize = 16;
fn new(
provider_id: &'a str,
model: &'a str,
request_sequence: u64,
request_input_tokens: usize,
max_tokens: usize,
) -> Self {
Self {
provider_id,
model,
request_sequence,
request_input_tokens,
max_tokens,
has_output: false,
pending_delta_text: String::new(),
projected_output_tokens: 0,
projection_source: ContextUsageSource::FallbackProjection,
last_emitted_tokens: request_input_tokens,
}
}
fn update_after_delta(&mut self, delta: &str) -> Option<OutputEvent> {
self.has_output = true;
self.pending_delta_text.push_str(delta);
if self.pending_delta_text.len() < Self::FLUSH_BYTES {
return None;
}
let pending_delta = std::mem::take(&mut self.pending_delta_text);
let projection = project_text_tokens(self.provider_id, self.model, &pending_delta);
self.projected_output_tokens = self
.projected_output_tokens
.saturating_add(projection.tokens);
self.projection_source = projection.source;
if self.projected_output_tokens.abs_diff(
self.last_emitted_tokens
.saturating_sub(self.request_input_tokens),
) < Self::EMIT_TOKEN_DELTA
{
return None;
}
let event = self.project_event_from_incremental();
let current_tokens = event_current_tokens(&event);
(current_tokens.abs_diff(self.last_emitted_tokens) >= Self::EMIT_TOKEN_DELTA).then(|| {
self.last_emitted_tokens = current_tokens;
event
})
}
fn force_event_if_changed(&mut self) -> Option<OutputEvent> {
if !self.has_output {
return None;
}
if !self.pending_delta_text.is_empty() {
let pending_delta = std::mem::take(&mut self.pending_delta_text);
let projection = project_text_tokens(self.provider_id, self.model, &pending_delta);
self.projected_output_tokens = self
.projected_output_tokens
.saturating_add(projection.tokens);
self.projection_source = projection.source;
}
let event = self.project_event_from_incremental();
let current_tokens = event_current_tokens(&event);
(current_tokens != self.last_emitted_tokens).then(|| {
self.last_emitted_tokens = current_tokens;
event
})
}
fn project_event_from_incremental(&self) -> OutputEvent {
OutputEvent::ContextUsage {
current_tokens: self
.request_input_tokens
.saturating_add(self.projected_output_tokens),
max_tokens: self.max_tokens,
reasoning_tokens: None,
source: self.projection_source,
request_sequence: self.request_sequence,
}
}
}
fn event_current_tokens(event: &OutputEvent) -> usize {
match event {
OutputEvent::ContextUsage { current_tokens, .. } => *current_tokens,
_ => unreachable!("context projection tracker only emits context usage"),
}
}
fn provider_usage_input_tokens(provider_id: &str, usage: &Usage) -> usize {
usize::try_from(normalized_input_tokens(provider_id, usage)).unwrap_or(usize::MAX)
}
fn derived_usage_total(provider_id: &str, usage: &Usage) -> u64 {
normalized_input_tokens(provider_id, usage).saturating_add(usage.output)
}
fn accumulate_total_tokens(output: &mut AgentRunOutput, usage: &Usage) {
output.total_tokens = Some(
output
.total_tokens
.unwrap_or_default()
.saturating_add(usage.total),
);
}
fn push_recovery_preview(preview: &mut String, delta: &str) {
const RECOVERY_REASONING_PREVIEW_LIMIT: usize = 4_000;
if preview.chars().count() >= RECOVERY_REASONING_PREVIEW_LIMIT {
return;
}
let remaining = RECOVERY_REASONING_PREVIEW_LIMIT.saturating_sub(preview.chars().count());
preview.extend(delta.chars().take(remaining));
}
fn sanitize_recovery_preview(text: &str) -> String {
const RECOVERY_REASONING_PREVIEW_LIMIT: usize = 4_000;
let redacted = redact_sensitive_text(text.trim());
redacted
.chars()
.take(RECOVERY_REASONING_PREVIEW_LIMIT)
.collect()
}
fn handle_reasoning_summary_completion(
collector: &mut ProviderEventCollector<'_, '_, '_>,
reasoning_summaries: &mut Vec<String>,
observed_reasoning_delta: &mut bool,
reasoning_preview: &mut String,
text: String,
identity: Option<(Option<String>, Option<String>)>,
provider_summary: bool,
) -> anyhow::Result<()> {
if let Err(error) = collector.assistant_chunk_batch.flush() {
collector
.session_persistence
.warn_once(collector.output_sink, &error)?;
}
let (item_id, turn_id) = identity.unwrap_or((None, None));
let event = if item_id.is_some() || turn_id.is_some() {
OutputEvent::ThinkingSummaryCompleteIdentified {
text: text.clone(),
item_id: item_id.clone(),
turn_id: turn_id.clone(),
}
} else {
OutputEvent::ThinkingSummaryComplete { text: text.clone() }
};
if let Some(sink) = collector.output_sink.as_deref_mut() {
if provider_summary {
sink.provider_reasoning_summary(&text)?;
}
sink.output_event(event)?;
}
let mut payload = json!({"text": text});
if let Some(item_id) = item_id {
payload["item_id"] = json!(item_id);
}
if let Some(turn_id) = turn_id {
payload["turn_id"] = json!(turn_id);
}
collector.session_persistence.try_record(
SessionEventKind::ReasoningSummary,
payload,
collector.output_sink,
)?;
reasoning_summaries.push(text);
*observed_reasoning_delta = false;
reasoning_preview.clear();
Ok(())
}
fn unsafe_tool_call_progress_from_error(error: &anyhow::Error) -> bool {
error.chain().any(|cause| {
cause
.to_string()
.contains("unsafe tool-call progress observed")
})
}
fn ttsr_match_from_error(error: &anyhow::Error) -> Option<Box<TtsrMatch>> {
error
.chain()
.find_map(|cause| cause.downcast_ref::<TtsrInterrupted>())
.map(|interrupted| Box::new(interrupted.ttsr_match.clone()))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::providers::error::ProviderError;
#[test]
fn compaction_usage_merges_presence_and_preserves_reported_totals() {
let mut accounting = CompactionUsage::default();
let mut sink = None;
accounting
.observe(
ANTHROPIC_PROVIDER,
crate::compaction::CompactionObservation::ProviderEvent(&ProviderEvent::Done),
0,
&mut sink,
)
.unwrap();
assert_eq!(accounting.total, None);
for (usage, presence, expected) in [
(
Usage {
input: 10,
cache_read: 20,
cache_write: 30,
..Usage::default()
},
crate::providers::UsagePresence {
input: true,
cache_read: true,
cache_write: true,
..Default::default()
},
None,
),
(
Usage {
output: 4,
..Usage::default()
},
crate::providers::UsagePresence {
output: true,
..Default::default()
},
Some(64),
),
(
Usage::default(),
crate::providers::UsagePresence {
total: true,
..Default::default()
},
Some(0),
),
] {
accounting
.observe(
ANTHROPIC_PROVIDER,
crate::compaction::CompactionObservation::ProviderEvent(
&ProviderEvent::UsageObserved(UsageObservation { usage, presence }),
),
0,
&mut sink,
)
.unwrap();
assert_eq!(accounting.total, expected);
}
let mut totals = RequestTokenTotals::default();
assert_eq!(totals.total(), None);
totals.observe(0, 64);
totals.observe(0, 0);
assert_eq!(totals.total(), Some(0));
totals.observe(1, 29);
totals.observe(1, 31);
totals.observe(0, 1000);
assert_eq!(totals.total(), Some(31));
}
#[test]
fn anthropic_family_usage_accounting_includes_cached_tokens() {
let usage = Usage {
input: 10,
cache_read: 20,
cache_write: 30,
..Usage::default()
};
assert_eq!(provider_usage_input_tokens(ANTHROPIC_PROVIDER, &usage), 60);
assert_eq!(provider_usage_input_tokens("custom", &usage), 10);
}
#[test]
fn normalized_usage_reports_provider_specific_cache_percentages() {
let openai = normalized_usage(
crate::providers::OPENAI_CODEX_PROVIDER,
&Usage {
input: 100,
cache_read: 80,
..Usage::default()
},
true,
);
assert_eq!(openai.effective_input, 100);
assert_eq!(openai.cache_read * 100 / openai.effective_input, 80);
let anthropic = normalized_usage(
ANTHROPIC_PROVIDER,
&Usage {
input: 20,
cache_read: 70,
cache_write: 10,
..Usage::default()
},
true,
);
assert_eq!(anthropic.effective_input, 100);
assert_eq!(anthropic.cache_read * 100 / anthropic.effective_input, 70);
}
#[test]
fn cache_coverage_requires_explicit_provider_components() {
let compatible = crate::providers::UsagePresence {
input: true,
cache_read: true,
..Default::default()
};
assert!(coverage_known("custom", compatible));
assert!(!coverage_known(
"custom",
crate::providers::UsagePresence {
input: false,
..compatible
}
));
assert!(!coverage_known(
"custom",
crate::providers::UsagePresence {
cache_read: false,
..compatible
}
));
assert!(!coverage_known(ANTHROPIC_PROVIDER, compatible));
assert!(coverage_known(
ANTHROPIC_PROVIDER,
crate::providers::UsagePresence {
cache_write: true,
..compatible
}
));
}
#[test]
fn auto_continue_allows_hidden_partial_tool_call_progress() {
let failure = ProviderStreamFailure {
error: ProviderError::stream_failed_incomplete(
"provider stream ended prematurely after partial response; response is incomplete: request or response body error; unsafe tool-call progress observed",
)
.into(),
partial_assistant_text: false,
observed_tool_call: false,
observed_function_response_item: false,
observed_reasoning_delta: false,
unsafe_tool_call_progress: true,
reasoning_preview: None,
partial_tool_call_summary: Some("partial provider tool-call progress observed; raw arguments omitted".to_string()),
provider_stream_trace: None,
cancelled: false,
ttsr_match: None,
};
assert!(failure.eligible_for_auto_continue());
}
}