1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "sort"
10 "strings"
11 "sync"
12 "time"
13
14 "github.com/bnema/zut/packages/provider"
15 "github.com/google/uuid"
16)
17
18// ResidentFactory is the host-owned construction boundary. It receives only
19// an already-resolved non-secret specification and returns an in-process
20// runtime; it never starts another zut binary.
21type ResidentFactory func(ResidentChildSpec, *ResidentJournal) (ResidentRuntime, error)
22
23type ResidentManager struct {
24 root string
25 factory ResidentFactory
26 prepareWorkspace func(context.Context, WorkspaceRequest) (WorkspaceHandle, error)
27 mu sync.Mutex
28 lifecycleMu sync.Mutex
29 closed bool
30 children map[string]*ResidentChild
31 recovered map[string]ResidentSnapshot
32 recoveredSpec map[string]ResidentChildSpec
33 pending map[string]struct{}
34 activeChildren map[string]struct{}
35 scheduler *ResidentScheduler
36 queueTimeout time.Duration
37 allowedRoots []string
38 dispatchMu sync.Mutex
39 onCompletion func(ResidentCompletion) bool
40 completionWaiters map[string][]chan ResidentCompletion
41 onAccepted func(ResidentChildSpec, string, string)
42 onUpdate func(string)
43 onHistoryUpdate func(string)
44 onActivity func(bool)
45}
46
47// SetCompletionObserver installs the host report channel. Return true only
48// when the host accepts the full report; false keeps tool waits self-contained.
49func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion) bool) {
50 if m == nil {
51 return
52 }
53 m.mu.Lock()
54 m.onCompletion = observer
55 m.mu.Unlock()
56}
57
58// WatchCompletion registers for one accepted resident turn. The caller must
59// register before making the turn visible, then call the returned cancel
60// function if it stops waiting before completion.
61func (m *ResidentManager) WatchCompletion(childID, turnID string) (<-chan ResidentCompletion, func()) {
62 result := make(chan ResidentCompletion, 1)
63 key := completionKey(childID, turnID)
64 if m == nil || key == "" {
65 close(result)
66 return result, func() {}
67 }
68 m.mu.Lock()
69 if m.closed {
70 m.mu.Unlock()
71 close(result)
72 return result, func() {}
73 }
74 if m.completionWaiters == nil {
75 m.completionWaiters = make(map[string][]chan ResidentCompletion)
76 }
77 m.completionWaiters[key] = append(m.completionWaiters[key], result)
78 m.mu.Unlock()
79 return result, func() {
80 m.mu.Lock()
81 waiters := m.completionWaiters[key]
82 for i, waiter := range waiters {
83 if waiter == result {
84 waiters = append(waiters[:i], waiters[i+1:]...)
85 break
86 }
87 }
88 if len(waiters) == 0 {
89 delete(m.completionWaiters, key)
90 } else {
91 m.completionWaiters[key] = waiters
92 }
93 m.mu.Unlock()
94 }
95}
96
97func (m *ResidentManager) reportCompletion(completion ResidentCompletion) {
98 if m == nil {
99 return
100 }
101 key := completionKey(completion.ChildID, completion.TurnID)
102 m.mu.Lock()
103 waiters := m.completionWaiters[key]
104 delete(m.completionWaiters, key)
105 observer := m.onCompletion
106 m.mu.Unlock()
107 // The host must accept the report before a tool wait can omit its copy.
108 // Observers run without manager locks, as before.
109 if observer != nil {
110 completion.HostUpdate = observer(completion)
111 }
112 for _, waiter := range waiters {
113 waiter <- completion
114 close(waiter)
115 }
116}
117
118// SetAcceptedObserver receives an accepted turn's ID and prompt only after
119// the authoritative child.accepted record has synchronized and before it can
120// be scheduled.
121func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string, string)) {
122 if m == nil {
123 return
124 }
125 m.mu.Lock()
126 m.onAccepted = observer
127 m.mu.Unlock()
128}
129
130// SetUpdateObserver receives child IDs whenever their visible state or live
131// projection changes. Observers run without manager or child locks and must
132// return promptly.
133func (m *ResidentManager) SetUpdateObserver(observer func(string)) {
134 if m == nil {
135 return
136 }
137 m.mu.Lock()
138 m.onUpdate = observer
139 m.mu.Unlock()
140}
141
142// SetHistoryUpdateObserver receives a child ID after a finalized event is
143// durably appended to its transcript. Stream deltas and state changes do not
144// trigger it, so observers can safely schedule bounded history reloads.
145func (m *ResidentManager) SetHistoryUpdateObserver(observer func(string)) {
146 if m == nil {
147 return
148 }
149 m.mu.Lock()
150 m.onHistoryUpdate = observer
151 m.mu.Unlock()
152}
153
154// SetActivityObserver receives the aggregate running-child state immediately
155// and after each lifecycle transition. It does not receive stream events, so
156// UI tickers can consume it without polling resident state.
157func (m *ResidentManager) SetActivityObserver(observer func(active bool)) {
158 if m == nil {
159 return
160 }
161 m.mu.Lock()
162 m.onActivity = observer
163 active := len(m.activeChildren) > 0
164 m.mu.Unlock()
165 if observer != nil {
166 observer(active)
167 }
168}
169
170func (m *ResidentManager) notifyActivity(childID string, state ResidentState) {
171 if m == nil {
172 return
173 }
174 activeChild := state == ResidentRunning
175 m.mu.Lock()
176 if m.activeChildren == nil {
177 m.activeChildren = make(map[string]struct{})
178 }
179 _, wasActive := m.activeChildren[childID]
180 if activeChild {
181 m.activeChildren[childID] = struct{}{}
182 } else {
183 delete(m.activeChildren, childID)
184 }
185 observer := m.onActivity
186 active := len(m.activeChildren) > 0
187 m.mu.Unlock()
188 if wasActive != activeChild && observer != nil {
189 observer(active)
190 }
191}
192
193func (m *ResidentManager) notifyUpdate(childID string, historyChanged bool) {
194 if m == nil {
195 return
196 }
197 m.mu.Lock()
198 observer := m.onUpdate
199 historyObserver := m.onHistoryUpdate
200 m.mu.Unlock()
201 if observer != nil {
202 observer(childID)
203 }
204 if historyChanged && historyObserver != nil {
205 historyObserver(childID)
206 }
207}
208
209// ResidentSnapshot is the manager's bounded public state projection. It
210// deliberately omits prompts, provider credentials, and filesystem paths.
211type ResidentSnapshot struct {
212 ID string
213 State ResidentState
214 Profile string
215 Provider string
216 Model string
217 WorkspaceMode WorkspaceMode
218 Required bool
219 UpdatedAt time.Time
220 TurnStartedAt time.Time
221 ActivityUpdatedAt time.Time
222 WaitingForModel bool
223 PendingFollowUps int
224 OwnedElsewhere bool
225 Usage provider.Usage
226 ContextUsed int
227 ContextMax int
228 Subscription bool
229}
230
231func residentSnapshot(child *ResidentChild) ResidentSnapshot {
232 if child == nil {
233 return ResidentSnapshot{}
234 }
235 live := child.Live()
236 return ResidentSnapshot{ID: child.spec.ID, State: child.State(), Profile: child.spec.Profile,
237 Provider: child.spec.Provider, Model: child.spec.Model,
238 WorkspaceMode: child.spec.WorkspaceMode, Required: child.spec.Required,
239 UpdatedAt: child.StateUpdatedAt(), TurnStartedAt: child.TurnStartedAt(),
240 ActivityUpdatedAt: child.ActivityUpdatedAt(), WaitingForModel: live.WaitingForModel, PendingFollowUps: child.PendingFollowUps(),
241 Usage: live.Usage, ContextUsed: live.ContextUsed, ContextMax: live.ContextMax, Subscription: live.Subscription}
242}
243
244func NewResidentManager(root string, factory ResidentFactory) *ResidentManager {
245 return NewResidentManagerWithLimit(root, 0, factory)
246}
247
248func NewResidentManagerWithLimit(root string, limit int, factory ResidentFactory) *ResidentManager {
249 return NewResidentManagerWithWorkspace(root, limit, PrepareWorkspace, factory)
250}
251
252// NewResidentManagerWithPolicy applies the resident-only host policy. A zero
253// queue timeout leaves accepted work queued until a scheduler slot is free.
254func NewResidentManagerWithPolicy(root string, policy SubagentPolicy, factory ResidentFactory) *ResidentManager {
255 return newResidentManager(root, policy, PrepareWorkspace, factory)
256}
257
258// NewResidentManagerWithWorkspace exposes the narrow allocation seam needed
259// to make workspace preparation part of spawn's durable commit boundary.
260func NewResidentManagerWithWorkspace(root string, limit int, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
261 return newResidentManager(root, SubagentPolicy{MaxConcurrent: limit}, prepare, factory)
262}
263
264func newResidentManager(root string, policy SubagentPolicy, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
265 if prepare == nil {
266 prepare = PrepareWorkspace
267 }
268 return &ResidentManager{
269 root: root,
270 factory: factory,
271 prepareWorkspace: prepare,
272 children: make(map[string]*ResidentChild),
273 recovered: make(map[string]ResidentSnapshot),
274 recoveredSpec: make(map[string]ResidentChildSpec),
275 pending: make(map[string]struct{}),
276 activeChildren: make(map[string]struct{}),
277 completionWaiters: make(map[string][]chan ResidentCompletion),
278 scheduler: NewResidentScheduler(policy.MaxConcurrent),
279 queueTimeout: policy.QueueTimeout,
280 allowedRoots: append([]string(nil), policy.AllowedRoots...),
281 }
282}
283
284// Spawn commits acceptance before publishing a child or starting its runner.
285func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, task string) (*ResidentChild, error) {
286 if m == nil || m.factory == nil {
287 return nil, errors.New("resident manager: no factory")
288 }
289 m.lifecycleMu.Lock()
290 defer m.lifecycleMu.Unlock()
291 m.mu.Lock()
292 if m.closed {
293 m.mu.Unlock()
294 return nil, errors.New("resident manager: closed")
295 }
296 if _, exists := m.children[spec.ID]; exists {
297 m.mu.Unlock()
298 return nil, errors.New("resident manager: duplicate child ID")
299 }
300 if _, exists := m.recovered[spec.ID]; exists {
301 m.mu.Unlock()
302 return nil, errors.New("resident manager: child journal already exists")
303 }
304 if _, exists := m.pending[spec.ID]; exists {
305 m.mu.Unlock()
306 return nil, errors.New("resident manager: child acceptance pending")
307 }
308 m.pending[spec.ID] = struct{}{}
309 m.mu.Unlock()
310 defer func() { m.mu.Lock(); delete(m.pending, spec.ID); m.mu.Unlock() }()
311 if spec.InitialTurnID == "" {
312 spec.InitialTurnID = uuid.NewString()
313 }
314 journal, err := OpenResidentJournal(m.root, spec.ID)
315 if err != nil {
316 return nil, err
317 }
318 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
319 if workspaceRoot == "" {
320 workspaceRoot = strings.TrimSpace(spec.Workspace)
321 }
322 if !m.workspaceAllowed(workspaceRoot) {
323 _ = journal.Close()
324 return nil, errors.New("resident manager: workspace is outside allowed roots")
325 }
326 var workspace WorkspaceHandle
327 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
328 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
329 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, spec.ID),
330 AgentID: spec.ID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots,
331 })
332 if err != nil {
333 _ = journal.Close()
334 return nil, fmt.Errorf("resident manager prepare workspace: %w", err)
335 }
336 spec.Workspace = workspace.Dir()
337 spec.WorkspaceMode = workspace.Mode()
338 }
339 if err := journal.Accept(spec, task); err != nil {
340 if workspace != nil {
341 _ = workspace.Cleanup(context.Background())
342 }
343 _ = journal.Close()
344 return nil, err
345 }
346 m.mu.Lock()
347 accepted := m.onAccepted
348 m.mu.Unlock()
349 if accepted != nil {
350 accepted(spec, spec.InitialTurnID, task)
351 }
352 runner, err := m.factory(spec, journal)
353 if err != nil {
354 _ = journal.RecordFailure(spec)
355 _ = journal.Close()
356 m.reportCompletion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
357 return nil, err
358 }
359 child := newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
360 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
361 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
362 m.mu.Lock()
363 if m.closed {
364 m.mu.Unlock()
365 _ = child.Close(context.Background())
366 return nil, errors.New("resident manager: closed")
367 }
368 m.children[spec.ID] = child
369 m.mu.Unlock()
370 if childAccepted, err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
371 m.mu.Lock()
372 if m.children[spec.ID] == child {
373 delete(m.children, spec.ID)
374 }
375 m.mu.Unlock()
376 _ = child.Close(context.Background())
377 if !childAccepted {
378 _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
379 m.reportCompletion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
380 }
381 return nil, err
382 }
383 return child, nil
384}
385
386// scheduledRuntime gates each turn on a global scheduler slot. Steering needs
387// no slot: it only reaches a turn that already holds one.
388type scheduledRuntime struct {
389 ResidentRuntime
390 run ResidentTurnRunner
391}
392
393func (r scheduledRuntime) Run(ctx context.Context, prompt string) error { return r.run(ctx, prompt) }
394
395func (m *ResidentManager) scheduledRunner(childID string, runtime ResidentRuntime) ResidentRuntime {
396 return scheduledRuntime{ResidentRuntime: runtime, run: m.scheduledRun(childID, runtime)}
397}
398
399func (m *ResidentManager) scheduledRun(childID string, runtime ResidentRuntime) ResidentTurnRunner {
400 return func(ctx context.Context, prompt string) error {
401 if ctx == nil {
402 ctx = context.Background()
403 }
404 ticket := m.scheduler.Enqueue(childID, prompt)
405 m.dispatch()
406 var timeout <-chan time.Time
407 var timer *time.Timer
408 if m.queueTimeout > 0 {
409 timer = time.NewTimer(m.queueTimeout)
410 timeout = timer.C
411 defer timer.Stop()
412 }
413 select {
414 case <-ctx.Done():
415 if m.scheduler.Cancel(ticket.Sequence) {
416 m.dispatch()
417 return ctx.Err()
418 }
419 // Admission won the race with cancellation. Consume the handoff
420 // and release its reservation without running provider work.
421 <-ticket.ready
422 _ = m.scheduler.Release(childID)
423 m.dispatch()
424 return ctx.Err()
425 case <-ticket.ready:
426 case <-timeout:
427 if m.scheduler.Cancel(ticket.Sequence) {
428 m.dispatch()
429 return fmt.Errorf("resident manager: queue timeout after %s", m.queueTimeout)
430 }
431 // Admission won the race with the timeout. The reservation is now
432 // owned by this runner and must be released below.
433 <-ticket.ready
434 }
435 defer func() {
436 _ = m.scheduler.Release(childID)
437 m.dispatch()
438 }()
439 return runtime.Run(ctx, prompt)
440 }
441}
442
443func (m *ResidentManager) dispatch() {
444 if m == nil || m.scheduler == nil {
445 return
446 }
447 m.dispatchMu.Lock()
448 defer m.dispatchMu.Unlock()
449 for {
450 if _, ok := m.scheduler.Admit(); !ok {
451 return
452 }
453 }
454}
455
456// Resume synchronizes a follow-up acceptance before making it visible to the
457// resident child. A disk-only child is rebuilt only for this explicit new
458// prompt; reconciliation itself never constructs or replays one.
459func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) error {
460 _, err := m.ResumeFollowUp(ctx, childID, prompt, ResumeSteer, uuid.NewString(), nil)
461 return err
462}
463
464// ResumeMode selects how a follow-up reaches a child that is running a turn.
465type ResumeMode string
466
467const (
468 // ResumeSteer injects the follow-up into the running turn at its next
469 // tool or model boundary. An idle child starts a new turn instead.
470 ResumeSteer ResumeMode = "steer"
471 // ResumeQueue runs the follow-up as a separate turn after the current one.
472 ResumeQueue ResumeMode = "queue"
473)
474
475// ResumeOutcome identifies the turn that will answer a follow-up.
476type ResumeOutcome struct {
477 // TurnID is the new turn for a queued follow-up, or the running turn
478 // that received a steered one.
479 TurnID string
480 Steered bool
481}
482
483// ResumeFollowUp delivers a follow-up by mode. A steered follow-up joins the
484// running turn, so it creates no new turn and no new completion. onSteered, when
485// set, runs before that turn can complete so a caller can watch its completion
486// without racing it. When steering is impossible, the follow-up is queued as a
487// new turn named turnID.
488func (m *ResidentManager) ResumeFollowUp(ctx context.Context, childID, prompt string, mode ResumeMode, turnID string, onSteered func(turnID string)) (ResumeOutcome, error) {
489 if mode == "" {
490 mode = ResumeSteer
491 }
492 if mode != ResumeSteer && mode != ResumeQueue {
493 return ResumeOutcome{}, fmt.Errorf("resident manager: unknown resume mode %q", mode)
494 }
495 if mode == ResumeSteer && m != nil && strings.TrimSpace(prompt) != "" {
496 m.lifecycleMu.Lock()
497 child := m.Get(strings.TrimSpace(childID))
498 if child != nil {
499 steered, activeTurn, err := child.steer(ctx, prompt, onSteered)
500 if err == nil && steered {
501 m.lifecycleMu.Unlock()
502 m.notifyUpdate(child.spec.ID, false)
503 return ResumeOutcome{TurnID: activeTurn, Steered: true}, nil
504 }
505 }
506 m.lifecycleMu.Unlock()
507 }
508 if strings.TrimSpace(turnID) == "" {
509 turnID = uuid.NewString()
510 }
511 if err := m.resumeWithTurn(ctx, childID, prompt, turnID); err != nil {
512 return ResumeOutcome{}, err
513 }
514 return ResumeOutcome{TurnID: turnID}, nil
515}
516
517// resumeWithTurn queues a follow-up turn with a caller-owned ID. The ID names the
518// accepted follow-up turn in its completion, so a caller that must observe the
519// outcome can subscribe with WatchCompletion before the turn can finish. The
520// ID must be unique per child: replay rejects a duplicate turn.accepted record,
521// so reusing one makes the journal unreplayable.
522func (m *ResidentManager) resumeWithTurn(ctx context.Context, childID, prompt, turnID string) error {
523 if strings.TrimSpace(turnID) == "" {
524 turnID = uuid.NewString()
525 }
526 if m == nil {
527 return errors.New("resident manager: unavailable")
528 }
529 m.lifecycleMu.Lock()
530 defer m.lifecycleMu.Unlock()
531 childID = strings.TrimSpace(childID)
532 prompt = strings.TrimSpace(prompt)
533 if childID == "" || prompt == "" {
534 return errors.New("resident manager: child ID and prompt are required")
535 }
536 m.mu.Lock()
537 if m.closed {
538 m.mu.Unlock()
539 return errors.New("resident manager: closed")
540 }
541 child := m.children[childID]
542 spec, recovered := m.recoveredSpec[childID]
543 snapshot := m.recovered[childID]
544 // A recovered foreign snapshot is advisory. Its owner may have exited since
545 // Reconcile; OpenResidentJournal below makes the authoritative fresh check.
546 if child == nil && snapshot.OwnedElsewhere {
547 recovered = true
548 }
549 if child == nil && recovered {
550 if _, pending := m.pending[childID]; pending {
551 m.mu.Unlock()
552 return errors.New("resident manager: child resume pending")
553 }
554 m.pending[childID] = struct{}{}
555 }
556 m.mu.Unlock()
557 if child == nil && recovered {
558 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
559 journal, err := OpenResidentJournal(m.root, childID)
560 if err != nil {
561 if errors.Is(err, ErrResidentLeaseBusy) {
562 return errors.New("resident manager: child is owned by another zut process")
563 }
564 return err
565 }
566 metadata, err := reconcileOwnedResidentJournal(journal)
567 if err != nil {
568 _ = journal.Close()
569 return err
570 }
571 journal.ConfigureUsage(metadata.ContextMax, metadata.Subscription)
572 records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
573 if err != nil || len(records) == 0 || records[0].Spec == nil {
574 _ = journal.Close()
575 return errors.New("resident manager: invalid accepted specification")
576 }
577 spec = *records[0].Spec
578 var workspace WorkspaceHandle
579 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
580 if workspaceRoot == "" {
581 workspaceRoot = strings.TrimSpace(spec.Workspace)
582 }
583 if !m.workspaceAllowed(workspaceRoot) {
584 _ = journal.Close()
585 return errors.New("resident manager: workspace is outside allowed roots")
586 }
587 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
588 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
589 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, childID),
590 AgentID: childID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots, ExistingPath: spec.Workspace,
591 })
592 if err != nil {
593 _ = journal.Close()
594 return fmt.Errorf("resident manager rebuild workspace: %w", err)
595 }
596 spec.Workspace = workspace.Dir()
597 spec.WorkspaceMode = workspace.Mode()
598 }
599 runner, err := m.factory(spec, journal)
600 if err != nil {
601 if workspace != nil && workspace.Mode() == WorkspaceShared {
602 _ = workspace.Cleanup(context.Background())
603 }
604 _ = journal.Close()
605 return fmt.Errorf("resident manager rebuild child: %w", err)
606 }
607 child = newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
608 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
609 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
610 m.mu.Lock()
611 if m.closed {
612 m.mu.Unlock()
613 _ = child.Close(context.Background())
614 return errors.New("resident manager: closed")
615 }
616 m.children[childID] = child
617 delete(m.recovered, childID)
618 delete(m.recoveredSpec, childID)
619 m.mu.Unlock()
620 }
621 if child == nil {
622 return errors.New("resident manager: child is not live")
623 }
624 if child.journal == nil {
625 _, err := child.resumeAccepted(ctx, turnID, prompt)
626 return err
627 }
628 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {
629 return err
630 }
631 m.mu.Lock()
632 accepted := m.onAccepted
633 m.mu.Unlock()
634 if accepted != nil {
635 accepted(child.spec, turnID, prompt)
636 }
637 if childAccepted, err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
638 if !childAccepted {
639 _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
640 m.reportCompletion(ResidentCompletion{ChildID: child.spec.ID, TurnID: turnID, Task: prompt, Err: err})
641 }
642 return err
643 }
644 return nil
645}
646
647func (m *ResidentManager) workspaceAllowed(root string) bool {
648 if len(m.allowedRoots) == 0 || strings.TrimSpace(root) == "" {
649 return true
650 }
651 for _, allowed := range m.allowedRoots {
652 if pathWithin(root, allowed) {
653 return true
654 }
655 }
656 return false
657}
658
659// Get returns the live child only. Durable children recovered from a previous
660// host are intentionally not auto-resumed.
661func (m *ResidentManager) Get(childID string) *ResidentChild {
662 if m == nil {
663 return nil
664 }
665 m.mu.Lock()
666 defer m.mu.Unlock()
667 return m.children[childID]
668}
669
670// State reports a live child's current execution state. A child with an
671// accepted turn awaiting global scheduler admission is queued, even though its
672// control loop has started preparing that turn.
673func (m *ResidentManager) State(childID string) (ResidentState, bool) {
674 child := m.Get(childID)
675 if child == nil {
676 return "", false
677 }
678 state := child.State()
679 if state == ResidentRunning && m.scheduler.Pending(childID) {
680 return ResidentQueued, true
681 }
682 return state, true
683}
684
685// Live returns an immutable unfinished-turn snapshot for one live child. It
686// never holds the manager lock while copying the child projection.
687func (m *ResidentManager) Live(childID string) (ResidentLiveSnapshot, bool) {
688 if m == nil {
689 return ResidentLiveSnapshot{}, false
690 }
691 m.mu.Lock()
692 child := m.children[childID]
693 m.mu.Unlock()
694 if child == nil {
695 return ResidentLiveSnapshot{}, false
696 }
697 return child.Live(), true
698}
699
700// Snapshot returns copies of the live children without holding the manager
701// lock while callers render or perform I/O.
702func (m *ResidentManager) Snapshot() []ResidentSnapshot {
703 children, _ := m.SnapshotPage(0, -1)
704 return children
705}
706
707// SnapshotPage returns a bounded, ID-sorted state projection. It avoids
708// materializing live snapshots that a renderer cannot display and reports the
709// total so callers can page without relying on a stale full list.
710func (m *ResidentManager) SnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
711 if m == nil {
712 return nil, 0
713 }
714 if offset < 0 {
715 offset = 0
716 }
717 m.mu.Lock()
718 live := make(map[string]*ResidentChild, len(m.children))
719 for id, child := range m.children {
720 live[id] = child
721 }
722 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
723 for id, snapshot := range m.recovered {
724 recovered[id] = snapshot
725 }
726 m.mu.Unlock()
727 ids := make([]string, 0, len(live)+len(recovered))
728 for id := range recovered {
729 ids = append(ids, id)
730 }
731 for id := range live {
732 if _, alreadyRecovered := recovered[id]; !alreadyRecovered {
733 ids = append(ids, id)
734 }
735 }
736 sort.Strings(ids)
737 total := len(ids)
738 if offset >= total || limit == 0 {
739 return nil, total
740 }
741 end := total
742 if limit > 0 && offset+limit < end {
743 end = offset + limit
744 }
745 result := make([]ResidentSnapshot, 0, end-offset)
746 for _, id := range ids[offset:end] {
747 if child := live[id]; child != nil {
748 result = append(result, residentSnapshot(child))
749 continue
750 }
751 if snapshot, ok := recovered[id]; ok {
752 result = append(result, snapshot)
753 }
754 }
755 return result, total
756}
757
758// RecentSnapshotPage returns a bounded state projection ordered by the latest
759// visible state transition, newest first. It is intended for the interactive
760// dashboard where operators need to identify the child that just changed.
761func (m *ResidentManager) RecentSnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
762 if m == nil {
763 return nil, 0
764 }
765 if offset < 0 {
766 offset = 0
767 }
768 m.mu.Lock()
769 live := make(map[string]*ResidentChild, len(m.children))
770 for id, child := range m.children {
771 live[id] = child
772 }
773 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
774 for id, snapshot := range m.recovered {
775 recovered[id] = snapshot
776 }
777 m.mu.Unlock()
778 ids := make([]string, 0, len(live)+len(recovered))
779 for id := range recovered {
780 ids = append(ids, id)
781 }
782 for id := range live {
783 if _, recoveredAlready := recovered[id]; !recoveredAlready {
784 ids = append(ids, id)
785 }
786 }
787 snapshots := make(map[string]ResidentSnapshot, len(ids))
788 for _, id := range ids {
789 if child := live[id]; child != nil {
790 snapshots[id] = residentSnapshot(child)
791 } else {
792 snapshots[id] = recovered[id]
793 }
794 }
795 sort.Slice(ids, func(i, j int) bool {
796 left, right := snapshots[ids[i]], snapshots[ids[j]]
797 if !left.UpdatedAt.Equal(right.UpdatedAt) {
798 return left.UpdatedAt.After(right.UpdatedAt)
799 }
800 return ids[i] < ids[j]
801 })
802 total := len(ids)
803 if offset >= total || limit == 0 {
804 return nil, total
805 }
806 end := total
807 if limit > 0 && offset+limit < end {
808 end = offset + limit
809 }
810 result := make([]ResidentSnapshot, 0, end-offset)
811 for _, id := range ids[offset:end] {
812 result = append(result, snapshots[id])
813 }
814 return result, total
815}
816
817// ActiveSnapshotPage returns only same-host queued and running children for
818// compact activity surfaces. It deliberately exposes no prompt, transcript, or path.
819func (m *ResidentManager) ActiveSnapshotPage(limit int) ([]ResidentSnapshot, int) {
820 if m == nil {
821 return nil, 0
822 }
823 m.mu.Lock()
824 live := make(map[string]*ResidentChild, len(m.children))
825 for id, child := range m.children {
826 live[id] = child
827 }
828 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
829 for id, snapshot := range m.recovered {
830 recovered[id] = snapshot
831 }
832 m.mu.Unlock()
833
834 result := make([]ResidentSnapshot, 0, len(live)+len(recovered))
835 for _, child := range live {
836 snapshot := residentSnapshot(child)
837 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
838 result = append(result, snapshot)
839 }
840 }
841 for id, snapshot := range recovered {
842 if _, exists := live[id]; exists {
843 continue
844 }
845 if snapshot.OwnedElsewhere {
846 continue
847 }
848 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
849 result = append(result, snapshot)
850 }
851 }
852 sort.Slice(result, func(i, j int) bool {
853 if !result[i].UpdatedAt.Equal(result[j].UpdatedAt) {
854 return result[i].UpdatedAt.After(result[j].UpdatedAt)
855 }
856 return result[i].ID < result[j].ID
857 })
858 total := len(result)
859 if limit > 0 && len(result) > limit {
860 result = result[:limit]
861 }
862 return result, total
863}
864
865// SnapshotFor returns metadata for one child without building a dashboard
866// projection. It keeps child-session headers independent of the total number
867// of resident children.
868func (m *ResidentManager) SnapshotFor(childID string) (ResidentSnapshot, bool) {
869 if m == nil {
870 return ResidentSnapshot{}, false
871 }
872 m.mu.Lock()
873 child := m.children[childID]
874 recovered, ok := m.recovered[childID]
875 m.mu.Unlock()
876 if child != nil {
877 return residentSnapshot(child), true
878 }
879 return recovered, ok
880}
881
882// Reconcile discovers journals left by an earlier host, repairs their bounded
883// projections, and marks queued/running work interrupted. It never creates a
884// live child or replays a prompt. Incompatible journals are left untouched
885// and excluded from discovery. Other per-child errors are returned for callers
886// to surface without hiding valid sibling journals.
887func (m *ResidentManager) Reconcile() []error {
888 if m == nil {
889 return []error{errors.New("resident manager: unavailable")}
890 }
891 m.lifecycleMu.Lock()
892 defer m.lifecycleMu.Unlock()
893 entries, err := os.ReadDir(m.root)
894 if err != nil {
895 if errors.Is(err, os.ErrNotExist) {
896 return nil
897 }
898 return []error{fmt.Errorf("resident manager read root: %w", err)}
899 }
900 sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
901 m.mu.Lock()
902 m.recovered = make(map[string]ResidentSnapshot)
903 m.recoveredSpec = make(map[string]ResidentChildSpec)
904 m.mu.Unlock()
905 var errs []error
906 for _, entry := range entries {
907 if !entry.IsDir() {
908 continue
909 }
910 // The old subprocess runtime stored children beneath an "agents"
911 // container. Resident children are direct entries in this root. Leave
912 // that legacy state untouched: it is neither read nor migrated.
913 if entry.Name() == "agents" {
914 continue
915 }
916 childID := entry.Name()
917 m.mu.Lock()
918 _, live := m.children[childID]
919 _, pending := m.pending[childID]
920 m.mu.Unlock()
921 if live || pending {
922 continue
923 }
924 childDir := filepath.Join(m.root, childID)
925 metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
926 // Unknown journal versions cannot be resumed. Leave them on disk
927 // without reporting the same incompatibility at every host startup.
928 // Match the exact sentinel so a joined journal-close error is still
929 // reported.
930 if reconcileErr == ErrIncompatibleResidentJournal {
931 continue
932 }
933 if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
934 foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
935 if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
936 errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
937 continue
938 }
939 if readErr == nil {
940 if foreign.Version != residentJournalVersion && foreign.Version != residentJournalVersionV2 {
941 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, ErrIncompatibleResidentJournal))
942 continue
943 }
944 // A v3 projection can never carry the removed budget
945 // state: the writer only persists live states. Treat it
946 // as corruption rather than resurfacing the zombie state.
947 if foreign.Version != residentJournalVersionV2 && foreign.State == ResidentState(residentV2OutcomeBudgetExhausted) {
948 errs = append(errs, fmt.Errorf("resident child %s: invalid terminal state", childID))
949 continue
950 }
951 }
952 state, updatedAt := foreign.State, foreign.UpdatedAt
953 if readErr != nil || state == "" {
954 // Ownership begins before child.accepted is durable. Do not read
955 // that live transcript merely to improve this transient snapshot.
956 state, updatedAt = ResidentQueued, time.Now().UTC()
957 }
958 if foreign.Version == residentJournalVersionV2 && state == ResidentState(residentV2OutcomeBudgetExhausted) {
959 // Historical budget stop from an older host: display it as
960 // failed, never as a distinct terminal state.
961 state = ResidentFailed
962 }
963 m.mu.Lock()
964 m.recovered[childID] = ResidentSnapshot{
965 ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true,
966 Usage: foreign.Usage, ContextUsed: foreign.ContextUsed, ContextMax: foreign.ContextMax, Subscription: foreign.Subscription,
967 }
968 m.mu.Unlock()
969 continue
970 }
971 if reconcileErr != nil {
972 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
973 continue
974 }
975 m.mu.Lock()
976 if _, live := m.children[childID]; !live {
977 m.recovered[childID] = ResidentSnapshot{
978 ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model,
979 WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt,
980 Usage: metadata.Usage, ContextUsed: metadata.ContextUsed, ContextMax: metadata.ContextMax, Subscription: metadata.Subscription,
981 }
982 m.recoveredSpec[childID] = spec
983 }
984 m.mu.Unlock()
985 }
986 return errs
987}
988
989// UnmetRequired returns required work that has not reached the successful idle
990// boundary. Failed and interrupted work intentionally remain unmet until an
991// explicit successful follow-up is accepted and completed.
992func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
993 all := m.Snapshot()
994 result := all[:0]
995 for _, snapshot := range all {
996 if snapshot.Required && snapshot.State != ResidentIdle && snapshot.State != ResidentCompleted {
997 result = append(result, snapshot)
998 }
999 }
1000 return result
1001}
1002
1003// Interrupt cancels only a live child's running turn and drops its queued
1004// follow-ups. The child stays live with its transcript, so a later resume
1005// continues with full context. It reports false when no turn is running.
1006func (m *ResidentManager) Interrupt(ctx context.Context, childID string) (bool, error) {
1007 if m == nil {
1008 return false, errors.New("resident manager: unavailable")
1009 }
1010 m.lifecycleMu.Lock()
1011 defer m.lifecycleMu.Unlock()
1012 child := m.Get(strings.TrimSpace(childID))
1013 if child == nil {
1014 return false, errors.New("resident manager: child is not live")
1015 }
1016 interrupted, err := child.interruptTurn(ctx)
1017 if err != nil {
1018 return false, err
1019 }
1020 m.notifyUpdate(child.spec.ID, false)
1021 return interrupted, nil
1022}
1023
1024// Stop cancels one live child and waits for its control loop. A disk-only
1025// interrupted child must be explicitly resumed by host construction instead.
1026func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
1027 if m == nil {
1028 return errors.New("resident manager: unavailable")
1029 }
1030 m.lifecycleMu.Lock()
1031 defer m.lifecycleMu.Unlock()
1032 childID = strings.TrimSpace(childID)
1033 m.mu.Lock()
1034 child := m.children[childID]
1035 m.mu.Unlock()
1036 if child == nil {
1037 return errors.New("resident manager: child is not live")
1038 }
1039 if err := child.Close(ctx); err != nil {
1040 return err
1041 }
1042 // A stopped child has released its resident agent and journal. Retain only
1043 // its durable spec so an explicit follow-up can rebuild a fresh resident
1044 // child; keeping the closed child live would make Resume append to a closed
1045 // journal.
1046 m.mu.Lock()
1047 if m.children[childID] == child {
1048 delete(m.children, childID)
1049 m.recovered[childID] = residentSnapshot(child)
1050 m.recoveredSpec[childID] = child.spec
1051 }
1052 m.mu.Unlock()
1053 m.notifyUpdate(childID, false)
1054 return nil
1055}
1056
1057// History reads bounded finalized resident history through the manager-owned
1058// state root. It accepts only a child ID, never a filesystem path.
1059func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
1060 if m == nil {
1061 return nil, errors.New("resident manager: unavailable")
1062 }
1063 dir, release, err := m.readableResidentDir(childID)
1064 if err != nil {
1065 return nil, err
1066 }
1067 defer release()
1068 return ReadResidentHistory(dir, limit)
1069}
1070
1071// HistoryPage reads one recent-first, bounded page through the manager-owned
1072// state root. Public consumers identify only a child and receive an opaque
1073// cursor for older content.
1074func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
1075 if m == nil {
1076 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
1077 }
1078 dir, release, err := m.readableResidentDir(childID)
1079 if err != nil {
1080 return ResidentHistoryPage{}, err
1081 }
1082 defer release()
1083 return ReadResidentHistoryPage(dir, olderCursor, limit)
1084}
1085
1086// Result returns the bounded latest-turn projection for a resident child.
1087func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
1088 if m == nil {
1089 return ResidentResult{}, errors.New("resident manager: unavailable")
1090 }
1091 dir, release, err := m.readableResidentDir(childID)
1092 if err != nil {
1093 return ResidentResult{}, err
1094 }
1095 defer release()
1096 return ReadResidentResult(filepath.Join(dir, residentResultName))
1097}
1098
1099func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
1100 dir, err := ResidentHistoryDir(m.root, childID)
1101 if err != nil {
1102 return "", nil, err
1103 }
1104 childID = strings.TrimSpace(childID)
1105 m.mu.Lock()
1106 live := m.children[childID] != nil
1107 m.mu.Unlock()
1108 if live {
1109 return dir, func() {}, nil
1110 }
1111 lease, err := acquireResidentLease(dir)
1112 if err != nil {
1113 if errors.Is(err, ErrResidentLeaseBusy) {
1114 return "", nil, errors.New("resident manager: child is owned by another zut process")
1115 }
1116 return "", nil, err
1117 }
1118 return dir, func() { _ = lease.Close() }, nil
1119}
1120
1121func (m *ResidentManager) Close(ctx context.Context) error {
1122 if m == nil {
1123 return nil
1124 }
1125 m.mu.Lock()
1126 m.closed = true
1127 children := make([]*ResidentChild, 0, len(m.children))
1128 for _, child := range m.children {
1129 children = append(children, child)
1130 }
1131 m.mu.Unlock()
1132 var errs []error
1133 for _, child := range children {
1134 if err := child.Close(ctx); err != nil {
1135 errs = append(errs, err)
1136 }
1137 }
1138 return errors.Join(errs...)
1139}