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