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 same-host queued and running children for
733// compact 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.OwnedElsewhere {
761 continue
762 }
763 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
764 result = append(result, snapshot)
765 }
766 }
767 sort.Slice(result, func(i, j int) bool {
768 if !result[i].UpdatedAt.Equal(result[j].UpdatedAt) {
769 return result[i].UpdatedAt.After(result[j].UpdatedAt)
770 }
771 return result[i].ID < result[j].ID
772 })
773 total := len(result)
774 if limit > 0 && len(result) > limit {
775 result = result[:limit]
776 }
777 return result, total
778}
779
780// SnapshotFor returns metadata for one child without building a dashboard
781// projection. It keeps child-session headers independent of the total number
782// of resident children.
783func (m *ResidentManager) SnapshotFor(childID string) (ResidentSnapshot, bool) {
784 if m == nil {
785 return ResidentSnapshot{}, false
786 }
787 m.mu.Lock()
788 child := m.children[childID]
789 recovered, ok := m.recovered[childID]
790 m.mu.Unlock()
791 if child != nil {
792 return residentSnapshot(child), true
793 }
794 return recovered, ok
795}
796
797// Reconcile discovers journals left by an earlier host, repairs their bounded
798// projections, and marks queued/running work interrupted. It never creates a
799// live child or replays a prompt. Per-child errors are returned for callers to
800// surface without hiding valid sibling journals.
801func (m *ResidentManager) Reconcile() []error {
802 if m == nil {
803 return []error{errors.New("resident manager: unavailable")}
804 }
805 m.lifecycleMu.Lock()
806 defer m.lifecycleMu.Unlock()
807 entries, err := os.ReadDir(m.root)
808 if err != nil {
809 if errors.Is(err, os.ErrNotExist) {
810 return nil
811 }
812 return []error{fmt.Errorf("resident manager read root: %w", err)}
813 }
814 sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
815 m.mu.Lock()
816 m.recovered = make(map[string]ResidentSnapshot)
817 m.recoveredSpec = make(map[string]ResidentChildSpec)
818 m.mu.Unlock()
819 var errs []error
820 for _, entry := range entries {
821 if !entry.IsDir() {
822 continue
823 }
824 // The old subprocess runtime stored children beneath an "agents"
825 // container. Resident children are direct entries in this root. Leave
826 // that legacy state untouched: it is neither read nor migrated.
827 if entry.Name() == "agents" {
828 continue
829 }
830 childID := entry.Name()
831 m.mu.Lock()
832 _, live := m.children[childID]
833 _, pending := m.pending[childID]
834 m.mu.Unlock()
835 if live || pending {
836 continue
837 }
838 childDir := filepath.Join(m.root, childID)
839 metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
840 if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
841 foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
842 if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
843 errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
844 continue
845 }
846 state, updatedAt := foreign.State, foreign.UpdatedAt
847 if readErr != nil || state == "" {
848 // Ownership begins before child.accepted is durable. Do not read
849 // that live transcript merely to improve this transient snapshot.
850 state, updatedAt = ResidentQueued, time.Now().UTC()
851 }
852 m.mu.Lock()
853 m.recovered[childID] = ResidentSnapshot{
854 ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true,
855 Usage: foreign.Usage, ContextUsed: foreign.ContextUsed, ContextMax: foreign.ContextMax, Subscription: foreign.Subscription,
856 }
857 m.mu.Unlock()
858 continue
859 }
860 if reconcileErr != nil {
861 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
862 continue
863 }
864 m.mu.Lock()
865 if _, live := m.children[childID]; !live {
866 m.recovered[childID] = ResidentSnapshot{
867 ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model,
868 WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt,
869 Usage: metadata.Usage, ContextUsed: metadata.ContextUsed, ContextMax: metadata.ContextMax, Subscription: metadata.Subscription,
870 }
871 m.recoveredSpec[childID] = spec
872 }
873 m.mu.Unlock()
874 }
875 return errs
876}
877
878// UnmetRequired returns required work that has not reached the successful idle
879// boundary. Failed and interrupted work intentionally remain unmet until an
880// explicit successful follow-up is accepted and completed.
881func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
882 all := m.Snapshot()
883 result := all[:0]
884 for _, snapshot := range all {
885 if snapshot.Required && snapshot.State != ResidentIdle {
886 result = append(result, snapshot)
887 }
888 }
889 return result
890}
891
892// Stop cancels one live child and waits for its control loop. A disk-only
893// interrupted child must be explicitly resumed by host construction instead.
894func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
895 if m == nil {
896 return errors.New("resident manager: unavailable")
897 }
898 m.lifecycleMu.Lock()
899 defer m.lifecycleMu.Unlock()
900 childID = strings.TrimSpace(childID)
901 m.mu.Lock()
902 child := m.children[childID]
903 m.mu.Unlock()
904 if child == nil {
905 return errors.New("resident manager: child is not live")
906 }
907 if err := child.Close(ctx); err != nil {
908 return err
909 }
910 // A stopped child has released its resident agent and journal. Retain only
911 // its durable spec so an explicit follow-up can rebuild a fresh resident
912 // child; keeping the closed child live would make Resume append to a closed
913 // journal.
914 m.mu.Lock()
915 if m.children[childID] == child {
916 delete(m.children, childID)
917 m.recovered[childID] = residentSnapshot(child)
918 m.recoveredSpec[childID] = child.spec
919 }
920 m.mu.Unlock()
921 m.notifyUpdate(childID, false)
922 return nil
923}
924
925// History reads bounded finalized resident history through the manager-owned
926// state root. It accepts only a child ID, never a filesystem path.
927func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
928 if m == nil {
929 return nil, errors.New("resident manager: unavailable")
930 }
931 dir, release, err := m.readableResidentDir(childID)
932 if err != nil {
933 return nil, err
934 }
935 defer release()
936 return ReadResidentHistory(dir, limit)
937}
938
939// HistoryPage reads one recent-first, bounded page through the manager-owned
940// state root. Public consumers identify only a child and receive an opaque
941// cursor for older content.
942func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
943 if m == nil {
944 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
945 }
946 dir, release, err := m.readableResidentDir(childID)
947 if err != nil {
948 return ResidentHistoryPage{}, err
949 }
950 defer release()
951 return ReadResidentHistoryPage(dir, olderCursor, limit)
952}
953
954// Result returns the bounded latest-turn projection for a resident child.
955func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
956 if m == nil {
957 return ResidentResult{}, errors.New("resident manager: unavailable")
958 }
959 dir, release, err := m.readableResidentDir(childID)
960 if err != nil {
961 return ResidentResult{}, err
962 }
963 defer release()
964 return ReadResidentResult(filepath.Join(dir, residentResultName))
965}
966
967func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
968 dir, err := ResidentHistoryDir(m.root, childID)
969 if err != nil {
970 return "", nil, err
971 }
972 childID = strings.TrimSpace(childID)
973 m.mu.Lock()
974 live := m.children[childID] != nil
975 m.mu.Unlock()
976 if live {
977 return dir, func() {}, nil
978 }
979 lease, err := acquireResidentLease(dir)
980 if err != nil {
981 if errors.Is(err, ErrResidentLeaseBusy) {
982 return "", nil, errors.New("resident manager: child is owned by another zut process")
983 }
984 return "", nil, err
985 }
986 return dir, func() { _ = lease.Close() }, nil
987}
988
989func (m *ResidentManager) Close(ctx context.Context) error {
990 if m == nil {
991 return nil
992 }
993 m.mu.Lock()
994 m.closed = true
995 children := make([]*ResidentChild, 0, len(m.children))
996 for _, child := range m.children {
997 children = append(children, child)
998 }
999 m.mu.Unlock()
1000 var errs []error
1001 for _, child := range children {
1002 if err := child.Close(ctx); err != nil {
1003 errs = append(errs, err)
1004 }
1005 }
1006 return errors.Join(errs...)
1007}