4ed3d6d fix(subagents): preserve durable delivery facts · zut

M docs/subagents.md +1 -1


@@ -137,7 +137,7 @@ Shared mode preserves the historical host working directory, so workers there mu
137	137
138	138		Every child retains narrowly scoped internal process control and durable task/result metadata. These details coordinate startup, cancellation, and recovery; they are not a user-facing progress model. Diagnostic progress comes only from paired trace boundaries and their projection into open operations, terminal facts, or an explicitly insufficient observation.
139	139
140		-	During a trace-observed open operation, interactive mode shows compact rows beside the input area: a themed spinner, the named profile (or agent ID), the most specific open operation, and its elapsed duration. A recent safe observation is shown alongside it when available, for example `assistant streaming · 1s ago`; normal mode never displays message text, reasoning text, tool arguments, or tool output. Rows default below the input and can move above it through running subagent position in `/settings` → tui settings. On short terminals, omitted operations collapse into one count summary so the editor remains visible. A process lifecycle, heartbeat, or generic `working` label never creates a row; use `/subagents` for the current factual operation or last observed fact.
	140	+	During a trace-observed open operation, interactive mode shows compact rows beside the input area: a themed spinner, the named profile (or agent ID), the most specific open operation, and its elapsed duration. A recent safe observation is shown alongside it when available, for example `assistant streaming 1s ago`; normal mode never displays message text, reasoning text, tool arguments, or tool output. Rows default below the input and can move above it through running subagent position in `/settings` → tui settings. On short terminals, omitted operations collapse into one count summary so the editor remains visible. A process lifecycle, heartbeat, or generic `working` label never creates a row; use `/subagents` for the current factual operation or last observed fact.
141	141
142	142		`/subagents` opens on workers from the current host session. Press Tab to switch to the all-sessions history, use Up/Down to select a worker, and press Right or l to expand a bounded live snapshot of its user messages, assistant output, and tool activity (Left or h collapses it). The dashboard derives its row budget from the terminal height, keeping the selected worker visible on short terminals. Press Enter for the complete transcript.
143	143


@@ -137,7 +137,7 @@ Shared mode preserves the historical host working directory, so workers there mu
137			137
138		Every child retains narrowly scoped internal process control and durable task/result metadata. These details coordinate startup, cancellation, and recovery; they are not a user-facing progress model. Diagnostic progress comes only from paired trace boundaries and their projection into open operations, terminal facts, or an explicitly insufficient observation.	138		Every child retains narrowly scoped internal process control and durable task/result metadata. These details coordinate startup, cancellation, and recovery; they are not a user-facing progress model. Diagnostic progress comes only from paired trace boundaries and their projection into open operations, terminal facts, or an explicitly insufficient observation.
139			139
140	-	During a trace-observed open operation, interactive mode shows compact rows beside the input area: a themed spinner, the named profile (or agent ID), the most specific open operation, and its elapsed duration. A recent safe observation is shown alongside it when available, for example `assistant streaming · 1s ago`; normal mode never displays message text, reasoning text, tool arguments, or tool output. Rows default below the input and can move above it through running subagent position in `/settings` → tui settings. On short terminals, omitted operations collapse into one count summary so the editor remains visible. A process lifecycle, heartbeat, or generic `working` label never creates a row; use `/subagents` for the current factual operation or last observed fact.	140	+	During a trace-observed open operation, interactive mode shows compact rows beside the input area: a themed spinner, the named profile (or agent ID), the most specific open operation, and its elapsed duration. A recent safe observation is shown alongside it when available, for example `assistant streaming 1s ago`; normal mode never displays message text, reasoning text, tool arguments, or tool output. Rows default below the input and can move above it through running subagent position in `/settings` → tui settings. On short terminals, omitted operations collapse into one count summary so the editor remains visible. A process lifecycle, heartbeat, or generic `working` label never creates a row; use `/subagents` for the current factual operation or last observed fact.
141			141
142		`/subagents` opens on workers from the current host session. Press Tab to switch to the all-sessions history, use Up/Down to select a worker, and press Right or l to expand a bounded live snapshot of its user messages, assistant output, and tool activity (Left or h collapses it). The dashboard derives its row budget from the terminal height, keeping the selected worker visible on short terminals. Press Enter for the complete transcript.	142		`/subagents` opens on workers from the current host session. Press Tab to switch to the all-sessions history, use Up/Down to select a worker, and press Right or l to expand a bounded live snapshot of its user messages, assistant output, and tool activity (Left or h collapses it). The dashboard derives its row budget from the terminal height, keeping the selected worker visible on short terminals. Press Enter for the complete transcript.
143			143

