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/google/uuid"
15)
16
17// ResidentFactory is the host-owned construction boundary. It receives only
18// an already-resolved non-secret specification and returns an in-process turn
19// runner; it never starts another zut binary.
20type ResidentFactory func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error)
21
22type ResidentManager struct {
23 root string
24 factory ResidentFactory
25 prepareWorkspace func(context.Context, WorkspaceRequest) (WorkspaceHandle, error)
26 mu sync.Mutex
27 lifecycleMu sync.Mutex
28 closed bool
29 children map[string]*ResidentChild
30 recovered map[string]ResidentSnapshot
31 recoveredSpec map[string]ResidentChildSpec
32 pending map[string]struct{}
33 activeChildren map[string]struct{}
34 scheduler *ResidentScheduler
35 queueTimeout time.Duration
36 allowedRoots []string
37 dispatchMu sync.Mutex
38 onCompletion func(ResidentCompletion)
39 onAccepted func(ResidentChildSpec, string)
40 onUpdate func(string)
41 onHistoryUpdate func(string)
42 onActivity func(bool)
43}
44
45func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion)) {
46 if m == nil {
47 return
48 }
49 m.mu.Lock()
50 m.onCompletion = observer
51 m.mu.Unlock()
52}
53
54// SetAcceptedObserver receives a prompt only after the authoritative
55// child.accepted record has synchronized, and before it can be scheduled.
56func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string)) {
57 if m == nil {
58 return
59 }
60 m.mu.Lock()
61 m.onAccepted = observer
62 m.mu.Unlock()
63}
64
65// SetUpdateObserver receives child IDs whenever their visible state or live
66// projection changes. Observers run without manager or child locks and must
67// return promptly.
68func (m *ResidentManager) SetUpdateObserver(observer func(string)) {
69 if m == nil {
70 return
71 }
72 m.mu.Lock()
73 m.onUpdate = observer
74 m.mu.Unlock()
75}
76
77// SetHistoryUpdateObserver receives a child ID after a finalized event is
78// durably appended to its transcript. Stream deltas and state changes do not
79// trigger it, so observers can safely schedule bounded history reloads.
80func (m *ResidentManager) SetHistoryUpdateObserver(observer func(string)) {
81 if m == nil {
82 return
83 }
84 m.mu.Lock()
85 m.onHistoryUpdate = observer
86 m.mu.Unlock()
87}
88
89// SetActivityObserver receives the aggregate running-child state immediately
90// and after each lifecycle transition. It does not receive stream events, so
91// UI tickers can consume it without polling resident state.
92func (m *ResidentManager) SetActivityObserver(observer func(active bool)) {
93 if m == nil {
94 return
95 }
96 m.mu.Lock()
97 m.onActivity = observer
98 active := len(m.activeChildren) > 0
99 m.mu.Unlock()
100 if observer != nil {
101 observer(active)
102 }
103}
104
105func (m *ResidentManager) notifyActivity(childID string, state ResidentState) {
106 if m == nil {
107 return
108 }
109 activeChild := state == ResidentRunning
110 m.mu.Lock()
111 if m.activeChildren == nil {
112 m.activeChildren = make(map[string]struct{})
113 }
114 _, wasActive := m.activeChildren[childID]
115 if activeChild {
116 m.activeChildren[childID] = struct{}{}
117 } else {
118 delete(m.activeChildren, childID)
119 }
120 observer := m.onActivity
121 active := len(m.activeChildren) > 0
122 m.mu.Unlock()
123 if wasActive != activeChild && observer != nil {
124 observer(active)
125 }
126}
127
128func (m *ResidentManager) notifyUpdate(childID string, historyChanged bool) {
129 if m == nil {
130 return
131 }
132 m.mu.Lock()
133 observer := m.onUpdate
134 historyObserver := m.onHistoryUpdate
135 m.mu.Unlock()
136 if observer != nil {
137 observer(childID)
138 }
139 if historyChanged && historyObserver != nil {
140 historyObserver(childID)
141 }
142}
143
144// ResidentSnapshot is the manager's bounded public state projection. It
145// deliberately omits prompts, provider credentials, and filesystem paths.
146type ResidentSnapshot struct {
147 ID string
148 State ResidentState
149 Profile string
150 Provider string
151 Model string
152 WorkspaceMode WorkspaceMode
153 Required bool
154 UpdatedAt time.Time
155 TurnStartedAt time.Time
156 ActivityUpdatedAt time.Time
157 WaitingForModel bool
158}
159
160func residentSnapshot(child *ResidentChild) ResidentSnapshot {
161 if child == nil {
162 return ResidentSnapshot{}
163 }
164 live := child.Live()
165 return ResidentSnapshot{ID: child.spec.ID, State: child.State(), Profile: child.spec.Profile,
166 Provider: child.spec.Provider, Model: child.spec.Model,
167 WorkspaceMode: child.spec.WorkspaceMode, Required: child.spec.Required,
168 UpdatedAt: child.StateUpdatedAt(), TurnStartedAt: child.TurnStartedAt(),
169 ActivityUpdatedAt: child.ActivityUpdatedAt(), WaitingForModel: live.WaitingForModel}
170}
171
172func NewResidentManager(root string, factory ResidentFactory) *ResidentManager {
173 return NewResidentManagerWithLimit(root, 0, factory)
174}
175
176func NewResidentManagerWithLimit(root string, limit int, factory ResidentFactory) *ResidentManager {
177 return NewResidentManagerWithWorkspace(root, limit, PrepareWorkspace, factory)
178}
179
180// NewResidentManagerWithPolicy applies the resident-only host policy. A zero
181// queue timeout leaves accepted work queued until a scheduler slot is free.
182func NewResidentManagerWithPolicy(root string, policy SubagentPolicy, factory ResidentFactory) *ResidentManager {
183 return newResidentManager(root, policy, PrepareWorkspace, factory)
184}
185
186// NewResidentManagerWithWorkspace exposes the narrow allocation seam needed
187// to make workspace preparation part of spawn's durable commit boundary.
188func NewResidentManagerWithWorkspace(root string, limit int, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
189 return newResidentManager(root, SubagentPolicy{MaxConcurrent: limit}, prepare, factory)
190}
191
192func newResidentManager(root string, policy SubagentPolicy, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
193 if prepare == nil {
194 prepare = PrepareWorkspace
195 }
196 return &ResidentManager{
197 root: root,
198 factory: factory,
199 prepareWorkspace: prepare,
200 children: make(map[string]*ResidentChild),
201 recovered: make(map[string]ResidentSnapshot),
202 recoveredSpec: make(map[string]ResidentChildSpec),
203 pending: make(map[string]struct{}),
204 activeChildren: make(map[string]struct{}),
205 scheduler: NewResidentScheduler(policy.MaxConcurrent),
206 queueTimeout: policy.QueueTimeout,
207 allowedRoots: append([]string(nil), policy.AllowedRoots...),
208 }
209}
210
211// Spawn commits acceptance before publishing a child or starting its runner.
212func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, task string) (*ResidentChild, error) {
213 if m == nil || m.factory == nil {
214 return nil, errors.New("resident manager: no factory")
215 }
216 m.lifecycleMu.Lock()
217 defer m.lifecycleMu.Unlock()
218 m.mu.Lock()
219 if m.closed {
220 m.mu.Unlock()
221 return nil, errors.New("resident manager: closed")
222 }
223 if _, exists := m.children[spec.ID]; exists {
224 m.mu.Unlock()
225 return nil, errors.New("resident manager: duplicate child ID")
226 }
227 if _, exists := m.recovered[spec.ID]; exists {
228 m.mu.Unlock()
229 return nil, errors.New("resident manager: child journal already exists")
230 }
231 if _, exists := m.pending[spec.ID]; exists {
232 m.mu.Unlock()
233 return nil, errors.New("resident manager: child acceptance pending")
234 }
235 m.pending[spec.ID] = struct{}{}
236 m.mu.Unlock()
237 defer func() { m.mu.Lock(); delete(m.pending, spec.ID); m.mu.Unlock() }()
238 if spec.InitialTurnID == "" {
239 spec.InitialTurnID = uuid.NewString()
240 }
241 journal, err := OpenResidentJournal(m.root, spec.ID)
242 if err != nil {
243 return nil, err
244 }
245 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
246 if workspaceRoot == "" {
247 workspaceRoot = strings.TrimSpace(spec.Workspace)
248 }
249 if !m.workspaceAllowed(workspaceRoot) {
250 _ = journal.Close()
251 return nil, errors.New("resident manager: workspace is outside allowed roots")
252 }
253 var workspace WorkspaceHandle
254 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
255 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
256 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, spec.ID),
257 AgentID: spec.ID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots,
258 })
259 if err != nil {
260 _ = journal.Close()
261 return nil, fmt.Errorf("resident manager prepare workspace: %w", err)
262 }
263 spec.Workspace = workspace.Dir()
264 spec.WorkspaceMode = workspace.Mode()
265 }
266 if err := journal.Accept(spec, task); err != nil {
267 if workspace != nil {
268 _ = workspace.Cleanup(context.Background())
269 }
270 _ = journal.Close()
271 return nil, err
272 }
273 m.mu.Lock()
274 accepted := m.onAccepted
275 m.mu.Unlock()
276 if accepted != nil {
277 accepted(spec, task)
278 }
279 runner, err := m.factory(spec, journal)
280 if err != nil {
281 _ = journal.RecordFailure(spec)
282 _ = journal.Close()
283 m.mu.Lock()
284 completion := m.onCompletion
285 m.mu.Unlock()
286 if completion != nil {
287 completion(ResidentCompletion{ChildID: spec.ID, Task: task, Err: err})
288 }
289 return nil, err
290 }
291 m.mu.Lock()
292 observer := m.onCompletion
293 m.mu.Unlock()
294 child := newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), observer)
295 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
296 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
297 m.mu.Lock()
298 if m.closed {
299 m.mu.Unlock()
300 _ = child.Close(context.Background())
301 return nil, errors.New("resident manager: closed")
302 }
303 m.children[spec.ID] = child
304 m.mu.Unlock()
305 if err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
306 _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
307 m.mu.Lock()
308 if m.children[spec.ID] == child {
309 delete(m.children, spec.ID)
310 }
311 m.mu.Unlock()
312 _ = child.Close(context.Background())
313 return nil, err
314 }
315 return child, nil
316}
317
318func (m *ResidentManager) scheduledRunner(childID string, runner ResidentTurnRunner) ResidentTurnRunner {
319 return func(ctx context.Context, prompt string) error {
320 if ctx == nil {
321 ctx = context.Background()
322 }
323 ticket := m.scheduler.Enqueue(childID, prompt)
324 m.dispatch()
325 var timeout <-chan time.Time
326 var timer *time.Timer
327 if m.queueTimeout > 0 {
328 timer = time.NewTimer(m.queueTimeout)
329 timeout = timer.C
330 defer timer.Stop()
331 }
332 select {
333 case <-ctx.Done():
334 if m.scheduler.Cancel(ticket.Sequence) {
335 m.dispatch()
336 return ctx.Err()
337 }
338 // Admission won the race with cancellation. Consume the handoff
339 // and release its reservation without running provider work.
340 <-ticket.ready
341 _ = m.scheduler.Release(childID)
342 m.dispatch()
343 return ctx.Err()
344 case <-ticket.ready:
345 case <-timeout:
346 if m.scheduler.Cancel(ticket.Sequence) {
347 m.dispatch()
348 return fmt.Errorf("resident manager: queue timeout after %s", m.queueTimeout)
349 }
350 // Admission won the race with the timeout. The reservation is now
351 // owned by this runner and must be released below.
352 <-ticket.ready
353 }
354 defer func() {
355 _ = m.scheduler.Release(childID)
356 m.dispatch()
357 }()
358 return runner(ctx, prompt)
359 }
360}
361
362func (m *ResidentManager) dispatch() {
363 if m == nil || m.scheduler == nil {
364 return
365 }
366 m.dispatchMu.Lock()
367 defer m.dispatchMu.Unlock()
368 for {
369 if _, ok := m.scheduler.Admit(); !ok {
370 return
371 }
372 }
373}
374
375// Resume synchronizes a follow-up acceptance before making it visible to the
376// resident child. A disk-only child is rebuilt only for this explicit new
377// prompt; reconciliation itself never constructs or replays one.
378func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) error {
379 if m == nil {
380 return errors.New("resident manager: unavailable")
381 }
382 m.lifecycleMu.Lock()
383 defer m.lifecycleMu.Unlock()
384 childID = strings.TrimSpace(childID)
385 prompt = strings.TrimSpace(prompt)
386 if childID == "" || prompt == "" {
387 return errors.New("resident manager: child ID and prompt are required")
388 }
389 m.mu.Lock()
390 if m.closed {
391 m.mu.Unlock()
392 return errors.New("resident manager: closed")
393 }
394 child := m.children[childID]
395 spec, recovered := m.recoveredSpec[childID]
396 if child == nil && recovered {
397 if _, pending := m.pending[childID]; pending {
398 m.mu.Unlock()
399 return errors.New("resident manager: child resume pending")
400 }
401 m.pending[childID] = struct{}{}
402 }
403 m.mu.Unlock()
404 if child == nil && recovered {
405 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
406 journal, err := OpenResidentJournal(m.root, childID)
407 if err != nil {
408 return err
409 }
410 var workspace WorkspaceHandle
411 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
412 if workspaceRoot == "" {
413 workspaceRoot = strings.TrimSpace(spec.Workspace)
414 }
415 if !m.workspaceAllowed(workspaceRoot) {
416 _ = journal.Close()
417 return errors.New("resident manager: workspace is outside allowed roots")
418 }
419 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
420 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
421 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, childID),
422 AgentID: childID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots, ExistingPath: spec.Workspace,
423 })
424 if err != nil {
425 _ = journal.Close()
426 return fmt.Errorf("resident manager rebuild workspace: %w", err)
427 }
428 spec.Workspace = workspace.Dir()
429 spec.WorkspaceMode = workspace.Mode()
430 }
431 runner, err := m.factory(spec, journal)
432 if err != nil {
433 if workspace != nil && workspace.Mode() == WorkspaceShared {
434 _ = workspace.Cleanup(context.Background())
435 }
436 _ = journal.Close()
437 return fmt.Errorf("resident manager rebuild child: %w", err)
438 }
439 m.mu.Lock()
440 observer := m.onCompletion
441 m.mu.Unlock()
442 child = newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), observer)
443 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
444 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
445 m.mu.Lock()
446 if m.closed {
447 m.mu.Unlock()
448 _ = child.Close(context.Background())
449 return errors.New("resident manager: closed")
450 }
451 m.children[childID] = child
452 delete(m.recovered, childID)
453 delete(m.recoveredSpec, childID)
454 m.mu.Unlock()
455 }
456 if child == nil {
457 return errors.New("resident manager: child is not live")
458 }
459 turnID := uuid.NewString()
460 if child.journal == nil {
461 return child.resumeAccepted(ctx, turnID, prompt)
462 }
463 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {
464 return err
465 }
466 m.mu.Lock()
467 accepted := m.onAccepted
468 m.mu.Unlock()
469 if accepted != nil {
470 accepted(child.spec, prompt)
471 }
472 if err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
473 _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
474 return err
475 }
476 return nil
477}
478
479func (m *ResidentManager) workspaceAllowed(root string) bool {
480 if len(m.allowedRoots) == 0 || strings.TrimSpace(root) == "" {
481 return true
482 }
483 for _, allowed := range m.allowedRoots {
484 if pathWithin(root, allowed) {
485 return true
486 }
487 }
488 return false
489}
490
491// Get returns the live child only. Durable children recovered from a previous
492// host are intentionally not auto-resumed.
493func (m *ResidentManager) Get(childID string) *ResidentChild {
494 if m == nil {
495 return nil
496 }
497 m.mu.Lock()
498 defer m.mu.Unlock()
499 return m.children[childID]
500}
501
502// Live returns an immutable unfinished-turn snapshot for one live child. It
503// never holds the manager lock while copying the child projection.
504func (m *ResidentManager) Live(childID string) (ResidentLiveSnapshot, bool) {
505 if m == nil {
506 return ResidentLiveSnapshot{}, false
507 }
508 m.mu.Lock()
509 child := m.children[childID]
510 m.mu.Unlock()
511 if child == nil {
512 return ResidentLiveSnapshot{}, false
513 }
514 return child.Live(), true
515}
516
517// Snapshot returns copies of the live children without holding the manager
518// lock while callers render or perform I/O.
519func (m *ResidentManager) Snapshot() []ResidentSnapshot {
520 children, _ := m.SnapshotPage(0, -1)
521 return children
522}
523
524// SnapshotPage returns a bounded, ID-sorted state projection. It avoids
525// materializing live snapshots that a renderer cannot display and reports the
526// total so callers can page without relying on a stale full list.
527func (m *ResidentManager) SnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
528 if m == nil {
529 return nil, 0
530 }
531 if offset < 0 {
532 offset = 0
533 }
534 m.mu.Lock()
535 live := make(map[string]*ResidentChild, len(m.children))
536 for id, child := range m.children {
537 live[id] = child
538 }
539 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
540 for id, snapshot := range m.recovered {
541 recovered[id] = snapshot
542 }
543 m.mu.Unlock()
544 ids := make([]string, 0, len(live)+len(recovered))
545 for id := range recovered {
546 ids = append(ids, id)
547 }
548 for id := range live {
549 if _, alreadyRecovered := recovered[id]; !alreadyRecovered {
550 ids = append(ids, id)
551 }
552 }
553 sort.Strings(ids)
554 total := len(ids)
555 if offset >= total || limit == 0 {
556 return nil, total
557 }
558 end := total
559 if limit > 0 && offset+limit < end {
560 end = offset + limit
561 }
562 result := make([]ResidentSnapshot, 0, end-offset)
563 for _, id := range ids[offset:end] {
564 if child := live[id]; child != nil {
565 result = append(result, residentSnapshot(child))
566 continue
567 }
568 if snapshot, ok := recovered[id]; ok {
569 result = append(result, snapshot)
570 }
571 }
572 return result, total
573}
574
575// RecentSnapshotPage returns a bounded state projection ordered by the latest
576// visible state transition, newest first. It is intended for the interactive
577// dashboard where operators need to identify the child that just changed.
578func (m *ResidentManager) RecentSnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
579 if m == nil {
580 return nil, 0
581 }
582 if offset < 0 {
583 offset = 0
584 }
585 m.mu.Lock()
586 live := make(map[string]*ResidentChild, len(m.children))
587 for id, child := range m.children {
588 live[id] = child
589 }
590 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
591 for id, snapshot := range m.recovered {
592 recovered[id] = snapshot
593 }
594 m.mu.Unlock()
595 ids := make([]string, 0, len(live)+len(recovered))
596 for id := range recovered {
597 ids = append(ids, id)
598 }
599 for id := range live {
600 if _, recoveredAlready := recovered[id]; !recoveredAlready {
601 ids = append(ids, id)
602 }
603 }
604 snapshots := make(map[string]ResidentSnapshot, len(ids))
605 for _, id := range ids {
606 if child := live[id]; child != nil {
607 snapshots[id] = residentSnapshot(child)
608 } else {
609 snapshots[id] = recovered[id]
610 }
611 }
612 sort.Slice(ids, func(i, j int) bool {
613 left, right := snapshots[ids[i]], snapshots[ids[j]]
614 if !left.UpdatedAt.Equal(right.UpdatedAt) {
615 return left.UpdatedAt.After(right.UpdatedAt)
616 }
617 return ids[i] < ids[j]
618 })
619 total := len(ids)
620 if offset >= total || limit == 0 {
621 return nil, total
622 }
623 end := total
624 if limit > 0 && offset+limit < end {
625 end = offset + limit
626 }
627 result := make([]ResidentSnapshot, 0, end-offset)
628 for _, id := range ids[offset:end] {
629 result = append(result, snapshots[id])
630 }
631 return result, total
632}
633
634// ActiveSnapshotPage returns only queued and running children for compact
635// activity surfaces. It deliberately exposes no prompt, transcript, or path.
636func (m *ResidentManager) ActiveSnapshotPage(limit int) ([]ResidentSnapshot, int) {
637 if m == nil {
638 return nil, 0
639 }
640 m.mu.Lock()
641 live := make(map[string]*ResidentChild, len(m.children))
642 for id, child := range m.children {
643 live[id] = child
644 }
645 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
646 for id, snapshot := range m.recovered {
647 recovered[id] = snapshot
648 }
649 m.mu.Unlock()
650
651 result := make([]ResidentSnapshot, 0, len(live)+len(recovered))
652 for _, child := range live {
653 snapshot := residentSnapshot(child)
654 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
655 result = append(result, snapshot)
656 }
657 }
658 for id, snapshot := range recovered {
659 if _, exists := live[id]; exists {
660 continue
661 }
662 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
663 result = append(result, snapshot)
664 }
665 }
666 sort.Slice(result, func(i, j int) bool {
667 if !result[i].UpdatedAt.Equal(result[j].UpdatedAt) {
668 return result[i].UpdatedAt.After(result[j].UpdatedAt)
669 }
670 return result[i].ID < result[j].ID
671 })
672 total := len(result)
673 if limit > 0 && len(result) > limit {
674 result = result[:limit]
675 }
676 return result, total
677}
678
679// SnapshotFor returns metadata for one child without building a dashboard
680// projection. It keeps child-session headers independent of the total number
681// of resident children.
682func (m *ResidentManager) SnapshotFor(childID string) (ResidentSnapshot, bool) {
683 if m == nil {
684 return ResidentSnapshot{}, false
685 }
686 m.mu.Lock()
687 child := m.children[childID]
688 recovered, ok := m.recovered[childID]
689 m.mu.Unlock()
690 if child != nil {
691 return residentSnapshot(child), true
692 }
693 return recovered, ok
694}
695
696// Reconcile discovers journals left by an earlier host, repairs their bounded
697// projections, and marks queued/running work interrupted. It never creates a
698// live child or replays a prompt. Per-child errors are returned for callers to
699// surface without hiding valid sibling journals.
700func (m *ResidentManager) Reconcile() []error {
701 if m == nil {
702 return []error{errors.New("resident manager: unavailable")}
703 }
704 m.lifecycleMu.Lock()
705 defer m.lifecycleMu.Unlock()
706 entries, err := os.ReadDir(m.root)
707 if err != nil {
708 if errors.Is(err, os.ErrNotExist) {
709 return nil
710 }
711 return []error{fmt.Errorf("resident manager read root: %w", err)}
712 }
713 sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
714 m.mu.Lock()
715 m.recovered = make(map[string]ResidentSnapshot)
716 m.recoveredSpec = make(map[string]ResidentChildSpec)
717 m.mu.Unlock()
718 var errs []error
719 for _, entry := range entries {
720 if !entry.IsDir() {
721 continue
722 }
723 // The old subprocess runtime stored children beneath an "agents"
724 // container. Resident children are direct entries in this root. Leave
725 // that legacy state untouched: it is neither read nor migrated.
726 if entry.Name() == "agents" {
727 continue
728 }
729 childID := entry.Name()
730 m.mu.Lock()
731 _, live := m.children[childID]
732 _, pending := m.pending[childID]
733 m.mu.Unlock()
734 if live || pending {
735 continue
736 }
737 metadata, reconcileErr := ReconcileResidentJournal(filepath.Join(m.root, childID))
738 if reconcileErr != nil {
739 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
740 continue
741 }
742 records, readErr := ReadResidentJournal(filepath.Join(m.root, childID, residentTranscriptName))
743 if readErr != nil || len(records) == 0 || records[0].Spec == nil {
744 errs = append(errs, fmt.Errorf("resident child %s: invalid accepted specification", childID))
745 continue
746 }
747 spec := records[0].Spec
748 m.mu.Lock()
749 if _, live := m.children[childID]; !live {
750 m.recovered[childID] = ResidentSnapshot{ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model, WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt}
751 m.recoveredSpec[childID] = *spec
752 }
753 m.mu.Unlock()
754 }
755 return errs
756}
757
758// UnmetRequired returns required work that has not reached the successful idle
759// boundary. Failed and interrupted work intentionally remain unmet until an
760// explicit successful follow-up is accepted and completed.
761func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
762 all := m.Snapshot()
763 result := all[:0]
764 for _, snapshot := range all {
765 if snapshot.Required && snapshot.State != ResidentIdle {
766 result = append(result, snapshot)
767 }
768 }
769 return result
770}
771
772// Stop cancels one live child and waits for its control loop. A disk-only
773// interrupted child must be explicitly resumed by host construction instead.
774func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
775 if m == nil {
776 return errors.New("resident manager: unavailable")
777 }
778 m.lifecycleMu.Lock()
779 defer m.lifecycleMu.Unlock()
780 childID = strings.TrimSpace(childID)
781 m.mu.Lock()
782 child := m.children[childID]
783 m.mu.Unlock()
784 if child == nil {
785 return errors.New("resident manager: child is not live")
786 }
787 if err := child.Close(ctx); err != nil {
788 return err
789 }
790 // A stopped child has released its resident agent and journal. Retain only
791 // its durable spec so an explicit follow-up can rebuild a fresh resident
792 // child; keeping the closed child live would make Resume append to a closed
793 // journal.
794 m.mu.Lock()
795 if m.children[childID] == child {
796 delete(m.children, childID)
797 m.recovered[childID] = residentSnapshot(child)
798 m.recoveredSpec[childID] = child.spec
799 }
800 m.mu.Unlock()
801 m.notifyUpdate(childID, false)
802 return nil
803}
804
805// History reads bounded finalized resident history through the manager-owned
806// state root. It accepts only a child ID, never a filesystem path.
807func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
808 if m == nil {
809 return nil, errors.New("resident manager: unavailable")
810 }
811 dir, err := ResidentHistoryDir(m.root, childID)
812 if err != nil {
813 return nil, err
814 }
815 return ReadResidentHistory(dir, limit)
816}
817
818// HistoryPage reads one recent-first, bounded page through the manager-owned
819// state root. Public consumers identify only a child and receive an opaque
820// cursor for older content.
821func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
822 if m == nil {
823 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
824 }
825 dir, err := ResidentHistoryDir(m.root, childID)
826 if err != nil {
827 return ResidentHistoryPage{}, err
828 }
829 return ReadResidentHistoryPage(dir, olderCursor, limit)
830}
831
832// Result returns the bounded latest-turn projection for a resident child.
833func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
834 if m == nil {
835 return ResidentResult{}, errors.New("resident manager: unavailable")
836 }
837 dir, err := ResidentHistoryDir(m.root, childID)
838 if err != nil {
839 return ResidentResult{}, err
840 }
841 return ReadResidentResult(filepath.Join(dir, residentResultName))
842}
843
844func (m *ResidentManager) Close(ctx context.Context) error {
845 if m == nil {
846 return nil
847 }
848 m.mu.Lock()
849 m.closed = true
850 children := make([]*ResidentChild, 0, len(m.children))
851 for _, child := range m.children {
852 children = append(children, child)
853 }
854 m.mu.Unlock()
855 var errs []error
856 for _, child := range children {
857 if err := child.Close(ctx); err != nil {
858 errs = append(errs, err)
859 }
860 }
861 return errors.Join(errs...)
862}