9265275 fix(subagents): harden deadline transcript recovery · zut

M packages/agent/subagents/agent.go +9 -3


@@ -445,11 +445,11 @@ func (a *Agent) setActivity(msg string) {
445	445		}
446	446
447	447		func (a *Agent) appendTranscript(chunk string) {
448		-	a.appendTranscriptLocked(chunk, "", false)
	448	+	a.appendTranscriptChunk(chunk, "", false)
449	449		}
450	450
451	451		func (a *Agent) appendUserMessage(text string) {
452		-	a.appendTranscriptLocked(text, "user: ", false)
	452	+	a.appendTranscriptChunk(text, "user: ", false)
453	453		}
454	454
455	455		func (a *Agent) appendAssistantMessage(text string) {

@@ -488,7 +488,7 @@ func (a *Agent) appendAssistantDelta(text string) {
488	488		a.markActivity(time.Now())
489	489		}
490	490
491		-	func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
	491	+	func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
492	492		if strings.TrimRight(chunk, "\n") == "" {
493	493		return
494	494		}

@@ -499,6 +499,12 @@ func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool)
499	499		a.markActivity(time.Now())
500	500		}
501	501
	502	+	func (a *Agent) resetStreamingAssistant() {
	503	+	a.mu.Lock()
	504	+	a.clearStreamingAssistantLocked()
	505	+	a.mu.Unlock()
	506	+	}
	507	+
502	508		func (a *Agent) replaceStreamingAssistantLocked(text string) {
503	509		a.removeStreamingAssistantLocked()
504	510		a.streamingAssistantStart = len(a.transcript)

 }
 func (a *Agent) appendTranscript(chunk string) {
-	a.appendTranscriptLocked(chunk, "", false)
+	a.appendTranscriptChunk(chunk, "", false)
 }
 func (a *Agent) appendUserMessage(text string) {
-	a.appendTranscriptLocked(text, "user: ", false)
+	a.appendTranscriptChunk(text, "user: ", false)
 }
 func (a *Agent) appendAssistantMessage(text string) {
 	a.markActivity(time.Now())
 }
-func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
+func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
 	if strings.TrimRight(chunk, "\n") == "" {
 		return
 	}
 	a.markActivity(time.Now())
 }
