detached cb3c4e4

feat(agent): coordinate delegated worker waves

commit
cb3c4e4eedd0855e066e6440805e2b2990d90622
author
brice · · 2026-08-11 03:56 UTC
committer
brice · · 2026-08-11 04:23 UTC
parent
75d4151
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
10 files · +522 -35
Unified
UnifiedSplit
M docs/subagents.md +1 -1
@@ -143,7 +143,7 @@ During an active delegated turn, interactive mode shows compact rows beside the
143143
144144 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.
145145
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:
147147
148148 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.
149149
@@ -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
286286 })
287287
288288 iv := newQueuedAutoSubagentsInteractive()
289+ release := iv.beginCompletionDeliveryHold()
289290 iv.TrackSubagentWorker(first, first.Task, false)
290291 const followUp = "queued follow-up"
291292 cancel := iv.PrepareResumedSubagentWorker(first, followUp, false)
@@ -297,6 +298,7 @@ func TestTrackResumedSubagentWorkerQueuesActiveFollowUpOnSameWorker(t *testing.T
297298 }
298299 first.OnTurnEnd(1, "")
299300 first.OnTurnEnd(2, "")
301+ release()
300302
301303 update := waitForQueuedPrompt(t, iv)
302304 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 (
1919
2020 "github.com/bnema/zut/packages/agent/extensions"
2121 "github.com/bnema/zut/packages/agent/extproto"
22+ "github.com/bnema/zut/packages/agent/internal/orchestration"
2223 "github.com/bnema/zut/packages/agent/modes/telegram"
2324 "github.com/bnema/zut/packages/agent/skills"
2425 "github.com/bnema/zut/packages/agent/subagents"
@@ -789,9 +790,12 @@ type Interactive struct {
789790 llamaConfigured bool
790791
791792 // 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.
794795 completionTracker *subagents.CompletionTracker
796+ turnCoordinator *orchestration.Coordinator
797+ coordinatorWorkerIDs map[string][]string
798+ coordinatorWorkerSeq uint64
795799 completionDeliveryMu sync.Mutex
796800 completionDeliveryRunning bool
797801 completionDeliveryRequest bool
@@ -4156,6 +4160,10 @@ func (i *Interactive) SubmitSlash(text string) {
41564160 // forwarded — because the queued-prompt path is text-only; a
41574161 // follow-up can expand the queue entry to carry images.
41584162 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) {
41594167 if cmd, ok := shellEscapeCommand(text); ok {
41604168 i.startShellEscape(i.runCtx, cmd)
41614169 return
@@ -4168,6 +4176,15 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
41684176 return
41694177 }
41704178 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+ }
41714188 i.maybeStartSessionTitle(i.runCtx, text)
41724189 i.mu.Lock()
41734190 if i.busy {
@@ -4213,6 +4230,8 @@ func (i *Interactive) CancelTurn() {
42134230 cancel := i.cancelTurn
42144231 i.mu.Unlock()
42154232 if cancel != nil {
4233+ i.cancelCoordinator()
4234+
42164235 i.mu.Lock()
42174236 handoff, persistHandoff := i.resetCompactContinuationLocked()
42184237 i.mu.Unlock()
@@ -8813,6 +8832,7 @@ func (i *Interactive) TrackStoppedSubagentWorker(a *subagents.Agent) {
88138832 return
88148833 }
88158834 tracker := i.ensureCompletionTracker()
8835+ i.registerCoordinatorWorker(a.ID)
88168836 tracker.TrackExit(a, a.Task)
88178837 i.invalidate()
88188838 i.requestCompletionDelivery()
@@ -8823,6 +8843,7 @@ func (i *Interactive) trackSubagentWorker(a *subagents.Agent, task string, follo
88238843 return nil
88248844 }
88258845 tracker := i.ensureCompletionTracker()
8846+ i.registerCoordinatorWorker(a.ID)
88268847 var cancel func()
88278848 if future {
88288849 cancel = tracker.TrackFutureTurn(a, task, followUp)
@@ -8845,6 +8866,126 @@ func (i *Interactive) ensureCompletionTracker() *subagents.CompletionTracker {
88458866 return i.completionTracker
88468867 }
88478868
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+
88488989 // requestCompletionDelivery starts at most one waiter for the current active
88498990 // set. A request arriving while that waiter is formatting or submitting is
88508991 // picked up before the waiter exits, so a later worker cannot be lost in the
@@ -8875,6 +9016,7 @@ func (i *Interactive) beginCompletionDeliveryHold() func() {
88759016 return func() {}
88769017 }
88779018 i.completionDeliveryMu.Lock()
9019+ i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
88789020 i.completionDeliveryHolds++
88799021 i.completionDeliveryMu.Unlock()
88809022 return i.releaseCompletionDeliveryHold
@@ -8885,15 +9027,20 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
88859027 return
88869028 }
88879029 start := false
9030+ var actions []orchestration.Action
88889031 i.completionDeliveryMu.Lock()
88899032 if i.completionDeliveryHolds > 0 {
88909033 i.completionDeliveryHolds--
88919034 }
9035+ if i.completionDeliveryHolds == 0 {
9036+ actions = i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
9037+ }
88929038 if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {
88939039 i.completionDeliveryRunning = true
88949040 start = true
88959041 }
88969042 i.completionDeliveryMu.Unlock()
9043+ i.executeCoordinatorActions(actions)
88979044 if start {
88989045 go i.deliverCompletionUpdates()
88999046 }
@@ -8918,8 +9065,19 @@ func (i *Interactive) deliverCompletionUpdates() {
89189065
89199066 batch, err := tracker.WaitIdle(i.completionWaitContext())
89209067 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)
89239081 }
89249082
89259083 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 bool790 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 one793+ // 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.CompletionTracker795 completionTracker *subagents.CompletionTracker
796+ turnCoordinator *orchestration.Coordinator
797+ coordinatorWorkerIDs map[string][]string
798+ coordinatorWorkerSeq uint64
795 completionDeliveryMu sync.Mutex799 completionDeliveryMu sync.Mutex
796 completionDeliveryRunning bool800 completionDeliveryRunning bool
797 completionDeliveryRequest bool801 completionDeliveryRequest bool
@@ -4156,6 +4160,10 @@ func (i *Interactive) SubmitSlash(text string) {
4156 // forwarded — because the queued-prompt path is text-only; a4160 // 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 return4169 return
@@ -4168,6 +4176,15 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
4168 return4176 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.cancelTurn4230 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 return8832 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 nil8843 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.completionTracker8866 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 active8989 // requestCompletionDelivery starts at most one waiter for the current active
8849 // set. A request arriving while that waiter is formatting or submitting is8990 // 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 the8991 // 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.releaseCompletionDeliveryHold9022 return i.releaseCompletionDeliveryHold
@@ -8885,15 +9027,20 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
8885 return9027 return
8886 }9028 }
8887 start := false9029 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 = true9039 i.completionDeliveryRunning = true
8894 start = true9040 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) {
149149 }
150150 }
151151
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+
152202 func TestRequiredWorkerGateRejectsTerminalParentResponseUntilRetrySucceeds(t *testing.T) {
153203 attempt := 0
154204 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 := 0203 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 (
44 "context"
55 "encoding/json"
66 "fmt"
7- "sort"
87 "strings"
98
109 "github.com/bnema/zut/packages/agent/subagents"
@@ -56,13 +55,11 @@ func (rt *subagentRuntime) WireRequiredWorkerGate(parent *core.Agent) {
5655 parent.BeforeAssistantMessage = func(text string) (bool, string, string) {
5756 unmet := rt.supervisor.UnmetRequirements()
5857 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), ""
6462 }
65- rt.clearRequiredBlock()
6663 if previousBeforeAssistant != nil {
6764 allowed, reason, replacement := previousBeforeAssistant(text)
6865 if !allowed {
@@ -118,28 +115,6 @@ func (rt *subagentRuntime) acknowledgeSatisfiedRequirements() {
118115
119116 const requiredWorkerContextRuneBudget = 6000
120117
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-
143118 func (rt *subagentRuntime) formatRequiredWorkerContext(required []subagents.AgentSnapshot) string {
144119 rt.requiredContextMu.Lock()
145120 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 = 6000116 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 {
4848 requiredReadyErr func() error
4949 requiredContextMu sync.Mutex
5050 reportedRequiredUnmet map[string]struct{}
51- lastRequiredBlock string
5251 settingsMu sync.RWMutex
5352 closeMu sync.Mutex
5453 closed bool
@@ -48,7 +48,6 @@ type subagentRuntime struct {
48 requiredReadyErr func() error48 requiredReadyErr func() error
49 requiredContextMu sync.Mutex49 requiredContextMu sync.Mutex
50 reportedRequiredUnmet map[string]struct{}50 reportedRequiredUnmet map[string]struct{}
51- lastRequiredBlock string
52 settingsMu sync.RWMutex51 settingsMu sync.RWMutex
53 closeMu sync.Mutex52 closeMu sync.Mutex
54 closed bool53 closed bool
M packages/agent/subagents/completion.go +29 -0
@@ -129,6 +129,35 @@ func (t *CompletionTracker) TrackExit(a *Agent, task string) {
129129 t.startProcessWaiter(state)
130130 }
131131
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+
132161 func (t *CompletionTracker) addTurn(a *Agent, task string, followUp, future bool) (*completionAgent, *completionEntry, bool) {
133162 entry := &completionEntry{task: task, followUp: followUp, expectedStep: expectedTurnStep(a, future), strictStep: future}
134163 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) {
282282 close(agent.done)
283283 }
284284
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+
285312 func TestCompletionTrackerWaitIdleCancellationLeavesActiveWork(t *testing.T) {
286313 agent := newCompletionTestAgent("worker-1", "task", StatusRunning)
287314 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()

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 476

↑↓ moveenter openesc close