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