fix(providerio): surface ErrStreamStalled in callers + widen flaky test margins

Address review:
- All four provider stream handlers (openai, codex, anthropic, gemini) now treat
  ErrStreamStalled in the same branch as ErrStreamIdle, via a new
  providerio.StreamTimeoutMessage helper that gives the stall a distinct, accurate
  message (it reports the content window and does NOT claim the upstream stopped
  sending data — keep-alives were still arriving). Previously a stalled stream fell
  through to the generic provider-error path.
- Widen the content-watchdog test timing margins (idle 100ms, data/keepalive every
  30ms) so CI/GC jitter can't trip the idle watchdog and mask the content path.
- Add a StreamTimeoutMessage unit test.

internal/providers/anthropic/provider.go

@@ -211,11 +211,11 @@ func (provider *Provider) stream(ctx context.Context, body []byte, events chan<-
 	err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, provider.streamIdleTimeout, func(data string) bool {
 		return provider.emitPayload(ctx, data, state, events)
 	})
-	if errors.Is(err, providerio.ErrStreamIdle) {
+	if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
 		state.closeOpen(ctx, events)
 		providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
 			Type:  zeroruntime.StreamEventError,
-			Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
+			Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
 		})
 		return
 	}

internal/providers/gemini/provider.go

@@ -158,10 +158,10 @@ func (provider *Provider) stream(ctx context.Context, body []byte, events chan<-
 	err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, provider.streamIdleTimeout, func(data string) bool {
 		return provider.emitPayload(ctx, data, &state, events)
 	})
-	if errors.Is(err, providerio.ErrStreamIdle) {
+	if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
 		providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
 			Type:  zeroruntime.StreamEventError,
-			Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
+			Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
 		})
 		return
 	}

internal/providers/openai/codex_responses.go

@@ -389,10 +389,10 @@ func (p *CodexProvider) streamResponses(
 	err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, inner.streamIdleTimeout, func(data string) bool {
 		return p.emitResponsesEvent(ctx, data, state, events)
 	})
-	if errors.Is(err, providerio.ErrStreamIdle) {
+	if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
 		providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
 			Type:  zeroruntime.StreamEventError,
-			Error: p.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", inner.streamIdleTimeout)),
+			Error: p.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, inner.streamIdleTimeout)),
 		})
 		return
 	}

internal/providers/openai/provider.go

@@ -218,12 +218,12 @@ func (provider *Provider) stream(ctx context.Context, body []byte, events chan<-
 	err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, provider.streamIdleTimeout, func(data string) bool {
 		return provider.emitPayload(ctx, data, state, events)
 	})
-	if errors.Is(err, providerio.ErrStreamIdle) {
+	if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
 		state.flushBufferedContent(events)
 		state.closeBufferedOpen(events)
 		sendEvent(ctx, events, zeroruntime.StreamEvent{
 			Type:  zeroruntime.StreamEventError,
-			Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
+			Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
 		})
 		return
 	}

internal/providers/providerio/providerio.go

@@ -23,6 +23,26 @@ const maxSSELineBytes = 16 * 1024 * 1024
 // closing the connection. Callers surface it as an idle-timeout error.
 var ErrStreamIdle = errors.New("idle timeout (upstream stopped sending data)")
 
