1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8 "sync"
9 "time"
10
11 "github.com/bnema/zut/packages/core"
12 "github.com/google/uuid"
13)
14
15// ResidentTurnRunner executes one already-accepted prompt. The child owns the
16// ordering and cancellation boundary; implementations own agent/provider work.
17type ResidentTurnRunner func(context.Context, string) error
18
19type residentPrompt struct {
20 turnID string
21 prompt string
22 ack chan error
23}
24
25type residentTurnResult struct {
26 turnID string
27 err error
28}
29
30// ResidentCompletion is a typed in-process terminal turn notification.
31type ResidentCompletion struct {
32 ChildID string
33 TurnID string
34 Task string
35 Err error
36 Summary string
37}
38
39// ResidentChild serializes prompt execution through one control goroutine.
40// No mutex is held while a runner performs provider or tool I/O. When a
41// journal is configured, every state boundary is committed by that goroutine.
42type ResidentChild struct {
43 runner ResidentTurnRunner
44 journal *ResidentJournal
45 workspace WorkspaceHandle
46 spec ResidentChildSpec
47 onCompletion func(ResidentCompletion)
48 onUpdate func(historyChanged bool)
49 onStateChange func(ResidentState)
50
51 ctx context.Context
52 cancel context.CancelFunc
53 inbox chan residentPrompt
54 done chan struct{}
55 once sync.Once
56
57 mu sync.RWMutex
58 state ResidentState
59 stateUpdatedAt time.Time
60 turnStartedAt time.Time
61 activityUpdatedAt time.Time
62 live *residentLiveProjection
63 cleanupWorkspace bool
64}
65
66func NewResidentChild(runner ResidentTurnRunner) *ResidentChild {
67 return newResidentChild(ResidentChildSpec{}, nil, runner)
68}
69
70func newJournaledResidentChild(spec ResidentChildSpec, journal *ResidentJournal, runner ResidentTurnRunner, onCompletion func(ResidentCompletion)) *ResidentChild {
71 child := newResidentChildWithWorkspace(spec, journal, nil, runner)
72 child.onCompletion = onCompletion
73 return child
74}
75
76func newJournaledResidentChildWithWorkspace(spec ResidentChildSpec, journal *ResidentJournal, workspace WorkspaceHandle, runner ResidentTurnRunner, onCompletion func(ResidentCompletion)) *ResidentChild {
77 child := newResidentChildWithWorkspace(spec, journal, workspace, runner)
78 child.onCompletion = onCompletion
79 return child
80}
81
82func newResidentChild(spec ResidentChildSpec, journal *ResidentJournal, runner ResidentTurnRunner) *ResidentChild {
83 return newResidentChildWithWorkspace(spec, journal, nil, runner)
84}
85
86func newResidentChildWithWorkspace(spec ResidentChildSpec, journal *ResidentJournal, workspace WorkspaceHandle, runner ResidentTurnRunner) *ResidentChild {
87 ctx, cancel := context.WithCancel(context.Background())
88 now := time.Now().UTC()
89 child := &ResidentChild{
90 runner: runner,
91 journal: journal,
92 workspace: workspace,
93 spec: spec,
94 ctx: ctx,
95 cancel: cancel,
96 inbox: make(chan residentPrompt),
97 done: make(chan struct{}),
98 state: ResidentQueued,
99 stateUpdatedAt: now,
100 activityUpdatedAt: now,
101 live: newResidentLiveProjection(),
102 }
103 if journal != nil {
104 child.live.SeedUsage(journal.usageSnapshot())
105 journal.SetEventObserver(func(event core.AgentEvent) {
106 child.live.Apply(event)
107 child.recordActivity()
108 child.notifyUpdate(residentEventChangesHistory(event))
109 })
110 }
111 go child.run()
112 return child
113}
114
115// setUpdateObserver receives state and live-projection changes after they are
116// visible to readers. historyChanged is true only after a finalized event was
117// appended to the resident transcript. It must not block resident execution.
118func (c *ResidentChild) setUpdateObserver(observer func(historyChanged bool)) {
119 if c == nil {
120 return
121 }
122 c.mu.Lock()
123 c.onUpdate = observer
124 c.mu.Unlock()
125}
126
127// setStateObserver receives lifecycle transitions after the child state lock
128// has been released. It is used by the manager to maintain cheap activity
129// notifications for independent UI animation.
130func (c *ResidentChild) setStateObserver(observer func(ResidentState)) {
131 if c == nil {
132 return
133 }
134 c.mu.Lock()
135 c.onStateChange = observer
136 c.mu.Unlock()
137}
138
139// Live returns an immutable copy of the unfinished visible turn.
140func (c *ResidentChild) Live() ResidentLiveSnapshot {
141 if c == nil {
142 return ResidentLiveSnapshot{}
143 }
144 return c.live.Snapshot()
145}
146
147func (c *ResidentChild) State() ResidentState {
148 if c == nil {
149 return ResidentStopped
150 }
151 c.mu.RLock()
152 defer c.mu.RUnlock()
153 return c.state
154}
155
156func (c *ResidentChild) StateUpdatedAt() time.Time {
157 if c == nil {
158 return time.Time{}
159 }
160 c.mu.RLock()
161 defer c.mu.RUnlock()
162 return c.stateUpdatedAt
163}
164
165// TurnStartedAt is the start time of the active or most recent resident turn.
166func (c *ResidentChild) TurnStartedAt() time.Time {
167 if c == nil {
168 return time.Time{}
169 }
170 c.mu.RLock()
171 defer c.mu.RUnlock()
172 return c.turnStartedAt
173}
174
175// ActivityUpdatedAt is refreshed by visible turn events so callers can show
176// how long the child has been waiting for its next observable action.
177func (c *ResidentChild) ActivityUpdatedAt() time.Time {
178 if c == nil {
179 return time.Time{}
180 }
181 c.mu.RLock()
182 defer c.mu.RUnlock()
183 return c.activityUpdatedAt
184}
185
186func (c *ResidentChild) setState(state ResidentState) {
187 c.mu.Lock()
188 now := time.Now().UTC()
189 c.state = state
190 c.stateUpdatedAt = now
191 c.activityUpdatedAt = now
192 observer := c.onUpdate
193 stateObserver := c.onStateChange
194 c.mu.Unlock()
195 if stateObserver != nil {
196 stateObserver(state)
197 }
198 if observer != nil {
199 observer(false)
200 }
201}
202
203func (c *ResidentChild) startTurn(turnID string) {
204 if c == nil {
205 return
206 }
207 c.live.Start(turnID)
208 c.mu.Lock()
209 now := time.Now().UTC()
210 c.state = ResidentRunning
211 c.stateUpdatedAt = now
212 c.turnStartedAt = now
213 c.activityUpdatedAt = now
214 observer := c.onUpdate
215 stateObserver := c.onStateChange
216 c.mu.Unlock()
217 if stateObserver != nil {
218 stateObserver(ResidentRunning)
219 }
220 if observer != nil {
221 observer(false)
222 }
223}
224
225func (c *ResidentChild) recordActivity() {
226 if c == nil {
227 return
228 }
229 c.mu.Lock()
230 c.activityUpdatedAt = time.Now().UTC()
231 c.mu.Unlock()
232}
233
234func (c *ResidentChild) notifyUpdate(historyChanged bool) {
235 if c == nil {
236 return
237 }
238 c.mu.RLock()
239 observer := c.onUpdate
240 c.mu.RUnlock()
241 if observer != nil {
242 observer(historyChanged)
243 }
244}
245
246func residentEventChangesHistory(event core.AgentEvent) bool {
247 switch event.(type) {
248 case core.EvUserMessage, core.EvAssistantMessage, core.EvToolCall, core.EvToolResult:
249 return true
250 default:
251 return false
252 }
253}
254
255// Resume queues a prompt for the legacy in-memory construction seam. Manager
256// callers must use resumeAccepted after the journal records acceptance.
257func (c *ResidentChild) Resume(ctx context.Context, prompt string) error {
258 return c.enqueue(ctx, residentPrompt{turnID: uuid.NewString(), prompt: prompt, ack: make(chan error, 1)})
259}
260
261// resumeAccepted queues durably accepted work and reports whether the child
262// took ownership of its terminal completion. Once it has taken ownership, it
263// emits the completion itself even when turn startup fails.
264func (c *ResidentChild) resumeAccepted(ctx context.Context, turnID, prompt string) (bool, error) {
265 return c.enqueueAccepted(ctx, residentPrompt{turnID: turnID, prompt: prompt, ack: make(chan error, 1)})
266}
267
268func (c *ResidentChild) enqueue(ctx context.Context, request residentPrompt) error {
269 _, err := c.enqueueAccepted(ctx, request)
270 return err
271}
272
273func (c *ResidentChild) enqueueAccepted(ctx context.Context, request residentPrompt) (bool, error) {
274 if c == nil || c.runner == nil {
275 return false, errors.New("resident child: no runner")
276 }
277 if ctx == nil {
278 ctx = context.Background()
279 }
280 request.prompt = strings.TrimSpace(request.prompt)
281 if request.prompt == "" {
282 return false, errors.New("resident child: prompt is empty")
283 }
284 if strings.TrimSpace(request.turnID) == "" {
285 return false, errors.New("resident child: missing turn ID")
286 }
287 select {
288 case <-ctx.Done():
289 return false, ctx.Err()
290 case <-c.ctx.Done():
291 return false, errors.New("resident child: closed")
292 case c.inbox <- request:
293 }
294 select {
295 case err := <-request.ack:
296 return true, err
297 case <-c.ctx.Done():
298 return true, errors.New("resident child: closed before prompt acceptance")
299 }
300}
301
302func (c *ResidentChild) Close(ctx context.Context) error {
303 if c == nil {
304 return nil
305 }
306 if ctx == nil {
307 ctx = context.Background()
308 }
309 c.once.Do(func() {
310 c.mu.Lock()
311 c.cleanupWorkspace = c.state == ResidentIdle || c.state == ResidentCompleted
312 c.mu.Unlock()
313 c.cancel()
314 })
315 select {
316 case <-ctx.Done():
317 return ctx.Err()
318 case <-c.done:
319 return nil
320 }
321}
322
323func (c *ResidentChild) run() {
324 defer close(c.done)
325 defer func() {
326 if c.journal != nil {
327 _ = c.journal.Close()
328 }
329 c.mu.RLock()
330 cleanupWorkspace := c.cleanupWorkspace
331 c.mu.RUnlock()
332 if c.workspace != nil && c.workspace.Mode() == WorkspaceWorktree && cleanupWorkspace {
333 _ = c.workspace.Cleanup(context.Background())
334 }
335 }()
336
337 var queue []residentPrompt
338 var interruptedPrompts []residentPrompt
339 results := make(chan residentTurnResult, 1)
340 var active residentPrompt
341 running, interrupted := false, false
342 for {
343 if !interrupted && !running && len(queue) > 0 {
344 active, queue = queue[0], queue[1:]
345 if c.journal != nil {
346 if err := c.journal.RecordTurnStarted(c.spec, active.turnID); err != nil {
347 terminalErr := fmt.Errorf("persist resident child start state: %w", err)
348 c.live.Finish(ResidentFailed)
349 c.setState(ResidentFailed)
350 c.cancel()
351 if c.onCompletion != nil {
352 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr})
353 for _, pending := range queue {
354 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr})
355 }
356 }
357 return
358 }
359 }
360 c.startTurn(active.turnID)
361 running = true
362 go func(request residentPrompt) {
363 results <- residentTurnResult{turnID: request.turnID, err: c.runner(c.ctx, request.prompt)}
364 }(active)
365 }
366
367 if interrupted && !running {
368 c.live.Finish(ResidentInterrupted)
369 c.setState(ResidentInterrupted)
370 if c.onCompletion != nil {
371 for _, prompt := range interruptedPrompts {
372 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: prompt.turnID, Task: prompt.prompt, Err: context.Canceled})
373 }
374 }
375 return
376 }
377
378 canceled := c.ctx.Done()
379 if interrupted {
380 canceled = nil
381 }
382 select {
383 case <-canceled:
384 if interrupted {
385 continue
386 }
387 interrupted = true
388 if running {
389 interruptedPrompts = append(interruptedPrompts, active)
390 if c.journal != nil {
391 _ = c.journal.RecordTurnInterrupted(c.spec, active.turnID)
392 }
393 }
394 for _, pending := range queue {
395 interruptedPrompts = append(interruptedPrompts, pending)
396 if c.journal != nil {
397 _ = c.journal.RecordTurnInterrupted(c.spec, pending.turnID)
398 }
399 }
400 queue = nil
401 case request := <-c.inbox:
402 if interrupted {
403 request.ack <- errors.New("resident child: closed")
404 continue
405 }
406 queue = append(queue, request)
407 request.ack <- nil
408 case result := <-results:
409 running = false
410 if interrupted {
411 continue
412 }
413 terminalErr := result.err
414 var capture *WorkspaceCapture
415 if (terminalErr == nil || errors.Is(terminalErr, ErrBudgetExceeded)) && c.workspace != nil && c.workspace.Mode() == WorkspaceWorktree {
416 captured, err := c.workspace.Capture(c.ctx)
417 if err != nil {
418 terminalErr = errors.Join(terminalErr, fmt.Errorf("capture resident worktree: %w", err))
419 } else {
420 capture = &captured
421 }
422 }
423 state := ResidentIdle
424 if terminalErr != nil {
425 state = ResidentFailed
426 if errors.Is(terminalErr, ErrBudgetExceeded) {
427 state = ResidentBudgetExhausted
428 }
429 }
430 summary := ""
431 persistenceFailed := false
432 if c.journal != nil {
433 projection, err := c.journal.finishTurn(c.spec, result.turnID, terminalErr, capture)
434 if err != nil {
435 terminalErr = fmt.Errorf("persist resident child terminal state: %w", err)
436 state, persistenceFailed = ResidentFailed, true
437 } else {
438 state, summary = projection.State, projection.Summary
439 if projection.Handoff != "" {
440 summary = projection.Handoff
441 }
442 }
443 }
444 c.live.Finish(state)
445 c.setState(state)
446 if c.onCompletion != nil {
447 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary})
448 if persistenceFailed {
449 for _, pending := range queue {
450 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr})
451 }
452 }
453 }
454 if persistenceFailed {
455 return
456 }
457 }
458 }
459}