Merge pull request #112 from bnema/feat/event-driven-autonomous-goals

feat(agent): coordinate delegated worker waves

commit: c63670ee39040ef3353b46c7d2c826224286b26b
author: bnema · 2mo ago · 2026-08-11 05:52 UTC
committer: GitHub · 2mo ago · 2026-08-11 05:52 UTC
merge of: 75d4151 + e9c105f · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

11 files · +525 -36

M docs/subagents.md +1 -1


@@ -143,7 +143,7 @@ During an active delegated turn, interactive mode shows compact rows beside the
143	143
144	144		The worker and supervisor communicate over newline-delimited JSON. Current messages use version `1` envelopes with `version`, `message_id`, `agent_id`, `turn_id`, `timestamp`, and `payload`. Unknown event names and payload fields are retained. Only versioned JSONL envelopes are accepted on the worker protocol.
145	145
146		-	A completed worker emits a `turn.result` event and writes `result.json`. Intentional cancellation records `shutdown_origin` as `targeted`, `session`, `deadline`, or `process`; manager updates describe that cancellation instead of reporting the worker as generically failed with `context canceled`. Every manager call returns immediately, and every worker outcome is delivered asynchronously through `[auto-subagents update]`; required outcome state is also injected as high-priority `[required-subagents update]` context. The primary remains available for coordination and independent work while children run. These host-delivered outcomes, rather than polling or inspecting state, are the signal for the primary to continue. Inline output is bounded; after delivery, inspect the full session through the stable references:
	146	+	A completed worker emits a `turn.result` event and writes `result.json`. Intentional cancellation records `shutdown_origin` as `targeted`, `session`, `deadline`, or `process`; manager updates describe that cancellation instead of reporting the worker as generically failed with `context canceled`. Optional manager calls return immediately. Required manager calls remain open until their required outcome is available in the current parent turn. When one manager turn starts multiple workers, their terminal outcomes are delivered as one ordered update after that worker wave is sealed and all registered turns are terminal; an early completion never wakes the manager by itself. Required outcome state is also injected as high-priority `[required-subagents update]` context. The primary remains available for coordination and independent work while optional children run. These host-delivered outcomes, rather than polling or inspecting state, are the signal for the primary to continue. Inline output is bounded; after delivery, inspect the full session through the stable references:
147	147
148	148		If a provider rejects a child request because its payload or context window is too large, the worker compacts its persisted transcript and continues the same request once. If compaction or that retry cannot fit, the result uses error code `context_limit` and directs the caller to narrow the task or reduce gathered context; it does not include the provider's raw request error.
149	149


@@ -143,7 +143,7 @@ During an active delegated turn, interactive mode shows compact rows beside the
143			143
144		The worker and supervisor communicate over newline-delimited JSON. Current messages use version `1` envelopes with `version`, `message_id`, `agent_id`, `turn_id`, `timestamp`, and `payload`. Unknown event names and payload fields are retained. Only versioned JSONL envelopes are accepted on the worker protocol.	144		The worker and supervisor communicate over newline-delimited JSON. Current messages use version `1` envelopes with `version`, `message_id`, `agent_id`, `turn_id`, `timestamp`, and `payload`. Unknown event names and payload fields are retained. Only versioned JSONL envelopes are accepted on the worker protocol.
145			145
146	-	A completed worker emits a `turn.result` event and writes `result.json`. Intentional cancellation records `shutdown_origin` as `targeted`, `session`, `deadline`, or `process`; manager updates describe that cancellation instead of reporting the worker as generically failed with `context canceled`. Every manager call returns immediately, and every worker outcome is delivered asynchronously through `[auto-subagents update]`; required outcome state is also injected as high-priority `[required-subagents update]` context. The primary remains available for coordination and independent work while children run. These host-delivered outcomes, rather than polling or inspecting state, are the signal for the primary to continue. Inline output is bounded; after delivery, inspect the full session through the stable references:	146	+	A completed worker emits a `turn.result` event and writes `result.json`. Intentional cancellation records `shutdown_origin` as `targeted`, `session`, `deadline`, or `process`; manager updates describe that cancellation instead of reporting the worker as generically failed with `context canceled`. Optional manager calls return immediately. Required manager calls remain open until their required outcome is available in the current parent turn. When one manager turn starts multiple workers, their terminal outcomes are delivered as one ordered update after that worker wave is sealed and all registered turns are terminal; an early completion never wakes the manager by itself. Required outcome state is also injected as high-priority `[required-subagents update]` context. The primary remains available for coordination and independent work while optional children run. These host-delivered outcomes, rather than polling or inspecting state, are the signal for the primary to continue. Inline output is bounded; after delivery, inspect the full session through the stable references:
147			147
148		If a provider rejects a child request because its payload or context window is too large, the worker compacts its persisted transcript and continues the same request once. If compaction or that retry cannot fit, the result uses error code `context_limit` and directs the caller to narrow the task or reduce gathered context; it does not include the provider's raw request error.	148		If a provider rejects a child request because its payload or context window is too large, the worker compacts its persisted transcript and continues the same request once. If compaction or that retry cannot fit, the result uses error code `context_limit` and directs the caller to narrow the task or reduce gathered context; it does not include the provider's raw request error.
149			149

A packages/agent/internal/orchestration/coordinator.go +180 -0

@@ -0,0 +1,180 @@
	1	+	// Package orchestration owns the deterministic policy between manager turns
	2	+	// and delegated worker lifecycles. It contains no goroutines, timers, or
	3	+	// provider work; presentation modes translate asynchronous callbacks into
	4	+	// events and execute the returned actions.
	5	+	package orchestration
	6	+
	7	+	import "github.com/bnema/zut/packages/agent/subagents"
	8	+
	9	+	// Coordinator collects one sealed worker wave per manager turn. A terminal
	10	+	// worker cannot wake the manager until the owning turn is sealed. Queued user
	11	+	// input wins the next wake; an active goal is used only when no user input or
	12	+	// worker result remains.
	13	+	type Coordinator struct {
	14	+	managerActive bool
	15	+	waveSealed bool
	16	+	stopped bool
	17	+	goalActive bool
	18	+	workers map[string]*worker
	19	+	workerOrder []string
	20	+	queuedUser []string
	21	+	}
	22	+
	23	+	type worker struct {
	24	+	terminal bool
	25	+	completion subagents.Completion
	26	+	}
	27	+
	28	+	type EventKind uint8
	29	+
	30	+	const (
	31	+	EventManagerStarted EventKind = iota + 1
	32	+	EventManagerFinished
	33	+	EventWorkerRegistered
	34	+	EventWorkerFinished
	35	+	EventUserInput
	36	+	EventGoalChanged
	37	+	EventCancelled
	38	+	EventShutdown
	39	+	)
	40	+
	41	+	type Event struct {
	42	+	Kind EventKind
	43	+	WorkerID string
	44	+	Completion subagents.Completion
	45	+	Text string
	46	+	GoalActive bool
	47	+	}
	48	+
	49	+	type ActionKind uint8
	50	+
	51	+	const (
	52	+	ActionRunManager ActionKind = iota + 1
	53	+	ActionWait
	54	+	ActionStop
	55	+	)
	56	+
	57	+	type WakeReason uint8
	58	+
	59	+	const (
	60	+	WakeUser WakeReason = iota + 1
	61	+	WakeWorkers
	62	+	WakeGoal
	63	+	)
	64	+
	65	+	type Action struct {
	66	+	Kind ActionKind
	67	+	Reason WakeReason
	68	+	Text string
	69	+	Completions []subagents.Completion
	70	+	}
	71	+
	72	+	type Result struct{ Actions []Action }
	73	+
	74	+	func New() *Coordinator {
	75	+	return &Coordinator{workers: make(map[string]*worker)}
	76	+	}
	77	+
	78	+	// AcceptsUserInput reports whether a sealed worker wave is awaiting its next
	79	+	// manager turn. Ordinary interactive turns continue to use their existing
	80	+	// queueing path; this only gives user input priority over a worker wake.
	81	+	func (c *Coordinator) AcceptsUserInput() bool {
	82	+	return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
	83	+	}
	84	+
	85	+	func (c *Coordinator) Apply(event Event) Result {
	86	+	if c == nil \|\| c.stopped {
	87	+	return Result{}
	88	+	}
	89	+	switch event.Kind {
	90	+	case EventManagerStarted:
	91	+	c.managerActive = true
	92	+	c.waveSealed = false
	93	+	c.workers = make(map[string]*worker)
	94	+	c.workerOrder = nil
	95	+	case EventManagerFinished:
	96	+	c.managerActive = false
	97	+	c.waveSealed = true
	98	+	if c.hasPendingWorkers() {
	99	+	return Result{Actions: []Action{{Kind: ActionWait}}}
	100	+	}
	101	+	return c.wakeIfIdle()
	102	+	case EventWorkerRegistered:
	103	+	if !c.managerActive \|\| c.waveSealed \|\| event.WorkerID == "" {
	104	+	return Result{}
	105	+	}
	106	+	if _, exists := c.workers[event.WorkerID]; !exists {
	107	+	c.workers[event.WorkerID] = &worker{}
	108	+	c.workerOrder = append(c.workerOrder, event.WorkerID)
	109	+	}
	110	+	case EventWorkerFinished:
	111	+	w, exists := c.workers[event.WorkerID]
	112	+	if !exists \|\| w.terminal {
	113	+	return Result{}
	114	+	}
	115	+	w.terminal = true
	116	+	w.completion = event.Completion
	117	+	if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() {
	118	+	return c.wakeIfIdle()
	119	+	}
	120	+	case EventUserInput:
	121	+	if event.Text == "" {
	122	+	return Result{}
	123	+	}
	124	+	c.queuedUser = append(c.queuedUser, event.Text)
	125	+	return c.wakeIfIdle()
	126	+	case EventGoalChanged:
	127	+	c.goalActive = event.GoalActive
	128	+	case EventCancelled, EventShutdown:
	129	+	c.stopped = true
	130	+	c.queuedUser = nil
	131	+	return Result{Actions: []Action{{Kind: ActionStop}}}
	132	+	}
	133	+	return Result{}
	134	+	}
	135	+
	136	+	func (c *Coordinator) wakeIfIdle() Result {
	137	+	if c.managerActive \|\| c.stopped \|\| c.hasPendingWorkers() {
	138	+	return Result{}
	139	+	}
	140	+	completions := c.completedWorkers()
	141	+	if len(c.queuedUser) != 0 {
	142	+	text := c.queuedUser[0]
	143	+	c.queuedUser = c.queuedUser[1:]
	144	+	return c.startManager(WakeUser, text, completions)
	145	+	}
	146	+	if len(completions) != 0 {
	147	+	return c.startManager(WakeWorkers, "", completions)
	148	+	}
	149	+	if c.goalActive {
	150	+	return c.startManager(WakeGoal, "", nil)
	151	+	}
	152	+	return Result{}
	153	+	}
	154	+
	155	+	func (c *Coordinator) startManager(reason WakeReason, text string, completions []subagents.Completion) Result {
	156	+	c.managerActive = true
	157	+	return Result{Actions: []Action{{Kind: ActionRunManager, Reason: reason, Text: text, Completions: completions}}}
	158	+	}
	159	+
	160	+	func (c *Coordinator) hasPendingWorkers() bool {
	161	+	for _, worker := range c.workers {
	162	+	if !worker.terminal {
	163	+	return true
	164	+	}
	165	+	}
	166	+	return false
	167	+	}
	168	+
	169	+	func (c *Coordinator) completedWorkers() []subagents.Completion {
	170	+	if c.hasPendingWorkers() {
	171	+	return nil
	172	+	}
	173	+	completions := make([]subagents.Completion, 0, len(c.workerOrder))
	174	+	for _, workerID := range c.workerOrder {
	175	+	if worker := c.workers[workerID]; worker != nil && worker.terminal {
	176	+	completions = append(completions, worker.completion)
	177	+	}
	178	+	}
	179	+	return completions
	180	+	}

