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.
5package orchestration
6
7import (
8 "github.com/bnema/zut/packages/agent/subagents"
9 "github.com/bnema/zut/packages/provider"
10)
11
12// Coordinator queues terminal worker results into an active manager turn.
13// When the manager is idle, it collects the remaining sealed worker wave.
14// Queued user input wins the next wake; an active goal is used only when no
15// user input or worker result remains.
16type Coordinator struct {
17 managerActive bool
18 waveSealed bool
19 stopped bool
20 goalActive bool
21 workers map[string]*worker
22 workerOrder []string
23 queuedUser []queuedUserInput
24}
25
26type queuedUserInput struct {
27 text string
28 images []provider.ImageBlock
29}
30
31type worker struct {
32 agentID string
33 delivered bool
34 terminal bool
35 completion subagents.Completion
36}
37
38type EventKind uint8
39
40const (
41 EventManagerStarted EventKind = iota + 1
42 EventManagerFinished
43 EventWorkerRegistered
44 EventWorkerFinished
45 EventUserInput
46 EventGoalChanged
47 EventCancelled
48 EventShutdown
49)
50
51type Event struct {
52 Kind EventKind
53 WorkerID string
54 AgentID string
55 Completion subagents.Completion
56 Text string
57 Images []provider.ImageBlock
58 GoalActive bool
59}
60
61type ActionKind uint8
62
63const (
64 ActionRunManager ActionKind = iota + 1
65 ActionWait
66 ActionStop
67 ActionQueueManager
68)
69
70type WakeReason uint8
71
72const (
73 WakeUser WakeReason = iota + 1
74 WakeWorkers
75 WakeGoal
76)
77
78type Action struct {
79 Kind ActionKind
80 Reason WakeReason
81 Text string
82 Images []provider.ImageBlock
83 Completions []subagents.Completion
84 ActiveAgentIDs []string
85}
86
87type Result struct{ Actions []Action }
88
89func New() *Coordinator {
90 return &Coordinator{workers: make(map[string]*worker)}
91}
92
93// AcceptsUserInput reports whether a sealed worker wave is awaiting its next
94// manager turn. Ordinary interactive turns continue to use their existing
95// queueing path; this only gives user input priority over a worker wake.
96func (c *Coordinator) AcceptsUserInput() bool {
97 return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
98}
99
100// HasPendingWorkers reports whether the current manager wave still owns work.
101// Callers use it to avoid starting a goal continuation outside the coordinator.
102func (c *Coordinator) HasPendingWorkers() bool {
103 return c != nil && !c.stopped && c.hasPendingWorkers()
104}
105
106func (c *Coordinator) Apply(event Event) Result {
107 if c == nil || c.stopped {
108 return Result{}
109 }
110 switch event.Kind {
111 case EventManagerStarted:
112 c.managerActive = true
113 c.waveSealed = false
114 c.dropDeliveredWorkers()
115 case EventManagerFinished:
116 c.managerActive = false
117 c.waveSealed = true
118 if c.hasPendingWorkers() {
119 return Result{Actions: []Action{{Kind: ActionWait}}}
120 }
121 return c.wakeIfIdle()
122 case EventWorkerRegistered:
123 if !c.managerActive || c.waveSealed || event.WorkerID == "" {
124 return Result{}
125 }
126 if _, exists := c.workers[event.WorkerID]; !exists {
127 c.workers[event.WorkerID] = &worker{agentID: event.AgentID}
128 c.workerOrder = append(c.workerOrder, event.WorkerID)
129 }
130 case EventWorkerFinished:
131 w, exists := c.workers[event.WorkerID]
132 if !exists || w.terminal {
133 return Result{}
134 }
135 w.terminal = true
136 w.completion = event.Completion
137 if c.managerActive {
138 w.delivered = true
139 return Result{Actions: []Action{{
140 Kind: ActionQueueManager,
141 Reason: WakeWorkers,
142 Completions: []subagents.Completion{event.Completion},
143 ActiveAgentIDs: c.activeAgentIDs(),
144 }}}
145 }
146 if c.waveSealed {
147 w.delivered = true
148 action := Action{
149 Kind: ActionRunManager,
150 Reason: WakeWorkers,
151 Completions: []subagents.Completion{event.Completion},
152 ActiveAgentIDs: c.activeAgentIDs(),
153 }
154 if len(c.queuedUser) != 0 {
155 input := c.queuedUser[0]
156 c.queuedUser = c.queuedUser[1:]
157 action.Reason = WakeUser
158 action.Text = input.text
159 action.Images = input.images
160 }
161 c.managerActive = true
162 return Result{Actions: []Action{action}}
163 }
164 case EventUserInput:
165 if event.Text == "" && len(event.Images) == 0 {
166 return Result{}
167 }
168 c.queuedUser = append(c.queuedUser, queuedUserInput{text: event.Text, images: cloneImages(event.Images)})
169 return c.wakeIfIdle()
170 case EventGoalChanged:
171 c.goalActive = event.GoalActive
172 case EventCancelled, EventShutdown:
173 c.stopped = true
174 c.queuedUser = nil
175 return Result{Actions: []Action{{Kind: ActionStop}}}
176 }
177 return Result{}
178}
179
180func (c *Coordinator) wakeIfIdle() Result {
181 if c.managerActive || c.stopped || c.hasPendingWorkers() {
182 return Result{}
183 }
184 completions := c.completedWorkers()
185 if len(c.queuedUser) != 0 {
186 input := c.queuedUser[0]
187 c.queuedUser = c.queuedUser[1:]
188 return c.startManager(WakeUser, input.text, input.images, completions)
189 }
190 if len(completions) != 0 {
191 return c.startManager(WakeWorkers, "", nil, completions)
192 }
193 if c.goalActive {
194 return c.startManager(WakeGoal, "", nil, nil)
195 }
196 return Result{}
197}
198
199func (c *Coordinator) startManager(reason WakeReason, text string, images []provider.ImageBlock, completions []subagents.Completion) Result {
200 c.managerActive = true
201 return Result{Actions: []Action{{Kind: ActionRunManager, Reason: reason, Text: text, Images: images, Completions: completions}}}
202}
203
204func cloneImages(images []provider.ImageBlock) []provider.ImageBlock {
205 if len(images) == 0 {
206 return nil
207 }
208 cloned := make([]provider.ImageBlock, len(images))
209 for index, image := range images {
210 cloned[index] = image
211 cloned[index].Data = append([]byte(nil), image.Data...)
212 }
213 return cloned
214}
215
216func (c *Coordinator) hasPendingWorkers() bool {
217 for _, worker := range c.workers {
218 if !worker.terminal {
219 return true
220 }
221 }
222 return false
223}
224
225func (c *Coordinator) activeAgentIDs() []string {
226 ids := make([]string, 0, len(c.workerOrder))
227 seen := make(map[string]struct{})
228 for _, workerID := range c.workerOrder {
229 worker := c.workers[workerID]
230 if worker == nil || worker.terminal || worker.agentID == "" {
231 continue
232 }
233 if _, exists := seen[worker.agentID]; exists {
234 continue
235 }
236 seen[worker.agentID] = struct{}{}
237 ids = append(ids, worker.agentID)
238 }
239 return ids
240}
241
242func (c *Coordinator) dropDeliveredWorkers() {
243 order := c.workerOrder[:0]
244 for _, workerID := range c.workerOrder {
245 worker := c.workers[workerID]
246 if worker == nil || worker.delivered {
247 delete(c.workers, workerID)
248 continue
249 }
250 order = append(order, workerID)
251 }
252 c.workerOrder = order
253}
254
255func (c *Coordinator) completedWorkers() []subagents.Completion {
256 if c.hasPendingWorkers() {
257 return nil
258 }
259 completions := make([]subagents.Completion, 0, len(c.workerOrder))
260 for _, workerID := range c.workerOrder {
261 if worker := c.workers[workerID]; worker != nil && worker.terminal && !worker.delivered {
262 completions = append(completions, worker.completion)
263 }
264 }
265 return completions
266}