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