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