M packages/agent/subagent_worker.go +1 -1


@@ -710,7 +710,7 @@ func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[
710	710		return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
711	711		}
712	712		if errors.Is(err, core.ErrStreamIdleTimeout) {
713		-	return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for one minute; the delegated turn stopped safely"}
	713	+	return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for five minutes; the delegated turn stopped safely"}
714	714		}
715	715		if errors.Is(err, context.DeadlineExceeded) \|\| shutdownOrigin == subagents.ShutdownOriginDeadline {
716	716		return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}


@@ -710,7 +710,7 @@ func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[
710		return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}	710		return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
711		}	711		}
712		if errors.Is(err, core.ErrStreamIdleTimeout) {	712		if errors.Is(err, core.ErrStreamIdleTimeout) {
713	-	return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for one minute; the delegated turn stopped safely"}	713	+	return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for five minutes; the delegated turn stopped safely"}
714		}	714		}
715		if errors.Is(err, context.DeadlineExceeded) \|\| shutdownOrigin == subagents.ShutdownOriginDeadline {	715		if errors.Is(err, context.DeadlineExceeded) \|\| shutdownOrigin == subagents.ShutdownOriginDeadline {
716		return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}	716		return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}

M packages/agent/subagents/projection.go +15 -4


@@ -179,8 +179,8 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView {
179	179		name, _ := event.Data["name"].(string)
180	180		provider, _ := event.Data["provider"].(string)
181	181		model, _ := event.Data["model"].(string)
182		-	attempt, _ := event.Data["attempt"].(int)
183		-	maxAttempts, _ := event.Data["max_attempts"].(int)
	182	+	attempt := traceInt(event.Data["attempt"])
	183	+	maxAttempts := traceInt(event.Data["max_attempts"])
184	184		open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, CallID: callID, Name: name, Provider: provider, Model: model, Attempt: attempt, MaxAttempts: maxAttempts, StartedAt: event.Timestamp}
185	185		}
186	186		if ends {

@@ -279,11 +279,22 @@ func traceObservation(event TraceEvent) *LiveObservation {
279	279		return nil
280	280		}
281	281
	282	+	func traceInt(value any) int {
	283	+	switch value := value.(type) {
	284	+	case int:
	285	+	return value
	286	+	case float64:
	287	+	return int(value)
	288	+	default:
	289	+	return 0
	290	+	}
	291	+	}
	292	+
282	293		func requestFact(event TraceEvent, outcome string) *RequestFact {
283	294		providerName, _ := event.Data["provider"].(string)
284	295		model, _ := event.Data["model"].(string)
285		-	attempt, _ := event.Data["attempt"].(int)
286		-	maxAttempts, _ := event.Data["max_attempts"].(int)
	296	+	attempt := traceInt(event.Data["attempt"])
	297	+	maxAttempts := traceInt(event.Data["max_attempts"])
287	298		errorCode, _ := event.Data["error_code"].(string)
288	299		return &RequestFact{TurnID: event.TurnID, Provider: providerName, Model: model, Attempt: attempt, MaxAttempts: maxAttempts, Outcome: outcome, ErrorCode: errorCode, StartedAt: event.Timestamp, EndedAt: event.Timestamp}
289	300		}

 			name, _ := event.Data["name"].(string)
 			provider, _ := event.Data["provider"].(string)
 			model, _ := event.Data["model"].(string)
-			attempt, _ := event.Data["attempt"].(int)
+			attempt := traceInt(event.Data["attempt"])
-			maxAttempts, _ := event.Data["max_attempts"].(int)
+			maxAttempts := traceInt(event.Data["max_attempts"])
 			open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, CallID: callID, Name: name, Provider: provider, Model: model, Attempt: attempt, MaxAttempts: maxAttempts, StartedAt: event.Timestamp}
 		}
 		if ends {
 	return nil
 }