+// ErrStreamStalled reports that a streaming upstream kept the connection alive
+// (SSE keep-alives reset the idle watchdog, so it never fired) but produced no
+// actual output for streamContentStallFactor × the idle timeout. Without this an
+// upstream that heartbeats-but-stalls — observed on chatgpt/gpt-5.x and ollama
+// reasoning models — would hang the agent indefinitely.
+var ErrStreamStalled = errors.New("stream stalled (upstream kept the connection alive but produced no output)")
+
+// StreamTimeoutMessage returns the human-readable detail for a stream-timeout
+// error (ErrStreamIdle or ErrStreamStalled), given the configured idle timeout.
+// Callers prepend their own "provider stream error: " prefix. A stalled stream
+// gets a distinct, actionable message — it did NOT stop sending data (keep-alives
+// kept arriving); it just produced no output, which usually means the model is
+// stuck or very slow.
+func StreamTimeoutMessage(err error, idleTimeout time.Duration) string {
+	if errors.Is(err, ErrStreamStalled) {
+		return fmt.Sprintf("no output for %s (the model kept the connection alive but produced nothing — it may be stuck; try a faster model or lower reasoning effort)", idleTimeout*streamContentStallFactor)
+	}
+	return fmt.Sprintf("idle timeout after %s (upstream stopped sending data)", idleTimeout)
+}
+
 // DefaultStreamIdleTimeout is the single source of truth for how long every
 // provider waits on a silent stream before aborting it. The watchdog only fires
 // on genuine silence — SSE keep-alive comments reset it (see
@@ -33,6 +53,14 @@ var ErrStreamIdle = errors.New("idle timeout (upstream stopped sending data)")
 // Override globally with ZERO_STREAM_IDLE_TIMEOUT.
 const DefaultStreamIdleTimeout = 5 * time.Minute
 
+// streamContentStallFactor bounds a heartbeat-but-no-output stream. Keep-alives
+// reset the idle watchdog (a heartbeating upstream is not "dead"), but if NO real
+// data line arrives for streamContentStallFactor × the idle timeout, the stream is
+// treated as stalled and aborted (ErrStreamStalled). It is > 1 so a slow-but-
+// producing request — one that emits data between heartbeats — is never killed; the
+// content watchdog only catches a stream that heartbeats while producing nothing.
+const streamContentStallFactor = 2
+
 // 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
