Merge pull request #155 from bnema/fix/retry-bare-stream-eof

fix(core): retry bare stream EOF errors

commit: 07cbc9804ac22208c26961540798438d0059ab3a
author: bnema · 2mo ago · 2026-08-23 09:42 UTC
committer: GitHub · 2mo ago · 2026-08-23 09:42 UTC
merge of: edaaa7c + c728340 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

2 files · +46 -8

M packages/core/agent.go +26 -8


@@ -5,6 +5,7 @@ import (
5	5		"encoding/json"
6	6		"errors"
7	7		"fmt"
	8	+	"io"
8	9		"strings"
9	10		"sync"
10	11		"time"

@@ -791,13 +792,14 @@ func (a *Agent) canRetryError(err error, attempt int) bool {
791	792		if msg == "" \|\| isNonRetryableProviderLimit(msg) {
792	793		return false
793	794		}
	795	+	if isStreamNetworkError(err, msg) {
	796	+	return true
	797	+	}
794	798		needles := []string{
795	799		"overloaded", "provider returned error", "rate limit", "ratelimit", "too many requests",
796	800		"429", "http 429", "500", "http 500", "502", "http 502", "503", "http 503", "504", "http 504",
797		-	"service unavailable", "server error", "internal error", "network error", "connection error",
798		-	"connection refused", "connection lost", "fetch failed", "upstream connect", "reset before headers",
799		-	"socket hang up", "ended without", "stream ended before", "did not get a response", "timed out",
800		-	"timeout", "terminated", "unexpected eof", "transport failure",
	801	+	"service unavailable", "server error", "internal error", "ended without", "stream ended before",
	802	+	"did not get a response", "timed out", "timeout", "terminated",
801	803		// OpenAI's ChatGPT/Codex backend returns this generic message (with a
802	804		// request ID) for transient server failures and explicitly says
803	805		// "You can retry your request".

@@ -820,6 +822,24 @@ func (a *Agent) canRetryError(err error, attempt int) bool {
820	822		return false
821	823		}
822	824
	825	+	// isStreamNetworkError detects transport errors from an opened provider stream.
	826	+	// Keep typed EOF checks because WebSocket readers return io.EOF on a peer close,
	827	+	// while provider-specific transports may only preserve a textual error.
	828	+	func isStreamNetworkError(err error, msg string) bool {
	829	+	if errors.Is(err, io.EOF) \|\| errors.Is(err, io.ErrUnexpectedEOF) {
	830	+	return true
	831	+	}
	832	+	for _, needle := range []string{
	833	+	"connection", "network", "unexpected eof", "broken pipe", "socket hang up",
	834	+	"transport failure", "upstream connect",
	835	+	} {
	836	+	if strings.Contains(msg, needle) {
	837	+	return true
	838	+	}
	839	+	}
	840	+	return false
	841	+	}
	842	+
823	843		func isNonRetryableProviderLimit(msg string) bool {
824	844		needles := []string{
825	845		"usage limit", "monthly usage limit", "freeusagelimit", "gousagelimit",

@@ -850,10 +870,8 @@ func classifyRetryReason(err error) RetryReason {
850	870		if errors.Is(err, context.DeadlineExceeded) \|\| strings.Contains(msg, "timeout") \|\| strings.Contains(msg, "timed out") {
851	871		return RetryReasonTimeout
852	872		}
853		-	for _, needle := range []string{"connection", "network", "unexpected eof", "broken pipe", "socket hang up", "transport failure", "upstream connect"} {
854		-	if strings.Contains(msg, needle) {
855		-	return RetryReasonNetwork
856		-	}
	873	+	if isStreamNetworkError(err, msg) {
	874	+	return RetryReasonNetwork
857	875		}
858	876		for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} {
859	877		if strings.Contains(msg, needle) {

 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"strings"
 	"sync"
 	"time"
 	if msg == "" || isNonRetryableProviderLimit(msg) {
 		return false
 	}
+	if isStreamNetworkError(err, msg) {
+		return true
+	}
 	needles := []string{
 		"overloaded", "provider returned error", "rate limit", "ratelimit", "too many requests",
 		"429", "http 429", "500", "http 500", "502", "http 502", "503", "http 503", "504", "http 504",
-		"service unavailable", "server error", "internal error", "network error", "connection error",
+		"service unavailable", "server error", "internal error", "ended without", "stream ended before",
-		"connection refused", "connection lost", "fetch failed", "upstream connect", "reset before headers",
+		"did not get a response", "timed out", "timeout", "terminated",
-		"socket hang up", "ended without", "stream ended before", "did not get a response", "timed out",
-		"timeout", "terminated", "unexpected eof", "transport failure",
 		// OpenAI's ChatGPT/Codex backend returns this generic message (with a
 		// request ID) for transient server failures and explicitly says
 		// "You can retry your request".
 	return false
 }
+// isStreamNetworkError detects transport errors from an opened provider stream.
+// Keep typed EOF checks because WebSocket readers return io.EOF on a peer close,
+// while provider-specific transports may only preserve a textual error.
+func isStreamNetworkError(err error, msg string) bool {
+	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+		return true
+	}
+	for _, needle := range []string{
+		"connection", "network", "unexpected eof", "broken pipe", "socket hang up",
+		"transport failure", "upstream connect",
+	} {
+		if strings.Contains(msg, needle) {
+			return true
+		}
+	}
+	return false
+}
 func isNonRetryableProviderLimit(msg string) bool {
 	needles := []string{
 		"usage limit", "monthly usage limit", "freeusagelimit", "gousagelimit",
 	if errors.Is(err, context.DeadlineExceeded) || strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out") {
 		return RetryReasonTimeout
 	}
-	for _, needle := range []string{"connection", "network", "unexpected eof", "broken pipe", "socket hang up", "transport failure", "upstream connect"} {
+	if isStreamNetworkError(err, msg) {
-		if strings.Contains(msg, needle) {
+		return RetryReasonNetwork
-			return RetryReasonNetwork
-		}
 	}
 	for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} {
 		if strings.Contains(msg, needle) {

M packages/core/agent_retry_test.go +20 -0


@@ -231,6 +231,26 @@ func TestAgentRetriesUnexpectedEOFStreamError(t *testing.T) {
231	231		}
232	232		}
233	233
	234	+	func TestAgentRetriesBareEOFStreamError(t *testing.T) {
	235	+	client := &retryFakeClient{firstErr: fmt.Errorf("read SSE: %w", io.EOF)}
	236	+	a := NewAgent(client, "fake-model", "system", Registry{})
	237	+	a.RetryBaseDelay = time.Millisecond
	238	+	var records []RetryLifecycleRecord
	239	+	a.OnRetryLifecycle = func(record RetryLifecycleRecord) {
	240	+	records = append(records, record)
	241	+	}
	242	+
	243	+	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
	244	+	t.Fatalf("Prompt returned %v", err)
	245	+	}
	246	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
	247	+	t.Fatalf("Stream calls = %d; want 2", got)
	248	+	}
	249	+	if len(records) != 2 \|\| records[0].Reason != RetryReasonNetwork \|\| records[1].Reason != RetryReasonNetwork {
	250	+	t.Fatalf("retry lifecycle = %#v; want network failure and retry", records)
	251	+	}
	252	+	}
	253	+
234	254		func TestAgentEmitsRetryLifecycleEvents(t *testing.T) {
235	255		client := &retryFakeClient{}
236	256		a := NewAgent(client, "fake-model", "system", Registry{})


@@ -231,6 +231,26 @@ func TestAgentRetriesUnexpectedEOFStreamError(t *testing.T) {
231	}	231		}
232	}	232		}
233		233
		234	+	func TestAgentRetriesBareEOFStreamError(t *testing.T) {
		235	+	client := &retryFakeClient{firstErr: fmt.Errorf("read SSE: %w", io.EOF)}
		236	+	a := NewAgent(client, "fake-model", "system", Registry{})
		237	+	a.RetryBaseDelay = time.Millisecond
		238	+	var records []RetryLifecycleRecord
		239	+	a.OnRetryLifecycle = func(record RetryLifecycleRecord) {
		240	+	records = append(records, record)
		241	+	}
		242	+
		243	+	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
		244	+	t.Fatalf("Prompt returned %v", err)
		245	+	}
		246	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
		247	+	t.Fatalf("Stream calls = %d; want 2", got)
		248	+	}
		249	+	if len(records) != 2 \|\| records[0].Reason != RetryReasonNetwork \|\| records[1].Reason != RetryReasonNetwork {
		250	+	t.Fatalf("retry lifecycle = %#v; want network failure and retry", records)
		251	+	}
		252	+	}
		253	+
234	func TestAgentEmitsRetryLifecycleEvents(t *testing.T) {	254		func TestAgentEmitsRetryLifecycleEvents(t *testing.T) {
235	client := &retryFakeClient{}	255		client := &retryFakeClient{}
236	a := NewAgent(client, "fake-model", "system", Registry{})	256		a := NewAgent(client, "fake-model", "system", Registry{})