Skip to content

Mark stream decode errors retryable #9723

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Abhijay007 merged 5 commits into from
Jun 13, 2026
Merged

Mark stream decode errors retryable #9723

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
12eb55a
Mark stream decode errors retryable
Jun 10, 2026
11d67f8
Classify OpenAI stream parse errors as retryable
Jun 10, 2026
1cc6c1a
Preserve stream API errors
Jun 11, 2026
cd3a0b2
Merge remote-tracking branch 'origin/main' into codex/retry-stream-de…
Jun 11, 2026
cd08528
Preserve Ollama stream provider errors
Jun 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions crates/goose-providers/src/errors.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,10 @@ pub enum ProviderError {
}

impl ProviderError {
pub fn stream_decode_error(error: impl std::fmt::Display) -> Self {
ProviderError::NetworkError(format!("Stream decode error: {error}"))
}

pub fn telemetry_type(&self) -> &'static str {
match self {
ProviderError::Authentication(_) => "auth",
Expand All @@ -73,11 +77,12 @@ impl ProviderError {
}

/// Recover a typed `ProviderError` from a streaming decode error, falling
/// back to `RequestFailed` for errors that did not originate as one.
/// back to a retryable stream decode error for errors that did not
/// originate as one.
pub fn from_stream_error(error: anyhow::Error) -> Self {
error
.downcast()
.unwrap_or_else(|e| ProviderError::RequestFailed(format!("Stream decode error: {e}")))
.unwrap_or_else(ProviderError::stream_decode_error)
}
}

Expand Down
8 changes: 6 additions & 2 deletions crates/goose-providers/src/formats/openai.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -866,7 +866,9 @@ fn strip_data_prefix(line: &str) -> Option<&str> {

fn parse_streaming_chunk(line: &str) -> Result<StreamingChunk, ProviderError> {
let value: Value = serde_json::from_str(line).map_err(|e| {
ProviderError::RequestFailed(format!("Failed to parse streaming chunk: {e}: {line:?}"))
ProviderError::stream_decode_error(format!(
"Failed to parse streaming chunk: {e}: {line:?}"
))
})?;

if let Some(error) = value.get("error") {
Expand All @@ -886,7 +888,9 @@ fn parse_streaming_chunk(line: &str) -> Result<StreamingChunk, ProviderError> {
}

serde_json::from_value(value).map_err(|e| {
ProviderError::RequestFailed(format!("Failed to parse streaming chunk: {e}: {line:?}"))
ProviderError::stream_decode_error(format!(
"Failed to parse streaming chunk: {e}: {line:?}"
))
})
}

Expand Down
5 changes: 4 additions & 1 deletion crates/goose/src/providers/chatgpt_codex.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1020,7 +1020,10 @@ impl Provider for ChatGptCodexProvider {
let message_stream = responses_api_to_streaming_message(framed);
pin!(message_stream);
while let Some(message) = message_stream.next().await {
let (message, usage) = message.map_err(|e| ProviderError::RequestFailed(format!("Stream decode error: {}", e)))?;
let (message, usage) = message.map_err(|e| {
e.downcast::<ProviderError>()
.unwrap_or_else(ProviderError::stream_decode_error)
})?;
yield (message, usage);
}
}))
Expand Down
4 changes: 3 additions & 1 deletion crates/goose/src/providers/formats/google.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -433,7 +433,9 @@ where
.get("status")
.and_then(|s| s.as_str())
.unwrap_or("UNKNOWN");
Err(anyhow::anyhow!("Google API error ({}): {}", status, message))?;
Err::<(), ProviderError>(ProviderError::RequestFailed(format!(
"Google API error ({status}): {message}"
)))?;
}