@@ -0,0 +1,180 @@
	1	+	// Package orchestration owns the deterministic policy between manager turns
	2	+	// and delegated worker lifecycles. It contains no goroutines, timers, or
	3	+	// provider work; presentation modes translate asynchronous callbacks into
	4	+	// events and execute the returned actions.
	5	+	package orchestration
	6	+
	7	+	import "github.com/bnema/zut/packages/agent/subagents"
	8	+
	9	+	// Coordinator collects one sealed worker wave per manager turn. A terminal
	10	+	// worker cannot wake the manager until the owning turn is sealed. Queued user
	11	+	// input wins the next wake; an active goal is used only when no user input or
	12	+	// worker result remains.
	13	+	type Coordinator struct {
	14	+	managerActive bool
	15	+	waveSealed bool
	16	+	stopped bool
	17	+	goalActive bool
	18	+	workers map[string]*worker
	19	+	workerOrder []string
	20	+	queuedUser []string
	21	+	}
	22	+
	23	+	type worker struct {
	24	+	terminal bool
	25	+	completion subagents.Completion
	26	+	}
	27	+
	28	+	type EventKind uint8
	29	+
	30	+	const (
	31	+	EventManagerStarted EventKind = iota + 1
	32	+	EventManagerFinished
	33	+	EventWorkerRegistered
	34	+	EventWorkerFinished
	35	+	EventUserInput
	36	+	EventGoalChanged
	37	+	EventCancelled
	38	+	EventShutdown
	39	+	)
	40	+
	41	+	type Event struct {
	42	+	Kind EventKind
	43	+	WorkerID string
	44	+	Completion subagents.Completion
	45	+	Text string
	46	+	GoalActive bool
	47	+	}
	48	+
	49	+	type ActionKind uint8
	50	+
	51	+	const (
	52	+	ActionRunManager ActionKind = iota + 1
	53	+	ActionWait
	54	+	ActionStop
	55	+	)
	56	+
	57	+	type WakeReason uint8
	58	+
	59	+	const (
	60	+	WakeUser WakeReason = iota + 1
	61	+	WakeWorkers
	62	+	WakeGoal
	63	+	)
	64	+
	65	+	type Action struct {
	66	+	Kind ActionKind
	67	+	Reason WakeReason
	68	+	Text string
	69	+	Completions []subagents.Completion
	70	+	}
	71	+
	72	+	type Result struct{ Actions []Action }
	73	+
	74	+	func New() *Coordinator {
	75	+	return &Coordinator{workers: make(map[string]*worker)}
	76	+	}
	77	+
	78	+	// AcceptsUserInput reports whether a sealed worker wave is awaiting its next
	79	+	// manager turn. Ordinary interactive turns continue to use their existing
	80	+	// queueing path; this only gives user input priority over a worker wake.
	81	+	func (c *Coordinator) AcceptsUserInput() bool {
	82	+	return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
	83	+	}
	84	+
	85	+	func (c *Coordinator) Apply(event Event) Result {
	86	+	if c == nil \|\| c.stopped {
	87	+	return Result{}
	88	+	}
	89	+	switch event.Kind {
	90	+	case EventManagerStarted:
	91	+	c.managerActive = true
	92	+	c.waveSealed = false
	93	+	c.workers = make(map[string]*worker)
	94	+	c.workerOrder = nil
	95	+	case EventManagerFinished:
	96	+	c.managerActive = false
	97	+	c.waveSealed = true
	98	+	if c.hasPendingWorkers() {
	99	+	return Result{Actions: []Action{{Kind: ActionWait}}}
	100	+	}
	101	+	return c.wakeIfIdle()
	102	+	case EventWorkerRegistered:
	103	+	if !c.managerActive \|\| c.waveSealed \|\| event.WorkerID == "" {
	104	+	return Result{}
	105	+	}
	106	+	if _, exists := c.workers[event.WorkerID]; !exists {
	107	+	c.workers[event.WorkerID] = &worker{}
	108	+	c.workerOrder = append(c.workerOrder, event.WorkerID)
	109	+	}
	110	+	case EventWorkerFinished:
	111	+	w, exists := c.workers[event.WorkerID]
	112	+	if !exists \|\| w.terminal {
	113	+	return Result{}
	114	+	}
	115	+	w.terminal = true
	116	+	w.completion = event.Completion
	117	+	if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() {
	118	+	return c.wakeIfIdle()
	119	+	}
	120	+	case EventUserInput:
	121	+	if event.Text == "" {
	122	+	return Result{}
	123	+	}
	124	+	c.queuedUser = append(c.queuedUser, event.Text)
	125	+	return c.wakeIfIdle()
	126	+	case EventGoalChanged:
	127	+	c.goalActive = event.GoalActive
	128	+	case EventCancelled, EventShutdown:
	129	+	c.stopped = true
	130	+	c.queuedUser = nil
	131	+	return Result{Actions: []Action{{Kind: ActionStop}}}
	132	+	}
	133	+	return Result{}
	134	+	}
	135	+
	136	+	func (c *Coordinator) wakeIfIdle() Result {
	137	+	if c.managerActive \|\| c.stopped \|\| c.hasPendingWorkers() {
	138	+	return Result{}
	139	+	}
	140	+	completions := c.completedWorkers()
	141	+	if len(c.queuedUser) != 0 {
	142	+	text := c.queuedUser[0]
	143	+	c.queuedUser = c.queuedUser[1:]
	144	+	return c.startManager(WakeUser, text, completions)
	145	+	}
	146	+	if len(completions) != 0 {
	147	+	return c.startManager(WakeWorkers, "", completions)
	148	+	}
	149	+	if c.goalActive {
	150	+	return c.startManager(WakeGoal, "", nil)
	151	+	}
	152	+	return Result{}
	153	+	}
	154	+
	155	+	func (c *Coordinator) startManager(reason WakeReason, text string, completions []subagents.Completion) Result {
	156	+	c.managerActive = true
	157	+	return Result{Actions: []Action{{Kind: ActionRunManager, Reason: reason, Text: text, Completions: completions}}}
	158	+	}
	159	+
	160	+	func (c *Coordinator) hasPendingWorkers() bool {
	161	+	for _, worker := range c.workers {
	162	+	if !worker.terminal {
	163	+	return true
	164	+	}
	165	+	}
	166	+	return false
	167	+	}
	168	+
	169	+	func (c *Coordinator) completedWorkers() []subagents.Completion {
	170	+	if c.hasPendingWorkers() {
	171	+	return nil
	172	+	}
	173	+	completions := make([]subagents.Completion, 0, len(c.workerOrder))
	174	+	for _, workerID := range c.workerOrder {
	175	+	if worker := c.workers[workerID]; worker != nil && worker.terminal {
	176	+	completions = append(completions, worker.completion)
	177	+	}
	178	+	}
	179	+	return completions
	180	+	}

A packages/agent/internal/orchestration/coordinator_test.go +67 -0

