68a8513 fix(subagents): bound silent provider streams · zut

M packages/agent/modes/subagents_dialog.go +16 -0


@@ -1306,6 +1306,22 @@ func (d *subagentsDialog) renderTranscript(th tui.Theme, width int) []string {
1306	1306		" " + th.FGColor(th.Muted, "dir: "+a.Dir),
1307	1307		" " + th.FGColor(th.Muted, "observation: "+d.traceSummary(a.ID)),
1308	1308		}
	1309	+	if request := d.traceView(a.ID).LastRequest; request != nil {
	1310	+	diagnostic := "provider request: " + request.Outcome
	1311	+	if request.Provider != "" \|\| request.Model != "" {
	1312	+	diagnostic += " · " + strings.Trim(strings.TrimSpace(request.Provider+" / "+request.Model), " /")
	1313	+	}
	1314	+	if request.Attempt > 0 {
	1315	+	diagnostic += fmt.Sprintf(" · attempt %d", request.Attempt)
	1316	+	if request.MaxAttempts > 0 {
	1317	+	diagnostic += fmt.Sprintf("/%d", request.MaxAttempts)
	1318	+	}
	1319	+	}
	1320	+	if request.ErrorCode != "" {
	1321	+	diagnostic += " · " + request.ErrorCode
	1322	+	}
	1323	+	header = append(header, " "+th.FGColor(th.Muted, diagnostic))
	1324	+	}
1309	1325		if a.Model != "" {
1310	1326		modelLine := "model: " + a.Model
1311	1327		if a.Provider != "" {


@@ -1306,6 +1306,22 @@ func (d *subagentsDialog) renderTranscript(th tui.Theme, width int) []string {
1306	" " + th.FGColor(th.Muted, "dir: "+a.Dir),	1306		" " + th.FGColor(th.Muted, "dir: "+a.Dir),
1307	" " + th.FGColor(th.Muted, "observation: "+d.traceSummary(a.ID)),	1307		" " + th.FGColor(th.Muted, "observation: "+d.traceSummary(a.ID)),
1308	}	1308		}
		1309	+	if request := d.traceView(a.ID).LastRequest; request != nil {
		1310	+	diagnostic := "provider request: " + request.Outcome
		1311	+	if request.Provider != "" \|\| request.Model != "" {
		1312	+	diagnostic += " · " + strings.Trim(strings.TrimSpace(request.Provider+" / "+request.Model), " /")
		1313	+	}
		1314	+	if request.Attempt > 0 {
		1315	+	diagnostic += fmt.Sprintf(" · attempt %d", request.Attempt)
		1316	+	if request.MaxAttempts > 0 {
		1317	+	diagnostic += fmt.Sprintf("/%d", request.MaxAttempts)
		1318	+	}
		1319	+	}
		1320	+	if request.ErrorCode != "" {
		1321	+	diagnostic += " · " + request.ErrorCode
		1322	+	}
		1323	+	header = append(header, " "+th.FGColor(th.Muted, diagnostic))
		1324	+	}
1309	if a.Model != "" {	1325		if a.Model != "" {
1310	modelLine := "model: " + a.Model	1326		modelLine := "model: " + a.Model
1311	if a.Provider != "" {	1327		if a.Provider != "" {

M packages/agent/modes/subagents_dialog_test.go +14 -0


@@ -46,6 +46,20 @@ func TestFormatSupervisorRowOmitsPriorTurnObservationAge(t *testing.T) {
46	46		}
47	47		}
48	48
	49	+	func TestSubagentsDialogTranscriptShowsProviderRequestDiagnostic(t *testing.T) {
	50	+	at := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
	51	+	d := newSubagentsDialog()
	52	+	d.Open(staticSnapshots(subagents.AgentSnapshot{ID: "review-1", Task: "review", Started: at}), nil, nil, nil, nil, nil, "")
	53	+	d.SetTraceViews(func() map[string]subagents.AgentTraceView {
	54	+	return map[string]subagents.AgentTraceView{"review-1": {LastRequest: &subagents.RequestFact{Provider: "openai-codex", Model: "gpt-5.6", Attempt: 2, MaxAttempts: 3, Outcome: "failed", ErrorCode: "deadline_exceeded"}}}
	55	+	})
	56	+	d.viewing = true
	57	+	joined := strings.Join(d.Render(tui.Theme{}, 120), "\n")
	58	+	if !strings.Contains(joined, "provider request: failed · openai-codex / gpt-5.6 · attempt 2/3 · deadline_exceeded") {
	59	+	t.Fatalf("diagnostic missing:\n%s", joined)
	60	+	}
	61	+	}
	62	+
49	63		func TestSubagentsDialogEmptyState(t *testing.T) {
50	64		d := newSubagentsDialog()
51	65		d.Open(staticSnapshots(), nil, nil, nil, nil, nil, "")


@@ -46,6 +46,20 @@ func TestFormatSupervisorRowOmitsPriorTurnObservationAge(t *testing.T) {
46	}	46		}
47	}	47		}
48		48
		49	+	func TestSubagentsDialogTranscriptShowsProviderRequestDiagnostic(t *testing.T) {
		50	+	at := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
		51	+	d := newSubagentsDialog()
		52	+	d.Open(staticSnapshots(subagents.AgentSnapshot{ID: "review-1", Task: "review", Started: at}), nil, nil, nil, nil, nil, "")
		53	+	d.SetTraceViews(func() map[string]subagents.AgentTraceView {
		54	+	return map[string]subagents.AgentTraceView{"review-1": {LastRequest: &subagents.RequestFact{Provider: "openai-codex", Model: "gpt-5.6", Attempt: 2, MaxAttempts: 3, Outcome: "failed", ErrorCode: "deadline_exceeded"}}}
		55	+	})
		56	+	d.viewing = true
		57	+	joined := strings.Join(d.Render(tui.Theme{}, 120), "\n")
		58	+	if !strings.Contains(joined, "provider request: failed · openai-codex / gpt-5.6 · attempt 2/3 · deadline_exceeded") {
		59	+	t.Fatalf("diagnostic missing:\n%s", joined)
		60	+	}
		61	+	}
		62	+
49	func TestSubagentsDialogEmptyState(t *testing.T) {	63		func TestSubagentsDialogEmptyState(t *testing.T) {
50	d := newSubagentsDialog()	64		d := newSubagentsDialog()
51	d.Open(staticSnapshots(), nil, nil, nil, nil, nil, "")	65		d.Open(staticSnapshots(), nil, nil, nil, nil, nil, "")

M packages/agent/subagent_worker.go +18 -7


@@ -117,9 +117,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
117	117		if err != nil {
118	118		return err
119	119		}
120		-	extMgr, stopExt := setupNonInteractiveExtensions(ctx, args, &r, version)
121		-	defer stopExt()
122		-
	120	+	// A delegated worker has one bounded task and must not inherit arbitrary
	121	+	// local extensions, their tools, hooks, or persisted extension state.
	122	+	// Extensions remain available to explicitly selected skills only through
	123	+	// the parent process; the worker runs with its resolved built-in registry.
123	124		ag := r.NewAgent()
124	125		initialAg := ag
125	126		defer func() {

@@ -128,7 +129,6 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
128	129		closeAgentLSP(initialAg)
129	130		}
130	131		}()
131		-	wireNonInteractiveAgentExtHooks(ctx, ag, extMgr)
132	132		sess, err := openOrCreateSession(ctx, args, r, ag, version)
133	133		if err != nil {
134	134		return err

@@ -136,12 +136,11 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
136	136		var sessionPersistence sessionPersistenceState
137	137		if sess != nil {
138	138		var providerName, model string
139		-	sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, extMgr, sess, ag)
	139	+	sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, nil, sess, ag)
140	140		if err != nil {
141	141		return err
142	142		}
143	143		r.Provider, r.Model = providerName, model
144		-	ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) }
145	144		// A prompt is acknowledged by the supervisor only after this callback
146	145		// has appended and synced the matching user message. This closes the
147	146		// crash window between accepting a follow-up and writing session.json.

@@ -165,7 +164,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
165	164		runErr = joinSessionCloseError(runErr, sess)
166	165		}()
167	166		}
168		-	announceSession(extMgr, sess)
	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
