4a12e83 fix(subagents): slide completed results into active parent turns (#179) · zut

M docs/subagents.md +11 -0


@@ -184,6 +184,17 @@ an explicit `wait` value. For an unwaited spawn, completion arrives through the
184	184		host’s typed completion update; `subagent_status` returns immediately and does
185	185		not wait for completion. Do not use sleep loops, repeated status calls, journal
186	186		files, or terminal UI inspection as a completion signal.
	187	+	In interactive mode, a result received while the primary is busy enters the
	188	+	visible sliding in queue and reaches the model at its next safe boundary,
	189	+	without interrupting a model request or tool call and without waiting for other
	190	+	children. Results already queued are not repeated when the parent turn ends.
	191	+	Undelivered results remain queued after a provider or compaction error for the
	192	+	next turn; they do not automatically retry the failed request. Explicit
	193	+	cancellation clears pending queue entries. The queue is in memory, while the
	194	+	child’s saved result remains available through `subagent_status`.
	195	+	When the primary is idle, remaining children are collected into one completion
	196	+	wave. Headless orchestration still waits for every accepted child before its
	197	+	next parent turn.
187	198		Successful completions include the final visible assistant summary, capped at
188	199		256 KiB; open the child session for the complete durable transcript.
189	200


@@ -184,6 +184,17 @@ an explicit `wait` value. For an unwaited spawn, completion arrives through the
184	host’s typed completion update; `subagent_status` returns immediately and does	184		host’s typed completion update; `subagent_status` returns immediately and does
185	not wait for completion. Do not use sleep loops, repeated status calls, journal	185		not wait for completion. Do not use sleep loops, repeated status calls, journal
186	files, or terminal UI inspection as a completion signal.	186		files, or terminal UI inspection as a completion signal.
		187	+	In interactive mode, a result received while the primary is busy enters the
		188	+	visible sliding in queue and reaches the model at its next safe boundary,
		189	+	without interrupting a model request or tool call and without waiting for other
		190	+	children. Results already queued are not repeated when the parent turn ends.
		191	+	Undelivered results remain queued after a provider or compaction error for the
		192	+	next turn; they do not automatically retry the failed request. Explicit
		193	+	cancellation clears pending queue entries. The queue is in memory, while the
		194	+	child’s saved result remains available through `subagent_status`.
		195	+	When the primary is idle, remaining children are collected into one completion
		196	+	wave. Headless orchestration still waits for every accepted child before its
		197	+	next parent turn.
187	Successful completions include the final visible assistant summary, capped at	198		Successful completions include the final visible assistant summary, capped at
188	256 KiB; open the child session for the complete durable transcript.	199		256 KiB; open the child session for the complete durable transcript.
189		200

M packages/agent/internal/orchestration/coordinator.go +11 -5


@@ -9,10 +9,10 @@ import (
9	9		"github.com/bnema/zut/packages/provider"
10	10		)
11	11
12		-	// Coordinator collects one sealed worker wave per manager turn. A terminal
13		-	// worker cannot wake the manager until the owning turn is sealed. Queued user
14		-	// input wins the next wake; an active goal is used only when no user input or
15		-	// worker result remains.
	12	+	// Coordinator queues terminal worker results into an active manager turn.
	13	+	// When the manager is idle, it collects the remaining sealed worker wave.
	14	+	// Queued user input wins the next wake; an active goal is used only when no
	15	+	// user input or worker result remains.
16	16		type Coordinator struct {
17	17		managerActive bool
18	18		waveSealed bool

@@ -29,6 +29,7 @@ type queuedUserInput struct {
29	29		}
30	30
31	31		type worker struct {
	32	+	delivered bool
32	33		terminal bool
33	34		completion subagents.Completion
34	35		}

@@ -61,6 +62,7 @@ const (
61	62		ActionRunManager ActionKind = iota + 1
62	63		ActionWait
63	64		ActionStop
	65	+	ActionQueueManager
64	66		)
65	67
66	68		type WakeReason uint8

@@ -130,6 +132,10 @@ func (c *Coordinator) Apply(event Event) Result {
130	132		}
131	133		w.terminal = true
132	134		w.completion = event.Completion
	135	+	if c.managerActive {
	136	+	w.delivered = true
	137	+	return Result{Actions: []Action{{Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{event.Completion}}}}
	138	+	}
133	139		if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() {
134	140		return c.wakeIfIdle()
135	141		}

@@ -200,7 +206,7 @@ func (c *Coordinator) completedWorkers() []subagents.Completion {
200	206		}
201	207		completions := make([]subagents.Completion, 0, len(c.workerOrder))
202	208		for _, workerID := range c.workerOrder {
203		-	if worker := c.workers[workerID]; worker != nil && worker.terminal {
	209	+	if worker := c.workers[workerID]; worker != nil && worker.terminal && !worker.delivered {
204	210		completions = append(completions, worker.completion)
205	211		}
206	212		}

 	"github.com/bnema/zut/packages/provider"
 )
-// Coordinator collects one sealed worker wave per manager turn. A terminal
+// Coordinator queues terminal worker results into an active manager turn.
-// worker cannot wake the manager until the owning turn is sealed. Queued user
+// When the manager is idle, it collects the remaining sealed worker wave.
-// input wins the next wake; an active goal is used only when no user input or
+// Queued user input wins the next wake; an active goal is used only when no
-// worker result remains.
+// user input or worker result remains.
 type Coordinator struct {
 	managerActive bool
 	waveSealed    bool
 }
 type worker struct {
+	delivered  bool
 	terminal   bool
 	completion subagents.Completion
 }
 	ActionRunManager ActionKind = iota + 1
 	ActionWait
 	ActionStop
+	ActionQueueManager
 )
 type WakeReason uint8
 		}
 		w.terminal = true
 		w.completion = event.Completion
+		if c.managerActive {
+			w.delivered = true
+			return Result{Actions: []Action{{Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{event.Completion}}}}
+		}
 		if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() {
 			return c.wakeIfIdle()
 		}
 	}
 	completions := make([]subagents.Completion, 0, len(c.workerOrder))
 	for _, workerID := range c.workerOrder {
-		if worker := c.workers[workerID]; worker != nil && worker.terminal {
+		if worker := c.workers[workerID]; worker != nil && worker.terminal && !worker.delivered {
 			completions = append(completions, worker.completion)
 		}
 	}

M packages/agent/internal/orchestration/coordinator_test.go +24 -0


@@ -20,6 +20,30 @@ func TestCoordinatorSealsWorkerWaveAndWakesOnce(t *testing.T) {
20	20		}})
21	21		}
22	22
	23	+	func TestCoordinatorSlidesCompletionIntoActiveManagerOnce(t *testing.T) {
	24	+	c := New()
	25	+	c.Apply(Event{Kind: EventManagerStarted})
	26	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"})
	27	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"})
	28	+	completion := subagents.Completion{AgentID: "a", Status: "completed"}
	29	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), []Action{{
	30	+	Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{completion},
	31	+	}})
	32	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), nil)
	33	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}})
	34	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b", Completion: subagents.Completion{AgentID: "b"}}), []Action{{
	35	+	Kind: ActionRunManager, Reason: WakeWorkers, Completions: []subagents.Completion{{AgentID: "b"}},
	36	+	}})
	37	+	}
	38	+
	39	+	func TestCoordinatorDoesNotWakeForAlreadyQueuedCompletion(t *testing.T) {
	40	+	c := New()
	41	+	c.Apply(Event{Kind: EventManagerStarted})
	42	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"})
	43	+	c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a"})
	44	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), nil)
	45	+	}
	46	+