@@ -0,0 +1,67 @@
	1	+	package orchestration
	2	+
	3	+	import (
	4	+	"testing"
	5	+
	6	+	"github.com/bnema/zut/packages/agent/subagents"
	7	+	)
	8	+
	9	+	func TestCoordinatorSealsWorkerWaveAndWakesOnce(t *testing.T) {
	10	+	c := New()
	11	+	c.Apply(Event{Kind: EventManagerStarted})
	12	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"})
	13	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"})
	14	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}})
	15	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b", Completion: subagents.Completion{AgentID: "b"}}), nil)
	16	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: subagents.Completion{AgentID: "a"}}), []Action{{
	17	+	Kind: ActionRunManager, Reason: WakeWorkers,
	18	+	Completions: []subagents.Completion{{AgentID: "a"}, {AgentID: "b"}},
	19	+	}})
	20	+	}
	21	+
	22	+	func TestCoordinatorGivesQueuedUserInputPriority(t *testing.T) {
	23	+	c := New()
	24	+	c.Apply(Event{Kind: EventManagerStarted})
	25	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "worker"})
	26	+	c.Apply(Event{Kind: EventManagerFinished})
	27	+	c.Apply(Event{Kind: EventUserInput, Text: "stop and explain"})
	28	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "worker", Completion: subagents.Completion{AgentID: "worker", Status: "failed"}}), []Action{{
	29	+	Kind: ActionRunManager, Reason: WakeUser, Text: "stop and explain",
	30	+	Completions: []subagents.Completion{{AgentID: "worker", Status: "failed"}},
	31	+	}})
	32	+	}
	33	+
	34	+	func TestCoordinatorUsesGoalOnlyWhenIdle(t *testing.T) {
	35	+	c := New()
	36	+	c.Apply(Event{Kind: EventGoalChanged, GoalActive: true})
	37	+	c.Apply(Event{Kind: EventManagerStarted})
	38	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionRunManager, Reason: WakeGoal}})
	39	+	}
	40	+
	41	+	func TestCoordinatorStopsAndIgnoresLateWorkerEvents(t *testing.T) {
	42	+	c := New()
	43	+	c.Apply(Event{Kind: EventManagerStarted})
	44	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "worker"})
	45	+	assertActions(t, c.Apply(Event{Kind: EventCancelled}), []Action{{Kind: ActionStop}})
	46	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "worker"}), nil)
	47	+	}
	48	+
	49	+	func assertActions(t *testing.T, got Result, want []Action) {
	50	+	t.Helper()
	51	+	if len(got.Actions) != len(want) {
	52	+	t.Fatalf("actions = %#v, want %#v", got.Actions, want)
	53	+	}
	54	+	for i := range want {
	55	+	if got.Actions[i].Kind != want[i].Kind \|\| got.Actions[i].Reason != want[i].Reason \|\| got.Actions[i].Text != want[i].Text {
	56	+	t.Fatalf("action %d = %#v, want %#v", i, got.Actions[i], want[i])
	57	+	}
	58	+	if len(got.Actions[i].Completions) != len(want[i].Completions) {
	59	+	t.Fatalf("action %d completions = %#v, want %#v", i, got.Actions[i].Completions, want[i].Completions)
	60	+	}
	61	+	for j := range want[i].Completions {
	62	+	if got.Actions[i].Completions[j].AgentID != want[i].Completions[j].AgentID \|\| got.Actions[i].Completions[j].Status != want[i].Completions[j].Status {
	63	+	t.Fatalf("action %d completion %d = %#v, want %#v", i, j, got.Actions[i].Completions[j], want[i].Completions[j])
	64	+	}
	65	+	}
	66	+	}
	67	+	}

@@ -0,0 +1,67 @@
	1	+	package orchestration
	2	+
	3	+	import (
	4	+	"testing"
	5	+
	6	+	"github.com/bnema/zut/packages/agent/subagents"
	7	+	)
	8	+
	9	+	func TestCoordinatorSealsWorkerWaveAndWakesOnce(t *testing.T) {
	10	+	c := New()
	11	+	c.Apply(Event{Kind: EventManagerStarted})
	12	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"})
	13	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"})
	14	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}})
	15	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b", Completion: subagents.Completion{AgentID: "b"}}), nil)
	16	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: subagents.Completion{AgentID: "a"}}), []Action{{
	17	+	Kind: ActionRunManager, Reason: WakeWorkers,
	18	+	Completions: []subagents.Completion{{AgentID: "a"}, {AgentID: "b"}},
	19	+	}})
	20	+	}
	21	+
	22	+	func TestCoordinatorGivesQueuedUserInputPriority(t *testing.T) {
	23	+	c := New()
	24	+	c.Apply(Event{Kind: EventManagerStarted})
	25	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "worker"})
	26	+	c.Apply(Event{Kind: EventManagerFinished})
	27	+	c.Apply(Event{Kind: EventUserInput, Text: "stop and explain"})
	28	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "worker", Completion: subagents.Completion{AgentID: "worker", Status: "failed"}}), []Action{{
	29	+	Kind: ActionRunManager, Reason: WakeUser, Text: "stop and explain",
	30	+	Completions: []subagents.Completion{{AgentID: "worker", Status: "failed"}},
	31	+	}})
	32	+	}
	33	+
	34	+	func TestCoordinatorUsesGoalOnlyWhenIdle(t *testing.T) {
	35	+	c := New()
	36	+	c.Apply(Event{Kind: EventGoalChanged, GoalActive: true})
	37	+	c.Apply(Event{Kind: EventManagerStarted})
	38	+	assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionRunManager, Reason: WakeGoal}})
	39	+	}
	40	+
	41	+	func TestCoordinatorStopsAndIgnoresLateWorkerEvents(t *testing.T) {
	42	+	c := New()
	43	+	c.Apply(Event{Kind: EventManagerStarted})
	44	+	c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "worker"})
	45	+	assertActions(t, c.Apply(Event{Kind: EventCancelled}), []Action{{Kind: ActionStop}})
	46	+	assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "worker"}), nil)
	47	+	}
	48	+
	49	+	func assertActions(t *testing.T, got Result, want []Action) {
	50	+	t.Helper()
	51	+	if len(got.Actions) != len(want) {
	52	+	t.Fatalf("actions = %#v, want %#v", got.Actions, want)
	53	+	}
	54	+	for i := range want {
	55	+	if got.Actions[i].Kind != want[i].Kind \|\| got.Actions[i].Reason != want[i].Reason \|\| got.Actions[i].Text != want[i].Text {
	56	+	t.Fatalf("action %d = %#v, want %#v", i, got.Actions[i], want[i])
	57	+	}
	58	+	if len(got.Actions[i].Completions) != len(want[i].Completions) {
	59	+	t.Fatalf("action %d completions = %#v, want %#v", i, got.Actions[i].Completions, want[i].Completions)
	60	+	}
	61	+	for j := range want[i].Completions {
	62	+	if got.Actions[i].Completions[j].AgentID != want[i].Completions[j].AgentID \|\| got.Actions[i].Completions[j].Status != want[i].Completions[j].Status {
	63	+	t.Fatalf("action %d completion %d = %#v, want %#v", i, j, got.Actions[i].Completions[j], want[i].Completions[j])
	64	+	}
	65	+	}
	66	+	}
	67	+	}

M packages/agent/modes/auto_subagents_test.go +2 -0


@@ -286,6 +286,7 @@ func TestTrackResumedSubagentWorkerQueuesActiveFollowUpOnSameWorker(t *testing.T
286	286		})
287	287
288	288		iv := newQueuedAutoSubagentsInteractive()
	289	+	release := iv.beginCompletionDeliveryHold()
289	290		iv.TrackSubagentWorker(first, first.Task, false)
290	291		const followUp = "queued follow-up"
291	292		cancel := iv.PrepareResumedSubagentWorker(first, followUp, false)

@@ -297,6 +298,7 @@ func TestTrackResumedSubagentWorkerQueuesActiveFollowUpOnSameWorker(t *testing.T
297	298		}
298	299		first.OnTurnEnd(1, "")
299	300		first.OnTurnEnd(2, "")
	301	+	release()
300	302
301	303		update := waitForQueuedPrompt(t, iv)
302	304		if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {


@@ -286,6 +286,7 @@ func TestTrackResumedSubagentWorkerQueuesActiveFollowUpOnSameWorker(t *testing.T
286	})	286		})
287		287
288	iv := newQueuedAutoSubagentsInteractive()	288		iv := newQueuedAutoSubagentsInteractive()
		289	+	release := iv.beginCompletionDeliveryHold()
289	iv.TrackSubagentWorker(first, first.Task, false)	290		iv.TrackSubagentWorker(first, first.Task, false)
290	const followUp = "queued follow-up"	291		const followUp = "queued follow-up"
291	cancel := iv.PrepareResumedSubagentWorker(first, followUp, false)	292		cancel := iv.PrepareResumedSubagentWorker(first, followUp, false)

@@ -297,6 +298,7 @@ func TestTrackResumedSubagentWorkerQueuesActiveFollowUpOnSameWorker(t *testing.T
297	}	298		}
298	first.OnTurnEnd(1, "")	299		first.OnTurnEnd(1, "")
299	first.OnTurnEnd(2, "")	300		first.OnTurnEnd(2, "")
		301	+	release()