if let Ok(usage) = get_usage(&chunk) {
Expand Down
27 changes: 16 additions & 11 deletions crates/goose/src/providers/formats/openai_responses.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
use crate::conversation::message::{Message, MessageContent};
use crate::mcp_utils::extract_text_from_resource;
use crate::model::ModelConfig;
use anyhow::{anyhow, Error};
use anyhow::Error;
use async_stream::try_stream;
use chrono;
use futures::Stream;
use goose_providers::conversation::token_usage::{ProviderUsage, Usage};
use goose_providers::errors::ProviderError;
use goose_providers::formats::openai::{
extract_reasoning_effort, is_openai_responses_model, openai_reasoning_effort_for_thinking,
};
Expand Down Expand Up @@ -249,11 +250,10 @@ fn is_known_responses_stream_event_type(event_type: &str) -> bool {

fn parse_responses_stream_event(data_line: &str) -> anyhow::Result<Option<ResponsesStreamEvent>> {
let raw_event: Value = serde_json::from_str(data_line).map_err(|e| {
anyhow!(
ProviderError::stream_decode_error(format!(
"Failed to parse Responses stream event: {}: {:?}",
e,
data_line
)
e, data_line
))
})?;

let Some(event_type) = raw_event.get("type").and_then(Value::as_str) else {
Expand All @@ -265,11 +265,10 @@ fn parse_responses_stream_event(data_line: &str) -> anyhow::Result<Option<Respon
}

let event = serde_json::from_value(raw_event).map_err(|e| {
anyhow!(
ProviderError::stream_decode_error(format!(
"Failed to parse Responses stream event: {}: {:?}",
e,
data_line
)
e, data_line
))
})?;
Ok(Some(event))
}
Expand Down Expand Up @@ -911,11 +910,17 @@ where
}

ResponsesStreamEvent::ResponseFailed { error, .. } => {
Err(anyhow!("Responses API failed: {:?}", error))?;
Err::<(), ProviderError>(ProviderError::RequestFailed(format!(
"Responses API failed: {:?}",
error
)))?;
}

ResponsesStreamEvent::Error { error } => {
Err(anyhow!("Responses API error: {:?}", error))?;
Err::<(), ProviderError>(ProviderError::RequestFailed(format!(
"Responses API error: {:?}",
error
)))?;
}

_ => {
Expand Down
7 changes: 4 additions & 3 deletions crates/goose/src/providers/gemini_oauth.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1015,9 +1015,10 @@ impl Provider for GeminiOAuthProvider {
let message_stream = response_to_streaming_message(raw_lines);
pin!(message_stream);
while let Some(message) = message_stream.next().await {
let (message, usage) = message.map_err(|e|
ProviderError::RequestFailed(format!("Stream decode error: {}", e))
)?;
let (message, usage) = message.map_err(|e| {
e.downcast::<ProviderError>()
.unwrap_or_else(ProviderError::stream_decode_error)
})?;
if message.is_some() || usage.is_some() {
log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?;
}
Expand Down
7 changes: 4 additions & 3 deletions crates/goose/src/providers/google.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,9 +232,10 @@ impl Provider for GoogleProvider {
let message_stream = response_to_streaming_message(framed);
pin!(message_stream);
while let Some(message) = message_stream.next().await {
let (message, usage) = message.map_err(|e|
ProviderError::RequestFailed(format!("Stream decode error: {}", e))
)?;
let (message, usage) = message.map_err(|e| {
e.downcast::<ProviderError>()
.unwrap_or_else(ProviderError::stream_decode_error)
})?;
if message.is_some() || usage.is_some() {
log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?;
}
Expand Down
4 changes: 1 addition & 3 deletions crates/goose/src/providers/ollama.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -473,9 +473,7 @@ fn stream_ollama(response: Response, mut log: RequestLog) -> Result<MessageStrea
pin!(message_stream);

while let Some(message) = message_stream.next().await {
let (message, usage) = message.map_err(|e|
ProviderError::RequestFailed(format!("Stream decode error: {}", e))
)?;
let (message, usage) = message.map_err(ProviderError::from_stream_error)?;
log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?;
yield (message, usage);
}
Expand Down
5 changes: 3 additions & 2 deletions crates/goose/src/providers/openai_compatible.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ pub fn stream_openai_compat(
while let Some(message) = message_stream.next().await {
let (message, usage) = message.map_err(|e|
e.downcast::<ProviderError>()
.unwrap_or_else(|e| ProviderError::RequestFailed(format!("Stream decode error: {e}")))
.unwrap_or_else(ProviderError::stream_decode_error)
Comment on lines 202 to +203

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Convert OpenAI parse failures before preserving errors

When an OpenAI-compatible SSE chunk is malformed or has an unexpected schema, formats/openai.rs::parse_streaming_chunk returns a ProviderError::RequestFailed("Failed to parse streaming chunk..."); this branch downcasts that error and returns it unchanged, so those stream decode failures still aren't classified as NetworkError like the other updated providers. In that malformed/truncated chunk scenario, any transient-only handling keyed off NetworkError still misses the main OpenAI-compatible chat/completions stream path.

Useful? React with 👍 / 👎.

)?;
log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?;
yield (message, usage);
Expand All @@ -223,7 +223,8 @@ pub fn stream_responses_compat(
pin!(message_stream);
while let Some(message) = message_stream.next().await {
let (message, usage) = message.map_err(|e|
ProviderError::RequestFailed(format!("Stream decode error: {e}"))
e.downcast::<ProviderError>()
.unwrap_or_else(ProviderError::stream_decode_error)
)?;
log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?;
yield (message, usage);
Expand Down
Loading