23	47		func TestCoordinatorGivesQueuedUserInputPriority(t *testing.T) {
24	48		c := New()
25	49		c.Apply(Event{Kind: EventManagerStarted})


@@ -20,6 +20,30 @@ func TestCoordinatorSealsWorkerWaveAndWakesOnce(t *testing.T) {
20	}})	20		}})
21	}	21		}
22		22
		23	+	func TestCoordinatorSlidesCompletionIntoActiveManagerOnce(t *testing.T) {
		24	+	c := New()
		25	+	c.Apply(Event{Kind: EventManagerStarted})
		26	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"})
		27	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"})
		28	+	completion := subagents.Completion{AgentID: "a", Status: "completed"}
		29	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), []Action{{
		30	+	Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{completion},
		31	+	}})
		32	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), nil)
		33	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}})
		34	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b", Completion: subagents.Completion{AgentID: "b"}}), []Action{{
		35	+	Kind: ActionRunManager, Reason: WakeWorkers, Completions: []subagents.Completion{{AgentID: "b"}},
		36	+	}})
		37	+	}
		38	+
		39	+	func TestCoordinatorDoesNotWakeForAlreadyQueuedCompletion(t *testing.T) {
		40	+	c := New()
		41	+	c.Apply(Event{Kind: EventManagerStarted})
		42	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"})
		43	+	c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a"})
		44	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), nil)
		45	+	}
		46	+
