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, child.journal.BudgetBaseline())
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, baseline int64) (BudgetSnapshot, string) {
244 return budgetSnapshotSince(usage, limit, baseline), 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 {
474 if _, pending := m.pending[childID]; pending {
475 m.mu.Unlock()
476 return errors.New("resident manager: child resume pending")
477 }
478 m.pending[childID] = struct{}{}
479 }
480 m.mu.Unlock()
481 if child != nil && residentSnapshot(child).Budget.State == BudgetExceeded && (child.State() == ResidentRunning || child.State() == ResidentQueued) {
482 return errors.New("resident manager: wait for the budget-exhausted turn's terminal notification before resuming")
483 }
484 if child == nil && recovered {
485 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
486 journal, err := OpenResidentJournal(m.root, childID)
487 if err != nil {
488 if errors.Is(err, ErrResidentLeaseBusy) {
489 return errors.New("resident manager: child is owned by another zut process")
490 }
491 return err
492 }
493 metadata, err := reconcileOwnedResidentJournal(journal)
494 if err != nil {
495 _ = journal.Close()
496 return err
497 }
498 journal.ConfigureUsage(metadata.ContextMax, metadata.Subscription)
499 records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
500 if err != nil || len(records) == 0 || records[0].Spec == nil {
501 _ = journal.Close()
502 return errors.New("resident manager: invalid accepted specification")
503 }
504 spec = *records[0].Spec
505 var workspace WorkspaceHandle
506 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
507 if workspaceRoot == "" {
508 workspaceRoot = strings.TrimSpace(spec.Workspace)
509 }
510 if !m.workspaceAllowed(workspaceRoot) {
511 _ = journal.Close()
512 return errors.New("resident manager: workspace is outside allowed roots")
513 }
514 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
515 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
516 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, childID),
517 AgentID: childID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots, ExistingPath: spec.Workspace,
518 })
519 if err != nil {
520 _ = journal.Close()
521 return fmt.Errorf("resident manager rebuild workspace: %w", err)
522 }
523 spec.Workspace = workspace.Dir()
524 spec.WorkspaceMode = workspace.Mode()
525 }
526 runner, err := m.factory(spec, journal)
527 if err != nil {
528 if workspace != nil && workspace.Mode() == WorkspaceShared {
529 _ = workspace.Cleanup(context.Background())
530 }
531 _ = journal.Close()
532 return fmt.Errorf("resident manager rebuild child: %w", err)
533 }
534 child = newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
535 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
536 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
537 m.mu.Lock()
538 if m.closed {
539 m.mu.Unlock()
540 _ = child.Close(context.Background())
541 return errors.New("resident manager: closed")
542 }
543 m.children[childID] = child
544 delete(m.recovered, childID)
545 delete(m.recoveredSpec, childID)
546 m.mu.Unlock()
547 }
548 if child == nil {
549 return errors.New("resident manager: child is not live")
550 }
551 turnID := uuid.NewString()
552 if child.journal == nil {
553 _, err := child.resumeAccepted(ctx, turnID, prompt)
554 return err
555 }
556 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {
557 return err
558 }
559 m.mu.Lock()
560 accepted := m.onAccepted
561 m.mu.Unlock()
562 if accepted != nil {
563 accepted(child.spec, turnID, prompt)
564 }
565 if childAccepted, err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
566 if !childAccepted {
567 _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
568 m.reportCompletion(ResidentCompletion{ChildID: child.spec.ID, TurnID: turnID, Task: prompt, Err: err})
569 }
570 return err
571 }
572 return nil
573}
574
575func (m *ResidentManager) workspaceAllowed(root string) bool {
576 if len(m.allowedRoots) == 0 || strings.TrimSpace(root) == "" {
577 return true
578 }
579 for _, allowed := range m.allowedRoots {
580 if pathWithin(root, allowed) {
581 return true
582 }
583 }
584 return false
585}
586
587// Get returns the live child only. Durable children recovered from a previous
588// host are intentionally not auto-resumed.
589func (m *ResidentManager) Get(childID string) *ResidentChild {
590 if m == nil {
591 return nil
592 }
593 m.mu.Lock()
594 defer m.mu.Unlock()
595 return m.children[childID]
596}
597
598// State reports a live child's current execution state. A child with an
599// accepted turn awaiting global scheduler admission is queued, even though its
600// control loop has started preparing that turn.
601func (m *ResidentManager) State(childID string) (ResidentState, bool) {
602 child := m.Get(childID)
603 if child == nil {
604 return "", false
605 }
606 state := child.State()
607 if state == ResidentRunning && m.scheduler.Pending(childID) {
608 return ResidentQueued, true
609 }
610 return state, true
611}
612
613// Live returns an immutable unfinished-turn snapshot for one live child. It
614// never holds the manager lock while copying the child projection.
615func (m *ResidentManager) Live(childID string) (ResidentLiveSnapshot, bool) {
616 if m == nil {
617 return ResidentLiveSnapshot{}, false
618 }
619 m.mu.Lock()
620 child := m.children[childID]
621 m.mu.Unlock()
622 if child == nil {
623 return ResidentLiveSnapshot{}, false
624 }
625 return child.Live(), true
626}
627
628// Snapshot returns copies of the live children without holding the manager
629// lock while callers render or perform I/O.
630func (m *ResidentManager) Snapshot() []ResidentSnapshot {
631 children, _ := m.SnapshotPage(0, -1)
632 return children
633}
634
635// SnapshotPage returns a bounded, ID-sorted state projection. It avoids
636// materializing live snapshots that a renderer cannot display and reports the
637// total so callers can page without relying on a stale full list.
638func (m *ResidentManager) SnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
639 if m == nil {
640 return nil, 0
641 }
642 if offset < 0 {
643 offset = 0
644 }
645 m.mu.Lock()
646 live := make(map[string]*ResidentChild, len(m.children))
647 for id, child := range m.children {
648 live[id] = child
649 }
650 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
651 for id, snapshot := range m.recovered {
652 recovered[id] = snapshot
653 }
654 m.mu.Unlock()
655 ids := make([]string, 0, len(live)+len(recovered))
656 for id := range recovered {
657 ids = append(ids, id)
658 }
659 for id := range live {
660 if _, alreadyRecovered := recovered[id]; !alreadyRecovered {
661 ids = append(ids, id)
662 }
663 }
664 sort.Strings(ids)
665 total := len(ids)
666 if offset >= total || limit == 0 {
667 return nil, total
668 }
669 end := total
670 if limit > 0 && offset+limit < end {
671 end = offset + limit
672 }
673 result := make([]ResidentSnapshot, 0, end-offset)
674 for _, id := range ids[offset:end] {
675 if child := live[id]; child != nil {
676 result = append(result, residentSnapshot(child))
677 continue
678 }
679 if snapshot, ok := recovered[id]; ok {
680 result = append(result, snapshot)
681 }
682 }
683 return result, total
684}
685
686// RecentSnapshotPage returns a bounded state projection ordered by the latest
687// visible state transition, newest first. It is intended for the interactive
688// dashboard where operators need to identify the child that just changed.
689func (m *ResidentManager) RecentSnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
690 if m == nil {
691 return nil, 0
692 }
693 if offset < 0 {
694 offset = 0
695 }
696 m.mu.Lock()
697 live := make(map[string]*ResidentChild, len(m.children))
698 for id, child := range m.children {
699 live[id] = child
700 }
701 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
702 for id, snapshot := range m.recovered {
703 recovered[id] = snapshot
704 }
705 m.mu.Unlock()
706 ids := make([]string, 0, len(live)+len(recovered))
707 for id := range recovered {
708 ids = append(ids, id)
709 }
710 for id := range live {
711 if _, recoveredAlready := recovered[id]; !recoveredAlready {
712 ids = append(ids, id)
713 }
714 }
715 snapshots := make(map[string]ResidentSnapshot, len(ids))
716 for _, id := range ids {
717 if child := live[id]; child != nil {
718 snapshots[id] = residentSnapshot(child)
719 } else {
720 snapshots[id] = recovered[id]
721 }
722 }
723 sort.Slice(ids, func(i, j int) bool {
724 left, right := snapshots[ids[i]], snapshots[ids[j]]
725 if !left.UpdatedAt.Equal(right.UpdatedAt) {
726 return left.UpdatedAt.After(right.UpdatedAt)
727 }
728 return ids[i] < ids[j]
729 })
730 total := len(ids)
731 if offset >= total || limit == 0 {
732 return nil, total
733 }
734 end := total
735 if limit > 0 && offset+limit < end {
736 end = offset + limit
737 }
738 result := make([]ResidentSnapshot, 0, end-offset)
739 for _, id := range ids[offset:end] {
740 result = append(result, snapshots[id])
741 }
742 return result, total
743}
744
745// ActiveSnapshotPage returns only same-host queued and running children for
746// compact activity surfaces. It deliberately exposes no prompt, transcript, or path.
747func (m *ResidentManager) ActiveSnapshotPage(limit int) ([]ResidentSnapshot, int) {
748 if m == nil {
749 return nil, 0
750 }
751 m.mu.Lock()
752 live := make(map[string]*ResidentChild, len(m.children))
753 for id, child := range m.children {
754 live[id] = child
755 }
756 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
757 for id, snapshot := range m.recovered {
758 recovered[id] = snapshot
759 }
760 m.mu.Unlock()
761
762 result := make([]ResidentSnapshot, 0, len(live)+len(recovered))
763 for _, child := range live {
764 snapshot := residentSnapshot(child)
765 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
766 result = append(result, snapshot)
767 }
768 }
769 for id, snapshot := range recovered {
770 if _, exists := live[id]; exists {
771 continue
772 }
773 if snapshot.OwnedElsewhere {
774 continue
775 }
776 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
777 result = append(result, snapshot)
778 }
779 }
780 sort.Slice(result, func(i, j int) bool {
781 if !result[i].UpdatedAt.Equal(result[j].UpdatedAt) {
782 return result[i].UpdatedAt.After(result[j].UpdatedAt)
783 }
784 return result[i].ID < result[j].ID
785 })
786 total := len(result)
787 if limit > 0 && len(result) > limit {
788 result = result[:limit]
789 }
790 return result, total
791}
792
793// SnapshotFor returns metadata for one child without building a dashboard
794// projection. It keeps child-session headers independent of the total number
795// of resident children.
796func (m *ResidentManager) SnapshotFor(childID string) (ResidentSnapshot, bool) {
797 if m == nil {
798 return ResidentSnapshot{}, false
799 }
800 m.mu.Lock()
801 child := m.children[childID]
802 recovered, ok := m.recovered[childID]
803 m.mu.Unlock()
804 if child != nil {
805 return residentSnapshot(child), true
806 }
807 return recovered, ok
808}
809
810// Reconcile discovers journals left by an earlier host, repairs their bounded
811// projections, and marks queued/running work interrupted. It never creates a
812// live child or replays a prompt. Per-child errors are returned for callers to
813// surface without hiding valid sibling journals.
814func (m *ResidentManager) Reconcile() []error {
815 if m == nil {
816 return []error{errors.New("resident manager: unavailable")}
817 }
818 m.lifecycleMu.Lock()
819 defer m.lifecycleMu.Unlock()
820 entries, err := os.ReadDir(m.root)
821 if err != nil {
822 if errors.Is(err, os.ErrNotExist) {
823 return nil
824 }
825 return []error{fmt.Errorf("resident manager read root: %w", err)}
826 }
827 sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
828 m.mu.Lock()
829 m.recovered = make(map[string]ResidentSnapshot)
830 m.recoveredSpec = make(map[string]ResidentChildSpec)
831 m.mu.Unlock()
832 var errs []error
833 for _, entry := range entries {
834 if !entry.IsDir() {
835 continue
836 }
837 // The old subprocess runtime stored children beneath an "agents"
838 // container. Resident children are direct entries in this root. Leave
839 // that legacy state untouched: it is neither read nor migrated.
840 if entry.Name() == "agents" {
841 continue
842 }
843 childID := entry.Name()
844 m.mu.Lock()
845 _, live := m.children[childID]
846 _, pending := m.pending[childID]
847 m.mu.Unlock()
848 if live || pending {
849 continue
850 }
851 childDir := filepath.Join(m.root, childID)
852 metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
853 if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
854 foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
855 if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
856 errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
857 continue
858 }
859 state, updatedAt := foreign.State, foreign.UpdatedAt
860 if readErr != nil || state == "" {
861 // Ownership begins before child.accepted is durable. Do not read
862 // that live transcript merely to improve this transient snapshot.
863 state, updatedAt = ResidentQueued, time.Now().UTC()
864 }
865 m.mu.Lock()
866 m.recovered[childID] = ResidentSnapshot{
867 ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true,
868 Usage: foreign.Usage, ContextUsed: foreign.ContextUsed, ContextMax: foreign.ContextMax, Subscription: foreign.Subscription,
869 }
870 m.mu.Unlock()
871 continue
872 }
873 if reconcileErr != nil {
874 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
875 continue
876 }
877 budget, budgetSource := residentBudgetSnapshot(metadata.Usage, spec.BudgetLimit, spec.BudgetSource, metadata.BudgetBaseline)
878 m.mu.Lock()
879 if _, live := m.children[childID]; !live {
880 m.recovered[childID] = ResidentSnapshot{
881 ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model,
882 WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt,
883 Usage: metadata.Usage, ContextUsed: metadata.ContextUsed, ContextMax: metadata.ContextMax, Subscription: metadata.Subscription,
884 Budget: budget, BudgetSource: budgetSource,
885 }
886 m.recoveredSpec[childID] = spec
887 }
888 m.mu.Unlock()
889 }
890 return errs
891}
892
893// UnmetRequired returns required work that has not reached the successful idle
894// boundary. Failed and interrupted work intentionally remain unmet until an
895// explicit successful follow-up is accepted and completed.
896func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
897 all := m.Snapshot()
898 result := all[:0]
899 for _, snapshot := range all {
900 if snapshot.Required && snapshot.State != ResidentIdle && snapshot.State != ResidentCompleted {
901 result = append(result, snapshot)
902 }
903 }
904 return result
905}
906
907// Stop cancels one live child and waits for its control loop. A disk-only
908// interrupted child must be explicitly resumed by host construction instead.
909func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
910 if m == nil {
911 return errors.New("resident manager: unavailable")
912 }
913 m.lifecycleMu.Lock()
914 defer m.lifecycleMu.Unlock()
915 childID = strings.TrimSpace(childID)
916 m.mu.Lock()
917 child := m.children[childID]
918 m.mu.Unlock()
919 if child == nil {
920 return errors.New("resident manager: child is not live")
921 }
922 if err := child.Close(ctx); err != nil {
923 return err
924 }
925 // A stopped child has released its resident agent and journal. Retain only
926 // its durable spec so an explicit follow-up can rebuild a fresh resident
927 // child; keeping the closed child live would make Resume append to a closed
928 // journal.
929 m.mu.Lock()
930 if m.children[childID] == child {
931 delete(m.children, childID)
932 m.recovered[childID] = residentSnapshot(child)
933 m.recoveredSpec[childID] = child.spec
934 }
935 m.mu.Unlock()
936 m.notifyUpdate(childID, false)
937 return nil
938}
939
940// History reads bounded finalized resident history through the manager-owned
941// state root. It accepts only a child ID, never a filesystem path.
942func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
943 if m == nil {
944 return nil, errors.New("resident manager: unavailable")
945 }
946 dir, release, err := m.readableResidentDir(childID)
947 if err != nil {
948 return nil, err
949 }
950 defer release()
951 return ReadResidentHistory(dir, limit)
952}
953
954// HistoryPage reads one recent-first, bounded page through the manager-owned
955// state root. Public consumers identify only a child and receive an opaque
956// cursor for older content.
957func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
958 if m == nil {
959 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
960 }
961 dir, release, err := m.readableResidentDir(childID)
962 if err != nil {
963 return ResidentHistoryPage{}, err
964 }
965 defer release()
966 return ReadResidentHistoryPage(dir, olderCursor, limit)
967}
968
969// Result returns the bounded latest-turn projection for a resident child.
970func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
971 if m == nil {
972 return ResidentResult{}, errors.New("resident manager: unavailable")
973 }
974 dir, release, err := m.readableResidentDir(childID)
975 if err != nil {
976 return ResidentResult{}, err
977 }
978 defer release()
979 return ReadResidentResult(filepath.Join(dir, residentResultName))
980}
981
982func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
983 dir, err := ResidentHistoryDir(m.root, childID)
984 if err != nil {
985 return "", nil, err
986 }
987 childID = strings.TrimSpace(childID)
988 m.mu.Lock()
989 live := m.children[childID] != nil
990 m.mu.Unlock()
991 if live {
992 return dir, func() {}, nil
993 }
994 lease, err := acquireResidentLease(dir)
995 if err != nil {
996 if errors.Is(err, ErrResidentLeaseBusy) {
997 return "", nil, errors.New("resident manager: child is owned by another zut process")
998 }
999 return "", nil, err
1000 }
1001 return dir, func() { _ = lease.Close() }, nil
1002}
1003
1004func (m *ResidentManager) Close(ctx context.Context) error {
1005 if m == nil {
1006 return nil
1007 }
1008 m.mu.Lock()
1009 m.closed = true
1010 children := make([]*ResidentChild, 0, len(m.children))
1011 for _, child := range m.children {
1012 children = append(children, child)
1013 }
1014 m.mu.Unlock()
1015 var errs []error
1016 for _, child := range children {
1017 if err := child.Close(ctx); err != nil {
1018 errs = append(errs, err)
1019 }
1020 }
1021 return errors.Join(errs...)
1022}