87f6ee8 fix(subagents): reconnect idle provider streams · zut

M packages/agent/modes/subagent_indicator_test.go +3 -3


@@ -21,7 +21,7 @@ func TestRenderSubagentActivityLinesShowsOnlyTraceOpenOperations(t *testing.T) {
21	21		"alive": {AgentID: "alive", LastEvent: subagents.TraceEvent{Type: "agent.ready", Timestamp: now.Add(-time.Second)}},
22	22		}
23	23		got := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", snapshots, views, 80, now))
24		-	want := []string{" / reviewer · tool open · 1m12s"}
	24	+	want := []string{" / reviewer · running tool · 1m12s"}
25	25		if strings.Join(got, "\n") != strings.Join(want, "\n") {
26	26		t.Fatalf("activity lines = %#v, want %#v", got, want)
27	27		}

@@ -32,7 +32,7 @@ func TestRenderSubagentActivityLinesShowsSafeLastObservation(t *testing.T) {
32	32		lines := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", []subagents.AgentSnapshot{{ID: "review-123", Subagent: "reviewer"}}, map[string]subagents.AgentTraceView{
33	33		"review-123": {PrimaryOperation: &subagents.Operation{Type: "turn.started", TurnID: "turn-1", StartedAt: now.Add(-time.Minute)}, LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)}},
34	34		}, 100, now))
35		-	if got := strings.Join(lines, "\n"); !strings.Contains(got, "turn open · assistant streaming 1s ago") \|\| strings.Contains(got, "delta") {
	35	+	if got := strings.Join(lines, "\n"); !strings.Contains(got, "processing turn · assistant streaming 1s ago") \|\| strings.Contains(got, "delta") {
36	36		t.Fatalf("activity lines = %#v", lines)
37	37		}
38	38		}

@@ -42,7 +42,7 @@ func TestRenderSubagentActivityLinesOmitsPriorTurnObservation(t *testing.T) {
42	42		lines := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", []subagents.AgentSnapshot{{ID: "review-123", Subagent: "reviewer"}}, map[string]subagents.AgentTraceView{
43	43		"review-123": {PrimaryOperation: &subagents.Operation{Type: "tool.started", TurnID: "turn-2", StartedAt: now.Add(-time.Minute)}, LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)}},
44	44		}, 100, now))
45		-	if got := strings.Join(lines, "\n"); strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "tool open") \|\| !strings.Contains(got, "1m") {
	45	+	if got := strings.Join(lines, "\n"); strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "running tool") \|\| !strings.Contains(got, "1m") {
46	46		t.Fatalf("activity lines = %#v", lines)
47	47		}
48	48		}


@@ -21,7 +21,7 @@ func TestRenderSubagentActivityLinesShowsOnlyTraceOpenOperations(t *testing.T) {
21		"alive": {AgentID: "alive", LastEvent: subagents.TraceEvent{Type: "agent.ready", Timestamp: now.Add(-time.Second)}},	21		"alive": {AgentID: "alive", LastEvent: subagents.TraceEvent{Type: "agent.ready", Timestamp: now.Add(-time.Second)}},
22		}	22		}
23		got := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", snapshots, views, 80, now))	23		got := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", snapshots, views, 80, now))
24	-	want := []string{" / reviewer · tool open · 1m12s"}	24	+	want := []string{" / reviewer · running tool · 1m12s"}
25		if strings.Join(got, "\n") != strings.Join(want, "\n") {	25		if strings.Join(got, "\n") != strings.Join(want, "\n") {
26		t.Fatalf("activity lines = %#v, want %#v", got, want)	26		t.Fatalf("activity lines = %#v, want %#v", got, want)
27		}	27		}

@@ -32,7 +32,7 @@ func TestRenderSubagentActivityLinesShowsSafeLastObservation(t *testing.T) {
32		lines := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", []subagents.AgentSnapshot{{ID: "review-123", Subagent: "reviewer"}}, map[string]subagents.AgentTraceView{	32		lines := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", []subagents.AgentSnapshot{{ID: "review-123", Subagent: "reviewer"}}, map[string]subagents.AgentTraceView{
33		"review-123": {PrimaryOperation: &subagents.Operation{Type: "turn.started", TurnID: "turn-1", StartedAt: now.Add(-time.Minute)}, LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)}},	33		"review-123": {PrimaryOperation: &subagents.Operation{Type: "turn.started", TurnID: "turn-1", StartedAt: now.Add(-time.Minute)}, LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)}},
34		}, 100, now))	34		}, 100, now))
35	-	if got := strings.Join(lines, "\n"); !strings.Contains(got, "turn open · assistant streaming 1s ago") \|\| strings.Contains(got, "delta") {	35	+	if got := strings.Join(lines, "\n"); !strings.Contains(got, "processing turn · assistant streaming 1s ago") \|\| strings.Contains(got, "delta") {
36		t.Fatalf("activity lines = %#v", lines)	36		t.Fatalf("activity lines = %#v", lines)
37		}	37		}
38		}	38		}

