01a09fb fix(subagents): preserve output after worker deadlines · zut

M docs/subagents.md +4 -0


@@ -123,6 +123,10 @@ subagent://<id>/result
123	123		subagent://<id>/patch
124	124		```
125	125
	126	+	`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.
	127	+
	128	+	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.
	129	+
126	130		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.
127	131
128	132		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.


@@ -123,6 +123,10 @@ subagent://<id>/result
123	subagent://<id>/patch	123		subagent://<id>/patch
124	```	124		```
125		125
		126	+	`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.
		127	+
		128	+	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.
		129	+
126	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.	130		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.
127		131
128	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.	132		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/subagents/agent.go +124 -30


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

@@ -449,17 +453,100 @@ func (a *Agent) appendUserMessage(text string) {
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())
453	489		}
454	490
455	491		func (a *Agent) appendTranscriptLocked(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) replaceStreamingAssistantLocked(text string) {
	503	+	a.removeStreamingAssistantLocked()
	504	+	a.streamingAssistantStart = len(a.transcript)
	505	+	a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
	506	+	a.boundTranscriptLocked()
	507	+	}
	508	+
	509	+	func (a *Agent) removeStreamingAssistantLocked() {
	510	+	if a.streamingAssistantLines == 0 {
	511	+	return
	512	+	}
	513	+	start := a.streamingAssistantStart
	514	+	if start < 0 {
	515	+	start = 0
	516	+	}
	517	+	if start > len(a.transcript) {
	518	+	start = len(a.transcript)
	519	+	}
	520	+	end := start + a.streamingAssistantLines
	521	+	if end > len(a.transcript) {
	522	+	end = len(a.transcript)
	523	+	}
	524	+	for _, line := range a.transcript[start:end] {
	525	+	a.outputBytes -= len(line) + 1
	526	+	a.outputLines--
	527	+	}
	528	+	a.transcript = append(a.transcript[:start], a.transcript[end:]...)
	529	+	a.streamingAssistantLines = 0
	530	+	if a.outputLines == 0 {
	531	+	a.outputBytes = 0
	532	+	}
	533	+	}
	534	+
	535	+	func (a *Agent) clearStreamingAssistantLocked() {
	536	+	a.streamingAssistantText = ""
	537	+	a.streamingAssistantTruncated = false
	538	+	a.streamingAssistantStart = 0
	539	+	a.streamingAssistantLines = 0
	540	+	}
	541	+
	542	+	func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
456	543		message := chunk
457	544		chunk = strings.TrimRight(chunk, "\n")
458	545		if chunk == "" {
459		-	return
	546	+	return 0
460	547		}
461		-	a.mu.Lock()
462		-	for _, line := range strings.Split(chunk, "\n") {
	548	+	lines := strings.Split(chunk, "\n")
	549	+	for _, line := range lines {
463	550		line = linePrefix + line
464	551		a.transcript = append(a.transcript, line)
465	552		a.outputBytes += len(line) + 1

@@ -468,35 +555,42 @@ func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool)
468	555		if assistant {
469	556		a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
470	557		}
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
	558	+	return len(lines)
	559	+	}
	560	+
	561	+	func (a *Agent) outputLimits() (byteCap, lineCap int) {
	562	+	byteCap = a.maxOutputBytes
478	563		if byteCap <= 0 {
479	564		byteCap = 500_000
480	565		}
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
	566	+	lineCap = a.maxOutputLines
	567	+	if lineCap <= 0 {
	568	+	lineCap = 2_000
487	569		}
488		-	for a.outputLines > 0 && a.outputBytes > byteCap {
	570	+	return byteCap, lineCap
	571	+	}
	572	+
	573	+	func (a *Agent) boundTranscriptLocked() {
	574	+	// Bound the inline dashboard copy. The durable event log and session
	575	+	// remain available through the logical history reference.
	576	+	byteCap, lineCap := a.outputLimits()
	577	+	for a.outputLines > lineCap \|\| (a.outputLines > 0 && a.outputBytes > byteCap) {
489	578		line := a.transcript[0]
490	579		a.transcript = a.transcript[1:]
491	580		a.outputBytes -= len(line) + 1
492	581		a.outputLines--
	582	+	if a.streamingAssistantLines > 0 {
	583	+	if a.streamingAssistantStart > 0 {
	584	+	a.streamingAssistantStart--
	585	+	} else {
	586	+	a.streamingAssistantLines--
	587	+	}
	588	+	}
493	589		a.outputTruncated = true
494	590		}
495	591		if a.outputLines == 0 {
496	592		a.outputBytes = 0
497	593		}
498		-	a.mu.Unlock()
499		-	a.markActivity(time.Now())
500	594		}
501	595
502	596		// 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

@@ -449,17 +453,100 @@ func (a *Agent) appendUserMessage(text string) {
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())
453		}	489		}
454			490
455		func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {	491		func (a *Agent) appendTranscriptLocked(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) replaceStreamingAssistantLocked(text string) {
			503	+	a.removeStreamingAssistantLocked()
			504	+	a.streamingAssistantStart = len(a.transcript)
			505	+	a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
			506	+	a.boundTranscriptLocked()
			507	+	}
			508	+
			509	+	func (a *Agent) removeStreamingAssistantLocked() {
			510	+	if a.streamingAssistantLines == 0 {
			511	+	return
			512	+	}
			513	+	start := a.streamingAssistantStart
			514	+	if start < 0 {
			515	+	start = 0
			516	+	}
			517	+	if start > len(a.transcript) {
			518	+	start = len(a.transcript)
			519	+	}
			520	+	end := start + a.streamingAssistantLines
			521	+	if end > len(a.transcript) {
			522	+	end = len(a.transcript)
			523	+	}
			524	+	for _, line := range a.transcript[start:end] {
			525	+	a.outputBytes -= len(line) + 1
			526	+	a.outputLines--
			527	+	}
			528	+	a.transcript = append(a.transcript[:start], a.transcript[end:]...)
			529	+	a.streamingAssistantLines = 0
			530	+	if a.outputLines == 0 {
			531	+	a.outputBytes = 0
			532	+	}
			533	+	}
			534	+
			535	+	func (a *Agent) clearStreamingAssistantLocked() {
			536	+	a.streamingAssistantText = ""
			537	+	a.streamingAssistantTruncated = false
			538	+	a.streamingAssistantStart = 0
			539	+	a.streamingAssistantLines = 0
			540	+	}
			541	+
			542	+	func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
456		message := chunk	543		message := chunk
457		chunk = strings.TrimRight(chunk, "\n")	544		chunk = strings.TrimRight(chunk, "\n")
458		if chunk == "" {	545		if chunk == "" {
459	-	return	546	+	return 0
460		}	547		}
461	-	a.mu.Lock()	548	+	lines := strings.Split(chunk, "\n")
462	-	for _, line := range strings.Split(chunk, "\n") {	549	+	for _, line := range lines {
463		line = linePrefix + line	550		line = linePrefix + line
464		a.transcript = append(a.transcript, line)	551		a.transcript = append(a.transcript, line)
465		a.outputBytes += len(line) + 1	552		a.outputBytes += len(line) + 1

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

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


@@ -678,6 +678,10 @@ func replayTranscriptIntoAgent(a *Agent, evs []Event) {
678	678
679	679		func replayEventTranscript(a *Agent, ev Event) {
680	680		switch ev.Type {
	681	+	case "message.delta":
	682	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
	683	+	a.appendAssistantDelta(delta)
	684	+	}
681	685		case "assistant_message", "user_message":
682	686		var text []string
683	687		if c, ok := ev.Data["content"].([]any); ok {


@@ -678,6 +678,10 @@ func replayTranscriptIntoAgent(a *Agent, evs []Event) {
678		678
679	func replayEventTranscript(a *Agent, ev Event) {	679		func replayEventTranscript(a *Agent, ev Event) {
680	switch ev.Type {	680		switch ev.Type {
		681	+	case "message.delta":
		682	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
		683	+	a.appendAssistantDelta(delta)
		684	+	}
681	case "assistant_message", "user_message":	685		case "assistant_message", "user_message":
682	var text []string	686		var text []string
683	if c, ok := ev.Data["content"].([]any); ok {	687		if c, ok := ev.Data["content"].([]any); ok {

M packages/agent/subagents/result.go +25 -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,19 @@ 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	+	// A canceled worker turn may emit its terminal result before its
	476	+	// provider turns accumulated text into a final assistant message.
	477	+	// 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	+	result.Output = partialOutput
	480	+	}
471	481		if result == nil {
472	482		turnID := a.CurrentTurnID()
473	483		if turnID == "" {
474	484		turnID = fmt.Sprintf("turn-%d", attempt)
475	485		}
476		-	a.mu.Lock()
477		-	output := a.lastAssistant
478		-	a.mu.Unlock()
	486	+	output := partialOutput
479	487		resultStatus := ResultSucceeded
480	488		switch {
481	489		case status == StatusFailed && errors.Is(runErr, context.Canceled):

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

 		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 != "" {
+		// A canceled 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.
+		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 worker was asked to stop cleanly before its process group was
+			// force-killed. Any streamed output received before that point remains
+			// in this result and HistoryRef(agentID), even when the worker's own
+			// cancellation result arrived before the parent classified the deadline.
+			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 +25 -9


@@ -269,21 +269,25 @@ 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	+	wantState TurnStatus
	276	+	wantCode string
	277	+	wantOutput string
276	278		}{
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},
	279	+	{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed, wantCode: "runner_failed"},
	280	+	{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
	281	+	{name: "deadline preserves recovery guidance", status: StatusFailed, runErr: context.DeadlineExceeded, wantState: ResultFailed, wantCode: "deadline_exceeded", wantOutput: "partial answer"},
	282	+	{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
280	283		}
281	284		for _, tt := range tests {
282	285		t.Run(tt.name, func(t *testing.T) {
283	286		stateDir := t.TempDir()
284	287		agent := &Agent{
285		-	ID: "agent-1",
286		-	stateDir: stateDir,
	288	+	ID: "agent-1",
	289	+	stateDir: stateDir,
	290	+	lastAssistant: tt.wantOutput,
287	291		result: &TurnResult{
288	292		AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
289	293		},

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

@@ -304,6 +314,12 @@ func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
304	314		if durable.Status != tt.wantState {
305	315		t.Fatalf("durable result status = %q, want %q", durable.Status, tt.wantState)
306	316		}
	317	+	if durable.Output != tt.wantOutput {
	318	+	t.Fatalf("durable result output = %q, want %q", durable.Output, tt.wantOutput)
	319	+	}
	320	+	if durable.Error == nil \|\| durable.Error.Code != tt.wantCode {
	321	+	t.Fatalf("durable result error = %#v, want code %q", durable.Error, tt.wantCode)
	322	+	}
307	323		})
308	324		}
309	325		}

 func TestEnsureResultReconcilesChildStatusWithRunnerError(t *testing.T) {
 	tests := []struct {
-		name      string
+		name       string
-		status    Status
+		status     Status
-		runErr    error
+		runErr     error
-		wantState TurnStatus
+		wantState  TurnStatus
+		wantCode   string
+		wantOutput string
 	}{
-		{name: "runner failure", status: StatusFailed, runErr: errors.New("runner failed"), wantState: ResultFailed},
+		{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},
+		{name: "runner cancellation", status: StatusFailed, runErr: context.Canceled, wantState: ResultCanceled, wantCode: "runner_failed"},
-		{name: "killed", status: StatusKilled, runErr: context.Canceled, wantState: ResultCanceled},
+		{name: "deadline preserves recovery guidance", status: StatusFailed, runErr: context.DeadlineExceeded, 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()
 			agent := &Agent{
-				ID:       "agent-1",
+				ID:            "agent-1",
-				stateDir: stateDir,
+				stateDir:      stateDir,
+				lastAssistant: tt.wantOutput,
 				result: &TurnResult{
 					AgentID: "agent-1", TurnID: "turn-1", Status: ResultSucceeded,
 				},
 			if result.Error == nil {
 				t.Fatal("runner error was not recorded")
 			}
+			if result.Error.Code != tt.wantCode {
+				t.Fatalf("result error code = %q, want %q", result.Error.Code, tt.wantCode)
+			}
+			if result.Output != tt.wantOutput {
+				t.Fatalf("result output = %q, want %q", result.Output, tt.wantOutput)
+			}
 			durable, err := readTurnResult(stateDir)
 			if err != nil {
 				t.Fatal(err)
 			if durable.Status != tt.wantState {
 				t.Fatalf("durable result status = %q, want %q", durable.Status, tt.wantState)
 			}
+			if durable.Output != tt.wantOutput {
+				t.Fatalf("durable result output = %q, want %q", durable.Output, tt.wantOutput)
+			}
+			if durable.Error == nil || durable.Error.Code != tt.wantCode {
+				t.Fatalf("durable result error = %#v, want code %q", durable.Error, tt.wantCode)
+			}
 		})
 	}
 }

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


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

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

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

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

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

@@ -711,6 +773,9 @@ func applyEventToSink(ev Event, sink Sink) {
711	773		userMessage(string)
712	774		assistantMessage(string)
713	775		}
	776	+	type streamingSink interface {
	777	+	assistantDelta(string)
	778	+	}
714	779		appendMessage := func(text string, assistant bool) {
715	780		if text == "" {
716	781		return

@@ -731,6 +796,14 @@ func applyEventToSink(ev Event, sink Sink) {
731	796		}
732	797
733	798		switch ev.Type {
	799	+	case "message.delta":
	800	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
	801	+	if streaming, ok := sink.(streamingSink); ok {
	802	+	streaming.assistantDelta(delta)
	803	+	} else {
	804	+	sink.Transcript(delta)
	805	+	}
	806	+	}
734	807		case "assistant_message", "user_message":
735	808		var text []string
736	809		if c, ok := ev.Data["content"].([]any); ok {


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

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

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

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

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

@@ -711,6 +773,9 @@ func applyEventToSink(ev Event, sink Sink) {
711		userMessage(string)	773		userMessage(string)
712		assistantMessage(string)	774		assistantMessage(string)
713		}	775		}
			776	+	type streamingSink interface {
			777	+	assistantDelta(string)
			778	+	}
714		appendMessage := func(text string, assistant bool) {	779		appendMessage := func(text string, assistant bool) {
715		if text == "" {	780		if text == "" {
716		return	781		return

@@ -731,6 +796,14 @@ func applyEventToSink(ev Event, sink Sink) {
731		}	796		}
732			797
733		switch ev.Type {	798		switch ev.Type {
			799	+	case "message.delta":
			800	+	if delta, _ := ev.Data["delta"].(string); delta != "" {
			801	+	if streaming, ok := sink.(streamingSink); ok {
			802	+	streaming.assistantDelta(delta)
			803	+	} else {
			804	+	sink.Transcript(delta)
			805	+	}
			806	+	}
734		case "assistant_message", "user_message":	807		case "assistant_message", "user_message":
735		var text []string	808		var text []string
736		if c, ok := ev.Data["content"].([]any); ok {	809		if c, ok := ev.Data["content"].([]any); ok {

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


@@ -12,25 +12,177 @@ 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	+	type controlledDeadlineContext struct{ done chan struct{} }
	29	+
	30	+	func (c *controlledDeadlineContext) Deadline() (time.Time, bool) {
	31	+	return time.Now().Add(time.Hour), true
	32	+	}
	33	+	func (c *controlledDeadlineContext) Done() <-chan struct{} { return c.done }
	34	+	func (c *controlledDeadlineContext) Err() error {
	35	+	select {
	36	+	case <-c.done:
	37	+	return context.DeadlineExceeded
	38	+	default:
	39	+	return nil
	40	+	}
	41	+	}
	42	+	func (c *controlledDeadlineContext) Value(any) any { return nil }
	43	+	func (c *controlledDeadlineContext) expire() {
	44	+	select {
	45	+	case <-c.done:
	46	+	default:
	47	+	close(c.done)
	48	+	}
	49	+	}
	50	+
	51	+	// TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput
	52	+	// verifies that deadline expiry shuts a worker down through its inbox rather
	53	+	// than killing it before its terminal events can be recorded.
	54	+	func TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput(t *testing.T) {
	55	+	if runtime.GOOS == "windows" {
	56	+	t.Skip("unix sockets not supported")
	57	+	}
	58	+	if testing.Short() {
	59	+	t.Skip("skip end-to-end runner test in -short mode")
	60	+	}
	61	+
	62	+	exe := buildStubChild(t)
	63	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
	64	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
	65	+
	66	+	root := t.TempDir()
	67	+	a := &Agent{
	68	+	ID: "deadline-test",
	69	+	Task: "long-running task",
	70	+	Dir: root,
	71	+	SessionPath: filepath.Join(root, "session.jsonl"),
	72	+	InboxPath: filepath.Join(root, "inbox.sock"),
	73	+	EventLogPath: filepath.Join(root, "events.jsonl"),
	74	+	}
	75	+	a.inbox = NewInbox(a.InboxPath)
	76	+	t.Cleanup(func() { _ = a.inbox.Close() })
	77	+
	78	+	r := &execRunner{
	79	+	agent: a,
	80	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
	81	+	GracePeriod: time.Second,
	82	+	}
	83	+	ctx := &controlledDeadlineContext{done: make(chan struct{})}
	84	+	defer ctx.expire()
	85	+	deltas := make(chan struct{}, 1)
	86	+	runDone := make(chan error, 1)
	87	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
	88	+	select {
	89	+	case <-deltas:
	90	+	case <-time.After(time.Second):
	91	+	t.Fatal("worker did not emit its initial delta")
	92	+	}
	93	+	ctx.expire()
	94	+	select {
	95	+	case err := <-runDone:
	96	+	if !errors.Is(err, context.DeadlineExceeded) {
	97	+	t.Fatalf("Run error = %v, want context deadline exceeded", err)
	98	+	}
	99	+	case <-time.After(time.Second):
	100	+	t.Fatal("Run did not return after deadline expiry")
	101	+	}
	102	+
	103	+	events, err := ReadEventLog(a.EventLogPath)
	104	+	if err != nil {
	105	+	t.Fatalf("ReadEventLog: %v", err)
	106	+	}
	107	+	var sawDelta, sawGracefulTurnEnd bool
	108	+	for _, event := range events {
	109	+	if event.Type == "message.delta" && event.Data["delta"] == "partial answer" {
	110	+	sawDelta = true
	111	+	}
	112	+	if event.Type == "turn_end" && event.Data["stop"] == "cancelled" {
	113	+	sawGracefulTurnEnd = true
	114	+	}
	115	+	}
	116	+	if !sawDelta \|\| !sawGracefulTurnEnd {
	117	+	t.Fatalf("deadline did not preserve a graceful partial turn: %s", formatEvents(events))
	118	+	}
	119	+
	120	+	snapshot := a.Snapshot()
	121	+	if snapshot.LastAssistant != "partial answer" {
	122	+	t.Fatalf("live partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
	123	+	}
	124	+	replayed := &Agent{}
	125	+	for _, event := range events {
	126	+	replayEventTranscript(replayed, event)
	127	+	}
	128	+	if snapshot := replayed.Snapshot(); snapshot.LastAssistant != "partial answer" {
	129	+	t.Fatalf("replayed partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
	130	+	}
	131	+	}
	132	+
	133	+	func TestExecRunnerCancelsBlockedWorker(t *testing.T) {
	134	+	if runtime.GOOS == "windows" {
	135	+	t.Skip("unix sockets not supported")
	136	+	}
	137	+	if testing.Short() {
	138	+	t.Skip("skip end-to-end runner test in -short mode")
	139	+	}
	140	+
	141	+	exe := buildStubChild(t)
	142	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
	143	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
	144	+
	145	+	root := t.TempDir()
	146	+	a := &Agent{
	147	+	ID: "cancel-test",
	148	+	Task: "blocked task",
	149	+	Dir: root,
	150	+	SessionPath: filepath.Join(root, "session.jsonl"),
	151	+	InboxPath: filepath.Join(root, "inbox.sock"),
	152	+	EventLogPath: filepath.Join(root, "events.jsonl"),
	153	+	}
	154	+	a.inbox = NewInbox(a.InboxPath)
	155	+	t.Cleanup(func() { _ = a.inbox.Close() })
	156	+
	157	+	r := &execRunner{
	158	+	agent: a,
	159	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
	160	+	}
	161	+	ctx, cancel := context.WithCancel(context.Background())
	162	+	defer cancel()
	163	+	deltas := make(chan struct{}, 1)
	164	+	runDone := make(chan error, 1)
	165	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
	166	+
	167	+	select {
	168	+	case <-deltas:
	169	+	case <-time.After(time.Second):
	170	+	t.Fatal("worker did not emit its initial delta")
	171	+	}
	172	+	cancel()
	173	+	select {
	174	+	case err := <-runDone:
	175	+	if !errors.Is(err, context.Canceled) {
	176	+	t.Fatalf("Run error = %v, want context canceled", err)
	177	+	}
	178	+	case <-time.After(time.Second):
	179	+	t.Fatal("Run did not return after context cancellation")
	180	+	}
	181	+	}
	182	+
	183	+	// TestRunnerEndToEndWithStubChild drives the daemon-mode runner through an
	184	+	// initial turn, a follow-up, and a graceful stop using the stub child. It is
	185	+	// skipped on platforms without Unix sockets.
34	186		func TestRunnerEndToEndWithStubChild(t *testing.T) {
35	187		if runtime.GOOS == "windows" {
36	188		t.Skip("unix sockets not supported")


@@ -12,25 +12,177 @@ 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	+	type controlledDeadlineContext struct{ done chan struct{} }
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	+	func (c *controlledDeadlineContext) Deadline() (time.Time, bool) {
31	-	// stub exits cleanly.	31	+	return time.Now().Add(time.Hour), true
32	-	//	32	+	}
33	-	// The test is skipped on platforms without unix sockets (Windows).	33	+	func (c *controlledDeadlineContext) Done() <-chan struct{} { return c.done }
			34	+	func (c *controlledDeadlineContext) Err() error {
			35	+	select {
			36	+	case <-c.done:
			37	+	return context.DeadlineExceeded
			38	+	default:
			39	+	return nil
			40	+	}
			41	+	}
			42	+	func (c *controlledDeadlineContext) Value(any) any { return nil }
			43	+	func (c *controlledDeadlineContext) expire() {
			44	+	select {
			45	+	case <-c.done:
			46	+	default:
			47	+	close(c.done)
			48	+	}
			49	+	}
			50	+
			51	+	// TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput
			52	+	// verifies that deadline expiry shuts a worker down through its inbox rather
			53	+	// than killing it before its terminal events can be recorded.
			54	+	func TestExecRunnerDeadlineGracefullyStopsWorkerAndPreservesStreamedOutput(t *testing.T) {
			55	+	if runtime.GOOS == "windows" {
			56	+	t.Skip("unix sockets not supported")
			57	+	}
			58	+	if testing.Short() {
			59	+	t.Skip("skip end-to-end runner test in -short mode")
			60	+	}
			61	+
			62	+	exe := buildStubChild(t)
			63	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
			64	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
			65	+
			66	+	root := t.TempDir()
			67	+	a := &Agent{
			68	+	ID: "deadline-test",
			69	+	Task: "long-running task",
			70	+	Dir: root,
			71	+	SessionPath: filepath.Join(root, "session.jsonl"),
			72	+	InboxPath: filepath.Join(root, "inbox.sock"),
			73	+	EventLogPath: filepath.Join(root, "events.jsonl"),
			74	+	}
			75	+	a.inbox = NewInbox(a.InboxPath)
			76	+	t.Cleanup(func() { _ = a.inbox.Close() })
			77	+
			78	+	r := &execRunner{
			79	+	agent: a,
			80	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
			81	+	GracePeriod: time.Second,
			82	+	}
			83	+	ctx := &controlledDeadlineContext{done: make(chan struct{})}
			84	+	defer ctx.expire()
			85	+	deltas := make(chan struct{}, 1)
			86	+	runDone := make(chan error, 1)
			87	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
			88	+	select {
			89	+	case <-deltas:
			90	+	case <-time.After(time.Second):
			91	+	t.Fatal("worker did not emit its initial delta")
			92	+	}
			93	+	ctx.expire()
			94	+	select {
			95	+	case err := <-runDone:
			96	+	if !errors.Is(err, context.DeadlineExceeded) {
			97	+	t.Fatalf("Run error = %v, want context deadline exceeded", err)
			98	+	}
			99	+	case <-time.After(time.Second):
			100	+	t.Fatal("Run did not return after deadline expiry")
			101	+	}
			102	+
			103	+	events, err := ReadEventLog(a.EventLogPath)
			104	+	if err != nil {
			105	+	t.Fatalf("ReadEventLog: %v", err)
			106	+	}
			107	+	var sawDelta, sawGracefulTurnEnd bool
			108	+	for _, event := range events {
			109	+	if event.Type == "message.delta" && event.Data["delta"] == "partial answer" {
			110	+	sawDelta = true
			111	+	}
			112	+	if event.Type == "turn_end" && event.Data["stop"] == "cancelled" {
			113	+	sawGracefulTurnEnd = true
			114	+	}
			115	+	}
			116	+	if !sawDelta \|\| !sawGracefulTurnEnd {
			117	+	t.Fatalf("deadline did not preserve a graceful partial turn: %s", formatEvents(events))
			118	+	}
			119	+
			120	+	snapshot := a.Snapshot()
			121	+	if snapshot.LastAssistant != "partial answer" {
			122	+	t.Fatalf("live partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
			123	+	}
			124	+	replayed := &Agent{}
			125	+	for _, event := range events {
			126	+	replayEventTranscript(replayed, event)
			127	+	}
			128	+	if snapshot := replayed.Snapshot(); snapshot.LastAssistant != "partial answer" {
			129	+	t.Fatalf("replayed partial output = %q, want %q", snapshot.LastAssistant, "partial answer")
			130	+	}
			131	+	}
			132	+
			133	+	func TestExecRunnerCancelsBlockedWorker(t *testing.T) {
			134	+	if runtime.GOOS == "windows" {
			135	+	t.Skip("unix sockets not supported")
			136	+	}
			137	+	if testing.Short() {
			138	+	t.Skip("skip end-to-end runner test in -short mode")
			139	+	}
			140	+
			141	+	exe := buildStubChild(t)
			142	+	t.Setenv("ZUT_STUB_PROTOCOL", "1")
			143	+	t.Setenv("ZUT_STUB_BLOCK_INITIAL", "1")
			144	+
			145	+	root := t.TempDir()
			146	+	a := &Agent{
			147	+	ID: "cancel-test",
			148	+	Task: "blocked task",
			149	+	Dir: root,
			150	+	SessionPath: filepath.Join(root, "session.jsonl"),
			151	+	InboxPath: filepath.Join(root, "inbox.sock"),
			152	+	EventLogPath: filepath.Join(root, "events.jsonl"),
			153	+	}
			154	+	a.inbox = NewInbox(a.InboxPath)
			155	+	t.Cleanup(func() { _ = a.inbox.Close() })
			156	+
			157	+	r := &execRunner{
			158	+	agent: a,
			159	+	Command: subagentWorkerArgs(subagentWorkerArgsOpts{Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task}),
			160	+	}
			161	+	ctx, cancel := context.WithCancel(context.Background())
			162	+	defer cancel()
			163	+	deltas := make(chan struct{}, 1)
			164	+	runDone := make(chan error, 1)
			165	+	go func() { runDone <- r.Run(ctx, notifyingAgentSink{agentSink: agentSink{a: a}, deltas: deltas}) }()
			166	+
			167	+	select {
			168	+	case <-deltas:
			169	+	case <-time.After(time.Second):
			170	+	t.Fatal("worker did not emit its initial delta")
			171	+	}
			172	+	cancel()
			173	+	select {
			174	+	case err := <-runDone:
			175	+	if !errors.Is(err, context.Canceled) {
			176	+	t.Fatalf("Run error = %v, want context canceled", err)
			177	+	}
			178	+	case <-time.After(time.Second):
			179	+	t.Fatal("Run did not return after context cancellation")
			180	+	}
			181	+	}
			182	+
			183	+	// TestRunnerEndToEndWithStubChild drives the daemon-mode runner through an
			184	+	// initial turn, a follow-up, and a graceful stop using the stub child. It is
			185	+	// skipped on platforms without Unix sockets.
34		func TestRunnerEndToEndWithStubChild(t *testing.T) {	186		func TestRunnerEndToEndWithStubChild(t *testing.T) {
35		if runtime.GOOS == "windows" {	187		if runtime.GOOS == "windows" {
36		t.Skip("unix sockets not supported")	188		t.Skip("unix sockets not supported")

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


@@ -58,6 +58,62 @@ 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	+
61	117		func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {
62	118		t.Setenv("OPENAI_API_KEY", "openai-secret")
63	119		t.Setenv("CUSTOM_PROVIDER_API_KEY", "custom-secret")


@@ -58,6 +58,62 @@ 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	+
61	func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {	117		func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {
62	t.Setenv("OPENAI_API_KEY", "openai-secret")	118		t.Setenv("OPENAI_API_KEY", "openai-secret")
63	t.Setenv("CUSTOM_PROVIDER_API_KEY", "custom-secret")	119		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	+	streamingOutputTruncated := a.streamingAssistantTruncated
1132	1137		finished := a.finished
1133	1138		a.mu.Unlock()
1134	1139

@@ -1143,7 +1148,7 @@ 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
	1151	+	outputTruncated := a.outputTruncated \|\| streamingOutputTruncated
1147	1152		a.lifecycleMu.Unlock()
1148	1153		return AgentSnapshot{
1149	1154		ID: a.ID, Task: a.Task, Dir: a.Dir,

@@ -1208,6 +1213,7 @@ func (s agentSink) Transcript(chunk string) {
1208	1213		}
1209	1214		func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1210	1215		func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
	1216	+	func (s agentSink) assistantDelta(text string) { s.a.appendAssistantDelta(text) }
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
+	streamingOutputTruncated := a.streamingAssistantTruncated
 	finished := a.finished
 	a.mu.Unlock()
 	resultRef := a.resultRef
 	patchRef := a.patchRef
 	changedFiles := append([]string(nil), a.changedFiles...)
-	outputTruncated := a.outputTruncated
+	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 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