1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "sort"
10 "strings"
11 "sync"
12 "time"
13
14 "github.com/bnema/zut/packages/provider"
15 "github.com/google/uuid"
16)
17
18// ResidentFactory is the host-owned construction boundary. It receives only
19// an already-resolved non-secret specification and returns an in-process turn
20// runner; it never starts another zut binary.
21type ResidentFactory func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error)
22
23type ResidentManager struct {
24 root string
25 factory ResidentFactory
26 prepareWorkspace func(context.Context, WorkspaceRequest) (WorkspaceHandle, error)
27 mu sync.Mutex
28 lifecycleMu sync.Mutex
29 closed bool
30 children map[string]*ResidentChild
31 recovered map[string]ResidentSnapshot
32 recoveredSpec map[string]ResidentChildSpec
33 pending map[string]struct{}
34 activeChildren map[string]struct{}
35 scheduler *ResidentScheduler
36 queueTimeout time.Duration
37 allowedRoots []string
38 dispatchMu sync.Mutex
39 onCompletion func(ResidentCompletion)
40 completionWaiters map[string][]chan ResidentCompletion
41 onAccepted func(ResidentChildSpec, string, string)
42 onUpdate func(string)
43 onHistoryUpdate func(string)
44 onActivity func(bool)
45}
46
47func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion)) {
48 if m == nil {
49 return
50 }
51 m.mu.Lock()
52 m.onCompletion = observer
53 m.mu.Unlock()
54}
55
56// WatchCompletion registers for one accepted resident turn. The caller must
57// register before making the turn visible, then call the returned cancel
58// function if it stops waiting before completion.
59func (m *ResidentManager) WatchCompletion(childID, turnID string) (<-chan ResidentCompletion, func()) {
60 result := make(chan ResidentCompletion, 1)
61 key := completionKey(childID, turnID)
62 if m == nil || key == "" {
63 close(result)
64 return result, func() {}
65 }
66 m.mu.Lock()
67 if m.closed {
68 m.mu.Unlock()
69 close(result)
70 return result, func() {}
71 }
72 if m.completionWaiters == nil {
73 m.completionWaiters = make(map[string][]chan ResidentCompletion)
74 }
75 m.completionWaiters[key] = append(m.completionWaiters[key], result)
76 m.mu.Unlock()
77 return result, func() {
78 m.mu.Lock()
79 waiters := m.completionWaiters[key]
80 for i, waiter := range waiters {
81 if waiter == result {
82 waiters = append(waiters[:i], waiters[i+1:]...)
83 break
84 }
85 }
86 if len(waiters) == 0 {
87 delete(m.completionWaiters, key)
88 } else {
89 m.completionWaiters[key] = waiters
90 }
91 m.mu.Unlock()
92 }
93}
94
95func (m *ResidentManager) reportCompletion(completion ResidentCompletion) {
96 if m == nil {
97 return
98 }
99 key := completionKey(completion.ChildID, completion.TurnID)
100 m.mu.Lock()
101 waiters := m.completionWaiters[key]
102 delete(m.completionWaiters, key)
103 observer := m.onCompletion
104 m.mu.Unlock()
105 for _, waiter := range waiters {
106 waiter <- completion
107 close(waiter)
108 }
109 if observer != nil {
110 observer(completion)
111 }
112}
113
114// SetAcceptedObserver receives an accepted turn's ID and prompt only after
115// the authoritative child.accepted record has synchronized and before it can
116// be scheduled.
117func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string, string)) {
118 if m == nil {
119 return
120 }
121 m.mu.Lock()
122 m.onAccepted = observer
123 m.mu.Unlock()
124}
125
126// SetUpdateObserver receives child IDs whenever their visible state or live
127// projection changes. Observers run without manager or child locks and must
128// return promptly.
129func (m *ResidentManager) SetUpdateObserver(observer func(string)) {
130 if m == nil {
131 return
132 }
133 m.mu.Lock()
134 m.onUpdate = observer
135 m.mu.Unlock()
136}
137
138// SetHistoryUpdateObserver receives a child ID after a finalized event is
139// durably appended to its transcript. Stream deltas and state changes do not
140// trigger it, so observers can safely schedule bounded history reloads.
141func (m *ResidentManager) SetHistoryUpdateObserver(observer func(string)) {
142 if m == nil {
143 return
144 }
145 m.mu.Lock()
146 m.onHistoryUpdate = observer
147 m.mu.Unlock()
148}
149
150// SetActivityObserver receives the aggregate running-child state immediately
151// and after each lifecycle transition. It does not receive stream events, so
152// UI tickers can consume it without polling resident state.
153func (m *ResidentManager) SetActivityObserver(observer func(active bool)) {
154 if m == nil {
155 return
156 }
157 m.mu.Lock()
158 m.onActivity = observer
159 active := len(m.activeChildren) > 0
160 m.mu.Unlock()
161 if observer != nil {
162 observer(active)
163 }
164}
165
166func (m *ResidentManager) notifyActivity(childID string, state ResidentState) {
167 if m == nil {
168 return
169 }
170 activeChild := state == ResidentRunning
171 m.mu.Lock()
172 if m.activeChildren == nil {
173 m.activeChildren = make(map[string]struct{})
174 }
175 _, wasActive := m.activeChildren[childID]
176 if activeChild {
177 m.activeChildren[childID] = struct{}{}
178 } else {
179 delete(m.activeChildren, childID)
180 }
181 observer := m.onActivity
182 active := len(m.activeChildren) > 0
183 m.mu.Unlock()
184 if wasActive != activeChild && observer != nil {
185 observer(active)
186 }
187}
188
189func (m *ResidentManager) notifyUpdate(childID string, historyChanged bool) {
190 if m == nil {
191 return
192 }
193 m.mu.Lock()
194 observer := m.onUpdate
195 historyObserver := m.onHistoryUpdate
196 m.mu.Unlock()
197 if observer != nil {
198 observer(childID)
199 }
200 if historyChanged && historyObserver != nil {
201 historyObserver(childID)
202 }
203}
204
205// ResidentSnapshot is the manager's bounded public state projection. It
206// deliberately omits prompts, provider credentials, and filesystem paths.
207type ResidentSnapshot struct {
208 ID string
209 State ResidentState
210 Profile string
211 Provider string
212 Model string
213 WorkspaceMode WorkspaceMode
214 Required bool
215 UpdatedAt time.Time
216 TurnStartedAt time.Time
217 ActivityUpdatedAt time.Time
218 WaitingForModel bool
219 OwnedElsewhere bool
220 Usage provider.Usage
221 ContextUsed int
222 ContextMax int
223 Subscription bool
224}
225
226func residentSnapshot(child *ResidentChild) ResidentSnapshot {
227 if child == nil {
228 return ResidentSnapshot{}
229 }
230 live := child.Live()
231 return ResidentSnapshot{ID: child.spec.ID, State: child.State(), Profile: child.spec.Profile,
232 Provider: child.spec.Provider, Model: child.spec.Model,
233 WorkspaceMode: child.spec.WorkspaceMode, Required: child.spec.Required,
234 UpdatedAt: child.StateUpdatedAt(), TurnStartedAt: child.TurnStartedAt(),
235 ActivityUpdatedAt: child.ActivityUpdatedAt(), WaitingForModel: live.WaitingForModel,
236 Usage: live.Usage, ContextUsed: live.ContextUsed, ContextMax: live.ContextMax, Subscription: live.Subscription}
237}
238
239func NewResidentManager(root string, factory ResidentFactory) *ResidentManager {
240 return NewResidentManagerWithLimit(root, 0, factory)
241}
242
243func NewResidentManagerWithLimit(root string, limit int, factory ResidentFactory) *ResidentManager {
244 return NewResidentManagerWithWorkspace(root, limit, PrepareWorkspace, factory)
245}
246
247// NewResidentManagerWithPolicy applies the resident-only host policy. A zero
248// queue timeout leaves accepted work queued until a scheduler slot is free.
249func NewResidentManagerWithPolicy(root string, policy SubagentPolicy, factory ResidentFactory) *ResidentManager {
250 return newResidentManager(root, policy, PrepareWorkspace, factory)
251}
252
253// NewResidentManagerWithWorkspace exposes the narrow allocation seam needed
254// to make workspace preparation part of spawn's durable commit boundary.
255func NewResidentManagerWithWorkspace(root string, limit int, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
256 return newResidentManager(root, SubagentPolicy{MaxConcurrent: limit}, prepare, factory)
257}
258
259func newResidentManager(root string, policy SubagentPolicy, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
260 if prepare == nil {
261 prepare = PrepareWorkspace
262 }
263 return &ResidentManager{
264 root: root,
265 factory: factory,
266 prepareWorkspace: prepare,
267 children: make(map[string]*ResidentChild),
268 recovered: make(map[string]ResidentSnapshot),
269 recoveredSpec: make(map[string]ResidentChildSpec),
270 pending: make(map[string]struct{}),
271 activeChildren: make(map[string]struct{}),
272 completionWaiters: make(map[string][]chan ResidentCompletion),
273 scheduler: NewResidentScheduler(policy.MaxConcurrent),
274 queueTimeout: policy.QueueTimeout,
275 allowedRoots: append([]string(nil), policy.AllowedRoots...),
276 }
277}
278
279// Spawn commits acceptance before publishing a child or starting its runner.
280func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, task string) (*ResidentChild, error) {
281 if m == nil || m.factory == nil {
282 return nil, errors.New("resident manager: no factory")
283 }
284 m.lifecycleMu.Lock()
285 defer m.lifecycleMu.Unlock()
286 m.mu.Lock()
287 if m.closed {
288 m.mu.Unlock()
289 return nil, errors.New("resident manager: closed")
290 }
291 if _, exists := m.children[spec.ID]; exists {
292 m.mu.Unlock()
293 return nil, errors.New("resident manager: duplicate child ID")
294 }
295 if _, exists := m.recovered[spec.ID]; exists {
296 m.mu.Unlock()
297 return nil, errors.New("resident manager: child journal already exists")
298 }
299 if _, exists := m.pending[spec.ID]; exists {
300 m.mu.Unlock()
301 return nil, errors.New("resident manager: child acceptance pending")
302 }
303 m.pending[spec.ID] = struct{}{}
304 m.mu.Unlock()
305 defer func() { m.mu.Lock(); delete(m.pending, spec.ID); m.mu.Unlock() }()
306 if spec.InitialTurnID == "" {
307 spec.InitialTurnID = uuid.NewString()
308 }
309 journal, err := OpenResidentJournal(m.root, spec.ID)
310 if err != nil {
311 return nil, err
312 }
313 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
314 if workspaceRoot == "" {
315 workspaceRoot = strings.TrimSpace(spec.Workspace)
316 }
317 if !m.workspaceAllowed(workspaceRoot) {
318 _ = journal.Close()
319 return nil, errors.New("resident manager: workspace is outside allowed roots")
320 }
321 var workspace WorkspaceHandle
322 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
323 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
324 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, spec.ID),
325 AgentID: spec.ID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots,
326 })
327 if err != nil {
328 _ = journal.Close()
329 return nil, fmt.Errorf("resident manager prepare workspace: %w", err)
330 }
331 spec.Workspace = workspace.Dir()
332 spec.WorkspaceMode = workspace.Mode()
333 }
334 if err := journal.Accept(spec, task); err != nil {
335 if workspace != nil {
336 _ = workspace.Cleanup(context.Background())
337 }
338 _ = journal.Close()
339 return nil, err
340 }
341 m.mu.Lock()
342 accepted := m.onAccepted
343 m.mu.Unlock()
344 if accepted != nil {
345 accepted(spec, spec.InitialTurnID, task)
346 }
347 runner, err := m.factory(spec, journal)
348 if err != nil {
349 _ = journal.RecordFailure(spec)
350 _ = journal.Close()
351 m.reportCompletion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
352 return nil, err
353 }
354 child := newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
355 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
356 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
357 m.mu.Lock()
358 if m.closed {
359 m.mu.Unlock()
360 _ = child.Close(context.Background())
361 return nil, errors.New("resident manager: closed")
362 }
363 m.children[spec.ID] = child
364 m.mu.Unlock()
365 if childAccepted, err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
366 m.mu.Lock()
367 if m.children[spec.ID] == child {
368 delete(m.children, spec.ID)
369 }
370 m.mu.Unlock()
371 _ = child.Close(context.Background())
372 if !childAccepted {
373 _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
374 m.reportCompletion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
375 }
376 return nil, err
377 }
378 return child, nil
379}
380
381func (m *ResidentManager) scheduledRunner(childID string, runner ResidentTurnRunner) ResidentTurnRunner {
382 return func(ctx context.Context, prompt string) error {
383 if ctx == nil {
384 ctx = context.Background()
385 }
386 ticket := m.scheduler.Enqueue(childID, prompt)
387 m.dispatch()
388 var timeout <-chan time.Time
389 var timer *time.Timer
390 if m.queueTimeout > 0 {
391 timer = time.NewTimer(m.queueTimeout)
392 timeout = timer.C
393 defer timer.Stop()
394 }
395 select {
396 case <-ctx.Done():
397 if m.scheduler.Cancel(ticket.Sequence) {
398 m.dispatch()
399 return ctx.Err()
400 }
401 // Admission won the race with cancellation. Consume the handoff
402 // and release its reservation without running provider work.
403 <-ticket.ready
404 _ = m.scheduler.Release(childID)
405 m.dispatch()
406 return ctx.Err()
407 case <-ticket.ready:
408 case <-timeout:
409 if m.scheduler.Cancel(ticket.Sequence) {
410 m.dispatch()
411 return fmt.Errorf("resident manager: queue timeout after %s", m.queueTimeout)
412 }
413 // Admission won the race with the timeout. The reservation is now
414 // owned by this runner and must be released below.
415 <-ticket.ready
416 }
417 defer func() {
418 _ = m.scheduler.Release(childID)
419 m.dispatch()
420 }()
421 return runner(ctx, prompt)
422 }
423}
424
425func (m *ResidentManager) dispatch() {
426 if m == nil || m.scheduler == nil {
427 return
428 }
429 m.dispatchMu.Lock()
430 defer m.dispatchMu.Unlock()
431 for {
432 if _, ok := m.scheduler.Admit(); !ok {
433 return
434 }
435 }
436}
437
438// Resume synchronizes a follow-up acceptance before making it visible to the
439// resident child. A disk-only child is rebuilt only for this explicit new
440// prompt; reconciliation itself never constructs or replays one.
441func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) error {
442 return m.ResumeWithTurn(ctx, childID, prompt, uuid.NewString())
443}
444
445// ResumeWithTurn is Resume with a caller-owned turn ID. The ID names the
446// accepted follow-up turn in its completion, so a caller that must observe the
447// outcome can subscribe with WatchCompletion before the turn can finish. The
448// ID must be unique per child: replay rejects a duplicate turn.accepted record,
449// so reusing one makes the journal unreplayable.
450func (m *ResidentManager) ResumeWithTurn(ctx context.Context, childID, prompt, turnID string) error {
451 if strings.TrimSpace(turnID) == "" {
452 turnID = uuid.NewString()
453 }
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 {
478 if _, pending := m.pending[childID]; pending {
479 m.mu.Unlock()
480 return errors.New("resident manager: child resume pending")
481 }
482 m.pending[childID] = struct{}{}
483 }
484 m.mu.Unlock()
485 if child == nil && recovered {
486 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
487 journal, err := OpenResidentJournal(m.root, childID)
488 if err != nil {
489 if errors.Is(err, ErrResidentLeaseBusy) {
490 return errors.New("resident manager: child is owned by another zut process")
491 }
492 return err
493 }
494 metadata, err := reconcileOwnedResidentJournal(journal)
495 if err != nil {
496 _ = journal.Close()
497 return err
498 }
499 journal.ConfigureUsage(metadata.ContextMax, metadata.Subscription)
500 records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
501 if err != nil || len(records) == 0 || records[0].Spec == nil {
502 _ = journal.Close()
503 return errors.New("resident manager: invalid accepted specification")
504 }
505 spec = *records[0].Spec
506 var workspace WorkspaceHandle
507 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
508 if workspaceRoot == "" {
509 workspaceRoot = strings.TrimSpace(spec.Workspace)
510 }
511 if !m.workspaceAllowed(workspaceRoot) {
512 _ = journal.Close()
513 return errors.New("resident manager: workspace is outside allowed roots")
514 }
515 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
516 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
517 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, childID),
518 AgentID: childID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots, ExistingPath: spec.Workspace,
519 })
520 if err != nil {
521 _ = journal.Close()
522 return fmt.Errorf("resident manager rebuild workspace: %w", err)
523 }
524 spec.Workspace = workspace.Dir()
525 spec.WorkspaceMode = workspace.Mode()
526 }
527 runner, err := m.factory(spec, journal)
528 if err != nil {
529 if workspace != nil && workspace.Mode() == WorkspaceShared {
530 _ = workspace.Cleanup(context.Background())
531 }
532 _ = journal.Close()
533 return fmt.Errorf("resident manager rebuild child: %w", err)
534 }
535 child = newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
536 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
537 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
538 m.mu.Lock()
539 if m.closed {
540 m.mu.Unlock()
541 _ = child.Close(context.Background())
542 return errors.New("resident manager: closed")
543 }
544 m.children[childID] = child
545 delete(m.recovered, childID)
546 delete(m.recoveredSpec, childID)
547 m.mu.Unlock()
548 }
549 if child == nil {
550 return errors.New("resident manager: child is not live")
551 }
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 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 // Unknown journal versions cannot be resumed. Leave them on disk
855 // without reporting the same incompatibility at every host startup.
856 // Match the exact sentinel so a joined journal-close error is still
857 // reported.
858 if reconcileErr == ErrIncompatibleResidentJournal {
859 continue
860 }
861 if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
862 foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
863 if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
864 errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
865 continue
866 }
867 if readErr == nil {
868 if foreign.Version != residentJournalVersion && foreign.Version != residentJournalVersionV2 {
869 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, ErrIncompatibleResidentJournal))
870 continue
871 }
872 // A v3 projection can never carry the removed budget
873 // state: the writer only persists live states. Treat it
874 // as corruption rather than resurfacing the zombie state.
875 if foreign.Version != residentJournalVersionV2 && foreign.State == ResidentState(residentV2OutcomeBudgetExhausted) {
876 errs = append(errs, fmt.Errorf("resident child %s: invalid terminal state", childID))
877 continue
878 }
879 }
880 state, updatedAt := foreign.State, foreign.UpdatedAt
881 if readErr != nil || state == "" {
882 // Ownership begins before child.accepted is durable. Do not read
883 // that live transcript merely to improve this transient snapshot.
884 state, updatedAt = ResidentQueued, time.Now().UTC()
885 }
886 if foreign.Version == residentJournalVersionV2 && state == ResidentState(residentV2OutcomeBudgetExhausted) {
887 // Historical budget stop from an older host: display it as
888 // failed, never as a distinct terminal state.
889 state = ResidentFailed
890 }
891 m.mu.Lock()
892 m.recovered[childID] = ResidentSnapshot{
893 ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true,
894 Usage: foreign.Usage, ContextUsed: foreign.ContextUsed, ContextMax: foreign.ContextMax, Subscription: foreign.Subscription,
895 }
896 m.mu.Unlock()
897 continue
898 }
899 if reconcileErr != nil {
900 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
901 continue
902 }
903 m.mu.Lock()
904 if _, live := m.children[childID]; !live {
905 m.recovered[childID] = ResidentSnapshot{
906 ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model,
907 WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt,
908 Usage: metadata.Usage, ContextUsed: metadata.ContextUsed, ContextMax: metadata.ContextMax, Subscription: metadata.Subscription,
909 }
910 m.recoveredSpec[childID] = spec
911 }
912 m.mu.Unlock()
913 }
914 return errs
915}
916
917// UnmetRequired returns required work that has not reached the successful idle
918// boundary. Failed and interrupted work intentionally remain unmet until an
919// explicit successful follow-up is accepted and completed.
920func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
921 all := m.Snapshot()
922 result := all[:0]
923 for _, snapshot := range all {
924 if snapshot.Required && snapshot.State != ResidentIdle && snapshot.State != ResidentCompleted {
925 result = append(result, snapshot)
926 }
927 }
928 return result
929}
930
931// Stop cancels one live child and waits for its control loop. A disk-only
932// interrupted child must be explicitly resumed by host construction instead.
933func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
934 if m == nil {
935 return errors.New("resident manager: unavailable")
936 }
937 m.lifecycleMu.Lock()
938 defer m.lifecycleMu.Unlock()
939 childID = strings.TrimSpace(childID)
940 m.mu.Lock()
941 child := m.children[childID]
942 m.mu.Unlock()
943 if child == nil {
944 return errors.New("resident manager: child is not live")
945 }
946 if err := child.Close(ctx); err != nil {
947 return err
948 }
949 // A stopped child has released its resident agent and journal. Retain only
950 // its durable spec so an explicit follow-up can rebuild a fresh resident
951 // child; keeping the closed child live would make Resume append to a closed
952 // journal.
953 m.mu.Lock()
954 if m.children[childID] == child {
955 delete(m.children, childID)
956 m.recovered[childID] = residentSnapshot(child)
957 m.recoveredSpec[childID] = child.spec
958 }
959 m.mu.Unlock()
960 m.notifyUpdate(childID, false)
961 return nil
962}
963
964// History reads bounded finalized resident history through the manager-owned
965// state root. It accepts only a child ID, never a filesystem path.
966func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
967 if m == nil {
968 return nil, errors.New("resident manager: unavailable")
969 }
970 dir, release, err := m.readableResidentDir(childID)
971 if err != nil {
972 return nil, err
973 }
974 defer release()
975 return ReadResidentHistory(dir, limit)
976}
977
978// HistoryPage reads one recent-first, bounded page through the manager-owned
979// state root. Public consumers identify only a child and receive an opaque
980// cursor for older content.
981func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
982 if m == nil {
983 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
984 }
985 dir, release, err := m.readableResidentDir(childID)
986 if err != nil {
987 return ResidentHistoryPage{}, err
988 }
989 defer release()
990 return ReadResidentHistoryPage(dir, olderCursor, limit)
991}
992
993// Result returns the bounded latest-turn projection for a resident child.
994func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
995 if m == nil {
996 return ResidentResult{}, errors.New("resident manager: unavailable")
997 }
998 dir, release, err := m.readableResidentDir(childID)
999 if err != nil {
1000 return ResidentResult{}, err
1001 }
1002 defer release()
1003 return ReadResidentResult(filepath.Join(dir, residentResultName))
1004}
1005
1006func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
1007 dir, err := ResidentHistoryDir(m.root, childID)
1008 if err != nil {
1009 return "", nil, err
1010 }
1011 childID = strings.TrimSpace(childID)
1012 m.mu.Lock()
1013 live := m.children[childID] != nil
1014 m.mu.Unlock()
1015 if live {
1016 return dir, func() {}, nil
1017 }
1018 lease, err := acquireResidentLease(dir)
1019 if err != nil {
1020 if errors.Is(err, ErrResidentLeaseBusy) {
1021 return "", nil, errors.New("resident manager: child is owned by another zut process")
1022 }
1023 return "", nil, err
1024 }
1025 return dir, func() { _ = lease.Close() }, nil
1026}
1027
1028func (m *ResidentManager) Close(ctx context.Context) error {
1029 if m == nil {
1030 return nil
1031 }
1032 m.mu.Lock()
1033 m.closed = true
1034 children := make([]*ResidentChild, 0, len(m.children))
1035 for _, child := range m.children {
1036 children = append(children, child)
1037 }
1038 m.mu.Unlock()
1039 var errs []error
1040 for _, child := range children {
1041 if err := child.Close(ctx); err != nil {
1042 errs = append(errs, err)
1043 }
1044 }
1045 return errors.Join(errs...)
1046}