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. Incompatible budget journals are left untouched
813// and excluded from discovery. Other per-child errors are returned for callers
814// to surface without hiding valid sibling journals.
815func (m *ResidentManager) Reconcile() []error {
816 if m == nil {
817 return []error{errors.New("resident manager: unavailable")}
818 }
819 m.lifecycleMu.Lock()
820 defer m.lifecycleMu.Unlock()
821 entries, err := os.ReadDir(m.root)
822 if err != nil {
823 if errors.Is(err, os.ErrNotExist) {
824 return nil
825 }
826 return []error{fmt.Errorf("resident manager read root: %w", err)}
827 }
828 sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
829 m.mu.Lock()
830 m.recovered = make(map[string]ResidentSnapshot)
831 m.recoveredSpec = make(map[string]ResidentChildSpec)
832 m.mu.Unlock()
833 var errs []error
834 for _, entry := range entries {
835 if !entry.IsDir() {
836 continue
837 }
838 // The old subprocess runtime stored children beneath an "agents"
839 // container. Resident children are direct entries in this root. Leave
840 // that legacy state untouched: it is neither read nor migrated.
841 if entry.Name() == "agents" {
842 continue
843 }
844 childID := entry.Name()
845 m.mu.Lock()
846 _, live := m.children[childID]
847 _, pending := m.pending[childID]
848 m.mu.Unlock()
849 if live || pending {
850 continue
851 }
852 childDir := filepath.Join(m.root, childID)
853 metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
854 // Old budget journals cannot be resumed. Leave them on disk without
855 // reporting the same incompatibility at every host startup. Match the
856 // exact sentinel so a joined journal-close error is still reported.
857 if reconcileErr == ErrIncompatibleResidentBudget {
858 continue
859 }
860 if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
861 foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
862 if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
863 errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
864 continue
865 }
866 state, updatedAt := foreign.State, foreign.UpdatedAt
867 if readErr != nil || state == "" {
868 // Ownership begins before child.accepted is durable. Do not read
869 // that live transcript merely to improve this transient snapshot.
870 state, updatedAt = ResidentQueued, time.Now().UTC()
871 }
872 m.mu.Lock()
873 m.recovered[childID] = ResidentSnapshot{
874 ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true,
875 Usage: foreign.Usage, ContextUsed: foreign.ContextUsed, ContextMax: foreign.ContextMax, Subscription: foreign.Subscription,
876 }
877 m.mu.Unlock()
878 continue
879 }
880 if reconcileErr != nil {
881 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
882 continue
883 }
884 budget, budgetSource := residentBudgetSnapshot(metadata.Usage, spec.BudgetLimit, spec.BudgetSource, metadata.BudgetBaseline)
885 m.mu.Lock()
886 if _, live := m.children[childID]; !live {
887 m.recovered[childID] = ResidentSnapshot{
888 ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model,
889 WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt,
890 Usage: metadata.Usage, ContextUsed: metadata.ContextUsed, ContextMax: metadata.ContextMax, Subscription: metadata.Subscription,
891 Budget: budget, BudgetSource: budgetSource,
892 }
893 m.recoveredSpec[childID] = spec
894 }
895 m.mu.Unlock()
896 }
897 return errs
898}
899
900// UnmetRequired returns required work that has not reached the successful idle
901// boundary. Failed and interrupted work intentionally remain unmet until an
902// explicit successful follow-up is accepted and completed.
903func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
904 all := m.Snapshot()
905 result := all[:0]
906 for _, snapshot := range all {
907 if snapshot.Required && snapshot.State != ResidentIdle && snapshot.State != ResidentCompleted {
908 result = append(result, snapshot)
909 }
910 }
911 return result
912}
913
914// Stop cancels one live child and waits for its control loop. A disk-only
915// interrupted child must be explicitly resumed by host construction instead.
916func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
917 if m == nil {
918 return errors.New("resident manager: unavailable")
919 }
920 m.lifecycleMu.Lock()
921 defer m.lifecycleMu.Unlock()
922 childID = strings.TrimSpace(childID)
923 m.mu.Lock()
924 child := m.children[childID]
925 m.mu.Unlock()
926 if child == nil {
927 return errors.New("resident manager: child is not live")
928 }
929 if err := child.Close(ctx); err != nil {
930 return err
931 }
932 // A stopped child has released its resident agent and journal. Retain only
933 // its durable spec so an explicit follow-up can rebuild a fresh resident
934 // child; keeping the closed child live would make Resume append to a closed
935 // journal.
936 m.mu.Lock()
937 if m.children[childID] == child {
938 delete(m.children, childID)
939 m.recovered[childID] = residentSnapshot(child)
940 m.recoveredSpec[childID] = child.spec
941 }
942 m.mu.Unlock()
943 m.notifyUpdate(childID, false)
944 return nil
945}
946
947// History reads bounded finalized resident history through the manager-owned
948// state root. It accepts only a child ID, never a filesystem path.
949func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
950 if m == nil {
951 return nil, errors.New("resident manager: unavailable")
952 }
953 dir, release, err := m.readableResidentDir(childID)
954 if err != nil {
955 return nil, err
956 }
957 defer release()
958 return ReadResidentHistory(dir, limit)
959}
960
961// HistoryPage reads one recent-first, bounded page through the manager-owned
962// state root. Public consumers identify only a child and receive an opaque
963// cursor for older content.
964func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
965 if m == nil {
966 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
967 }
968 dir, release, err := m.readableResidentDir(childID)
969 if err != nil {
970 return ResidentHistoryPage{}, err
971 }
972 defer release()
973 return ReadResidentHistoryPage(dir, olderCursor, limit)
974}
975
976// Result returns the bounded latest-turn projection for a resident child.
977func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
978 if m == nil {
979 return ResidentResult{}, errors.New("resident manager: unavailable")
980 }
981 dir, release, err := m.readableResidentDir(childID)
982 if err != nil {
983 return ResidentResult{}, err
984 }
985 defer release()
986 return ReadResidentResult(filepath.Join(dir, residentResultName))
987}
988
989func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
990 dir, err := ResidentHistoryDir(m.root, childID)
991 if err != nil {
992 return "", nil, err
993 }
994 childID = strings.TrimSpace(childID)
995 m.mu.Lock()
996 live := m.children[childID] != nil
997 m.mu.Unlock()
998 if live {
999 return dir, func() {}, nil
1000 }
1001 lease, err := acquireResidentLease(dir)
1002 if err != nil {
1003 if errors.Is(err, ErrResidentLeaseBusy) {
1004 return "", nil, errors.New("resident manager: child is owned by another zut process")
1005 }
1006 return "", nil, err
1007 }
1008 return dir, func() { _ = lease.Close() }, nil
1009}
1010
1011func (m *ResidentManager) Close(ctx context.Context) error {
1012 if m == nil {
1013 return nil
1014 }
1015 m.mu.Lock()
1016 m.closed = true
1017 children := make([]*ResidentChild, 0, len(m.children))
1018 for _, child := range m.children {
1019 children = append(children, child)
1020 }
1021 m.mu.Unlock()
1022 var errs []error
1023 for _, child := range children {
1024 if err := child.Close(ctx); err != nil {
1025 errs = append(errs, err)
1026 }
1027 }
1028 return errors.Join(errs...)
1029}