+func traceInt(value any) int {
+	switch value := value.(type) {
+	case int:
+		return value
+	case float64:
+		return int(value)
+	default:
+		return 0
+	}
+}
 func requestFact(event TraceEvent, outcome string) *RequestFact {
 	providerName, _ := event.Data["provider"].(string)
 	model, _ := event.Data["model"].(string)
-	attempt, _ := event.Data["attempt"].(int)
+	attempt := traceInt(event.Data["attempt"])
-	maxAttempts, _ := event.Data["max_attempts"].(int)
+	maxAttempts := traceInt(event.Data["max_attempts"])
 	errorCode, _ := event.Data["error_code"].(string)
 	return &RequestFact{TurnID: event.TurnID, Provider: providerName, Model: model, Attempt: attempt, MaxAttempts: maxAttempts, Outcome: outcome, ErrorCode: errorCode, StartedAt: event.Timestamp, EndedAt: event.Timestamp}
 }

M packages/agent/subagents/projection_test.go +21 -0

@@ -1,11 +1,32 @@
1	1		package subagents
2	2
3	3		import (
	4	+	"os"
	5	+	"path/filepath"
4	6		"reflect"
5	7		"testing"
6	8		"time"
7	9		)
8	10
	11	+	func TestProjectTraceRetainsJSONDecodedRequestAttempts(t *testing.T) {
	12	+	bundle := t.TempDir()
	13	+	if err := os.WriteFile(filepath.Join(bundle, "manifest.json"), []byte(`{"version":1,"trace_file":"trace.jsonl"}`), 0o600); err != nil {
	14	+	t.Fatal(err)
	15	+	}
	16	+	line := `{"seq":1,"timestamp":"2026-08-12T07:00:00Z","type":"provider.request.started","agent_id":"agent-1","turn_id":"turn-1","data":{"attempt":2,"max_attempts":6}}` + "\n"
	17	+	if err := os.WriteFile(filepath.Join(bundle, "trace.jsonl"), []byte(line), 0o600); err != nil {
	18	+	t.Fatal(err)
	19	+	}
	20	+	events, err := ReadTrace(bundle)
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	operation := ProjectTrace(events)["agent-1"].PrimaryOperation
	25	+	if operation == nil \|\| operation.Attempt != 2 \|\| operation.MaxAttempts != 6 {
	26	+	t.Fatalf("projected operation = %#v, want attempt 2/6", operation)
	27	+	}
	28	+	}
	29	+