23	func TestCoordinatorGivesQueuedUserInputPriority(t *testing.T) {	47		func TestCoordinatorGivesQueuedUserInputPriority(t *testing.T) {
24	c := New()	48		c := New()
25	c.Apply(Event{Kind: EventManagerStarted})	49		c.Apply(Event{Kind: EventManagerStarted})

M packages/agent/modes/interactive.go +3 -4


@@ -2102,10 +2102,9 @@ type telegramSenderAdapter struct {
2102	2102		// picked up before the waiter exits, so a later worker cannot be lost in the
2103	2103		// handoff between batches.
2104	2104
2105		-	// beginCompletionDeliveryHold keeps all completions observed during one
2106		-	// parent model turn together. Tool calls are executed sequentially inside the
2107		-	// core agent, so the release point is the owning registration boundary for
2108		-	// same-parent-turn batching rather than a timing-based debounce.
	2105	+	// beginCompletionDeliveryHold marks the owning parent turn's registration
	2106	+	// boundary. Results slide into an active turn; release seals the remaining
	2107	+	// worker wave for idle-parent batching rather than delaying active delivery.
2109	2108
2110	2109		// autoSubagentsAddenda returns the prompt blocks owned by subagent prompting,
2111	2110		// in the same order Resolve appends them to a new agent. The profile manifest


@@ -2102,10 +2102,9 @@ type telegramSenderAdapter struct {
2102		// picked up before the waiter exits, so a later worker cannot be lost in the	2102		// picked up before the waiter exits, so a later worker cannot be lost in the
2103		// handoff between batches.	2103		// handoff between batches.
2104			2104
2105	-	// beginCompletionDeliveryHold keeps all completions observed during one	2105	+	// beginCompletionDeliveryHold marks the owning parent turn's registration
2106	-	// parent model turn together. Tool calls are executed sequentially inside the	2106	+	// boundary. Results slide into an active turn; release seals the remaining
2107	-	// core agent, so the release point is the owning registration boundary for	2107	+	// worker wave for idle-parent batching rather than delaying active delivery.
2108	-	// same-parent-turn batching rather than a timing-based debounce.
2109			2108
2110		// autoSubagentsAddenda returns the prompt blocks owned by subagent prompting,	2109		// autoSubagentsAddenda returns the prompt blocks owned by subagent prompting,
2111		// in the same order Resolve appends them to a new agent. The profile manifest	2110		// in the same order Resolve appends them to a new agent. The profile manifest

M packages/agent/modes/interactive_compaction.go +6 -17


@@ -269,8 +269,7 @@ func (i *Interactive) runCompact(parent context.Context, request compactContinua
269	269		// continues the user message already in the transcript. Pre-turn
270	270		// compaction preserves its complete text-and-images request. Regular
271	271		// prompts typed during compaction remain in the host queue.
272		-	var next string
273		-	var nextImages []provider.ImageBlock
	272	+	var next core.QueuedMessage
274	273		var hasNext bool
275	274		var continueExisting bool
276	275		var continueAutomatically bool

@@ -286,21 +285,15 @@ func (i *Interactive) runCompact(parent context.Context, request compactContinua
286	285		} else {
287	286		i.statusOK = "compaction cancelled"
288	287		}
289		-	i.queued = nil // drop queue on cancel
	288	+	i.discardQueuedMessagesLocked(false)
290	289		i.clearPendingCompactTurnLocked()
291	290		handoff, persistHandoff = i.resetCompactContinuationLocked()
292		-	if i.agent != nil {
293		-	i.agent.DrainQueuedMessages()
294		-	}
295	291		case err != nil:
296	292		i.statusErr = "compaction failed: " + err.Error()
297	293		i.statusOK = ""
298		-	i.queued = nil // drop queue on error
	294	+	i.discardQueuedMessagesLocked(true)
299	295		i.clearPendingCompactTurnLocked()
300	296		handoff, persistHandoff = i.resetCompactContinuationLocked()
301		-	if i.agent != nil {
302		-	i.agent.DrainQueuedMessages()
303		-	}
304	297		default:
305	298		i.statusErr = ""
306	299		// Read token count from the compaction message meta.

@@ -329,8 +322,7 @@ func (i *Interactive) runCompact(parent context.Context, request compactContinua
329	322		continueExisting = true
330	323		i.continueAfterCompact = false
331	324		case i.hasPendingCompactPrompt:
332		-	next = i.pendingCompactPrompt
333		-	nextImages = i.pendingCompactImages
	325	+	next = core.QueuedMessage{Text: i.pendingCompactPrompt, Images: i.pendingCompactImages}
334	326		i.pendingCompactPrompt = ""
335	327		i.pendingCompactImages = nil
336	328		i.hasPendingCompactPrompt = false

@@ -341,10 +333,7 @@ func (i *Interactive) runCompact(parent context.Context, request compactContinua
341	333		handoff, persistHandoff = i.setCompactContinuationLocked(compactContinuationState{reason: continuationReason})
342	334		continueAutomatically = true
343	335		case len(i.queued) > 0:
344		-	queued := i.queued[0]
345		-	i.queued = i.queued[1:]
346		-	next = queued.Text
347		-	nextImages = queued.Images
	336	+	next, i.queued = i.queued[0], i.queued[1:]
348	337		hasNext = true
349	338		handoff, persistHandoff = i.resetCompactContinuationLocked()
350	339		case continuationReason == compactContinuationStructuralTail \|\| continuationReason == compactContinuationStatusRescue:

@@ -395,7 +384,7 @@ func (i *Interactive) runCompact(parent context.Context, request compactContinua
395	384		case continueExisting:
396	385		i.startTurnRequest(p, "", nil, true, true)
397	386		case hasNext:
398		-	i.startTurnWithImages(p, next, nextImages)
	387	+	i.startQueuedTurn(p, next)
399	388		case continueAutomatically:
400	389		i.startAutoCompactContinuation(p)
401	390		case continueGoal:

 		// continues the user message already in the transcript. Pre-turn
 		// compaction preserves its complete text-and-images request. Regular
 		// prompts typed during compaction remain in the host queue.
-		var next string
+		var next core.QueuedMessage
-		var nextImages []provider.ImageBlock
 		var hasNext bool
 		var continueExisting bool
 		var continueAutomatically bool
 			} else {
 				i.statusOK = "compaction cancelled"
 			}
-			i.queued = nil // drop queue on cancel
+			i.discardQueuedMessagesLocked(false)
 			i.clearPendingCompactTurnLocked()
 			handoff, persistHandoff = i.resetCompactContinuationLocked()
-			if i.agent != nil {
-				i.agent.DrainQueuedMessages()
-			}
 		case err != nil:
 			i.statusErr = "compaction failed: " + err.Error()
 			i.statusOK = ""
-			i.queued = nil // drop queue on error
+			i.discardQueuedMessagesLocked(true)
 			i.clearPendingCompactTurnLocked()
 			handoff, persistHandoff = i.resetCompactContinuationLocked()
-			if i.agent != nil {
-				i.agent.DrainQueuedMessages()
-			}
 		default:
 			i.statusErr = ""
 			// Read token count from the compaction message meta.
 				continueExisting = true
 				i.continueAfterCompact = false
 			case i.hasPendingCompactPrompt:
-				next = i.pendingCompactPrompt
+				next = core.QueuedMessage{Text: i.pendingCompactPrompt, Images: i.pendingCompactImages}
-				nextImages = i.pendingCompactImages
 				i.pendingCompactPrompt = ""
 				i.pendingCompactImages = nil
 				i.hasPendingCompactPrompt = false
 				handoff, persistHandoff = i.setCompactContinuationLocked(compactContinuationState{reason: continuationReason})
 				continueAutomatically = true
 			case len(i.queued) > 0:
-				queued := i.queued[0]
+				next, i.queued = i.queued[0], i.queued[1:]
-				i.queued = i.queued[1:]
-				next = queued.Text
-				nextImages = queued.Images
 				hasNext = true
 				handoff, persistHandoff = i.resetCompactContinuationLocked()
 			case continuationReason == compactContinuationStructuralTail || continuationReason == compactContinuationStatusRescue:
 			case continueExisting:
 				i.startTurnRequest(p, "", nil, true, true)
 			case hasNext:
-				i.startTurnWithImages(p, next, nextImages)
+				i.startQueuedTurn(p, next)
 			case continueAutomatically:
 				i.startAutoCompactContinuation(p)
 			case continueGoal:

A packages/agent/modes/interactive_completion_error_test.go +129 -0

@@ -0,0 +1,129 @@
	1	+	package modes
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"strings"
	7	+	"sync/atomic"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/bnema/zut/packages/agent/subagents"
	12	+	"github.com/bnema/zut/packages/core"
	13	+	"github.com/bnema/zut/packages/provider"
	14	+	)
	15	+
	16	+	type completionErrorClient struct {
	17	+	calls atomic.Int32
	18	+	started chan struct{}
	19	+	release chan struct{}
	20	+	requests chan provider.Request
	21	+	}
	22	+
	23	+	func (*completionErrorClient) Name() string { return "completion-error-test" }
	24	+	func (c *completionErrorClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
	25	+	if c.calls.Add(1) == 1 {
	26	+	close(c.started)
	27	+	select {
	28	+	case <-ctx.Done():
	29	+	return nil, ctx.Err()
	30	+	case <-c.release:
	31	+	return nil, errors.New("synthetic request failure")
	32	+	}
	33	+	}
	34	+	c.requests <- req
	35	+	out := make(chan provider.Event, 1)
	36	+	out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "done"}}}}
	37	+	close(out)
	38	+	return out, nil
	39	+	}
	40	+
	41	+	func TestCompletionSurvivesParentError(t *testing.T) {
	42	+	for _, name := range []string{"provider", "compaction", "idle-pre-turn-compaction"} {
	43	+	t.Run(name, func(t *testing.T) {
	44	+	ctx, cancel := context.WithCancel(context.Background())
	45	+	defer cancel()
	46	+	client := &completionErrorClient{started: make(chan struct{}), release: make(chan struct{}), requests: make(chan provider.Request, 4)}
	47	+	ag := core.NewAgent(client, "test-model", "", nil)
	48	+	ag.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "existing task"}}}})
	49	+	threshold := 70
	50	+	i := NewInteractive(InteractiveConfig{Agent: ag, Provider: "anthropic", Model: "claude-sonnet-4-5", AutoCompactThreshold: &threshold})
	51	+	i.runCtx = ctx
	52	+	report := func() {
	53	+	i.TrackResidentSubagent("worker", "turn")
	54	+	i.ReportResidentSubagent(subagents.ResidentCompletion{ChildID: "worker", TurnID: "turn", Summary: "retained evidence"})
	55	+	}
	56	+	switch name {
	57	+	case "idle-pre-turn-compaction":
	58	+	i.lastCtxInput = 150000
	59	+	report()
	60	+	case "compaction":
	61	+	i.runCompact(ctx, compactContinuationRequest{origin: compactOriginManual})
	62	+	default:
	63	+	i.startTurn(ctx, "work")
	64	+	}
	65	+	select {
	66	+	case <-client.started:
	67	+	case <-time.After(2 * time.Second):
	68	+	t.Fatal("request did not start")
	69	+	}
	70	+	if name != "idle-pre-turn-compaction" {
	71	+	report()
	72	+	}
	73	+	deadline := time.Now().Add(2 * time.Second)
	74	+	for {
	75	+	i.mu.Lock()
	76	+	count := len(i.queued) + ag.QueuedMessageCount()
	77	+	i.mu.Unlock()
	78	+	if count != 0 {
	79	+	break
	80	+	}
	81	+	if time.Now().After(deadline) {
	82	+	t.Fatal("completion did not enter queue")
	83	+	}
	84	+	time.Sleep(time.Millisecond)
	85	+	}
	86	+	i.submitOrQueue("stale user input", nil, true)
	87	+	close(client.release)
	88	+	waitInteractiveIdle(t, i)
	89	+	pending := ag.PendingQueuedMessages()
	90	+	if len(pending) != 1 \|\| !pending[0].HostEvent \|\| !strings.Contains(pending[0].Text, "retained evidence") {
	91	+	t.Fatalf("pending after error = %#v", pending)
	92	+	}
	93	+	if client.calls.Load() != 1 {
	94	+	t.Fatal("error automatically retried")
	95	+	}
	96	+	i.mu.Lock()
	97	+	i.lastCtxInput = 0 // the explicit retry no longer needs compaction
	98	+	i.mu.Unlock()
	99	+	i.startTurn(ctx, "retry now")
	100	+	select {
	101	+	case req := <-client.requests:
	102	+	if got := requestUserTextCount(req, pending[0].Text); got != 1 {
	103	+	t.Fatalf("completion copies = %d, want 1", got)
	104	+	}
	105	+	if requestUserTextCount(req, "stale user input") != 0 {
	106	+	t.Fatal("stale user input survived error")
	107	+	}
	108	+	case <-time.After(2 * time.Second):
	109	+	t.Fatal("explicit retry did not start")
	110	+	}
	111	+	waitInteractiveIdle(t, i)
	112	+	if ag.QueuedMessageCount() != 0 {
	113	+	t.Fatal("completion remained queued after delivery")
	114	+	}
	115	+	})
	116	+	}
	117	+	}
	118	+
	119	+	func TestExplicitCancellationDiscardsQueuedHostEvents(t *testing.T) {
	120	+	ag := core.NewAgent(nil, "test-model", "", nil)
	121	+	ag.QueuePrompt(core.QueuedMessage{Text: "agent event", HostEvent: true})
	122	+	i := &Interactive{agent: ag, queued: []core.QueuedMessage{{Text: "host event", HostEvent: true}}}
	123	+	i.mu.Lock()
	124	+	i.discardQueuedMessagesLocked(false)
	125	+	i.mu.Unlock()
	126	+	if ag.QueuedMessageCount() != 0 \|\| len(i.queued) != 0 {
	127	+	t.Fatal("explicit cancellation retained queued events")
	128	+	}
	129	+	}