+func (a *Agent) resetStreamingAssistant() {
+	a.mu.Lock()
+	a.clearStreamingAssistantLocked()
+	a.mu.Unlock()
+}
 func (a *Agent) replaceStreamingAssistantLocked(text string) {
 	a.removeStreamingAssistantLocked()
 	a.streamingAssistantStart = len(a.transcript)

M packages/agent/subagents/agent_test.go +31 -1

@@ -1,6 +1,9 @@
1	1		package subagents
2	2
3		-	import "testing"
	3	+	import (
	4	+	"sync"
	5	+	"testing"
	6	+	)
4	7
5	8		func TestAgentAttemptValueAndTranscriptAccounting(t *testing.T) {
6	9		a := &Agent{maxOutputBytes: 6, maxOutputLines: 10}

@@ -47,3 +50,30 @@ func TestAgentTranscriptLineCapAccounting(t *testing.T) {
47	50		t.Fatalf("line-cap accounting = (%d bytes, %d lines), want (10, 2)", a.outputBytes, a.outputLines)
48	51		}
49	52		}
	53	+
	54	+	func TestAgentSnapshotConcurrentTranscriptTruncation(t *testing.T) {
	55	+	a := &Agent{maxOutputBytes: 8, maxOutputLines: 1}
	56	+	start := make(chan struct{})
	57	+	var wg sync.WaitGroup
	58	+	wg.Add(2)
	59	+	go func() {
	60	+	defer wg.Done()
	61	+	<-start
	62	+	for range 10_000 {
	63	+	a.appendTranscript("overflow")
	64	+	}
	65	+	}()
	66	+	go func() {
	67	+	defer wg.Done()
	68	+	<-start
	69	+	for range 10_000 {
	70	+	_ = a.Snapshot()
	71	+	}
	72	+	}()
	73	+	close(start)
	74	+	wg.Wait()
	75	+
	76	+	if !a.Snapshot().OutputTruncated {
	77	+	t.Fatal("snapshot did not report transcript truncation")
	78	+	}
	79	+	}

@@ -1,6 +1,9 @@
1		package subagents	1		package subagents
2			2
3	-	import "testing"	3	+	import (
			4	+	"sync"
			5	+	"testing"
			6	+	)
4			7
5		func TestAgentAttemptValueAndTranscriptAccounting(t *testing.T) {	8		func TestAgentAttemptValueAndTranscriptAccounting(t *testing.T) {
6		a := &Agent{maxOutputBytes: 6, maxOutputLines: 10}	9		a := &Agent{maxOutputBytes: 6, maxOutputLines: 10}

@@ -47,3 +50,30 @@ func TestAgentTranscriptLineCapAccounting(t *testing.T) {
47		t.Fatalf("line-cap accounting = (%d bytes, %d lines), want (10, 2)", a.outputBytes, a.outputLines)	50		t.Fatalf("line-cap accounting = (%d bytes, %d lines), want (10, 2)", a.outputBytes, a.outputLines)
48		}	51		}
49		}	52		}
			53	+
			54	+	func TestAgentSnapshotConcurrentTranscriptTruncation(t *testing.T) {
			55	+	a := &Agent{maxOutputBytes: 8, maxOutputLines: 1}
			56	+	start := make(chan struct{})
			57	+	var wg sync.WaitGroup
			58	+	wg.Add(2)
			59	+	go func() {
			60	+	defer wg.Done()
			61	+	<-start
			62	+	for range 10_000 {
			63	+	a.appendTranscript("overflow")
			64	+	}
			65	+	}()
			66	+	go func() {
			67	+	defer wg.Done()
			68	+	<-start
			69	+	for range 10_000 {
			70	+	_ = a.Snapshot()
			71	+	}
			72	+	}()
			73	+	close(start)
			74	+	wg.Wait()
			75	+
			76	+	if !a.Snapshot().OutputTruncated {
			77	+	t.Fatal("snapshot did not report transcript truncation")
			78	+	}
			79	+	}

M packages/agent/subagents/persist.go +4 -2