300		302
301	update := waitForQueuedPrompt(t, iv)	303		update := waitForQueuedPrompt(t, iv)
302	if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {	304		if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {

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


@@ -19,6 +19,7 @@ import (
19	19
20	20		"github.com/bnema/zut/packages/agent/extensions"
21	21		"github.com/bnema/zut/packages/agent/extproto"
	22	+	"github.com/bnema/zut/packages/agent/internal/orchestration"
22	23		"github.com/bnema/zut/packages/agent/modes/telegram"
23	24		"github.com/bnema/zut/packages/agent/skills"
24	25		"github.com/bnema/zut/packages/agent/subagents"

@@ -789,9 +790,12 @@ type Interactive struct {
789	790		llamaConfigured bool
790	791
791	792		// completionTracker observes auto-subagent turn and process completion.
792		-	// Delivery is kept separate so every active set is injected as one
793		-	// summary turn, preserving the existing queue behavior.
	793	+	// turnCoordinator seals worker waves at manager-turn boundaries and decides
	794	+	// the next wake without relying on timing or polling.
794	795		completionTracker *subagents.CompletionTracker
	796	+	turnCoordinator *orchestration.Coordinator
	797	+	coordinatorWorkerIDs map[string][]string
	798	+	coordinatorWorkerSeq uint64
795	799		completionDeliveryMu sync.Mutex
796	800		completionDeliveryRunning bool
797	801		completionDeliveryRequest bool

@@ -4156,6 +4160,10 @@ func (i *Interactive) SubmitSlash(text string) {
4156	4160		// forwarded — because the queued-prompt path is text-only; a
4157	4161		// follow-up can expand the queue entry to carry images.
4158	4162		func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
	4163	+	i.submitOrQueue(text, images, true)
	4164	+	}
	4165	+
	4166	+	func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {
4159	4167		if cmd, ok := shellEscapeCommand(text); ok {
4160	4168		i.startShellEscape(i.runCtx, cmd)
4161	4169		return

@@ -4168,6 +4176,15 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
4168	4176		return
4169	4177		}
4170	4178		i.mu.Unlock()
	4179	+	// User input is an orchestration event, not merely a queue append. When
	4180	+	// the manager is dormant behind a sealed worker wave, the coordinator makes
	4181	+	// this the next manager turn and carries any completed worker summary with it.
	4182	+	if userInput && i.coordinatorAcceptsUserInput() {
	4183	+	if actions := i.applyCoordinator(orchestration.Event{Kind: orchestration.EventUserInput, Text: text}); len(actions) != 0 {
	4184	+	i.executeCoordinatorActions(actions)
	4185	+	return
	4186	+	}
	4187	+	}
4171	4188		i.maybeStartSessionTitle(i.runCtx, text)
4172	4189		i.mu.Lock()
4173	4190		if i.busy {

@@ -4213,6 +4230,8 @@ func (i *Interactive) CancelTurn() {
4213	4230		cancel := i.cancelTurn
4214	4231		i.mu.Unlock()
4215	4232		if cancel != nil {
	4233	+	i.cancelCoordinator()
	4234	+
4216	4235		i.mu.Lock()
4217	4236		handoff, persistHandoff := i.resetCompactContinuationLocked()
4218	4237		i.mu.Unlock()

@@ -8813,6 +8832,7 @@ func (i Interactive) TrackStoppedSubagentWorker(a subagents.Agent) {
8813	8832		return
8814	8833		}
8815	8834		tracker := i.ensureCompletionTracker()
	8835	+	i.registerCoordinatorWorker(a.ID)
8816	8836		tracker.TrackExit(a, a.Task)
8817	8837		i.invalidate()
8818	8838		i.requestCompletionDelivery()

@@ -8823,6 +8843,7 @@ func (i Interactive) trackSubagentWorker(a subagents.Agent, task string, follo
8823	8843		return nil
8824	8844		}
8825	8845		tracker := i.ensureCompletionTracker()
	8846	+	i.registerCoordinatorWorker(a.ID)
8826	8847		var cancel func()
8827	8848		if future {
8828	8849		cancel = tracker.TrackFutureTurn(a, task, followUp)

@@ -8845,6 +8866,126 @@ func (i Interactive) ensureCompletionTracker() subagents.CompletionTracker {
8845	8866		return i.completionTracker
8846	8867		}
8847	8868
	8869	+	func (i Interactive) ensureCoordinatorLocked() orchestration.Coordinator {
	8870	+	if i.turnCoordinator == nil {
	8871	+	i.turnCoordinator = orchestration.New()
	8872	+	}
	8873	+	return i.turnCoordinator
	8874	+	}
	8875	+
	8876	+	func (i *Interactive) coordinatorAcceptsUserInput() bool {
	8877	+	if i == nil {
	8878	+	return false
	8879	+	}
	8880	+	i.completionDeliveryMu.Lock()
	8881	+	accepts := i.ensureCoordinatorLocked().AcceptsUserInput()
	8882	+	i.completionDeliveryMu.Unlock()
	8883	+	return accepts
	8884	+	}
	8885	+
	8886	+	func (i *Interactive) applyCoordinator(event orchestration.Event) []orchestration.Action {
	8887	+	if i == nil {
	8888	+	return nil
	8889	+	}
	8890	+	i.completionDeliveryMu.Lock()
	8891	+	result := i.ensureCoordinatorLocked().Apply(event)
	8892	+	i.completionDeliveryMu.Unlock()
	8893	+	return result.Actions
	8894	+	}
	8895	+
	8896	+	// registerCoordinatorWorker associates a tracker registration with the open
	8897	+	// manager wave. Direct embedders that register a worker outside startTurn get
	8898	+	// a one-worker sealed wave, retaining the exported tracker API's old behavior.
	8899	+	func (i *Interactive) registerCoordinatorWorker(workerID string) {
	8900	+	if i == nil \|\| workerID == "" {
	8901	+	return
	8902	+	}
	8903	+	i.completionDeliveryMu.Lock()
	8904	+	coordinator := i.ensureCoordinatorLocked()
	8905	+	implicitWave := i.completionDeliveryHolds == 0
	8906	+	if implicitWave {
	8907	+	coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
	8908	+	}
	8909	+	i.coordinatorWorkerSeq++
	8910	+	registrationID := fmt.Sprintf("%s#%d", workerID, i.coordinatorWorkerSeq)
	8911	+	if i.coordinatorWorkerIDs == nil {
	8912	+	i.coordinatorWorkerIDs = make(map[string][]string)
	8913	+	}
	8914	+	i.coordinatorWorkerIDs[workerID] = append(i.coordinatorWorkerIDs[workerID], registrationID)
	8915	+	coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID})
	8916	+	var actions []orchestration.Action
	8917	+	if implicitWave {
	8918	+	actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
	8919	+	}
	8920	+	i.completionDeliveryMu.Unlock()
	8921	+	i.executeCoordinatorActions(actions)
	8922	+	}
	8923	+
	8924	+	func (i *Interactive) cancelCoordinator() {
	8925	+	if i == nil {
	8926	+	return
	8927	+	}
	8928	+	i.completionDeliveryMu.Lock()
	8929	+	coordinator := i.ensureCoordinatorLocked()
	8930	+	actions := coordinator.Apply(orchestration.Event{Kind: orchestration.EventCancelled}).Actions
	8931	+	// A later user turn starts a fresh wave. Drop tracker registrations as well
	8932	+	// as coordinator identities so late outcomes cannot match a new worker with
	8933	+	// the same Agent.ID.
	8934	+	if i.completionTracker != nil {
	8935	+	i.completionTracker.Reset()
	8936	+	}
	8937	+	i.turnCoordinator = orchestration.New()
	8938	+	i.coordinatorWorkerIDs = nil
	8939	+	i.completionDeliveryMu.Unlock()
	8940	+	i.executeCoordinatorActions(actions)
	8941	+	}
	8942	+
	8943	+	func (i *Interactive) takeCoordinatorWorkerID(agentID string) string {
	8944	+	if i == nil \|\| agentID == "" {
	8945	+	return ""
	8946	+	}
	8947	+	i.completionDeliveryMu.Lock()
	8948	+	defer i.completionDeliveryMu.Unlock()
	8949	+	ids := i.coordinatorWorkerIDs[agentID]
	8950	+	if len(ids) == 0 {
	8951	+	return ""
	8952	+	}
	8953	+	workerID := ids[0]
	8954	+	if len(ids) == 1 {
	8955	+	delete(i.coordinatorWorkerIDs, agentID)
	8956	+	} else {
	8957	+	i.coordinatorWorkerIDs[agentID] = ids[1:]
	8958	+	}
	8959	+	return workerID
	8960	+	}
	8961	+
	8962	+	func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) {
	8963	+	for _, action := range actions {
	8964	+	if action.Kind != orchestration.ActionRunManager {
	8965	+	continue
	8966	+	}
	8967	+	prompt := action.Text
	8968	+	if len(action.Completions) != 0 {
	8969	+	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."
	8970	+	update := subagents.FormatCompletionUpdate(action.Completions, instruction)
	8971	+	if prompt != "" {
	8972	+	prompt = update + "\n\nQueued user request:\n" + prompt
	8973	+	} else {
	8974	+	prompt = update
	8975	+	}
	8976	+	}
	8977	+	if prompt != "" {
	8978	+	i.submitOrQueue(prompt, nil, false)
	8979	+	} else if action.Reason == orchestration.WakeGoal {
	8980	+	parent := i.runCtx
	8981	+	if parent == nil {
	8982	+	parent = context.Background()
	8983	+	}
	8984	+	i.requestGoalContinuationIfIdle(parent)
	8985	+	}
	8986	+	}
	8987	+	}
	8988	+
8848	8989		// requestCompletionDelivery starts at most one waiter for the current active
8849	8990		// set. A request arriving while that waiter is formatting or submitting is
8850	8991		// picked up before the waiter exits, so a later worker cannot be lost in the

@@ -8875,6 +9016,7 @@ func (i *Interactive) beginCompletionDeliveryHold() func() {
8875	9016		return func() {}
8876	9017		}
8877	9018		i.completionDeliveryMu.Lock()
	9019	+	i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
8878	9020		i.completionDeliveryHolds++
8879	9021		i.completionDeliveryMu.Unlock()
8880	9022		return i.releaseCompletionDeliveryHold

@@ -8885,15 +9027,20 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
8885	9027		return
8886	9028		}
8887	9029		start := false
	9030	+	var actions []orchestration.Action
8888	9031		i.completionDeliveryMu.Lock()
8889	9032		if i.completionDeliveryHolds > 0 {
8890	9033		i.completionDeliveryHolds--
8891	9034		}
	9035	+	if i.completionDeliveryHolds == 0 {
	9036	+	actions = i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
	9037	+	}
8892	9038		if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {
8893	9039		i.completionDeliveryRunning = true
8894	9040		start = true
8895	9041		}
8896	9042		i.completionDeliveryMu.Unlock()
	9043	+	i.executeCoordinatorActions(actions)
8897	9044		if start {
8898	9045		go i.deliverCompletionUpdates()
8899	9046		}

@@ -8918,8 +9065,19 @@ func (i *Interactive) deliverCompletionUpdates() {
8918	9065
8919	9066		batch, err := tracker.WaitIdle(i.completionWaitContext())
8920	9067		if err == nil && len(batch) != 0 {
8921		-	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."
8922		-	i.SubmitOrQueue(subagents.FormatCompletionUpdate(batch, instruction), nil)
	9068	+	var actions []orchestration.Action
	9069	+	for _, completion := range batch {
	9070	+	workerID := i.takeCoordinatorWorkerID(completion.AgentID)
	9071	+	if workerID == "" {
	9072	+	continue
	9073	+	}
	9074	+	actions = append(actions, i.applyCoordinator(orchestration.Event{
	9075	+	Kind: orchestration.EventWorkerFinished,
	9076	+	WorkerID: workerID,
	9077	+	Completion: completion,
	9078	+	})...)
	9079	+	}
	9080	+	i.executeCoordinatorActions(actions)
8923	9081		}
8924	9082
8925	9083		i.completionDeliveryMu.Lock()


@@ -19,6 +19,7 @@ import (
19			19
20		"github.com/bnema/zut/packages/agent/extensions"	20		"github.com/bnema/zut/packages/agent/extensions"
21		"github.com/bnema/zut/packages/agent/extproto"	21		"github.com/bnema/zut/packages/agent/extproto"
			22	+	"github.com/bnema/zut/packages/agent/internal/orchestration"
22		"github.com/bnema/zut/packages/agent/modes/telegram"	23		"github.com/bnema/zut/packages/agent/modes/telegram"
23		"github.com/bnema/zut/packages/agent/skills"	24		"github.com/bnema/zut/packages/agent/skills"
24		"github.com/bnema/zut/packages/agent/subagents"	25		"github.com/bnema/zut/packages/agent/subagents"

@@ -789,9 +790,12 @@ type Interactive struct {
789		llamaConfigured bool	790		llamaConfigured bool
790			791
791		// completionTracker observes auto-subagent turn and process completion.	792		// completionTracker observes auto-subagent turn and process completion.
792	-	// Delivery is kept separate so every active set is injected as one	793	+	// turnCoordinator seals worker waves at manager-turn boundaries and decides
793	-	// summary turn, preserving the existing queue behavior.	794	+	// the next wake without relying on timing or polling.
794		completionTracker *subagents.CompletionTracker	795		completionTracker *subagents.CompletionTracker
			796	+	turnCoordinator *orchestration.Coordinator
			797	+	coordinatorWorkerIDs map[string][]string
			798	+	coordinatorWorkerSeq uint64
795		completionDeliveryMu sync.Mutex	799		completionDeliveryMu sync.Mutex
796		completionDeliveryRunning bool	800		completionDeliveryRunning bool
797		completionDeliveryRequest bool	801		completionDeliveryRequest bool

@@ -4156,6 +4160,10 @@ func (i *Interactive) SubmitSlash(text string) {
4156		// forwarded — because the queued-prompt path is text-only; a	4160		// forwarded — because the queued-prompt path is text-only; a
4157		// follow-up can expand the queue entry to carry images.	4161		// follow-up can expand the queue entry to carry images.
4158		func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {	4162		func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
			4163	+	i.submitOrQueue(text, images, true)
			4164	+	}
			4165	+
			4166	+	func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {
4159		if cmd, ok := shellEscapeCommand(text); ok {	4167		if cmd, ok := shellEscapeCommand(text); ok {
4160		i.startShellEscape(i.runCtx, cmd)	4168		i.startShellEscape(i.runCtx, cmd)
4161		return	4169		return

@@ -4168,6 +4176,15 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
4168		return	4176		return
4169		}	4177		}
4170		i.mu.Unlock()	4178		i.mu.Unlock()
			4179	+	// User input is an orchestration event, not merely a queue append. When
			4180	+	// the manager is dormant behind a sealed worker wave, the coordinator makes
			4181	+	// this the next manager turn and carries any completed worker summary with it.
			4182	+	if userInput && i.coordinatorAcceptsUserInput() {
			4183	+	if actions := i.applyCoordinator(orchestration.Event{Kind: orchestration.EventUserInput, Text: text}); len(actions) != 0 {
			4184	+	i.executeCoordinatorActions(actions)
			4185	+	return
			4186	+	}
			4187	+	}
4171		i.maybeStartSessionTitle(i.runCtx, text)	4188		i.maybeStartSessionTitle(i.runCtx, text)
4172		i.mu.Lock()	4189		i.mu.Lock()
4173		if i.busy {	4190		if i.busy {

@@ -4213,6 +4230,8 @@ func (i *Interactive) CancelTurn() {
4213		cancel := i.cancelTurn	4230		cancel := i.cancelTurn
4214		i.mu.Unlock()	4231		i.mu.Unlock()
4215		if cancel != nil {	4232		if cancel != nil {
			4233	+	i.cancelCoordinator()
			4234	+
4216		i.mu.Lock()	4235		i.mu.Lock()
4217		handoff, persistHandoff := i.resetCompactContinuationLocked()	4236		handoff, persistHandoff := i.resetCompactContinuationLocked()
4218		i.mu.Unlock()	4237		i.mu.Unlock()

@@ -8813,6 +8832,7 @@ func (i Interactive) TrackStoppedSubagentWorker(a subagents.Agent) {
8813		return	8832		return
8814		}	8833		}
8815		tracker := i.ensureCompletionTracker()	8834		tracker := i.ensureCompletionTracker()
			8835	+	i.registerCoordinatorWorker(a.ID)
8816		tracker.TrackExit(a, a.Task)	8836		tracker.TrackExit(a, a.Task)
8817		i.invalidate()	8837		i.invalidate()
8818		i.requestCompletionDelivery()	8838		i.requestCompletionDelivery()

@@ -8823,6 +8843,7 @@ func (i Interactive) trackSubagentWorker(a subagents.Agent, task string, follo
8823		return nil	8843		return nil
8824		}	8844		}
8825		tracker := i.ensureCompletionTracker()	8845		tracker := i.ensureCompletionTracker()
			8846	+	i.registerCoordinatorWorker(a.ID)
8826		var cancel func()	8847		var cancel func()
8827		if future {	8848		if future {
8828		cancel = tracker.TrackFutureTurn(a, task, followUp)	8849		cancel = tracker.TrackFutureTurn(a, task, followUp)

@@ -8845,6 +8866,126 @@ func (i Interactive) ensureCompletionTracker() subagents.CompletionTracker {
8845		return i.completionTracker	8866		return i.completionTracker
8846		}	8867		}
8847			8868
			8869	+	func (i Interactive) ensureCoordinatorLocked() orchestration.Coordinator {
			8870	+	if i.turnCoordinator == nil {
			8871	+	i.turnCoordinator = orchestration.New()
			8872	+	}
			8873	+	return i.turnCoordinator
			8874	+	}
			8875	+
			8876	+	func (i *Interactive) coordinatorAcceptsUserInput() bool {
			8877	+	if i == nil {
			8878	+	return false
			8879	+	}
			8880	+	i.completionDeliveryMu.Lock()
			8881	+	accepts := i.ensureCoordinatorLocked().AcceptsUserInput()
			8882	+	i.completionDeliveryMu.Unlock()
			8883	+	return accepts
			8884	+	}
			8885	+
			8886	+	func (i *Interactive) applyCoordinator(event orchestration.Event) []orchestration.Action {
			8887	+	if i == nil {
			8888	+	return nil
			8889	+	}
			8890	+	i.completionDeliveryMu.Lock()
			8891	+	result := i.ensureCoordinatorLocked().Apply(event)
			8892	+	i.completionDeliveryMu.Unlock()
			8893	+	return result.Actions
			8894	+	}
			8895	+
			8896	+	// registerCoordinatorWorker associates a tracker registration with the open
			8897	+	// manager wave. Direct embedders that register a worker outside startTurn get
			8898	+	// a one-worker sealed wave, retaining the exported tracker API's old behavior.
			8899	+	func (i *Interactive) registerCoordinatorWorker(workerID string) {
			8900	+	if i == nil \|\| workerID == "" {
			8901	+	return
			8902	+	}
			8903	+	i.completionDeliveryMu.Lock()
			8904	+	coordinator := i.ensureCoordinatorLocked()
			8905	+	implicitWave := i.completionDeliveryHolds == 0
			8906	+	if implicitWave {
			8907	+	coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
			8908	+	}
			8909	+	i.coordinatorWorkerSeq++
			8910	+	registrationID := fmt.Sprintf("%s#%d", workerID, i.coordinatorWorkerSeq)
			8911	+	if i.coordinatorWorkerIDs == nil {
			8912	+	i.coordinatorWorkerIDs = make(map[string][]string)
			8913	+	}
			8914	+	i.coordinatorWorkerIDs[workerID] = append(i.coordinatorWorkerIDs[workerID], registrationID)
			8915	+	coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID})
			8916	+	var actions []orchestration.Action
			8917	+	if implicitWave {
			8918	+	actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
			8919	+	}
			8920	+	i.completionDeliveryMu.Unlock()
			8921	+	i.executeCoordinatorActions(actions)
			8922	+	}
			8923	+
			8924	+	func (i *Interactive) cancelCoordinator() {
			8925	+	if i == nil {
			8926	+	return
			8927	+	}
			8928	+	i.completionDeliveryMu.Lock()
			8929	+	coordinator := i.ensureCoordinatorLocked()
			8930	+	actions := coordinator.Apply(orchestration.Event{Kind: orchestration.EventCancelled}).Actions
			8931	+	// A later user turn starts a fresh wave. Drop tracker registrations as well
			8932	+	// as coordinator identities so late outcomes cannot match a new worker with
			8933	+	// the same Agent.ID.
			8934	+	if i.completionTracker != nil {
			8935	+	i.completionTracker.Reset()
			8936	+	}
			8937	+	i.turnCoordinator = orchestration.New()
			8938	+	i.coordinatorWorkerIDs = nil
			8939	+	i.completionDeliveryMu.Unlock()
			8940	+	i.executeCoordinatorActions(actions)
			8941	+	}
			8942	+
			8943	+	func (i *Interactive) takeCoordinatorWorkerID(agentID string) string {
			8944	+	if i == nil \|\| agentID == "" {
			8945	+	return ""
			8946	+	}
			8947	+	i.completionDeliveryMu.Lock()
			8948	+	defer i.completionDeliveryMu.Unlock()
			8949	+	ids := i.coordinatorWorkerIDs[agentID]
			8950	+	if len(ids) == 0 {
			8951	+	return ""
			8952	+	}
			8953	+	workerID := ids[0]
			8954	+	if len(ids) == 1 {
			8955	+	delete(i.coordinatorWorkerIDs, agentID)
			8956	+	} else {
			8957	+	i.coordinatorWorkerIDs[agentID] = ids[1:]
			8958	+	}
			8959	+	return workerID
			8960	+	}
			8961	+
			8962	+	func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) {
			8963	+	for _, action := range actions {
			8964	+	if action.Kind != orchestration.ActionRunManager {
			8965	+	continue
			8966	+	}
			8967	+	prompt := action.Text
			8968	+	if len(action.Completions) != 0 {
			8969	+	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."
			8970	+	update := subagents.FormatCompletionUpdate(action.Completions, instruction)
			8971	+	if prompt != "" {
			8972	+	prompt = update + "\n\nQueued user request:\n" + prompt
			8973	+	} else {
			8974	+	prompt = update
			8975	+	}
			8976	+	}
			8977	+	if prompt != "" {
			8978	+	i.submitOrQueue(prompt, nil, false)
			8979	+	} else if action.Reason == orchestration.WakeGoal {
			8980	+	parent := i.runCtx
			8981	+	if parent == nil {
			8982	+	parent = context.Background()
			8983	+	}
			8984	+	i.requestGoalContinuationIfIdle(parent)
			8985	+	}
			8986	+	}
			8987	+	}
			8988	+
8848		// requestCompletionDelivery starts at most one waiter for the current active	8989		// requestCompletionDelivery starts at most one waiter for the current active
8849		// set. A request arriving while that waiter is formatting or submitting is	8990		// set. A request arriving while that waiter is formatting or submitting is
8850		// picked up before the waiter exits, so a later worker cannot be lost in the	8991		// picked up before the waiter exits, so a later worker cannot be lost in the

@@ -8875,6 +9016,7 @@ func (i *Interactive) beginCompletionDeliveryHold() func() {
8875		return func() {}	9016		return func() {}
8876		}	9017		}
8877		i.completionDeliveryMu.Lock()	9018		i.completionDeliveryMu.Lock()
			9019	+	i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
8878		i.completionDeliveryHolds++	9020		i.completionDeliveryHolds++
8879		i.completionDeliveryMu.Unlock()	9021		i.completionDeliveryMu.Unlock()
8880		return i.releaseCompletionDeliveryHold	9022		return i.releaseCompletionDeliveryHold

@@ -8885,15 +9027,20 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
8885		return	9027		return
8886		}	9028		}
8887		start := false	9029		start := false
			9030	+	var actions []orchestration.Action
8888		i.completionDeliveryMu.Lock()	9031		i.completionDeliveryMu.Lock()
8889		if i.completionDeliveryHolds > 0 {	9032		if i.completionDeliveryHolds > 0 {
8890		i.completionDeliveryHolds--	9033		i.completionDeliveryHolds--
8891		}	9034		}
			9035	+	if i.completionDeliveryHolds == 0 {
			9036	+	actions = i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
			9037	+	}
8892		if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {	9038		if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {
8893		i.completionDeliveryRunning = true	9039		i.completionDeliveryRunning = true
8894		start = true	9040		start = true
8895		}	9041		}
8896		i.completionDeliveryMu.Unlock()	9042		i.completionDeliveryMu.Unlock()
			9043	+	i.executeCoordinatorActions(actions)
8897		if start {	9044		if start {
8898		go i.deliverCompletionUpdates()	9045		go i.deliverCompletionUpdates()
8899		}	9046		}

@@ -8918,8 +9065,19 @@ func (i *Interactive) deliverCompletionUpdates() {
8918			9065
8919		batch, err := tracker.WaitIdle(i.completionWaitContext())	9066		batch, err := tracker.WaitIdle(i.completionWaitContext())
8920		if err == nil && len(batch) != 0 {	9067		if err == nil && len(batch) != 0 {
8921	-	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."	9068	+	var actions []orchestration.Action
8922	-	i.SubmitOrQueue(subagents.FormatCompletionUpdate(batch, instruction), nil)	9069	+	for _, completion := range batch {
			9070	+	workerID := i.takeCoordinatorWorkerID(completion.AgentID)
			9071	+	if workerID == "" {
			9072	+	continue
			9073	+	}
			9074	+	actions = append(actions, i.applyCoordinator(orchestration.Event{
			9075	+	Kind: orchestration.EventWorkerFinished,
			9076	+	WorkerID: workerID,
			9077	+	Completion: completion,
			9078	+	})...)
			9079	+	}
			9080	+	i.executeCoordinatorActions(actions)
8923		}	9081		}
8924			9082
8925		i.completionDeliveryMu.Lock()	9083		i.completionDeliveryMu.Lock()

M packages/agent/subagent_requirement_gate_test.go +50 -0


@@ -149,6 +149,56 @@ func TestRequiredWorkerGateTreatsManualTerminalRemovalAsWaiver(t *testing.T) {
149	149		}
150	150		}
151	151
	152	+	func TestRequiredWorkerGateWaitsForEveryRequiredWorker(t *testing.T) {
	153	+	releases := []chan struct{}{make(chan struct{}), make(chan struct{})}
	154	+	started := make(chan struct{}, len(releases))
	155	+	runnerIndex := 0
	156	+	cfg := newRuntimeTestConfig(t.TempDir(), t.TempDir(), func(*subagents.Agent) subagents.Runner {
	157	+	index := runnerIndex
	158	+	runnerIndex++
	159	+	return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
	160	+	started <- struct{}{}
	161	+	select {
	162	+	case <-releases[index]:
	163	+	return nil
	164	+	case <-ctx.Done():
	165	+	return ctx.Err()
	166	+	}
	167	+	})
	168	+	})
	169	+	rt := newSubagentRuntime(cfg)
	170	+	defer func() { _ = rt.Close(context.Background()) }()
	171	+	rt.SetActiveSession("parent-session")
	172	+	first, err := rt.Supervisor().SpawnReq(context.Background(), subagents.SpawnRequest{
	173	+	Task: "first required check", RootSessionID: "parent-session", Required: true,
	174	+	})
	175	+	if err != nil {
	176	+	t.Fatal(err)
	177	+	}
	178	+	second, err := rt.Supervisor().SpawnReq(context.Background(), subagents.SpawnRequest{
	179	+	Task: "second required check", RootSessionID: "parent-session", Required: true,
	180	+	})
	181	+	if err != nil {
	182	+	t.Fatal(err)
	183	+	}
	184	+	<-started
	185	+	<-started
	186	+
	187	+	parent := &core.Agent{}
	188	+	rt.WireRequiredWorkerGate(parent)
	189	+	close(releases[0])
	190	+	first.Wait()
	191	+	if allowed, reason, _ := parent.BeforeAssistantMessage("first worker completed"); allowed \|\| !strings.Contains(reason, second.ID) {
	192	+	t.Fatalf("terminal response after one required worker = (%v, %q), want remaining worker block", allowed, reason)
	193	+	}
	194	+
	195	+	close(releases[1])
	196	+	second.Wait()
	197	+	if allowed, reason, _ := parent.BeforeAssistantMessage("both workers completed"); !allowed \|\| reason != "" {
	198	+	t.Fatalf("terminal response after all required workers = (%v, %q), want allowed", allowed, reason)
	199	+	}
	200	+	}
	201	+
152	202		func TestRequiredWorkerGateRejectsTerminalParentResponseUntilRetrySucceeds(t *testing.T) {
153	203		attempt := 0
154	204		cfg := newRuntimeTestConfig(t.TempDir(), t.TempDir(), func(*subagents.Agent) subagents.Runner {


@@ -149,6 +149,56 @@ func TestRequiredWorkerGateTreatsManualTerminalRemovalAsWaiver(t *testing.T) {
149	}	149		}
150	}	150		}
151		151
		152	+	func TestRequiredWorkerGateWaitsForEveryRequiredWorker(t *testing.T) {
		153	+	releases := []chan struct{}{make(chan struct{}), make(chan struct{})}
		154	+	started := make(chan struct{}, len(releases))
		155	+	runnerIndex := 0
		156	+	cfg := newRuntimeTestConfig(t.TempDir(), t.TempDir(), func(*subagents.Agent) subagents.Runner {
		157	+	index := runnerIndex
		158	+	runnerIndex++
		159	+	return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
		160	+	started <- struct{}{}
		161	+	select {
		162	+	case <-releases[index]:
		163	+	return nil
		164	+	case <-ctx.Done():
		165	+	return ctx.Err()
		166	+	}
		167	+	})
		168	+	})
		169	+	rt := newSubagentRuntime(cfg)
		170	+	defer func() { _ = rt.Close(context.Background()) }()
		171	+	rt.SetActiveSession("parent-session")
		172	+	first, err := rt.Supervisor().SpawnReq(context.Background(), subagents.SpawnRequest{
		173	+	Task: "first required check", RootSessionID: "parent-session", Required: true,
		174	+	})
		175	+	if err != nil {
		176	+	t.Fatal(err)
		177	+	}
		178	+	second, err := rt.Supervisor().SpawnReq(context.Background(), subagents.SpawnRequest{
		179	+	Task: "second required check", RootSessionID: "parent-session", Required: true,
		180	+	})
		181	+	if err != nil {
		182	+	t.Fatal(err)
		183	+	}
		184	+	<-started
		185	+	<-started
		186	+
		187	+	parent := &core.Agent{}
		188	+	rt.WireRequiredWorkerGate(parent)
		189	+	close(releases[0])
		190	+	first.Wait()
		191	+	if allowed, reason, _ := parent.BeforeAssistantMessage("first worker completed"); allowed \|\| !strings.Contains(reason, second.ID) {
		192	+	t.Fatalf("terminal response after one required worker = (%v, %q), want remaining worker block", allowed, reason)
		193	+	}
		194	+
		195	+	close(releases[1])
		196	+	second.Wait()
		197	+	if allowed, reason, _ := parent.BeforeAssistantMessage("both workers completed"); !allowed \|\| reason != "" {
		198	+	t.Fatalf("terminal response after all required workers = (%v, %q), want allowed", allowed, reason)
		199	+	}
		200	+	}
		201	+
152	func TestRequiredWorkerGateRejectsTerminalParentResponseUntilRetrySucceeds(t *testing.T) {	202		func TestRequiredWorkerGateRejectsTerminalParentResponseUntilRetrySucceeds(t *testing.T) {
153	attempt := 0	203		attempt := 0
154	cfg := newRuntimeTestConfig(t.TempDir(), t.TempDir(), func(*subagents.Agent) subagents.Runner {	204		cfg := newRuntimeTestConfig(t.TempDir(), t.TempDir(), func(*subagents.Agent) subagents.Runner {

M packages/agent/subagent_requirements.go +4 -29


@@ -4,7 +4,6 @@ import (
4	4		"context"
5	5		"encoding/json"
6	6		"fmt"
7		-	"sort"
8	7		"strings"
9	8
10	9		"github.com/bnema/zut/packages/agent/subagents"

@@ -56,13 +55,11 @@ func (rt subagentRuntime) WireRequiredWorkerGate(parent core.Agent) {
56	55		parent.BeforeAssistantMessage = func(text string) (bool, string, string) {
57	56		unmet := rt.supervisor.UnmetRequirements()
58	57		if len(unmet) != 0 {
59		-	reason := formatRequiredWorkerBlock(unmet)
60		-	if rt.noteRequiredBlock(unmet) {
61		-	parent.QueueMessage(reason)
62		-	}
63		-	return false, reason, ""
	58	+	// Worker completion is delivered through the coordinator-owned tracker.
	59	+	// Do not queue a synthetic reminder here: it races that terminal wave
	60	+	// and can wake the parent before every required result is available.
	61	+	return false, formatRequiredWorkerBlock(unmet), ""
64	62		}
65		-	rt.clearRequiredBlock()
66	63		if previousBeforeAssistant != nil {
67	64		allowed, reason, replacement := previousBeforeAssistant(text)
68	65		if !allowed {

@@ -118,28 +115,6 @@ func (rt *subagentRuntime) acknowledgeSatisfiedRequirements() {
118	115
119	116		const requiredWorkerContextRuneBudget = 6000
120	117
121		-	func (rt *subagentRuntime) noteRequiredBlock(unmet []subagents.AgentSnapshot) bool {
122		-	ids := make([]string, 0, len(unmet))
123		-	for _, snapshot := range unmet {
124		-	ids = append(ids, fmt.Sprintf("%s:%s:%d:%s", snapshot.ID, snapshot.Requirement.State, snapshot.Requirement.TargetTurn, snapshot.Requirement.ErrorCode))
125		-	}
126		-	sort.Strings(ids)
127		-	key := strings.Join(ids, "\n")
128		-	rt.requiredContextMu.Lock()
129		-	defer rt.requiredContextMu.Unlock()
130		-	if key == rt.lastRequiredBlock {
131		-	return false
132		-	}
133		-	rt.lastRequiredBlock = key
134		-	return true
135		-	}
136		-
137		-	func (rt *subagentRuntime) clearRequiredBlock() {
138		-	rt.requiredContextMu.Lock()
139		-	rt.lastRequiredBlock = ""
140		-	rt.requiredContextMu.Unlock()
141		-	}
142		-
143	118		func (rt *subagentRuntime) formatRequiredWorkerContext(required []subagents.AgentSnapshot) string {
144	119		rt.requiredContextMu.Lock()
145	120		defer rt.requiredContextMu.Unlock()


@@ -4,7 +4,6 @@ import (
4		"context"	4		"context"
5		"encoding/json"	5		"encoding/json"
6		"fmt"	6		"fmt"
7	-	"sort"
8		"strings"	7		"strings"
9			8
10		"github.com/bnema/zut/packages/agent/subagents"	9		"github.com/bnema/zut/packages/agent/subagents"

@@ -56,13 +55,11 @@ func (rt subagentRuntime) WireRequiredWorkerGate(parent core.Agent) {
56		parent.BeforeAssistantMessage = func(text string) (bool, string, string) {	55		parent.BeforeAssistantMessage = func(text string) (bool, string, string) {
57		unmet := rt.supervisor.UnmetRequirements()	56		unmet := rt.supervisor.UnmetRequirements()
58		if len(unmet) != 0 {	57		if len(unmet) != 0 {
59	-	reason := formatRequiredWorkerBlock(unmet)	58	+	// Worker completion is delivered through the coordinator-owned tracker.
60	-	if rt.noteRequiredBlock(unmet) {	59	+	// Do not queue a synthetic reminder here: it races that terminal wave
61	-	parent.QueueMessage(reason)	60	+	// and can wake the parent before every required result is available.
62	-	}	61	+	return false, formatRequiredWorkerBlock(unmet), ""
63	-	return false, reason, ""
64		}	62		}
65	-	rt.clearRequiredBlock()
66		if previousBeforeAssistant != nil {	63		if previousBeforeAssistant != nil {
67		allowed, reason, replacement := previousBeforeAssistant(text)	64		allowed, reason, replacement := previousBeforeAssistant(text)
68		if !allowed {	65		if !allowed {

@@ -118,28 +115,6 @@ func (rt *subagentRuntime) acknowledgeSatisfiedRequirements() {
118			115
119		const requiredWorkerContextRuneBudget = 6000	116		const requiredWorkerContextRuneBudget = 6000
120			117
121	-	func (rt *subagentRuntime) noteRequiredBlock(unmet []subagents.AgentSnapshot) bool {
122	-	ids := make([]string, 0, len(unmet))
123	-	for _, snapshot := range unmet {
124	-	ids = append(ids, fmt.Sprintf("%s:%s:%d:%s", snapshot.ID, snapshot.Requirement.State, snapshot.Requirement.TargetTurn, snapshot.Requirement.ErrorCode))
125	-	}
126	-	sort.Strings(ids)
127	-	key := strings.Join(ids, "\n")
128	-	rt.requiredContextMu.Lock()
129	-	defer rt.requiredContextMu.Unlock()
130	-	if key == rt.lastRequiredBlock {
131	-	return false
132	-	}
133	-	rt.lastRequiredBlock = key
134	-	return true
135	-	}
136	-
137	-	func (rt *subagentRuntime) clearRequiredBlock() {
138	-	rt.requiredContextMu.Lock()
139	-	rt.lastRequiredBlock = ""
140	-	rt.requiredContextMu.Unlock()
141	-	}
142	-
143		func (rt *subagentRuntime) formatRequiredWorkerContext(required []subagents.AgentSnapshot) string {	118		func (rt *subagentRuntime) formatRequiredWorkerContext(required []subagents.AgentSnapshot) string {
144		rt.requiredContextMu.Lock()	119		rt.requiredContextMu.Lock()
145		defer rt.requiredContextMu.Unlock()	120		defer rt.requiredContextMu.Unlock()

M packages/agent/subagent_runtime.go +0 -1


@@ -48,7 +48,6 @@ type subagentRuntime struct {
48	48		requiredReadyErr func() error
49	49		requiredContextMu sync.Mutex
50	50		reportedRequiredUnmet map[string]struct{}
51		-	lastRequiredBlock string
52	51		settingsMu sync.RWMutex
53	52		closeMu sync.Mutex
54	53		closed bool


@@ -48,7 +48,6 @@ type subagentRuntime struct {
48		requiredReadyErr func() error	48	requiredReadyErr func() error
49		requiredContextMu sync.Mutex	49	requiredContextMu sync.Mutex
50		reportedRequiredUnmet map[string]struct{}	50	reportedRequiredUnmet map[string]struct{}
51	-	lastRequiredBlock string
52		settingsMu sync.RWMutex	51	settingsMu sync.RWMutex
53		closeMu sync.Mutex	52	closeMu sync.Mutex
54		closed bool	53	closed bool

M packages/agent/subagents/completion.go +29 -0


@@ -129,6 +129,35 @@ func (t CompletionTracker) TrackExit(a Agent, task string) {
129	129		t.startProcessWaiter(state)
130	130		}
131	131
	132	+	// Reset discards all active and ready completions. Hosts use it when the
	133	+	// owning turn is cancelled so a late result cannot be attributed to a later
	134	+	// registration for the same worker.
	135	+	func (t *CompletionTracker) Reset() {
	136	+	if t == nil {
	137	+	return
	138	+	}
	139	+	var cancels []context.CancelFunc
	140	+	t.mu.Lock()
	141	+	t.initLocked()
	142	+	for _, state := range t.active {
	143	+	for _, entry := range state.entries {
	144	+	entry.done = true
	145	+	}
	146	+	if state.waitCancel != nil {
	147	+	cancels = append(cancels, state.waitCancel)
	148	+	state.waitCancel = nil
	149	+	}
	150	+	state.waitingForProcess = false
	151	+	}
	152	+	t.active = make(map[Agent]completionAgent)
	153	+	t.ready = nil
	154	+	t.signalLocked()
	155	+	t.mu.Unlock()
	156	+	for _, cancel := range cancels {
	157	+	cancel()
	158	+	}
	159	+	}
	160	+
132	161		func (t CompletionTracker) addTurn(a Agent, task string, followUp, future bool) (completionAgent, completionEntry, bool) {
133	162		entry := &completionEntry{task: task, followUp: followUp, expectedStep: expectedTurnStep(a, future), strictStep: future}
134	163		t.mu.Lock()


@@ -129,6 +129,35 @@ func (t CompletionTracker) TrackExit(a Agent, task string) {
129	t.startProcessWaiter(state)	129		t.startProcessWaiter(state)
130	}	130		}
131		131
		132	+	// Reset discards all active and ready completions. Hosts use it when the
		133	+	// owning turn is cancelled so a late result cannot be attributed to a later
		134	+	// registration for the same worker.
		135	+	func (t *CompletionTracker) Reset() {
		136	+	if t == nil {
		137	+	return
		138	+	}
		139	+	var cancels []context.CancelFunc
		140	+	t.mu.Lock()
		141	+	t.initLocked()
		142	+	for _, state := range t.active {
		143	+	for _, entry := range state.entries {
		144	+	entry.done = true
		145	+	}
		146	+	if state.waitCancel != nil {
		147	+	cancels = append(cancels, state.waitCancel)
		148	+	state.waitCancel = nil
		149	+	}
		150	+	state.waitingForProcess = false
		151	+	}
		152	+	t.active = make(map[Agent]completionAgent)
		153	+	t.ready = nil
		154	+	t.signalLocked()
		155	+	t.mu.Unlock()
		156	+	for _, cancel := range cancels {
		157	+	cancel()
		158	+	}
		159	+	}
		160	+
132	func (t CompletionTracker) addTurn(a Agent, task string, followUp, future bool) (completionAgent, completionEntry, bool) {	161		func (t CompletionTracker) addTurn(a Agent, task string, followUp, future bool) (completionAgent, completionEntry, bool) {
133	entry := &completionEntry{task: task, followUp: followUp, expectedStep: expectedTurnStep(a, future), strictStep: future}	162		entry := &completionEntry{task: task, followUp: followUp, expectedStep: expectedTurnStep(a, future), strictStep: future}
134	t.mu.Lock()	163		t.mu.Lock()

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


@@ -282,6 +282,33 @@ func TestCompletionTrackerKeepsCompletionsBeforeWaiting(t *testing.T) {
282	282		close(agent.done)
283	283		}
284	284
	285	+	func TestCompletionTrackerResetDropsLateAndBufferedCompletions(t *testing.T) {
	286	+	agent := newCompletionTestAgent("worker-1", "task", StatusRunning)
	287	+	t.Cleanup(func() { close(agent.done) })
	288	+	tracker := NewCompletionTracker()
	289	+	tracker.TrackTurn(agent, agent.Task, false)
	290	+	agent.OnTurnEnd(1, "")
	291	+
	292	+	tracker.Reset()
	293	+	batch, err := tracker.WaitIdle(context.Background())
	294	+	if err != nil {
	295	+	t.Fatal(err)
	296	+	}
	297	+	if len(batch) != 0 {
	298	+	t.Fatalf("reset batch = %+v, want no stale completion", batch)
	299	+	}
	300	+	if tracker.Pending() != 0 {
	301	+	t.Fatalf("pending after reset = %d, want 0", tracker.Pending())
	302	+	}
	303	+
	304	+	tracker.TrackTurn(agent, "new task", true)
	305	+	agent.OnTurnEnd(2, "")
	306	+	batch, err = tracker.WaitIdle(context.Background())
	307	+	if err != nil \|\| len(batch) != 1 \|\| batch[0].Task != "new task" {
	308	+	t.Fatalf("new registration completion = (%+v, %v), want one fresh completion", batch, err)
	309	+	}
	310	+	}
	311	+
285	312		func TestCompletionTrackerWaitIdleCancellationLeavesActiveWork(t *testing.T) {
286	313		agent := newCompletionTestAgent("worker-1", "task", StatusRunning)
287	314		tracker := NewCompletionTracker()


@@ -282,6 +282,33 @@ func TestCompletionTrackerKeepsCompletionsBeforeWaiting(t *testing.T) {
282	close(agent.done)	282		close(agent.done)
283	}	283		}
284		284
		285	+	func TestCompletionTrackerResetDropsLateAndBufferedCompletions(t *testing.T) {
		286	+	agent := newCompletionTestAgent("worker-1", "task", StatusRunning)
		287	+	t.Cleanup(func() { close(agent.done) })
		288	+	tracker := NewCompletionTracker()
		289	+	tracker.TrackTurn(agent, agent.Task, false)
		290	+	agent.OnTurnEnd(1, "")
		291	+
		292	+	tracker.Reset()
		293	+	batch, err := tracker.WaitIdle(context.Background())
		294	+	if err != nil {
		295	+	t.Fatal(err)
		296	+	}
		297	+	if len(batch) != 0 {
		298	+	t.Fatalf("reset batch = %+v, want no stale completion", batch)
		299	+	}
		300	+	if tracker.Pending() != 0 {
		301	+	t.Fatalf("pending after reset = %d, want 0", tracker.Pending())
		302	+	}
		303	+
		304	+	tracker.TrackTurn(agent, "new task", true)
		305	+	agent.OnTurnEnd(2, "")
		306	+	batch, err = tracker.WaitIdle(context.Background())
		307	+	if err != nil \|\| len(batch) != 1 \|\| batch[0].Task != "new task" {
		308	+	t.Fatalf("new registration completion = (%+v, %v), want one fresh completion", batch, err)
		309	+	}
		310	+	}
		311	+
285	func TestCompletionTrackerWaitIdleCancellationLeavesActiveWork(t *testing.T) {	312		func TestCompletionTrackerWaitIdleCancellationLeavesActiveWork(t *testing.T) {
286	agent := newCompletionTestAgent("worker-1", "task", StatusRunning)	313		agent := newCompletionTestAgent("worker-1", "task", StatusRunning)
287	tracker := NewCompletionTracker()	314		tracker := NewCompletionTracker()

M packages/agent/subagents/scheduler_test.go +3 -1


@@ -173,7 +173,9 @@ func TestStopQueuedAgentBeforeRunnerAdmission(t *testing.T) {
173	173
174	174		func TestCanceledQueuedAgentIsRemovedAndGetsResult(t *testing.T) {
175	175		root := t.TempDir()
176		-	started := make(chan struct{})
	176	+	// The runner may start before the test reaches its receive. Buffer the
	177	+	// one-shot readiness signal so scheduling cannot turn it into a lost wakeup.
	178	+	started := make(chan struct{}, 1)
177	179		f := New(Config{
178	180		Root: root, RepoRoot: root,
179	181		Policy: SubagentPolicy{MaxConcurrent: 1, IdleTimeout: time.Hour},

 func TestCanceledQueuedAgentIsRemovedAndGetsResult(t *testing.T) {
 	root := t.TempDir()
-	started := make(chan struct{})
+	// The runner may start before the test reaches its receive. Buffer the
+	// one-shot readiness signal so scheduling cannot turn it into a lost wakeup.
+	started := make(chan struct{}, 1)
 	f := New(Config{
 		Root: root, RepoRoot: root,
 		Policy: SubagentPolicy{MaxConcurrent: 1, IdleTimeout: time.Hour},