coordinator.go RawHistoryFolder

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 "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.
13type 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
23type worker struct {
24	terminal   bool
25	completion subagents.Completion
26}
27
28type EventKind uint8
29
30const (
31	EventManagerStarted EventKind = iota + 1
32	EventManagerFinished
33	EventWorkerRegistered
34	EventWorkerFinished
35	EventUserInput
36	EventGoalChanged
37	EventCancelled
38	EventShutdown
39)
40
41type Event struct {
42	Kind       EventKind
43	WorkerID   string
44	Completion subagents.Completion
45	Text       string
46	GoalActive bool
47}
48
49type ActionKind uint8
50
51const (
52	ActionRunManager ActionKind = iota + 1
53	ActionWait
54	ActionStop
55)
56
57type WakeReason uint8
58
59const (
60	WakeUser WakeReason = iota + 1
61	WakeWorkers
62	WakeGoal
63)
64
65type Action struct {
66	Kind        ActionKind
67	Reason      WakeReason
68	Text        string
69	Completions []subagents.Completion
70}
71
72type Result struct{ Actions []Action }
73
74func 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.
81func (c *Coordinator) AcceptsUserInput() bool {
82	return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
83}
84
85// HasPendingWorkers reports whether the current manager wave still owns work.
86// Callers use it to avoid starting a goal continuation outside the coordinator.
87func (c *Coordinator) HasPendingWorkers() bool {
88	return c != nil && !c.stopped && c.hasPendingWorkers()
89}
90
91func (c *Coordinator) Apply(event Event) Result {
92	if c == nil || c.stopped {
93		return Result{}
94	}
95	switch event.Kind {
96	case EventManagerStarted:
97		c.managerActive = true
98		c.waveSealed = false
99		c.workers = make(map[string]*worker)
100		c.workerOrder = nil
101	case EventManagerFinished:
102		c.managerActive = false
103		c.waveSealed = true
104		if c.hasPendingWorkers() {
105			return Result{Actions: []Action{{Kind: ActionWait}}}
106		}
107		return c.wakeIfIdle()
108	case EventWorkerRegistered:
109		if !c.managerActive || c.waveSealed || event.WorkerID == "" {
110			return Result{}
111		}
112		if _, exists := c.workers[event.WorkerID]; !exists {
113			c.workers[event.WorkerID] = &worker{}
114			c.workerOrder = append(c.workerOrder, event.WorkerID)
115		}
116	case EventWorkerFinished:
117		w, exists := c.workers[event.WorkerID]
118		if !exists || w.terminal {
119			return Result{}
120		}
121		w.terminal = true
122		w.completion = event.Completion
123		if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() {
124			return c.wakeIfIdle()
125		}
126	case EventUserInput:
127		if event.Text == "" {
128			return Result{}
129		}
130		c.queuedUser = append(c.queuedUser, event.Text)
131		return c.wakeIfIdle()
132	case EventGoalChanged:
133		c.goalActive = event.GoalActive
134	case EventCancelled, EventShutdown:
135		c.stopped = true
136		c.queuedUser = nil
137		return Result{Actions: []Action{{Kind: ActionStop}}}
138	}
139	return Result{}
140}
141
142func (c *Coordinator) wakeIfIdle() Result {
143	if c.managerActive || c.stopped || c.hasPendingWorkers() {
144		return Result{}
145	}
146	completions := c.completedWorkers()
147	if len(c.queuedUser) != 0 {
148		text := c.queuedUser[0]
149		c.queuedUser = c.queuedUser[1:]
150		return c.startManager(WakeUser, text, completions)
151	}
152	if len(completions) != 0 {
153		return c.startManager(WakeWorkers, "", completions)
154	}
155	if c.goalActive {
156		return c.startManager(WakeGoal, "", nil)
157	}
158	return Result{}
159}
160
161func (c *Coordinator) startManager(reason WakeReason, text string, completions []subagents.Completion) Result {
162	c.managerActive = true
163	return Result{Actions: []Action{{Kind: ActionRunManager, Reason: reason, Text: text, Completions: completions}}}
164}
165
166func (c *Coordinator) hasPendingWorkers() bool {
167	for _, worker := range c.workers {
168		if !worker.terminal {
169			return true
170		}
171	}
172	return false
173}
174
175func (c *Coordinator) completedWorkers() []subagents.Completion {
176	if c.hasPendingWorkers() {
177		return nil
178	}
179	completions := make([]subagents.Completion, 0, len(c.workerOrder))
180	for _, workerID := range c.workerOrder {
181		if worker := c.workers[workerID]; worker != nil && worker.terminal {
182			completions = append(completions, worker.completion)
183		}
184	}
185	return completions
186}