@@ -42,7 +42,7 @@ func TestRenderSubagentActivityLinesOmitsPriorTurnObservation(t *testing.T) {
42		lines := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", []subagents.AgentSnapshot{{ID: "review-123", Subagent: "reviewer"}}, map[string]subagents.AgentTraceView{	42		lines := plainActivityLines(renderSubagentActivityLines(tui.Dark, "/", []subagents.AgentSnapshot{{ID: "review-123", Subagent: "reviewer"}}, map[string]subagents.AgentTraceView{
43		"review-123": {PrimaryOperation: &subagents.Operation{Type: "tool.started", TurnID: "turn-2", StartedAt: now.Add(-time.Minute)}, LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)}},	43		"review-123": {PrimaryOperation: &subagents.Operation{Type: "tool.started", TurnID: "turn-2", StartedAt: now.Add(-time.Minute)}, LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)}},
44		}, 100, now))	44		}, 100, now))
45	-	if got := strings.Join(lines, "\n"); strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "tool open") \|\| !strings.Contains(got, "1m") {	45	+	if got := strings.Join(lines, "\n"); strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "running tool") \|\| !strings.Contains(got, "1m") {
46		t.Fatalf("activity lines = %#v", lines)	46		t.Fatalf("activity lines = %#v", lines)
47		}	47		}
48		}	48		}

M packages/agent/modes/subagents_dialog_test.go +4 -4


@@ -41,7 +41,7 @@ func TestFormatSupervisorRowOmitsPriorTurnObservationAge(t *testing.T) {
41	41		PrimaryOperation: &subagents.Operation{Type: "tool.started", TurnID: "turn-2", StartedAt: now.Add(-time.Minute)},
42	42		LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)},
43	43		}, 120)
44		-	if strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "tool open") \|\| !strings.Contains(got, "1m") {
	44	+	if strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "running tool") \|\| !strings.Contains(got, "1m") {
45	45		t.Fatalf("row = %q", got)
46	46		}
47	47		}

@@ -88,7 +88,7 @@ func TestSubagentsDialogRendersRows(t *testing.T) {
88	88		})
89	89		d.Open(staticSnapshots(rows...), nil, nil, nil, nil, nil, "")
90	90		out := strings.Join(d.Render(tui.Theme{}, 100), "\n")
91		-	for _, want := range []string{"alpha-1", "beta-2", "tool open", "completed", "OPERATION", "TASK"} {
	91	+	for _, want := range []string{"alpha-1", "beta-2", "running tool", "completed", "OPERATION", "TASK"} {
92	92		if !strings.Contains(out, want) {
93	93		t.Fatalf("render missing %q:\n%s", want, out)
94	94		}

@@ -1350,7 +1350,7 @@ func TestSupervisorTranscriptBusySpinnerRenders(t *testing.T) {
1350	1350		d.HandleKey(tui.Key{Kind: tui.KeyEnter})
1351	1351
1352	1352		out := strings.Join(d.Render(tui.Theme{}, 80), "\n")
1353		-	if !strings.Contains(out, "tool open") {
	1353	+	if !strings.Contains(out, "running tool") {
1354	1354		t.Fatalf("busy spinner did not surface an open tool near the editor:\n%s", out)
1355	1355		}
1356	1356		if d.transcriptSpin == nil {

@@ -1363,7 +1363,7 @@ func TestSupervisorTranscriptBusySpinnerRenders(t *testing.T) {
1363	1363		if d.transcriptSpin != nil {
1364	1364		t.Error("spinner stayed live after agent reported idle")
1365	1365		}
1366		-	if strings.Contains(out, "tool open, ") {
	1366	+	if strings.Contains(out, "running tool, ") {
1367	1367		t.Errorf("busy line still rendered after operation finished:\n%s", out)
1368	1368		}
1369	1369		}


@@ -41,7 +41,7 @@ func TestFormatSupervisorRowOmitsPriorTurnObservationAge(t *testing.T) {
41		PrimaryOperation: &subagents.Operation{Type: "tool.started", TurnID: "turn-2", StartedAt: now.Add(-time.Minute)},	41		PrimaryOperation: &subagents.Operation{Type: "tool.started", TurnID: "turn-2", StartedAt: now.Add(-time.Minute)},
42		LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)},	42		LastObservation: &subagents.LiveObservation{Type: "assistant.stream.observed", TurnID: "turn-1", At: now.Add(-time.Second)},
43		}, 120)	43		}, 120)
44	-	if strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "tool open") \|\| !strings.Contains(got, "1m") {	44	+	if strings.Contains(got, "assistant streaming") \|\| !strings.Contains(got, "running tool") \|\| !strings.Contains(got, "1m") {
45		t.Fatalf("row = %q", got)	45		t.Fatalf("row = %q", got)
46		}	46		}
47		}	47		}

