From 5f3ccea465a8ee0847fae182d27d23776b9120d5 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Thu, 27 Aug 2026 07:10:03 +0000 Subject: [PATCH 1/8] fix(agent): retry mid-stream transport aborts before tool dispatch Connect-time streamWithReconnect and the CollectStream stall path already recover from transient disconnects and idle timeouts, but a failure DURING CollectStream that is a transport abort (Windows wsarecv/WSAECONNABORTED, connection reset by peer, forcibly closed) still aborted the turn and forced a manual continue. Classify those mid-stream aborts via shouldReconnect (single-sourced) and reuse the existing stall-retry loop with the same safety rules: no forwarded visible prose, empty collected.Text, and error before tool dispatch. Bound unchanged (maxStreamStallRetries=1); transport aborts surface reconnect wording, stalls keep the stall notice. Fixes #973 --- internal/agent/loop.go | 31 ++++-- internal/agent/midstream_retry_test.go | 135 +++++++++++++++++++++++++ internal/agent/reconnect.go | 22 ++++ internal/agent/reconnect_test.go | 4 + 4 files changed, 183 insertions(+), 9 deletions(-) create mode 100644 internal/agent/midstream_retry_test.go diff --git a/internal/agent/loop.go b/internal/agent/loop.go index fe691ac4c..d4e473bd3 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -25,14 +25,16 @@ const maxTurnsAnswer = "Agent reached maximum number of turns without a final an const maxTurnsFinalAnswerPrompt = "You have reached the tool-turn limit. Do not call tools. Give a concise final answer now: summarize what you completed, what you found, and any remaining blockers." // maxStreamStallRetries bounds how many times a turn that timed out (idle/stall) -// WITH NO OUTPUT yet is re-issued on a fresh connection before giving up. Only -// the no-output case is retried (a partial turn would duplicate), so this is a -// safe recovery for a stalled/dead pooled connection. +// OR hit a mid-stream transport abort (#973) WITH NO OUTPUT yet is re-issued on +// a fresh connection before giving up. Only the no-output case is retried (a +// partial turn would duplicate), so this is a safe recovery for a stalled/dead +// pooled connection or a socket abort before tool dispatch. // // Set to 1 (not 2): each attempt can itself idle for the full stream timeout // (~5min) before the stall is even detected, so 2 retries left an interactive // session frozen for ~15min. One retry keeps the common single-hiccup recovery -// while bounding the worst case to ~2× the idle timeout. +// while bounding the worst case to ~2× the idle timeout. Mid-stream transport +// aborts share this bound (do not raise it for #973). const maxStreamStallRetries = 1 const ( @@ -450,9 +452,10 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) result.Messages = copyMessages(messages) return result, ctx.Err() } - // A stream idle/stall timeout is safely re-issued when the turn committed NO - // answer text — no forwarded visible prose (forwardedVisibleText) and no - // collected final text (collected.Text). This covers two cases: + // A stream idle/stall timeout OR mid-stream transport abort (#973 — wsarecv / + // connection reset / forcibly closed) is safely re-issued when the turn + // committed NO answer text — no forwarded visible prose (forwardedVisibleText) + // and no collected final text (collected.Text). This covers three cases: // 1. Nothing streamed at all before the connection died (the original // macOS stale-pooled-connection hang past the response-header timeout). // 2. The model streamed transient reasoning and began a tool call (e.g. a @@ -463,13 +466,23 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) // re-render is transient previews, not duplicated answer text. This is // why the gate no longer excludes collected.ToolCalls: an incomplete // tool call from a timed-out stream is discard-and-retry, not output. + // 3. Mid-stream socket abort after connect succeeded (Windows WSAECONNABORTED / + // wsarecv, connection reset by peer). Same no-tool-executed safety: the + // error returns before tool dispatch, so a bounded retry is safe. // A turn that forwarded real prose is NOT retried (it would duplicate visible // answer text) and falls through to the error return below. Capped + // exponential backoff, with a user-visible notice per attempt. for attempt := 1; attempt <= maxStreamStallRetries && - isStreamTimeoutError(collected.Error) && !forwardedVisibleText && + (isStreamTimeoutError(collected.Error) || isMidStreamTransportAbort(collected.Error)) && + !forwardedVisibleText && collected.Text == ""; attempt++ { - if notify := stallRetryNoticeFor(options); notify != nil { + var notify reconnectNotifier + if isStreamTimeoutError(collected.Error) { + notify = stallRetryNoticeFor(options) + } else { + notify = reconnectNoticeFor(options) + } + if notify != nil { notify(attempt, maxStreamStallRetries) } if err := sleepWithContext(ctx, backoffFor(attempt)); err != nil { diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go new file mode 100644 index 000000000..37f941288 --- /dev/null +++ b/internal/agent/midstream_retry_test.go @@ -0,0 +1,135 @@ +package agent + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/Gitlawb/zero/internal/tools" + "github.com/Gitlawb/zero/internal/zeroruntime" +) + +// midStreamAbortProvider connects successfully but emits a transport-abort +// StreamEventError on the first abortBefore calls, then succeeds with "done". +// Mirrors stallProvider shapes so mid-stream abort retries stay in parity with +// the idle/stall path (#973). +type midStreamAbortProvider struct { + calls int32 + abortBefore int32 + abortError string + partialText string + partialToolCall string +} + +func (p *midStreamAbortProvider) StreamCompletion(_ context.Context, _ zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) { + n := atomic.AddInt32(&p.calls, 1) + ch := make(chan zeroruntime.StreamEvent, 5) + if n <= p.abortBefore { + if p.partialText != "" { + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventText, Content: p.partialText} + } + if p.partialToolCall != "" { + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventToolCallStart, ToolCallID: "tc_1", ToolName: p.partialToolCall} + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventToolCallDelta, ToolCallID: "tc_1", ArgumentsFragment: `{"path":"x.html","content":" Date: Thu, 27 Aug 2026 09:32:03 +0000 Subject: [PATCH 2/8] fix(agent): align mid-stream retry comment with no-answer-text gate Address CodeRabbit review on #976. --- internal/agent/loop.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/internal/agent/loop.go b/internal/agent/loop.go index d4e473bd3..ec7b4a4c7 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -25,10 +25,12 @@ const maxTurnsAnswer = "Agent reached maximum number of turns without a final an const maxTurnsFinalAnswerPrompt = "You have reached the tool-turn limit. Do not call tools. Give a concise final answer now: summarize what you completed, what you found, and any remaining blockers." // maxStreamStallRetries bounds how many times a turn that timed out (idle/stall) -// OR hit a mid-stream transport abort (#973) WITH NO OUTPUT yet is re-issued on -// a fresh connection before giving up. Only the no-output case is retried (a -// partial turn would duplicate), so this is a safe recovery for a stalled/dead -// pooled connection or a socket abort before tool dispatch. +// OR hit a mid-stream transport abort (#973) with no answer text yet is re-issued +// on a fresh connection before giving up. Incomplete tool-call previews are +// allowed through the gate (they are never dispatched before the error return). +// The safety claim is only that visible answer prose is not duplicated on retry — +// not that every partial stream shape is excluded. Covers a stalled/dead pooled +// connection or a socket abort before tool dispatch. // // Set to 1 (not 2): each attempt can itself idle for the full stream timeout // (~5min) before the stall is even detected, so 2 retries left an interactive From b7fdbc8d805996b88cab2b7e56b410989072b5c0 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Thu, 27 Aug 2026 09:32:04 +0000 Subject: [PATCH 3/8] test(agent): assert no tool dispatch and mid-stream abort exhaustion Address CodeRabbit review on #976. --- internal/agent/midstream_retry_test.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go index 37f941288..c60092174 100644 --- a/internal/agent/midstream_retry_test.go +++ b/internal/agent/midstream_retry_test.go @@ -95,11 +95,13 @@ func TestRunDoesNotRetryMidStreamAbortAfterPartialOutput(t *testing.T) { // incomplete call is never executed or committed before the error return. func TestRunRetriesMidStreamAbortAfterIncompleteToolCall(t *testing.T) { starts := 0 + dispatched := 0 p := &midStreamAbortProvider{abortBefore: 1, partialToolCall: "write_file"} result, err := Run(context.Background(), "go", p, Options{ Registry: tools.NewRegistry(), OnToolCallStart: func(string, string) { starts++ }, OnToolCallDelta: func(string, string) {}, + OnToolCall: func(ToolCall) { dispatched++ }, }) if err != nil { t.Fatalf("a mid-stream abort mid-incomplete-tool-call should retry to success, got %v", err) @@ -113,6 +115,22 @@ func TestRunRetriesMidStreamAbortAfterIncompleteToolCall(t *testing.T) { if starts != 1 { t.Fatalf("the incomplete tool call should have been forwarded once before the abort, got %d starts", starts) } + if dispatched != 0 { + t.Fatalf("OnToolCall must never fire for an incomplete aborted tool call (no dispatch), got %d", dispatched) + } +} + +// A persistent mid-stream transport abort surfaces an error after exhausting the +// capped retries (1 initial stream + maxStreamStallRetries retries = 2 attempts). +func TestRunGivesUpAfterMaxMidStreamAbortRetries(t *testing.T) { + p := &midStreamAbortProvider{abortBefore: 99} + _, err := Run(context.Background(), "go", p, Options{Registry: tools.NewRegistry()}) + if err == nil { + t.Fatal("a persistent mid-stream abort must surface an error after exhausting retries") + } + if got := atomic.LoadInt32(&p.calls); got != int32(1+maxStreamStallRetries) { + t.Fatalf("want %d calls (1 + %d retries), got %d", 1+maxStreamStallRetries, maxStreamStallRetries, got) + } } func TestIsMidStreamTransportAbort(t *testing.T) { From 975489fc291f07a96983537065d3106c3c6e0109 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Thu, 27 Aug 2026 17:15:38 +0000 Subject: [PATCH 4/8] fix(agent): narrow mid-stream abort gate and keep Canceled sentinel Address Vasanthdev2004 review on #976: - isMidStreamTransportAbort no longer delegates to shouldReconnect. Mid-stream retries match abort/reset/EOF/close needles only, not connect-phase timeout or connection refused (a slow healthy server must not cost a second prefill). - Recheck ctx.Err() after a retried CollectStream so ACP still sees errors.Is(err, context.Canceled). - Pin wsarecv without the aborted substring, the retry bound at 1, reconnect notice wording, forwardedVisibleText, header-timeout non-retry, and the cancel-during-retry sentinel. --- internal/agent/loop.go | 21 +++- internal/agent/midstream_retry_test.go | 165 +++++++++++++++++++++++-- internal/agent/reconnect.go | 40 ++++-- internal/agent/reconnect_test.go | 2 +- 4 files changed, 206 insertions(+), 22 deletions(-) diff --git a/internal/agent/loop.go b/internal/agent/loop.go index ec7b4a4c7..0cb3a3d4a 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -455,9 +455,13 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) return result, ctx.Err() } // A stream idle/stall timeout OR mid-stream transport abort (#973 — wsarecv / - // connection reset / forcibly closed) is safely re-issued when the turn - // committed NO answer text — no forwarded visible prose (forwardedVisibleText) - // and no collected final text (collected.Text). This covers three cases: + // connection reset / unexpected EOF / connection closed / broken pipe) is + // safely re-issued when the turn committed NO answer text — no forwarded + // visible prose (forwardedVisibleText) and no collected final text + // (collected.Text). The abort gate matches midStreamAbortNeedles only, not + // the full shouldReconnect list: connect-phase timeouts and connection + // refused are NOT retried here (a slow healthy server must not cost a + // second prefill). This covers three cases: // 1. Nothing streamed at all before the connection died (the original // macOS stale-pooled-connection hang past the response-header timeout). // 2. The model streamed transient reasoning and began a tool call (e.g. a @@ -503,6 +507,13 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) collected = zeroruntime.CollectStreamWithOptions(ctx, retryStream, forwardingOpts) stallGenSpan.End() } + // Recheck ctx after a retried CollectStream: helpers.go stores + // ctx.Err().Error() in collected.Error, and errors.New of that string + // drops the context.Canceled sentinel ACP branches on. + if ctx.Err() != nil { + result.Messages = copyMessages(messages) + return result, ctx.Err() + } if collected.Error != "" { // Route a reissued stream's non-stall error through the SAME recovery as // the initial stream (image-rejection wrapping / context-limit compaction) @@ -513,6 +524,10 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) result.Messages = copyMessages(messages) return result, stop } + if ctx.Err() != nil { + result.Messages = copyMessages(messages) + return result, ctx.Err() + } if collected.Error != "" { result.Messages = copyMessages(messages) return result, errors.New(collected.Error) diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go index c60092174..d338ed4fe 100644 --- a/internal/agent/midstream_retry_test.go +++ b/internal/agent/midstream_retry_test.go @@ -2,8 +2,11 @@ package agent import ( "context" + "errors" + "strings" "sync/atomic" "testing" + "time" "github.com/Gitlawb/zero/internal/tools" "github.com/Gitlawb/zero/internal/zeroruntime" @@ -11,20 +14,37 @@ import ( // midStreamAbortProvider connects successfully but emits a transport-abort // StreamEventError on the first abortBefore calls, then succeeds with "done". -// Mirrors stallProvider shapes so mid-stream abort retries stay in parity with -// the idle/stall path (#973). +// hangOnCall, if > 0, makes that 1-based call block until ctx is done so a +// cancel during the retried CollectStream can be reproduced. type midStreamAbortProvider struct { calls int32 abortBefore int32 abortError string partialText string + emptyTextEvent bool partialToolCall string + hangOnCall int32 + started chan struct{} } -func (p *midStreamAbortProvider) StreamCompletion(_ context.Context, _ zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) { +func (p *midStreamAbortProvider) StreamCompletion(ctx context.Context, _ zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) { n := atomic.AddInt32(&p.calls, 1) + if p.hangOnCall > 0 && n == p.hangOnCall { + if p.started != nil { + close(p.started) + } + ch := make(chan zeroruntime.StreamEvent) + go func() { + <-ctx.Done() + close(ch) + }() + return ch, nil + } ch := make(chan zeroruntime.StreamEvent, 5) if n <= p.abortBefore { + if p.emptyTextEvent { + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventText, Content: ""} + } if p.partialText != "" { ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventText, Content: p.partialText} } @@ -80,19 +100,60 @@ func TestRunRetriesMidStreamWindowsAbort(t *testing.T) { } } +// wsarecv alone (no "connection was aborted" substring) must still retry, so +// deleting the wsarecv needle cannot go green. +func TestRunRetriesMidStreamWsarecvNeedle(t *testing.T) { + p := &midStreamAbortProvider{ + abortBefore: 1, + abortError: "provider stream error: read: wsarecv: 10053", + } + result, err := Run(context.Background(), "go", p, Options{Registry: tools.NewRegistry()}) + if err != nil { + t.Fatalf("wsarecv without 'connection was aborted' should retry to success, got %v", err) + } + if result.FinalAnswer != "done" { + t.Fatalf("final answer = %q, want %q", result.FinalAnswer, "done") + } + if got := atomic.LoadInt32(&p.calls); got != 2 { + t.Fatalf("want 2 calls (1 abort + 1 retry), got %d", got) + } +} + func TestRunDoesNotRetryMidStreamAbortAfterPartialOutput(t *testing.T) { + sawText := 0 p := &midStreamAbortProvider{abortBefore: 1, partialText: "partial"} - _, err := Run(context.Background(), "go", p, Options{Registry: tools.NewRegistry()}) + _, err := Run(context.Background(), "go", p, Options{ + Registry: tools.NewRegistry(), + OnText: func(string) { sawText++ }, + }) if err == nil { t.Fatal("a mid-stream abort after partial output must NOT be retried; want an error") } if got := atomic.LoadInt32(&p.calls); got != 1 { t.Fatalf("partial-then-abort must not retry, got %d calls", got) } + if sawText == 0 { + t.Fatal("OnText must have fired so forwardedVisibleText is the named guard") + } +} + +// Empty StreamEventText still invokes OnText, so forwardedVisibleText is true +// while collected.Text stays empty. Neutralising !forwardedVisibleText in the +// gate would retry this; collected.Text == "" would not catch it. +func TestRunDoesNotRetryMidStreamAbortAfterForwardedEmptyText(t *testing.T) { + p := &midStreamAbortProvider{abortBefore: 1, emptyTextEvent: true} + _, err := Run(context.Background(), "go", p, Options{ + Registry: tools.NewRegistry(), + OnText: func(string) {}, + }) + if err == nil { + t.Fatal("forwarded visible text (even empty chunk) must block retry") + } + if got := atomic.LoadInt32(&p.calls); got != 1 { + t.Fatalf("forwardedVisibleText must block retry, got %d calls", got) + } } -// Incomplete tool call then abort SHOULD retry (parity with stall path): the -// incomplete call is never executed or committed before the error return. func TestRunRetriesMidStreamAbortAfterIncompleteToolCall(t *testing.T) { starts := 0 dispatched := 0 @@ -120,31 +181,113 @@ func TestRunRetriesMidStreamAbortAfterIncompleteToolCall(t *testing.T) { } } -// A persistent mid-stream transport abort surfaces an error after exhausting the -// capped retries (1 initial stream + maxStreamStallRetries retries = 2 attempts). func TestRunGivesUpAfterMaxMidStreamAbortRetries(t *testing.T) { + if maxStreamStallRetries != 1 { + t.Fatalf("maxStreamStallRetries = %d, want 1 (#973 must not raise this bound)", maxStreamStallRetries) + } + defer func(orig time.Duration) { streamReconnectBase = orig }(streamReconnectBase) + streamReconnectBase = time.Millisecond p := &midStreamAbortProvider{abortBefore: 99} _, err := Run(context.Background(), "go", p, Options{Registry: tools.NewRegistry()}) if err == nil { t.Fatal("a persistent mid-stream abort must surface an error after exhausting retries") } - if got := atomic.LoadInt32(&p.calls); got != int32(1+maxStreamStallRetries) { - t.Fatalf("want %d calls (1 + %d retries), got %d", 1+maxStreamStallRetries, maxStreamStallRetries, got) + if got := atomic.LoadInt32(&p.calls); got != 2 { + t.Fatalf("want 2 calls (1 abort + 1 retry); pin the bound at 1, got %d", got) + } +} + +func TestRunMidStreamAbortUsesReconnectNotice(t *testing.T) { + defer func(orig time.Duration) { streamReconnectBase = orig }(streamReconnectBase) + streamReconnectBase = time.Millisecond + var notices string + p := &midStreamAbortProvider{abortBefore: 1} + result, err := Run(context.Background(), "go", p, Options{ + Registry: tools.NewRegistry(), + OnReasoning: func(s string) { notices += s }, + }) + if err != nil { + t.Fatalf("retry should succeed, got %v", err) + } + if result.FinalAnswer != "done" { + t.Fatalf("final answer = %q, want %q", result.FinalAnswer, "done") + } + lower := strings.ToLower(notices) + if !strings.Contains(lower, "connection lost") || !strings.Contains(lower, "reconnecting") { + t.Fatalf("transport abort must use reconnect wording, notices = %q", notices) + } + if strings.Contains(lower, "model stalled") { + t.Fatalf("transport abort must not use stall wording, notices = %q", notices) + } +} + +func TestRunCancelDuringMidStreamRetryPreservesContextCanceled(t *testing.T) { + defer func(orig time.Duration) { streamReconnectBase = orig }(streamReconnectBase) + streamReconnectBase = time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + started := make(chan struct{}) + p := &midStreamAbortProvider{abortBefore: 1, hangOnCall: 2, started: started} + errCh := make(chan error, 1) + go func() { + _, err := Run(ctx, "go", p, Options{Registry: tools.NewRegistry()}) + errCh <- err + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for retried stream") + } + cancel() + err := <-errCh + if err == nil { + t.Fatal("want context.Canceled after cancel during retried stream") + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancel during retried stream must keep the sentinel, got %q (errors.Is Canceled = false)", err) + } +} + +func TestRunDoesNotRetryResponseHeaderTimeout(t *testing.T) { + p := &midStreamAbortProvider{ + abortBefore: 99, + abortError: "net/http: timeout awaiting response headers", + } + _, err := Run(context.Background(), "go", p, Options{Registry: tools.NewRegistry()}) + if err == nil { + t.Fatal("a response-header timeout must not be retried as a mid-stream abort") + } + if got := atomic.LoadInt32(&p.calls); got != 1 { + t.Fatalf("header timeout must not retry, got %d calls", got) } } func TestIsMidStreamTransportAbort(t *testing.T) { aborts := []string{ "provider stream error: read: connection reset by peer", + "provider stream error: unexpected EOF", + "provider stream error: read: wsarecv: 10053", "provider stream error: read: wsarecv: An established connection was aborted by the software in your host machine", "An existing connection was forcibly closed by the remote host", + "provider stream error: read: connection closed", + "write: broken pipe", + "provider stream error: server closed the connection", } for _, m := range aborts { if !isMidStreamTransportAbort(m) { t.Fatalf("want mid-stream transport abort: %q", m) } } - notAborts := []string{"", "context length exceeded", "rate limit error: slow down", "model not found"} + notAborts := []string{ + "", + "context length exceeded", + "rate limit error: slow down", + "model not found", + "net/http: timeout awaiting response headers", + "dial tcp 10.0.0.1:443: connect: connection refused", + "i/o timeout", + "504 Gateway Timeout", + } for _, m := range notAborts { if isMidStreamTransportAbort(m) { t.Fatalf("must NOT classify as mid-stream transport abort: %q", m) diff --git a/internal/agent/reconnect.go b/internal/agent/reconnect.go index c9653c294..ba472783e 100644 --- a/internal/agent/reconnect.go +++ b/internal/agent/reconnect.go @@ -2,7 +2,6 @@ package agent import ( "context" - "errors" "fmt" "math/rand" "strings" @@ -158,20 +157,47 @@ func shouldReconnect(ctx context.Context, err error) bool { return false } +// midStreamAbortNeedles are the transport-abort classes retried AFTER connect +// succeeded and the stream body has started. This is a subset of shouldReconnect: +// connect-phase signals (timeout, connection refused, "temporarily unavailable") +// stay on the connect retry path, because matching them here would re-prefill a +// healthy-but-slow server (ollama cloud header timeouts) and tell the user the +// connection was lost. Issue #973 names WSAECONNABORTED / connection reset / +// unexpected stream EOF. +var midStreamAbortNeedles = []string{ + "eof", + "unexpected end", + "connection reset", + "broken pipe", + "connection closed", + "server closed", + "wsarecv", + "connection was aborted", + "forcibly closed", +} + // isMidStreamTransportAbort reports whether a collected stream error string is a -// retryable mid-stream transport abort (connection reset, Windows wsarecv / -// WSAECONNABORTED, forcibly closed, etc.). Classification is single-sourced -// through shouldReconnect so connect-time reconnect and post-connect CollectStream -// retries stay in lockstep. +// retryable mid-stream transport abort. It does NOT delegate to shouldReconnect: +// that list is a connect-phase argument ("no response was received"). The stream +// body has already started here, so only abort/reset/EOF/close classes retry. // // Used for failures DURING the stream body AFTER a successful connect and BEFORE // tool dispatch — the incomplete turn committed no answer text and executed no // tools, so a bounded re-issue is safe (same safety rules as the stall path). func isMidStreamTransportAbort(message string) bool { - if message == "" { + lowered := strings.ToLower(strings.TrimSpace(message)) + if lowered == "" || isContextLimitError(lowered) { + return false + } + if errhint.HasStatusCode(lowered, "500", "502", "503", "504") { return false } - return shouldReconnect(context.Background(), errors.New(message)) + for _, needle := range midStreamAbortNeedles { + if strings.Contains(lowered, needle) { + return true + } + } + return false } // backoffFor is the deterministic exponential base delay for a 1-based attempt, diff --git a/internal/agent/reconnect_test.go b/internal/agent/reconnect_test.go index 01cc9040e..cc4d762bf 100644 --- a/internal/agent/reconnect_test.go +++ b/internal/agent/reconnect_test.go @@ -92,7 +92,7 @@ func TestShouldReconnectClassification(t *testing.T) { disconnects := []string{ "unexpected EOF", "connection reset by peer", "broken pipe", "i/o timeout", "server closed the connection", "connection refused", - // Windows mid-stream aborts (#973). + // Windows socket aborts also remain reconnectable at connect time. "wsarecv: An established connection was aborted by the software in your host machine", "read: connection was aborted", "An existing connection was forcibly closed by the remote host", From 6edde0ed86acfd80eb9a1f0d5546314812b04861 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Thu, 27 Aug 2026 17:47:17 +0000 Subject: [PATCH 5/8] fix(agent): do not treat empty OnText as committed answer prose CodeRabbit nit on #976: OnText("") was setting forwardedVisibleText and blocking an eligible mid-stream abort retry even though collected.Text stayed empty. Only non-empty text counts as forwarded visible prose. --- internal/agent/loop.go | 6 +++++- internal/agent/midstream_retry_test.go | 20 +++++++++++--------- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 0cb3a3d4a..0042ab1f3 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -378,7 +378,11 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) rec.StampFirstToken() } if onText != nil { - forwardedVisibleText = true + // Zero-length chunks are not committed answer prose; counting + // them would block an eligible mid-stream abort retry. + if s != "" { + forwardedVisibleText = true + } onText(s) } } diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go index d338ed4fe..2b4f9eaeb 100644 --- a/internal/agent/midstream_retry_test.go +++ b/internal/agent/midstream_retry_test.go @@ -137,20 +137,22 @@ func TestRunDoesNotRetryMidStreamAbortAfterPartialOutput(t *testing.T) { } } -// Empty StreamEventText still invokes OnText, so forwardedVisibleText is true -// while collected.Text stays empty. Neutralising !forwardedVisibleText in the -// gate would retry this; collected.Text == "" would not catch it. -func TestRunDoesNotRetryMidStreamAbortAfterForwardedEmptyText(t *testing.T) { +// A zero-length OnText chunk is not committed answer prose, so an eligible +// transport abort still retries. +func TestRunRetriesMidStreamAbortAfterEmptyTextEvent(t *testing.T) { p := &midStreamAbortProvider{abortBefore: 1, emptyTextEvent: true} - _, err := Run(context.Background(), "go", p, Options{ + result, err := Run(context.Background(), "go", p, Options{ Registry: tools.NewRegistry(), OnText: func(string) {}, }) - if err == nil { - t.Fatal("forwarded visible text (even empty chunk) must block retry") + if err != nil { + t.Fatalf("empty OnText must not block mid-stream abort retry, got %v", err) } - if got := atomic.LoadInt32(&p.calls); got != 1 { - t.Fatalf("forwardedVisibleText must block retry, got %d calls", got) + if result.FinalAnswer != "done" { + t.Fatalf("final answer = %q, want %q", result.FinalAnswer, "done") + } + if got := atomic.LoadInt32(&p.calls); got != 2 { + t.Fatalf("want 2 calls (1 abort + 1 retry), got %d", got) } } From 0aa6fa1f4ac6952b2ac3ec9f2e1a42e6d93223a0 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Tue, 1 Sep 2026 03:34:37 -0400 Subject: [PATCH 6/8] fix(agent): keep Canceled across replacement connect backoff A mid-stream abort retry that then failed to reconnect returned the network error when the user cancelled during streamWithReconnect's backoff, so ACP treated a cancelled turn as an internal error. Count the post-connect reissue as a reconnect even when the replacement connect succeeds immediately. --- internal/agent/loop.go | 7 +++ internal/agent/midstream_retry_test.go | 73 ++++++++++++++++++++++++++ 2 files changed, 80 insertions(+) diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 0042ab1f3..85e84ed16 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -491,6 +491,10 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) notify = stallRetryNoticeFor(options) } else { notify = reconnectNoticeFor(options) + // Count this post-connect reissue even when the replacement + // connect succeeds immediately. streamWithReconnect only + // increments reconnect_count after a failed connect attempt. + options.Trace.Counter(trace.CounterReconnectCount, 1) } if notify != nil { notify(attempt, maxStreamStallRetries) @@ -505,6 +509,9 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) retryStream, retryErr := streamWithReconnect(ctx, provider, retryRequest, reconnectNoticeFor(options)) if retryErr != nil { result.Messages = copyMessages(messages) + if ctx.Err() != nil { + return result, ctx.Err() + } return result, retryErr } stallGenSpan := options.Trace.Span(trace.SpanGeneration) diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go index 2b4f9eaeb..7046d658a 100644 --- a/internal/agent/midstream_retry_test.go +++ b/internal/agent/midstream_retry_test.go @@ -9,6 +9,7 @@ import ( "time" "github.com/Gitlawb/zero/internal/tools" + "github.com/Gitlawb/zero/internal/trace" "github.com/Gitlawb/zero/internal/zeroruntime" ) @@ -16,6 +17,8 @@ import ( // StreamEventError on the first abortBefore calls, then succeeds with "done". // hangOnCall, if > 0, makes that 1-based call block until ctx is done so a // cancel during the retried CollectStream can be reproduced. +// connectFailFrom, if > 0, makes that 1-based call and later return a +// reconnectable connect error so streamWithReconnect enters its backoff. type midStreamAbortProvider struct { calls int32 abortBefore int32 @@ -24,11 +27,27 @@ type midStreamAbortProvider struct { emptyTextEvent bool partialToolCall string hangOnCall int32 + connectFailFrom int32 + connectErr string started chan struct{} } func (p *midStreamAbortProvider) StreamCompletion(ctx context.Context, _ zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) { n := atomic.AddInt32(&p.calls, 1) + if p.connectFailFrom > 0 && n >= p.connectFailFrom { + if p.started != nil { + select { + case <-p.started: + default: + close(p.started) + } + } + errMsg := p.connectErr + if errMsg == "" { + errMsg = "connection reset by peer" + } + return nil, errors.New(errMsg) + } if p.hangOnCall > 0 && n == p.hangOnCall { if p.started != nil { close(p.started) @@ -250,6 +269,60 @@ func TestRunCancelDuringMidStreamRetryPreservesContextCanceled(t *testing.T) { } } +func TestRunCancelDuringReplacementConnectBackoffPreservesContextCanceled(t *testing.T) { + defer func(orig time.Duration) { streamReconnectBase = orig }(streamReconnectBase) + streamReconnectBase = 200 * time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + started := make(chan struct{}) + p := &midStreamAbortProvider{abortBefore: 1, connectFailFrom: 2, started: started} + errCh := make(chan error, 1) + go func() { + _, err := Run(ctx, "go", p, Options{Registry: tools.NewRegistry()}) + errCh <- err + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for replacement connect failure") + } + cancel() + err := <-errCh + if err == nil { + t.Fatal("want context.Canceled after cancel during replacement connect backoff") + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancel during replacement connect backoff must keep the sentinel, got %q (errors.Is Canceled = false)", err) + } +} + +func TestRunMidStreamAbortRecordsReconnectCountOnImmediateReplacement(t *testing.T) { + defer func(orig time.Duration) { streamReconnectBase = orig }(streamReconnectBase) + streamReconnectBase = time.Millisecond + rec := trace.NewRecorder("midstream-abort", "run-1", "test") + p := &midStreamAbortProvider{abortBefore: 1} + result, err := Run(context.Background(), "go", p, Options{ + Registry: tools.NewRegistry(), + Trace: rec, + }) + if err != nil { + t.Fatalf("retry should succeed, got %v", err) + } + if result.FinalAnswer != "done" { + t.Fatalf("final answer = %q, want %q", result.FinalAnswer, "done") + } + if got := atomic.LoadInt32(&p.calls); got != 2 { + t.Fatalf("want 2 calls (1 abort + 1 immediate replacement), got %d", got) + } + gotTrace := rec.Finish() + if got := gotTrace.Counter(trace.CounterReconnectCount); got != 1 { + t.Fatalf("reconnect_count = %d, want 1 for a successful replacement connection", got) + } + if got := gotTrace.Counter(trace.CounterModelRequests); got != 2 { + t.Fatalf("model_requests = %d, want 2 (initial + replacement)", got) + } +} + func TestRunDoesNotRetryResponseHeaderTimeout(t *testing.T) { p := &midStreamAbortProvider{ abortBefore: 99, From e4069b484b0ec64a685f22a55bc94f6bd95c2423 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Tue, 1 Sep 2026 18:02:51 -0400 Subject: [PATCH 7/8] fix(agent): match EOF transport aborts with word boundaries --- internal/agent/midstream_retry_test.go | 55 ++++++++++++++++++++++++++ internal/agent/reconnect.go | 26 +++++++++++- 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go index 7046d658a..f14c1a7be 100644 --- a/internal/agent/midstream_retry_test.go +++ b/internal/agent/midstream_retry_test.go @@ -362,6 +362,8 @@ func TestIsMidStreamTransportAbort(t *testing.T) { "dial tcp 10.0.0.1:443: connect: connection refused", "i/o timeout", "504 Gateway Timeout", + "provider request error: request does not satisfy oneOf schema", + "schema error: property oneOf is invalid", } for _, m := range notAborts { if isMidStreamTransportAbort(m) { @@ -369,3 +371,56 @@ func TestIsMidStreamTransportAbort(t *testing.T) { } } } + +func TestRunDoesNotRetryApplicationErrorContainingEOFSubstring(t *testing.T) { + p := &midStreamAbortProvider{ + abortBefore: 1, + abortError: "provider request error: request does not satisfy oneOf schema", + } + var notices string + opts := Options{ + Registry: tools.NewRegistry(), + OnReasoning: func(s string) { notices += s }, + } + _, err := Run(context.Background(), "go", p, opts) + if err == nil { + t.Fatal("expected application error, got nil") + } + if got := atomic.LoadInt32(&p.calls); got != 1 { + t.Fatalf("application error with 'oneOf' must not retry (got %d calls, want 1)", got) + } + if notices != "" { + t.Fatalf("expected no reconnect notices, got %q", notices) + } + if !strings.Contains(err.Error(), "oneOf") { + t.Fatalf("expected original error preserved, got %v", err) + } +} + +func TestRunRetriesMidStreamUnexpectedEOF(t *testing.T) { + defer func(orig time.Duration) { streamReconnectBase = orig }(streamReconnectBase) + streamReconnectBase = time.Millisecond + p := &midStreamAbortProvider{ + abortBefore: 1, + abortError: "provider stream error: unexpected EOF", + } + var notices string + opts := Options{ + Registry: tools.NewRegistry(), + OnReasoning: func(s string) { notices += s }, + } + result, err := Run(context.Background(), "go", p, opts) + if err != nil { + t.Fatalf("unexpected EOF should retry to success, got %v", err) + } + if result.FinalAnswer != "done" { + t.Fatalf("final answer = %q, want %q", result.FinalAnswer, "done") + } + if got := atomic.LoadInt32(&p.calls); got != 2 { + t.Fatalf("want 2 calls (1 abort + 1 retry), got %d", got) + } + lower := strings.ToLower(notices) + if !strings.Contains(lower, "connection lost") || !strings.Contains(lower, "reconnecting") { + t.Fatalf("expected reconnect notice, got %q", notices) + } +} diff --git a/internal/agent/reconnect.go b/internal/agent/reconnect.go index ba472783e..a2c817a7a 100644 --- a/internal/agent/reconnect.go +++ b/internal/agent/reconnect.go @@ -165,7 +165,6 @@ func shouldReconnect(ctx context.Context, err error) bool { // connection was lost. Issue #973 names WSAECONNABORTED / connection reset / // unexpected stream EOF. var midStreamAbortNeedles = []string{ - "eof", "unexpected end", "connection reset", "broken pipe", @@ -192,6 +191,9 @@ func isMidStreamTransportAbort(message string) bool { if errhint.HasStatusCode(lowered, "500", "502", "503", "504") { return false } + if containsWordBoundary(lowered, "eof") { + return true + } for _, needle := range midStreamAbortNeedles { if strings.Contains(lowered, needle) { return true @@ -200,6 +202,28 @@ func isMidStreamTransportAbort(message string) bool { return false } +func containsWordBoundary(text, word string) bool { + start := 0 + for { + idx := strings.Index(text[start:], word) + if idx < 0 { + return false + } + pos := start + idx + endPos := pos + len(word) + leftBoundary := pos == 0 || !isAlphaNumByte(text[pos-1]) + rightBoundary := endPos == len(text) || !isAlphaNumByte(text[endPos]) + if leftBoundary && rightBoundary { + return true + } + start = pos + 1 + } +} + +func isAlphaNumByte(c byte) bool { + return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' +} + // backoffFor is the deterministic exponential base delay for a 1-based attempt, // capped at streamReconnectMax. Jitter is layered on separately (jitteredBackoff). func backoffFor(attempt int) time.Duration { From 33a6d2877e4f4035add27a169cae541b19a168b4 Mon Sep 17 00:00:00 2001 From: cairn-intern Date: Wed, 2 Sep 2026 04:36:32 -0400 Subject: [PATCH 8/8] fix(agent): reject classified provider errors before matching transport phrases --- internal/agent/midstream_retry_test.go | 28 ++++++++++++++++++++++++++ internal/agent/reconnect.go | 11 ++++++++++ 2 files changed, 39 insertions(+) diff --git a/internal/agent/midstream_retry_test.go b/internal/agent/midstream_retry_test.go index f14c1a7be..13f0839f6 100644 --- a/internal/agent/midstream_retry_test.go +++ b/internal/agent/midstream_retry_test.go @@ -363,6 +363,9 @@ func TestIsMidStreamTransportAbort(t *testing.T) { "i/o timeout", "504 Gateway Timeout", "provider request error: request does not satisfy oneOf schema", + "provider request error: connection closed is not a supported finish reason", + "auth error: connection reset by peer", + "rate limit error: server closed", "schema error: property oneOf is invalid", } for _, m := range notAborts { @@ -372,6 +375,31 @@ func TestIsMidStreamTransportAbort(t *testing.T) { } } +func TestRunDoesNotRetryClassifiedProviderErrorWithSocketPhrase(t *testing.T) { + p := &midStreamAbortProvider{ + abortBefore: 1, + abortError: "provider request error: connection closed is not a supported finish reason", + } + var notices string + opts := Options{ + Registry: tools.NewRegistry(), + OnReasoning: func(s string) { notices += s }, + } + _, err := Run(context.Background(), "go", p, opts) + if err == nil { + t.Fatal("expected application error, got nil") + } + if got := atomic.LoadInt32(&p.calls); got != 1 { + t.Fatalf("classified provider error with 'connection closed' must not retry (got %d calls, want 1)", got) + } + if notices != "" { + t.Fatalf("expected no reconnect notices, got %q", notices) + } + if !strings.Contains(err.Error(), "connection closed") { + t.Fatalf("expected original error preserved, got %v", err) + } +} + func TestRunDoesNotRetryApplicationErrorContainingEOFSubstring(t *testing.T) { p := &midStreamAbortProvider{ abortBefore: 1, diff --git a/internal/agent/reconnect.go b/internal/agent/reconnect.go index a2c817a7a..bf7897c50 100644 --- a/internal/agent/reconnect.go +++ b/internal/agent/reconnect.go @@ -175,6 +175,12 @@ var midStreamAbortNeedles = []string{ "forcibly closed", } +var classifiedNonTransportPrefixes = []string{ + "provider request error:", + "auth error:", + "rate limit error:", +} + // isMidStreamTransportAbort reports whether a collected stream error string is a // retryable mid-stream transport abort. It does NOT delegate to shouldReconnect: // that list is a connect-phase argument ("no response was received"). The stream @@ -188,6 +194,11 @@ func isMidStreamTransportAbort(message string) bool { if lowered == "" || isContextLimitError(lowered) { return false } + for _, prefix := range classifiedNonTransportPrefixes { + if strings.Contains(lowered, prefix) { + return false + } + } if errhint.HasStatusCode(lowered, "500", "502", "503", "504") { return false }