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 delivered bool
33 terminal bool
34 completion subagents.Completion
35}
36
37type EventKind uint8
38
39const (
40 EventManagerStarted EventKind = iota + 1
41 EventManagerFinished
42 EventWorkerRegistered
43 EventWorkerFinished
44 EventUserInput
45 EventGoalChanged
46 EventCancelled
47 EventShutdown
48)
49
50type Event struct {
51 Kind EventKind
52 WorkerID string
53 Completion subagents.Completion
54 Text string
55 Images []provider.ImageBlock
56 GoalActive bool
57}
58
59type ActionKind uint8
60
61const (
62 ActionRunManager ActionKind = iota + 1
63 ActionWait
64 ActionStop
65 ActionQueueManager
66)
67
68type WakeReason uint8
69
70const (
71 WakeUser WakeReason = iota + 1
72 WakeWorkers
73 WakeGoal
74)
75
76type Action struct {
77 Kind ActionKind
78 Reason WakeReason
79 Text string
80 Images []provider.ImageBlock
81 Completions []subagents.Completion
82}
83
84type Result struct{ Actions []Action }
85
86func New() *Coordinator {
87 return &Coordinator{workers: make(map[string]*worker)}
88}
89
90// AcceptsUserInput reports whether a sealed worker wave is awaiting its next
91// manager turn. Ordinary interactive turns continue to use their existing
92// queueing path; this only gives user input priority over a worker wake.
93func (c *Coordinator) AcceptsUserInput() bool {
94 return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
95}
96
97// HasPendingWorkers reports whether the current manager wave still owns work.
98// Callers use it to avoid starting a goal continuation outside the coordinator.
99func (c *Coordinator) HasPendingWorkers() bool {
100 return c != nil && !c.stopped && c.hasPendingWorkers()
101}
102
103func (c *Coordinator) Apply(event Event) Result {
104 if c == nil || c.stopped {
105 return Result{}
106 }
107 switch event.Kind {
108 case EventManagerStarted:
109 c.managerActive = true
110 c.waveSealed = false
111 c.workers = make(map[string]*worker)
112 c.workerOrder = nil
113 case EventManagerFinished:
114 c.managerActive = false
115 c.waveSealed = true
116 if c.hasPendingWorkers() {
117 return Result{Actions: []Action{{Kind: ActionWait}}}
118 }
119 return c.wakeIfIdle()
120 case EventWorkerRegistered:
121 if !c.managerActive || c.waveSealed || event.WorkerID == "" {
122 return Result{}
123 }
124 if _, exists := c.workers[event.WorkerID]; !exists {
125 c.workers[event.WorkerID] = &worker{}
126 c.workerOrder = append(c.workerOrder, event.WorkerID)
127 }
128 case EventWorkerFinished:
129 w, exists := c.workers[event.WorkerID]
130 if !exists || w.terminal {
131 return Result{}
132 }
133 w.terminal = true
134 w.completion = event.Completion
135 if c.managerActive {
136 w.delivered = true
137 return Result{Actions: []Action{{Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{event.Completion}}}}
138 }
139 if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() {
140 return c.wakeIfIdle()
141 }
142 case EventUserInput:
143 if event.Text == "" && len(event.Images) == 0 {
144 return Result{}
145 }
146 c.queuedUser = append(c.queuedUser, queuedUserInput{text: event.Text, images: cloneImages(event.Images)})
147 return c.wakeIfIdle()
148 case EventGoalChanged:
149 c.goalActive = event.GoalActive
150 case EventCancelled, EventShutdown:
151 c.stopped = true
152 c.queuedUser = nil
153 return Result{Actions: []Action{{Kind: ActionStop}}}
154 }
155 return Result{}
156}
157
158func (c *Coordinator) wakeIfIdle() Result {
159 if c.managerActive || c.stopped || c.hasPendingWorkers() {
160 return Result{}
161 }
162 completions := c.completedWorkers()
163 if len(c.queuedUser) != 0 {
164 input := c.queuedUser[0]
165 c.queuedUser = c.queuedUser[1:]
166 return c.startManager(WakeUser, input.text, input.images, completions)
167 }
168 if len(completions) != 0 {
169 return c.startManager(WakeWorkers, "", nil, completions)
170 }
171 if c.goalActive {
172 return c.startManager(WakeGoal, "", nil, nil)
173 }
174 return Result{}
175}
176
177func (c *Coordinator) startManager(reason WakeReason, text string, images []provider.ImageBlock, completions []subagents.Completion) Result {
178 c.managerActive = true
179 return Result{Actions: []Action{{Kind: ActionRunManager, Reason: reason, Text: text, Images: images, Completions: completions}}}
180}
181
182func cloneImages(images []provider.ImageBlock) []provider.ImageBlock {
183 if len(images) == 0 {
184 return nil
185 }
186 cloned := make([]provider.ImageBlock, len(images))
187 for index, image := range images {
188 cloned[index] = image
189 cloned[index].Data = append([]byte(nil), image.Data...)
190 }
191 return cloned
192}
193
194func (c *Coordinator) hasPendingWorkers() bool {
195 for _, worker := range c.workers {
196 if !worker.terminal {
197 return true
198 }
199 }
200 return false
201}
202
203func (c *Coordinator) completedWorkers() []subagents.Completion {
204 if c.hasPendingWorkers() {
205 return nil
206 }
207 completions := make([]subagents.Completion, 0, len(c.workerOrder))
208 for _, workerID := range c.workerOrder {
209 if worker := c.workers[workerID]; worker != nil && worker.terminal && !worker.delivered {
210 completions = append(completions, worker.completion)
211 }
212 }
213 return completions
214}