use crate::{
agent::{
cancellation::{AgentCancellation, is_run_canceled},
ttsr::TtsrInterrupted,
},
providers::{
HttpRequest, HttpTransport, ProviderEvent,
error::{ProviderError, provider_stream_trace_from_error, retryable_provider_error},
transport::provider_stream_deadline_after,
},
};
use std::{
sync::atomic::{AtomicU64, Ordering},
time::{Duration, Instant},
};
const PROVIDER_STREAM_MAX_ATTEMPTS: usize = 3;
const RATE_LIMIT_RETRY_BACKOFFS: [Duration; 3] = [
Duration::from_secs(2),
Duration::from_secs(10),
Duration::from_secs(20),
];
const RATE_LIMIT_MAX_ATTEMPTS: usize = RATE_LIMIT_RETRY_BACKOFFS.len() + 1;
#[cfg(not(test))]
pub(super) const PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT: Duration = Duration::from_secs(60);
#[cfg(test)]
pub(super) const PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT: Duration = Duration::from_millis(10);
#[derive(Debug)]
pub(super) struct StreamAttemptError {
pub(super) error: anyhow::Error,
pub(super) made_semantic_progress: bool,
pub(super) unsafe_recovery_progress: bool,
pub(super) attempts_used: usize,
}
pub(super) trait ProviderStreamParser {
fn push_chunk_outcome(
&mut self,
chunk: &str,
) -> anyhow::Result<crate::providers::stream::StreamParseOutcome>;
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>>;
fn response_model(&self) -> Option<String> {
None
}
}
impl ProviderStreamParser for crate::providers::stream::StreamParser {
fn push_chunk_outcome(
&mut self,
chunk: &str,
) -> anyhow::Result<crate::providers::stream::StreamParseOutcome> {
self.push_chunk_outcome(chunk)
}
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
self.finish()
}
fn response_model(&self) -> Option<String> {
crate::providers::stream::StreamParser::response_model(self)
}
}
#[cfg(test)]
pub(super) fn stream_with_transport<T: HttpTransport>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
stream_with_transport_parser(
transport,
http_request,
cancellation,
semantic_progress_timeout,
crate::providers::stream::StreamParser::default(),
on_event,
)
}
pub(super) fn stream_with_transport_parser<T: HttpTransport, P: ProviderStreamParser + Default>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
parser: P,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
stream_with_transport_attempt_result_with_parser::<T, P>(
transport,
http_request,
cancellation,
semantic_progress_timeout,
parser,
on_event,
)
.map_err(|attempt_error| attempt_error.error)
}
pub(super) fn stream_with_transport_attempt_result<T: HttpTransport>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> Result<(), StreamAttemptError> {
stream_with_transport_attempt_result_with_attempt_offset(
transport,
http_request,
cancellation,
semantic_progress_timeout,
0,
on_event,
)
}
pub(super) fn stream_with_transport_attempt_result_with_attempt_offset<T: HttpTransport>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
attempt_offset: usize,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> Result<(), StreamAttemptError> {
stream_with_transport_attempt_result_with_parser_and_sleep::<
T,
crate::providers::stream::StreamParser,
_,
>(
transport,
http_request,
cancellation,
semantic_progress_timeout,
crate::providers::stream::StreamParser::default(),
attempt_offset,
on_event,
&mut sleep_cancellable,
)
}
fn stream_with_transport_attempt_result_with_parser<
T: HttpTransport,
P: ProviderStreamParser + Default,
>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
parser: P,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> Result<(), StreamAttemptError> {
let mut sleep = sleep_cancellable;
stream_with_transport_attempt_result_with_parser_and_sleep::<T, P, _>(
transport,
http_request,
cancellation,
semantic_progress_timeout,
parser,
0,
on_event,
&mut sleep,
)
}
#[allow(clippy::too_many_arguments)]
fn stream_with_transport_attempt_result_with_parser_and_sleep<
T: HttpTransport,
P: ProviderStreamParser + Default,
S: FnMut(Duration, &AgentCancellation) -> anyhow::Result<()>,
>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
parser: P,
attempt_offset: usize,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
sleep: &mut S,
) -> Result<(), StreamAttemptError> {
let mut last_retryable_error = None;
let mut first_attempt_parser = Some(parser);
for local_attempt in 1..=RATE_LIMIT_MAX_ATTEMPTS {
let attempt = attempt_offset + local_attempt;
if let Err(error) = cancellation.check() {
return Err(StreamAttemptError {
error,
made_semantic_progress: false,
unsafe_recovery_progress: false,
attempts_used: local_attempt,
});
}
let mut parser = first_attempt_parser.take().unwrap_or_default();
match stream_attempt(
transport,
http_request.clone(),
cancellation,
semantic_progress_timeout,
&mut parser,
on_event,
attempt,
) {
Ok(()) => return Ok(()),
Err(attempt_error)
if !attempt_error.made_semantic_progress
&& retryable_provider_error(&attempt_error.error) =>
{
let Some(backoff) = provider_retry_backoff_for(&attempt_error.error, local_attempt)
else {
let error = attempt_error.error;
return Err(StreamAttemptError {
error: anyhow::anyhow!(
"provider request failed after {attempt} attempts: {error}"
),
made_semantic_progress: false,
unsafe_recovery_progress: false,
attempts_used: local_attempt,
});
};
last_retryable_error = Some(attempt_error.error);
if let Err(error) = sleep(backoff, cancellation) {
return Err(StreamAttemptError {
error,
made_semantic_progress: false,
unsafe_recovery_progress: false,
attempts_used: local_attempt,
});
}
}
Err(mut attempt_error) => {
attempt_error.attempts_used = local_attempt;
return Err(attempt_error);
}
}
}
let error = last_retryable_error
.map(|error| error.to_string())
.unwrap_or_else(|| "unknown provider retry failure".to_string());
Err(StreamAttemptError {
error: anyhow::anyhow!(
"provider request failed after {attempt_offset} + {RATE_LIMIT_MAX_ATTEMPTS} attempts: {error}"
),
made_semantic_progress: false,
unsafe_recovery_progress: false,
attempts_used: RATE_LIMIT_MAX_ATTEMPTS,
})
}
fn stream_attempt<T: HttpTransport, P: ProviderStreamParser>(
transport: &T,
http_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
parser: &mut P,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
attempt: usize,
) -> Result<(), StreamAttemptError> {
let mut made_semantic_progress = false;
let mut last_semantic_progress = Instant::now();
let mut unsafe_recovery_progress = false;
let semantic_deadline =
AtomicU64::new(provider_stream_deadline_after(semantic_progress_timeout));
let is_responses = http_request.url.contains("/responses");
let requested_model = http_request
.body
.get("model")
.and_then(serde_json::Value::as_str)
.unwrap_or("unknown")
.to_string();
let provider = if http_request.url.contains("chatgpt.com") {
"openai-codex"
} else {
"openai-compatible"
};
let mut request_id = None;
let stream_result = transport.stream_json_cancellable_with_response_metadata(
http_request,
cancellation,
&semantic_deadline,
&mut |id| request_id = id,
&mut |chunk| {
cancellation.check()?;
let outcome = parser.push_chunk_outcome(chunk)?;
unsafe_recovery_progress |= outcome.unsafe_recovery_progress;
if outcome.semantic_progress {
made_semantic_progress = true;
last_semantic_progress = Instant::now();
semantic_deadline.store(
provider_stream_deadline_after(semantic_progress_timeout),
Ordering::SeqCst,
);
} else if last_semantic_progress.elapsed() >= semantic_progress_timeout {
return Err(ProviderError::stream_terminal(
"provider stream no semantic progress before timeout",
)
.into());
}
for event in outcome.events {
on_event(event)?;
cancellation.check()?;
}
cancellation.check()?;
Ok(())
},
);
if let Err(error) = stream_result {
let error =
incomplete_stream_error(error, made_semantic_progress, unsafe_recovery_progress);
if is_responses {
emit_response_identity(
on_event,
response_identity(
provider,
attempt,
&requested_model,
parser.response_model(),
request_id,
"error",
),
made_semantic_progress,
unsafe_recovery_progress,
)?;
}
return Err(StreamAttemptError {
error,
made_semantic_progress,
unsafe_recovery_progress,
attempts_used: 1,
});
}
let finish_events = match parser.finish() {
Ok(events) => events,
Err(error) => {
let error =
incomplete_stream_error(error, made_semantic_progress, unsafe_recovery_progress);
if is_responses {
emit_response_identity(
on_event,
response_identity(
provider,
attempt,
&requested_model,
parser.response_model(),
request_id,
"error",
),
made_semantic_progress,
unsafe_recovery_progress,
)?;
}
return Err(StreamAttemptError {
error,
made_semantic_progress,
unsafe_recovery_progress,
attempts_used: 1,
});
}
};
made_semantic_progress |= !finish_events.is_empty();
for event in finish_events {
on_event(event).map_err(|error| StreamAttemptError {
error,
made_semantic_progress,
unsafe_recovery_progress,
attempts_used: 1,
})?;
}
if is_responses {
emit_response_identity(
on_event,
response_identity(
provider,
attempt,
&requested_model,
parser.response_model(),
request_id,
"success",
),
made_semantic_progress,
unsafe_recovery_progress,
)?;
}
Ok(())
}
fn response_identity(
provider: &str,
attempt: usize,
requested_model: &str,
provider_response_model: Option<String>,
request_id: Option<String>,
outcome: &str,
) -> crate::providers::error::ResponseAttemptIdentity {
crate::providers::error::ResponseAttemptIdentity {
schema_version: 1,
provider: provider.to_string(),
attempt,
requested_model: crate::providers::error::bounded_response_identity_string(requested_model),
provider_response_model: provider_response_model
.map(|value| crate::providers::error::bounded_response_identity_string(&value)),
request_id: request_id
.map(|value| crate::providers::error::bounded_response_identity_string(&value)),
outcome: crate::providers::error::bounded_response_identity_string(outcome),
}
}
fn emit_response_identity(
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
identity: crate::providers::error::ResponseAttemptIdentity,
made_semantic_progress: bool,
unsafe_recovery_progress: bool,
) -> Result<(), StreamAttemptError> {
on_event(ProviderEvent::ResponseIdentity(identity)).map_err(|error| StreamAttemptError {
error,
made_semantic_progress,
unsafe_recovery_progress,
attempts_used: 1,
})
}
fn incomplete_stream_error(
error: anyhow::Error,
made_semantic_progress: bool,
unsafe_recovery_progress: bool,
) -> anyhow::Error {
if !made_semantic_progress
|| is_run_canceled(&error)
|| contains_actionable_provider_error(&error)
{
return error;
}
if error.chain().any(|cause| cause.is::<TtsrInterrupted>()) {
return error;
}
let stream_trace = provider_stream_trace_from_error(&error);
let unsafe_suffix = if unsafe_recovery_progress {
"; unsafe tool-call progress observed"
} else {
""
};
let mut wrapped = ProviderError::stream_failed_incomplete(format!(
"provider stream ended prematurely after partial response; response is incomplete: {error}{unsafe_suffix}"
));
if let Some(trace) = stream_trace {
wrapped = wrapped.with_stream_trace(trace);
}
wrapped.into()
}
fn contains_actionable_provider_error(error: &anyhow::Error) -> bool {
error
.downcast_ref::<ProviderError>()
.and_then(ProviderError::http_status_code)
.is_some()
}
pub(super) fn sleep_cancellable(
duration: Duration,
cancellation: &AgentCancellation,
) -> anyhow::Result<()> {
const CANCEL_POLL_INTERVAL: Duration = Duration::from_millis(25);
let start = Instant::now();
loop {
cancellation.check()?;
let elapsed = start.elapsed();
if elapsed >= duration {
return Ok(());
}
std::thread::sleep(duration.saturating_sub(elapsed).min(CANCEL_POLL_INTERVAL));
}
}
fn provider_retry_backoff_for(error: &anyhow::Error, attempt: usize) -> Option<Duration> {
if !retryable_provider_error(error) {
return None;
}
if is_rate_limited(error) {
return RATE_LIMIT_RETRY_BACKOFFS
.get(attempt.checked_sub(1)?)
.copied();
}
(attempt < PROVIDER_STREAM_MAX_ATTEMPTS).then(|| provider_retry_backoff(attempt))
}
fn is_rate_limited(error: &anyhow::Error) -> bool {
error
.downcast_ref::<ProviderError>()
.and_then(ProviderError::http_status_code)
== Some(429)
}
static RETRY_JITTER_COUNTER: AtomicU64 = AtomicU64::new(0);
fn provider_retry_backoff(attempt: usize) -> Duration {
let base_ms: u64 = match attempt {
1 => 250,
_ => 750,
};
let bound_ms = base_ms / 5;
let span = bound_ms.saturating_mul(2).saturating_add(1);
let offset =
(RETRY_JITTER_COUNTER.fetch_add(1, Ordering::Relaxed) % span) as i64 - bound_ms as i64;
Duration::from_millis((base_ms as i64 + offset).max(1) as u64)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::providers::{
error::{
ProviderStreamTrace, ProviderStreamTracePendingTool, provider_stream_trace_from_error,
},
stream::StreamParseOutcome,
};
use serde_json::Value;
use std::{
collections::{BTreeMap, VecDeque},
sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicU64, Ordering},
},
};
#[derive(Default)]
struct ProgressParser;
impl ProviderStreamParser for ProgressParser {
fn push_chunk_outcome(&mut self, _chunk: &str) -> anyhow::Result<StreamParseOutcome> {
Ok(StreamParseOutcome {
events: vec![ProviderEvent::TextDelta("partial".to_string())],
semantic_progress: true,
unsafe_recovery_progress: false,
})
}
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
Ok(Vec::new())
}
}
struct ErrorAfterChunkTransport;
impl HttpTransport for ErrorAfterChunkTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
on_chunk("data")?;
Err(ProviderError::transport("connection reset").into())
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
struct HttpStatusAfterChunkTransport;
impl HttpTransport for HttpStatusAfterChunkTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
on_chunk("data")?;
Err(ProviderError::http_status(
401,
format!(
"provider request failed for {} with status 401 Unauthorized: {}",
crate::providers::CODEX_RESPONSES_URL,
crate::providers::error::CODEX_SESSION_EXPIRED_MESSAGE
),
)
.into())
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
#[derive(Default)]
struct UnsafeProgressParser;
impl ProviderStreamParser for UnsafeProgressParser {
fn push_chunk_outcome(&mut self, _chunk: &str) -> anyhow::Result<StreamParseOutcome> {
Ok(StreamParseOutcome {
events: Vec::new(),
semantic_progress: true,
unsafe_recovery_progress: true,
})
}
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
Ok(Vec::new())
}
}
struct ErrorAfterUnsafeProgressTransport;
impl HttpTransport for ErrorAfterUnsafeProgressTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
on_chunk("data")?;
Err(ProviderError::transport("connection reset").into())
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
struct ImmediateErrorTransport;
impl HttpTransport for ImmediateErrorTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
_on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
anyhow::bail!("connection reset")
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
enum ScriptStep {
RateLimit,
Done,
}
struct ScriptedTransport {
steps: Mutex<VecDeque<ScriptStep>>,
attempts: Arc<Mutex<usize>>,
}
impl ScriptedTransport {
fn new(steps: Vec<ScriptStep>) -> Self {
Self {
steps: Mutex::new(steps.into()),
attempts: Arc::new(Mutex::new(0)),
}
}
fn attempts_handle(&self) -> Arc<Mutex<usize>> {
Arc::clone(&self.attempts)
}
}
impl HttpTransport for ScriptedTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
*self.attempts.lock().unwrap() += 1;
match self.steps.lock().unwrap().pop_front().unwrap() {
ScriptStep::RateLimit => Err(rate_limit_error()),
ScriptStep::Done => on_chunk("data: [DONE]\n\n"),
}
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
struct SingleChunkTransport;
impl HttpTransport for SingleChunkTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
on_chunk("data")
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
struct SemanticProgressResetTransport;
impl HttpTransport for SemanticProgressResetTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable_with_semantic_deadline(
request,
cancellation,
&AtomicU64::new(0),
on_chunk,
)
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
_request: HttpRequest,
cancellation: &AgentCancellation,
semantic_deadline: &AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
cancellation.check()?;
let original_deadline = semantic_deadline.load(Ordering::SeqCst);
assert!(original_deadline > 0);
std::thread::sleep(Duration::from_millis(2));
on_chunk("first")?;
let refreshed_deadline = semantic_deadline.load(Ordering::SeqCst);
assert!(refreshed_deadline > original_deadline);
while super::super::transport::provider_stream_deadline_after(Duration::ZERO)
<= original_deadline
{
std::thread::yield_now();
}
assert!(
super::super::transport::provider_stream_deadline_after(Duration::ZERO)
< refreshed_deadline,
"deadline reset did not allow progress after original deadline"
);
cancellation.check()?;
on_chunk("second")?;
let second_deadline = semantic_deadline.load(Ordering::SeqCst);
assert!(second_deadline > refreshed_deadline);
while super::super::transport::provider_stream_deadline_after(Duration::ZERO)
<= second_deadline
{
std::thread::yield_now();
}
std::thread::sleep(Duration::from_millis(5));
cancellation.check()?;
on_chunk("third")
}
}
#[derive(Default)]
struct ExpiringProgressParser {
chunks: usize,
}
impl ProviderStreamParser for ExpiringProgressParser {
fn push_chunk_outcome(&mut self, _chunk: &str) -> anyhow::Result<StreamParseOutcome> {
self.chunks += 1;
Ok(StreamParseOutcome {
events: vec![ProviderEvent::TextDelta("partial".to_string())],
semantic_progress: self.chunks < 3,
unsafe_recovery_progress: false,
})
}
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
Ok(Vec::new())
}
}
#[derive(Clone)]
struct RecordingParser {
used: Arc<AtomicBool>,
}
impl Default for RecordingParser {
fn default() -> Self {
Self {
used: Arc::new(AtomicBool::new(false)),
}
}
}
impl ProviderStreamParser for RecordingParser {
fn push_chunk_outcome(&mut self, _chunk: &str) -> anyhow::Result<StreamParseOutcome> {
self.used.store(true, Ordering::SeqCst);
Ok(StreamParseOutcome {
events: vec![ProviderEvent::Done],
semantic_progress: true,
unsafe_recovery_progress: false,
})
}
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
Ok(Vec::new())
}
}
fn request() -> HttpRequest {
HttpRequest {
method: "POST".to_string(),
url: "https://provider.test/v1/chat/completions".to_string(),
headers: BTreeMap::new(),
body: Value::Null,
}
}
fn rate_limit_error() -> anyhow::Error {
ProviderError::http_status(429, "rate limit").into()
}
fn status_error(status: u16) -> anyhow::Error {
ProviderError::http_status(status, format!("http {status}")).into()
}
#[derive(Default)]
struct TraceErrorParser;
impl ProviderStreamParser for TraceErrorParser {
fn push_chunk_outcome(&mut self, _chunk: &str) -> anyhow::Result<StreamParseOutcome> {
Ok(StreamParseOutcome {
events: vec![ProviderEvent::TextDelta("partial".to_string())],
semantic_progress: true,
unsafe_recovery_progress: true,
})
}
fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
Err(ProviderError::stream_terminal("traced terminal")
.with_stream_trace(test_trace())
.into())
}
}
fn test_trace() -> ProviderStreamTrace {
ProviderStreamTrace {
schema_version: 1,
provider: "anthropic".to_string(),
failure_context: "message_stop".to_string(),
message_delta_stop_reason: Some("tool_use".to_string()),
recent_events: Vec::new(),
pending_tool_count: 1,
pending_tools_truncated: false,
pending_tools: vec![ProviderStreamTracePendingTool {
index: 1,
id: "toolu_1".to_string(),
name: "read".to_string(),
argument_bytes: 13,
argument_sha256: "a".repeat(64),
}],
}
}
#[test]
fn incomplete_stream_error_preserves_provider_stream_trace() {
let attempt_error = stream_with_transport_attempt_result_with_parser::<
SingleChunkTransport,
TraceErrorParser,
>(
&SingleChunkTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
TraceErrorParser,
&mut |_| Ok(()),
)
.unwrap_err();
assert!(attempt_error.made_semantic_progress);
assert!(attempt_error.unsafe_recovery_progress);
assert_eq!(
provider_stream_trace_from_error(&attempt_error.error),
Some(test_trace())
);
assert!(
attempt_error
.error
.to_string()
.contains("unsafe tool-call progress")
);
}
#[test]
fn rate_limit_retry_backoff_uses_progressive_schedule() {
assert_eq!(
provider_retry_backoff_for(&rate_limit_error(), 1),
Some(Duration::from_secs(2))
);
assert_eq!(
provider_retry_backoff_for(&rate_limit_error(), 2),
Some(Duration::from_secs(10))
);
assert_eq!(
provider_retry_backoff_for(&rate_limit_error(), 3),
Some(Duration::from_secs(20))
);
assert_eq!(provider_retry_backoff_for(&rate_limit_error(), 4), None);
}
#[test]
fn non_rate_limit_retry_backoff_preserves_existing_schedule() {
for attempt in [1, 2] {
let backoff = provider_retry_backoff_for(&status_error(503), attempt).unwrap();
let range = if attempt == 1 {
Duration::from_millis(200)..=Duration::from_millis(300)
} else {
Duration::from_millis(600)..=Duration::from_millis(900)
};
assert!(range.contains(&backoff), "{backoff:?}");
}
assert_eq!(provider_retry_backoff_for(&status_error(503), 3), None);
}
#[test]
fn non_retryable_status_has_no_retry_backoff() {
assert_eq!(provider_retry_backoff_for(&status_error(401), 1), None);
}
#[test]
fn rate_limit_retries_three_times_with_progressive_backoff_before_success() {
let transport = ScriptedTransport::new(vec![
ScriptStep::RateLimit,
ScriptStep::RateLimit,
ScriptStep::RateLimit,
ScriptStep::Done,
]);
let attempts = transport.attempts_handle();
let mut sleeps = Vec::new();
let mut events = Vec::new();
stream_with_transport_attempt_result_with_parser_and_sleep::<
ScriptedTransport,
crate::providers::stream::StreamParser,
_,
>(
&transport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
crate::providers::stream::StreamParser::default(),
0,
&mut |event| {
events.push(event);
Ok(())
},
&mut |duration, _| {
sleeps.push(duration);
Ok(())
},
)
.unwrap();
assert_eq!(*attempts.lock().unwrap(), 4);
assert_eq!(
sleeps,
vec![
Duration::from_secs(2),
Duration::from_secs(10),
Duration::from_secs(20)
]
);
assert_eq!(events, vec![ProviderEvent::Done]);
}
#[test]
fn rate_limit_failure_stops_after_three_retries() {
let transport = ScriptedTransport::new(vec![
ScriptStep::RateLimit,
ScriptStep::RateLimit,
ScriptStep::RateLimit,
ScriptStep::RateLimit,
]);
let attempts = transport.attempts_handle();
let mut sleeps = Vec::new();
let error = stream_with_transport_attempt_result_with_parser_and_sleep::<
ScriptedTransport,
crate::providers::stream::StreamParser,
_,
>(
&transport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
crate::providers::stream::StreamParser::default(),
0,
&mut |_| Ok(()),
&mut |duration, _| {
sleeps.push(duration);
Ok(())
},
)
.unwrap_err()
.error
.to_string();
assert_eq!(*attempts.lock().unwrap(), 4);
assert_eq!(
sleeps,
vec![
Duration::from_secs(2),
Duration::from_secs(10),
Duration::from_secs(20)
]
);
assert!(error.contains("after 4 attempts"), "{error}");
}
#[test]
fn supplied_parser_is_used_for_first_attempt() {
let parser = RecordingParser::default();
let used = Arc::clone(&parser.used);
let mut events = Vec::new();
stream_with_transport_parser(
&SingleChunkTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
parser,
&mut |event| {
events.push(event);
Ok(())
},
)
.unwrap();
assert!(used.load(Ordering::SeqCst));
assert_eq!(events, vec![ProviderEvent::Done]);
}
#[test]
fn semantic_progress_resets_deadline_after_byte_silence() {
let mut events = Vec::new();
let attempt_error = stream_with_transport_attempt_result_with_parser::<
SemanticProgressResetTransport,
ExpiringProgressParser,
>(
&SemanticProgressResetTransport,
request(),
&AgentCancellation::default(),
Duration::from_millis(30),
ExpiringProgressParser::default(),
&mut |event| {
events.push(event);
Ok(())
},
)
.unwrap_err();
let error = attempt_error.error.to_string();
assert!(attempt_error.made_semantic_progress);
assert_eq!(events.len(), 2);
assert!(error.contains("no semantic progress"), "{error}");
}
#[test]
fn retry_backoff_stays_within_bounded_jitter_range() {
for _ in 0..128 {
let first = provider_retry_backoff(1);
assert!(
(Duration::from_millis(200)..=Duration::from_millis(300)).contains(&first),
"{first:?}"
);
let later = provider_retry_backoff(2);
assert!(
(Duration::from_millis(600)..=Duration::from_millis(900)).contains(&later),
"{later:?}"
);
}
}
#[test]
fn stream_error_after_partial_progress_reports_incomplete_response() {
let mut events = Vec::new();
let attempt_error = stream_with_transport_attempt_result_with_parser::<
ErrorAfterChunkTransport,
ProgressParser,
>(
&ErrorAfterChunkTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
ProgressParser,
&mut |event| {
events.push(event);
Ok(())
},
)
.unwrap_err();
let error = attempt_error.error.to_string();
assert!(attempt_error.made_semantic_progress);
assert!(matches!(events.as_slice(), [ProviderEvent::TextDelta(text)] if text == "partial"));
assert!(error.contains("partial response"), "{error}");
assert!(error.contains("incomplete"), "{error}");
assert!(error.contains("connection reset"), "{error}");
assert!(
attempt_error
.error
.downcast_ref::<ProviderError>()
.is_some()
);
}
#[test]
fn ttsr_interrupted_preserves_typed_error_through_stream_pipeline() {
let error = stream_with_transport_parser(
&SingleChunkTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
ProgressParser,
&mut |event| match event {
ProviderEvent::TextDelta(_) => Err(TtsrInterrupted::test_instance().into()),
_ => Ok(()),
},
)
.unwrap_err();
assert!(error.chain().any(|cause| cause.is::<TtsrInterrupted>()));
assert!(error.downcast_ref::<ProviderError>().is_none());
}
#[test]
fn stream_error_propagates_unsafe_recovery_progress_flag() {
let attempt_error = stream_with_transport_attempt_result_with_parser::<
ErrorAfterUnsafeProgressTransport,
UnsafeProgressParser,
>(
&ErrorAfterUnsafeProgressTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
UnsafeProgressParser,
&mut |_| Ok(()),
)
.unwrap_err();
let error = attempt_error.error.to_string();
assert!(attempt_error.made_semantic_progress);
assert!(attempt_error.unsafe_recovery_progress);
assert!(
error.contains("unsafe tool-call progress observed"),
"{error}"
);
}
#[test]
fn actionable_provider_error_after_partial_progress_keeps_original_identity() {
let attempt_error = stream_with_transport_attempt_result_with_parser::<
HttpStatusAfterChunkTransport,
ProgressParser,
>(
&HttpStatusAfterChunkTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
ProgressParser,
&mut |_| Ok(()),
)
.unwrap_err();
let error = attempt_error.error.to_string();
assert!(attempt_error.made_semantic_progress);
assert!(!error.contains("partial response"), "{error}");
assert!(
attempt_error
.error
.downcast_ref::<ProviderError>()
.is_some_and(ProviderError::is_codex_session_expired_401)
);
}
#[test]
fn stream_error_before_progress_keeps_clean_failure_diagnostic() {
let attempt_error = stream_with_transport_attempt_result_with_parser::<
ImmediateErrorTransport,
ProgressParser,
>(
&ImmediateErrorTransport,
request(),
&AgentCancellation::default(),
PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
ProgressParser,
&mut |_| Ok(()),
)
.unwrap_err();
let error = attempt_error.error.to_string();
assert!(!attempt_error.made_semantic_progress);
assert_eq!(error, "connection reset");
}
}