@@ -0,0 +1,129 @@
	1	+	package modes
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"strings"
	7	+	"sync/atomic"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/bnema/zut/packages/agent/subagents"
	12	+	"github.com/bnema/zut/packages/core"
	13	+	"github.com/bnema/zut/packages/provider"
	14	+	)
	15	+
	16	+	type completionErrorClient struct {
	17	+	calls atomic.Int32
	18	+	started chan struct{}
	19	+	release chan struct{}
	20	+	requests chan provider.Request
	21	+	}
	22	+
	23	+	func (*completionErrorClient) Name() string { return "completion-error-test" }
	24	+	func (c *completionErrorClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
	25	+	if c.calls.Add(1) == 1 {
	26	+	close(c.started)
	27	+	select {
	28	+	case <-ctx.Done():
	29	+	return nil, ctx.Err()
	30	+	case <-c.release:
	31	+	return nil, errors.New("synthetic request failure")
	32	+	}
	33	+	}
	34	+	c.requests <- req
	35	+	out := make(chan provider.Event, 1)
	36	+	out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "done"}}}}
	37	+	close(out)
	38	+	return out, nil
	39	+	}
	40	+
	41	+	func TestCompletionSurvivesParentError(t *testing.T) {
	42	+	for _, name := range []string{"provider", "compaction", "idle-pre-turn-compaction"} {
	43	+	t.Run(name, func(t *testing.T) {
	44	+	ctx, cancel := context.WithCancel(context.Background())
	45	+	defer cancel()
	46	+	client := &completionErrorClient{started: make(chan struct{}), release: make(chan struct{}), requests: make(chan provider.Request, 4)}
	47	+	ag := core.NewAgent(client, "test-model", "", nil)
	48	+	ag.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "existing task"}}}})
	49	+	threshold := 70
	50	+	i := NewInteractive(InteractiveConfig{Agent: ag, Provider: "anthropic", Model: "claude-sonnet-4-5", AutoCompactThreshold: &threshold})
	51	+	i.runCtx = ctx
	52	+	report := func() {
	53	+	i.TrackResidentSubagent("worker", "turn")
	54	+	i.ReportResidentSubagent(subagents.ResidentCompletion{ChildID: "worker", TurnID: "turn", Summary: "retained evidence"})
	55	+	}
	56	+	switch name {
	57	+	case "idle-pre-turn-compaction":
	58	+	i.lastCtxInput = 150000
	59	+	report()
	60	+	case "compaction":
	61	+	i.runCompact(ctx, compactContinuationRequest{origin: compactOriginManual})
	62	+	default:
	63	+	i.startTurn(ctx, "work")
	64	+	}
	65	+	select {
	66	+	case <-client.started:
	67	+	case <-time.After(2 * time.Second):
	68	+	t.Fatal("request did not start")
	69	+	}
	70	+	if name != "idle-pre-turn-compaction" {
	71	+	report()
	72	+	}
	73	+	deadline := time.Now().Add(2 * time.Second)
	74	+	for {
	75	+	i.mu.Lock()
	76	+	count := len(i.queued) + ag.QueuedMessageCount()
	77	+	i.mu.Unlock()
	78	+	if count != 0 {
	79	+	break
	80	+	}
	81	+	if time.Now().After(deadline) {
	82	+	t.Fatal("completion did not enter queue")
	83	+	}
	84	+	time.Sleep(time.Millisecond)
	85	+	}
	86	+	i.submitOrQueue("stale user input", nil, true)
	87	+	close(client.release)
	88	+	waitInteractiveIdle(t, i)
	89	+	pending := ag.PendingQueuedMessages()
	90	+	if len(pending) != 1 \|\| !pending[0].HostEvent \|\| !strings.Contains(pending[0].Text, "retained evidence") {
	91	+	t.Fatalf("pending after error = %#v", pending)
	92	+	}
	93	+	if client.calls.Load() != 1 {
	94	+	t.Fatal("error automatically retried")
	95	+	}
	96	+	i.mu.Lock()
	97	+	i.lastCtxInput = 0 // the explicit retry no longer needs compaction
	98	+	i.mu.Unlock()
	99	+	i.startTurn(ctx, "retry now")
	100	+	select {
	101	+	case req := <-client.requests:
	102	+	if got := requestUserTextCount(req, pending[0].Text); got != 1 {
	103	+	t.Fatalf("completion copies = %d, want 1", got)
	104	+	}
	105	+	if requestUserTextCount(req, "stale user input") != 0 {
	106	+	t.Fatal("stale user input survived error")
	107	+	}
	108	+	case <-time.After(2 * time.Second):
	109	+	t.Fatal("explicit retry did not start")
	110	+	}
	111	+	waitInteractiveIdle(t, i)
	112	+	if ag.QueuedMessageCount() != 0 {
	113	+	t.Fatal("completion remained queued after delivery")
	114	+	}
	115	+	})
	116	+	}
	117	+	}
	118	+
	119	+	func TestExplicitCancellationDiscardsQueuedHostEvents(t *testing.T) {
	120	+	ag := core.NewAgent(nil, "test-model", "", nil)
	121	+	ag.QueuePrompt(core.QueuedMessage{Text: "agent event", HostEvent: true})
	122	+	i := &Interactive{agent: ag, queued: []core.QueuedMessage{{Text: "host event", HostEvent: true}}}
	123	+	i.mu.Lock()
	124	+	i.discardQueuedMessagesLocked(false)
	125	+	i.mu.Unlock()
	126	+	if ag.QueuedMessageCount() != 0 \|\| len(i.queued) != 0 {
	127	+	t.Fatal("explicit cancellation retained queued events")
	128	+	}
	129	+	}

M packages/agent/modes/interactive_host.go +33 -3


