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