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