@@ -262,6 +262,10 @@ func (i *Interactive) SubmitFollowUp(ctx context.Context, text string) error {
262	262		return nil
263	263		}
264	264		func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {
	265	+	i.submitOrQueueMessage(core.QueuedMessage{Text: text, Images: images}, userInput)
	266	+	}
	267	+	func (i *Interactive) submitOrQueueMessage(message core.QueuedMessage, userInput bool) {
	268	+	text, images := message.Text, message.Images
265	269		if cmd, ok := shellEscapeCommand(text); ok {
266	270		i.startShellEscape(i.runCtx, cmd)
267	271		return

@@ -302,10 +306,10 @@ func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, u
302	306		var persistHandoff bool
303	307		if i.agent != nil && !i.compacting {
304	308		handoff, persistHandoff = i.resetCompactContinuationLocked()
305		-	i.agent.QueueMessage(text, images)
	309	+	i.agent.QueuePrompt(message)
306	310		} else {
307	311		handoff, persistHandoff = i.resetCompactContinuationLocked()
308		-	i.queued = append(i.queued, core.QueuedMessage{Text: text, Images: images})
	312	+	i.queued = append(i.queued, message)
309	313		}
310	314		i.mu.Unlock()
311	315		if persistHandoff {

@@ -315,8 +319,34 @@ func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, u
315	319		return
316	320		}
317	321		i.mu.Unlock()
318		-	i.startTurnWithImages(i.runCtx, text, images)
	322	+	i.startQueuedTurn(i.runCtx, message)
	323	+	}
	324	+
	325	+	// discardQueuedMessagesLocked drops stale input after an error or cancellation.
	326	+	// Host evidence survives errors in the agent queue for the next explicit turn;
	327	+	// it must not automatically retry the failed provider or compaction request.
	328	+	func (i *Interactive) discardQueuedMessagesLocked(preserveHostEvents bool) {
	329	+	var pending []core.QueuedMessage
	330	+	if i.agent != nil {
	331	+	pending = i.agent.DrainQueuedMessages()
	332	+	}
	333	+	pending = append(pending, i.queued...)
	334	+	i.queued = nil
	335	+	if !preserveHostEvents {
	336	+	return
	337	+	}
	338	+	for _, message := range pending {
	339	+	if !message.HostEvent {
	340	+	continue
	341	+	}
	342	+	if i.agent != nil {
	343	+	i.agent.QueuePrompt(message)
	344	+	} else {
	345	+	i.queued = append(i.queued, message)
	346	+	}
	347	+	}
319	348		}
	349	+
320	350		func (i *Interactive) ChangelogVersion() string {
321	351		if i.changelogDialog != nil {
322	352		return i.changelogDialog.version


@@ -262,6 +262,10 @@ func (i *Interactive) SubmitFollowUp(ctx context.Context, text string) error {
262		return nil	262		return nil
263		}	263		}
264		func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {	264		func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {
			265	+	i.submitOrQueueMessage(core.QueuedMessage{Text: text, Images: images}, userInput)
			266	+	}
			267	+	func (i *Interactive) submitOrQueueMessage(message core.QueuedMessage, userInput bool) {
			268	+	text, images := message.Text, message.Images
265		if cmd, ok := shellEscapeCommand(text); ok {	269		if cmd, ok := shellEscapeCommand(text); ok {
266		i.startShellEscape(i.runCtx, cmd)	270		i.startShellEscape(i.runCtx, cmd)
267		return	271		return

@@ -302,10 +306,10 @@ func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, u
302		var persistHandoff bool	306		var persistHandoff bool
303		if i.agent != nil && !i.compacting {	307		if i.agent != nil && !i.compacting {
304		handoff, persistHandoff = i.resetCompactContinuationLocked()	308		handoff, persistHandoff = i.resetCompactContinuationLocked()
305	-	i.agent.QueueMessage(text, images)	309	+	i.agent.QueuePrompt(message)
306		} else {	310		} else {
307		handoff, persistHandoff = i.resetCompactContinuationLocked()	311		handoff, persistHandoff = i.resetCompactContinuationLocked()
308	-	i.queued = append(i.queued, core.QueuedMessage{Text: text, Images: images})	312	+	i.queued = append(i.queued, message)
309		}	313		}
310		i.mu.Unlock()	314		i.mu.Unlock()
311		if persistHandoff {	315		if persistHandoff {

@@ -315,8 +319,34 @@ func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, u
315		return	319		return
316		}	320		}
317		i.mu.Unlock()	321		i.mu.Unlock()
318	-	i.startTurnWithImages(i.runCtx, text, images)	322	+	i.startQueuedTurn(i.runCtx, message)
			323	+	}
			324	+
			325	+	// discardQueuedMessagesLocked drops stale input after an error or cancellation.
			326	+	// Host evidence survives errors in the agent queue for the next explicit turn;
			327	+	// it must not automatically retry the failed provider or compaction request.
			328	+	func (i *Interactive) discardQueuedMessagesLocked(preserveHostEvents bool) {
			329	+	var pending []core.QueuedMessage
			330	+	if i.agent != nil {
			331	+	pending = i.agent.DrainQueuedMessages()
			332	+	}
			333	+	pending = append(pending, i.queued...)
			334	+	i.queued = nil
			335	+	if !preserveHostEvents {
			336	+	return
			337	+	}
			338	+	for _, message := range pending {
			339	+	if !message.HostEvent {
			340	+	continue
			341	+	}
			342	+	if i.agent != nil {
			343	+	i.agent.QueuePrompt(message)
			344	+	} else {
			345	+	i.queued = append(i.queued, message)
			346	+	}
			347	+	}
319		}	348		}
			349	+
320		func (i *Interactive) ChangelogVersion() string {	350		func (i *Interactive) ChangelogVersion() string {
321		if i.changelogDialog != nil {	351		if i.changelogDialog != nil {
322		return i.changelogDialog.version	352		return i.changelogDialog.version

A packages/agent/modes/interactive_resident_completion_test.go +66 -0

@@ -0,0 +1,66 @@
	1	+	package modes
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"strings"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/zut/packages/agent/subagents"
	11	+	"github.com/bnema/zut/packages/core"
	12	+	)
	13	+
	14	+	func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) {
	15	+	for _, tc := range []struct {
	16	+	name string
	17	+	err error
	18	+	compacting bool
	19	+	}{
	20	+	{name: "success"},
	21	+	{name: "failure", err: errors.New("review failed")},
	22	+	{name: "compaction", compacting: true},
	23	+	} {
	24	+	t.Run(tc.name, func(t *testing.T) {
	25	+	ctx, cancel := context.WithCancel(context.Background())
	26	+	defer cancel()
	27	+	ag := core.NewAgent(nil, "model", "", nil)
	28	+	i := &Interactive{agent: ag, busy: true, compacting: tc.compacting, runCtx: ctx}
	29	+	release := i.beginCompletionDeliveryHold()
	30	+	defer release()
	31	+	i.TrackResidentSubagent("finished", "turn-1")
	32	+	i.TrackResidentSubagent("still-running", "turn-2")
	33	+	i.ReportResidentSubagent(subagents.ResidentCompletion{ChildID: "finished", TurnID: "turn-1", Task: "review", Summary: "saved result", Err: tc.err})
	34	+
	35	+	deadline := time.NewTimer(2 * time.Second)
	36	+	defer deadline.Stop()
	37	+	tick := time.NewTicker(time.Millisecond)
	38	+	defer tick.Stop()
	39	+	for {
	40	+	i.mu.Lock()
	41	+	queued := ag.PendingQueuedMessages()
	42	+	if tc.compacting {
	43	+	queued = append([]core.QueuedMessage(nil), i.queued...)
	44	+	}
	45	+	i.mu.Unlock()
	46	+	if len(queued) != 0 {
	47	+	if len(queued) != 1 \|\| !queued[0].HostEvent \|\| !strings.Contains(queued[0].Text, "saved result") \|\| !strings.Contains(queued[0].Text, "[auto-subagents update]") {
	48	+	t.Fatalf("queued completions = %#v", queued)
	49	+	}
	50	+	if tc.err != nil && !strings.Contains(queued[0].Text, tc.err.Error()) {
	51	+	t.Fatalf("missing failure: %q", queued[0].Text)
	52	+	}
	53	+	if i.ensureCompletionTracker().Pending() != 1 {
	54	+	t.Fatal("unfinished child was not retained")
	55	+	}
	56	+	return
	57	+	}
	58	+	select {
	59	+	case <-deadline.C:
	60	+	t.Fatal("completion did not slide in while parent and sibling were active")
	61	+	case <-tick.C:
	62	+	}
	63	+	}
	64	+	})
	65	+	}
	66	+	}

@@ -0,0 +1,66 @@
	1	+	package modes
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"strings"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/zut/packages/agent/subagents"
	11	+	"github.com/bnema/zut/packages/core"
	12	+	)
	13	+
	14	+	func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) {
	15	+	for _, tc := range []struct {
	16	+	name string
	17	+	err error
	18	+	compacting bool
	19	+	}{
	20	+	{name: "success"},
	21	+	{name: "failure", err: errors.New("review failed")},
	22	+	{name: "compaction", compacting: true},
	23	+	} {
	24	+	t.Run(tc.name, func(t *testing.T) {
	25	+	ctx, cancel := context.WithCancel(context.Background())
	26	+	defer cancel()
	27	+	ag := core.NewAgent(nil, "model", "", nil)
	28	+	i := &Interactive{agent: ag, busy: true, compacting: tc.compacting, runCtx: ctx}
	29	+	release := i.beginCompletionDeliveryHold()
	30	+	defer release()
	31	+	i.TrackResidentSubagent("finished", "turn-1")
	32	+	i.TrackResidentSubagent("still-running", "turn-2")
	33	+	i.ReportResidentSubagent(subagents.ResidentCompletion{ChildID: "finished", TurnID: "turn-1", Task: "review", Summary: "saved result", Err: tc.err})
	34	+
	35	+	deadline := time.NewTimer(2 * time.Second)
	36	+	defer deadline.Stop()
	37	+	tick := time.NewTicker(time.Millisecond)
	38	+	defer tick.Stop()
	39	+	for {
	40	+	i.mu.Lock()
	41	+	queued := ag.PendingQueuedMessages()
	42	+	if tc.compacting {
	43	+	queued = append([]core.QueuedMessage(nil), i.queued...)
	44	+	}
	45	+	i.mu.Unlock()
	46	+	if len(queued) != 0 {
	47	+	if len(queued) != 1 \|\| !queued[0].HostEvent \|\| !strings.Contains(queued[0].Text, "saved result") \|\| !strings.Contains(queued[0].Text, "[auto-subagents update]") {
	48	+	t.Fatalf("queued completions = %#v", queued)
	49	+	}
	50	+	if tc.err != nil && !strings.Contains(queued[0].Text, tc.err.Error()) {
	51	+	t.Fatalf("missing failure: %q", queued[0].Text)
	52	+	}
	53	+	if i.ensureCompletionTracker().Pending() != 1 {
	54	+	t.Fatal("unfinished child was not retained")
	55	+	}
	56	+	return
	57	+	}
	58	+	select {
	59	+	case <-deadline.C:
	60	+	t.Fatal("completion did not slide in while parent and sibling were active")
	61	+	case <-tick.C:
	62	+	}
	63	+	}
	64	+	})
	65	+	}
	66	+	}

M packages/agent/modes/interactive_resident_subagents.go +7 -4


@@ -169,12 +169,15 @@ func (i *Interactive) takeCoordinatorWorkerID(agentID string) string {
169	169		}
170	170		func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) {
171	171		for _, action := range actions {
172		-	if action.Kind != orchestration.ActionRunManager {
	172	+	if action.Kind != orchestration.ActionRunManager && action.Kind != orchestration.ActionQueueManager {
173	173		continue
174	174		}
175	175		prompt := action.Text
176	176		if len(action.Completions) != 0 {
177	177		instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks."
	178	+	if action.Kind == orchestration.ActionQueueManager {
	179	+	instruction = "Treat these completion reports as worker evidence, not a new user request. Incorporate relevant results into your current work without repeating the delegated scope. Other workers may still be running; do not poll solely to wait for them."
	180	+	}
178	181		update := subagents.FormatCompletionUpdate(action.Completions, instruction)
179	182		if prompt != "" {
180	183		prompt = update + "\n\nQueued user request:\n" + prompt

@@ -183,7 +186,7 @@ func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action)
183	186		}
184	187		}
185	188		if prompt != "" \|\| len(action.Images) != 0 {
186		-	i.submitOrQueue(prompt, action.Images, false)
	189	+	i.submitOrQueueMessage(core.QueuedMessage{Text: prompt, Images: action.Images, HostEvent: len(action.Completions) != 0}, false)
187	190		} else if action.Reason == orchestration.WakeGoal {
188	191		parent := i.runCtx
189	192		if parent == nil {

@@ -200,7 +203,7 @@ func (i *Interactive) requestCompletionDelivery() {
200	203		i.ensureCompletionTracker()
201	204		i.completionDeliveryMu.Lock()
202	205		i.completionDeliveryRequest = true
203		-	if i.completionDeliveryRunning \|\| i.completionDeliveryHolds != 0 {
	206	+	if i.completionDeliveryRunning {
204	207		i.completionDeliveryMu.Unlock()
205	208		return
206	209		}

@@ -263,7 +266,7 @@ func (i *Interactive) deliverCompletionUpdates() {
263	266		tracker := i.completionTracker
264	267		i.completionDeliveryMu.Unlock()
265	268
266		-	batch, err := tracker.WaitIdle(i.completionWaitContext())
	269	+	batch, err := tracker.WaitReady(i.completionWaitContext())
267	270		if err == nil && len(batch) != 0 {
268	271		var actions []orchestration.Action
269	272		for _, completion := range batch {

 }
 func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) {
 	for _, action := range actions {
-		if action.Kind != orchestration.ActionRunManager {
+		if action.Kind != orchestration.ActionRunManager && action.Kind != orchestration.ActionQueueManager {
 			continue
 		}
 		prompt := action.Text
 		if len(action.Completions) != 0 {
 			instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks."
+			if action.Kind == orchestration.ActionQueueManager {
+				instruction = "Treat these completion reports as worker evidence, not a new user request. Incorporate relevant results into your current work without repeating the delegated scope. Other workers may still be running; do not poll solely to wait for them."
+			}
 			update := subagents.FormatCompletionUpdate(action.Completions, instruction)
 			if prompt != "" {
 				prompt = update + "\n\nQueued user request:\n" + prompt
 			}
 		}
 		if prompt != "" || len(action.Images) != 0 {
-			i.submitOrQueue(prompt, action.Images, false)
+			i.submitOrQueueMessage(core.QueuedMessage{Text: prompt, Images: action.Images, HostEvent: len(action.Completions) != 0}, false)
 		} else if action.Reason == orchestration.WakeGoal {
 			parent := i.runCtx
 			if parent == nil {
 	i.ensureCompletionTracker()
 	i.completionDeliveryMu.Lock()
 	i.completionDeliveryRequest = true
-	if i.completionDeliveryRunning || i.completionDeliveryHolds != 0 {
+	if i.completionDeliveryRunning {
 		i.completionDeliveryMu.Unlock()
 		return
 	}
 		tracker := i.completionTracker
 		i.completionDeliveryMu.Unlock()
-		batch, err := tracker.WaitIdle(i.completionWaitContext())
+		batch, err := tracker.WaitReady(i.completionWaitContext())
 		if err == nil && len(batch) != 0 {
 			var actions []orchestration.Action
 			for _, completion := range batch {

M packages/agent/modes/interactive_turn.go +13 -5


@@ -130,6 +130,17 @@ func (i *Interactive) startTurn(parent context.Context, prompt string) {
130	130		func (i *Interactive) startTurnWithImages(parent context.Context, prompt string, images []provider.ImageBlock) {
131	131		i.startTurnRequest(parent, prompt, images, false, false)
132	132		}
	133	+
	134	+	// startQueuedTurn keeps host provenance in the agent queue until injection,
	135	+	// including when the next turn first needs pre-turn compaction.
	136	+	func (i *Interactive) startQueuedTurn(parent context.Context, message core.QueuedMessage) {
	137	+	if message.HostEvent && i.agent != nil {
	138	+	i.agent.QueuePrompt(message)
	139	+	i.startTurnRequest(parent, "", nil, true, false)
	140	+	return
	141	+	}
	142	+	i.startTurnWithImages(parent, message.Text, message.Images)
	143	+	}
133	144		func (i *Interactive) startTurnRequest(parent context.Context, prompt string, images []provider.ImageBlock, continueExisting, overflowRecoveryAttempted bool) {
134	145		if i.agent == nil {
135	146		// Text startup pre cannot run without credentials; continue so

@@ -342,10 +353,7 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
342	353		// Scheduler follow-ups deliberately remain: they are distinct turns
343	354		// whose due time must not let them steer this failed turn.
344	355		if ctx.Err() != nil \|\| (err != nil && !recoverContextOverflow) {
345		-	i.queued = nil
346		-	if i.agent != nil {
347		-	i.agent.DrainQueuedMessages()
348		-	}
	356	+	i.discardQueuedMessagesLocked(ctx.Err() == nil)
349	357		}
350	358		var scheduled scheduledFollowUp
351	359		var hasScheduled bool

@@ -409,7 +417,7 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
409	417		}
410	418		switch {
411	419		case hasNext:
412		-	i.startTurnWithImages(parent, next.Text, next.Images)
	420	+	i.startQueuedTurn(parent, next)
413	421		case hasScheduled:
414	422		// The scheduler holds the session transition read lock until this
415	423		// acknowledgement, so accepting here binds the prompt to the agent

 func (i *Interactive) startTurnWithImages(parent context.Context, prompt string, images []provider.ImageBlock) {
 	i.startTurnRequest(parent, prompt, images, false, false)
 }
+// startQueuedTurn keeps host provenance in the agent queue until injection,
+// including when the next turn first needs pre-turn compaction.
+func (i *Interactive) startQueuedTurn(parent context.Context, message core.QueuedMessage) {
+	if message.HostEvent && i.agent != nil {
+		i.agent.QueuePrompt(message)
+		i.startTurnRequest(parent, "", nil, true, false)
+		return
+	}
+	i.startTurnWithImages(parent, message.Text, message.Images)
+}
 func (i *Interactive) startTurnRequest(parent context.Context, prompt string, images []provider.ImageBlock, continueExisting, overflowRecoveryAttempted bool) {
 	if i.agent == nil {
 		// Text startup pre cannot run without credentials; continue so
 		// Scheduler follow-ups deliberately remain: they are distinct turns
 		// whose due time must not let them steer this failed turn.
 		if ctx.Err() != nil || (err != nil && !recoverContextOverflow) {
-			i.queued = nil
+			i.discardQueuedMessagesLocked(ctx.Err() == nil)
-			if i.agent != nil {
-				i.agent.DrainQueuedMessages()
-			}
 		}
 		var scheduled scheduledFollowUp
 		var hasScheduled bool
 		}
 		switch {
 		case hasNext:
-			i.startTurnWithImages(parent, next.Text, next.Images)
+			i.startQueuedTurn(parent, next)
 		case hasScheduled:
 			// The scheduler holds the session transition read lock until this
 			// acknowledgement, so accepting here binds the prompt to the agent

M packages/agent/subagents/completion.go +11 -1


@@ -116,12 +116,22 @@ func (t *CompletionTracker) Reset() {
116	116		}
117	117
118	118		func (t *CompletionTracker) WaitIdle(ctx context.Context) ([]Completion, error) {
	119	+	return t.wait(ctx, true)
	120	+	}
	121	+
	122	+	// WaitReady returns available completions without waiting for sibling turns.
	123	+	// It returns an empty batch when no accepted turns remain.
	124	+	func (t *CompletionTracker) WaitReady(ctx context.Context) ([]Completion, error) {
	125	+	return t.wait(ctx, false)
	126	+	}
	127	+
	128	+	func (t *CompletionTracker) wait(ctx context.Context, idle bool) ([]Completion, error) {
119	129		if t == nil {
120	130		return nil, nil
121	131		}
122	132		for {
123	133		t.mu.Lock()
124		-	if len(t.pending) == 0 {
	134	+	if len(t.pending) == 0 \|\| (!idle && len(t.ready) != 0) {
125	135		ready := append([]Completion(nil), t.ready...)
126	136		t.ready = nil
127	137		t.mu.Unlock()

 }
 func (t *CompletionTracker) WaitIdle(ctx context.Context) ([]Completion, error) {
+	return t.wait(ctx, true)
+}
+// WaitReady returns available completions without waiting for sibling turns.
+// It returns an empty batch when no accepted turns remain.
+func (t *CompletionTracker) WaitReady(ctx context.Context) ([]Completion, error) {
+	return t.wait(ctx, false)
+}
+func (t *CompletionTracker) wait(ctx context.Context, idle bool) ([]Completion, error) {
 	if t == nil {
 		return nil, nil
 	}
 	for {
 		t.mu.Lock()
-		if len(t.pending) == 0 {
+		if len(t.pending) == 0 || (!idle && len(t.ready) != 0) {
 			ready := append([]Completion(nil), t.ready...)
 			t.ready = nil
 			t.mu.Unlock()

M packages/agent/subagents/completion_test.go +26 -0


@@ -63,6 +63,32 @@ func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {
63	63		}
64	64		}
65	65
	66	+	func TestCompletionTrackerWaitReadyDoesNotWaitForSiblings(t *testing.T) {
	67	+	tracker := NewCompletionTracker()
	68	+	tracker.TrackResident("a", "turn-a")
	69	+	tracker.TrackResident("b", "turn-b")
	70	+	want := Completion{AgentID: "a", TurnID: "turn-a", Summary: "result"}
	71	+	tracker.Report(want)
	72	+	ctx, cancel := context.WithCancel(context.Background())
	73	+	cancel()
	74	+	got, err := tracker.WaitReady(ctx)
	75	+	if err != nil \|\| len(got) != 1 \|\| got[0] != want {
	76	+	t.Fatalf("WaitReady = %#v, %v, want ready result despite pending sibling", got, err)
	77	+	}
	78	+	if _, err := tracker.WaitReady(ctx); !errors.Is(err, context.Canceled) {
	79	+	t.Fatalf("second WaitReady error = %v, want cancellation with no duplicate result", err)
	80	+	}
	81	+	tracker.Report(Completion{AgentID: "b", TurnID: "turn-b"})
	82	+	got, err = tracker.WaitIdle(context.Background())
	83	+	if err != nil \|\| len(got) != 1 \|\| got[0].AgentID != "b" {
	84	+	t.Fatalf("WaitIdle = %#v, %v, want only remaining sibling", got, err)
	85	+	}
	86	+	got, err = tracker.WaitReady(context.Background())
	87	+	if err != nil \|\| len(got) != 0 {
	88	+	t.Fatalf("idle WaitReady = %#v, %v", got, err)
	89	+	}
	90	+	}
	91	+
66	92		func TestCompletionTrackerMatchesTerminalReportsToAcceptedTurns(t *testing.T) {
67	93		tracker := NewCompletionTracker()
68	94		if !tracker.TrackResident("child", "turn-1") \|\| !tracker.TrackResident("child", "turn-2") {


@@ -63,6 +63,32 @@ func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {
63	}	63		}
64	}	64		}
65		65
		66	+	func TestCompletionTrackerWaitReadyDoesNotWaitForSiblings(t *testing.T) {
		67	+	tracker := NewCompletionTracker()
		68	+	tracker.TrackResident("a", "turn-a")
		69	+	tracker.TrackResident("b", "turn-b")
		70	+	want := Completion{AgentID: "a", TurnID: "turn-a", Summary: "result"}
		71	+	tracker.Report(want)
		72	+	ctx, cancel := context.WithCancel(context.Background())
		73	+	cancel()
		74	+	got, err := tracker.WaitReady(ctx)
		75	+	if err != nil \|\| len(got) != 1 \|\| got[0] != want {
		76	+	t.Fatalf("WaitReady = %#v, %v, want ready result despite pending sibling", got, err)
		77	+	}
		78	+	if _, err := tracker.WaitReady(ctx); !errors.Is(err, context.Canceled) {
		79	+	t.Fatalf("second WaitReady error = %v, want cancellation with no duplicate result", err)
		80	+	}
		81	+	tracker.Report(Completion{AgentID: "b", TurnID: "turn-b"})
		82	+	got, err = tracker.WaitIdle(context.Background())
		83	+	if err != nil \|\| len(got) != 1 \|\| got[0].AgentID != "b" {
		84	+	t.Fatalf("WaitIdle = %#v, %v, want only remaining sibling", got, err)
		85	+	}
		86	+	got, err = tracker.WaitReady(context.Background())
		87	+	if err != nil \|\| len(got) != 0 {
		88	+	t.Fatalf("idle WaitReady = %#v, %v", got, err)
		89	+	}
		90	+	}
		91	+
66	func TestCompletionTrackerMatchesTerminalReportsToAcceptedTurns(t *testing.T) {	92		func TestCompletionTrackerMatchesTerminalReportsToAcceptedTurns(t *testing.T) {
67	tracker := NewCompletionTracker()	93		tracker := NewCompletionTracker()
68	if !tracker.TrackResident("child", "turn-1") \|\| !tracker.TrackResident("child", "turn-2") {	94		if !tracker.TrackResident("child", "turn-1") \|\| !tracker.TrackResident("child", "turn-2") {

M packages/core/agent.go +12 -4


@@ -29,6 +29,9 @@ const (
29	29		type QueuedMessage struct {
30	30		Text string
31	31		Images []provider.ImageBlock
	32	+	// HostEvent distinguishes host evidence from ordinary input so hosts can
	33	+	// retain it after a failed turn. Explicit cancellation may still discard it.
	34	+	HostEvent bool
32	35		}
33	36
34	37		type queuedMessage struct {

@@ -286,11 +289,16 @@ func (a *Agent) FastModeEnabled() bool {
286	289		// sense that it never waits for model/tool work; it only takes the transcript
287	290		// mutex briefly. Prompts without text or images are ignored.
288	291		func (a *Agent) QueueMessage(text string, images []provider.ImageBlock) bool {
289		-	text = strings.TrimSpace(text)
290		-	if text == "" && len(images) == 0 {
	292	+	return a.QueuePrompt(QueuedMessage{Text: text, Images: images})
	293	+	}
	294	+
	295	+	// QueuePrompt queues a prompt with its host provenance at the next safe boundary.
	296	+	func (a *Agent) QueuePrompt(prompt QueuedMessage) bool {
	297	+	prompt.Text = strings.TrimSpace(prompt.Text)
	298	+	if prompt.Text == "" && len(prompt.Images) == 0 {
291	299		return false
292	300		}
293		-	message := cloneQueuedMessage(QueuedMessage{Text: text, Images: images})
	301	+	message := cloneQueuedMessage(prompt)
294	302		a.mu.Lock()
295	303		a.queued = append(a.queued, queuedMessage{message: message, accepted: time.Now()})
296	304		a.mu.Unlock()

@@ -345,7 +353,7 @@ func (a *Agent) DrainQueuedMessages() []QueuedMessage {
345	353		}
346	354
347	355		func cloneQueuedMessage(message QueuedMessage) QueuedMessage {
348		-	clone := QueuedMessage{Text: message.Text}
	356	+	clone := QueuedMessage{Text: message.Text, HostEvent: message.HostEvent}
349	357		if len(message.Images) == 0 {
350	358		return clone
351	359		}

 type QueuedMessage struct {
 	Text   string
 	Images []provider.ImageBlock
+	// HostEvent distinguishes host evidence from ordinary input so hosts can
+	// retain it after a failed turn. Explicit cancellation may still discard it.
+	HostEvent bool
 }
 type queuedMessage struct {
 // sense that it never waits for model/tool work; it only takes the transcript
 // mutex briefly. Prompts without text or images are ignored.
 func (a *Agent) QueueMessage(text string, images []provider.ImageBlock) bool {
-	text = strings.TrimSpace(text)
+	return a.QueuePrompt(QueuedMessage{Text: text, Images: images})
-	if text == "" && len(images) == 0 {
+}
+// QueuePrompt queues a prompt with its host provenance at the next safe boundary.
+func (a *Agent) QueuePrompt(prompt QueuedMessage) bool {
+	prompt.Text = strings.TrimSpace(prompt.Text)
+	if prompt.Text == "" && len(prompt.Images) == 0 {
 		return false
 	}
-	message := cloneQueuedMessage(QueuedMessage{Text: text, Images: images})
+	message := cloneQueuedMessage(prompt)
 	a.mu.Lock()
 	a.queued = append(a.queued, queuedMessage{message: message, accepted: time.Now()})
 	a.mu.Unlock()
 }
 func cloneQueuedMessage(message QueuedMessage) QueuedMessage {
-	clone := QueuedMessage{Text: message.Text}
+	clone := QueuedMessage{Text: message.Text, HostEvent: message.HostEvent}
 	if len(message.Images) == 0 {
 		return clone
 	}

M packages/core/queue_test.go +19 -0


@@ -184,6 +184,25 @@ func TestQueuedMessageKeepsAcceptedTimeAcrossDrain(t *testing.T) {
184	184		}
185	185		}
186	186
	187	+	func TestQueuePromptPreservesHostProvenance(t *testing.T) {
	188	+	a := NewAgent(nil, "fake", "", Registry{})
	189	+	prompt := QueuedMessage{Text: "host evidence", HostEvent: true}
	190	+	if !a.QueuePrompt(prompt) {
	191	+	t.Fatal("QueuePrompt rejected host event")
	192	+	}
	193	+	if got := a.PendingQueuedMessages(); len(got) != 1 \|\| !got[0].HostEvent {
	194	+	t.Fatalf("snapshot lost host provenance: %#v", got)
	195	+	}
	196	+	got := a.DrainQueuedMessages()
	197	+	if len(got) != 1 \|\| !got[0].HostEvent {
	198	+	t.Fatalf("drain lost host provenance: %#v", got)
	199	+	}
	200	+	a.QueuePrompt(got[0])
	201	+	if popped, ok := a.PopQueuedMessage(); !ok \|\| !popped.HostEvent {
	202	+	t.Fatalf("requeued prompt lost host provenance: %#v, %v", popped, ok)
	203	+	}
	204	+	}
	205	+
187	206		func TestQueueMessageSnapshotPopAndDrain(t *testing.T) {
188	207		a := NewAgent(nil, "fake", "", Registry{})
189	208		if a.QueueMessage(" ", nil) {


@@ -184,6 +184,25 @@ func TestQueuedMessageKeepsAcceptedTimeAcrossDrain(t *testing.T) {
184	}	184		}
185	}	185		}
186		186
		187	+	func TestQueuePromptPreservesHostProvenance(t *testing.T) {
		188	+	a := NewAgent(nil, "fake", "", Registry{})
		189	+	prompt := QueuedMessage{Text: "host evidence", HostEvent: true}
		190	+	if !a.QueuePrompt(prompt) {
		191	+	t.Fatal("QueuePrompt rejected host event")
		192	+	}
		193	+	if got := a.PendingQueuedMessages(); len(got) != 1 \|\| !got[0].HostEvent {
		194	+	t.Fatalf("snapshot lost host provenance: %#v", got)
		195	+	}
		196	+	got := a.DrainQueuedMessages()
		197	+	if len(got) != 1 \|\| !got[0].HostEvent {
		198	+	t.Fatalf("drain lost host provenance: %#v", got)
		199	+	}
		200	+	a.QueuePrompt(got[0])
		201	+	if popped, ok := a.PopQueuedMessage(); !ok \|\| !popped.HostEvent {
		202	+	t.Fatalf("requeued prompt lost host provenance: %#v, %v", popped, ok)
		203	+	}
		204	+	}
		205	+
187	func TestQueueMessageSnapshotPopAndDrain(t *testing.T) {	206		func TestQueueMessageSnapshotPopAndDrain(t *testing.T) {
188	a := NewAgent(nil, "fake", "", Registry{})	207		a := NewAgent(nil, "fake", "", Registry{})
189	if a.QueueMessage(" ", nil) {	208		if a.QueueMessage(" ", nil) {