169	171
170	172		// Resolve retains metadata for catalog models and for valid synthesized
171	173		// local/routed/open-catalog models. The active agent also survives a session

@@ -254,7 +256,13 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
254	256		mu.Unlock()
255	257
256	258		turnStarted := false
	259	+	toolStarted := false
	260	+	ag.MaxRetries = 2
257	261		sink := func(ev core.AgentEvent) {
	262	+	if ev.Type() == "tool_execution_started" && !toolStarted {
	263	+	toolStarted = true
	264	+	ag.MaxRetries = 0
	265	+	}
258	266		data := subagentEventData(ev)
259	267		if ev.Type() == "user_message" && !turnStarted {
260	268		// Prompt invokes OnMessageAppended before delivering this sink

@@ -707,6 +715,9 @@ func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[
707	715		if provider.IsContextOverflowError(err) {
708	716		return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
709	717		}
	718	+	if errors.Is(err, core.ErrStreamIdleTimeout) {
	719	+	return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for one minute; the delegated turn stopped safely"}
	720	+	}
710	721		if errors.Is(err, context.DeadlineExceeded) \|\| shutdownOrigin == subagents.ShutdownOriginDeadline {
711	722		return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}
712	723		}


@@ -117,9 +117,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
117		if err != nil {	117		if err != nil {
118		return err	118		return err
119		}	119		}
120	-	extMgr, stopExt := setupNonInteractiveExtensions(ctx, args, &r, version)	120	+	// A delegated worker has one bounded task and must not inherit arbitrary
121	-	defer stopExt()	121	+	// local extensions, their tools, hooks, or persisted extension state.
122	-		122	+	// Extensions remain available to explicitly selected skills only through
			123	+	// the parent process; the worker runs with its resolved built-in registry.
123		ag := r.NewAgent()	124		ag := r.NewAgent()
124		initialAg := ag	125		initialAg := ag
125		defer func() {	126		defer func() {

@@ -128,7 +129,6 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
128		closeAgentLSP(initialAg)	129		closeAgentLSP(initialAg)
129		}	130		}
130		}()	131		}()
131	-	wireNonInteractiveAgentExtHooks(ctx, ag, extMgr)
132		sess, err := openOrCreateSession(ctx, args, r, ag, version)	132		sess, err := openOrCreateSession(ctx, args, r, ag, version)
133		if err != nil {	133		if err != nil {
134		return err	134		return err

@@ -136,12 +136,11 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
136		var sessionPersistence sessionPersistenceState	136		var sessionPersistence sessionPersistenceState
137		if sess != nil {	137		if sess != nil {
138		var providerName, model string	138		var providerName, model string
139	-	sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, extMgr, sess, ag)	139	+	sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, nil, sess, ag)
140		if err != nil {	140		if err != nil {
141		return err	141		return err
142		}	142		}
143		r.Provider, r.Model = providerName, model	143		r.Provider, r.Model = providerName, model
144	-	ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) }
145		// A prompt is acknowledged by the supervisor only after this callback	144		// A prompt is acknowledged by the supervisor only after this callback
146		// has appended and synced the matching user message. This closes the	145		// has appended and synced the matching user message. This closes the
147		// crash window between accepting a follow-up and writing session.json.	146		// crash window between accepting a follow-up and writing session.json.

@@ -165,7 +164,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
165		runErr = joinSessionCloseError(runErr, sess)	164		runErr = joinSessionCloseError(runErr, sess)
166		}()	165		}()
167		}	166		}
168	-	announceSession(extMgr, sess)	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
169			171
170		// Resolve retains metadata for catalog models and for valid synthesized	172		// Resolve retains metadata for catalog models and for valid synthesized
171		// local/routed/open-catalog models. The active agent also survives a session	173		// local/routed/open-catalog models. The active agent also survives a session

@@ -254,7 +256,13 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
254		mu.Unlock()	256		mu.Unlock()
255			257
256		turnStarted := false	258		turnStarted := false
			259	+	toolStarted := false
			260	+	ag.MaxRetries = 2
257		sink := func(ev core.AgentEvent) {	261		sink := func(ev core.AgentEvent) {
			262	+	if ev.Type() == "tool_execution_started" && !toolStarted {
			263	+	toolStarted = true
			264	+	ag.MaxRetries = 0
			265	+	}
258		data := subagentEventData(ev)	266		data := subagentEventData(ev)
259		if ev.Type() == "user_message" && !turnStarted {	267		if ev.Type() == "user_message" && !turnStarted {
260		// Prompt invokes OnMessageAppended before delivering this sink	268		// Prompt invokes OnMessageAppended before delivering this sink

@@ -707,6 +715,9 @@ func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[
707		if provider.IsContextOverflowError(err) {	715		if provider.IsContextOverflowError(err) {
708		return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}	716		return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
709		}	717		}
			718	+	if errors.Is(err, core.ErrStreamIdleTimeout) {
			719	+	return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for one minute; the delegated turn stopped safely"}
			720	+	}
710		if errors.Is(err, context.DeadlineExceeded) \|\| shutdownOrigin == subagents.ShutdownOriginDeadline {	721		if errors.Is(err, context.DeadlineExceeded) \|\| shutdownOrigin == subagents.ShutdownOriginDeadline {
711		return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}	722		return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}
712		}	723		}

M packages/agent/subagent_worker_test.go +7 -0


@@ -52,6 +52,13 @@ func TestSubagentTurnContextResetsDeadlineAndHonorsShutdown(t *testing.T) {
52	52		}
53	53		}
54	54
	55	+	func TestResultErrorPayloadClassifiesStreamIdleTimeout(t *testing.T) {
	56	+	payload := resultErrorPayload(core.ErrStreamIdleTimeout, "")
	57	+	if payload["code"] != "stream_idle_timeout" {
	58	+	t.Fatalf("idle timeout error code = %v, want stream_idle_timeout", payload["code"])
	59	+	}
	60	+	}
	61	+
55	62		func TestResultErrorPayloadClassifiesTurnDeadline(t *testing.T) {
56	63		payload := resultErrorPayload(context.DeadlineExceeded, "")
57	64		if payload["code"] != "deadline_exceeded" {


@@ -52,6 +52,13 @@ func TestSubagentTurnContextResetsDeadlineAndHonorsShutdown(t *testing.T) {
52	}	52		}
53	}	53		}
54		54
		55	+	func TestResultErrorPayloadClassifiesStreamIdleTimeout(t *testing.T) {
		56	+	payload := resultErrorPayload(core.ErrStreamIdleTimeout, "")
		57	+	if payload["code"] != "stream_idle_timeout" {
		58	+	t.Fatalf("idle timeout error code = %v, want stream_idle_timeout", payload["code"])
		59	+	}
		60	+	}
		61	+