@@ -88,7 +88,7 @@ func TestSubagentsDialogRendersRows(t *testing.T) {
88		})	88		})
89		d.Open(staticSnapshots(rows...), nil, nil, nil, nil, nil, "")	89		d.Open(staticSnapshots(rows...), nil, nil, nil, nil, nil, "")
90		out := strings.Join(d.Render(tui.Theme{}, 100), "\n")	90		out := strings.Join(d.Render(tui.Theme{}, 100), "\n")
91	-	for _, want := range []string{"alpha-1", "beta-2", "tool open", "completed", "OPERATION", "TASK"} {	91	+	for _, want := range []string{"alpha-1", "beta-2", "running tool", "completed", "OPERATION", "TASK"} {
92		if !strings.Contains(out, want) {	92		if !strings.Contains(out, want) {
93		t.Fatalf("render missing %q:\n%s", want, out)	93		t.Fatalf("render missing %q:\n%s", want, out)
94		}	94		}

@@ -1350,7 +1350,7 @@ func TestSupervisorTranscriptBusySpinnerRenders(t *testing.T) {
1350		d.HandleKey(tui.Key{Kind: tui.KeyEnter})	1350		d.HandleKey(tui.Key{Kind: tui.KeyEnter})
1351			1351
1352		out := strings.Join(d.Render(tui.Theme{}, 80), "\n")	1352		out := strings.Join(d.Render(tui.Theme{}, 80), "\n")
1353	-	if !strings.Contains(out, "tool open") {	1353	+	if !strings.Contains(out, "running tool") {
1354		t.Fatalf("busy spinner did not surface an open tool near the editor:\n%s", out)	1354		t.Fatalf("busy spinner did not surface an open tool near the editor:\n%s", out)
1355		}	1355		}
1356		if d.transcriptSpin == nil {	1356		if d.transcriptSpin == nil {

@@ -1363,7 +1363,7 @@ func TestSupervisorTranscriptBusySpinnerRenders(t *testing.T) {
1363		if d.transcriptSpin != nil {	1363		if d.transcriptSpin != nil {
1364		t.Error("spinner stayed live after agent reported idle")	1364		t.Error("spinner stayed live after agent reported idle")
1365		}	1365		}
1366	-	if strings.Contains(out, "tool open, ") {	1366	+	if strings.Contains(out, "running tool, ") {
1367		t.Errorf("busy line still rendered after operation finished:\n%s", out)	1367		t.Errorf("busy line still rendered after operation finished:\n%s", out)
1368		}	1368		}
1369		}	1369		}

M packages/agent/subagent_worker.go +5 -11


@@ -164,11 +164,6 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
164	164		runErr = joinSessionCloseError(runErr, sess)
165	165		}()
166	166		}
167		-	// A silent provider stream is neither progress nor a safe reason to leave
168		-	// a delegated worker occupied until its broad turn deadline. Set this after
169		-	// session resume because resume may rebuild the agent for another model.
170		-	ag.StreamIdleTimeout = time.Minute
171		-
172	167		// Resolve retains metadata for catalog models and for valid synthesized
173	168		// local/routed/open-catalog models. The active agent also survives a session
174	169		// resume that rebuilt it for a different provider/model pair.

@@ -256,13 +251,12 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
256	251		mu.Unlock()
257	252
258	253		turnStarted := false
259		-	toolStarted := false
260		-	ag.MaxRetries = 2
	254	+	// Match Codex's Responses stream policy: five minutes without an
	255	+	// event marks only the current stream as lost, and up to five retries
	256	+	// reconnect the sampling request without ending the delegated turn.
	257	+	ag.StreamIdleTimeout = 5 * time.Minute
	258	+	ag.MaxRetries = 5
261	259		sink := func(ev core.AgentEvent) {
262		-	if ev.Type() == "tool_execution_started" && !toolStarted {
263		-	toolStarted = true
264		-	ag.MaxRetries = 0
265		-	}
266	260		data := subagentEventData(ev)
267	261		if ev.Type() == "user_message" && !turnStarted {
268	262		// Prompt invokes OnMessageAppended before delivering this sink

 			runErr = joinSessionCloseError(runErr, sess)
 		}()
 	}
-	// A silent provider stream is neither progress nor a safe reason to leave
-	// a delegated worker occupied until its broad turn deadline. Set this after
-	// session resume because resume may rebuild the agent for another model.
-	ag.StreamIdleTimeout = time.Minute
 	// Resolve retains metadata for catalog models and for valid synthesized
 	// local/routed/open-catalog models. The active agent also survives a session
 	// resume that rebuilt it for a different provider/model pair.
 		mu.Unlock()
 		turnStarted := false
-		toolStarted := false
+		// Match Codex's Responses stream policy: five minutes without an
-		ag.MaxRetries = 2
+		// event marks only the current stream as lost, and up to five retries
+		// reconnect the sampling request without ending the delegated turn.
+		ag.StreamIdleTimeout = 5 * time.Minute
+		ag.MaxRetries = 5
 		sink := func(ev core.AgentEvent) {
-			if ev.Type() == "tool_execution_started" && !toolStarted {
-				toolStarted = true
-				ag.MaxRetries = 0
-			}
 			data := subagentEventData(ev)
 			if ev.Type() == "user_message" && !turnStarted {
 				// Prompt invokes OnMessageAppended before delivering this sink

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


@@ -34,13 +34,23 @@ func (o Operation) Duration(now time.Time) time.Duration {
34	34		return now.Sub(o.StartedAt)
35	35		}
36	36
37		-	// Label returns the shared concise label for an open operation.
	37	+	// Label returns a concise user-facing description of an open operation.
38	38		func (o Operation) Label() string {
39		-	label := strings.TrimSuffix(o.Type, ".started")
40		-	if o.Name != "" {
41		-	label += " " + o.Name
	39	+	switch o.Type {
	40	+	case "provider.request.started":
	41	+	return "waiting for model response"
	42	+	case "tool.started":
	43	+	if o.Name != "" {
	44	+	return "running " + o.Name
	45	+	}
	46	+	return "running tool"
	47	+	case "turn.started":
	48	+	return "processing turn"
	49	+	case "agent.wait.started":
	50	+	return "waiting"
	51	+	default:
	52	+	return strings.ReplaceAll(strings.TrimSuffix(o.Type, ".started"), ".", " ")
42	53		}
43		-	return label + " open"
44	54		}
45	55
46	56		// LiveObservation is the last safe, factual signal received from a worker.

 	return now.Sub(o.StartedAt)
 }
-// Label returns the shared concise label for an open operation.
+// Label returns a concise user-facing description of an open operation.
 func (o Operation) Label() string {
-	label := strings.TrimSuffix(o.Type, ".started")
+	switch o.Type {
-	if o.Name != "" {
+	case "provider.request.started":
-		label += " " + o.Name
+		return "waiting for model response"
+	case "tool.started":
+		if o.Name != "" {
+			return "running " + o.Name
+		}
+		return "running tool"
+	case "turn.started":
+		return "processing turn"
+	case "agent.wait.started":
+		return "waiting"
+	default:
+		return strings.ReplaceAll(strings.TrimSuffix(o.Type, ".started"), ".", " ")
 	}
-	return label + " open"
 }
 // LiveObservation is the last safe, factual signal received from a worker.

M packages/agent/subagents/projection_test.go +19 -3


@@ -75,11 +75,27 @@ func TestProjectTraceSelectsSpecificActivityAndProjectsResultDelivery(t *testing
75	75		if view.Result == nil \|\| !view.Result.Available \|\| !view.Result.Delivered \|\| view.Result.Ref != "subagent://agent-1/result" {
76	76		t.Fatalf("result = %#v", view.Result)
77	77		}
78		-	if got, want := view.Summary(), "tool bash open · assistant streaming"; got != want {
	78	+	if got, want := view.Summary(), "running bash · assistant streaming"; got != want {
79	79		t.Fatalf("summary = %q, want %q", got, want)
80	80		}
81	81		}
82	82
	83	+	func TestOperationLabelUsesHumanFacingActivity(t *testing.T) {
	84	+	cases := []struct {
	85	+	operation Operation
	86	+	want string
	87	+	}{
	88	+	{operation: Operation{Type: "provider.request.started"}, want: "waiting for model response"},
	89	+	{operation: Operation{Type: "tool.started", Name: "bash"}, want: "running bash"},
	90	+	{operation: Operation{Type: "turn.started"}, want: "processing turn"},
	91	+	}
	92	+	for _, tc := range cases {
	93	+	if got := tc.operation.Label(); got != tc.want {
	94	+	t.Fatalf("Label(%q) = %q, want %q", tc.operation.Type, got, tc.want)
	95	+	}
	96	+	}
	97	+	}
	98	+
83	99		func TestProjectTraceRetainsSafeProviderAttemptDiagnostic(t *testing.T) {
84	100		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
85	101		view := ProjectTrace([]TraceEvent{

@@ -108,7 +124,7 @@ func TestProjectTraceOmitsObservationForAnotherToolInSameTurn(t *testing.T) {
108	124		if observation := view.ObservationFor(*view.PrimaryOperation); observation != nil {
109	125		t.Fatalf("another tool's observation = %#v, want omitted", observation)
110	126		}
111		-	if got, want := view.Summary(), "tool bash open"; got != want {
	127	+	if got, want := view.Summary(), "running bash"; got != want {
112	128		t.Fatalf("summary = %q, want %q", got, want)
113	129		}
114	130		}

@@ -128,7 +144,7 @@ func TestProjectTraceOmitsPriorTurnObservation(t *testing.T) {
128	144		if observation := view.ObservationFor(*view.PrimaryOperation); observation != nil {
129	145		t.Fatalf("prior turn observation = %#v, want omitted", observation)
130	146		}
131		-	if got, want := view.Summary(), "turn open"; got != want {
	147	+	if got, want := view.Summary(), "processing turn"; got != want {
132	148		t.Fatalf("summary = %q, want %q", got, want)
133	149		}
134	150		}

 	if view.Result == nil || !view.Result.Available || !view.Result.Delivered || view.Result.Ref != "subagent://agent-1/result" {
 		t.Fatalf("result = %#v", view.Result)
 	}
-	if got, want := view.Summary(), "tool bash open · assistant streaming"; got != want {
+	if got, want := view.Summary(), "running bash · assistant streaming"; got != want {
 		t.Fatalf("summary = %q, want %q", got, want)
 	}
 }
+func TestOperationLabelUsesHumanFacingActivity(t *testing.T) {
+	cases := []struct {
+		operation Operation
+		want      string
+	}{
+		{operation: Operation{Type: "provider.request.started"}, want: "waiting for model response"},
+		{operation: Operation{Type: "tool.started", Name: "bash"}, want: "running bash"},
+		{operation: Operation{Type: "turn.started"}, want: "processing turn"},
+	}
+	for _, tc := range cases {
+		if got := tc.operation.Label(); got != tc.want {
+			t.Fatalf("Label(%q) = %q, want %q", tc.operation.Type, got, tc.want)
+		}
+	}
+}
 func TestProjectTraceRetainsSafeProviderAttemptDiagnostic(t *testing.T) {
 	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
 	view := ProjectTrace([]TraceEvent{
 	if observation := view.ObservationFor(*view.PrimaryOperation); observation != nil {
 		t.Fatalf("another tool's observation = %#v, want omitted", observation)
 	}
-	if got, want := view.Summary(), "tool bash open"; got != want {
+	if got, want := view.Summary(), "running bash"; got != want {
 		t.Fatalf("summary = %q, want %q", got, want)
 	}
 }
 	if observation := view.ObservationFor(*view.PrimaryOperation); observation != nil {
 		t.Fatalf("prior turn observation = %#v, want omitted", observation)
 	}
-	if got, want := view.Summary(), "turn open"; got != want {
+	if got, want := view.Summary(), "processing turn"; got != want {
 		t.Fatalf("summary = %q, want %q", got, want)
 	}
 }

M packages/agent/tools/subagent_status.go +1 -1


@@ -184,7 +184,7 @@ func publicSubagentStatus(snapshot subagents.AgentSnapshot, view subagents.Agent
184	184		if snapshot.Result != nil {
185	185		entry.Result = &subagentStatusResult{
186	186		State: publicResultState(snapshot.Result.Status),
187		-	Available: true,
	187	+	Available: snapshot.ResultRef != "",
188	188		Ref: snapshot.ResultRef,
189	189		}
190	190		}


@@ -184,7 +184,7 @@ func publicSubagentStatus(snapshot subagents.AgentSnapshot, view subagents.Agent
184		if snapshot.Result != nil {	184		if snapshot.Result != nil {
185		entry.Result = &subagentStatusResult{	185		entry.Result = &subagentStatusResult{
186		State: publicResultState(snapshot.Result.Status),	186		State: publicResultState(snapshot.Result.Status),
187	-	Available: true,	187	+	Available: snapshot.ResultRef != "",
188		Ref: snapshot.ResultRef,	188		Ref: snapshot.ResultRef,
189		}	189		}
190		}	190		}

M packages/agent/tools/subagent_status_test.go +10 -0


@@ -86,6 +86,16 @@ func TestPublicSubagentStatusUsesOnlyPrimaryOperation(t *testing.T) {
86	86		}
87	87		}
88	88
	89	+	func TestPublicSubagentStatusDoesNotInferAvailabilityWithoutResultReference(t *testing.T) {
	90	+	entry := publicSubagentStatus(subagents.AgentSnapshot{
	91	+	ID: "agent-1",
	92	+	Result: &subagents.TurnResult{Status: subagents.ResultFailed},
	93	+	}, subagents.AgentTraceView{})
	94	+	if entry.Result == nil \|\| entry.Result.Available {
	95	+	t.Fatalf("result availability = %#v, want unavailable without result ref", entry.Result)
	96	+	}
	97	+	}
	98	+
89	99		func TestPublicSubagentStatusIncludesResultDeliveryFacts(t *testing.T) {
90	100		entry := publicSubagentStatus(subagents.AgentSnapshot{ID: "agent-1"}, subagents.AgentTraceView{
91	101		Result: &subagents.ResultFact{Available: true, Ref: "subagent://agent-1/result"},


@@ -86,6 +86,16 @@ func TestPublicSubagentStatusUsesOnlyPrimaryOperation(t *testing.T) {
86	}	86		}
87	}	87		}
88		88
		89	+	func TestPublicSubagentStatusDoesNotInferAvailabilityWithoutResultReference(t *testing.T) {
		90	+	entry := publicSubagentStatus(subagents.AgentSnapshot{
		91	+	ID: "agent-1",
		92	+	Result: &subagents.TurnResult{Status: subagents.ResultFailed},
		93	+	}, subagents.AgentTraceView{})
		94	+	if entry.Result == nil \|\| entry.Result.Available {
		95	+	t.Fatalf("result availability = %#v, want unavailable without result ref", entry.Result)
		96	+	}
		97	+	}
		98	+
89	func TestPublicSubagentStatusIncludesResultDeliveryFacts(t *testing.T) {	99		func TestPublicSubagentStatusIncludesResultDeliveryFacts(t *testing.T) {
90	entry := publicSubagentStatus(subagents.AgentSnapshot{ID: "agent-1"}, subagents.AgentTraceView{	100		entry := publicSubagentStatus(subagents.AgentSnapshot{ID: "agent-1"}, subagents.AgentTraceView{
91	Result: &subagents.ResultFact{Available: true, Ref: "subagent://agent-1/result"},	101		Result: &subagents.ResultFact{Available: true, Ref: "subagent://agent-1/result"},

M packages/core/agent.go +35 -55


@@ -952,76 +952,56 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext
952	952		timer.Reset(a.StreamIdleTimeout)
953	953		}
954	954		}
	955	+	handleStreamEvent := func(ev provider.Event) (done bool) {
	956	+	if resetIdle != nil {
	957	+	resetIdle()
	958	+	}
	959	+	switch e := ev.(type) {
	960	+	case provider.EventStart:
	961	+	// nothing
	962	+	case provider.EventTextDelta:
	963	+	sink(EvTextDelta{Delta: e.Delta})
	964	+	case provider.EventToolStart:
	965	+	sink(EvToolUseStart{ID: e.ID, Name: e.Name})
	966	+	case provider.EventToolArgs:
	967	+	sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
	968	+	case provider.EventToolEnd:
	969	+	sink(EvToolUseEnd{ID: e.ID})
	970	+	case provider.EventUsage:
	971	+	cum := a.addUsage(e.Usage)
	972	+	sink(EvUsage{Usage: e.Usage, Cumulative: cum})
	973	+	if a.OnUsage != nil {
	974	+	a.OnUsage(cum)
	975	+	}
	976	+	case provider.EventDone:
	977	+	stop = e.Stop
	978	+	finalErr = e.Err
	979	+	finalMsg = e.Message
	980	+	return true
	981	+	}
	982	+	return false
	983	+	}
955	984		for {
956		-	// Prefer an already-buffered terminal provider event over a concurrent
957		-	// context cancellation so retry classification retains its real cause.
	985	+	// Consume no more than one buffered event before honoring cancellation.
958	986		select {
959	987		case ev, ok := <-stream:
960		-	if !ok {
	988	+	if !ok \|\| handleStreamEvent(ev) {
961	989		goto streamDone
962	990		}
963		-	if resetIdle != nil {
964		-	resetIdle()
965		-	}
966		-	switch e := ev.(type) {
967		-	case provider.EventStart:
968		-	// nothing
969		-	case provider.EventTextDelta:
970		-	sink(EvTextDelta{Delta: e.Delta})
971		-	case provider.EventToolStart:
972		-	sink(EvToolUseStart{ID: e.ID, Name: e.Name})
973		-	case provider.EventToolArgs:
974		-	sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
975		-	case provider.EventToolEnd:
976		-	sink(EvToolUseEnd{ID: e.ID})
977		-	case provider.EventUsage:
978		-	cum := a.addUsage(e.Usage)
979		-	sink(EvUsage{Usage: e.Usage, Cumulative: cum})
980		-	if a.OnUsage != nil {
981		-	a.OnUsage(cum)
982		-	}
983		-	case provider.EventDone:
984		-	stop = e.Stop
985		-	finalErr = e.Err
986		-	finalMsg = e.Message
987		-	}
988		-	continue
989	991		default:
990	992		}
	993	+	if err := streamCtx.Err(); err != nil {
	994	+	return provider.StopError, finalMsg, err
	995	+	}
991	996		select {
992	997		case <-streamCtx.Done():
993	998		return provider.StopError, finalMsg, streamCtx.Err()
994	999		case <-idle:
995	1000		return provider.StopError, finalMsg, ErrStreamIdleTimeout
996	1001		case ev, ok := <-stream:
997		-	if !ok {
	1002	+	if !ok \|\| handleStreamEvent(ev) {
998	1003		goto streamDone
999	1004		}
1000		-	if resetIdle != nil {
1001		-	resetIdle()
1002		-	}
1003		-	switch e := ev.(type) {
1004		-	case provider.EventStart:
1005		-	// nothing
1006		-	case provider.EventTextDelta:
1007		-	sink(EvTextDelta{Delta: e.Delta})
1008		-	case provider.EventToolStart:
1009		-	sink(EvToolUseStart{ID: e.ID, Name: e.Name})
1010		-	case provider.EventToolArgs:
1011		-	sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
1012		-	case provider.EventToolEnd:
1013		-	sink(EvToolUseEnd{ID: e.ID})
1014		-	case provider.EventUsage:
1015		-	cum := a.addUsage(e.Usage)
1016		-	sink(EvUsage{Usage: e.Usage, Cumulative: cum})
1017		-	if a.OnUsage != nil {
1018		-	a.OnUsage(cum)
1019		-	}
1020		-	case provider.EventDone:
1021		-	stop = e.Stop
1022		-	finalErr = e.Err
1023		-	finalMsg = e.Message
1024		-	}
1025	1005		}
1026	1006		}
1027	1007		streamDone:

 			timer.Reset(a.StreamIdleTimeout)
 		}
 	}
+	handleStreamEvent := func(ev provider.Event) (done bool) {
+		if resetIdle != nil {
+			resetIdle()
+		}
+		switch e := ev.(type) {
+		case provider.EventStart:
+			// nothing
+		case provider.EventTextDelta:
+			sink(EvTextDelta{Delta: e.Delta})
+		case provider.EventToolStart:
+			sink(EvToolUseStart{ID: e.ID, Name: e.Name})
+		case provider.EventToolArgs:
+			sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
+		case provider.EventToolEnd:
+			sink(EvToolUseEnd{ID: e.ID})
+		case provider.EventUsage:
+			cum := a.addUsage(e.Usage)
+			sink(EvUsage{Usage: e.Usage, Cumulative: cum})
+			if a.OnUsage != nil {
+				a.OnUsage(cum)
+			}
+		case provider.EventDone:
+			stop = e.Stop
+			finalErr = e.Err
+			finalMsg = e.Message
+			return true
+		}
+		return false
+	}
 	for {
-		// Prefer an already-buffered terminal provider event over a concurrent
+		// Consume no more than one buffered event before honoring cancellation.
-		// context cancellation so retry classification retains its real cause.
 		select {
 		case ev, ok := <-stream:
-			if !ok {
+			if !ok || handleStreamEvent(ev) {
 				goto streamDone
 			}
-			if resetIdle != nil {
-				resetIdle()
-			}
-			switch e := ev.(type) {
-			case provider.EventStart:
-				// nothing
-			case provider.EventTextDelta:
-				sink(EvTextDelta{Delta: e.Delta})
-			case provider.EventToolStart:
-				sink(EvToolUseStart{ID: e.ID, Name: e.Name})
-			case provider.EventToolArgs:
-				sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
-			case provider.EventToolEnd:
-				sink(EvToolUseEnd{ID: e.ID})
-			case provider.EventUsage:
-				cum := a.addUsage(e.Usage)
-				sink(EvUsage{Usage: e.Usage, Cumulative: cum})
-				if a.OnUsage != nil {
-					a.OnUsage(cum)
-				}
-			case provider.EventDone:
-				stop = e.Stop
-				finalErr = e.Err
-				finalMsg = e.Message
-			}
-			continue
 		default:
 		}
+		if err := streamCtx.Err(); err != nil {
+			return provider.StopError, finalMsg, err
+		}
 		select {
 		case <-streamCtx.Done():
 			return provider.StopError, finalMsg, streamCtx.Err()
 		case <-idle:
 			return provider.StopError, finalMsg, ErrStreamIdleTimeout
 		case ev, ok := <-stream:
-			if !ok {
+			if !ok || handleStreamEvent(ev) {
 				goto streamDone
 			}
-			if resetIdle != nil {
-				resetIdle()
-			}
-			switch e := ev.(type) {
-			case provider.EventStart:
-				// nothing
-			case provider.EventTextDelta:
-				sink(EvTextDelta{Delta: e.Delta})
-			case provider.EventToolStart:
-				sink(EvToolUseStart{ID: e.ID, Name: e.Name})
-			case provider.EventToolArgs:
-				sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
-			case provider.EventToolEnd:
-				sink(EvToolUseEnd{ID: e.ID})
-			case provider.EventUsage:
-				cum := a.addUsage(e.Usage)
-				sink(EvUsage{Usage: e.Usage, Cumulative: cum})
-				if a.OnUsage != nil {
-					a.OnUsage(cum)
-				}
-			case provider.EventDone:
-				stop = e.Stop
-				finalErr = e.Err
-				finalMsg = e.Message
-			}
 		}
 	}
 streamDone:

M packages/core/agent_retry_test.go +38 -0


@@ -61,6 +61,28 @@ func (silentStreamClient) Stream(ctx context.Context, _ provider.Request) (<-cha
61	61		return out, nil
62	62		}
63	63
	64	+	type readableAfterCancellationClient struct {
	65	+	cancel context.CancelFunc
	66	+	}
	67	+
	68	+	func (c readableAfterCancellationClient) Name() string { return "readable-after-cancellation" }
	69	+
	70	+	func (c readableAfterCancellationClient) Stream(_ context.Context, _ provider.Request) (<-chan provider.Event, error) {
	71	+	out := make(chan provider.Event, 1)
	72	+	out <- provider.EventStart{}
	73	+	c.cancel()
	74	+	return out, nil
	75	+	}
	76	+
	77	+	func TestAgentStopsReadingStreamAfterCancellation(t *testing.T) {
	78	+	ctx, cancel := context.WithCancel(context.Background())
	79	+	a := NewAgent(readableAfterCancellationClient{cancel: cancel}, "fake-model", "system", Registry{})
	80	+	a.MaxRetries = 0
	81	+	if err := a.Prompt(ctx, "hello", nil, nil); !errors.Is(err, context.Canceled) {
	82	+	t.Fatalf("Prompt error = %v, want context.Canceled", err)
	83	+	}
	84	+	}
	85	+
64	86		func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
65	87		a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})
66	88		a.MaxRetries = 0

@@ -70,6 +92,22 @@ func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
70	92		}
71	93		}
72	94
	95	+	func TestAgentRetriesSilentStreamWithoutEndingTurn(t *testing.T) {
	96	+	client := &retryFakeClient{firstErr: ErrStreamIdleTimeout}
	97	+	a := NewAgent(client, "fake-model", "system", Registry{})
	98	+	a.MaxRetries = 1
	99	+	a.RetryBaseDelay = time.Millisecond
	100	+	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
	101	+	t.Fatalf("Prompt returned %v", err)
	102	+	}
	103	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
	104	+	t.Fatalf("Stream calls = %d; want reconnect after idle timeout", got)
	105	+	}
	106	+	if got := extractText(a.Messages()[1]); got != "ok" {
	107	+	t.Fatalf("final assistant text = %q; want ok", got)
	108	+	}
	109	+	}
	110	+
73	111		func TestAgentRetriesOverloadedStreamError(t *testing.T) {
74	112		client := &retryFakeClient{}
75	113		a := NewAgent(client, "fake-model", "system", Registry{})


@@ -61,6 +61,28 @@ func (silentStreamClient) Stream(ctx context.Context, _ provider.Request) (<-cha
61	return out, nil	61		return out, nil
62	}	62		}
63		63
		64	+	type readableAfterCancellationClient struct {
		65	+	cancel context.CancelFunc
		66	+	}
		67	+
		68	+	func (c readableAfterCancellationClient) Name() string { return "readable-after-cancellation" }
		69	+
		70	+	func (c readableAfterCancellationClient) Stream(_ context.Context, _ provider.Request) (<-chan provider.Event, error) {
		71	+	out := make(chan provider.Event, 1)
		72	+	out <- provider.EventStart{}
		73	+	c.cancel()
		74	+	return out, nil
		75	+	}
		76	+
		77	+	func TestAgentStopsReadingStreamAfterCancellation(t *testing.T) {
		78	+	ctx, cancel := context.WithCancel(context.Background())
		79	+	a := NewAgent(readableAfterCancellationClient{cancel: cancel}, "fake-model", "system", Registry{})
		80	+	a.MaxRetries = 0
		81	+	if err := a.Prompt(ctx, "hello", nil, nil); !errors.Is(err, context.Canceled) {
		82	+	t.Fatalf("Prompt error = %v, want context.Canceled", err)
		83	+	}
		84	+	}
		85	+
64	func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {	86		func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
65	a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})	87		a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})
66	a.MaxRetries = 0	88		a.MaxRetries = 0

@@ -70,6 +92,22 @@ func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
70	}	92		}
71	}	93		}
72		94
		95	+	func TestAgentRetriesSilentStreamWithoutEndingTurn(t *testing.T) {
		96	+	client := &retryFakeClient{firstErr: ErrStreamIdleTimeout}
		97	+	a := NewAgent(client, "fake-model", "system", Registry{})
		98	+	a.MaxRetries = 1
		99	+	a.RetryBaseDelay = time.Millisecond
		100	+	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
		101	+	t.Fatalf("Prompt returned %v", err)
		102	+	}
		103	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
		104	+	t.Fatalf("Stream calls = %d; want reconnect after idle timeout", got)
		105	+	}
		106	+	if got := extractText(a.Messages()[1]); got != "ok" {
		107	+	t.Fatalf("final assistant text = %q; want ok", got)
		108	+	}
		109	+	}
		110	+
73	func TestAgentRetriesOverloadedStreamError(t *testing.T) {	111		func TestAgentRetriesOverloadedStreamError(t *testing.T) {
74	client := &retryFakeClient{}	112		client := &retryFakeClient{}
75	a := NewAgent(client, "fake-model", "system", Registry{})	113		a := NewAgent(client, "fake-model", "system", Registry{})