6c26357 Merge pull request #72 from bnema/fix/subagent-timeout-transcript-recovery · zut

M docs/subagents.md +4 -0


@@ -125,6 +125,10 @@ subagent://<id>/result
125	125		subagent://<id>/patch
126	126		```
127	127
	128	+	`events.jsonl` is appended as the worker emits events, including partial `message.delta` output. The dashboard and `subagent://<id>/history` replay those deltas, so output received before a worker finishes remains recoverable rather than waiting for a final assistant message.
	129	+
	130	+	When a subagent deadline expires, the supervisor requests worker shutdown and allows the configured grace period for the worker to cancel its turn and write its session and result. The terminal result remains a failure with error code `deadline_exceeded`, preserves any output already received, and directs the caller to its history. If the worker does not stop during that grace period, the supervisor forcefully cancels it to preserve the timeout limit.
	131	+
128	132		Use `/subagents resume-session <id>` to continue the existing session without replaying its original task. Use `/subagents restart-task <id>` only when intentionally starting the stored task again. Cancellation requests graceful shutdown first, then forcefully cancels after the configured grace period.
129	133
130	134		When auto-subagents is enabled and permitted by launch-time policy, the model can manage workers with `subagent_status`, `subagent_stop`, and `subagent_resume`. Call `subagent_status` with `{}` to list workers visible to the active supervisor session, or pass `{"agent_id":"<id-or-unique-prefix>"}` to query one worker. Queries use in-process snapshots and never wait for a worker turn or process to finish; repeated calls or state inspection are not completion signals. The JSON response contains the worker id, a normalized lifecycle state (`starting`, `running`, `completed`, `failed`, `cancelled`, or `detached`), the underlying `process_state` and `turn_state`, start/update/finish timestamps, a bounded first-line task summary, and terminal-result metadata/reference when available. It intentionally omits prompts, transcripts, result output, credentials, provider settings, and filesystem paths.


@@ -125,6 +125,10 @@ subagent://<id>/result
125	subagent://<id>/patch	125		subagent://<id>/patch
126	```	126		```
127		127
		128	+	`events.jsonl` is appended as the worker emits events, including partial `message.delta` output. The dashboard and `subagent://<id>/history` replay those deltas, so output received before a worker finishes remains recoverable rather than waiting for a final assistant message.
		129	+
		130	+	When a subagent deadline expires, the supervisor requests worker shutdown and allows the configured grace period for the worker to cancel its turn and write its session and result. The terminal result remains a failure with error code `deadline_exceeded`, preserves any output already received, and directs the caller to its history. If the worker does not stop during that grace period, the supervisor forcefully cancels it to preserve the timeout limit.
		131	+
128	Use `/subagents resume-session <id>` to continue the existing session without replaying its original task. Use `/subagents restart-task <id>` only when intentionally starting the stored task again. Cancellation requests graceful shutdown first, then forcefully cancels after the configured grace period.	132		Use `/subagents resume-session <id>` to continue the existing session without replaying its original task. Use `/subagents restart-task <id>` only when intentionally starting the stored task again. Cancellation requests graceful shutdown first, then forcefully cancels after the configured grace period.
129		133
130	When auto-subagents is enabled and permitted by launch-time policy, the model can manage workers with `subagent_status`, `subagent_stop`, and `subagent_resume`. Call `subagent_status` with `{}` to list workers visible to the active supervisor session, or pass `{"agent_id":"<id-or-unique-prefix>"}` to query one worker. Queries use in-process snapshots and never wait for a worker turn or process to finish; repeated calls or state inspection are not completion signals. The JSON response contains the worker id, a normalized lifecycle state (`starting`, `running`, `completed`, `failed`, `cancelled`, or `detached`), the underlying `process_state` and `turn_state`, start/update/finish timestamps, a bounded first-line task summary, and terminal-result metadata/reference when available. It intentionally omits prompts, transcripts, result output, credentials, provider settings, and filesystem paths.	134		When auto-subagents is enabled and permitted by launch-time policy, the model can manage workers with `subagent_status`, `subagent_stop`, and `subagent_resume`. Call `subagent_status` with `{}` to list workers visible to the active supervisor session, or pass `{"agent_id":"<id-or-unique-prefix>"}` to query one worker. Queries use in-process snapshots and never wait for a worker turn or process to finish; repeated calls or state inspection are not completion signals. The JSON response contains the worker id, a normalized lifecycle state (`starting`, `running`, `completed`, `failed`, `cancelled`, or `detached`), the underlying `process_state` and `turn_state`, start/update/finish timestamps, a bounded first-line task summary, and terminal-result metadata/reference when available. It intentionally omits prompts, transcripts, result output, credentials, provider settings, and filesystem paths.

M packages/agent/modes/auto_subagents_test.go +3 -1


@@ -13,6 +13,8 @@ import (
13	13		"github.com/bnema/zut/packages/provider"
14	14		)
15	15
	16	+	const autoSubagentsTestTimeout = 10 * time.Second
	17	+
16	18		func TestAutoSubagentsSystemPromptTogglesProfileManifestWithDelegationGuidance(t *testing.T) {
17	19		const onDemand = "delegate only on an explicit user request"
18	20		iv := &Interactive{

@@ -332,7 +334,7 @@ func newQueuedAutoSubagentsInteractive() *Interactive {
332	334
333	335		func waitForQueuedPrompt(t testing.T, iv Interactive) string {
334	336		t.Helper()
335		-	deadline := time.NewTimer(time.Second)
	337	+	deadline := time.NewTimer(autoSubagentsTestTimeout)
336	338		defer deadline.Stop()
337	339		ticker := time.NewTicker(time.Millisecond)
338	340		defer ticker.Stop()

 	"github.com/bnema/zut/packages/provider"
 )
+const autoSubagentsTestTimeout = 10 * time.Second
 func TestAutoSubagentsSystemPromptTogglesProfileManifestWithDelegationGuidance(t *testing.T) {
 	const onDemand = "delegate only on an explicit user request"
 	iv := &Interactive{
 func waitForQueuedPrompt(t *testing.T, iv *Interactive) string {
 	t.Helper()
-	deadline := time.NewTimer(time.Second)
+	deadline := time.NewTimer(autoSubagentsTestTimeout)
 	defer deadline.Stop()
 	ticker := time.NewTicker(time.Millisecond)
 	defer ticker.Stop()

M packages/agent/modes/session_dialog_test.go +8 -6


@@ -13,6 +13,8 @@ import (
13	13		"github.com/bnema/zut/packages/tui"
14	14		)
15	15
	16	+	const sessionDialogTestTimeout = 10 * time.Second
	17	+
16	18		func TestFormatSessionRowPlainSanitizesControlBytes(t *testing.T) {
17	19		row := formatSessionRowPlain(core.SessionSummary{
18	20		Provider: "test\x1b]0;bad\a",

@@ -149,12 +151,12 @@ func TestSessionDialogCancellationDoesNotEmitFinished(t *testing.T) {
149	151		t.Fatalf("first load event = %+v, open %v; want sessionLoadStarted", event, ok)
150	152		}
151	153		d.ApplyLoad(event)
152		-	case <-time.After(time.Second):
	154	+	case <-time.After(sessionDialogTestTimeout):
153	155		t.Fatal("timed out waiting for sessionLoadStarted")
154	156		}
155	157		select {
156	158		case <-entered:
157		-	case <-time.After(time.Second):
	159	+	case <-time.After(sessionDialogTestTimeout):
158	160		t.Fatal("timed out waiting for the session worker")
159	161		}
160	162

@@ -173,7 +175,7 @@ func TestSessionDialogCancellationDoesNotEmitFinished(t *testing.T) {
173	175		t.Fatal("canceled load emitted sessionLoadFinished")
174	176		}
175	177		d.ApplyLoad(event)
176		-	case <-time.After(time.Second):
	178	+	case <-time.After(sessionDialogTestTimeout):
177	179		t.Fatal("timed out waiting for canceled session load to close")
178	180		}
179	181		}

@@ -221,12 +223,12 @@ func TestSessionDialogPreCanceledOpenPreservesPreviousLoadBarrier(t *testing.T)
221	223		t.Fatalf("first load event = %+v, open %v; want sessionLoadStarted", event, ok)
222	224		}
223	225		d.ApplyLoad(event)
224		-	case <-time.After(time.Second):
	226	+	case <-time.After(sessionDialogTestTimeout):
225	227		t.Fatal("timed out waiting for the first load to start")
226	228		}
227	229		select {
228	230		case <-entered:
229		-	case <-time.After(time.Second):
	231	+	case <-time.After(sessionDialogTestTimeout):
230	232		t.Fatal("timed out waiting for the first worker")
231	233		}
232	234

@@ -253,7 +255,7 @@ func TestSessionDialogPreCanceledOpenPreservesPreviousLoadBarrier(t *testing.T)
253	255		t.Fatalf("third load event = %+v, open %v; want sessionLoadStarted after barrier", event, ok)
254	256		}
255	257		d.ApplyLoad(event)
256		-	case <-time.After(time.Second):
	258	+	case <-time.After(sessionDialogTestTimeout):
257	259		t.Fatal("timed out waiting for third load after releasing the first")
258	260		}
259	261		cancelThird()

 	"github.com/bnema/zut/packages/tui"
 )
+const sessionDialogTestTimeout = 10 * time.Second
 func TestFormatSessionRowPlainSanitizesControlBytes(t *testing.T) {
 	row := formatSessionRowPlain(core.SessionSummary{
 		Provider:      "test\x1b]0;bad\a",
 			t.Fatalf("first load event = %+v, open %v; want sessionLoadStarted", event, ok)
 		}
 		d.ApplyLoad(event)
-	case <-time.After(time.Second):
+	case <-time.After(sessionDialogTestTimeout):
 		t.Fatal("timed out waiting for sessionLoadStarted")
 	}
 	select {
 	case <-entered:
-	case <-time.After(time.Second):
+	case <-time.After(sessionDialogTestTimeout):
 		t.Fatal("timed out waiting for the session worker")
 	}
 				t.Fatal("canceled load emitted sessionLoadFinished")
 			}
 			d.ApplyLoad(event)
-		case <-time.After(time.Second):
+		case <-time.After(sessionDialogTestTimeout):
 			t.Fatal("timed out waiting for canceled session load to close")
 		}
 	}
 			t.Fatalf("first load event = %+v, open %v; want sessionLoadStarted", event, ok)
 		}
 		d.ApplyLoad(event)