55	func TestResultErrorPayloadClassifiesTurnDeadline(t *testing.T) {	62		func TestResultErrorPayloadClassifiesTurnDeadline(t *testing.T) {
56	payload := resultErrorPayload(context.DeadlineExceeded, "")	63		payload := resultErrorPayload(context.DeadlineExceeded, "")
57	if payload["code"] != "deadline_exceeded" {	64		if payload["code"] != "deadline_exceeded" {

M packages/agent/subagents/projection.go +53 -8


@@ -10,12 +10,16 @@ import (
10	10		// boundary. Completed operations are deliberately not retained in the public
11	11		// projection: their terminal trace fact remains available as LastEvent.
12	12		type Operation struct {
13		-	Type string
14		-	AgentID string
15		-	TurnID string
16		-	CallID string
17		-	Name string
18		-	StartedAt time.Time
	13	+	Type string
	14	+	AgentID string
	15	+	TurnID string
	16	+	CallID string
	17	+	Name string
	18	+	Provider string
	19	+	Model string
	20	+	Attempt int
	21	+	MaxAttempts int
	22	+	StartedAt time.Time
19	23		}
20	24
21	25		// Open is always true because the projection exposes only unmatched starts.

@@ -82,6 +86,20 @@ type ResultFact struct {
82	86		At time.Time
83	87		}
84	88
	89	+	// RequestFact is a bounded provider-attempt diagnostic. It excludes request
	90	+	// bodies, streamed text, tool arguments, and raw provider errors.
	91	+	type RequestFact struct {
	92	+	TurnID string
	93	+	Provider string
	94	+	Model string
	95	+	Attempt int
	96	+	MaxAttempts int
	97	+	Outcome string
	98	+	ErrorCode string
	99	+	StartedAt time.Time
	100	+	EndedAt time.Time
	101	+	}
	102	+
85	103		// AgentTraceView is the only user-facing execution state. It intentionally
86	104		// contains operations and terminal facts rather than lifecycle guesses such as
87	105		// working, idle, or alive.

@@ -93,6 +111,7 @@ type AgentTraceView struct {
93	111		PrimaryOperation *Operation
94	112		Terminal string
95	113		Result *ResultFact
	114	+	LastRequest *RequestFact
96	115		}
97	116
98	117		// Summary returns the most specific factual activity or terminal result fact.

@@ -140,14 +159,31 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView {
140	159		view.LastEvent = event
141	160		key, starts, ends, terminal := traceBoundary(event)
142	161		if starts {
	162	+	if event.Type == "provider.request.started" {
	163	+	view.LastRequest = requestFact(event, "open")
	164	+	}
143	165		if open[event.AgentID] == nil {
144	166		open[event.AgentID] = make(map[string]Operation)
145	167		}
146	168		callID, _ := event.Data["call_id"].(string)
147	169		name, _ := event.Data["name"].(string)
148		-	open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, CallID: callID, Name: name, StartedAt: event.Timestamp}
	170	+	provider, _ := event.Data["provider"].(string)
	171	+	model, _ := event.Data["model"].(string)
	172	+	attempt, _ := event.Data["attempt"].(int)
	173	+	maxAttempts, _ := event.Data["max_attempts"].(int)
	174	+	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}
149	175		}
150	176		if ends {
	177	+	if event.Type == "provider.request.finished" \|\| event.Type == "provider.request.failed" \|\| event.Type == "provider.request.cancelled" \|\| event.Type == "provider.request.retry.scheduled" {
	178	+	if active, found := open[event.AgentID][key]; found {
	179	+	outcome := strings.TrimPrefix(event.Type, "provider.request.")
	180	+	request := requestFact(event, outcome)
	181	+	request.Provider, request.Model = active.Provider, active.Model
	182	+	request.Attempt, request.MaxAttempts = active.Attempt, active.MaxAttempts
	183	+	request.StartedAt = active.StartedAt
	184	+	view.LastRequest = request
	185	+	}
	186	+	}
151	187		delete(open[event.AgentID], key)
152	188		}
153	189		if observation := traceObservation(event); observation != nil {

@@ -233,6 +269,15 @@ func traceObservation(event TraceEvent) *LiveObservation {
233	269		return nil
234	270		}
235	271
	272	+	func requestFact(event TraceEvent, outcome string) *RequestFact {
	273	+	providerName, _ := event.Data["provider"].(string)
	274	+	model, _ := event.Data["model"].(string)
	275	+	attempt, _ := event.Data["attempt"].(int)
	276	+	maxAttempts, _ := event.Data["max_attempts"].(int)
	277	+	errorCode, _ := event.Data["error_code"].(string)
	278	+	return &RequestFact{TurnID: event.TurnID, Provider: providerName, Model: model, Attempt: attempt, MaxAttempts: maxAttempts, Outcome: outcome, ErrorCode: errorCode, StartedAt: event.Timestamp, EndedAt: event.Timestamp}
	279	+	}
	280	+
236	281		func traceResultFact(event TraceEvent, previous ResultFact) ResultFact {
237	282		if event.Type != "result.available" && event.Type != "result.delivered" && event.Type != "result.delivery.failed" {
238	283		return previous

@@ -270,7 +315,7 @@ func traceBoundary(event TraceEvent) (key string, starts, ends bool, terminal st
270	315		return "tool:" + event.TurnID + ":" + callID, false, true, ""
271	316		case "provider.request.started":
272	317		return "provider:" + event.TurnID, true, false, ""
273		-	case "provider.request.finished", "provider.request.failed", "provider.request.cancelled":
	318	+	case "provider.request.finished", "provider.request.failed", "provider.request.cancelled", "provider.request.retry.scheduled":
274	319		return "provider:" + event.TurnID, false, true, ""
275	320		case "agent.wait.started":
276	321		return "wait:" + event.TurnID, true, false, ""

 // boundary. Completed operations are deliberately not retained in the public
 // projection: their terminal trace fact remains available as LastEvent.
 type Operation struct {
-	Type      string
+	Type        string
-	AgentID   string
+	AgentID     string
-	TurnID    string
+	TurnID      string
-	CallID    string
+	CallID      string
-	Name      string
+	Name        string
-	StartedAt time.Time
+	Provider    string
+	Model       string
+	Attempt     int
+	MaxAttempts int
+	StartedAt   time.Time
 }
 // Open is always true because the projection exposes only unmatched starts.
 	At        time.Time
 }
+// RequestFact is a bounded provider-attempt diagnostic. It excludes request
+// bodies, streamed text, tool arguments, and raw provider errors.
+type RequestFact struct {
+	TurnID      string
+	Provider    string
+	Model       string
+	Attempt     int
+	MaxAttempts int
+	Outcome     string
+	ErrorCode   string
+	StartedAt   time.Time
+	EndedAt     time.Time
+}
 // AgentTraceView is the only user-facing execution state. It intentionally
 // contains operations and terminal facts rather than lifecycle guesses such as
 // working, idle, or alive.
 	PrimaryOperation *Operation
 	Terminal         string
 	Result           *ResultFact
+	LastRequest      *RequestFact
 }
 // Summary returns the most specific factual activity or terminal result fact.
 		view.LastEvent = event
 		key, starts, ends, terminal := traceBoundary(event)
 		if starts {
+			if event.Type == "provider.request.started" {
+				view.LastRequest = requestFact(event, "open")
+			}
 			if open[event.AgentID] == nil {
 				open[event.AgentID] = make(map[string]Operation)
 			}
 			callID, _ := event.Data["call_id"].(string)
 			name, _ := event.Data["name"].(string)
-			open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, CallID: callID, Name: name, StartedAt: event.Timestamp}
+			provider, _ := event.Data["provider"].(string)
+			model, _ := event.Data["model"].(string)
+			attempt, _ := event.Data["attempt"].(int)
+			maxAttempts, _ := event.Data["max_attempts"].(int)
+			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 {
+			if event.Type == "provider.request.finished" || event.Type == "provider.request.failed" || event.Type == "provider.request.cancelled" || event.Type == "provider.request.retry.scheduled" {
+				if active, found := open[event.AgentID][key]; found {
+					outcome := strings.TrimPrefix(event.Type, "provider.request.")
+					request := requestFact(event, outcome)
+					request.Provider, request.Model = active.Provider, active.Model
+					request.Attempt, request.MaxAttempts = active.Attempt, active.MaxAttempts
+					request.StartedAt = active.StartedAt
+					view.LastRequest = request
+				}
+			}
 			delete(open[event.AgentID], key)
 		}
 		if observation := traceObservation(event); observation != nil {
 	return nil
 }
+func requestFact(event TraceEvent, outcome string) *RequestFact {
+	providerName, _ := event.Data["provider"].(string)
+	model, _ := event.Data["model"].(string)
+	attempt, _ := event.Data["attempt"].(int)
+	maxAttempts, _ := event.Data["max_attempts"].(int)
+	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}
+}
 func traceResultFact(event TraceEvent, previous *ResultFact) *ResultFact {
 	if event.Type != "result.available" && event.Type != "result.delivered" && event.Type != "result.delivery.failed" {
 		return previous
 		return "tool:" + event.TurnID + ":" + callID, false, true, ""
 	case "provider.request.started":
 		return "provider:" + event.TurnID, true, false, ""
-	case "provider.request.finished", "provider.request.failed", "provider.request.cancelled":
+	case "provider.request.finished", "provider.request.failed", "provider.request.cancelled", "provider.request.retry.scheduled":
 		return "provider:" + event.TurnID, false, true, ""
 	case "agent.wait.started":
 		return "wait:" + event.TurnID, true, false, ""

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

@@ -1,6 +1,7 @@
1	1		package subagents
2	2
3	3		import (
	4	+	"reflect"
4	5		"testing"
5	6		"time"
6	7		)

@@ -79,6 +80,20 @@ func TestProjectTraceSelectsSpecificActivityAndProjectsResultDelivery(t *testing
79	80		}
80	81		}
81	82
	83	+	func TestProjectTraceRetainsSafeProviderAttemptDiagnostic(t *testing.T) {
	84	+	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
	85	+	view := ProjectTrace([]TraceEvent{
	86	+	{Timestamp: started, Type: "provider.request.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"provider": "openai-codex", "model": "gpt-5.6", "attempt": 2, "max_attempts": 4}},
	87	+	{Timestamp: started.Add(time.Second), Type: "provider.request.failed", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"error_code": "deadline_exceeded"}},
	88	+	})["agent-1"]
	89	+	if len(view.OpenOperations) != 0 {
	90	+	t.Fatalf("open operations = %#v, want none", view.OpenOperations)
	91	+	}
	92	+	if got, want := view.LastRequest, (&RequestFact{TurnID: "turn-1", Provider: "openai-codex", Model: "gpt-5.6", Attempt: 2, MaxAttempts: 4, Outcome: "failed", ErrorCode: "deadline_exceeded", StartedAt: started, EndedAt: started.Add(time.Second)}); !reflect.DeepEqual(got, want) {
	93	+	t.Fatalf("request diagnostic = %#v, want %#v", got, want)
	94	+	}
	95	+	}
	96	+
82	97		func TestProjectTraceOmitsObservationForAnotherToolInSameTurn(t *testing.T) {
83	98		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
84	99		view := ProjectTrace([]TraceEvent{

@@ -1,6 +1,7 @@
1	package subagents	1		package subagents
2		2
3	import (	3		import (
		4	+	"reflect"
4	"testing"	5		"testing"
5	"time"	6		"time"
6	)	7		)

@@ -79,6 +80,20 @@ func TestProjectTraceSelectsSpecificActivityAndProjectsResultDelivery(t *testing
79	}	80		}
80	}	81		}
81		82
		83	+	func TestProjectTraceRetainsSafeProviderAttemptDiagnostic(t *testing.T) {
		84	+	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
		85	+	view := ProjectTrace([]TraceEvent{
		86	+	{Timestamp: started, Type: "provider.request.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"provider": "openai-codex", "model": "gpt-5.6", "attempt": 2, "max_attempts": 4}},
		87	+	{Timestamp: started.Add(time.Second), Type: "provider.request.failed", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"error_code": "deadline_exceeded"}},
		88	+	})["agent-1"]
		89	+	if len(view.OpenOperations) != 0 {
		90	+	t.Fatalf("open operations = %#v, want none", view.OpenOperations)
		91	+	}
		92	+	if got, want := view.LastRequest, (&RequestFact{TurnID: "turn-1", Provider: "openai-codex", Model: "gpt-5.6", Attempt: 2, MaxAttempts: 4, Outcome: "failed", ErrorCode: "deadline_exceeded", StartedAt: started, EndedAt: started.Add(time.Second)}); !reflect.DeepEqual(got, want) {
		93	+	t.Fatalf("request diagnostic = %#v, want %#v", got, want)
		94	+	}
		95	+	}
		96	+
82	func TestProjectTraceOmitsObservationForAnotherToolInSameTurn(t *testing.T) {	97		func TestProjectTraceOmitsObservationForAnotherToolInSameTurn(t *testing.T) {
83	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)	98		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
84	view := ProjectTrace([]TraceEvent{	99		view := ProjectTrace([]TraceEvent{

M packages/agent/subagents/runner.go +73 -9


@@ -903,30 +903,58 @@ func recordWorkerTrace(agent *Agent, ev Event) {
903	903		if agent == nil {
904	904		return
905	905		}
	906	+	// assistant_start only marks the beginning of a provider response. It is
	907	+	// neither streamed content nor an open operation, and recording it as an
	908	+	// unknown protocol observation makes it replace useful live activity.
	909	+	if ev.Type == "assistant_start" \|\| ev.Type == EventAgentHeartbeat \|\| ev.Type == "agent_heartbeat" {
	910	+	return
	911	+	}
906	912		typeName := strings.ReplaceAll(ev.Type, "_", ".")
907	913		traceType := "worker.protocol.observed"
908	914		nestedTurn, _ := ev.Data["nested_turn"].(bool)
909	915		switch ev.Type {
910		-	case "turn_start", "turn_end":
	916	+	case "turn_start":
911	917		if nestedTurn {
912		-	if ev.Type == "turn_start" {
913		-	traceType = "provider.request.started"
914		-	} else {
915		-	traceType = "provider.request.finished"
916		-	}
917		-	} else if ev.Type == "turn_start" {
	918	+	traceType = "provider.request.started"
	919	+	} else {
918	920		traceType = "turn.started"
	921	+	}
	922	+	case "turn_end":
	923	+	// Each delegated turn has completed its provider/model loop by this
	924	+	// point. Close the provider operation even when the terminal result was
	925	+	// emitted through a separate turn.result event.
	926	+	if errMsg, _ := ev.Data["error"].(string); errMsg != "" {
	927	+	traceType = "provider.request.failed"
919	928		} else {
920		-	traceType = "turn.finished"
	929	+	traceType = "provider.request.finished"
921	930		}
922	931		case EventTurnStarted:
923		-	traceType = "turn.started"
	932	+	// The worker canonicalizes core turn_start to turn.started. Its
	933	+	// nested_turn marker keeps that provider-loop boundary distinct from
	934	+	// the delegated worker turn, which also uses turn.started.
	935	+	if nestedTurn {
	936	+	traceType = "provider.request.started"
	937	+	} else {
	938	+	traceType = "turn.started"
	939	+	}
924	940		case EventTurnResult:
925	941		traceType = "turn.finished"
	942	+	if status, _ := ev.Data["status"].(string); status == "failed" {
	943	+	traceType = "turn.failed"
	944	+	} else if status == "canceled" {
	945	+	traceType = "turn.cancelled"
	946	+	}
926	947		case "turn_failed", EventTurnFailed:
927	948		traceType = "turn.failed"
	949	+	case "turn_cancelled", "turn.canceled":
	950	+	traceType = "turn.cancelled"
928	951		case "request_started":
929	952		traceType = "provider.request.started"
	953	+	case "retry_scheduled":
	954	+	traceType = "provider.request.retry.scheduled"
	955	+	case "error":
	956	+	// Error events do not necessarily belong to a provider request; the
	957	+	// following terminal turn event closes any request factually.
930	958		case "tool_call", EventToolStarted:
931	959		traceType = "tool.started"
932	960		case "tool_result", EventToolFinished:

@@ -945,6 +973,21 @@ func recordWorkerTrace(agent *Agent, ev Event) {
945	973		if status, ok := ev.Data["status"].(string); ok && status != "" {
946	974		data["status"] = status
947	975		}
	976	+	for _, key := range []string{"provider", "model"} {
	977	+	if value, ok := ev.Data[key].(string); ok && value != "" {
	978	+	data[key] = value
	979	+	}
	980	+	}
	981	+	for _, key := range []string{"attempt", "max_attempts"} {
	982	+	if value, ok := ev.Data[key].(float64); ok && value > 0 {
	983	+	data[key] = int(value)
	984	+	} else if value, ok := ev.Data[key].(int); ok && value > 0 {
	985	+	data[key] = value
	986	+	}
	987	+	}
	988	+	if errorCode := traceErrorCode(ev.Data); errorCode != "" {
	989	+	data["error_code"] = errorCode
	990	+	}
948	991		callID, _ := ev.Data["call_id"].(string)
949	992		if callID == "" {
950	993		callID, _ = ev.Data["id"].(string)

@@ -955,6 +998,27 @@ func recordWorkerTrace(agent *Agent, ev Event) {
955	998		agent.recordTrace(TraceEvent{Type: traceType, Timestamp: ev.Time, TurnID: ev.TurnID, Data: data})
956	999		}
957	1000
	1001	+	func traceErrorCode(data map[string]any) string {
	1002	+	value, _ := data["error"].(string)
	1003	+	value = strings.ToLower(value)
	1004	+	switch {
	1005	+	case strings.Contains(value, "deadline") \|\| strings.Contains(value, "timeout"):
	1006	+	return "deadline_exceeded"
	1007	+	case strings.Contains(value, "canceled") \|\| strings.Contains(value, "cancelled"):
	1008	+	return "cancelled"
	1009	+	case strings.Contains(value, "quota") \|\| strings.Contains(value, "billing"):
	1010	+	return "quota"
	1011	+	case strings.Contains(value, "rate limit") \|\| strings.Contains(value, "429"):
	1012	+	return "rate_limit"
	1013	+	case strings.Contains(value, "context") && strings.Contains(value, "limit"):
	1014	+	return "context_limit"
	1015	+	case value != "":
	1016	+	return "provider_error"
	1017	+	default:
	1018	+	return ""
	1019	+	}
	1020	+	}
	1021	+
958	1022		// applyEventToSink translates an Event into Sink updates. Only a
959	1023		// few event types are interpreted; the rest still land in the
960	1024		// durable log via the caller.


@@ -903,30 +903,58 @@ func recordWorkerTrace(agent *Agent, ev Event) {
903		if agent == nil {	903		if agent == nil {
904		return	904		return
905		}	905		}
			906	+	// assistant_start only marks the beginning of a provider response. It is
			907	+	// neither streamed content nor an open operation, and recording it as an
			908	+	// unknown protocol observation makes it replace useful live activity.
			909	+	if ev.Type == "assistant_start" \|\| ev.Type == EventAgentHeartbeat \|\| ev.Type == "agent_heartbeat" {
			910	+	return
			911	+	}
906		typeName := strings.ReplaceAll(ev.Type, "_", ".")	912		typeName := strings.ReplaceAll(ev.Type, "_", ".")
907		traceType := "worker.protocol.observed"	913		traceType := "worker.protocol.observed"
908		nestedTurn, _ := ev.Data["nested_turn"].(bool)	914		nestedTurn, _ := ev.Data["nested_turn"].(bool)
909		switch ev.Type {	915		switch ev.Type {
910	-	case "turn_start", "turn_end":	916	+	case "turn_start":
911		if nestedTurn {	917		if nestedTurn {
912	-	if ev.Type == "turn_start" {	918	+	traceType = "provider.request.started"
913	-	traceType = "provider.request.started"	919	+	} else {
914	-	} else {
915	-	traceType = "provider.request.finished"
916	-	}
917	-	} else if ev.Type == "turn_start" {
918		traceType = "turn.started"	920		traceType = "turn.started"
			921	+	}
			922	+	case "turn_end":
			923	+	// Each delegated turn has completed its provider/model loop by this
			924	+	// point. Close the provider operation even when the terminal result was
			925	+	// emitted through a separate turn.result event.
			926	+	if errMsg, _ := ev.Data["error"].(string); errMsg != "" {
			927	+	traceType = "provider.request.failed"
919		} else {	928		} else {
920	-	traceType = "turn.finished"	929	+	traceType = "provider.request.finished"
921		}	930		}
922		case EventTurnStarted:	931		case EventTurnStarted:
923	-	traceType = "turn.started"	932	+	// The worker canonicalizes core turn_start to turn.started. Its
			933	+	// nested_turn marker keeps that provider-loop boundary distinct from
			934	+	// the delegated worker turn, which also uses turn.started.
			935	+	if nestedTurn {
			936	+	traceType = "provider.request.started"
			937	+	} else {
			938	+	traceType = "turn.started"
			939	+	}
924		case EventTurnResult:	940		case EventTurnResult:
925		traceType = "turn.finished"	941		traceType = "turn.finished"
			942	+	if status, _ := ev.Data["status"].(string); status == "failed" {
			943	+	traceType = "turn.failed"
			944	+	} else if status == "canceled" {
			945	+	traceType = "turn.cancelled"
			946	+	}
926		case "turn_failed", EventTurnFailed:	947		case "turn_failed", EventTurnFailed:
927		traceType = "turn.failed"	948		traceType = "turn.failed"
			949	+	case "turn_cancelled", "turn.canceled":
			950	+	traceType = "turn.cancelled"
928		case "request_started":	951		case "request_started":
929		traceType = "provider.request.started"	952		traceType = "provider.request.started"
			953	+	case "retry_scheduled":
			954	+	traceType = "provider.request.retry.scheduled"
			955	+	case "error":
			956	+	// Error events do not necessarily belong to a provider request; the
			957	+	// following terminal turn event closes any request factually.
930		case "tool_call", EventToolStarted:	958		case "tool_call", EventToolStarted:
931		traceType = "tool.started"	959		traceType = "tool.started"
932		case "tool_result", EventToolFinished:	960		case "tool_result", EventToolFinished:

@@ -945,6 +973,21 @@ func recordWorkerTrace(agent *Agent, ev Event) {
945		if status, ok := ev.Data["status"].(string); ok && status != "" {	973		if status, ok := ev.Data["status"].(string); ok && status != "" {
946		data["status"] = status	974		data["status"] = status
947		}	975		}
			976	+	for _, key := range []string{"provider", "model"} {
			977	+	if value, ok := ev.Data[key].(string); ok && value != "" {
			978	+	data[key] = value
			979	+	}
			980	+	}
			981	+	for _, key := range []string{"attempt", "max_attempts"} {
			982	+	if value, ok := ev.Data[key].(float64); ok && value > 0 {
			983	+	data[key] = int(value)
			984	+	} else if value, ok := ev.Data[key].(int); ok && value > 0 {
			985	+	data[key] = value
			986	+	}
			987	+	}
			988	+	if errorCode := traceErrorCode(ev.Data); errorCode != "" {
			989	+	data["error_code"] = errorCode
			990	+	}
948		callID, _ := ev.Data["call_id"].(string)	991		callID, _ := ev.Data["call_id"].(string)
949		if callID == "" {	992		if callID == "" {
950		callID, _ = ev.Data["id"].(string)	993		callID, _ = ev.Data["id"].(string)

@@ -955,6 +998,27 @@ func recordWorkerTrace(agent *Agent, ev Event) {
955		agent.recordTrace(TraceEvent{Type: traceType, Timestamp: ev.Time, TurnID: ev.TurnID, Data: data})	998		agent.recordTrace(TraceEvent{Type: traceType, Timestamp: ev.Time, TurnID: ev.TurnID, Data: data})
956		}	999		}
957			1000
			1001	+	func traceErrorCode(data map[string]any) string {
			1002	+	value, _ := data["error"].(string)
			1003	+	value = strings.ToLower(value)
			1004	+	switch {
			1005	+	case strings.Contains(value, "deadline") \|\| strings.Contains(value, "timeout"):
			1006	+	return "deadline_exceeded"
			1007	+	case strings.Contains(value, "canceled") \|\| strings.Contains(value, "cancelled"):
			1008	+	return "cancelled"
			1009	+	case strings.Contains(value, "quota") \|\| strings.Contains(value, "billing"):
			1010	+	return "quota"
			1011	+	case strings.Contains(value, "rate limit") \|\| strings.Contains(value, "429"):
			1012	+	return "rate_limit"
			1013	+	case strings.Contains(value, "context") && strings.Contains(value, "limit"):
			1014	+	return "context_limit"
			1015	+	case value != "":
			1016	+	return "provider_error"
			1017	+	default:
			1018	+	return ""
			1019	+	}
			1020	+	}
			1021	+
958		// applyEventToSink translates an Event into Sink updates. Only a	1022		// applyEventToSink translates an Event into Sink updates. Only a
959		// few event types are interpreted; the rest still land in the	1023		// few event types are interpreted; the rest still land in the
960		// durable log via the caller.	1024		// durable log via the caller.

M packages/agent/subagents/runner_test.go +23 -1


@@ -806,8 +806,12 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
806	806		t.Cleanup(func() { _ = trace.Close() })
807	807		agent := &Agent{ID: "agent-1", trace: trace}
808	808		recordWorkerTrace(agent, Event{Type: EventMessageDelta, TurnID: "turn-1", Data: map[string]any{"delta": "sensitive answer"}})
	809	+	recordWorkerTrace(agent, Event{Type: "assistant_start", TurnID: "turn-1"})
	810	+	recordWorkerTrace(agent, Event{Type: EventAgentHeartbeat, TurnID: "turn-1"})
809	811		recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1", Data: map[string]any{"provider": "test", "model": "test"}})
810		-	recordWorkerTrace(agent, Event{Type: "turn_start", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
	812	+	// The worker canonicalizes core turn_start to EventTurnStarted while
	813	+	// retaining nested_turn, so this exercises the supervisor's real input.
	814	+	recordWorkerTrace(agent, Event{Type: EventTurnStarted, TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
811	815		recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
812	816		recordWorkerTrace(agent, Event{Type: "tool_call", TurnID: "turn-1", Data: map[string]any{"id": "call-1", "name": "bash", "args": "sensitive arguments"}})
813	817		recordWorkerTrace(agent, Event{Type: "future_event", TurnID: "turn-1", Data: map[string]any{"secret": "hidden"}})

@@ -857,6 +861,24 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
857	861		}
858	862		}
859	863
	864	+	func TestRecordWorkerTraceClosesProviderRequestWhenTurnEnds(t *testing.T) {
	865	+	trace := NewMemoryTraceWriter()
	866	+	t.Cleanup(func() { _ = trace.Close() })
	867	+	agent := &Agent{ID: "agent-1", trace: trace}
	868	+	recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1"})
	869	+	recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"error": "provider unavailable"}})
	870	+	if err := trace.Flush(); err != nil {
	871	+	t.Fatal(err)
	872	+	}
	873	+	view := ProjectTrace(trace.Events())["agent-1"]
	874	+	if len(view.OpenOperations) != 0 {
	875	+	t.Fatalf("open operations = %#v, want none", view.OpenOperations)
	876	+	}
	877	+	if view.LastEvent.Type != "provider.request.failed" {
	878	+	t.Fatalf("last event = %#v, want provider request failure", view.LastEvent)
	879	+	}
	880	+	}
	881	+
860	882		func TestApplyEventToSinkProjectsToolActivityIntoTranscript(t *testing.T) {
861	883		events := []Event{
862	884		{Type: EventToolStarted, Data: map[string]any{"name": "bash"}},


@@ -806,8 +806,12 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
806		t.Cleanup(func() { _ = trace.Close() })	806		t.Cleanup(func() { _ = trace.Close() })
807		agent := &Agent{ID: "agent-1", trace: trace}	807		agent := &Agent{ID: "agent-1", trace: trace}
808		recordWorkerTrace(agent, Event{Type: EventMessageDelta, TurnID: "turn-1", Data: map[string]any{"delta": "sensitive answer"}})	808		recordWorkerTrace(agent, Event{Type: EventMessageDelta, TurnID: "turn-1", Data: map[string]any{"delta": "sensitive answer"}})
			809	+	recordWorkerTrace(agent, Event{Type: "assistant_start", TurnID: "turn-1"})
			810	+	recordWorkerTrace(agent, Event{Type: EventAgentHeartbeat, TurnID: "turn-1"})
809		recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1", Data: map[string]any{"provider": "test", "model": "test"}})	811		recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1", Data: map[string]any{"provider": "test", "model": "test"}})
810	-	recordWorkerTrace(agent, Event{Type: "turn_start", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})	812	+	// The worker canonicalizes core turn_start to EventTurnStarted while
			813	+	// retaining nested_turn, so this exercises the supervisor's real input.
			814	+	recordWorkerTrace(agent, Event{Type: EventTurnStarted, TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
811		recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})	815		recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
812		recordWorkerTrace(agent, Event{Type: "tool_call", TurnID: "turn-1", Data: map[string]any{"id": "call-1", "name": "bash", "args": "sensitive arguments"}})	816		recordWorkerTrace(agent, Event{Type: "tool_call", TurnID: "turn-1", Data: map[string]any{"id": "call-1", "name": "bash", "args": "sensitive arguments"}})
813		recordWorkerTrace(agent, Event{Type: "future_event", TurnID: "turn-1", Data: map[string]any{"secret": "hidden"}})	817		recordWorkerTrace(agent, Event{Type: "future_event", TurnID: "turn-1", Data: map[string]any{"secret": "hidden"}})

@@ -857,6 +861,24 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
857		}	861		}
858		}	862		}
859			863
			864	+	func TestRecordWorkerTraceClosesProviderRequestWhenTurnEnds(t *testing.T) {
			865	+	trace := NewMemoryTraceWriter()
			866	+	t.Cleanup(func() { _ = trace.Close() })
			867	+	agent := &Agent{ID: "agent-1", trace: trace}
			868	+	recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1"})
			869	+	recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"error": "provider unavailable"}})
			870	+	if err := trace.Flush(); err != nil {
			871	+	t.Fatal(err)
			872	+	}
			873	+	view := ProjectTrace(trace.Events())["agent-1"]
			874	+	if len(view.OpenOperations) != 0 {
			875	+	t.Fatalf("open operations = %#v, want none", view.OpenOperations)
			876	+	}
			877	+	if view.LastEvent.Type != "provider.request.failed" {
			878	+	t.Fatalf("last event = %#v, want provider request failure", view.LastEvent)
			879	+	}
			880	+	}
			881	+
860		func TestApplyEventToSinkProjectsToolActivityIntoTranscript(t *testing.T) {	882		func TestApplyEventToSinkProjectsToolActivityIntoTranscript(t *testing.T) {
861		events := []Event{	883		events := []Event{
862		{Type: EventToolStarted, Data: map[string]any{"name": "bash"}},	884		{Type: EventToolStarted, Data: map[string]any{"name": "bash"}},

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


@@ -13,6 +13,8 @@ import (
13	13		"github.com/bnema/zut/packages/provider"
14	14		)
15	15
	16	+	var ErrStreamIdleTimeout = errors.New("provider stream idle timeout")
	17	+
16	18		type queuedMessage struct {
17	19		text string
18	20		accepted time.Time

@@ -95,6 +97,9 @@ type Agent struct {
95	97		// RetryBaseDelay is doubled for each attempt; zero uses 2s.
96	98		MaxRetries int
97	99		RetryBaseDelay time.Duration
	100	+	// StreamIdleTimeout cancels an individual provider stream when it emits no
	101	+	// events for this duration. Zero leaves the stream unbounded.
	102	+	StreamIdleTimeout time.Duration
98	103
99	104		// OnEvent, if set, mirrors every AgentEvent the loop emits to
100	105		// this callback in addition to the per-Prompt sink. Used by the

@@ -916,7 +921,9 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext
916	921		Attempt: attempt + 1,
917	922		MaxAttempts: maxAttempts,
918	923		})
919		-	stream, err := a.Client.Stream(ctx, req)
	924	+	streamCtx, cancelStream := context.WithCancel(ctx)
	925	+	defer cancelStream()
	926	+	stream, err := a.Client.Stream(streamCtx, req)
920	927		if err != nil {
921	928		return provider.StopError, provider.Message{}, err
922	929		}

@@ -929,30 +936,95 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext
929	936		finalMsg provider.Message
930	937		)
931	938
932		-	for ev := range stream {
933		-	switch e := ev.(type) {
934		-	case provider.EventStart:
935		-	// nothing
936		-	case provider.EventTextDelta:
937		-	sink(EvTextDelta{Delta: e.Delta})
938		-	case provider.EventToolStart:
939		-	sink(EvToolUseStart{ID: e.ID, Name: e.Name})
940		-	case provider.EventToolArgs:
941		-	sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
942		-	case provider.EventToolEnd:
943		-	sink(EvToolUseEnd{ID: e.ID})
944		-	case provider.EventUsage:
945		-	cum := a.addUsage(e.Usage)
946		-	sink(EvUsage{Usage: e.Usage, Cumulative: cum})
947		-	if a.OnUsage != nil {
948		-	a.OnUsage(cum)
	939	+	var idle <-chan time.Time
	940	+	var resetIdle func()
	941	+	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)
	953	+	}
	954	+	}
	955	+	for {
	956	+	// Prefer an already-buffered terminal provider event over a concurrent
	957	+	// context cancellation so retry classification retains its real cause.
	958	+	select {
	959	+	case ev, ok := <-stream:
	960	+	if !ok {
	961	+	goto streamDone
	962	+	}
	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	+	default:
	990	+	}
	991	+	select {
	992	+	case <-streamCtx.Done():
	993	+	return provider.StopError, finalMsg, streamCtx.Err()
	994	+	case <-idle:
	995	+	return provider.StopError, finalMsg, ErrStreamIdleTimeout
	996	+	case ev, ok := <-stream:
	997	+	if !ok {
	998	+	goto streamDone
	999	+	}
	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
949	1024		}
950		-	case provider.EventDone:
951		-	stop = e.Stop
952		-	finalErr = e.Err
953		-	finalMsg = e.Message
954	1025		}
955	1026		}
	1027	+	streamDone:
956	1028
957	1029		// Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)
958	1030		// produce partial content. When the partial message is text only we


@@ -13,6 +13,8 @@ import (
13		"github.com/bnema/zut/packages/provider"	13		"github.com/bnema/zut/packages/provider"
14		)	14		)
15			15
			16	+	var ErrStreamIdleTimeout = errors.New("provider stream idle timeout")
			17	+
16		type queuedMessage struct {	18		type queuedMessage struct {
17		text string	19		text string
18		accepted time.Time	20		accepted time.Time

@@ -95,6 +97,9 @@ type Agent struct {
95		// RetryBaseDelay is doubled for each attempt; zero uses 2s.	97		// RetryBaseDelay is doubled for each attempt; zero uses 2s.
96		MaxRetries int	98		MaxRetries int
97		RetryBaseDelay time.Duration	99		RetryBaseDelay time.Duration
			100	+	// StreamIdleTimeout cancels an individual provider stream when it emits no
			101	+	// events for this duration. Zero leaves the stream unbounded.
			102	+	StreamIdleTimeout time.Duration
98			103
99		// OnEvent, if set, mirrors every AgentEvent the loop emits to	104		// OnEvent, if set, mirrors every AgentEvent the loop emits to
100		// this callback in addition to the per-Prompt sink. Used by the	105		// this callback in addition to the per-Prompt sink. Used by the

@@ -916,7 +921,9 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext
916		Attempt: attempt + 1,	921		Attempt: attempt + 1,
917		MaxAttempts: maxAttempts,	922		MaxAttempts: maxAttempts,
918		})	923		})
919	-	stream, err := a.Client.Stream(ctx, req)	924	+	streamCtx, cancelStream := context.WithCancel(ctx)
			925	+	defer cancelStream()
			926	+	stream, err := a.Client.Stream(streamCtx, req)
920		if err != nil {	927		if err != nil {
921		return provider.StopError, provider.Message{}, err	928		return provider.StopError, provider.Message{}, err
922		}	929		}

@@ -929,30 +936,95 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext
929		finalMsg provider.Message	936		finalMsg provider.Message
930		)	937		)
931			938
932	-	for ev := range stream {	939	+	var idle <-chan time.Time
933	-	switch e := ev.(type) {	940	+	var resetIdle func()
934	-	case provider.EventStart:	941	+	if a.StreamIdleTimeout > 0 {
935	-	// nothing	942	+	timer := time.NewTimer(a.StreamIdleTimeout)
936	-	case provider.EventTextDelta:	943	+	defer timer.Stop()
937	-	sink(EvTextDelta{Delta: e.Delta})	944	+	idle = timer.C
938	-	case provider.EventToolStart:	945	+	resetIdle = func() {
939	-	sink(EvToolUseStart{ID: e.ID, Name: e.Name})	946	+	if !timer.Stop() {
940	-	case provider.EventToolArgs:	947	+	select {
941	-	sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})	948	+	case <-timer.C:
942	-	case provider.EventToolEnd:	949	+	default:
943	-	sink(EvToolUseEnd{ID: e.ID})	950	+	}
944	-	case provider.EventUsage:	951	+	}
945	-	cum := a.addUsage(e.Usage)	952	+	timer.Reset(a.StreamIdleTimeout)
946	-	sink(EvUsage{Usage: e.Usage, Cumulative: cum})	953	+	}
947	-	if a.OnUsage != nil {	954	+	}
948	-	a.OnUsage(cum)	955	+	for {
			956	+	// Prefer an already-buffered terminal provider event over a concurrent
			957	+	// context cancellation so retry classification retains its real cause.
			958	+	select {
			959	+	case ev, ok := <-stream:
			960	+	if !ok {
			961	+	goto streamDone
			962	+	}
			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	+	default:
			990	+	}
			991	+	select {
			992	+	case <-streamCtx.Done():
			993	+	return provider.StopError, finalMsg, streamCtx.Err()
			994	+	case <-idle:
			995	+	return provider.StopError, finalMsg, ErrStreamIdleTimeout
			996	+	case ev, ok := <-stream:
			997	+	if !ok {
			998	+	goto streamDone
			999	+	}
			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
949		}	1024		}
950	-	case provider.EventDone:
951	-	stop = e.Stop
952	-	finalErr = e.Err
953	-	finalMsg = e.Message
954		}	1025		}
955		}	1026		}
			1027	+	streamDone:
956			1028
957		// Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)	1029		// Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)
958		// produce partial content. When the partial message is text only we	1030		// produce partial content. When the partial message is text only we

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


@@ -48,6 +48,28 @@ func (c *retryFakeClient) Stream(ctx context.Context, req provider.Request) (<-c
48	48		return out, nil
49	49		}
50	50
	51	+	type silentStreamClient struct{}
	52	+
	53	+	func (silentStreamClient) Name() string { return "silent-stream" }
	54	+
	55	+	func (silentStreamClient) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Event, error) {
	56	+	out := make(chan provider.Event)
	57	+	go func() {
	58	+	defer close(out)
	59	+	<-ctx.Done()
	60	+	}()
	61	+	return out, nil
	62	+	}
	63	+
	64	+	func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
	65	+	a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})
	66	+	a.MaxRetries = 0
	67	+	a.StreamIdleTimeout = 10 * time.Millisecond
	68	+	if err := a.Prompt(context.Background(), "hello", nil, nil); !errors.Is(err, ErrStreamIdleTimeout) {
	69	+	t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
	70	+	}
	71	+	}
	72	+
51	73		func TestAgentRetriesOverloadedStreamError(t *testing.T) {
52	74		client := &retryFakeClient{}
53	75		a := NewAgent(client, "fake-model", "system", Registry{})


@@ -48,6 +48,28 @@ func (c *retryFakeClient) Stream(ctx context.Context, req provider.Request) (<-c
48	return out, nil	48		return out, nil
49	}	49		}
50		50
		51	+	type silentStreamClient struct{}
		52	+
		53	+	func (silentStreamClient) Name() string { return "silent-stream" }
		54	+
		55	+	func (silentStreamClient) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Event, error) {
		56	+	out := make(chan provider.Event)
		57	+	go func() {
		58	+	defer close(out)
		59	+	<-ctx.Done()
		60	+	}()
		61	+	return out, nil
		62	+	}
		63	+
		64	+	func TestAgentStopsSilentStreamAtIdleDeadline(t *testing.T) {
		65	+	a := NewAgent(silentStreamClient{}, "fake-model", "system", Registry{})
		66	+	a.MaxRetries = 0
		67	+	a.StreamIdleTimeout = 10 * time.Millisecond
		68	+	if err := a.Prompt(context.Background(), "hello", nil, nil); !errors.Is(err, ErrStreamIdleTimeout) {
		69	+	t.Fatalf("Prompt error = %v, want ErrStreamIdleTimeout", err)
		70	+	}
		71	+	}
		72	+
51	func TestAgentRetriesOverloadedStreamError(t *testing.T) {	73		func TestAgentRetriesOverloadedStreamError(t *testing.T) {
52	client := &retryFakeClient{}	74		client := &retryFakeClient{}
53	a := NewAgent(client, "fake-model", "system", Registry{})	75		a := NewAgent(client, "fake-model", "system", Registry{})