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