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/factory_test.go

@@ -264,6 +264,13 @@ func TestNewRejectsUnsupportedProviderKind(t *testing.T) {
 }
 
 func TestNewRoutesChatGPTCatalogToCodexProvider(t *testing.T) {
+	// Isolate the OAuth token store to an empty temp path so the factory reads no
+	// stored login — otherwise this test picks up the developer's real chatgpt
+	// OAuth token and the "want empty chatgpt-account-id" assertion fails locally
+	// (it still passes in CI, where no login is stored). Mirrors the isolation in
+	// TestNewRoutesChatGPTCatalogWithStoredAccountID.
+	t.Setenv("ZERO_OAUTH_TOKENS_PATH", t.TempDir()+"/tokens.json")
+
 	transport := &captureTransport{
 		responseBody: "data: [DONE]\n\n",
 	}

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)
+	}
+}

internal/swarm/tools.go

@@ -103,6 +103,25 @@ func errResult(format string, a ...any) tools.Result {
 	return tools.Result{Status: tools.StatusError, Output: "Error: " + fmt.Sprintf(format, a...)}
 }
 
+// agentTypeProperty builds the schema for an agent-type parameter, constraining it
+// to the roster's currently registered types (the built-ins plus any user-registered
+// agents) via an enum so a model can't spawn or hand off to an unknown type — the
+// schema teaches the valid set up front instead of the model learning it from a
+// failed call. Falls back to a free-form string if the roster is unavailable or
+// empty (so the tool never advertises an empty, un-satisfiable enum).
+func agentTypeProperty(sw *Swarm, description string) tools.PropertySchema {
+	prop := tools.PropertySchema{Type: "string", Description: description}
+	if sw == nil {
+		return prop
+	}
+	if registry := sw.Registry(); registry != nil {
+		if types := registry.AgentTypes(); len(types) > 0 {
+			prop.Enum = types
+		}
+	}
+	return prop
+}
+
 // ---- swarm_spawn -----------------------------------------------------------
 
 type spawnTool struct {
@@ -118,7 +137,7 @@ func (t *spawnTool) Parameters() tools.Schema {
 	return tools.Schema{
 		Type: "object",
 		Properties: map[string]tools.PropertySchema{
-			"agent_type": {Type: "string", Description: "Roster agent type to spawn (e.g. teammate, subagent)."},
+			"agent_type": agentTypeProperty(t.sw, "Roster agent type to spawn."),
 			"task":       {Type: "string", Description: "The focused task/briefing for the member."},
 			"team":       {Type: "string", Description: "Team to spawn into. Defaults to \"default\"."},
 		},
@@ -353,7 +372,7 @@ func (t *handoffTool) Parameters() tools.Schema {
 		Type: "object",
 		Properties: map[string]tools.PropertySchema{
 			"task_id":       {Type: "string", Description: "Task id to hand off."},
-			"to_agent_type": {Type: "string", Description: "Roster agent type to take over."},
+			"to_agent_type": agentTypeProperty(t.sw, "Roster agent type to take over."),
 			"note":          {Type: "string", Description: "Optional handoff note delivered to the new member's inbox."},
 			"team":          {Type: "string", Description: "Team the task belongs to. Defaults to \"default\"."},
 		},

internal/swarm/tools_test.go

@@ -2,6 +2,7 @@ package swarm
 
 import (
 	"context"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -10,6 +11,46 @@ import (
 	"github.com/Gitlawb/zero/internal/tools"
 )
 
+func TestSpawnAndHandoffAgentTypeEnumReflectsRoster(t *testing.T) {
+	sw := newSwarmFor(t, newLauncher(okFor))
+	want := sw.Registry().AgentTypes() // sorted built-ins: subagent, teammate
+
+	spawnEnum := (&spawnTool{sw: sw}).Parameters().Properties["agent_type"].Enum
+	if !slices.Equal(spawnEnum, want) {
+		t.Fatalf("spawn agent_type enum = %v, want roster %v", spawnEnum, want)
+	}
+	if !slices.Contains(spawnEnum, "subagent") || !slices.Contains(spawnEnum, "teammate") {
+		t.Fatalf("enum missing built-in roster types: %v", spawnEnum)
+	}
+	handoffEnum := (&handoffTool{sw: sw}).Parameters().Properties["to_agent_type"].Enum
+	if !slices.Equal(handoffEnum, want) {
+		t.Fatalf("handoff to_agent_type enum = %v, want roster %v", handoffEnum, want)
+	}
+
+	// A user-registered custom type must extend the enum (not a hardcoded list).
+	if err := sw.Registry().Register(Definition{AgentType: "researcher"}); err != nil {
+		t.Fatalf("register custom agent: %v", err)
+	}
+	spawnEnum = (&spawnTool{sw: sw}).Parameters().Properties["agent_type"].Enum
+	if !slices.Contains(spawnEnum, "researcher") {
+		t.Fatalf("custom agent type not reflected in spawn enum: %v", spawnEnum)
+	}
+	handoffEnum = (&handoffTool{sw: sw}).Parameters().Properties["to_agent_type"].Enum
+	if !slices.Contains(handoffEnum, "researcher") {
+		t.Fatalf("custom agent type not reflected in handoff enum: %v", handoffEnum)
+	}
+}
+
+func TestAgentTypePropertyFallsBackWithoutRoster(t *testing.T) {
+	prop := agentTypeProperty(nil, "desc")
+	if prop.Type != "string" || prop.Description != "desc" {
+		t.Fatalf("unexpected property: %+v", prop)
+	}
+	if prop.Enum != nil {
+		t.Fatalf("nil swarm must not advertise an (empty) enum, got %v", prop.Enum)
+	}
+}
+
 func TestCollapseRuneSafeTruncation(t *testing.T) {
 	// 300 multi-byte runes: truncation must not split a rune (no invalid UTF-8).
 	got := collapse(strings.Repeat("🚀", 300))