@@ -202,21 +230,36 @@ func ScanSSEDataWithContext(
 	// The idle watchdog is optional. When idleTimeout <= 0 it is disabled, but we
 	// STILL run the goroutine + select loop so ctx cancellation is always honored
 	// (a nil idleC channel simply never fires in the select).
-	var idleC <-chan time.Time
+	var idleC, contentC <-chan time.Time
 	resetIdle := func() {}
+	resetContent := func() {}
 	if idleTimeout > 0 {
+		// reset drains the timer's channel if it already fired before re-arming, so
+		// a stale tick can't trip the watchdog after activity resumed.
+		reset := func(t *time.Timer, d time.Duration) func() {
+			return func() {
+				if !t.Stop() {
+					select {
+					case <-t.C:
+					default:
+					}
+				}
+				t.Reset(d)
+			}
+		}
 		idle := time.NewTimer(idleTimeout)
 		defer idle.Stop()
 		idleC = idle.C
-		resetIdle = func() {
-			if !idle.Stop() {
-				select {
-				case <-idle.C:
-				default:
-				}
-			}
-			idle.Reset(idleTimeout)
-		}
+		resetIdle = reset(idle, 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
+		// hanging forever.
+		contentTimeout := idleTimeout * streamContentStallFactor
+		content := time.NewTimer(contentTimeout)
+		defer content.Stop()
+		contentC = content.C
+		resetContent = reset(content, contentTimeout)
 	}
 
 	for {
@@ -232,6 +275,12 @@ func ScanSSEDataWithContext(
 			// a timeout instead of blocking the agent forever.
 			cancel()
 			return ErrStreamIdle
+		case <-contentC:
+			// Upstream kept heartbeating (idle watchdog never fired) but produced
+			// no output for the content window. Treat as stalled and abort rather
+			// than hanging forever.
+			cancel()
+			return ErrStreamStalled
 		case item, ok := <-payloads:
 			if !ok {
 				// Reader finished: deliver its terminal status (EOF -> nil,
@@ -246,8 +295,11 @@ func ScanSSEDataWithContext(
 			}
 			resetIdle()
 			if item.keepAlive {
+				// A heartbeat keeps the connection "alive" (idle watchdog) but is
+				// NOT output, so it must not reset the content watchdog.
 				continue
 			}
+			resetContent()
 			if !handle(item.data) {
 				// The provider asked to stop (e.g. it already emitted an error
 				// for this payload). Abort the read and end like ScanSSEData:

internal/providers/providerio/providerio_test.go

@@ -201,3 +201,106 @@ func TestUpstreamUnreachable(t *testing.T) {
 		})
 	}
 }
+
+// A heartbeating-but-output-less upstream must not hang forever: SSE keep-alives
+// feed the idle watchdog (so it never fires), but the content watchdog
+// (idleTimeout × streamContentStallFactor) aborts with ErrStreamStalled when no
+// real data line arrives. This is the gpt-5.x / ollama "still generating forever"
+// hang.
+func TestScanSSEDataWithContextAbortsOnContentStall(t *testing.T) {
+	pr, pw := io.Pipe()
+	defer func() { _ = pw.Close() }()
+
+	stop := make(chan struct{})
+	defer close(stop)
+	go func() {
+		// Heartbeat every 30ms — comfortably under the 100ms idle timeout, so idle
+		// never fires under CI/GC jitter — but never send a data line.
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			if _, err := io.WriteString(pw, ": keep-alive\n\n"); err != nil {
+				return
+			}
+			time.Sleep(30 * time.Millisecond)
+		}
+	}()
+
+	cancelled := false
+	cancel := func() { cancelled = true }
+
+	done := make(chan error, 1)
+	go func() {
+		// idle 100ms → content stall at 200ms. Keep-alives reset idle but not content.
+		done <- ScanSSEDataWithContext(context.Background(), cancel, pr, 100*time.Millisecond, func(string) bool { return true })
+	}()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, ErrStreamStalled) {
+			t.Fatalf("err = %v, want ErrStreamStalled", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("ScanSSEDataWithContext hung on a heartbeat-but-no-output stream")
+	}
+	if !cancelled {
+		t.Fatal("content stall did not cancel the request context")
+	}
+}
+
+// Real data lines reset the content watchdog, so a slow-but-producing stream that
+// runs longer than the content window is never aborted.
+func TestScanSSEDataWithContextContentResetsOnData(t *testing.T) {
+	pr, pw := io.Pipe()
+
+	go func() {
+		// One data line every 30ms for ~300ms (well past the 200ms content window,
+		// with comfortable slack under the 100ms idle timeout), so the content
+		// watchdog keeps resetting, then close cleanly.
+		for i := 0; i < 10; i++ {
+			if _, err := io.WriteString(pw, "data: chunk\n\n"); err != nil {
+				return
+			}
+			time.Sleep(30 * time.Millisecond)
+		}
+		_ = pw.Close()
+	}()
+
+	n := 0
+	done := make(chan error, 1)
+	go func() {
+		done <- ScanSSEDataWithContext(context.Background(), func() {}, pr, 100*time.Millisecond, func(string) bool { n++; return true })
+	}()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("err = %v, want nil (data kept the stream alive)", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("ScanSSEDataWithContext hung on a producing stream")
+	}
+	if n != 10 {
+		t.Fatalf("handled %d data lines, want 10", n)
+	}
+}
+
+// StreamTimeoutMessage must give a stalled stream a distinct, accurate detail —
+// it reports the content window (idle × factor) and must NOT claim the upstream
+// "stopped sending data" (keep-alives were still arriving).
+func TestStreamTimeoutMessage(t *testing.T) {
+	idle := 5 * time.Minute
+	if msg := StreamTimeoutMessage(ErrStreamIdle, idle); !strings.Contains(msg, "idle timeout after 5m") {
+		t.Fatalf("idle message = %q, want it to mention the 5m idle timeout", msg)
+	}
+	stalled := StreamTimeoutMessage(ErrStreamStalled, idle)
+	if !strings.Contains(stalled, "no output for 10m") {
+		t.Fatalf("stalled message = %q, want it to report the 10m content window", stalled)
+	}
+	if strings.Contains(stalled, "stopped sending data") {
+		t.Fatalf("stalled message must not claim the upstream stopped sending data: %q", stalled)
+	}
+}