fix(core): bound silent stream failures (#173)

* fix(core): bound silent stream failures

* test(core): guard empty durable events

commit: 809ffbab89efd1b5666679626461200d038c4ba0
author: bnema · 48d ago · 2026-09-04 13:58 UTC
committer: GitHub · 48d ago · 2026-09-04 13:58 UTC
parent: 73d6864
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

3 files · +72 -10

M README.md +1 -1


@@ -505,7 +505,7 @@ When a turn fails because of a recoverable provider error — expired token (`40
505	505
506	506		The picker is the same vertical list / fuzzy filter UI as `/model`, but it only shows models from providers you're currently logged in to (env vars, `auth.json`, Kimi CLI fallback, ollama). The failed model is excluded. Press `↑`/`↓` to choose, `enter` to retry the same prompt on the new model, `esc` to dismiss.
507	507
508		-	Before the actual provider request fires, the shared HTTP streaming clients used by OpenAI / Anthropic / Kimi / DeepSeek / Google / OpenAI-Codex also do up to four silent retries with short backoff (250ms, then 750ms, 750ms, and 750ms) on transient HTTP failures and connection-reset / EOF-before-headers errors. Once a stream is open, five reconnect attempts and a five-minute idle timeout independently guard stalled or interrupted streams. Most edge-proxy blips disappear without you ever seeing the rescue picker.
	508	+	Before the actual provider request fires, the shared HTTP streaming clients used by OpenAI / Anthropic / Kimi / DeepSeek / Google / OpenAI-Codex also do up to four silent retries with short backoff (250ms, then 750ms, 750ms, and 750ms) on transient HTTP failures and connection-reset / EOF-before-headers errors. Once a stream is open, up to five reconnect attempts guard interrupted streams. Independently, five minutes without a provider event fails the turn instead of restarting the idle deadline on each retry. Most edge-proxy blips disappear without you ever seeing the rescue picker.
509	509
510	510		A rescue retry always drops launch-time `--api-key` and `--base-url` before rebuilding the agent. Those overrides are usually the reason the rescue triggered (bad key, typo'd base URL, corporate gateway only valid for the originally-picked provider), so the retry re-resolves credentials from env vars / `auth.json` / provider defaults instead. Use `/model` if you want overrides to stick.
511	511


@@ -505,7 +505,7 @@ When a turn fails because of a recoverable provider error — expired token (`40
505			505
506		The picker is the same vertical list / fuzzy filter UI as `/model`, but it only shows models from providers you're currently logged in to (env vars, `auth.json`, Kimi CLI fallback, ollama). The failed model is excluded. Press `↑`/`↓` to choose, `enter` to retry the same prompt on the new model, `esc` to dismiss.	506		The picker is the same vertical list / fuzzy filter UI as `/model`, but it only shows models from providers you're currently logged in to (env vars, `auth.json`, Kimi CLI fallback, ollama). The failed model is excluded. Press `↑`/`↓` to choose, `enter` to retry the same prompt on the new model, `esc` to dismiss.
507			507
508	-	Before the actual provider request fires, the shared HTTP streaming clients used by OpenAI / Anthropic / Kimi / DeepSeek / Google / OpenAI-Codex also do up to four silent retries with short backoff (250ms, then 750ms, 750ms, and 750ms) on transient HTTP failures and connection-reset / EOF-before-headers errors. Once a stream is open, five reconnect attempts and a five-minute idle timeout independently guard stalled or interrupted streams. Most edge-proxy blips disappear without you ever seeing the rescue picker.	508	+	Before the actual provider request fires, the shared HTTP streaming clients used by OpenAI / Anthropic / Kimi / DeepSeek / Google / OpenAI-Codex also do up to four silent retries with short backoff (250ms, then 750ms, 750ms, and 750ms) on transient HTTP failures and connection-reset / EOF-before-headers errors. Once a stream is open, up to five reconnect attempts guard interrupted streams. Independently, five minutes without a provider event fails the turn instead of restarting the idle deadline on each retry. Most edge-proxy blips disappear without you ever seeing the rescue picker.
509			509
510		A rescue retry always drops launch-time `--api-key` and `--base-url` before rebuilding the agent. Those overrides are usually the reason the rescue triggered (bad key, typo'd base URL, corporate gateway only valid for the originally-picked provider), so the retry re-resolves credentials from env vars / `auth.json` / provider defaults instead. Use `/model` if you want overrides to stick.	510		A rescue retry always drops launch-time `--api-key` and `--base-url` before rebuilding the agent. Those overrides are usually the reason the rescue triggered (bad key, typo'd base URL, corporate gateway only valid for the originally-picked provider), so the retry re-resolves credentials from env vars / `auth.json` / provider defaults instead. Use `/model` if you want overrides to stick.
511			511

M packages/core/agent.go +5 -1


@@ -785,7 +785,11 @@ func (a *Agent) canRetryError(err error, attempt int) bool {
785	785		if errors.As(err, &openErr) {
786	786		return false
787	787		}
788		-	if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {
	788	+	// An idle timeout already bounds a full period without provider progress.
	789	+	// Retrying it would restart that deadline for every attempt, allowing one
	790	+	// silent model continuation to keep an agent running for more than thirty
	791	+	// minutes under the default retry policy.
	792	+	if errors.Is(err, ErrStreamIdleTimeout) \|\| errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {
789	793		return false
790	794		}
791	795		msg := strings.ToLower(err.Error())

 	if errors.As(err, &openErr) {
 		return false
 	}
-	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+	// An idle timeout already bounds a full period without provider progress.
+	// Retrying it would restart that deadline for every attempt, allowing one
+	// silent model continuation to keep an agent running for more than thirty
+	// minutes under the default retry policy.
+	if errors.Is(err, ErrStreamIdleTimeout) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return false
 	}
 	msg := strings.ToLower(err.Error())

M packages/core/agent_retry_test.go +66 -8


@@ -2,6 +2,7 @@ package core
2	2
3	3		import (
4	4		"context"
	5	+	"encoding/json"
5	6		"errors"
6	7		"fmt"
7	8		"io"

@@ -73,6 +74,32 @@ func (activeStreamClient) Stream(context.Context, provider.Request) (<-chan prov
73	74		return out, nil
74	75		}
75	76
	77	+	type postToolSilentClient struct {
	78	+	calls atomic.Int32
	79	+	}
	80	+
	81	+	func (c *postToolSilentClient) Name() string { return "post-tool-silent" }
	82	+
	83	+	func (c *postToolSilentClient) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Event, error) {
	84	+	if c.calls.Add(1) == 1 {
	85	+	out := make(chan provider.Event, 1)
	86	+	out <- provider.EventDone{Stop: provider.StopToolUse, Message: provider.Message{
	87	+	Role: provider.RoleAssistant,
	88	+	Content: []provider.Content{provider.ToolCallBlock{
	89	+	ID: "call-1", Name: "result", Arguments: json.RawMessage(`{}`),
	90	+	}},
	91	+	}}
	92	+	close(out)
	93	+	return out, nil
	94	+	}
	95	+	out := make(chan provider.Event)
	96	+	go func() {
	97	+	defer close(out)
	98	+	<-ctx.Done()
	99	+	}()
	100	+	return out, nil
	101	+	}
	102	+
76	103		type requestOpenFailureClient struct{ calls int32 }
77	104
78	105		func (c *requestOpenFailureClient) Name() string { return "request-open-failure" }

@@ -159,19 +186,50 @@ func TestAgentResetsIdleDeadlineForEveryStreamEvent(t *testing.T) {
159	186		}
160	187		}
161	188
162		-	func TestAgentRetriesSilentStreamWithoutEndingTurn(t *testing.T) {
	189	+	func TestAgentDoesNotRetrySilentStream(t *testing.T) {
163	190		client := &retryFakeClient{firstErr: ErrStreamIdleTimeout}
164	191		a := NewAgent(client, "fake-model", "system", Registry{})
165		-	a.MaxRetries = 1
166	192		a.RetryBaseDelay = time.Millisecond
167		-	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
168		-	t.Fatalf("Prompt returned %v", err)
	193	+	if err := a.Prompt(context.Background(), "hello", nil, nil); !errors.Is(err, ErrStreamIdleTimeout) {
	194	+	t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
169	195		}
170		-	if got := atomic.LoadInt32(&client.calls); got != 2 {
171		-	t.Fatalf("Stream calls = %d; want reconnect after idle timeout", got)
	196	+	if got := atomic.LoadInt32(&client.calls); got != 1 {
	197	+	t.Fatalf("Stream calls = %d; want no reconnect after idle deadline", got)
172	198		}
173		-	if got := extractText(a.Messages()[1]); got != "ok" {
174		-	t.Fatalf("final assistant text = %q; want ok", got)
	199	+	}
	200	+
	201	+	func TestAgentFailsSilentContinuationAfterToolResult(t *testing.T) {
	202	+	client := &postToolSilentClient{}
	203	+	a := NewAgent(client, "fake-model", "system", Registry{
	204	+	"result": &resultTool{result: ToolResult{Content: []provider.Content{provider.TextBlock{Text: "ok"}}}},
	205	+	})
	206	+	a.RetryBaseDelay = time.Millisecond
	207	+	var timers atomic.Int32
	208	+	a.streamIdleTimer = func(time.Duration) (<-chan time.Time, func(), func()) {
	209	+	idle := make(chan time.Time, 1)
	210	+	if timers.Add(1) > 1 {
	211	+	idle <- time.Now()
	212	+	}
	213	+	return idle, func() {}, func() {}
	214	+	}
	215	+	var durable []string
	216	+	err := a.Prompt(context.Background(), "review", nil, func(event AgentEvent) {
	217	+	switch event.(type) {
	218	+	case EvUserMessage, EvAssistantMessage, EvToolCall, EvToolResult:
	219	+	durable = append(durable, event.Type())
	220	+	}
	221	+	})
	222	+	if !errors.Is(err, ErrStreamIdleTimeout) {
	223	+	t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
	224	+	}
	225	+	if got := client.calls.Load(); got != 2 {
	226	+	t.Fatalf("provider calls = %d, want tool request plus one bounded continuation", got)
	227	+	}
	228	+	if len(durable) == 0 {
	229	+	t.Fatal("agent emitted no durable events")
	230	+	}
	231	+	if got, want := durable[len(durable)-1], "tool_result"; got != want {
	232	+	t.Fatalf("last durable event = %q, want %q; events = %v", got, want, durable)
175	233		}
176	234		}
177	235


@@ -2,6 +2,7 @@ package core
2			2
3		import (	3		import (
4		"context"	4		"context"
			5	+	"encoding/json"
5		"errors"	6		"errors"
6		"fmt"	7		"fmt"
7		"io"	8		"io"

@@ -73,6 +74,32 @@ func (activeStreamClient) Stream(context.Context, provider.Request) (<-chan prov
73		return out, nil	74		return out, nil
74		}	75		}
75			76
			77	+	type postToolSilentClient struct {
			78	+	calls atomic.Int32
			79	+	}
			80	+
			81	+	func (c *postToolSilentClient) Name() string { return "post-tool-silent" }
			82	+
			83	+	func (c *postToolSilentClient) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Event, error) {
			84	+	if c.calls.Add(1) == 1 {
			85	+	out := make(chan provider.Event, 1)
			86	+	out <- provider.EventDone{Stop: provider.StopToolUse, Message: provider.Message{
			87	+	Role: provider.RoleAssistant,
			88	+	Content: []provider.Content{provider.ToolCallBlock{
			89	+	ID: "call-1", Name: "result", Arguments: json.RawMessage(`{}`),
			90	+	}},
			91	+	}}
			92	+	close(out)
			93	+	return out, nil
			94	+	}
			95	+	out := make(chan provider.Event)
			96	+	go func() {
			97	+	defer close(out)
			98	+	<-ctx.Done()
			99	+	}()
			100	+	return out, nil
			101	+	}
			102	+
76		type requestOpenFailureClient struct{ calls int32 }	103		type requestOpenFailureClient struct{ calls int32 }
77			104
78		func (c *requestOpenFailureClient) Name() string { return "request-open-failure" }	105		func (c *requestOpenFailureClient) Name() string { return "request-open-failure" }

@@ -159,19 +186,50 @@ func TestAgentResetsIdleDeadlineForEveryStreamEvent(t *testing.T) {
159		}	186		}
160		}	187		}
161			188
162	-	func TestAgentRetriesSilentStreamWithoutEndingTurn(t *testing.T) {	189	+	func TestAgentDoesNotRetrySilentStream(t *testing.T) {
163		client := &retryFakeClient{firstErr: ErrStreamIdleTimeout}	190		client := &retryFakeClient{firstErr: ErrStreamIdleTimeout}
164		a := NewAgent(client, "fake-model", "system", Registry{})	191		a := NewAgent(client, "fake-model", "system", Registry{})
165	-	a.MaxRetries = 1
166		a.RetryBaseDelay = time.Millisecond	192		a.RetryBaseDelay = time.Millisecond
167	-	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {	193	+	if err := a.Prompt(context.Background(), "hello", nil, nil); !errors.Is(err, ErrStreamIdleTimeout) {
168	-	t.Fatalf("Prompt returned %v", err)	194	+	t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
169		}	195		}
170	-	if got := atomic.LoadInt32(&client.calls); got != 2 {	196	+	if got := atomic.LoadInt32(&client.calls); got != 1 {
171	-	t.Fatalf("Stream calls = %d; want reconnect after idle timeout", got)	197	+	t.Fatalf("Stream calls = %d; want no reconnect after idle deadline", got)
172		}	198		}
173	-	if got := extractText(a.Messages()[1]); got != "ok" {	199	+	}
174	-	t.Fatalf("final assistant text = %q; want ok", got)	200	+
			201	+	func TestAgentFailsSilentContinuationAfterToolResult(t *testing.T) {
			202	+	client := &postToolSilentClient{}
			203	+	a := NewAgent(client, "fake-model", "system", Registry{
			204	+	"result": &resultTool{result: ToolResult{Content: []provider.Content{provider.TextBlock{Text: "ok"}}}},
			205	+	})
			206	+	a.RetryBaseDelay = time.Millisecond
			207	+	var timers atomic.Int32
			208	+	a.streamIdleTimer = func(time.Duration) (<-chan time.Time, func(), func()) {
			209	+	idle := make(chan time.Time, 1)
			210	+	if timers.Add(1) > 1 {
			211	+	idle <- time.Now()
			212	+	}
			213	+	return idle, func() {}, func() {}
			214	+	}
			215	+	var durable []string
			216	+	err := a.Prompt(context.Background(), "review", nil, func(event AgentEvent) {
			217	+	switch event.(type) {
			218	+	case EvUserMessage, EvAssistantMessage, EvToolCall, EvToolResult:
			219	+	durable = append(durable, event.Type())
			220	+	}
			221	+	})
			222	+	if !errors.Is(err, ErrStreamIdleTimeout) {
			223	+	t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
			224	+	}
			225	+	if got := client.calls.Load(); got != 2 {
			226	+	t.Fatalf("provider calls = %d, want tool request plus one bounded continuation", got)
			227	+	}
			228	+	if len(durable) == 0 {
			229	+	t.Fatal("agent emitted no durable events")
			230	+	}
			231	+	if got, want := durable[len(durable)-1], "tool_result"; got != want {
			232	+	t.Fatalf("last durable event = %q, want %q; events = %v", got, want, durable)
175		}	233		}
176		}	234		}
177			235