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// ResidentRuntime owns one child's agent. The child owns turn ordering and
16// cancellation; the runtime owns agent/provider work.
17//
18// Run executes one accepted turn. Steer injects a follow-up into the turn Run
19// is executing and reports false when no turn can still receive it; a true
20// result guarantees Run delivers the follow-up before it returns successfully.
21// PendingSteers counts follow-ups not yet delivered, and DrainSteers removes
22// and returns them after a turn ended without delivering them.
23type ResidentRuntime interface {
24 Run(ctx context.Context, prompt string) error
25 Steer(prompt string) bool
26 PendingSteers() int
27 DrainSteers() []string
28}
29
30// ResidentTurnRunner is a runtime that runs turns but cannot be steered, so
31// every follow-up becomes a separate queued turn.
32type ResidentTurnRunner func(context.Context, string) error
33
34func (r ResidentTurnRunner) Run(ctx context.Context, prompt string) error { return r(ctx, prompt) }
35func (ResidentTurnRunner) Steer(string) bool { return false }
36func (ResidentTurnRunner) PendingSteers() int { return 0 }
37func (ResidentTurnRunner) DrainSteers() []string { return nil }
38
39type residentPrompt struct {
40 turnID string
41 prompt string
42 ack chan error
43}
44
45type residentControlKind int
46
47const (
48 residentControlSteer residentControlKind = iota + 1
49 residentControlInterrupt
50)
51
52// residentControl is a request handled by the control goroutine, so it is
53// ordered with turn start and completion.
54type residentControl struct {
55 kind residentControlKind
56 prompt string
57 // onSteered runs on the control goroutine before the active turn can
58 // report its completion, so a caller can subscribe without a race.
59 onSteered func(turnID string)
60 reply chan residentControlReply
61}
62
63type residentControlReply struct {
64 ok bool
65 turnID string
66}
67
68type residentTurnResult struct {
69 turnID string
70 err error
71}
72
73// ResidentCompletion is a typed in-process terminal turn notification.
74type ResidentCompletion struct {
75 ChildID string
76 TurnID string
77 Task string
78 Err error
79 Summary string
80 // HostUpdate is set by the manager after the host completion observer
81 // accepts this outcome. Tool waits then omit the full report.
82 HostUpdate bool
83 // NotStarted marks a queued follow-up that was dropped before the child
84 // ever ran it, for example because the child was stopped or interrupted.
85 NotStarted bool
86 // Undelivered lists follow-ups steered into this turn that the child never
87 // read because the turn ended first.
88 Undelivered []string
89}
90
91// ResidentChild serializes prompt execution through one control goroutine.
92// No mutex is held while a runner performs provider or tool I/O. When a
93// journal is configured, every state boundary is committed by that goroutine.
94type ResidentChild struct {
95 runtime ResidentRuntime
96 journal *ResidentJournal
97 workspace WorkspaceHandle
98 spec ResidentChildSpec
99 onCompletion func(ResidentCompletion)
100 onUpdate func(historyChanged bool)
101 onStateChange func(ResidentState)
102
103 ctx context.Context
104 cancel context.CancelFunc
105 inbox chan residentPrompt
106 control chan residentControl
107 done chan struct{}
108 once sync.Once
109
110 mu sync.RWMutex
111 state ResidentState
112 stateUpdatedAt time.Time
113 turnStartedAt time.Time
114 activityUpdatedAt time.Time
115 live *residentLiveProjection
116 cleanupWorkspace bool
117 queuedTurns int
118}
119
120func NewResidentChild(runtime ResidentRuntime) *ResidentChild {
121 return newResidentChild(ResidentChildSpec{}, nil, runtime)
122}
123
124func newJournaledResidentChild(spec ResidentChildSpec, journal *ResidentJournal, runtime ResidentRuntime, onCompletion func(ResidentCompletion)) *ResidentChild {
125 child := newResidentChildWithWorkspace(spec, journal, nil, runtime)
126 child.onCompletion = onCompletion
127 return child
128}
129
130func newJournaledResidentChildWithWorkspace(spec ResidentChildSpec, journal *ResidentJournal, workspace WorkspaceHandle, runtime ResidentRuntime, onCompletion func(ResidentCompletion)) *ResidentChild {
131 child := newResidentChildWithWorkspace(spec, journal, workspace, runtime)
132 child.onCompletion = onCompletion
133 return child
134}
135
136func newResidentChild(spec ResidentChildSpec, journal *ResidentJournal, runtime ResidentRuntime) *ResidentChild {
137 return newResidentChildWithWorkspace(spec, journal, nil, runtime)
138}
139
140func newResidentChildWithWorkspace(spec ResidentChildSpec, journal *ResidentJournal, workspace WorkspaceHandle, runtime ResidentRuntime) *ResidentChild {
141 ctx, cancel := context.WithCancel(context.Background())
142 now := time.Now().UTC()
143 child := &ResidentChild{
144 runtime: runtime,
145 journal: journal,
146 workspace: workspace,
147 spec: spec,
148 ctx: ctx,
149 cancel: cancel,
150 inbox: make(chan residentPrompt),
151 control: make(chan residentControl),
152 done: make(chan struct{}),
153 state: ResidentQueued,
154 stateUpdatedAt: now,
155 activityUpdatedAt: now,
156 live: newResidentLiveProjection(),
157 }
158 if journal != nil {
159 child.live.SeedUsage(journal.usageSnapshot())
160 journal.SetEventObserver(func(event core.AgentEvent) {
161 child.live.Apply(event)
162 child.recordActivity()
163 child.notifyUpdate(residentEventChangesHistory(event))
164 })
165 }
166 go child.run()
167 return child
168}
169
170// setUpdateObserver receives state and live-projection changes after they are
171// visible to readers. historyChanged is true only after a finalized event was
172// appended to the resident transcript. It must not block resident execution.
173func (c *ResidentChild) setUpdateObserver(observer func(historyChanged bool)) {
174 if c == nil {
175 return
176 }
177 c.mu.Lock()
178 c.onUpdate = observer
179 c.mu.Unlock()
180}
181
182// setStateObserver receives lifecycle transitions after the child state lock
183// has been released. It is used by the manager to maintain cheap activity
184// notifications for independent UI animation.
185func (c *ResidentChild) setStateObserver(observer func(ResidentState)) {
186 if c == nil {
187 return
188 }
189 c.mu.Lock()
190 c.onStateChange = observer
191 c.mu.Unlock()
192}
193
194// Live returns an immutable copy of the unfinished visible turn.
195func (c *ResidentChild) Live() ResidentLiveSnapshot {
196 if c == nil {
197 return ResidentLiveSnapshot{}
198 }
199 return c.live.Snapshot()
200}
201
202func (c *ResidentChild) State() ResidentState {
203 if c == nil {
204 return ResidentStopped
205 }
206 c.mu.RLock()
207 defer c.mu.RUnlock()
208 return c.state
209}
210
211func (c *ResidentChild) StateUpdatedAt() time.Time {
212 if c == nil {
213 return time.Time{}
214 }
215 c.mu.RLock()
216 defer c.mu.RUnlock()
217 return c.stateUpdatedAt
218}
219
220// TurnStartedAt is the start time of the active or most recent resident turn.
221func (c *ResidentChild) TurnStartedAt() time.Time {
222 if c == nil {
223 return time.Time{}
224 }
225 c.mu.RLock()
226 defer c.mu.RUnlock()
227 return c.turnStartedAt
228}
229
230// ActivityUpdatedAt is refreshed by visible turn events so callers can show
231// how long the child has been waiting for its next observable action.
232func (c *ResidentChild) ActivityUpdatedAt() time.Time {
233 if c == nil {
234 return time.Time{}
235 }
236 c.mu.RLock()
237 defer c.mu.RUnlock()
238 return c.activityUpdatedAt
239}
240
241// PendingFollowUps counts follow-ups the child has accepted but not yet read:
242// queued turns plus messages steered into the running turn.
243func (c *ResidentChild) PendingFollowUps() int {
244 if c == nil {
245 return 0
246 }
247 c.mu.RLock()
248 queued := c.queuedTurns
249 c.mu.RUnlock()
250 if c.runtime != nil {
251 queued += c.runtime.PendingSteers()
252 }
253 return queued
254}
255
256func (c *ResidentChild) setQueuedTurns(n int) {
257 c.mu.Lock()
258 c.queuedTurns = n
259 c.mu.Unlock()
260}
261
262func (c *ResidentChild) setState(state ResidentState) {
263 c.mu.Lock()
264 now := time.Now().UTC()
265 c.state = state
266 c.stateUpdatedAt = now
267 c.activityUpdatedAt = now
268 observer := c.onUpdate
269 stateObserver := c.onStateChange
270 c.mu.Unlock()
271 if stateObserver != nil {
272 stateObserver(state)
273 }
274 if observer != nil {
275 observer(false)
276 }
277}
278
279func (c *ResidentChild) startTurn(turnID string) {
280 if c == nil {
281 return
282 }
283 c.live.Start(turnID)
284 c.mu.Lock()
285 now := time.Now().UTC()
286 c.state = ResidentRunning
287 c.stateUpdatedAt = now
288 c.turnStartedAt = now
289 c.activityUpdatedAt = now
290 observer := c.onUpdate
291 stateObserver := c.onStateChange
292 c.mu.Unlock()
293 if stateObserver != nil {
294 stateObserver(ResidentRunning)
295 }
296 if observer != nil {
297 observer(false)
298 }
299}
300
301func (c *ResidentChild) recordActivity() {
302 if c == nil {
303 return
304 }
305 c.mu.Lock()
306 c.activityUpdatedAt = time.Now().UTC()
307 c.mu.Unlock()
308}
309
310func (c *ResidentChild) notifyUpdate(historyChanged bool) {
311 if c == nil {
312 return
313 }
314 c.mu.RLock()
315 observer := c.onUpdate
316 c.mu.RUnlock()
317 if observer != nil {
318 observer(historyChanged)
319 }
320}
321
322func residentEventChangesHistory(event core.AgentEvent) bool {
323 switch event.(type) {
324 case core.EvUserMessage, core.EvAssistantMessage, core.EvToolCall, core.EvToolResult:
325 return true
326 default:
327 return false
328 }
329}
330
331// Resume queues a prompt for the legacy in-memory construction seam. Manager
332// callers must use resumeAccepted after the journal records acceptance.
333func (c *ResidentChild) Resume(ctx context.Context, prompt string) error {
334 return c.enqueue(ctx, residentPrompt{turnID: uuid.NewString(), prompt: prompt, ack: make(chan error, 1)})
335}
336
337// resumeAccepted queues durably accepted work and reports whether the child
338// took ownership of its terminal completion. Once it has taken ownership, it
339// emits the completion itself even when turn startup fails.
340func (c *ResidentChild) resumeAccepted(ctx context.Context, turnID, prompt string) (bool, error) {
341 return c.enqueueAccepted(ctx, residentPrompt{turnID: turnID, prompt: prompt, ack: make(chan error, 1)})
342}
343
344func (c *ResidentChild) enqueue(ctx context.Context, request residentPrompt) error {
345 _, err := c.enqueueAccepted(ctx, request)
346 return err
347}
348
349// steer injects a follow-up into the running turn. It reports false, leaving
350// the child unchanged, when no turn is executing or earlier follow-ups are
351// still queued; the caller must then queue the follow-up as a new turn.
352func (c *ResidentChild) steer(ctx context.Context, prompt string, onSteered func(turnID string)) (bool, string, error) {
353 reply, err := c.sendControl(ctx, residentControl{kind: residentControlSteer, prompt: strings.TrimSpace(prompt), onSteered: onSteered})
354 return reply.ok, reply.turnID, err
355}
356
357// interruptTurn cancels only the running turn. The child stays alive with its
358// transcript, and queued follow-ups are dropped. It reports false when no turn
359// is running.
360func (c *ResidentChild) interruptTurn(ctx context.Context) (bool, error) {
361 reply, err := c.sendControl(ctx, residentControl{kind: residentControlInterrupt})
362 return reply.ok, err
363}
364
365func (c *ResidentChild) sendControl(ctx context.Context, request residentControl) (residentControlReply, error) {
366 if c == nil || c.runtime == nil {
367 return residentControlReply{}, errors.New("resident child: no runtime")
368 }
369 if ctx == nil {
370 ctx = context.Background()
371 }
372 request.reply = make(chan residentControlReply, 1)
373 select {
374 case <-ctx.Done():
375 return residentControlReply{}, ctx.Err()
376 case <-c.done:
377 return residentControlReply{}, errors.New("resident child: closed")
378 case c.control <- request:
379 }
380 return <-request.reply, nil
381}
382
383func (c *ResidentChild) enqueueAccepted(ctx context.Context, request residentPrompt) (bool, error) {
384 if c == nil || c.runtime == nil {
385 return false, errors.New("resident child: no runner")
386 }
387 if ctx == nil {
388 ctx = context.Background()
389 }
390 request.prompt = strings.TrimSpace(request.prompt)
391 if request.prompt == "" {
392 return false, errors.New("resident child: prompt is empty")
393 }
394 if strings.TrimSpace(request.turnID) == "" {
395 return false, errors.New("resident child: missing turn ID")
396 }
397 select {
398 case <-ctx.Done():
399 return false, ctx.Err()
400 case <-c.ctx.Done():
401 return false, errors.New("resident child: closed")
402 case c.inbox <- request:
403 }
404 select {
405 case err := <-request.ack:
406 return true, err
407 case <-c.ctx.Done():
408 return true, errors.New("resident child: closed before prompt acceptance")
409 }
410}
411
412func (c *ResidentChild) Close(ctx context.Context) error {
413 if c == nil {
414 return nil
415 }
416 if ctx == nil {
417 ctx = context.Background()
418 }
419 c.once.Do(func() {
420 c.mu.Lock()
421 c.cleanupWorkspace = c.state == ResidentIdle || c.state == ResidentCompleted
422 c.mu.Unlock()
423 c.cancel()
424 })
425 select {
426 case <-ctx.Done():
427 return ctx.Err()
428 case <-c.done:
429 return nil
430 }
431}
432
433func (c *ResidentChild) run() {
434 defer close(c.done)
435 defer func() {
436 if c.journal != nil {
437 _ = c.journal.Close()
438 }
439 c.mu.RLock()
440 cleanupWorkspace := c.cleanupWorkspace
441 c.mu.RUnlock()
442 if c.workspace != nil && c.workspace.Mode() == WorkspaceWorktree && cleanupWorkspace {
443 _ = c.workspace.Cleanup(context.Background())
444 }
445 }()
446
447 var queue []residentPrompt
448 var interruptedPrompts []residentPrompt
449 var droppedPrompts []residentPrompt
450 results := make(chan residentTurnResult, 1)
451 var active residentPrompt
452 turnCancel := context.CancelFunc(func() {})
453 defer func() { turnCancel() }()
454 running, interrupted, turnInterrupted := false, false, false
455 // dropQueued reports queued follow-ups that never started. Each was
456 // accepted and tracked, so each still gets exactly one completion, marked
457 // NotStarted so the parent does not mistake it for interrupted work.
458 dropQueued := func(err error) {
459 for _, pending := range queue {
460 if c.journal != nil {
461 _ = c.journal.RecordTurnInterrupted(c.spec, pending.turnID)
462 }
463 if c.onCompletion != nil {
464 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: err, NotStarted: true})
465 }
466 }
467 queue = nil
468 c.setQueuedTurns(0)
469 }
470 for {
471 c.setQueuedTurns(len(queue))
472 if !interrupted && !running && len(queue) > 0 {
473 active, queue = queue[0], queue[1:]
474 c.setQueuedTurns(len(queue))
475 if c.journal != nil {
476 if err := c.journal.RecordTurnStarted(c.spec, active.turnID); err != nil {
477 terminalErr := fmt.Errorf("persist resident child start state: %w", err)
478 c.live.Finish(ResidentFailed)
479 c.setState(ResidentFailed)
480 c.cancel()
481 if c.onCompletion != nil {
482 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr})
483 for _, pending := range queue {
484 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr})
485 }
486 }
487 return
488 }
489 }
490 c.startTurn(active.turnID)
491 running, turnInterrupted = true, false
492 turnCtx, cancelTurn := context.WithCancel(c.ctx)
493 turnCancel = cancelTurn
494 go func(ctx context.Context, request residentPrompt) {
495 results <- residentTurnResult{turnID: request.turnID, err: c.runtime.Run(ctx, request.prompt)}
496 }(turnCtx, active)
497 }
498
499 if interrupted && !running {
500 c.live.Finish(ResidentInterrupted)
501 c.setState(ResidentInterrupted)
502 if c.onCompletion != nil {
503 for _, prompt := range interruptedPrompts {
504 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: prompt.turnID, Task: prompt.prompt, Err: context.Canceled, Undelivered: c.runtime.DrainSteers()})
505 }
506 for _, prompt := range droppedPrompts {
507 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: prompt.turnID, Task: prompt.prompt, Err: context.Canceled, NotStarted: true})
508 }
509 }
510 return
511 }
512
513 canceled := c.ctx.Done()
514 if interrupted {
515 canceled = nil
516 }
517 select {
518 case <-canceled:
519 if interrupted {
520 continue
521 }
522 interrupted = true
523 if running {
524 interruptedPrompts = append(interruptedPrompts, active)
525 if c.journal != nil {
526 _ = c.journal.RecordTurnInterrupted(c.spec, active.turnID)
527 }
528 }
529 for _, pending := range queue {
530 droppedPrompts = append(droppedPrompts, pending)
531 if c.journal != nil {
532 _ = c.journal.RecordTurnInterrupted(c.spec, pending.turnID)
533 }
534 }
535 queue = nil
536 c.setQueuedTurns(0)
537 case request := <-c.control:
538 switch request.kind {
539 case residentControlSteer:
540 // Steering past queued turns would reorder follow-ups, and an
541 // interrupted turn can no longer read new input.
542 if running && !interrupted && !turnInterrupted && len(queue) == 0 && request.prompt != "" && c.runtime.Steer(request.prompt) {
543 if request.onSteered != nil {
544 request.onSteered(active.turnID)
545 }
546 c.notifyUpdate(false)
547 request.reply <- residentControlReply{ok: true, turnID: active.turnID}
548 } else {
549 request.reply <- residentControlReply{}
550 }
551 case residentControlInterrupt:
552 if !running || interrupted || turnInterrupted {
553 request.reply <- residentControlReply{}
554 continue
555 }
556 turnInterrupted = true
557 turnCancel()
558 dropQueued(context.Canceled)
559 request.reply <- residentControlReply{ok: true, turnID: active.turnID}
560 }
561 case request := <-c.inbox:
562 if interrupted {
563 request.ack <- errors.New("resident child: closed")
564 continue
565 }
566 queue = append(queue, request)
567 c.setQueuedTurns(len(queue))
568 request.ack <- nil
569 case result := <-results:
570 running = false
571 turnCancel()
572 if interrupted {
573 continue
574 }
575 terminalErr := result.err
576 if turnInterrupted && terminalErr == nil {
577 // The turn finished before cancellation reached it; the
578 // manager still asked for an interruption.
579 terminalErr = context.Canceled
580 }
581 undelivered := c.runtime.DrainSteers()
582 var capture *WorkspaceCapture
583 if (terminalErr == nil || !errors.Is(terminalErr, context.Canceled)) && c.workspace != nil && c.workspace.Mode() == WorkspaceWorktree {
584 captured, err := c.workspace.Capture(c.ctx)
585 if err != nil {
586 terminalErr = errors.Join(terminalErr, fmt.Errorf("capture resident worktree: %w", err))
587 } else {
588 capture = &captured
589 }
590 }
591 state := ResidentIdle
592 if terminalErr != nil {
593 state = ResidentFailed
594 }
595 summary := ""
596 persistenceFailed := false
597 if c.journal != nil {
598 projection, err := c.journal.finishTurn(c.spec, result.turnID, terminalErr, capture)
599 if err != nil {
600 terminalErr = fmt.Errorf("persist resident child terminal state: %w", err)
601 state, persistenceFailed = ResidentFailed, true
602 } else {
603 state, summary = projection.State, projection.Summary
604 if projection.Handoff != "" {
605 summary = projection.Handoff
606 }
607 }
608 }
609 c.live.Finish(state)
610 c.setState(state)
611 if c.onCompletion != nil {
612 c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary, Undelivered: undelivered})
613 }
614 if persistenceFailed {
615 dropQueued(terminalErr)
616 return
617 }
618 }
619 }
620}