@@ -598,6 +598,7 @@ func resumePromptAcknowledged(events []Event, acceptedAt time.Time) bool {
598	598		func replayEventsIntoAgent(a *Agent, evs []Event) {
599	599		terminal := false
600	600		for _, ev := range evs {
	601	+	replayEventTranscript(a, ev)
601	602		switch ev.Type {
602	603		case EventTurnStarted, "turn_start":
603	604		turnID := ev.TurnID

@@ -627,8 +628,6 @@ func replayEventsIntoAgent(a *Agent, evs []Event) {
627	628		a.status = StatusDone
628	629		a.activity = "done (offline)"
629	630		a.mu.Unlock()
630		-	case "assistant_message", "user_message", "stdout", "stderr", "error":
631		-	replayEventTranscript(a, ev)
632	631		case "agent_stopped":
633	632		terminal = true
634	633		reason, _ := ev.Data["reason"].(string)

@@ -677,6 +676,9 @@ func replayTranscriptIntoAgent(a *Agent, evs []Event) {
677	676		}
678	677
679	678		func replayEventTranscript(a *Agent, ev Event) {
	679	+	if isAssistantStreamBoundary(ev.Type) {
	680	+	a.resetStreamingAssistant()
	681	+	}
680	682		switch ev.Type {
681	683		case "message.delta":
682	684		if delta, _ := ev.Data["delta"].(string); delta != "" {

 func replayEventsIntoAgent(a *Agent, evs []Event) {
 	terminal := false
 	for _, ev := range evs {
+		replayEventTranscript(a, ev)
 		switch ev.Type {
 		case EventTurnStarted, "turn_start":
 			turnID := ev.TurnID
 			a.status = StatusDone
 			a.activity = "done (offline)"
 			a.mu.Unlock()
-		case "assistant_message", "user_message", "stdout", "stderr", "error":
-			replayEventTranscript(a, ev)
 		case "agent_stopped":
 			terminal = true
 			reason, _ := ev.Data["reason"].(string)
 }
 func replayEventTranscript(a *Agent, ev Event) {
+	if isAssistantStreamBoundary(ev.Type) {
+		a.resetStreamingAssistant()
+	}
 	switch ev.Type {
 	case "message.delta":
 		if delta, _ := ev.Data["delta"].(string); delta != "" {

M packages/agent/subagents/result.go +9 -8


@@ -471,11 +471,12 @@ func (f Supervisor) ensureResult(a Agent, status Status, runErr error) {
471	471		result = nil
472	472		}
473	473		}
474		-	if result != nil && result.Output == "" && partialOutput != "" {
475		-	// A canceled worker turn may emit its terminal result before its
476		-	// provider turns accumulated text into a final assistant message.
	474	+	if result != nil && result.Output == "" && partialOutput != "" &&
	475	+	(result.Status == ResultFailed \|\| result.Status == ResultCanceled \|\| status == StatusFailed \|\| status == StatusKilled) {
	476	+	// A canceled or failed worker turn may emit its terminal result before
	477	+	// its provider turns accumulated text into a final assistant message.
477	478		// The event stream has already projected that text onto the Agent, so
478		-	// retain it in the durable result rather than returning an empty timeout.
	479	+	// retain it in the durable result rather than returning an empty failure.
479	480		result.Output = partialOutput
480	481		}
481	482		if result == nil {

@@ -521,10 +522,10 @@ func (f Supervisor) ensureResult(a Agent, status Status, runErr error) {
521	522		result.Status = ResultFailed
522	523		}
523	524		if errors.Is(runErr, context.DeadlineExceeded) {
524		-	// The worker was asked to stop cleanly before its process group was
525		-	// force-killed. Any streamed output received before that point remains
526		-	// in this result and HistoryRef(agentID), even when the worker's own
527		-	// cancellation result arrived before the parent classified the deadline.
	525	+	// The parent deadline is authoritative for terminal categorization, so
	526	+	// it intentionally overrides any child result error, including
	527	+	// context_limit. The worker was first asked to stop cleanly, and any
	528	+	// streamed output remains in this result and HistoryRef(agentID).
528	529		message := "subagent deadline exceeded; inspect history for received output"
529	530		if result.Output != "" {
530	531		message = "subagent deadline exceeded; partial output is preserved in the result and history"

 			result = nil
 		}
 	}
-	if result != nil && result.Output == "" && partialOutput != "" {
+	if result != nil && result.Output == "" && partialOutput != "" &&
-		// A canceled worker turn may emit its terminal result before its
+		(result.Status == ResultFailed || result.Status == ResultCanceled || status == StatusFailed || status == StatusKilled) {
-		// provider turns accumulated text into a final assistant message.
+		// A canceled or failed worker turn may emit its terminal result before
+		// its provider turns accumulated text into a final assistant message.
 		// The event stream has already projected that text onto the Agent, so
-		// retain it in the durable result rather than returning an empty timeout.
+		// retain it in the durable result rather than returning an empty failure.
 		result.Output = partialOutput
 	}
 	if result == nil {
 			result.Status = ResultFailed
 		}
 		if errors.Is(runErr, context.DeadlineExceeded) {
-			// The worker was asked to stop cleanly before its process group was
+			// The parent deadline is authoritative for terminal categorization, so
-			// force-killed. Any streamed output received before that point remains
+			// it intentionally overrides any child result error, including
-			// in this result and HistoryRef(agentID), even when the worker's own
+			// context_limit. The worker was first asked to stop cleanly, and any
-			// cancellation result arrived before the parent classified the deadline.
+			// streamed output remains in this result and HistoryRef(agentID).
 			message := "subagent deadline exceeded; inspect history for received output"
 			if result.Output != "" {
 				message = "subagent deadline exceeded; partial output is preserved in the result and history"

M packages/agent/subagents/result_test.go +27 -1


@@ -272,6 +272,7 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
272	272		name string
273	273		status Status
274	274		runErr error
	275	+	childCode string
275	276		wantState TurnStatus
276	277		wantCode string
277	278		wantOutput string

@@ -279,17 +280,24 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
279	280		{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed, wantCode: "runner_failed"},
280	281		{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
281	282		{name: "deadline preserves recovery guidance", status: StatusFailed, runErr: context.DeadlineExceeded, wantState: ResultFailed, wantCode: "deadline_exceeded", wantOutput: "partial answer"},
	283	+	{name: "deadline overrides child error", status: StatusFailed, runErr: context.DeadlineExceeded, childCode: "context_limit", wantState: ResultFailed, wantCode: "deadline_exceeded", wantOutput: "partial answer"},
282	284		{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
283	285		}
284	286		for _, tt := range tests {
285	287		t.Run(tt.name, func(t *testing.T) {
286	288		stateDir := t.TempDir()
	289	+	childStatus := ResultSucceeded
	290	+	var childError *ResultError
	291	+	if tt.childCode != "" {
	292	+	childStatus = ResultFailed
	293	+	childError = &ResultError{Code: tt.childCode, Message: "child error"}
	294	+	}
287	295		agent := &Agent{
288	296		ID: "agent-1",
289	297		stateDir: stateDir,
290	298		lastAssistant: tt.wantOutput,
291	299		result: &TurnResult{
292		-	AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
	300	+	AgentID: "agent-1", TurnID: "turn-1", Status: childStatus, Error: childError,
293	301		},
294	302		}
295	303		supervisor := &Supervisor{}

@@ -323,3 +331,21 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
323	331		})
324	332		}
325	333		}
	334	+
	335	+	func TestEnsureResultDoesNotBackfillSuccessfulEmptyOutput(t *testing.T) {
	336	+	stateDir := t.TempDir()
	337	+	agent := &Agent{
	338	+	ID: "agent-1",
	339	+	stateDir: stateDir,
	340	+	lastAssistant: "streamed text",
	341	+	result: &TurnResult{
	342	+	AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
	343	+	},
	344	+	}
	345	+
	346	+	(&Supervisor{}).ensureResult(agent, StatusDone, nil)
	347	+
	348	+	if result := agent.Result(); result == nil \|\| result.Output != "" {
	349	+	t.Fatalf("successful result = %#v, want intentionally empty output", result)
	350	+	}
	351	+	}

 		name       string
 		status     Status
 		runErr     error
+		childCode  string
 		wantState  TurnStatus
 		wantCode   string
 		wantOutput string
 		{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed, wantCode: "runner_failed"},
 		{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
 		{name: "deadline preserves recovery guidance", status: StatusFailed, runErr: context.DeadlineExceeded, wantState: ResultFailed, wantCode: "deadline_exceeded", wantOutput: "partial answer"},
+		{name: "deadline overrides child error", status: StatusFailed, runErr: context.DeadlineExceeded, childCode: "context_limit", wantState: ResultFailed, wantCode: "deadline_exceeded", wantOutput: "partial answer"},
 		{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			stateDir := t.TempDir()
+			childStatus := ResultSucceeded
+			var childError *ResultError
+			if tt.childCode != "" {
+				childStatus = ResultFailed
+				childError = &ResultError{Code: tt.childCode, Message: "child error"}
+			}
 			agent := &Agent{
 				ID:            "agent-1",
 				stateDir:      stateDir,
 				lastAssistant: tt.wantOutput,
 				result: &TurnResult{
-					AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
+					AgentID: "agent-1", TurnID: "turn-1", Status: childStatus, Error: childError,
 				},
 			}
 			supervisor := &Supervisor{}
 		})
 	}
 }
+func TestEnsureResultDoesNotBackfillSuccessfulEmptyOutput(t *testing.T) {
+	stateDir := t.TempDir()
+	agent := &Agent{
+		ID:            "agent-1",
+		stateDir:      stateDir,
+		lastAssistant: "streamed text",
+		result: &TurnResult{
+			AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
+		},
+	}
+	(&Supervisor{}).ensureResult(agent, StatusDone, nil)
+	if result := agent.Result(); result == nil || result.Output != "" {
+		t.Fatalf("successful result = %#v, want intentionally empty output", result)
+	}
+}

M packages/agent/subagents/runner.go +20 -2


@@ -5,6 +5,7 @@ import (
5	5		"bytes"
6	6		"context"
7	7		"encoding/json"
	8	+	"errors"
8	9		"fmt"
9	10		"os"
10	11		"os/exec"

@@ -478,7 +479,7 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
478	479		}
479	480		if ctxErr := ctx.Err(); ctxErr != nil {
480	481		reason := "cancelled"
481		-	if ctxErr == context.DeadlineExceeded {
	482	+	if errors.Is(ctxErr, context.DeadlineExceeded) {
482	483		reason = "deadline"
483	484		}
484	485		appendLog(NewEvent("agent_stopped", map[string]any{"reason": reason}))

@@ -511,7 +512,7 @@ func (r execRunner) stopOnContextDone(ctx context.Context, cmd exec.Cmd, runne
511	512		return
512	513		case <-ctx.Done():
513	514		}
514		-	if ctx.Err() != context.DeadlineExceeded {
	515	+	if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
515	516		if cmd.Cancel != nil {
516	517		_ = cmd.Cancel()
517	518		return

@@ -765,6 +766,15 @@ func notifyPromptTurnEnd(a *Agent, ev Event) {
765	766		go fn(int(step), errMsg)
766	767		}
767	768
	769	+	func isAssistantStreamBoundary(eventType string) bool {
	770	+	switch eventType {
	771	+	case EventTurnStarted, "turn_start", EventTurnResult, "turn_result", EventTurnFailed, "turn_failed", "turn_end":
	772	+	return true
	773	+	default:
	774	+	return false
	775	+	}
	776	+	}
	777	+
768	778		// applyEventToSink translates an Event into Sink updates. Only a
769	779		// few event types are interpreted; the rest still land in the
770	780		// durable log via the caller.

@@ -776,6 +786,14 @@ func applyEventToSink(ev Event, sink Sink) {
776	786		type streamingSink interface {
777	787		assistantDelta(string)
778	788		}
	789	+	type streamResetSink interface {
	790	+	resetStreamingAssistant()
	791	+	}
	792	+	if isAssistantStreamBoundary(ev.Type) {
	793	+	if streaming, ok := sink.(streamResetSink); ok {
	794	+	streaming.resetStreamingAssistant()
	795	+	}
	796	+	}
779	797		appendMessage := func(text string, assistant bool) {
780	798		if text == "" {
781	799		return

 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
 	}
 	if ctxErr := ctx.Err(); ctxErr != nil {
 		reason := "cancelled"
-		if ctxErr == context.DeadlineExceeded {
+		if errors.Is(ctxErr, context.DeadlineExceeded) {
 			reason = "deadline"
 		}
 		appendLog(NewEvent("agent_stopped", map[string]any{"reason": reason}))
 		return
 	case <-ctx.Done():
 	}
-	if ctx.Err() != context.DeadlineExceeded {
+	if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
 		if cmd.Cancel != nil {
 			_ = cmd.Cancel()
 			return
 	go fn(int(step), errMsg)
 }
+func isAssistantStreamBoundary(eventType string) bool {
+	switch eventType {
+	case EventTurnStarted, "turn_start", EventTurnResult, "turn_result", EventTurnFailed, "turn_failed", "turn_end":
+		return true
+	default:
+		return false
+	}
+}
 // applyEventToSink translates an Event into Sink updates. Only a
 // few event types are interpreted; the rest still land in the
 // durable log via the caller.
 	type streamingSink interface {
 		assistantDelta(string)
 	}
+	type streamResetSink interface {
+		resetStreamingAssistant()
+	}
+	if isAssistantStreamBoundary(ev.Type) {
+		if streaming, ok := sink.(streamResetSink); ok {
+			streaming.resetStreamingAssistant()
+		}
+	}
 	appendMessage := func(text string, assistant bool) {
 		if text == "" {
 			return

M packages/agent/subagents/runner_e2e_test.go +6 -4


@@ -25,6 +25,8 @@ func (s notifyingAgentSink) assistantDelta(text string) {
25	25		}
26	26		}
27	27
	28	+	const runnerProcessTimeout = 10 * time.Second
	29	+
28	30		type controlledDeadlineContext struct{ done chan struct{} }
29	31
30	32		func (c *controlledDeadlineContext) Deadline() (time.Time, bool) {

@@ -87,7 +89,7 @@ func TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput(t *te
87	89		go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
88	90		select {
89	91		case <-deltas:
90		-	case <-time.After(time.Second):
	92	+	case <-time.After(runnerProcessTimeout):
91	93		t.Fatal("worker did not emit its initial delta")
92	94		}
93	95		ctx.expire()

@@ -96,7 +98,7 @@ func TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput(t *te
96	98		if !errors.Is(err, context.DeadlineExceeded) {
97	99		t.Fatalf("Run error = %v, want context deadline exceeded", err)
98	100		}
99		-	case <-time.After(time.Second):
	101	+	case <-time.After(runnerProcessTimeout):
100	102		t.Fatal("Run did not return after deadline expiry")
101	103		}
102	104

@@ -166,7 +168,7 @@ func TestExecRunnerCancelsBlockedWorker(t *testing.T) {
166	168
167	169		select {
168	170		case <-deltas:
169		-	case <-time.After(time.Second):
	171	+	case <-time.After(runnerProcessTimeout):
170	172		t.Fatal("worker did not emit its initial delta")
171	173		}
172	174		cancel()

@@ -175,7 +177,7 @@ func TestExecRunnerCancelsBlockedWorker(t *testing.T) {
175	177		if !errors.Is(err, context.Canceled) {
176	178		t.Fatalf("Run error = %v, want context canceled", err)
177	179		}
178		-	case <-time.After(time.Second):
	180	+	case <-time.After(runnerProcessTimeout):
179	181		t.Fatal("Run did not return after context cancellation")
180	182		}
181	183		}

 	}
 }
+const runnerProcessTimeout = 10 * time.Second
 type controlledDeadlineContext struct{ done chan struct{} }
 func (c *controlledDeadlineContext) Deadline() (time.Time, bool) {
 	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
 	select {
 	case <-deltas:
-	case <-time.After(time.Second):
+	case <-time.After(runnerProcessTimeout):
 		t.Fatal("worker did not emit its initial delta")
 	}
 	ctx.expire()
 		if !errors.Is(err, context.DeadlineExceeded) {
 			t.Fatalf("Run error = %v, want context deadline exceeded", err)
 		}
-	case <-time.After(time.Second):
+	case <-time.After(runnerProcessTimeout):
 		t.Fatal("Run did not return after deadline expiry")
 	}
 	select {
 	case <-deltas:
-	case <-time.After(time.Second):
+	case <-time.After(runnerProcessTimeout):
 		t.Fatal("worker did not emit its initial delta")
 	}
 	cancel()
 		if !errors.Is(err, context.Canceled) {
 			t.Fatalf("Run error = %v, want context canceled", err)
 		}
-	case <-time.After(time.Second):
+	case <-time.After(runnerProcessTimeout):
 		t.Fatal("Run did not return after context cancellation")
 	}
 }

M packages/agent/subagents/runner_test.go +44 -0


@@ -112,6 +112,50 @@ func TestMessageDeltaBoundsLiveProjection(t *testing.T) {
112	112		if !snapshot.OutputTruncated {
113	113		t.Fatal("snapshot did not expose truncated streamed output")
114	114		}
	115	+
	116	+	events := []Event{
	117	+	NewEvent(EventMessageDelta, map[string]any{"delta": "first partial"}),
	118	+	NewEvent("turn_end", map[string]any{"stop": "cancelled"}),
	119	+	NewEvent(EventMessageDelta, map[string]any{"delta": "second partial"}),
	120	+	}
	121	+	want := "first partial\nsecond partial"
	122	+	assertProjection := func(t testing.T, agent Agent) {
	123	+	t.Helper()
	124	+	if got := strings.Join(agent.Transcript(), "\n"); got != want {
	125	+	t.Fatalf("transcript = %q, want %q", got, want)
	126	+	}
	127	+	agent.mu.Lock()
	128	+	truncated := agent.streamingAssistantTruncated
	129	+	agent.mu.Unlock()
	130	+	if truncated {
	131	+	t.Fatal("fresh second-turn stream was incorrectly truncated")
	132	+	}
	133	+	}
	134	+
	135	+	live := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
	136	+	for _, event := range events {
	137	+	applyEventToSink(event, agentSink{a: live})
	138	+	}
	139	+	assertProjection(t, live)
	140	+
	141	+	replayed := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
	142	+	replayTranscriptIntoAgent(replayed, events)
	143	+	assertProjection(t, replayed)
	144	+
	145	+	legacy := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
	146	+	replayEventsIntoAgent(legacy, events)
	147	+	assertProjection(t, legacy)
	148	+
	149	+	truncatedTurn := &Agent{maxOutputBytes: 16, maxOutputLines: 100, transcriptLoaded: true}
	150	+	applyEventToSink(NewEvent(EventMessageDelta, map[string]any{"delta": strings.Repeat("x", 64)}), agentSink{a: truncatedTurn})
	151	+	applyEventToSink(NewEvent("turn_end", map[string]any{"stop": "cancelled"}), agentSink{a: truncatedTurn})
	152	+	truncatedTurn.mu.Lock()
	153	+	streamText := truncatedTurn.streamingAssistantText
	154	+	streamTruncated := truncatedTurn.streamingAssistantTruncated
	155	+	truncatedTurn.mu.Unlock()
	156	+	if streamText != "" \|\| streamTruncated {
	157	+	t.Fatalf("stream state after turn boundary = (%q, %v), want empty and not truncated", streamText, streamTruncated)
	158	+	}
115	159		}
116	160
117	161		func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {


@@ -112,6 +112,50 @@ func TestMessageDeltaBoundsLiveProjection(t *testing.T) {
112	if !snapshot.OutputTruncated {	112		if !snapshot.OutputTruncated {
113	t.Fatal("snapshot did not expose truncated streamed output")	113		t.Fatal("snapshot did not expose truncated streamed output")
114	}	114		}
		115	+
		116	+	events := []Event{
		117	+	NewEvent(EventMessageDelta, map[string]any{"delta": "first partial"}),
		118	+	NewEvent("turn_end", map[string]any{"stop": "cancelled"}),
		119	+	NewEvent(EventMessageDelta, map[string]any{"delta": "second partial"}),
		120	+	}
		121	+	want := "first partial\nsecond partial"
		122	+	assertProjection := func(t testing.T, agent Agent) {
		123	+	t.Helper()
		124	+	if got := strings.Join(agent.Transcript(), "\n"); got != want {
		125	+	t.Fatalf("transcript = %q, want %q", got, want)
		126	+	}
		127	+	agent.mu.Lock()
		128	+	truncated := agent.streamingAssistantTruncated
		129	+	agent.mu.Unlock()
		130	+	if truncated {
		131	+	t.Fatal("fresh second-turn stream was incorrectly truncated")
		132	+	}
		133	+	}
		134	+
		135	+	live := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
		136	+	for _, event := range events {
		137	+	applyEventToSink(event, agentSink{a: live})
		138	+	}
		139	+	assertProjection(t, live)
		140	+
		141	+	replayed := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
		142	+	replayTranscriptIntoAgent(replayed, events)
		143	+	assertProjection(t, replayed)
		144	+
		145	+	legacy := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
		146	+	replayEventsIntoAgent(legacy, events)
		147	+	assertProjection(t, legacy)
		148	+
		149	+	truncatedTurn := &Agent{maxOutputBytes: 16, maxOutputLines: 100, transcriptLoaded: true}
		150	+	applyEventToSink(NewEvent(EventMessageDelta, map[string]any{"delta": strings.Repeat("x", 64)}), agentSink{a: truncatedTurn})
		151	+	applyEventToSink(NewEvent("turn_end", map[string]any{"stop": "cancelled"}), agentSink{a: truncatedTurn})
		152	+	truncatedTurn.mu.Lock()
		153	+	streamText := truncatedTurn.streamingAssistantText
		154	+	streamTruncated := truncatedTurn.streamingAssistantTruncated
		155	+	truncatedTurn.mu.Unlock()
		156	+	if streamText != "" \|\| streamTruncated {
		157	+	t.Fatalf("stream state after turn boundary = (%q, %v), want empty and not truncated", streamText, streamTruncated)
		158	+	}
115	}	159		}
116		160
117	func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {	161		func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {

M packages/agent/subagents/supervisor.go +2 -2


@@ -1133,7 +1133,7 @@ func (a *Agent) Snapshot() AgentSnapshot {
1133	1133		status := a.status
1134	1134		activity := a.activity
1135	1135		lastAssistant := a.lastAssistant
1136		-	streamingOutputTruncated := a.streamingAssistantTruncated
	1136	+	outputTruncated := a.outputTruncated \|\| a.streamingAssistantTruncated
1137	1137		finished := a.finished
1138	1138		a.mu.Unlock()
1139	1139

@@ -1148,7 +1148,6 @@ func (a *Agent) Snapshot() AgentSnapshot {
1148	1148		resultRef := a.resultRef
1149	1149		patchRef := a.patchRef
1150	1150		changedFiles := append([]string(nil), a.changedFiles...)
1151		-	outputTruncated := a.outputTruncated \|\| streamingOutputTruncated
1152	1151		a.lifecycleMu.Unlock()
1153	1152		return AgentSnapshot{
1154	1153		ID: a.ID, Task: a.Task, Dir: a.Dir,

@@ -1214,6 +1213,7 @@ func (s agentSink) Transcript(chunk string) {
1214	1213		func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1215	1214		func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1216	1215		func (s agentSink) assistantDelta(text string) { s.a.appendAssistantDelta(text) }
	1216	+	func (s agentSink) resetStreamingAssistant() { s.a.resetStreamingAssistant() }
1217	1217
1218	1218		func truncate(s string, n int) string {
1219	1219		runes := []rune(s)

 	status := a.status
 	activity := a.activity
 	lastAssistant := a.lastAssistant
-	streamingOutputTruncated := a.streamingAssistantTruncated
+	outputTruncated := a.outputTruncated || a.streamingAssistantTruncated
 	finished := a.finished
 	a.mu.Unlock()
 	resultRef := a.resultRef
 	patchRef := a.patchRef
 	changedFiles := append([]string(nil), a.changedFiles...)
-	outputTruncated := a.outputTruncated || streamingOutputTruncated
 	a.lifecycleMu.Unlock()
 	return AgentSnapshot{
 		ID: a.ID, Task: a.Task, Dir: a.Dir,
 func (s agentSink) userMessage(text string)      { s.a.appendUserMessage(text) }
 func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
 func (s agentSink) assistantDelta(text string)   { s.a.appendAssistantDelta(text) }
+func (s agentSink) resetStreamingAssistant()     { s.a.resetStreamingAssistant() }
 func truncate(s string, n int) string {
 	runes := []rune(s)