diff --git a/internal/agent/guardrails.go b/internal/agent/guardrails.go index 089dc818e..5657aa3c4 100644 --- a/internal/agent/guardrails.go +++ b/internal/agent/guardrails.go @@ -2,6 +2,7 @@ package agent import ( "encoding/json" + "fmt" "strconv" "strings" @@ -284,12 +285,67 @@ 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. +// 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 +// 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 +356,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 +417,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 +483,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..e8e564fe4 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,96 @@ 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)") + } + // 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, +// 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..b969993cf 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -25,7 +25,20 @@ 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 +// 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" @@ -324,6 +337,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 +412,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, emptyStreamDetail(collected)) + 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/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"` } 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) + } +} 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) }