From 92ab65aa5fc295191a378956e7ca20d0700ce435 Mon Sep 17 00:00:00 2001 From: Gnanam Date: Thu, 2 Jul 2026 19:54:31 +0530 Subject: [PATCH 1/4] fix(agent): retry provider-empty streams and name the fault when they persist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A backend can answer a completion request with HTTP 200 and an SSE stream that carries nothing at all — no text, no tool calls, no reasoning. Observed live on the ollama cloud relay under load (200 in ~1s, straight to [DONE]; healthy calls take 11s+): three such responses arrived within two seconds, the no-output guard struck out, and the run died with the generic "Agent stopped after 3 turns with no output" — no error, no hint the PROVIDER failed. On the reporting machine 577 of 1034 sessions (56%) ended exactly this way, and the same silent handling produced multi-minute "Working · thinking" hangs in the TUI plan flow. Distinguish that PROVIDER-empty shape from a behavioral empty turn (the model streamed reasoning but committed no answer — its trace proves the backend worked) via a new CollectedStream.ReasoningEmitted signal: - providerEmptyStream turns are re-issued in-turn with backoff (bounded by maxEmptyStreamRetries, mirroring the stall-retry pattern) before they may count as a no-output strike — the condition is transient backend state and usually recovers. Nothing from the empty stream was forwarded or committed to history, so the retry is clean. A user-visible notice ("provider returned an empty response — retrying n/m") rides the reasoning channel like the stall/reconnect notices. - When every guard strike was provider-empty, the stop answer names the backend fault and what to do about it, instead of reading as the agent giving up. Mixed or behavioral strikes keep the existing message, and IsNoProgressStop recognizes the new answer so session titling and resume filtering treat it like the other guard stops. Tests: provider-empty strikes retry 3× each and stop with the provider message (recognized by IsNoProgressStop, distinct from the generic marker); a transient empty recovers on the in-turn retry with no extra turn counted; mixed provider/behavioral strikes keep the generic message; the existing counter-reset and guard tests now drive BEHAVIORAL empties (reasoning-only turns) and are byte-identical in outcome. --- internal/agent/guardrails.go | 56 ++++++++++++++ internal/agent/guardrails_test.go | 122 +++++++++++++++++++++++++++--- internal/agent/loop.go | 46 +++++++++++ internal/agent/reconnect.go | 16 ++++ internal/zeroruntime/helpers.go | 10 +++ 5 files changed, 239 insertions(+), 11 deletions(-) diff --git a/internal/agent/guardrails.go b/internal/agent/guardrails.go index 089dc818e..facb3cef9 100644 --- a/internal/agent/guardrails.go +++ b/internal/agent/guardrails.go @@ -284,12 +284,48 @@ const ( noOutputStopSuffix = "to avoid consuming tokens without making progress." ) +// providerEmptyStream reports that a cleanly-completed turn carried NOTHING — +// no visible text, no tool calls (real or dropped), and no reasoning signal. +// That shape is a transport/backend fault (e.g. the ollama cloud relay +// answering 200 with an instantly-empty SSE under load), not model behavior: +// a model that is generating always leaves SOME trace (text, a tool call +// attempt, or reasoning deltas). +func providerEmptyStream(collected zeroruntime.CollectedStream) bool { + return strings.TrimSpace(collected.Text) == "" && + len(collected.ToolCalls) == 0 && + collected.DroppedToolCalls == 0 && + len(collected.ReasoningBlocks) == 0 && + !collected.ReasoningEmitted +} + +// allEmptyTurnsProviderEmpty reports that every strike toward the no-output +// guard was a provider-empty stream, so the stop message should name the +// backend fault rather than generic agent no-progress. +func (state *guardState) allEmptyTurnsProviderEmpty() bool { + return state.emptyTurns > 0 && state.providerEmptyTurns == state.emptyTurns +} + // noOutputStopAnswer is the final answer returned when the no-output guard // stops the run. The turn count is interpolated at the call site. func noOutputStopAnswer(turns int) string { return noOutputStopPrefix + strconv.Itoa(turns) + " turns " + noOutputStopMarker + " " + noOutputStopSuffix } +// providerEmptyStopAnswer is the final answer when every guard strike was a +// provider-empty stream: the backend repeatedly answered with contentless +// completions (already retried with backoff inside each turn). It tells the +// user the truth — the provider failed, not the agent — and what to do next. +func providerEmptyStopAnswer(turns int) string { + return providerEmptyStopPrefix + + strconv.Itoa(turns) + " consecutive attempts, each already retried with backoff. " + + "The backend is likely rate-limiting or degraded right now — try again shortly, or switch providers/models (zero providers use )." +} + +// providerEmptyStopPrefix is the stable head of providerEmptyStopAnswer, used +// by IsNoProgressStop so session titling / resume filtering treat the +// provider-empty stop like the other guard answers (not real content). +const providerEmptyStopPrefix = "The model provider returned an empty response (no text, no tool calls, no reasoning) on " + // IsNoProgressStop reports whether content IS the no-output guardrail stop answer // (a run that produced no visible text and no tool calls). It matches the EXACT // structure noOutputStopAnswer emits — prefix + " turns " + marker + " " + @@ -300,6 +336,15 @@ func noOutputStopAnswer(turns int) string { // and skip its title generation. func IsNoProgressStop(content string) bool { trimmed := strings.TrimSpace(content) + // The provider-empty variant: stable prefix + a bare integer count. + if rest, ok := strings.CutPrefix(trimmed, providerEmptyStopPrefix); ok { + if sep := strings.Index(rest, " consecutive attempts"); sep > 0 { + if _, err := strconv.Atoi(rest[:sep]); err == nil { + return true + } + } + return false + } if !strings.HasPrefix(trimmed, noOutputStopPrefix) { return false } @@ -352,6 +397,7 @@ func toolOnlyProgressReminder(turns int) string { // purely from tool-call names and per-turn output, matching what the loop holds. type guardState struct { emptyTurns int + providerEmptyTurns int totalToolCalls int toolCallsSincePlanUpdate int planEverCalled bool @@ -417,8 +463,18 @@ func (state *guardState) observeTurn(collected zeroruntime.CollectedStream) (sto if hasToolCalls || hasVisibleText { state.emptyTurns = 0 + state.providerEmptyTurns = 0 } else { state.emptyTurns++ + // Track separately whether the empty turn was a PROVIDER-empty stream + // (nothing at all came back — not even reasoning) vs a behavioral empty + // (the model thought/streamed but committed no answer). When every + // strike was provider-empty, the stop message names the backend fault + // instead of blaming the agent — the difference between the user + // retrying/switching providers and filing a "the agent gives up" bug. + if providerEmptyStream(collected) { + state.providerEmptyTurns++ + } } if hasToolCalls && !hasVisibleText { state.toolOnlyTurns++ diff --git a/internal/agent/guardrails_test.go b/internal/agent/guardrails_test.go index fd7758df0..0401f24ca 100644 --- a/internal/agent/guardrails_test.go +++ b/internal/agent/guardrails_test.go @@ -9,11 +9,23 @@ import ( "github.com/Gitlawb/zero/internal/zeroruntime" ) -// emptyTurn is a stream that produces no visible text and no tool calls. +// emptyTurn is a PROVIDER-empty stream: it completes cleanly carrying nothing +// at all (the ollama-cloud 200+[DONE] shape). The loop retries these in-turn +// before counting a no-output strike. func emptyTurn() []zeroruntime.StreamEvent { return []zeroruntime.StreamEvent{{Type: zeroruntime.StreamEventDone}} } +// reasoningOnlyTurn is a BEHAVIORAL empty turn: the model streamed reasoning +// (so the provider clearly worked) but committed no text and no tool calls. +// These are NOT retried in-turn; they count directly toward the no-output guard. +func reasoningOnlyTurn() []zeroruntime.StreamEvent { + return []zeroruntime.StreamEvent{ + {Type: zeroruntime.StreamEventReasoning, Content: "thinking…"}, + {Type: zeroruntime.StreamEventDone}, + } +} + // textTurn produces a turn with visible assistant text. func textTurn(content string) []zeroruntime.StreamEvent { return []zeroruntime.StreamEvent{ @@ -45,9 +57,9 @@ func countUserMessagesContaining(messages []zeroruntime.Message, needle string) func TestRunStopsAfterConsecutiveEmptyTurns(t *testing.T) { provider := &mockProvider{ turns: [][]zeroruntime.StreamEvent{ - emptyTurn(), - emptyTurn(), - emptyTurn(), + reasoningOnlyTurn(), + reasoningOnlyTurn(), + reasoningOnlyTurn(), // A 4th turn exists but must never be requested. textTurn("should never reach here"), }, @@ -77,10 +89,10 @@ func TestRunStopsAfterConsecutiveEmptyTurns(t *testing.T) { func TestRunResetsEmptyTurnCounterOnVisibleOutput(t *testing.T) { provider := &mockProvider{ turns: [][]zeroruntime.StreamEvent{ - emptyTurn(), - emptyTurn(), + reasoningOnlyTurn(), + reasoningOnlyTurn(), textTurn("here is real progress"), // resets the counter and is the final answer - emptyTurn(), + reasoningOnlyTurn(), }, } @@ -109,11 +121,11 @@ func TestRunResetsEmptyTurnCounterOnToolCall(t *testing.T) { provider := &mockProvider{ turns: [][]zeroruntime.StreamEvent{ - emptyTurn(), - emptyTurn(), + reasoningOnlyTurn(), + reasoningOnlyTurn(), toolTurn("call-1", "read_file", `{"path":"notes.txt"}`), // resets counter - emptyTurn(), - emptyTurn(), + reasoningOnlyTurn(), + reasoningOnlyTurn(), textTurn("done"), }, } @@ -455,3 +467,91 @@ func TestRunInjectsToolFailureHintWithSchema(t *testing.T) { t.Fatalf("expected a tool-failure hint on the 3rd turn, messages: %+v", provider.requests[2].Messages) } } + +// A provider-empty stream (clean completion carrying nothing at all) is +// retried in-turn with backoff before it may count as a no-output strike; when +// the backend keeps answering empty, the run stops with a message that names +// the PROVIDER fault — not the generic "agent made no progress" text. This is +// the exact live failure observed on the ollama cloud relay (HTTP 200 in ~1s, +// SSE straight to [DONE]), which silently killed 56% of sessions on the +// reporting machine. +func TestRunRetriesProviderEmptyStreamsThenStopsWithProviderMessage(t *testing.T) { + turns := make([][]zeroruntime.StreamEvent, 0, 9) + for i := 0; i < 9; i++ { // 3 strikes × (1 initial + 2 in-turn retries) + turns = append(turns, emptyTurn()) + } + provider := &mockProvider{turns: turns} + + result, err := Run(context.Background(), "go", provider, Options{ + Registry: tools.NewRegistry(), + MaxTurns: 12, + }) + if err != nil { + t.Fatal(err) + } + wantRequests := maxEmptyTurns * (1 + maxEmptyStreamRetries) + if len(provider.requests) != wantRequests { + t.Fatalf("expected %d requests (%d strikes × %d attempts), got %d", + wantRequests, maxEmptyTurns, 1+maxEmptyStreamRetries, len(provider.requests)) + } + if !strings.Contains(result.FinalAnswer, "provider returned an empty response") { + t.Fatalf("expected the provider-empty stop message, got %q", result.FinalAnswer) + } + if strings.Contains(result.FinalAnswer, noOutputStopMarker) { + t.Fatalf("provider fault must not be reported as generic agent no-progress: %q", result.FinalAnswer) + } + if !IsNoProgressStop(result.FinalAnswer) { + t.Fatal("the provider-empty stop answer must be recognized by IsNoProgressStop (titling/resume filters)") + } +} + +// A transient empty response recovers on the in-turn retry: no strike, no stop, +// the retried turn's real output is the answer. +func TestRunRecoversWhenEmptyStreamRetrySucceeds(t *testing.T) { + provider := &mockProvider{ + turns: [][]zeroruntime.StreamEvent{ + emptyTurn(), // initial attempt: backend hiccup + textTurn("recovered"), // in-turn retry gets the real answer + }, + } + + result, err := Run(context.Background(), "go", provider, Options{ + Registry: tools.NewRegistry(), + MaxTurns: 12, + }) + if err != nil { + t.Fatal(err) + } + if len(provider.requests) != 2 { + t.Fatalf("expected 2 requests (initial + one retry), got %d", len(provider.requests)) + } + if result.FinalAnswer != "recovered" { + t.Fatalf("expected the retried turn's text as final answer, got %q", result.FinalAnswer) + } + if result.Turns != 1 { + t.Fatalf("an in-turn retry is the SAME turn, want 1 turn, got %d", result.Turns) + } +} + +// Mixed strikes (provider-empty + behavioral reasoning-only) must NOT claim a +// provider fault — the generic no-output message stays. +func TestRunMixedEmptyTurnsKeepGenericStopMessage(t *testing.T) { + provider := &mockProvider{ + turns: [][]zeroruntime.StreamEvent{ + reasoningOnlyTurn(), // strike 1: behavioral + emptyTurn(), emptyTurn(), emptyTurn(), // strike 2: provider-empty ×(1+2 retries) + reasoningOnlyTurn(), // strike 3: behavioral + }, + } + + result, err := Run(context.Background(), "go", provider, Options{ + Registry: tools.NewRegistry(), + MaxTurns: 12, + }) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(result.FinalAnswer, noOutputStopMarker) { + t.Fatalf("mixed strikes must use the generic no-output message, got %q", result.FinalAnswer) + } +} diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 29c99a7d3..38f13c86c 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -27,6 +27,14 @@ const maxTurnsFinalAnswerPrompt = "You have reached the tool-turn limit. Do not // safe recovery for a stalled/dead pooled connection. const maxStreamStallRetries = 2 +// maxEmptyStreamRetries caps in-turn re-issues of a request whose stream +// completed cleanly but carried nothing at all (providerEmptyStream). Observed +// live on the ollama cloud relay under load: HTTP 200 in ~1s with an SSE that +// goes straight to [DONE]. The condition is transient backend state, so a +// short backoff-retry usually recovers; without it three such responses +// arrived inside two seconds and killed the run via the no-output guard. +const maxEmptyStreamRetries = 2 + const ( toolResultMetaControl = "control" toolResultControlSpecReview = "spec_review_required" @@ -324,6 +332,35 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) } collected = zeroruntime.CollectStreamWithOptions(ctx, retryStream, forwardingOpts) } + // Provider-empty retry: the stream completed CLEANLY but carried nothing + // — no text, no tool calls, no reasoning. That is not model behavior (a + // generating model always leaves some trace); it is a backend answering + // with a contentless completion (seen on the ollama cloud relay under + // load: 200 + instant [DONE]). Transient, so re-issue with backoff before + // letting it count as a no-output strike. Nothing from the empty stream + // was forwarded or committed to history, so the retry is clean. + for attempt := 1; attempt <= maxEmptyStreamRetries && + collected.Error == "" && !forwardedVisibleText && + providerEmptyStream(collected); attempt++ { + if notify := emptyRetryNoticeFor(options); notify != nil { + notify(attempt, maxEmptyStreamRetries) + } + if err := sleepWithContext(ctx, backoffFor(attempt)); err != nil { + result.Messages = copyMessages(messages) + return result, err + } + retryRequest := zeroruntime.CompletionRequest{ + Messages: copyMessages(messages), + Tools: exposed, + ReasoningEffort: options.ReasoningEffort, + } + retryStream, retryErr := streamWithReconnect(ctx, provider, retryRequest, reconnectNoticeFor(options)) + if retryErr != nil { + result.Messages = copyMessages(messages) + return result, retryErr + } + collected = zeroruntime.CollectStreamWithOptions(ctx, retryStream, forwardingOpts) + } 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) @@ -370,6 +407,15 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) // A truly-empty turn (no text, no tool calls, no dropped calls) is // counted toward the runaway cap so we stop before burning maxTurns. if guards.observeTurn(collected) { + // Every strike being a provider-empty stream means the BACKEND + // failed repeatedly (each attempt already retried with backoff + // above) — say so, instead of the generic no-progress message + // that reads as the agent giving up. + if guards.allEmptyTurnsProviderEmpty() { + result.FinalAnswer = providerEmptyStopAnswer(result.Turns) + result.Messages = copyMessages(messages) + return result, nil + } result.FinalAnswer = noOutputStopAnswer(result.Turns) result.Messages = copyMessages(messages) return result, nil diff --git a/internal/agent/reconnect.go b/internal/agent/reconnect.go index 641366a95..b0a3053a5 100644 --- a/internal/agent/reconnect.go +++ b/internal/agent/reconnect.go @@ -55,6 +55,22 @@ func stallRetryNoticeFor(options Options) reconnectNotifier { } } +// emptyRetryNoticeFor builds a notifier for the loop's provider-empty retry +// (a stream that completed cleanly but carried NOTHING — no text, no tool +// calls, no reasoning). Distinct wording from the stall notice because nothing +// timed out: the backend answered instantly with an empty completion (observed +// on the ollama cloud relay under load, and possible on any gateway having a +// bad moment). Surfaced through OnReasoning, the non-content channel. Nil when +// there is no reasoning sink (the retry still happens silently). +func emptyRetryNoticeFor(options Options) reconnectNotifier { + if options.OnReasoning == nil { + return nil + } + return func(attempt, max int) { + options.OnReasoning(fmt.Sprintf("\n[provider returned an empty response — retrying %d/%d…]\n", attempt, max)) + } +} + // streamWithReconnect issues request via provider.StreamCompletion and, on a // transient disconnect error, retries the connect up to maxStreamReconnects // times with exponential backoff. It returns the live stream on success, or the diff --git a/internal/zeroruntime/helpers.go b/internal/zeroruntime/helpers.go index 01e9188a1..3794c06ad 100644 --- a/internal/zeroruntime/helpers.go +++ b/internal/zeroruntime/helpers.go @@ -21,6 +21,12 @@ type CollectedStream struct { // thinking blocks) that must be replayed on the next turn. Empty for providers // or runs without extended thinking. ReasoningBlocks []ReasoningBlock + // ReasoningEmitted reports that the stream carried ANY reasoning signal + // (live reasoning deltas or preserved blocks). It distinguishes "the model + // thought but produced no answer" (a behavioral empty turn) from "the + // provider returned a contentless stream" (a transport/backend fault worth + // retrying) — the two need different handling in the agent loop. + ReasoningEmitted bool } // Truncated reports whether the response ended for a non-normal reason (the @@ -120,6 +126,7 @@ func CollectStreamWithOptions(ctx context.Context, events <-chan StreamEvent, op // accumulate them regardless of type so they survive for replay. if len(event.ReasoningBlocks) > 0 { collected.ReasoningBlocks = append(collected.ReasoningBlocks, event.ReasoningBlocks...) + collected.ReasoningEmitted = true } switch event.Type { @@ -132,6 +139,9 @@ func CollectStreamWithOptions(ctx context.Context, events <-chan StreamEvent, op options.OnText(event.Content) } case StreamEventReasoning: + if event.Content != "" { + collected.ReasoningEmitted = true + } if options.OnReasoning != nil { options.OnReasoning(event.Content) } From 48318dfb57271bf047e4fef541fe14e8d0c5352d Mon Sep 17 00:00:00 2001 From: Gnanam Date: Thu, 2 Jul 2026 19:56:07 +0530 Subject: [PATCH 2/4] fix(openai): parse the `reasoning` delta field alongside `reasoning_content` Reasoning/thinking deltas arrive under different keys depending on the backend dialect: DeepSeek-style servers emit `reasoning_content` (already handled), while ollama's OpenAI-compat endpoint, OpenRouter, and most gateways emit `reasoning`. Zero only parsed the former, so on those backends every thinking token was silently dropped: live capture against ollama glm-5.2:cloud showed 28-227 reasoning deltas per turn discarded even in successful runs. The model looked frozen for minutes behind a dead "thinking" spinner, the discarded thinking was never replayed into the next turn's context, and a reasoning-only turn was indistinguishable from a dead provider (it now feeds CollectedStream.ReasoningEmitted, which the agent loop uses to tell backend faults from behavioral empty turns). Emit StreamEventReasoning from whichever key is present; no backend sends both. The regression test replays the exact delta shape captured live from ollama. --- internal/providers/openai/provider.go | 16 +++++++++++-- internal/providers/openai/provider_test.go | 27 ++++++++++++++++++++++ internal/providers/openai/types.go | 10 +++++++- 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/internal/providers/openai/provider.go b/internal/providers/openai/provider.go index 49de6a0a4..2bd1d7d27 100644 --- a/internal/providers/openai/provider.go +++ b/internal/providers/openai/provider.go @@ -276,6 +276,16 @@ func (provider *Provider) emitPayload(ctx context.Context, data string, state *t return true } +// firstNonEmptyString returns the first argument that is not empty. +func firstNonEmptyString(values ...string) string { + for _, value := range values { + if value != "" { + return value + } + } + return "" +} + func (provider *Provider) emitChunk( ctx context.Context, chunk streamChunk, @@ -283,10 +293,12 @@ func (provider *Provider) emitChunk( events chan<- zeroruntime.StreamEvent, ) { for _, choice := range chunk.Choices { - if choice.Delta.ReasoningContent != "" { + // Two dialects for the same thing (see streamDelta): prefer + // `reasoning_content`, fall back to `reasoning`. No backend sends both. + if reasoning := firstNonEmptyString(choice.Delta.ReasoningContent, choice.Delta.Reasoning); reasoning != "" { sendEvent(ctx, events, zeroruntime.StreamEvent{ Type: zeroruntime.StreamEventReasoning, - Content: choice.Delta.ReasoningContent, + Content: reasoning, }) } if choice.Delta.Content != "" { diff --git a/internal/providers/openai/provider_test.go b/internal/providers/openai/provider_test.go index 69d7e9027..01cc87b40 100644 --- a/internal/providers/openai/provider_test.go +++ b/internal/providers/openai/provider_test.go @@ -227,6 +227,33 @@ func TestStreamCompletionEmitsReasoningContentDeltas(t *testing.T) { } } +// Ollama's OpenAI-compat endpoint, OpenRouter, and most gateways stream +// thinking under `reasoning` (not DeepSeek's `reasoning_content`). Dropping it +// made every thinking token on those backends vanish: the model looked frozen +// for minutes and reasoning-only turns were indistinguishable from a dead +// provider. The exact shape below is a live capture from ollama glm-5.2:cloud. +func TestStreamCompletionEmitsOllamaStyleReasoningDeltas(t *testing.T) { + provider := newTestProvider(t, func(w http.ResponseWriter, r *http.Request) { + writeSSE(w, `{"choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning":"The user"},"finish_reason":null}]}`) + writeSSE(w, `{"choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning":" wants hello"},"finish_reason":null}]}`) + writeSSE(w, `{"choices":[{"index":0,"delta":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`) + writeSSE(w, `[DONE]`) + }) + + events := collectProviderEvents(t, provider) + reasoning := eventsOfType(events, zeroruntime.StreamEventReasoning) + if len(reasoning) != 2 { + t.Fatalf("reasoning events = %#v, want two reasoning deltas (ollama `reasoning` field regression)", reasoning) + } + if reasoning[0].Content != "The user" || reasoning[1].Content != " wants hello" { + t.Fatalf("unexpected reasoning events: %#v", reasoning) + } + text := eventsOfType(events, zeroruntime.StreamEventText) + if len(text) != 1 || text[0].Content != "hello" { + t.Fatalf("content after thinking must still stream as text, got %#v", text) + } +} + func TestStreamCompletionEmitsReasoningBeforeRegularContent(t *testing.T) { provider := newTestProvider(t, func(w http.ResponseWriter, r *http.Request) { writeSSE(w, `{"choices":[{"delta":{"reasoning_content":"Thinking. ","content":"Answer."}}]}`) diff --git a/internal/providers/openai/types.go b/internal/providers/openai/types.go index f093ce30b..8be080f21 100644 --- a/internal/providers/openai/types.go +++ b/internal/providers/openai/types.go @@ -76,8 +76,16 @@ type streamChoice struct { } type streamDelta struct { - Content string `json:"content"` + Content string `json:"content"` + // Reasoning/thinking deltas arrive under DIFFERENT keys depending on the + // backend dialect: DeepSeek-style servers emit `reasoning_content`, while + // ollama's OpenAI-compat endpoint, OpenRouter, and most gateways emit + // `reasoning`. Parse both — dropping `reasoning` made every thinking token + // on those backends silently vanish (the model looked frozen for minutes, + // its thinking was never replayed into the next turn's context, and a + // reasoning-only turn was indistinguishable from a dead provider). ReasoningContent string `json:"reasoning_content"` + Reasoning string `json:"reasoning"` ToolCalls []streamToolCallDelta `json:"tool_calls"` } From ae43d92d2a663592ecd7bfc6eca904ad84a62bbb Mon Sep 17 00:00:00 2001 From: Gnanam Date: Thu, 2 Jul 2026 19:58:56 +0530 Subject: [PATCH 3/4] feat(agent): carry wire-level diagnostics in the provider-empty stop answer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Append the last empty attempt's response shape (finish_reason, output/reasoning token counts) to the provider-empty stop message. This failure class previously took a live traffic capture to diagnose — the sessions recorded only the generic guard text, with nothing about what the backend actually returned. Now the session log itself says e.g. "[last attempt: finish_reason=stop, output_tokens=0, reasoning_tokens=0]", which distinguishes an instantly-empty relay response from a truncated or filtered one at a glance. --- internal/agent/guardrails.go | 24 ++++++++++++++++++++++-- internal/agent/guardrails_test.go | 5 +++++ internal/agent/loop.go | 2 +- 3 files changed, 28 insertions(+), 3 deletions(-) diff --git a/internal/agent/guardrails.go b/internal/agent/guardrails.go index facb3cef9..5657aa3c4 100644 --- a/internal/agent/guardrails.go +++ b/internal/agent/guardrails.go @@ -2,6 +2,7 @@ package agent import ( "encoding/json" + "fmt" "strconv" "strings" @@ -315,10 +316,29 @@ func noOutputStopAnswer(turns int) string { // provider-empty stream: the backend repeatedly answered with contentless // completions (already retried with backoff inside each turn). It tells the // user the truth — the provider failed, not the agent — and what to do next. -func providerEmptyStopAnswer(turns int) string { - return providerEmptyStopPrefix + +// detail carries wire-level diagnostics from the last empty attempt +// (finish reason, token counts) so the condition is debuggable straight from +// the session log — this exact failure class previously took a live traffic +// capture to diagnose because nothing about the response shape was recorded. +func providerEmptyStopAnswer(turns int, detail string) string { + answer := providerEmptyStopPrefix + strconv.Itoa(turns) + " consecutive attempts, each already retried with backoff. " + "The backend is likely rate-limiting or degraded right now — try again shortly, or switch providers/models (zero providers use )." + if detail != "" { + answer += " [last attempt: " + detail + "]" + } + return answer +} + +// emptyStreamDetail renders the wire-level shape of an empty attempt for +// providerEmptyStopAnswer. +func emptyStreamDetail(collected zeroruntime.CollectedStream) string { + finish := collected.FinishReason + if finish == "" { + finish = "stop" + } + return fmt.Sprintf("finish_reason=%s, output_tokens=%d, reasoning_tokens=%d", + finish, collected.Usage.OutputTokens, collected.Usage.ReasoningTokens) } // providerEmptyStopPrefix is the stable head of providerEmptyStopAnswer, used diff --git a/internal/agent/guardrails_test.go b/internal/agent/guardrails_test.go index 0401f24ca..e8e564fe4 100644 --- a/internal/agent/guardrails_test.go +++ b/internal/agent/guardrails_test.go @@ -503,6 +503,11 @@ func TestRunRetriesProviderEmptyStreamsThenStopsWithProviderMessage(t *testing.T if !IsNoProgressStop(result.FinalAnswer) { t.Fatal("the provider-empty stop answer must be recognized by IsNoProgressStop (titling/resume filters)") } + // Wire-level diagnostics must land in the message (and therefore the + // session log) so this failure class is debuggable without a traffic capture. + if !strings.Contains(result.FinalAnswer, "output_tokens=0") || !strings.Contains(result.FinalAnswer, "finish_reason=") { + t.Fatalf("stop answer must carry last-attempt diagnostics, got %q", result.FinalAnswer) + } } // A transient empty response recovers on the in-turn retry: no strike, no stop, diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 38f13c86c..fe221455b 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -412,7 +412,7 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options) // above) — say so, instead of the generic no-progress message // that reads as the agent giving up. if guards.allEmptyTurnsProviderEmpty() { - result.FinalAnswer = providerEmptyStopAnswer(result.Turns) + result.FinalAnswer = providerEmptyStopAnswer(result.Turns, emptyStreamDetail(collected)) result.Messages = copyMessages(messages) return result, nil } From 132788fd3626dd190fe2fef9d34de97e00201f38 Mon Sep 17 00:00:00 2001 From: Gnanam Date: Thu, 2 Jul 2026 20:16:55 +0530 Subject: [PATCH 4/4] fix(providerio): detect a mid-stream dead connection at a tighter gap timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live capture of the reported "stuck for 33 minutes": the model call streamed ~119KB, then the socket went byte-frozen forever while the TUI showed a dead "Working · thinking" spinner. The only bound on that shape was the full 5-minute idle window — per attempt — so one turn could blind- wait 15+ minutes across its stall retries before surfacing anything. Split the watchdog window by stream phase: - BEFORE the first payload the full idle window still applies (slow cloud proxies legitimately withhold output until the upstream model produces its first token — the reason the idle default is 5m). - AFTER any payload has arrived the backend has proven it streams and heartbeats, so the re-arm window tightens to the gap timeout (default 2m, ZERO_STREAM_GAP_TIMEOUT, clamped to the idle window, disabled with it). A byte-frozen socket mid-stream is a dead connection; waiting the full idle for it just multiplies the user's blind wait per retry. Keep-alives reset the gap like they reset idle, so a slow-but-alive stream is never cut — this only fires on true byte silence. Raise the in-turn stall retries from 2 to 3 (4 attempts total): Codex ships stream_max_retries=5 for the same failure class, and opencode's unbounded session retry has looped forever on it — 4 hard-capped attempts with the faster detection bounds a typical mid-stream death at roughly six minutes to recovery-or-honest-error instead of fifteen-plus blind. All providers benefit (openai/anthropic/gemini share the SSE scanner). Tests: gap applies only after the first payload (frozen-after-first-byte aborts at the tightened window; silent-from-the-start still gets the full idle), resolver default/clamp/env/off semantics, and the existing idle, content-stall, and cancel watchdog tests unchanged. --- internal/agent/loop.go | 7 +- internal/providers/providerio/providerio.go | 56 ++++++++++++- .../providers/providerio/providerio_test.go | 83 +++++++++++++++++++ 3 files changed, 144 insertions(+), 2 deletions(-) diff --git a/internal/agent/loop.go b/internal/agent/loop.go index fe221455b..b969993cf 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -25,7 +25,12 @@ const maxTurnsFinalAnswerPrompt = "You have reached the tool-turn limit. Do not // 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. -const maxStreamStallRetries = 2 +// maxStreamStallRetries: 3 in-turn re-issues (4 attempts total) approaches +// Codex's stream_max_retries=5 posture while staying hard-capped (opencode's +// unbounded session retry has produced infinite loops on this same failure). +// Affordable now that a mid-stream death is detected at the gap timeout +// (~2m) instead of the full 5m idle window. +const maxStreamStallRetries = 3 // maxEmptyStreamRetries caps in-turn re-issues of a request whose stream // completed cleanly but carried nothing at all (providerEmptyStream). Observed diff --git a/internal/providers/providerio/providerio.go b/internal/providers/providerio/providerio.go index d3199876d..6d97b830b 100644 --- a/internal/providers/providerio/providerio.go +++ b/internal/providers/providerio/providerio.go @@ -82,6 +82,53 @@ func ContentStallTimeout(idleTimeout time.Duration) time.Duration { return idleTimeout + extra } +// DefaultStreamGapTimeout bounds the byte-silent gap AFTER a stream has +// started delivering payloads. Once ANY payload (data or keep-alive) has +// arrived, the backend has proven it heartbeats/streams — so a subsequent gap +// with zero bytes is almost always a dead connection (live capture of the +// reported "stuck for 33 minutes": ~119KB streamed, then the socket went +// byte-frozen forever). Detecting that at 2 minutes instead of the 5-minute +// idle default turns a ~15-minute blind wait (3 × 5m attempts) into a ~6-minute +// bounded recovery, without touching the pre-first-byte window that slow cloud +// proxies legitimately need (they may withhold output until the upstream model +// produces its first token). Override with ZERO_STREAM_GAP_TIMEOUT. For +// comparison: Codex ships the same 5m idle with 5 stream retries; opencode +// suffers this exact freeze (idle HTTPS socket in "working" state) with no +// mid-stream gap detection at all. +const DefaultStreamGapTimeout = 2 * time.Minute + +// streamGapTimeoutEnv overrides the post-first-payload gap timeout. Accepts the +// same forms as ZERO_STREAM_IDLE_TIMEOUT; "0"/"off" disables the tightened gap +// (the plain idle timeout then applies for the whole stream). +const streamGapTimeoutEnv = "ZERO_STREAM_GAP_TIMEOUT" + +// ResolveStreamGapTimeout selects the effective post-first-payload gap timeout +// for a stream whose idle timeout resolved to idleTimeout. It is clamped to +// never EXCEED the idle timeout (a gap larger than idle would be dead config), +// and disabled entirely when the idle watchdog is disabled. +func ResolveStreamGapTimeout(idleTimeout time.Duration) time.Duration { + if idleTimeout <= 0 { + return 0 + } + gap := DefaultStreamGapTimeout + if raw := strings.TrimSpace(os.Getenv(streamGapTimeoutEnv)); raw != "" { + switch strings.ToLower(raw) { + case "0", "off", "none", "disabled": + return idleTimeout // no tightened gap: idle applies throughout + default: + if d, err := time.ParseDuration(raw); err == nil && d > 0 { + gap = d + } else if secs, err := strconv.Atoi(raw); err == nil && secs > 0 { + gap = time.Duration(secs) * time.Second + } + } + } + if gap > idleTimeout { + return idleTimeout + } + return gap +} + // streamIdleTimeoutEnv is the global override for the stream idle timeout. It // accepts a Go duration ("5m", "300s", "90s") or a bare number of seconds // ("300"). A value of "0", "off", "none", or "disabled" turns the watchdog off @@ -317,7 +364,14 @@ func ScanSSEDataWithContext( idle := time.NewTimer(idleTimeout) defer idle.Stop() idleC = idle.C - resetIdle = reset(idle, idleTimeout) + // The timer was armed above with the FULL idle window, which governs + // until the first payload (slow cloud proxies may withhold output until + // the upstream model produces its first token). resetIdle only runs when + // a payload HAS arrived — the backend has proven it streams/heartbeats — + // so every re-arm uses the tighter gap window: a byte-frozen socket + // mid-stream is a dead connection, and waiting the full idle for it just + // multiplies the user's blind wait per retry. + resetIdle = reset(idle, ResolveStreamGapTimeout(idleTimeout)) // Content watchdog: only real data lines reset it (keep-alives do not), so a // stream that heartbeats without producing output is bounded instead of diff --git a/internal/providers/providerio/providerio_test.go b/internal/providers/providerio/providerio_test.go index 33f539a3f..1a029576f 100644 --- a/internal/providers/providerio/providerio_test.go +++ b/internal/providers/providerio/providerio_test.go @@ -371,3 +371,86 @@ func TestHTTPClientReturnsStallHardenedSharedClient(t *testing.T) { t.Fatal("an explicit client must be returned unchanged") } } + +// Once a stream has delivered its first payload, a byte-silent gap is bounded +// by the TIGHTER gap timeout, not the full idle window — a mid-stream frozen +// socket (live-captured "stuck" shape: bytes flowed, then nothing forever) +// must be detected fast so retries aren't multiplied by five-minute waits. +func TestScanSSETightensGapAfterFirstPayload(t *testing.T) { + t.Setenv("ZERO_STREAM_GAP_TIMEOUT", "80ms") + pr, pw := io.Pipe() + defer func() { _ = pw.Close() }() + + go func() { + _, _ = io.WriteString(pw, "data: first\n\n") + // then: byte-frozen forever + }() + + start := time.Now() + done := make(chan error, 1) + go func() { + done <- ScanSSEDataWithContext(context.Background(), func() {}, pr, 2*time.Second, func(string) bool { return true }) + }() + select { + case err := <-done: + if !errors.Is(err, ErrStreamIdle) { + t.Fatalf("err = %v, want ErrStreamIdle", err) + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("gap abort took %s — the tightened window (80ms) did not apply after the first payload", elapsed) + } + case <-time.After(5 * time.Second): + t.Fatal("stream not aborted") + } +} + +// BEFORE the first payload the FULL idle window applies: slow cloud proxies may +// withhold output until the upstream model emits its first token, so the gap +// timeout must NOT govern the wait for the first byte. +func TestScanSSEKeepsFullIdleBeforeFirstPayload(t *testing.T) { + t.Setenv("ZERO_STREAM_GAP_TIMEOUT", "30ms") + pr, pw := io.Pipe() + defer func() { _ = pw.Close() }() + // never write anything + + start := time.Now() + done := make(chan error, 1) + go func() { + done <- ScanSSEDataWithContext(context.Background(), func() {}, pr, 300*time.Millisecond, func(string) bool { return true }) + }() + select { + case err := <-done: + if !errors.Is(err, ErrStreamIdle) { + t.Fatalf("err = %v, want ErrStreamIdle", err) + } + if elapsed := time.Since(start); elapsed < 250*time.Millisecond { + t.Fatalf("aborted after %s — the 30ms gap timeout must not apply before the first payload", elapsed) + } + case <-time.After(5 * time.Second): + t.Fatal("stream not aborted") + } +} + +func TestResolveStreamGapTimeout(t *testing.T) { + t.Setenv("ZERO_STREAM_GAP_TIMEOUT", "") + if got := ResolveStreamGapTimeout(5 * time.Minute); got != DefaultStreamGapTimeout { + t.Fatalf("default gap = %s, want %s", got, DefaultStreamGapTimeout) + } + // Clamped: the gap never exceeds the idle window. + if got := ResolveStreamGapTimeout(time.Minute); got != time.Minute { + t.Fatalf("gap = %s, want clamp to the 1m idle", got) + } + // Disabled idle disables the gap too. + if got := ResolveStreamGapTimeout(0); got != 0 { + t.Fatalf("gap with idle off = %s, want 0", got) + } + // Env override wins; "off" falls back to the idle window for the whole stream. + t.Setenv("ZERO_STREAM_GAP_TIMEOUT", "45s") + if got := ResolveStreamGapTimeout(5 * time.Minute); got != 45*time.Second { + t.Fatalf("env gap = %s, want 45s", got) + } + t.Setenv("ZERO_STREAM_GAP_TIMEOUT", "off") + if got := ResolveStreamGapTimeout(5 * time.Minute); got != 5*time.Minute { + t.Fatalf("off gap = %s, want the idle window", got) + } +}