9	30		func TestProjectTraceReportsOpenOperationInsteadOfLifecycleState(t *testing.T) {
10	31		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
11	32		views := ProjectTrace([]TraceEvent{

@@ -1,11 +1,32 @@
1	package subagents	1		package subagents
2		2
3	import (	3		import (
		4	+	"os"
		5	+	"path/filepath"
4	"reflect"	6		"reflect"
5	"testing"	7		"testing"
6	"time"	8		"time"
7	)	9		)
8		10
		11	+	func TestProjectTraceRetainsJSONDecodedRequestAttempts(t *testing.T) {
		12	+	bundle := t.TempDir()
		13	+	if err := os.WriteFile(filepath.Join(bundle, "manifest.json"), []byte(`{"version":1,"trace_file":"trace.jsonl"}`), 0o600); err != nil {
		14	+	t.Fatal(err)
		15	+	}
		16	+	line := `{"seq":1,"timestamp":"2026-08-12T07:00:00Z","type":"provider.request.started","agent_id":"agent-1","turn_id":"turn-1","data":{"attempt":2,"max_attempts":6}}` + "\n"
		17	+	if err := os.WriteFile(filepath.Join(bundle, "trace.jsonl"), []byte(line), 0o600); err != nil {
		18	+	t.Fatal(err)
		19	+	}
		20	+	events, err := ReadTrace(bundle)
		21	+	if err != nil {
		22	+	t.Fatal(err)
		23	+	}
		24	+	operation := ProjectTrace(events)["agent-1"].PrimaryOperation
		25	+	if operation == nil \|\| operation.Attempt != 2 \|\| operation.MaxAttempts != 6 {
		26	+	t.Fatalf("projected operation = %#v, want attempt 2/6", operation)
		27	+	}
		28	+	}
		29	+
9	func TestProjectTraceReportsOpenOperationInsteadOfLifecycleState(t *testing.T) {	30		func TestProjectTraceReportsOpenOperationInsteadOfLifecycleState(t *testing.T) {
10	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)	31		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
11	views := ProjectTrace([]TraceEvent{	32		views := ProjectTrace([]TraceEvent{

M packages/agent/subagents/requirement.go +4 -0


@@ -2,6 +2,7 @@ package subagents
2	2
3	3		import (
4	4		"context"
	5	+	"errors"
5	6		"fmt"
6	7		"sort"
7	8		"strconv"

@@ -310,6 +311,9 @@ func (f *Supervisor) MarkRequirementNotified(id string) error {
310	311		if a == nil {
311	312		return fmt.Errorf("subagents: no such agent %q", id)
312	313		}
	314	+	if strings.TrimSpace(a.Snapshot().ResultRef) == "" {
	315	+	return errors.New("subagents: required result is not durably available")
	316	+	}
313	317		if a.markRequirementNotified() {
314	318		a.recordTrace(TraceEvent{Type: "result.delivered", TurnID: a.CurrentTurnID(), Data: map[string]any{"ref": ResultRef(a.ID)}})
315	319		}


@@ -2,6 +2,7 @@ package subagents
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"errors"
5	"fmt"	6		"fmt"
6	"sort"	7		"sort"
7	"strconv"	8		"strconv"

@@ -310,6 +311,9 @@ func (f *Supervisor) MarkRequirementNotified(id string) error {
310	if a == nil {	311		if a == nil {
311	return fmt.Errorf("subagents: no such agent %q", id)	312		return fmt.Errorf("subagents: no such agent %q", id)
312	}	313		}
		314	+	if strings.TrimSpace(a.Snapshot().ResultRef) == "" {
		315	+	return errors.New("subagents: required result is not durably available")
		316	+	}
313	if a.markRequirementNotified() {	317		if a.markRequirementNotified() {
314	a.recordTrace(TraceEvent{Type: "result.delivered", TurnID: a.CurrentTurnID(), Data: map[string]any{"ref": ResultRef(a.ID)}})	318		a.recordTrace(TraceEvent{Type: "result.delivered", TurnID: a.CurrentTurnID(), Data: map[string]any{"ref": ResultRef(a.ID)}})
315	}	319		}

M packages/agent/subagents/requirement_test.go +19 -1


@@ -221,7 +221,7 @@ func TestRequiredOutcomePersistsAcrossReload(t *testing.T) {
221	221		func TestRequirementNotificationRecordsDeliveryExactlyOnce(t *testing.T) {
222	222		trace := NewMemoryTraceWriter()
223	223		t.Cleanup(func() { _ = trace.Close() })
224		-	a := &Agent{ID: "agent-1", trace: trace, requirement: RequirementSnapshot{Required: true, State: RequirementSatisfied, TargetTurn: 1}}
	224	+	a := &Agent{ID: "agent-1", trace: trace, resultRef: ResultRef("agent-1"), requirement: RequirementSnapshot{Required: true, State: RequirementSatisfied, TargetTurn: 1}}
225	225		s := &Supervisor{agents: map[string]*Agent{a.ID: a}}
226	226		if err := s.MarkRequirementNotified(a.ID); err != nil {
227	227		t.Fatal(err)

@@ -243,6 +243,24 @@ func TestRequirementNotificationRecordsDeliveryExactlyOnce(t *testing.T) {
243	243		}
244	244		}
245	245
	246	+	func TestRequirementNotificationRejectsMissingDurableResult(t *testing.T) {
	247	+	trace := NewMemoryTraceWriter()
	248	+	t.Cleanup(func() { _ = trace.Close() })
	249	+	a := &Agent{ID: "agent-1", trace: trace, requirement: RequirementSnapshot{Required: true, State: RequirementSatisfied, TargetTurn: 1}}
	250	+	s := &Supervisor{agents: map[string]*Agent{a.ID: a}}
	251	+	if err := s.MarkRequirementNotified(a.ID); err == nil {
	252	+	t.Fatal("MarkRequirementNotified succeeded without durable result")
	253	+	}
	254	+	if a.Snapshot().Requirement.Notified {
	255	+	t.Fatal("requirement marked notified without durable result")
	256	+	}
	257	+	for _, event := range trace.Events() {
	258	+	if event.Type == "result.delivered" {
	259	+	t.Fatalf("unexpected delivery trace: %#v", event)
	260	+	}
	261	+	}
	262	+	}
	263	+
246	264		func TestRequirementNotificationPersistsAndRetryClearsIt(t *testing.T) {
247	265		root := t.TempDir()
248	266		first := New(Config{


@@ -221,7 +221,7 @@ func TestRequiredOutcomePersistsAcrossReload(t *testing.T) {
221		func TestRequirementNotificationRecordsDeliveryExactlyOnce(t *testing.T) {	221		func TestRequirementNotificationRecordsDeliveryExactlyOnce(t *testing.T) {
222		trace := NewMemoryTraceWriter()	222		trace := NewMemoryTraceWriter()
223		t.Cleanup(func() { _ = trace.Close() })	223		t.Cleanup(func() { _ = trace.Close() })
224	-	a := &Agent{ID: "agent-1", trace: trace, requirement: RequirementSnapshot{Required: true, State: RequirementSatisfied, TargetTurn: 1}}	224	+	a := &Agent{ID: "agent-1", trace: trace, resultRef: ResultRef("agent-1"), requirement: RequirementSnapshot{Required: true, State: RequirementSatisfied, TargetTurn: 1}}
225		s := &Supervisor{agents: map[string]*Agent{a.ID: a}}	225		s := &Supervisor{agents: map[string]*Agent{a.ID: a}}
226		if err := s.MarkRequirementNotified(a.ID); err != nil {	226		if err := s.MarkRequirementNotified(a.ID); err != nil {
227		t.Fatal(err)	227		t.Fatal(err)

@@ -243,6 +243,24 @@ func TestRequirementNotificationRecordsDeliveryExactlyOnce(t *testing.T) {
243		}	243		}
244		}	244		}
245			245
			246	+	func TestRequirementNotificationRejectsMissingDurableResult(t *testing.T) {
			247	+	trace := NewMemoryTraceWriter()
			248	+	t.Cleanup(func() { _ = trace.Close() })
			249	+	a := &Agent{ID: "agent-1", trace: trace, requirement: RequirementSnapshot{Required: true, State: RequirementSatisfied, TargetTurn: 1}}
			250	+	s := &Supervisor{agents: map[string]*Agent{a.ID: a}}
			251	+	if err := s.MarkRequirementNotified(a.ID); err == nil {
			252	+	t.Fatal("MarkRequirementNotified succeeded without durable result")
			253	+	}
			254	+	if a.Snapshot().Requirement.Notified {
			255	+	t.Fatal("requirement marked notified without durable result")
			256	+	}
			257	+	for _, event := range trace.Events() {
			258	+	if event.Type == "result.delivered" {
			259	+	t.Fatalf("unexpected delivery trace: %#v", event)
			260	+	}
			261	+	}
			262	+	}
			263	+
246		func TestRequirementNotificationPersistsAndRetryClearsIt(t *testing.T) {	264		func TestRequirementNotificationPersistsAndRetryClearsIt(t *testing.T) {
247		root := t.TempDir()	265		root := t.TempDir()
248		first := New(Config{	266		first := New(Config{

M packages/core/agent.go +22 -11


@@ -100,6 +100,7 @@ type Agent struct {
100	100		// StreamIdleTimeout cancels an individual provider stream when it emits no
101	101		// events for this duration. Zero leaves the stream unbounded.
102	102		StreamIdleTimeout time.Duration
	103	+	streamIdleTimer func(time.Duration) (<-chan time.Time, func(), func())
103	104
104	105		// OnEvent, if set, mirrors every AgentEvent the loop emits to
105	106		// this callback in addition to the per-Prompt sink. Used by the

@@ -871,6 +872,21 @@ func retryReasonFromProvider(reason provider.RequestFailureReason) RetryReason {
871	872
872	873		// oneTurn calls the LLM once, forwards events, returns the stop reason
873	874		// and the assembled assistant message (already appended to the transcript).
	875	+	func newStreamIdleTimer(timeout time.Duration) (<-chan time.Time, func(), func()) {
	876	+	timer := time.NewTimer(timeout)
	877	+	stop := func() { timer.Stop() }
	878	+	reset := func() {
	879	+	if !timer.Stop() {
	880	+	select {
	881	+	case <-timer.C:
	882	+	default:
	883	+	}
	884	+	}
	885	+	timer.Reset(timeout)
	886	+	}
	887	+	return timer.C, stop, reset
	888	+	}
	889	+
874	890		func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext string, attempt, maxAttempts int) (provider.StopReason, provider.Message, error) {
875	891		fastMode := a.FastModeEnabled()
876	892		if err := provider.ValidateFastMode(a.Client.Name(), fastMode); err != nil {

@@ -939,18 +955,13 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext
939	955		var idle <-chan time.Time
940	956		var resetIdle func()
941	957		if a.StreamIdleTimeout > 0 {
942		-	timer := time.NewTimer(a.StreamIdleTimeout)
943		-	defer timer.Stop()
944		-	idle = timer.C
945		-	resetIdle = func() {
946		-	if !timer.Stop() {
947		-	select {
948		-	case <-timer.C:
949		-	default:
950		-	}
951		-	}
952		-	timer.Reset(a.StreamIdleTimeout)
	958	+	newTimer := a.streamIdleTimer
	959	+	if newTimer == nil {
	960	+	newTimer = newStreamIdleTimer
953	961		}
	962	+	var stopIdle func()
	963	+	idle, stopIdle, resetIdle = newTimer(a.StreamIdleTimeout)
	964	+	defer stopIdle()
954	965		}
955	966		handleStreamEvent := func(ev provider.Event) (done bool) {
956	967		if resetIdle != nil {

 	// StreamIdleTimeout cancels an individual provider stream when it emits no
 	// events for this duration. Zero leaves the stream unbounded.
 	StreamIdleTimeout time.Duration
+	streamIdleTimer   func(time.Duration) (<-chan time.Time, func(), func())
 	// OnEvent, if set, mirrors every AgentEvent the loop emits to
 	// this callback in addition to the per-Prompt sink. Used by the
 // oneTurn calls the LLM once, forwards events, returns the stop reason
 // and the assembled assistant message (already appended to the transcript).
+func newStreamIdleTimer(timeout time.Duration) (<-chan time.Time, func(), func()) {
+	timer := time.NewTimer(timeout)
+	stop := func() { timer.Stop() }
+	reset := func() {
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+		timer.Reset(timeout)
+	}
+	return timer.C, stop, reset
+}
 func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext string, attempt, maxAttempts int) (provider.StopReason, provider.Message, error) {
 	fastMode := a.FastModeEnabled()
 	if err := provider.ValidateFastMode(a.Client.Name(), fastMode); err != nil {
 	var idle <-chan time.Time
 	var resetIdle func()
 	if a.StreamIdleTimeout > 0 {
-		timer := time.NewTimer(a.StreamIdleTimeout)
+		newTimer := a.streamIdleTimer
-		defer timer.Stop()
+		if newTimer == nil {
-		idle = timer.C
+			newTimer = newStreamIdleTimer
-		resetIdle = func() {
-			if !timer.Stop() {
-				select {
-				case <-timer.C:
-				default:
-				}
-			}
-			timer.Reset(a.StreamIdleTimeout)
 		}
+		var stopIdle func()
+		idle, stopIdle, resetIdle = newTimer(a.StreamIdleTimeout)
+		defer stopIdle()
 	}
 	handleStreamEvent := func(ev provider.Event) (done bool) {
 		if resetIdle != nil {

M packages/core/agent_retry_test.go +6 -1


@@ -86,7 +86,12 @@ func TestAgentStopsReadingStreamAfterCancellation(t *testing.T) {
86	86		func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
87	87		a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})
88	88		a.MaxRetries = 0
89		-	a.StreamIdleTimeout = 10 * time.Millisecond
	89	+	a.StreamIdleTimeout = time.Minute
	90	+	idle := make(chan time.Time, 1)
	91	+	a.streamIdleTimer = func(time.Duration) (<-chan time.Time, func(), func()) {
	92	+	idle <- time.Now()
	93	+	return idle, func() {}, func() {}
	94	+	}
90	95		if err := a.Prompt(context.Background(), "hello", nil, nil); !errors.Is(err, ErrStreamIdleTimeout) {
91	96		t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
92	97		}

 func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
 	a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})
 	a.MaxRetries = 0
-	a.StreamIdleTimeout = 10 * time.Millisecond
+	a.StreamIdleTimeout = time.Minute
+	idle := make(chan time.Time, 1)
+	a.streamIdleTimer = func(time.Duration) (<-chan time.Time, func(), func()) {
+		idle <- time.Now()
+		return idle, func() {}, func() {}
+	}
 	if err := a.Prompt(context.Background(), "hello", nil, nil); !errors.Is(err, ErrStreamIdleTimeout) {
 		t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
 	}