-	case <-time.After(time.Second):
+	case <-time.After(sessionDialogTestTimeout):
 		t.Fatal("timed out waiting for the first load to start")
 	}
 	select {
 	case <-entered:
-	case <-time.After(time.Second):
+	case <-time.After(sessionDialogTestTimeout):
 		t.Fatal("timed out waiting for the first worker")
 	}
 			t.Fatalf("third load event = %+v, open %v; want sessionLoadStarted after barrier", event, ok)
 		}
 		d.ApplyLoad(event)
-	case <-time.After(time.Second):
+	case <-time.After(sessionDialogTestTimeout):
 		t.Fatal("timed out waiting for third load after releasing the first")
 	}
 	cancelThird()

M packages/agent/subagents/agent.go +133 -33


@@ -116,16 +116,20 @@ type Agent struct {
116	116		// Supervisor.Spawn so callers do not need to manage the socket directly.
117	117		inbox *Inbox
118	118
119		-	mu sync.Mutex
120		-	status Status
121		-	activity string
122		-	transcript []string
123		-	lastAssistant string
124		-	finished time.Time
125		-	lastErr error
126		-	transcriptLoaded bool
127		-	legacyEventState bool
128		-	transcriptMu sync.Mutex
	119	+	mu sync.Mutex
	120	+	status Status
	121	+	activity string
	122	+	transcript []string
	123	+	lastAssistant string
	124	+	streamingAssistantText string
	125	+	streamingAssistantTruncated bool
	126	+	streamingAssistantStart int
	127	+	streamingAssistantLines int
	128	+	finished time.Time
	129	+	lastErr error
	130	+	transcriptLoaded bool
	131	+	legacyEventState bool
	132	+	transcriptMu sync.Mutex
129	133
130	134		lifecycleMu sync.Mutex
131	135		processState ProcessState

@@ -441,25 +445,114 @@ func (a *Agent) setActivity(msg string) {
441	445		}
442	446
443	447		func (a *Agent) appendTranscript(chunk string) {
444		-	a.appendTranscriptLocked(chunk, "", false)
	448	+	a.appendTranscriptChunk(chunk, "", false)
445	449		}
446	450
447	451		func (a *Agent) appendUserMessage(text string) {
448		-	a.appendTranscriptLocked(text, "user: ", false)
	452	+	a.appendTranscriptChunk(text, "user: ", false)
449	453		}
450	454
451	455		func (a *Agent) appendAssistantMessage(text string) {
452		-	a.appendTranscriptLocked(text, "", true)
	456	+	if text == "" {
	457	+	return
	458	+	}
	459	+	a.mu.Lock()
	460	+	if a.streamingAssistantText != "" {
	461	+	a.replaceStreamingAssistantLocked(text)
	462	+	a.clearStreamingAssistantLocked()
	463	+	} else {
	464	+	a.appendTranscriptLinesLocked(text, "", true)
	465	+	a.boundTranscriptLocked()
	466	+	}
	467	+	a.mu.Unlock()
	468	+	a.markActivity(time.Now())
	469	+	}
	470	+
	471	+	// appendAssistantDelta exposes an in-progress assistant response in the
	472	+	// dashboard without duplicating it when the final assistant_message arrives.
	473	+	// The JSONL event log remains the durable source; this maintains its live and
	474	+	// replayed in-memory projection.
	475	+	func (a *Agent) appendAssistantDelta(text string) {
	476	+	if text == "" {
	477	+	return
	478	+	}
	479	+	a.mu.Lock()
	480	+	if !a.streamingAssistantTruncated {
	481	+	value := a.streamingAssistantText + text
	482	+	byteCap, lineCap := a.outputLimits()
	483	+	a.streamingAssistantText = boundInlineText(value, byteCap, lineCap)
	484	+	a.streamingAssistantTruncated = a.streamingAssistantText != value
	485	+	}
	486	+	a.replaceStreamingAssistantLocked(a.streamingAssistantText)
	487	+	a.mu.Unlock()
	488	+	a.markActivity(time.Now())
	489	+	}
	490	+
	491	+	func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
	492	+	if strings.TrimRight(chunk, "\n") == "" {
	493	+	return
	494	+	}
	495	+	a.mu.Lock()
	496	+	a.appendTranscriptLinesLocked(chunk, linePrefix, assistant)
	497	+	a.boundTranscriptLocked()
	498	+	a.mu.Unlock()
	499	+	a.markActivity(time.Now())
	500	+	}
	501	+
	502	+	func (a *Agent) resetStreamingAssistant() {
	503	+	a.mu.Lock()
	504	+	a.clearStreamingAssistantLocked()
	505	+	a.mu.Unlock()
	506	+	}
	507	+
	508	+	func (a *Agent) replaceStreamingAssistantLocked(text string) {
	509	+	a.removeStreamingAssistantLocked()
	510	+	a.streamingAssistantStart = len(a.transcript)
	511	+	a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
	512	+	a.boundTranscriptLocked()
	513	+	}
	514	+
	515	+	func (a *Agent) removeStreamingAssistantLocked() {
	516	+	if a.streamingAssistantLines == 0 {
	517	+	return
	518	+	}
	519	+	start := a.streamingAssistantStart
	520	+	if start < 0 {
	521	+	start = 0
	522	+	}
	523	+	if start > len(a.transcript) {
	524	+	start = len(a.transcript)
	525	+	}
	526	+	end := start + a.streamingAssistantLines
	527	+	if end > len(a.transcript) {
	528	+	end = len(a.transcript)
	529	+	}
	530	+	for _, line := range a.transcript[start:end] {
	531	+	a.outputBytes -= len(line) + 1
	532	+	a.outputLines--
	533	+	}
	534	+	a.transcript = append(a.transcript[:start], a.transcript[end:]...)
	535	+	a.streamingAssistantLines = 0
	536	+	if a.outputLines == 0 {
	537	+	a.outputBytes = 0
	538	+	}
	539	+	}
	540	+
	541	+	func (a *Agent) clearStreamingAssistantLocked() {
	542	+	a.streamingAssistantText = ""
	543	+	a.streamingAssistantTruncated = false
	544	+	a.streamingAssistantStart = 0
	545	+	a.streamingAssistantLines = 0
453	546		}
454	547
455		-	func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
	548	+	func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
