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