456	549		message := chunk
457	550		chunk = strings.TrimRight(chunk, "\n")
458	551		if chunk == "" {
459		-	return
	552	+	return 0
460	553		}
461		-	a.mu.Lock()
462		-	for _, line := range strings.Split(chunk, "\n") {
	554	+	lines := strings.Split(chunk, "\n")
	555	+	for _, line := range lines {
463	556		line = linePrefix + line
464	557		a.transcript = append(a.transcript, line)
465	558		a.outputBytes += len(line) + 1

@@ -468,35 +561,42 @@ func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool)
468	561		if assistant {
469	562		a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
470	563		}
471		-	// Bound the inline dashboard copy. The durable event log and session
472		-	// remain available through the logical history reference.
473		-	lineCap := a.maxOutputLines
474		-	if lineCap <= 0 {
475		-	lineCap = 2_000
476		-	}
477		-	byteCap := a.maxOutputBytes
	564	+	return len(lines)
	565	+	}
	566	+
	567	+	func (a *Agent) outputLimits() (byteCap, lineCap int) {
	568	+	byteCap = a.maxOutputBytes
478	569		if byteCap <= 0 {
479	570		byteCap = 500_000
480	571		}
481		-	for a.outputLines > lineCap {
482		-	line := a.transcript[0]
483		-	a.transcript = a.transcript[1:]
484		-	a.outputBytes -= len(line) + 1
485		-	a.outputLines--
486		-	a.outputTruncated = true
	572	+	lineCap = a.maxOutputLines
	573	+	if lineCap <= 0 {
	574	+	lineCap = 2_000
487	575		}
488		-	for a.outputLines > 0 && a.outputBytes > byteCap {
	576	+	return byteCap, lineCap
	577	+	}
	578	+
	579	+	func (a *Agent) boundTranscriptLocked() {
	580	+	// Bound the inline dashboard copy. The durable event log and session
	581	+	// remain available through the logical history reference.
	582	+	byteCap, lineCap := a.outputLimits()
	583	+	for a.outputLines > lineCap \|\| (a.outputLines > 0 && a.outputBytes > byteCap) {
489	584		line := a.transcript[0]
490	585		a.transcript = a.transcript[1:]
491	586		a.outputBytes -= len(line) + 1
492	587		a.outputLines--
	588	+	if a.streamingAssistantLines > 0 {
	589	+	if a.streamingAssistantStart > 0 {
	590	+	a.streamingAssistantStart--
	591	+	} else {
	592	+	a.streamingAssistantLines--
	593	+	}
	594	+	}
493	595		a.outputTruncated = true
494	596		}
495	597		if a.outputLines == 0 {
496	598		a.outputBytes = 0
497	599		}
498		-	a.mu.Unlock()
499		-	a.markActivity(time.Now())
500	600		}
501	601
502	602		// newAgentID returns a short, mostly-collision-free identifier of the


@@ -116,16 +116,20 @@ type Agent struct {
116		// Supervisor.Spawn so callers do not need to manage the socket directly.	116		// Supervisor.Spawn so callers do not need to manage the socket directly.
117		inbox *Inbox	117		inbox *Inbox
118			118
119	-	mu sync.Mutex	119	+	mu sync.Mutex
120	-	status Status	120	+	status Status
121	-	activity string	121	+	activity string
122	-	transcript []string	122	+	transcript []string
123	-	lastAssistant string	123	+	lastAssistant string
124	-	finished time.Time	124	+	streamingAssistantText string
125	-	lastErr error	125	+	streamingAssistantTruncated bool
126	-	transcriptLoaded bool	126	+	streamingAssistantStart int
127	-	legacyEventState bool	127	+	streamingAssistantLines int
128	-	transcriptMu sync.Mutex	128	+	finished time.Time
			129	+	lastErr error
			130	+	transcriptLoaded bool
			131	+	legacyEventState bool
			132	+	transcriptMu sync.Mutex
129			133
130		lifecycleMu sync.Mutex	134		lifecycleMu sync.Mutex
131		processState ProcessState	135		processState ProcessState

@@ -441,25 +445,114 @@ func (a *Agent) setActivity(msg string) {
441		}	445		}
442			446
443		func (a *Agent) appendTranscript(chunk string) {	447		func (a *Agent) appendTranscript(chunk string) {
444	-	a.appendTranscriptLocked(chunk, "", false)	448	+	a.appendTranscriptChunk(chunk, "", false)
445		}	449		}
446			450
447		func (a *Agent) appendUserMessage(text string) {	451		func (a *Agent) appendUserMessage(text string) {
448	-	a.appendTranscriptLocked(text, "user: ", false)	452	+	a.appendTranscriptChunk(text, "user: ", false)
449		}	453		}
450			454
451		func (a *Agent) appendAssistantMessage(text string) {	455		func (a *Agent) appendAssistantMessage(text string) {
452	-	a.appendTranscriptLocked(text, "", true)	456	+	if text == "" {
			457	+	return
			458	+	}
			459	+	a.mu.Lock()
			460	+	if a.streamingAssistantText != "" {
			461	+	a.replaceStreamingAssistantLocked(text)
			462	+	a.clearStreamingAssistantLocked()
			463	+	} else {
			464	+	a.appendTranscriptLinesLocked(text, "", true)
			465	+	a.boundTranscriptLocked()
			466	+	}
			467	+	a.mu.Unlock()
			468	+	a.markActivity(time.Now())
			469	+	}
			470	+
			471	+	// appendAssistantDelta exposes an in-progress assistant response in the
			472	+	// dashboard without duplicating it when the final assistant_message arrives.
			473	+	// The JSONL event log remains the durable source; this maintains its live and
			474	+	// replayed in-memory projection.
			475	+	func (a *Agent) appendAssistantDelta(text string) {
			476	+	if text == "" {
			477	+	return
			478	+	}
			479	+	a.mu.Lock()
			480	+	if !a.streamingAssistantTruncated {
			481	+	value := a.streamingAssistantText + text
			482	+	byteCap, lineCap := a.outputLimits()
			483	+	a.streamingAssistantText = boundInlineText(value, byteCap, lineCap)
			484	+	a.streamingAssistantTruncated = a.streamingAssistantText != value
			485	+	}
			486	+	a.replaceStreamingAssistantLocked(a.streamingAssistantText)
			487	+	a.mu.Unlock()
			488	+	a.markActivity(time.Now())
			489	+	}
			490	+
			491	+	func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
			492	+	if strings.TrimRight(chunk, "\n") == "" {
			493	+	return
			494	+	}
			495	+	a.mu.Lock()
			496	+	a.appendTranscriptLinesLocked(chunk, linePrefix, assistant)
			497	+	a.boundTranscriptLocked()
			498	+	a.mu.Unlock()
			499	+	a.markActivity(time.Now())
			500	+	}
			501	+
			502	+	func (a *Agent) resetStreamingAssistant() {
			503	+	a.mu.Lock()
			504	+	a.clearStreamingAssistantLocked()
			505	+	a.mu.Unlock()
			506	+	}
			507	+
			508	+	func (a *Agent) replaceStreamingAssistantLocked(text string) {
			509	+	a.removeStreamingAssistantLocked()
			510	+	a.streamingAssistantStart = len(a.transcript)
			511	+	a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
			512	+	a.boundTranscriptLocked()
			513	+	}
			514	+
			515	+	func (a *Agent) removeStreamingAssistantLocked() {
			516	+	if a.streamingAssistantLines == 0 {
			517	+	return
			518	+	}
			519	+	start := a.streamingAssistantStart
			520	+	if start < 0 {
			521	+	start = 0
			522	+	}
			523	+	if start > len(a.transcript) {
			524	+	start = len(a.transcript)
			525	+	}
			526	+	end := start + a.streamingAssistantLines
			527	+	if end > len(a.transcript) {
			528	+	end = len(a.transcript)
			529	+	}
			530	+	for _, line := range a.transcript[start:end] {
			531	+	a.outputBytes -= len(line) + 1
			532	+	a.outputLines--
			533	+	}
			534	+	a.transcript = append(a.transcript[:start], a.transcript[end:]...)
			535	+	a.streamingAssistantLines = 0
			536	+	if a.outputLines == 0 {
			537	+	a.outputBytes = 0
			538	+	}
			539	+	}
			540	+
			541	+	func (a *Agent) clearStreamingAssistantLocked() {
			542	+	a.streamingAssistantText = ""
			543	+	a.streamingAssistantTruncated = false
			544	+	a.streamingAssistantStart = 0
			545	+	a.streamingAssistantLines = 0
453		}	546		}
454			547
455	-	func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {	548	+	func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
456		message := chunk	549		message := chunk
457		chunk = strings.TrimRight(chunk, "\n")	550		chunk = strings.TrimRight(chunk, "\n")
458		if chunk == "" {	551		if chunk == "" {
459	-	return	552	+	return 0
460		}	553		}
461	-	a.mu.Lock()	554	+	lines := strings.Split(chunk, "\n")
462	-	for _, line := range strings.Split(chunk, "\n") {	555	+	for _, line := range lines {
463		line = linePrefix + line	556		line = linePrefix + line
464		a.transcript = append(a.transcript, line)	557		a.transcript = append(a.transcript, line)
465		a.outputBytes += len(line) + 1	558		a.outputBytes += len(line) + 1

@@ -468,35 +561,42 @@ func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool)
468		if assistant {	561		if assistant {
469		a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)	562		a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
470		}	563		}
471	-	// Bound the inline dashboard copy. The durable event log and session	564	+	return len(lines)
472	-	// remain available through the logical history reference.	565	+	}
473	-	lineCap := a.maxOutputLines	566	+
474	-	if lineCap <= 0 {	567	+	func (a *Agent) outputLimits() (byteCap, lineCap int) {
475	-	lineCap = 2_000	568	+	byteCap = a.maxOutputBytes
476	-	}
477	-	byteCap := a.maxOutputBytes
478		if byteCap <= 0 {	569		if byteCap <= 0 {
479		byteCap = 500_000	570		byteCap = 500_000
480		}	571		}
481	-	for a.outputLines > lineCap {	572	+	lineCap = a.maxOutputLines
482	-	line := a.transcript[0]	573	+	if lineCap <= 0 {
483	-	a.transcript = a.transcript[1:]	574	+	lineCap = 2_000
484	-	a.outputBytes -= len(line) + 1
485	-	a.outputLines--
486	-	a.outputTruncated = true
487		}	575		}
488	-	for a.outputLines > 0 && a.outputBytes > byteCap {	576	+	return byteCap, lineCap
			577	+	}
			578	+
			579	+	func (a *Agent) boundTranscriptLocked() {
			580	+	// Bound the inline dashboard copy. The durable event log and session
			581	+	// remain available through the logical history reference.
			582	+	byteCap, lineCap := a.outputLimits()
			583	+	for a.outputLines > lineCap \|\| (a.outputLines > 0 && a.outputBytes > byteCap) {
489		line := a.transcript[0]	584		line := a.transcript[0]
490		a.transcript = a.transcript[1:]	585		a.transcript = a.transcript[1:]
491		a.outputBytes -= len(line) + 1	586		a.outputBytes -= len(line) + 1
492		a.outputLines--	587		a.outputLines--
			588	+	if a.streamingAssistantLines > 0 {
			589	+	if a.streamingAssistantStart > 0 {
			590	+	a.streamingAssistantStart--
			591	+	} else {
			592	+	a.streamingAssistantLines--
			593	+	}
			594	+	}
493		a.outputTruncated = true	595		a.outputTruncated = true
494		}	596		}
495		if a.outputLines == 0 {	597		if a.outputLines == 0 {
496		a.outputBytes = 0	598		a.outputBytes = 0
497		}	599		}
498	-	a.mu.Unlock()
499	-	a.markActivity(time.Now())
500		}	600		}
501			601
502		// newAgentID returns a short, mostly-collision-free identifier of the	602		// newAgentID returns a short, mostly-collision-free identifier of the

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 +8 -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,7 +676,14 @@ 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 {
	683	+	case "message.delta":
	684	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
	685	+	a.appendAssistantDelta(delta)
	686	+	}
681	687		case "assistant_message", "user_message":
682	688		var text []string
683	689		if c, ok := ev.Data["content"].([]any); ok {

 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 != "" {
+			a.appendAssistantDelta(delta)
+		}
 	case "assistant_message", "user_message":
 		var text []string
 		if c, ok := ev.Data["content"].([]any); ok {

M packages/agent/subagents/result.go +26 -4


@@ -443,6 +443,9 @@ func (f Supervisor) ensureResult(a Agent, status Status, runErr error) {
443	443		return
444	444		}
445	445		attempt := a.AttemptValue()
	446	+	a.mu.Lock()
	447	+	partialOutput := a.lastAssistant
	448	+	a.mu.Unlock()
446	449		result := a.Result()
447	450		if result != nil {
448	451		if err := validateTurnResultAgent(result, a.ID); err != nil {

@@ -468,14 +471,20 @@ func (f Supervisor) ensureResult(a Agent, status Status, runErr error) {
468	471		result = nil
469	472		}
470	473		}
	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.
	478	+	// The event stream has already projected that text onto the Agent, so
	479	+	// retain it in the durable result rather than returning an empty failure.
	480	+	result.Output = partialOutput
	481	+	}
471	482		if result == nil {
472	483		turnID := a.CurrentTurnID()
473	484		if turnID == "" {
474	485		turnID = fmt.Sprintf("turn-%d", attempt)
475	486		}
476		-	a.mu.Lock()
477		-	output := a.lastAssistant
478		-	a.mu.Unlock()
	487	+	output := partialOutput
479	488		resultStatus := ResultSucceeded
480	489		switch {
481	490		case status == StatusFailed && errors.Is(runErr, context.Canceled):

@@ -512,7 +521,20 @@ func (f Supervisor) ensureResult(a Agent, status Status, runErr error) {
512	521		} else {
513	522		result.Status = ResultFailed
514	523		}
515		-	if result.Error == nil {
	524	+	if errors.Is(runErr, context.DeadlineExceeded) {
	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).
	529	+	message := "subagent deadline exceeded; inspect history for received output"
	530	+	if result.Output != "" {
	531	+	message = "subagent deadline exceeded; partial output is preserved in the result and history"
	532	+	}
	533	+	result.Error = &ResultError{
	534	+	Code: "deadline_exceeded",
	535	+	Message: message,
	536	+	}
	537	+	} else if result.Error == nil {
516	538		result.Error = &ResultError{Code: "runner_failed", Message: truncate(runErr.Error(), 500)}
517	539		}
518	540		}

 		return
 	}
 	attempt := a.AttemptValue()
+	a.mu.Lock()
+	partialOutput := a.lastAssistant
+	a.mu.Unlock()
 	result := a.Result()
 	if result != nil {
 		if err := validateTurnResultAgent(result, a.ID); err != nil {
 			result = nil
 		}
 	}
+	if result != nil && result.Output == "" && partialOutput != "" &&
+		(result.Status == ResultFailed || result.Status == ResultCanceled || status == StatusFailed || status == StatusKilled) {
+		// 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 failure.
+		result.Output = partialOutput
+	}
 	if result == nil {
 		turnID := a.CurrentTurnID()
 		if turnID == "" {
 			turnID = fmt.Sprintf("turn-%d", attempt)
 		}
-		a.mu.Lock()
+		output := partialOutput
-		output := a.lastAssistant
-		a.mu.Unlock()
 		resultStatus := ResultSucceeded
 		switch {
 		case status == StatusFailed && errors.Is(runErr, context.Canceled):
 		} else {
 			result.Status = ResultFailed
 		}
-		if result.Error == nil {
+		if errors.Is(runErr, context.DeadlineExceeded) {
+			// The parent deadline is authoritative for terminal categorization, so
+			// it intentionally overrides any child result error, including
+			// context_limit. The worker was first asked to stop cleanly, and any
+			// 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"
+			}
+			result.Error = &ResultError{
+				Code:    "deadline_exceeded",
+				Message: message,
+			}
+		} else if result.Error == nil {
 			result.Error = &ResultError{Code: "runner_failed", Message: truncate(runErr.Error(), 500)}
 		}
 	}

M packages/agent/subagents/result_test.go +52 -10


@@ -269,23 +269,35 @@ func checkCaptureFailureResult(t testing.T, agent Agent, stateDir string) {
269	269
270	270		func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
271	271		tests := []struct {
272		-	name string
273		-	status Status
274		-	runErr error
275		-	wantState TurnStatus
	272	+	name string
	273	+	status Status
	274	+	runErr error
	275	+	childCode string
	276	+	wantState TurnStatus
	277	+	wantCode string
	278	+	wantOutput string
276	279		}{
277		-	{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed},
278		-	{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled},
279		-	{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled},
	280	+	{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed, wantCode: "runner_failed"},
	281	+	{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
	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"},
	284	+	{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
280	285		}
281	286		for _, tt := range tests {
282	287		t.Run(tt.name, func(t *testing.T) {
283	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	+	}
284	295		agent := &Agent{
285		-	ID: "agent-1",
286		-	stateDir: stateDir,
	296	+	ID: "agent-1",
	297	+	stateDir: stateDir,
	298	+	lastAssistant: tt.wantOutput,
287	299		result: &TurnResult{
288		-	AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
	300	+	AgentID: "agent-1", TurnID: "turn-1", Status: childStatus, Error: childError,
289	301		},
290	302		}
291	303		supervisor := &Supervisor{}

@@ -297,6 +309,12 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
297	309		if result.Error == nil {
298	310		t.Fatal("runner error was not recorded")
299	311		}
	312	+	if result.Error.Code != tt.wantCode {
	313	+	t.Fatalf("result error code = %q, want %q", result.Error.Code, tt.wantCode)
	314	+	}
	315	+	if result.Output != tt.wantOutput {
	316	+	t.Fatalf("result output = %q, want %q", result.Output, tt.wantOutput)
	317	+	}
300	318		durable, err := readTurnResult(stateDir)
301	319		if err != nil {
302	320		t.Fatal(err)

@@ -304,6 +322,30 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
304	322		if durable.Status != tt.wantState {
305	323		t.Fatalf("durable result status = %q, want %q", durable.Status, tt.wantState)
306	324		}
	325	+	if durable.Output != tt.wantOutput {
	326	+	t.Fatalf("durable result output = %q, want %q", durable.Output, tt.wantOutput)
	327	+	}
	328	+	if durable.Error == nil \|\| durable.Error.Code != tt.wantCode {
	329	+	t.Fatalf("durable result error = %#v, want code %q", durable.Error, tt.wantCode)
	330	+	}
307	331		})
308	332		}
309	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	+	}


@@ -269,23 +269,35 @@ func checkCaptureFailureResult(t testing.T, agent Agent, stateDir string) {
269			269
270		func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {	270		func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
271		tests := []struct {	271		tests := []struct {
272	-	name string	272	+	name string
273	-	status Status	273	+	status Status
274	-	runErr error	274	+	runErr error
275	-	wantState TurnStatus	275	+	childCode string
			276	+	wantState TurnStatus
			277	+	wantCode string
			278	+	wantOutput string
276		}{	279		}{
277	-	{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed},	280	+	{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed, wantCode: "runner_failed"},
278	-	{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled},	281	+	{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
279	-	{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled},	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"},
			284	+	{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
280		}	285		}
281		for _, tt := range tests {	286		for _, tt := range tests {
282		t.Run(tt.name, func(t *testing.T) {	287		t.Run(tt.name, func(t *testing.T) {
283		stateDir := t.TempDir()	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	+	}
284		agent := &Agent{	295		agent := &Agent{
285	-	ID: "agent-1",	296	+	ID: "agent-1",
286	-	stateDir: stateDir,	297	+	stateDir: stateDir,
			298	+	lastAssistant: tt.wantOutput,
287		result: &TurnResult{	299		result: &TurnResult{
288	-	AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,	300	+	AgentID: "agent-1", TurnID: "turn-1", Status: childStatus, Error: childError,
289		},	301		},
290		}	302		}
291		supervisor := &Supervisor{}	303		supervisor := &Supervisor{}

@@ -297,6 +309,12 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
297		if result.Error == nil {	309		if result.Error == nil {
298		t.Fatal("runner error was not recorded")	310		t.Fatal("runner error was not recorded")
299		}	311		}
			312	+	if result.Error.Code != tt.wantCode {
			313	+	t.Fatalf("result error code = %q, want %q", result.Error.Code, tt.wantCode)
			314	+	}
			315	+	if result.Output != tt.wantOutput {
			316	+	t.Fatalf("result output = %q, want %q", result.Output, tt.wantOutput)
			317	+	}
300		durable, err := readTurnResult(stateDir)	318		durable, err := readTurnResult(stateDir)
301		if err != nil {	319		if err != nil {
302		t.Fatal(err)	320		t.Fatal(err)

@@ -304,6 +322,30 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
304		if durable.Status != tt.wantState {	322		if durable.Status != tt.wantState {
305		t.Fatalf("durable result status = %q, want %q", durable.Status, tt.wantState)	323		t.Fatalf("durable result status = %q, want %q", durable.Status, tt.wantState)
306		}	324		}
			325	+	if durable.Output != tt.wantOutput {
			326	+	t.Fatalf("durable result output = %q, want %q", durable.Output, tt.wantOutput)
			327	+	}
			328	+	if durable.Error == nil \|\| durable.Error.Code != tt.wantCode {
			329	+	t.Fatalf("durable result error = %#v, want code %q", durable.Error, tt.wantCode)
			330	+	}
307		})	331		})
308		}	332		}
309		}	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	+	}

M packages/agent/subagents/runner.go +95 -4


@@ -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"

@@ -52,6 +53,11 @@ type execRunner struct {
52	53		// would litter the user's repo — every agent's Dir points
53	54		// at it directly.
54	55		SessionPath string
	56	+
	57	+	// GracePeriod is the time a deadline-expired worker gets to handle an
	58	+	// inbox shutdown and write its session/result before the process group is
	59	+	// force-killed. Zero uses the standard ten-second cancellation grace.
	60	+	GracePeriod time.Duration
55	61		}
56	62
57	63		// subagentWorkerArgsOpts captures every dynamic input to subagentWorkerArgs.

@@ -302,7 +308,14 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
302	308		args = defaultChildArgs(exe, r.agent, sessionPath, inboxPath)
303	309		}
304	310
305		-	cmd := exec.CommandContext(ctx, args[0], args[1:]...)
	311	+	// Do not bind the process directly to ctx: CommandContext sends an
	312	+	// immediate SIGKILL at the deadline, which prevents the worker from
	313	+	// handling cancellation and persisting the session it has built so far.
	314	+	// A deadline watcher below requests the worker's graceful inbox shutdown
	315	+	// first, then force-stops it only after the bounded grace period. Use a
	316	+	// non-canceling CommandContext so configureWorkerProcess can retain its
	317	+	// Cmd.Cancel fallback without tying it to the worker lifetime context.
	318	+	cmd := exec.CommandContext(context.Background(), args[0], args[1:]...)
306	319		cmd.Dir = r.agent.Dir
307	320		cmd.Env = append(workerEnvironment(r.agent.Provider),
308	321		"ZUT_SUBAGENT_AGENT_ID="+r.agent.ID,

@@ -353,6 +366,10 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
353	366		r.agent.persistFn(r.agent)
354	367		}
355	368
	369	+	runnerDone := make(chan struct{})
	370	+	defer close(runnerDone)
	371	+	go r.stopOnContextDone(ctx, cmd, runnerDone)
	372	+
356	373		var logErrMu sync.Mutex
357	374		var logErr error
358	375		var stopOnLogErr sync.Once

@@ -460,12 +477,16 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
460	477		if ee, ok := err.(*exec.ExitError); ok {
461	478		exit = ee.ExitCode()
462	479		}
463		-	if err != nil && ctx.Err() != nil {
464		-	appendLog(NewEvent("agent_stopped", map[string]any{"reason": "cancelled"}))
	480	+	if ctxErr := ctx.Err(); ctxErr != nil {
	481	+	reason := "cancelled"
	482	+	if errors.Is(ctxErr, context.DeadlineExceeded) {
	483	+	reason = "deadline"
	484	+	}
	485	+	appendLog(NewEvent("agent_stopped", map[string]any{"reason": reason}))
465	486		if logErr := firstLogErr(); logErr != nil {
466	487		return fmt.Errorf("subagent event log: %w", logErr)
467	488		}
468		-	return ctx.Err()
	489	+	return ctxErr
469	490		}
470	491		if err != nil {
471	492		appendLog(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": exit, "error": err.Error()}))

@@ -482,6 +503,48 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
482	503		return nil
483	504		}
484	505
	506	+	// stopOnContextDone stops a worker when its run context ends. A deadline gets
	507	+	// one bounded chance to process an inbox shutdown and persist its session;
	508	+	// explicit cancellation force-stops immediately so Run cannot remain blocked.
	509	+	func (r execRunner) stopOnContextDone(ctx context.Context, cmd exec.Cmd, runnerDone <-chan struct{}) {
	510	+	select {
	511	+	case <-runnerDone:
	512	+	return
	513	+	case <-ctx.Done():
	514	+	}
	515	+	if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
	516	+	if cmd.Cancel != nil {
	517	+	_ = cmd.Cancel()
	518	+	return
	519	+	}
	520	+	_ = killProcessGroup(cmd)
	521	+	return
	522	+	}
	523	+
	524	+	grace := r.GracePeriod
	525	+	if grace <= 0 {
	526	+	grace = 10 * time.Second
	527	+	}
	528	+	shutdownCtx, cancel := context.WithTimeout(context.Background(), grace)
	529	+	defer cancel()
	530	+	if r.agent != nil && r.agent.inbox != nil {
	531	+	_ = r.agent.inbox.SendCommandContext(shutdownCtx, NewCommand(
	532	+	CommandAgentShutdown, r.agent.ID, r.agent.CurrentTurnID(), AgentShutdownPayload{},
	533	+	))
	534	+	}
	535	+
	536	+	select {
	537	+	case <-runnerDone:
	538	+	return
	539	+	case <-shutdownCtx.Done():
	540	+	}
	541	+	if cmd.Cancel != nil {
	542	+	_ = cmd.Cancel()
	543	+	return
	544	+	}
	545	+	_ = killProcessGroup(cmd)
	546	+	}
	547	+
485	548		func workerEnvironment(provider string) []string {
486	549		// Provider credentials are resolved by the supervisor and transferred
487	550		// over the child's stdin. Do not inherit API-key/token variables into the

@@ -703,6 +766,15 @@ func notifyPromptTurnEnd(a *Agent, ev Event) {
703	766		go fn(int(step), errMsg)
704	767		}
705	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	+
706	778		// applyEventToSink translates an Event into Sink updates. Only a
707	779		// few event types are interpreted; the rest still land in the
708	780		// durable log via the caller.

@@ -711,6 +783,17 @@ func applyEventToSink(ev Event, sink Sink) {
711	783		userMessage(string)
712	784		assistantMessage(string)
713	785		}
	786	+	type streamingSink interface {
	787	+	assistantDelta(string)
	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	+	}
714	797		appendMessage := func(text string, assistant bool) {
715	798		if text == "" {
716	799		return

@@ -731,6 +814,14 @@ func applyEventToSink(ev Event, sink Sink) {
731	814		}
732	815
733	816		switch ev.Type {
	817	+	case "message.delta":
	818	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
	819	+	if streaming, ok := sink.(streamingSink); ok {
	820	+	streaming.assistantDelta(delta)
	821	+	} else {
	822	+	sink.Transcript(delta)
	823	+	}
	824	+	}
734	825		case "assistant_message", "user_message":
735	826		var text []string
736	827		if c, ok := ev.Data["content"].([]any); ok {


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

@@ -52,6 +53,11 @@ type execRunner struct {
52		// would litter the user's repo — every agent's Dir points	53		// would litter the user's repo — every agent's Dir points
53		// at it directly.	54		// at it directly.
54		SessionPath string	55		SessionPath string
			56	+
			57	+	// GracePeriod is the time a deadline-expired worker gets to handle an
			58	+	// inbox shutdown and write its session/result before the process group is
			59	+	// force-killed. Zero uses the standard ten-second cancellation grace.
			60	+	GracePeriod time.Duration
55		}	61		}
56			62
57		// subagentWorkerArgsOpts captures every dynamic input to subagentWorkerArgs.	63		// subagentWorkerArgsOpts captures every dynamic input to subagentWorkerArgs.

@@ -302,7 +308,14 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
302		args = defaultChildArgs(exe, r.agent, sessionPath, inboxPath)	308		args = defaultChildArgs(exe, r.agent, sessionPath, inboxPath)
303		}	309		}
304			310
305	-	cmd := exec.CommandContext(ctx, args[0], args[1:]...)	311	+	// Do not bind the process directly to ctx: CommandContext sends an
			312	+	// immediate SIGKILL at the deadline, which prevents the worker from
			313	+	// handling cancellation and persisting the session it has built so far.
			314	+	// A deadline watcher below requests the worker's graceful inbox shutdown
			315	+	// first, then force-stops it only after the bounded grace period. Use a
			316	+	// non-canceling CommandContext so configureWorkerProcess can retain its
			317	+	// Cmd.Cancel fallback without tying it to the worker lifetime context.
			318	+	cmd := exec.CommandContext(context.Background(), args[0], args[1:]...)
306		cmd.Dir = r.agent.Dir	319		cmd.Dir = r.agent.Dir
307		cmd.Env = append(workerEnvironment(r.agent.Provider),	320		cmd.Env = append(workerEnvironment(r.agent.Provider),
308		"ZUT_SUBAGENT_AGENT_ID="+r.agent.ID,	321		"ZUT_SUBAGENT_AGENT_ID="+r.agent.ID,

@@ -353,6 +366,10 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
353		r.agent.persistFn(r.agent)	366		r.agent.persistFn(r.agent)
354		}	367		}
355			368
			369	+	runnerDone := make(chan struct{})
			370	+	defer close(runnerDone)
			371	+	go r.stopOnContextDone(ctx, cmd, runnerDone)
			372	+
356		var logErrMu sync.Mutex	373		var logErrMu sync.Mutex
357		var logErr error	374		var logErr error
358		var stopOnLogErr sync.Once	375		var stopOnLogErr sync.Once

@@ -460,12 +477,16 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
460		if ee, ok := err.(*exec.ExitError); ok {	477		if ee, ok := err.(*exec.ExitError); ok {
461		exit = ee.ExitCode()	478		exit = ee.ExitCode()
462		}	479		}
463	-	if err != nil && ctx.Err() != nil {	480	+	if ctxErr := ctx.Err(); ctxErr != nil {
464	-	appendLog(NewEvent("agent_stopped", map[string]any{"reason": "cancelled"}))	481	+	reason := "cancelled"
			482	+	if errors.Is(ctxErr, context.DeadlineExceeded) {
			483	+	reason = "deadline"
			484	+	}
			485	+	appendLog(NewEvent("agent_stopped", map[string]any{"reason": reason}))
465		if logErr := firstLogErr(); logErr != nil {	486		if logErr := firstLogErr(); logErr != nil {
466		return fmt.Errorf("subagent event log: %w", logErr)	487		return fmt.Errorf("subagent event log: %w", logErr)
467		}	488		}
468	-	return ctx.Err()	489	+	return ctxErr
469		}	490		}
470		if err != nil {	491		if err != nil {
471		appendLog(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": exit, "error": err.Error()}))	492		appendLog(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": exit, "error": err.Error()}))

@@ -482,6 +503,48 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error {
482		return nil	503		return nil
483		}	504		}
484			505
			506	+	// stopOnContextDone stops a worker when its run context ends. A deadline gets
			507	+	// one bounded chance to process an inbox shutdown and persist its session;
			508	+	// explicit cancellation force-stops immediately so Run cannot remain blocked.
			509	+	func (r execRunner) stopOnContextDone(ctx context.Context, cmd exec.Cmd, runnerDone <-chan struct{}) {
			510	+	select {
			511	+	case <-runnerDone:
			512	+	return
			513	+	case <-ctx.Done():
			514	+	}
			515	+	if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
			516	+	if cmd.Cancel != nil {
			517	+	_ = cmd.Cancel()
			518	+	return
			519	+	}
			520	+	_ = killProcessGroup(cmd)
			521	+	return
			522	+	}
			523	+
			524	+	grace := r.GracePeriod
			525	+	if grace <= 0 {
			526	+	grace = 10 * time.Second
			527	+	}
			528	+	shutdownCtx, cancel := context.WithTimeout(context.Background(), grace)
			529	+	defer cancel()
			530	+	if r.agent != nil && r.agent.inbox != nil {
			531	+	_ = r.agent.inbox.SendCommandContext(shutdownCtx, NewCommand(
			532	+	CommandAgentShutdown, r.agent.ID, r.agent.CurrentTurnID(), AgentShutdownPayload{},
			533	+	))
			534	+	}
			535	+
			536	+	select {
			537	+	case <-runnerDone:
			538	+	return
			539	+	case <-shutdownCtx.Done():
			540	+	}
			541	+	if cmd.Cancel != nil {
			542	+	_ = cmd.Cancel()
			543	+	return
			544	+	}
			545	+	_ = killProcessGroup(cmd)
			546	+	}
			547	+
485		func workerEnvironment(provider string) []string {	548		func workerEnvironment(provider string) []string {
486		// Provider credentials are resolved by the supervisor and transferred	549		// Provider credentials are resolved by the supervisor and transferred
487		// over the child's stdin. Do not inherit API-key/token variables into the	550		// over the child's stdin. Do not inherit API-key/token variables into the

@@ -703,6 +766,15 @@ func notifyPromptTurnEnd(a *Agent, ev Event) {
703		go fn(int(step), errMsg)	766		go fn(int(step), errMsg)
704		}	767		}
705			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	+
706		// applyEventToSink translates an Event into Sink updates. Only a	778		// applyEventToSink translates an Event into Sink updates. Only a
707		// few event types are interpreted; the rest still land in the	779		// few event types are interpreted; the rest still land in the
708		// durable log via the caller.	780		// durable log via the caller.

@@ -711,6 +783,17 @@ func applyEventToSink(ev Event, sink Sink) {
711		userMessage(string)	783		userMessage(string)
712		assistantMessage(string)	784		assistantMessage(string)
713		}	785		}
			786	+	type streamingSink interface {
			787	+	assistantDelta(string)
			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	+	}
714		appendMessage := func(text string, assistant bool) {	797		appendMessage := func(text string, assistant bool) {
715		if text == "" {	798		if text == "" {
716		return	799		return

@@ -731,6 +814,14 @@ func applyEventToSink(ev Event, sink Sink) {
731		}	814		}
732			815
733		switch ev.Type {	816		switch ev.Type {
			817	+	case "message.delta":
			818	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
			819	+	if streaming, ok := sink.(streamingSink); ok {
			820	+	streaming.assistantDelta(delta)
			821	+	} else {
			822	+	sink.Transcript(delta)
			823	+	}
			824	+	}
734		case "assistant_message", "user_message":	825		case "assistant_message", "user_message":
735		var text []string	826		var text []string
736		if c, ok := ev.Data["content"].([]any); ok {	827		if c, ok := ev.Data["content"].([]any); ok {

M packages/agent/subagents/runner_e2e_test.go +181 -19


@@ -12,25 +12,187 @@ import (
12	12		"time"
13	13		)
14	14
15		-	// TestRunnerEndToEndWithStubChild is the integration test for the
16		-	// new daemon-mode runner. It compiles the stubchild binary in
17		-	// testdata/cmd/stubchild, points an execRunner at it, and drives
18		-	// a Supervisor through one Spawn + one SendUserTurn + Stop cycle.
19		-	//
20		-	// What this proves:
21		-	//
22		-	// - The default argv shape (subagentWorkerArgs) is one the child can
23		-	// actually parse — locks the shape against silent breakage.
24		-	// - The stdout JSONL parser ingests events and writes them to the
25		-	// durable log.
26		-	// - applyEventToSink turns events into Activity / Transcript
27		-	// updates the dashboard reads.
28		-	// - The supervisor inbox dials the child's socket and a follow-up
29		-	// line round-trips back as a user_message event.
30		-	// - Stop closes the inbox AND cancels the child's context so the
31		-	// stub exits cleanly.
32		-	//
33		-	// The test is skipped on platforms without unix sockets (Windows).
	15	+	type notifyingAgentSink struct {
	16	+	agentSink
	17	+	deltas chan<- struct{}
	18	+	}
	19	+
	20	+	func (s notifyingAgentSink) assistantDelta(text string) {
	21	+	s.agentSink.assistantDelta(text)
	22	+	select {
	23	+	case s.deltas <- struct{}{}:
	24	+	default:
	25	+	}
	26	+	}
	27	+
	28	+	const runnerProcessTimeout = 10 * time.Second
	29	+
	30	+	type controlledDeadlineContext struct{ done chan struct{} }
	31	+
	32	+	func (c *controlledDeadlineContext) Deadline() (time.Time, bool) {
	33	+	return time.Now().Add(time.Hour), true
	34	+	}
	35	+	func (c *controlledDeadlineContext) Done() <-chan struct{} { return c.done }
	36	+	func (c *controlledDeadlineContext) Err() error {
	37	+	select {
	38	+	case <-c.done:
	39	+	return context.DeadlineExceeded
	40	+	default:
	41	+	return nil
	42	+	}
	43	+	}
	44	+	func (c *controlledDeadlineContext) Value(any) any { return nil }
	45	+	func (c *controlledDeadlineContext) expire() {
	46	+	select {
	47	+	case <-c.done:
	48	+	default:
	49	+	close(c.done)
	50	+	}
	51	+	}
	52	+
	53	+	// TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput
	54	+	// verifies that deadline expiry shuts a worker down through its inbox rather
	55	+	// than killing it before its terminal events can be recorded.
	56	+	func TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput(t *testing.T) {
	57	+	if runtime.GOOS == "windows" {
	58	+	t.Skip("unix sockets not supported")
	59	+	}
	60	+	if testing.Short() {
	61	+	t.Skip("skip end-to-end runner test in -short mode")
	62	+	}
	63	+
	64	+	exe := buildStubChild(t)
	65	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
	66	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
	67	+
	68	+	root := t.TempDir()
	69	+	inboxPath, err := inboxSocketPath(root, "deadline-test")
	70	+	if err != nil {
	71	+	t.Fatalf("inboxSocketPath: %v", err)
	72	+	}
	73	+	a := &Agent{
	74	+	ID: "deadline-test",
	75	+	Task: "long-running task",
	76	+	Dir: root,
	77	+	SessionPath: filepath.Join(root, "session.jsonl"),
	78	+	InboxPath: inboxPath,
	79	+	EventLogPath: filepath.Join(root, "events.jsonl"),
	80	+	}
	81	+	a.inbox = NewInbox(a.InboxPath)
	82	+	t.Cleanup(func() { _ = a.inbox.Close() })
	83	+
	84	+	r := &execRunner{
	85	+	agent: a,
	86	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
	87	+	GracePeriod: time.Second,
	88	+	}
	89	+	ctx := &controlledDeadlineContext{done: make(chan struct{})}
	90	+	defer ctx.expire()
	91	+	deltas := make(chan struct{}, 1)
	92	+	runDone := make(chan error, 1)
	93	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
	94	+	select {
	95	+	case <-deltas:
	96	+	case <-time.After(runnerProcessTimeout):
	97	+	t.Fatal("worker did not emit its initial delta")
	98	+	}
	99	+	ctx.expire()
	100	+	select {
	101	+	case err := <-runDone:
	102	+	if !errors.Is(err, context.DeadlineExceeded) {
	103	+	t.Fatalf("Run error = %v, want context deadline exceeded", err)
	104	+	}
	105	+	case <-time.After(runnerProcessTimeout):
	106	+	t.Fatal("Run did not return after deadline expiry")
	107	+	}
	108	+
	109	+	events, err := ReadEventLog(a.EventLogPath)
	110	+	if err != nil {
	111	+	t.Fatalf("ReadEventLog: %v", err)
	112	+	}
	113	+	var sawDelta, sawGracefulTurnEnd bool
	114	+	for _, event := range events {
	115	+	if event.Type == "message.delta" && event.Data["delta"] == "partial answer" {
	116	+	sawDelta = true
	117	+	}
	118	+	if event.Type == "turn_end" && event.Data["stop"] == "cancelled" {
	119	+	sawGracefulTurnEnd = true
	120	+	}
	121	+	}
	122	+	if !sawDelta \|\| !sawGracefulTurnEnd {
	123	+	t.Fatalf("deadline did not preserve a graceful partial turn: %s", formatEvents(events))
	124	+	}
	125	+
	126	+	snapshot := a.Snapshot()
	127	+	if snapshot.LastAssistant != "partial answer" {
	128	+	t.Fatalf("live partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
	129	+	}
	130	+	replayed := &Agent{}
	131	+	for _, event := range events {
	132	+	replayEventTranscript(replayed, event)
	133	+	}
	134	+	if snapshot := replayed.Snapshot(); snapshot.LastAssistant != "partial answer" {
	135	+	t.Fatalf("replayed partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
	136	+	}
	137	+	}
	138	+
	139	+	func TestExecRunnerCancelsBlockedWorker(t *testing.T) {
	140	+	if runtime.GOOS == "windows" {
	141	+	t.Skip("unix sockets not supported")
	142	+	}
	143	+	if testing.Short() {
	144	+	t.Skip("skip end-to-end runner test in -short mode")
	145	+	}
	146	+
	147	+	exe := buildStubChild(t)
	148	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
	149	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
	150	+
	151	+	root := t.TempDir()
	152	+	inboxPath, err := inboxSocketPath(root, "cancel-test")
	153	+	if err != nil {
	154	+	t.Fatalf("inboxSocketPath: %v", err)
	155	+	}
	156	+	a := &Agent{
	157	+	ID: "cancel-test",
	158	+	Task: "blocked task",
	159	+	Dir: root,
	160	+	SessionPath: filepath.Join(root, "session.jsonl"),
	161	+	InboxPath: inboxPath,
	162	+	EventLogPath: filepath.Join(root, "events.jsonl"),
	163	+	}
	164	+	a.inbox = NewInbox(a.InboxPath)
	165	+	t.Cleanup(func() { _ = a.inbox.Close() })
	166	+
	167	+	r := &execRunner{
	168	+	agent: a,
	169	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
	170	+	}
	171	+	ctx, cancel := context.WithCancel(context.Background())
	172	+	defer cancel()
	173	+	deltas := make(chan struct{}, 1)
	174	+	runDone := make(chan error, 1)
	175	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
	176	+
	177	+	select {
	178	+	case <-deltas:
	179	+	case <-time.After(runnerProcessTimeout):
	180	+	t.Fatal("worker did not emit its initial delta")
	181	+	}
	182	+	cancel()
	183	+	select {
	184	+	case err := <-runDone:
	185	+	if !errors.Is(err, context.Canceled) {
	186	+	t.Fatalf("Run error = %v, want context canceled", err)
	187	+	}
	188	+	case <-time.After(runnerProcessTimeout):
	189	+	t.Fatal("Run did not return after context cancellation")
	190	+	}
	191	+	}
	192	+
	193	+	// TestRunnerEndToEndWithStubChild drives the daemon-mode runner through an
	194	+	// initial turn, a follow-up, and a graceful stop using the stub child. It is
	195	+	// skipped on platforms without Unix sockets.
34	196		func TestRunnerEndToEndWithStubChild(t *testing.T) {
35	197		if runtime.GOOS == "windows" {
36	198		t.Skip("unix sockets not supported")


@@ -12,25 +12,187 @@ import (
12		"time"	12		"time"
13		)	13		)
14			14
15	-	// TestRunnerEndToEndWithStubChild is the integration test for the	15	+	type notifyingAgentSink struct {
16	-	// new daemon-mode runner. It compiles the stubchild binary in	16	+	agentSink
17	-	// testdata/cmd/stubchild, points an execRunner at it, and drives	17	+	deltas chan<- struct{}
18	-	// a Supervisor through one Spawn + one SendUserTurn + Stop cycle.	18	+	}
19	-	//	19	+
20	-	// What this proves:	20	+	func (s notifyingAgentSink) assistantDelta(text string) {
21	-	//	21	+	s.agentSink.assistantDelta(text)
22	-	// - The default argv shape (subagentWorkerArgs) is one the child can	22	+	select {
23	-	// actually parse — locks the shape against silent breakage.	23	+	case s.deltas <- struct{}{}:
24	-	// - The stdout JSONL parser ingests events and writes them to the	24	+	default:
25	-	// durable log.	25	+	}
26	-	// - applyEventToSink turns events into Activity / Transcript	26	+	}
27	-	// updates the dashboard reads.	27	+
28	-	// - The supervisor inbox dials the child's socket and a follow-up	28	+	const runnerProcessTimeout = 10 * time.Second
29	-	// line round-trips back as a user_message event.	29	+
30	-	// - Stop closes the inbox AND cancels the child's context so the	30	+	type controlledDeadlineContext struct{ done chan struct{} }
31	-	// stub exits cleanly.	31	+
32	-	//	32	+	func (c *controlledDeadlineContext) Deadline() (time.Time, bool) {
33	-	// The test is skipped on platforms without unix sockets (Windows).	33	+	return time.Now().Add(time.Hour), true
			34	+	}
			35	+	func (c *controlledDeadlineContext) Done() <-chan struct{} { return c.done }
			36	+	func (c *controlledDeadlineContext) Err() error {
			37	+	select {
			38	+	case <-c.done:
			39	+	return context.DeadlineExceeded
			40	+	default:
			41	+	return nil
			42	+	}
			43	+	}
			44	+	func (c *controlledDeadlineContext) Value(any) any { return nil }
			45	+	func (c *controlledDeadlineContext) expire() {
			46	+	select {
			47	+	case <-c.done:
			48	+	default:
			49	+	close(c.done)
			50	+	}
			51	+	}
			52	+
			53	+	// TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput
			54	+	// verifies that deadline expiry shuts a worker down through its inbox rather
			55	+	// than killing it before its terminal events can be recorded.
			56	+	func TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput(t *testing.T) {
			57	+	if runtime.GOOS == "windows" {
			58	+	t.Skip("unix sockets not supported")
			59	+	}
			60	+	if testing.Short() {
			61	+	t.Skip("skip end-to-end runner test in -short mode")
			62	+	}
			63	+
			64	+	exe := buildStubChild(t)
			65	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
			66	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
			67	+
			68	+	root := t.TempDir()
			69	+	inboxPath, err := inboxSocketPath(root, "deadline-test")
			70	+	if err != nil {
			71	+	t.Fatalf("inboxSocketPath: %v", err)
			72	+	}
			73	+	a := &Agent{
			74	+	ID: "deadline-test",
			75	+	Task: "long-running task",
			76	+	Dir: root,
			77	+	SessionPath: filepath.Join(root, "session.jsonl"),
			78	+	InboxPath: inboxPath,
			79	+	EventLogPath: filepath.Join(root, "events.jsonl"),
			80	+	}
			81	+	a.inbox = NewInbox(a.InboxPath)
			82	+	t.Cleanup(func() { _ = a.inbox.Close() })
			83	+
			84	+	r := &execRunner{
			85	+	agent: a,
			86	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
			87	+	GracePeriod: time.Second,
			88	+	}
			89	+	ctx := &controlledDeadlineContext{done: make(chan struct{})}
			90	+	defer ctx.expire()
			91	+	deltas := make(chan struct{}, 1)
			92	+	runDone := make(chan error, 1)
			93	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
			94	+	select {
			95	+	case <-deltas:
			96	+	case <-time.After(runnerProcessTimeout):
			97	+	t.Fatal("worker did not emit its initial delta")
			98	+	}
			99	+	ctx.expire()
			100	+	select {
			101	+	case err := <-runDone:
			102	+	if !errors.Is(err, context.DeadlineExceeded) {
			103	+	t.Fatalf("Run error = %v, want context deadline exceeded", err)
			104	+	}
			105	+	case <-time.After(runnerProcessTimeout):
			106	+	t.Fatal("Run did not return after deadline expiry")
			107	+	}
			108	+
			109	+	events, err := ReadEventLog(a.EventLogPath)
			110	+	if err != nil {
			111	+	t.Fatalf("ReadEventLog: %v", err)
			112	+	}
			113	+	var sawDelta, sawGracefulTurnEnd bool
			114	+	for _, event := range events {
			115	+	if event.Type == "message.delta" && event.Data["delta"] == "partial answer" {
			116	+	sawDelta = true
			117	+	}
			118	+	if event.Type == "turn_end" && event.Data["stop"] == "cancelled" {
			119	+	sawGracefulTurnEnd = true
			120	+	}
			121	+	}
			122	+	if !sawDelta \|\| !sawGracefulTurnEnd {
			123	+	t.Fatalf("deadline did not preserve a graceful partial turn: %s", formatEvents(events))
			124	+	}
			125	+
			126	+	snapshot := a.Snapshot()
			127	+	if snapshot.LastAssistant != "partial answer" {
			128	+	t.Fatalf("live partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
			129	+	}
			130	+	replayed := &Agent{}
			131	+	for _, event := range events {
			132	+	replayEventTranscript(replayed, event)
			133	+	}
			134	+	if snapshot := replayed.Snapshot(); snapshot.LastAssistant != "partial answer" {
			135	+	t.Fatalf("replayed partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
			136	+	}
			137	+	}
			138	+
			139	+	func TestExecRunnerCancelsBlockedWorker(t *testing.T) {
			140	+	if runtime.GOOS == "windows" {
			141	+	t.Skip("unix sockets not supported")
			142	+	}
			143	+	if testing.Short() {
			144	+	t.Skip("skip end-to-end runner test in -short mode")
			145	+	}
			146	+
			147	+	exe := buildStubChild(t)
			148	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
			149	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
			150	+
			151	+	root := t.TempDir()
			152	+	inboxPath, err := inboxSocketPath(root, "cancel-test")
			153	+	if err != nil {
			154	+	t.Fatalf("inboxSocketPath: %v", err)
			155	+	}
			156	+	a := &Agent{
			157	+	ID: "cancel-test",
			158	+	Task: "blocked task",
			159	+	Dir: root,
			160	+	SessionPath: filepath.Join(root, "session.jsonl"),
			161	+	InboxPath: inboxPath,
			162	+	EventLogPath: filepath.Join(root, "events.jsonl"),
			163	+	}
			164	+	a.inbox = NewInbox(a.InboxPath)
			165	+	t.Cleanup(func() { _ = a.inbox.Close() })
			166	+
			167	+	r := &execRunner{
			168	+	agent: a,
			169	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
			170	+	}
			171	+	ctx, cancel := context.WithCancel(context.Background())
			172	+	defer cancel()
			173	+	deltas := make(chan struct{}, 1)
			174	+	runDone := make(chan error, 1)
			175	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
			176	+
			177	+	select {
			178	+	case <-deltas:
			179	+	case <-time.After(runnerProcessTimeout):
			180	+	t.Fatal("worker did not emit its initial delta")
			181	+	}
			182	+	cancel()
			183	+	select {
			184	+	case err := <-runDone:
			185	+	if !errors.Is(err, context.Canceled) {
			186	+	t.Fatalf("Run error = %v, want context canceled", err)
			187	+	}
			188	+	case <-time.After(runnerProcessTimeout):
			189	+	t.Fatal("Run did not return after context cancellation")
			190	+	}
			191	+	}
			192	+
			193	+	// TestRunnerEndToEndWithStubChild drives the daemon-mode runner through an
			194	+	// initial turn, a follow-up, and a graceful stop using the stub child. It is
			195	+	// skipped on platforms without Unix sockets.
34		func TestRunnerEndToEndWithStubChild(t *testing.T) {	196		func TestRunnerEndToEndWithStubChild(t *testing.T) {
35		if runtime.GOOS == "windows" {	197		if runtime.GOOS == "windows" {
36		t.Skip("unix sockets not supported")	198		t.Skip("unix sockets not supported")

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


@@ -58,6 +58,106 @@ func TestCredentialStdinHelperProcess(t *testing.T) {
58	58		os.Exit(0)
59	59		}
60	60
	61	+	// TestMessageDeltaStreamsAndReplaysWithoutDuplicatingFinalMessage verifies
	62	+	// that the durable event stream has the same visible result while live and
	63	+	// after history replay.
	64	+	func TestMessageDeltaStreamsAndReplaysWithoutDuplicatingFinalMessage(t *testing.T) {
	65	+	events := []Event{
	66	+	NewEvent("message.delta", map[string]any{"delta": "partial "}),
	67	+	NewEvent("message.delta", map[string]any{"delta": "answer"}),
	68	+	NewEvent("assistant_message", map[string]any{"content": []any{
	69	+	map[string]any{"type": "text", "text": "partial answer"},
	70	+	}}),
	71	+	}
	72	+
	73	+	live := &Agent{transcriptLoaded: true}
	74	+	for _, event := range events {
	75	+	applyEventToSink(event, agentSink{a: live})
	76	+	}
	77	+	if got := strings.Join(live.Transcript(), "\n"); got != "partial answer" {
	78	+	t.Fatalf("live transcript = %q, want final assistant message once", got)
	79	+	}
	80	+	if got := live.Snapshot().LastAssistant; got != "partial answer" {
	81	+	t.Fatalf("live assistant output = %q, want %q", got, "partial answer")
	82	+	}
	83	+
	84	+	replayed := &Agent{transcriptLoaded: true}
	85	+	for _, event := range events {
	86	+	replayEventTranscript(replayed, event)
	87	+	}
	88	+	if got := strings.Join(replayed.Transcript(), "\n"); got != "partial answer" {
	89	+	t.Fatalf("replayed transcript = %q, want final assistant message once", got)
	90	+	}
	91	+	}
	92	+
	93	+	func TestMessageDeltaBoundsLiveProjection(t *testing.T) {
	94	+	a := &Agent{maxOutputBytes: 32, maxOutputLines: 100, transcriptLoaded: true}
	95	+	applyEventToSink(NewEvent("message.delta", map[string]any{"delta": strings.Repeat("x", 64)}), agentSink{a: a})
	96	+	applyEventToSink(NewEvent("message.delta", map[string]any{"delta": "discarded"}), agentSink{a: a})
	97	+
	98	+	a.mu.Lock()
	99	+	streamed := a.streamingAssistantText
	100	+	truncated := a.streamingAssistantTruncated
	101	+	a.mu.Unlock()
	102	+	if !truncated {
	103	+	t.Fatal("streamed output was not marked truncated")
	104	+	}
	105	+	if got := len([]byte(streamed)); got > 32 {
	106	+	t.Fatalf("streamed output length = %d, want at most 32", got)
	107	+	}
	108	+	snapshot := a.Snapshot()
	109	+	if got := snapshot.LastAssistant; got != streamed {
	110	+	t.Fatalf("live assistant output = %q, want bounded stream %q", got, streamed)
	111	+	}
	112	+	if !snapshot.OutputTruncated {
	113	+	t.Fatal("snapshot did not expose truncated streamed output")
	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	+	}
	159	+	}
	160	+
61	161		func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {
62	162		t.Setenv("OPENAI_API_KEY", "openai-secret")
63	163		t.Setenv("CUSTOM_PROVIDER_API_KEY", "custom-secret")


@@ -58,6 +58,106 @@ func TestCredentialStdinHelperProcess(t *testing.T) {
58	os.Exit(0)	58		os.Exit(0)
59	}	59		}
60		60
		61	+	// TestMessageDeltaStreamsAndReplaysWithoutDuplicatingFinalMessage verifies
		62	+	// that the durable event stream has the same visible result while live and
		63	+	// after history replay.
		64	+	func TestMessageDeltaStreamsAndReplaysWithoutDuplicatingFinalMessage(t *testing.T) {
		65	+	events := []Event{
		66	+	NewEvent("message.delta", map[string]any{"delta": "partial "}),
		67	+	NewEvent("message.delta", map[string]any{"delta": "answer"}),
		68	+	NewEvent("assistant_message", map[string]any{"content": []any{
		69	+	map[string]any{"type": "text", "text": "partial answer"},
		70	+	}}),
		71	+	}
		72	+
		73	+	live := &Agent{transcriptLoaded: true}
		74	+	for _, event := range events {
		75	+	applyEventToSink(event, agentSink{a: live})
		76	+	}
		77	+	if got := strings.Join(live.Transcript(), "\n"); got != "partial answer" {
		78	+	t.Fatalf("live transcript = %q, want final assistant message once", got)
		79	+	}
		80	+	if got := live.Snapshot().LastAssistant; got != "partial answer" {
		81	+	t.Fatalf("live assistant output = %q, want %q", got, "partial answer")
		82	+	}
		83	+
		84	+	replayed := &Agent{transcriptLoaded: true}
		85	+	for _, event := range events {
		86	+	replayEventTranscript(replayed, event)
		87	+	}
		88	+	if got := strings.Join(replayed.Transcript(), "\n"); got != "partial answer" {
		89	+	t.Fatalf("replayed transcript = %q, want final assistant message once", got)
		90	+	}
		91	+	}
		92	+
		93	+	func TestMessageDeltaBoundsLiveProjection(t *testing.T) {
		94	+	a := &Agent{maxOutputBytes: 32, maxOutputLines: 100, transcriptLoaded: true}
		95	+	applyEventToSink(NewEvent("message.delta", map[string]any{"delta": strings.Repeat("x", 64)}), agentSink{a: a})
		96	+	applyEventToSink(NewEvent("message.delta", map[string]any{"delta": "discarded"}), agentSink{a: a})
		97	+
		98	+	a.mu.Lock()
		99	+	streamed := a.streamingAssistantText
		100	+	truncated := a.streamingAssistantTruncated
		101	+	a.mu.Unlock()
		102	+	if !truncated {
		103	+	t.Fatal("streamed output was not marked truncated")
		104	+	}
		105	+	if got := len([]byte(streamed)); got > 32 {
		106	+	t.Fatalf("streamed output length = %d, want at most 32", got)
		107	+	}
		108	+	snapshot := a.Snapshot()
		109	+	if got := snapshot.LastAssistant; got != streamed {
		110	+	t.Fatalf("live assistant output = %q, want bounded stream %q", got, streamed)
		111	+	}
		112	+	if !snapshot.OutputTruncated {
		113	+	t.Fatal("snapshot did not expose truncated streamed output")
		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	+	}
		159	+	}
		160	+
61	func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {	161		func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {
62	t.Setenv("OPENAI_API_KEY", "openai-secret")	162		t.Setenv("OPENAI_API_KEY", "openai-secret")
63	t.Setenv("CUSTOM_PROVIDER_API_KEY", "custom-secret")	163		t.Setenv("CUSTOM_PROVIDER_API_KEY", "custom-secret")

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


@@ -155,7 +155,11 @@ func New(cfg Config) *Supervisor {
155	155		}
156	156		if cfg.NewRunner == nil {
157	157		cfg.NewRunner = func(a *Agent) Runner {
158		-	return &execRunner{agent: a, resolveCredential: cfg.ResolveCredential}
	158	+	return &execRunner{
	159	+	agent: a,
	160	+	resolveCredential: cfg.ResolveCredential,
	161	+	GracePeriod: cfg.Policy.CancelGracePeriod,
	162	+	}
159	163		}
160	164		}
161	165		cfg.Policy.normalize()

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

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

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

 	}
 	if cfg.NewRunner == nil {
 		cfg.NewRunner = func(a *Agent) Runner {
-			return &execRunner{agent: a, resolveCredential: cfg.ResolveCredential}
+			return &execRunner{
+				agent:             a,
+				resolveCredential: cfg.ResolveCredential,
+				GracePeriod:       cfg.Policy.CancelGracePeriod,
+			}
 		}
 	}
 	cfg.Policy.normalize()
 	status := a.status
 	activity := a.activity
 	lastAssistant := a.lastAssistant
+	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
 	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)

M packages/agent/subagents/testdata/cmd/stubchild/main.go +20 -1


@@ -72,8 +72,20 @@ func main() {
72	72
73	73		// Initial task lives in the positional. Process it as the first
74	74		// user turn so the supervisor's "initial task" path is exercised.
	75	+	initialTurnOpen := false
75	76		if task := flag.Arg(0); task != "" {
76		-	runTurn(emit, task, 0)
	77	+	if os.Getenv("ZUT_STUB_BLOCK_INITIAL") == "1" {
	78	+	initialTurnOpen = true
	79	+	emit("turn_start", map[string]any{"step": 0})
	80	+	emit("user_message", map[string]any{
	81	+	"content": []any{
	82	+	map[string]any{"type": "text", "text": task},
	83	+	},
	84	+	})
	85	+	emit("message.delta", map[string]any{"delta": "partial answer"})
	86	+	} else {
	87	+	runTurn(emit, task, 0)
	88	+	}
77	89		}
78	90
79	91		turn := 1

@@ -91,10 +103,17 @@ func main() {
91	103		if parseErr == nil {
92	104		switch command.Type {
93	105		case subagents.CommandAgentShutdown:
	106	+	if initialTurnOpen {
	107	+	emit("turn_end", map[string]any{"stop": "cancelled"})
	108	+	initialTurnOpen = false
	109	+	}
94	110		emit("agent_stopped", map[string]any{"reason": "shutdown"})
95	111		_ = c.Close()
96	112		return
97	113		case subagents.CommandTurnCancel:
	114	+	if initialTurnOpen {
	115	+	initialTurnOpen = false
	116	+	}
98	117		emit("turn_end", map[string]any{"stop": "cancelled"})
99	118		case subagents.CommandTurnStart:
100	119		var payload subagents.TurnStartPayload


@@ -72,8 +72,20 @@ func main() {
72			72
73		// Initial task lives in the positional. Process it as the first	73		// Initial task lives in the positional. Process it as the first
74		// user turn so the supervisor's "initial task" path is exercised.	74		// user turn so the supervisor's "initial task" path is exercised.
			75	+	initialTurnOpen := false
75		if task := flag.Arg(0); task != "" {	76		if task := flag.Arg(0); task != "" {
76	-	runTurn(emit, task, 0)	77	+	if os.Getenv("ZUT_STUB_BLOCK_INITIAL") == "1" {
			78	+	initialTurnOpen = true
			79	+	emit("turn_start", map[string]any{"step": 0})
			80	+	emit("user_message", map[string]any{
			81	+	"content": []any{
			82	+	map[string]any{"type": "text", "text": task},
			83	+	},
			84	+	})
			85	+	emit("message.delta", map[string]any{"delta": "partial answer"})
			86	+	} else {
			87	+	runTurn(emit, task, 0)
			88	+	}
77		}	89		}
78			90
79		turn := 1	91		turn := 1

@@ -91,10 +103,17 @@ func main() {
91		if parseErr == nil {	103		if parseErr == nil {
92		switch command.Type {	104		switch command.Type {
93		case subagents.CommandAgentShutdown:	105		case subagents.CommandAgentShutdown:
			106	+	if initialTurnOpen {
			107	+	emit("turn_end", map[string]any{"stop": "cancelled"})
			108	+	initialTurnOpen = false
			109	+	}
94		emit("agent_stopped", map[string]any{"reason": "shutdown"})	110		emit("agent_stopped", map[string]any{"reason": "shutdown"})
95		_ = c.Close()	111		_ = c.Close()
96		return	112		return
97		case subagents.CommandTurnCancel:	113		case subagents.CommandTurnCancel:
			114	+	if initialTurnOpen {
			115	+	initialTurnOpen = false
			116	+	}
98		emit("turn_end", map[string]any{"stop": "cancelled"})	117		emit("turn_end", map[string]any{"stop": "cancelled"})
99		case subagents.CommandTurnStart:	118		case subagents.CommandTurnStart:
100		var payload subagents.TurnStartPayload	119		var payload subagents.TurnStartPayload