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
20// runtime; it never starts another zut binary.
21type ResidentFactory func(ResidentChildSpec, *ResidentJournal) (ResidentRuntime, 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 PendingFollowUps int
220 OwnedElsewhere bool
221 Usage provider.Usage
222 ContextUsed int
223 ContextMax int
224 Subscription bool
225}
226
227func residentSnapshot(child *ResidentChild) ResidentSnapshot {
228 if child == nil {
229 return ResidentSnapshot{}
230 }
231 live := child.Live()
232 return ResidentSnapshot{ID: child.spec.ID, State: child.State(), Profile: child.spec.Profile,
233 Provider: child.spec.Provider, Model: child.spec.Model,
234 WorkspaceMode: child.spec.WorkspaceMode, Required: child.spec.Required,
235 UpdatedAt: child.StateUpdatedAt(), TurnStartedAt: child.TurnStartedAt(),
236 ActivityUpdatedAt: child.ActivityUpdatedAt(), WaitingForModel: live.WaitingForModel, PendingFollowUps: child.PendingFollowUps(),
237 Usage: live.Usage, ContextUsed: live.ContextUsed, ContextMax: live.ContextMax, Subscription: live.Subscription}
238}
239
240func NewResidentManager(root string, factory ResidentFactory) *ResidentManager {
241 return NewResidentManagerWithLimit(root, 0, factory)
242}
243
244func NewResidentManagerWithLimit(root string, limit int, factory ResidentFactory) *ResidentManager {
245 return NewResidentManagerWithWorkspace(root, limit, PrepareWorkspace, factory)
246}
247
248// NewResidentManagerWithPolicy applies the resident-only host policy. A zero
249// queue timeout leaves accepted work queued until a scheduler slot is free.
250func NewResidentManagerWithPolicy(root string, policy SubagentPolicy, factory ResidentFactory) *ResidentManager {
251 return newResidentManager(root, policy, PrepareWorkspace, factory)
252}
253
254// NewResidentManagerWithWorkspace exposes the narrow allocation seam needed
255// to make workspace preparation part of spawn's durable commit boundary.
256func NewResidentManagerWithWorkspace(root string, limit int, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
257 return newResidentManager(root, SubagentPolicy{MaxConcurrent: limit}, prepare, factory)
258}
259
260func newResidentManager(root string, policy SubagentPolicy, prepare func(context.Context, WorkspaceRequest) (WorkspaceHandle, error), factory ResidentFactory) *ResidentManager {
261 if prepare == nil {
262 prepare = PrepareWorkspace
263 }
264 return &ResidentManager{
265 root: root,
266 factory: factory,
267 prepareWorkspace: prepare,
268 children: make(map[string]*ResidentChild),
269 recovered: make(map[string]ResidentSnapshot),
270 recoveredSpec: make(map[string]ResidentChildSpec),
271 pending: make(map[string]struct{}),
272 activeChildren: make(map[string]struct{}),
273 completionWaiters: make(map[string][]chan ResidentCompletion),
274 scheduler: NewResidentScheduler(policy.MaxConcurrent),
275 queueTimeout: policy.QueueTimeout,
276 allowedRoots: append([]string(nil), policy.AllowedRoots...),
277 }
278}
279
280// Spawn commits acceptance before publishing a child or starting its runner.
281func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, task string) (*ResidentChild, error) {
282 if m == nil || m.factory == nil {
283 return nil, errors.New("resident manager: no factory")
284 }
285 m.lifecycleMu.Lock()
286 defer m.lifecycleMu.Unlock()
287 m.mu.Lock()
288 if m.closed {
289 m.mu.Unlock()
290 return nil, errors.New("resident manager: closed")
291 }
292 if _, exists := m.children[spec.ID]; exists {
293 m.mu.Unlock()
294 return nil, errors.New("resident manager: duplicate child ID")
295 }
296 if _, exists := m.recovered[spec.ID]; exists {
297 m.mu.Unlock()
298 return nil, errors.New("resident manager: child journal already exists")
299 }
300 if _, exists := m.pending[spec.ID]; exists {
301 m.mu.Unlock()
302 return nil, errors.New("resident manager: child acceptance pending")
303 }
304 m.pending[spec.ID] = struct{}{}
305 m.mu.Unlock()
306 defer func() { m.mu.Lock(); delete(m.pending, spec.ID); m.mu.Unlock() }()
307 if spec.InitialTurnID == "" {
308 spec.InitialTurnID = uuid.NewString()
309 }
310 journal, err := OpenResidentJournal(m.root, spec.ID)
311 if err != nil {
312 return nil, err
313 }
314 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
315 if workspaceRoot == "" {
316 workspaceRoot = strings.TrimSpace(spec.Workspace)
317 }
318 if !m.workspaceAllowed(workspaceRoot) {
319 _ = journal.Close()
320 return nil, errors.New("resident manager: workspace is outside allowed roots")
321 }
322 var workspace WorkspaceHandle
323 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
324 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
325 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, spec.ID),
326 AgentID: spec.ID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots,
327 })
328 if err != nil {
329 _ = journal.Close()
330 return nil, fmt.Errorf("resident manager prepare workspace: %w", err)
331 }
332 spec.Workspace = workspace.Dir()
333 spec.WorkspaceMode = workspace.Mode()
334 }
335 if err := journal.Accept(spec, task); err != nil {
336 if workspace != nil {
337 _ = workspace.Cleanup(context.Background())
338 }
339 _ = journal.Close()
340 return nil, err
341 }
342 m.mu.Lock()
343 accepted := m.onAccepted
344 m.mu.Unlock()
345 if accepted != nil {
346 accepted(spec, spec.InitialTurnID, task)
347 }
348 runner, err := m.factory(spec, journal)
349 if err != nil {
350 _ = journal.RecordFailure(spec)
351 _ = journal.Close()
352 m.reportCompletion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
353 return nil, err
354 }
355 child := newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
356 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
357 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
358 m.mu.Lock()
359 if m.closed {
360 m.mu.Unlock()
361 _ = child.Close(context.Background())
362 return nil, errors.New("resident manager: closed")
363 }
364 m.children[spec.ID] = child
365 m.mu.Unlock()
366 if childAccepted, err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
367 m.mu.Lock()
368 if m.children[spec.ID] == child {
369 delete(m.children, spec.ID)
370 }
371 m.mu.Unlock()
372 _ = child.Close(context.Background())
373 if !childAccepted {
374 _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
375 m.reportCompletion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
376 }
377 return nil, err
378 }
379 return child, nil
380}
381
382// scheduledRuntime gates each turn on a global scheduler slot. Steering needs
383// no slot: it only reaches a turn that already holds one.
384type scheduledRuntime struct {
385 ResidentRuntime
386 run ResidentTurnRunner
387}
388
389func (r scheduledRuntime) Run(ctx context.Context, prompt string) error { return r.run(ctx, prompt) }
390
391func (m *ResidentManager) scheduledRunner(childID string, runtime ResidentRuntime) ResidentRuntime {
392 return scheduledRuntime{ResidentRuntime: runtime, run: m.scheduledRun(childID, runtime)}
393}
394
395func (m *ResidentManager) scheduledRun(childID string, runtime ResidentRuntime) ResidentTurnRunner {
396 return func(ctx context.Context, prompt string) error {
397 if ctx == nil {
398 ctx = context.Background()
399 }
400 ticket := m.scheduler.Enqueue(childID, prompt)
401 m.dispatch()
402 var timeout <-chan time.Time
403 var timer *time.Timer
404 if m.queueTimeout > 0 {
405 timer = time.NewTimer(m.queueTimeout)
406 timeout = timer.C
407 defer timer.Stop()
408 }
409 select {
410 case <-ctx.Done():
411 if m.scheduler.Cancel(ticket.Sequence) {
412 m.dispatch()
413 return ctx.Err()
414 }
415 // Admission won the race with cancellation. Consume the handoff
416 // and release its reservation without running provider work.
417 <-ticket.ready
418 _ = m.scheduler.Release(childID)
419 m.dispatch()
420 return ctx.Err()
421 case <-ticket.ready:
422 case <-timeout:
423 if m.scheduler.Cancel(ticket.Sequence) {
424 m.dispatch()
425 return fmt.Errorf("resident manager: queue timeout after %s", m.queueTimeout)
426 }
427 // Admission won the race with the timeout. The reservation is now
428 // owned by this runner and must be released below.
429 <-ticket.ready
430 }
431 defer func() {
432 _ = m.scheduler.Release(childID)
433 m.dispatch()
434 }()
435 return runtime.Run(ctx, prompt)
436 }
437}
438
439func (m *ResidentManager) dispatch() {
440 if m == nil || m.scheduler == nil {
441 return
442 }
443 m.dispatchMu.Lock()
444 defer m.dispatchMu.Unlock()
445 for {
446 if _, ok := m.scheduler.Admit(); !ok {
447 return
448 }
449 }
450}
451
452// Resume synchronizes a follow-up acceptance before making it visible to the
453// resident child. A disk-only child is rebuilt only for this explicit new
454// prompt; reconciliation itself never constructs or replays one.
455func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) error {
456 _, err := m.ResumeFollowUp(ctx, childID, prompt, ResumeSteer, uuid.NewString(), nil)
457 return err
458}
459
460// ResumeMode selects how a follow-up reaches a child that is running a turn.
461type ResumeMode string
462
463const (
464 // ResumeSteer injects the follow-up into the running turn at its next
465 // tool or model boundary. An idle child starts a new turn instead.
466 ResumeSteer ResumeMode = "steer"
467 // ResumeQueue runs the follow-up as a separate turn after the current one.
468 ResumeQueue ResumeMode = "queue"
469)
470
471// ResumeOutcome identifies the turn that will answer a follow-up.
472type ResumeOutcome struct {
473 // TurnID is the new turn for a queued follow-up, or the running turn
474 // that received a steered one.
475 TurnID string
476 Steered bool
477}
478
479// ResumeFollowUp delivers a follow-up by mode. A steered follow-up joins the
480// running turn, so it creates no new turn and no new completion. onSteered, when
481// set, runs before that turn can complete so a caller can watch its completion
482// without racing it. When steering is impossible, the follow-up is queued as a
483// new turn named turnID.
484func (m *ResidentManager) ResumeFollowUp(ctx context.Context, childID, prompt string, mode ResumeMode, turnID string, onSteered func(turnID string)) (ResumeOutcome, error) {
485 if mode == "" {
486 mode = ResumeSteer
487 }
488 if mode != ResumeSteer && mode != ResumeQueue {
489 return ResumeOutcome{}, fmt.Errorf("resident manager: unknown resume mode %q", mode)
490 }
491 if mode == ResumeSteer && m != nil && strings.TrimSpace(prompt) != "" {
492 m.lifecycleMu.Lock()
493 child := m.Get(strings.TrimSpace(childID))
494 if child != nil {
495 steered, activeTurn, err := child.steer(ctx, prompt, onSteered)
496 if err == nil && steered {
497 m.lifecycleMu.Unlock()
498 m.notifyUpdate(child.spec.ID, false)
499 return ResumeOutcome{TurnID: activeTurn, Steered: true}, nil
500 }
501 }
502 m.lifecycleMu.Unlock()
503 }
504 if strings.TrimSpace(turnID) == "" {
505 turnID = uuid.NewString()
506 }
507 if err := m.resumeWithTurn(ctx, childID, prompt, turnID); err != nil {
508 return ResumeOutcome{}, err
509 }
510 return ResumeOutcome{TurnID: turnID}, nil
511}
512
513// resumeWithTurn queues a follow-up turn with a caller-owned ID. The ID names the
514// accepted follow-up turn in its completion, so a caller that must observe the
515// outcome can subscribe with WatchCompletion before the turn can finish. The
516// ID must be unique per child: replay rejects a duplicate turn.accepted record,
517// so reusing one makes the journal unreplayable.
518func (m *ResidentManager) resumeWithTurn(ctx context.Context, childID, prompt, turnID string) error {
519 if strings.TrimSpace(turnID) == "" {
520 turnID = uuid.NewString()
521 }
522 if m == nil {
523 return errors.New("resident manager: unavailable")
524 }
525 m.lifecycleMu.Lock()
526 defer m.lifecycleMu.Unlock()
527 childID = strings.TrimSpace(childID)
528 prompt = strings.TrimSpace(prompt)
529 if childID == "" || prompt == "" {
530 return errors.New("resident manager: child ID and prompt are required")
531 }
532 m.mu.Lock()
533 if m.closed {
534 m.mu.Unlock()
535 return errors.New("resident manager: closed")
536 }
537 child := m.children[childID]
538 spec, recovered := m.recoveredSpec[childID]
539 snapshot := m.recovered[childID]
540 // A recovered foreign snapshot is advisory. Its owner may have exited since
541 // Reconcile; OpenResidentJournal below makes the authoritative fresh check.
542 if child == nil && snapshot.OwnedElsewhere {
543 recovered = true
544 }
545 if child == nil && recovered {
546 if _, pending := m.pending[childID]; pending {
547 m.mu.Unlock()
548 return errors.New("resident manager: child resume pending")
549 }
550 m.pending[childID] = struct{}{}
551 }
552 m.mu.Unlock()
553 if child == nil && recovered {
554 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
555 journal, err := OpenResidentJournal(m.root, childID)
556 if err != nil {
557 if errors.Is(err, ErrResidentLeaseBusy) {
558 return errors.New("resident manager: child is owned by another zut process")
559 }
560 return err
561 }
562 metadata, err := reconcileOwnedResidentJournal(journal)
563 if err != nil {
564 _ = journal.Close()
565 return err
566 }
567 journal.ConfigureUsage(metadata.ContextMax, metadata.Subscription)
568 records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
569 if err != nil || len(records) == 0 || records[0].Spec == nil {
570 _ = journal.Close()
571 return errors.New("resident manager: invalid accepted specification")
572 }
573 spec = *records[0].Spec
574 var workspace WorkspaceHandle
575 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
576 if workspaceRoot == "" {
577 workspaceRoot = strings.TrimSpace(spec.Workspace)
578 }
579 if !m.workspaceAllowed(workspaceRoot) {
580 _ = journal.Close()
581 return errors.New("resident manager: workspace is outside allowed roots")
582 }
583 if workspaceRoot != "" || spec.WorkspaceMode == WorkspaceWorktree {
584 workspace, err = m.prepareWorkspace(ctx, WorkspaceRequest{
585 Mode: spec.WorkspaceMode, RepositoryRoot: workspaceRoot, StateDir: filepath.Join(m.root, childID),
586 AgentID: childID, Base: spec.WorkspaceBase, Capture: spec.WorkspaceCapture, AllowedRoots: m.allowedRoots, ExistingPath: spec.Workspace,
587 })
588 if err != nil {
589 _ = journal.Close()
590 return fmt.Errorf("resident manager rebuild workspace: %w", err)
591 }
592 spec.Workspace = workspace.Dir()
593 spec.WorkspaceMode = workspace.Mode()
594 }
595 runner, err := m.factory(spec, journal)
596 if err != nil {
597 if workspace != nil && workspace.Mode() == WorkspaceShared {
598 _ = workspace.Cleanup(context.Background())
599 }
600 _ = journal.Close()
601 return fmt.Errorf("resident manager rebuild child: %w", err)
602 }
603 child = newJournaledResidentChildWithWorkspace(spec, journal, workspace, m.scheduledRunner(spec.ID, runner), m.reportCompletion)
604 child.setUpdateObserver(func(historyChanged bool) { m.notifyUpdate(spec.ID, historyChanged) })
605 child.setStateObserver(func(state ResidentState) { m.notifyActivity(spec.ID, state) })
606 m.mu.Lock()
607 if m.closed {
608 m.mu.Unlock()
609 _ = child.Close(context.Background())
610 return errors.New("resident manager: closed")
611 }
612 m.children[childID] = child
613 delete(m.recovered, childID)
614 delete(m.recoveredSpec, childID)
615 m.mu.Unlock()
616 }
617 if child == nil {
618 return errors.New("resident manager: child is not live")
619 }
620 if child.journal == nil {
621 _, err := child.resumeAccepted(ctx, turnID, prompt)
622 return err
623 }
624 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {
625 return err
626 }
627 m.mu.Lock()
628 accepted := m.onAccepted
629 m.mu.Unlock()
630 if accepted != nil {
631 accepted(child.spec, turnID, prompt)
632 }
633 if childAccepted, err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
634 if !childAccepted {
635 _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
636 m.reportCompletion(ResidentCompletion{ChildID: child.spec.ID, TurnID: turnID, Task: prompt, Err: err})
637 }
638 return err
639 }
640 return nil
641}
642
643func (m *ResidentManager) workspaceAllowed(root string) bool {
644 if len(m.allowedRoots) == 0 || strings.TrimSpace(root) == "" {
645 return true
646 }
647 for _, allowed := range m.allowedRoots {
648 if pathWithin(root, allowed) {
649 return true
650 }
651 }
652 return false
653}
654
655// Get returns the live child only. Durable children recovered from a previous
656// host are intentionally not auto-resumed.
657func (m *ResidentManager) Get(childID string) *ResidentChild {
658 if m == nil {
659 return nil
660 }
661 m.mu.Lock()
662 defer m.mu.Unlock()
663 return m.children[childID]
664}
665
666// State reports a live child's current execution state. A child with an
667// accepted turn awaiting global scheduler admission is queued, even though its
668// control loop has started preparing that turn.
669func (m *ResidentManager) State(childID string) (ResidentState, bool) {
670 child := m.Get(childID)
671 if child == nil {
672 return "", false
673 }
674 state := child.State()
675 if state == ResidentRunning && m.scheduler.Pending(childID) {
676 return ResidentQueued, true
677 }
678 return state, true
679}
680
681// Live returns an immutable unfinished-turn snapshot for one live child. It
682// never holds the manager lock while copying the child projection.
683func (m *ResidentManager) Live(childID string) (ResidentLiveSnapshot, bool) {
684 if m == nil {
685 return ResidentLiveSnapshot{}, false
686 }
687 m.mu.Lock()
688 child := m.children[childID]
689 m.mu.Unlock()
690 if child == nil {
691 return ResidentLiveSnapshot{}, false
692 }
693 return child.Live(), true
694}
695
696// Snapshot returns copies of the live children without holding the manager
697// lock while callers render or perform I/O.
698func (m *ResidentManager) Snapshot() []ResidentSnapshot {
699 children, _ := m.SnapshotPage(0, -1)
700 return children
701}
702
703// SnapshotPage returns a bounded, ID-sorted state projection. It avoids
704// materializing live snapshots that a renderer cannot display and reports the
705// total so callers can page without relying on a stale full list.
706func (m *ResidentManager) SnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
707 if m == nil {
708 return nil, 0
709 }
710 if offset < 0 {
711 offset = 0
712 }
713 m.mu.Lock()
714 live := make(map[string]*ResidentChild, len(m.children))
715 for id, child := range m.children {
716 live[id] = child
717 }
718 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
719 for id, snapshot := range m.recovered {
720 recovered[id] = snapshot
721 }
722 m.mu.Unlock()
723 ids := make([]string, 0, len(live)+len(recovered))
724 for id := range recovered {
725 ids = append(ids, id)
726 }
727 for id := range live {
728 if _, alreadyRecovered := recovered[id]; !alreadyRecovered {
729 ids = append(ids, id)
730 }
731 }
732 sort.Strings(ids)
733 total := len(ids)
734 if offset >= total || limit == 0 {
735 return nil, total
736 }
737 end := total
738 if limit > 0 && offset+limit < end {
739 end = offset + limit
740 }
741 result := make([]ResidentSnapshot, 0, end-offset)
742 for _, id := range ids[offset:end] {
743 if child := live[id]; child != nil {
744 result = append(result, residentSnapshot(child))
745 continue
746 }
747 if snapshot, ok := recovered[id]; ok {
748 result = append(result, snapshot)
749 }
750 }
751 return result, total
752}
753
754// RecentSnapshotPage returns a bounded state projection ordered by the latest
755// visible state transition, newest first. It is intended for the interactive
756// dashboard where operators need to identify the child that just changed.
757func (m *ResidentManager) RecentSnapshotPage(offset, limit int) ([]ResidentSnapshot, int) {
758 if m == nil {
759 return nil, 0
760 }
761 if offset < 0 {
762 offset = 0
763 }
764 m.mu.Lock()
765 live := make(map[string]*ResidentChild, len(m.children))
766 for id, child := range m.children {
767 live[id] = child
768 }
769 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
770 for id, snapshot := range m.recovered {
771 recovered[id] = snapshot
772 }
773 m.mu.Unlock()
774 ids := make([]string, 0, len(live)+len(recovered))
775 for id := range recovered {
776 ids = append(ids, id)
777 }
778 for id := range live {
779 if _, recoveredAlready := recovered[id]; !recoveredAlready {
780 ids = append(ids, id)
781 }
782 }
783 snapshots := make(map[string]ResidentSnapshot, len(ids))
784 for _, id := range ids {
785 if child := live[id]; child != nil {
786 snapshots[id] = residentSnapshot(child)
787 } else {
788 snapshots[id] = recovered[id]
789 }
790 }
791 sort.Slice(ids, func(i, j int) bool {
792 left, right := snapshots[ids[i]], snapshots[ids[j]]
793 if !left.UpdatedAt.Equal(right.UpdatedAt) {
794 return left.UpdatedAt.After(right.UpdatedAt)
795 }
796 return ids[i] < ids[j]
797 })
798 total := len(ids)
799 if offset >= total || limit == 0 {
800 return nil, total
801 }
802 end := total
803 if limit > 0 && offset+limit < end {
804 end = offset + limit
805 }
806 result := make([]ResidentSnapshot, 0, end-offset)
807 for _, id := range ids[offset:end] {
808 result = append(result, snapshots[id])
809 }
810 return result, total
811}
812
813// ActiveSnapshotPage returns only same-host queued and running children for
814// compact activity surfaces. It deliberately exposes no prompt, transcript, or path.
815func (m *ResidentManager) ActiveSnapshotPage(limit int) ([]ResidentSnapshot, int) {
816 if m == nil {
817 return nil, 0
818 }
819 m.mu.Lock()
820 live := make(map[string]*ResidentChild, len(m.children))
821 for id, child := range m.children {
822 live[id] = child
823 }
824 recovered := make(map[string]ResidentSnapshot, len(m.recovered))
825 for id, snapshot := range m.recovered {
826 recovered[id] = snapshot
827 }
828 m.mu.Unlock()
829
830 result := make([]ResidentSnapshot, 0, len(live)+len(recovered))
831 for _, child := range live {
832 snapshot := residentSnapshot(child)
833 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
834 result = append(result, snapshot)
835 }
836 }
837 for id, snapshot := range recovered {
838 if _, exists := live[id]; exists {
839 continue
840 }
841 if snapshot.OwnedElsewhere {
842 continue
843 }
844 if snapshot.State == ResidentQueued || snapshot.State == ResidentRunning {
845 result = append(result, snapshot)
846 }
847 }
848 sort.Slice(result, func(i, j int) bool {
849 if !result[i].UpdatedAt.Equal(result[j].UpdatedAt) {
850 return result[i].UpdatedAt.After(result[j].UpdatedAt)
851 }
852 return result[i].ID < result[j].ID
853 })
854 total := len(result)
855 if limit > 0 && len(result) > limit {
856 result = result[:limit]
857 }
858 return result, total
859}
860
861// SnapshotFor returns metadata for one child without building a dashboard
862// projection. It keeps child-session headers independent of the total number
863// of resident children.
864func (m *ResidentManager) SnapshotFor(childID string) (ResidentSnapshot, bool) {
865 if m == nil {
866 return ResidentSnapshot{}, false
867 }
868 m.mu.Lock()
869 child := m.children[childID]
870 recovered, ok := m.recovered[childID]
871 m.mu.Unlock()
872 if child != nil {
873 return residentSnapshot(child), true
874 }
875 return recovered, ok
876}
877
878// Reconcile discovers journals left by an earlier host, repairs their bounded
879// projections, and marks queued/running work interrupted. It never creates a
880// live child or replays a prompt. Incompatible journals are left untouched
881// and excluded from discovery. Other per-child errors are returned for callers
882// to surface without hiding valid sibling journals.
883func (m *ResidentManager) Reconcile() []error {
884 if m == nil {
885 return []error{errors.New("resident manager: unavailable")}
886 }
887 m.lifecycleMu.Lock()
888 defer m.lifecycleMu.Unlock()
889 entries, err := os.ReadDir(m.root)
890 if err != nil {
891 if errors.Is(err, os.ErrNotExist) {
892 return nil
893 }
894 return []error{fmt.Errorf("resident manager read root: %w", err)}
895 }
896 sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
897 m.mu.Lock()
898 m.recovered = make(map[string]ResidentSnapshot)
899 m.recoveredSpec = make(map[string]ResidentChildSpec)
900 m.mu.Unlock()
901 var errs []error
902 for _, entry := range entries {
903 if !entry.IsDir() {
904 continue
905 }
906 // The old subprocess runtime stored children beneath an "agents"
907 // container. Resident children are direct entries in this root. Leave
908 // that legacy state untouched: it is neither read nor migrated.
909 if entry.Name() == "agents" {
910 continue
911 }
912 childID := entry.Name()
913 m.mu.Lock()
914 _, live := m.children[childID]
915 _, pending := m.pending[childID]
916 m.mu.Unlock()
917 if live || pending {
918 continue
919 }
920 childDir := filepath.Join(m.root, childID)
921 metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
922 // Unknown journal versions cannot be resumed. Leave them on disk
923 // without reporting the same incompatibility at every host startup.
924 // Match the exact sentinel so a joined journal-close error is still
925 // reported.
926 if reconcileErr == ErrIncompatibleResidentJournal {
927 continue
928 }
929 if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
930 foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
931 if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
932 errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
933 continue
934 }
935 if readErr == nil {
936 if foreign.Version != residentJournalVersion && foreign.Version != residentJournalVersionV2 {
937 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, ErrIncompatibleResidentJournal))
938 continue
939 }
940 // A v3 projection can never carry the removed budget
941 // state: the writer only persists live states. Treat it
942 // as corruption rather than resurfacing the zombie state.
943 if foreign.Version != residentJournalVersionV2 && foreign.State == ResidentState(residentV2OutcomeBudgetExhausted) {
944 errs = append(errs, fmt.Errorf("resident child %s: invalid terminal state", childID))
945 continue
946 }
947 }
948 state, updatedAt := foreign.State, foreign.UpdatedAt
949 if readErr != nil || state == "" {
950 // Ownership begins before child.accepted is durable. Do not read
951 // that live transcript merely to improve this transient snapshot.
952 state, updatedAt = ResidentQueued, time.Now().UTC()
953 }
954 if foreign.Version == residentJournalVersionV2 && state == ResidentState(residentV2OutcomeBudgetExhausted) {
955 // Historical budget stop from an older host: display it as
956 // failed, never as a distinct terminal state.
957 state = ResidentFailed
958 }
959 m.mu.Lock()
960 m.recovered[childID] = ResidentSnapshot{
961 ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true,
962 Usage: foreign.Usage, ContextUsed: foreign.ContextUsed, ContextMax: foreign.ContextMax, Subscription: foreign.Subscription,
963 }
964 m.mu.Unlock()
965 continue
966 }
967 if reconcileErr != nil {
968 errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
969 continue
970 }
971 m.mu.Lock()
972 if _, live := m.children[childID]; !live {
973 m.recovered[childID] = ResidentSnapshot{
974 ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model,
975 WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt,
976 Usage: metadata.Usage, ContextUsed: metadata.ContextUsed, ContextMax: metadata.ContextMax, Subscription: metadata.Subscription,
977 }
978 m.recoveredSpec[childID] = spec
979 }
980 m.mu.Unlock()
981 }
982 return errs
983}
984
985// UnmetRequired returns required work that has not reached the successful idle
986// boundary. Failed and interrupted work intentionally remain unmet until an
987// explicit successful follow-up is accepted and completed.
988func (m *ResidentManager) UnmetRequired() []ResidentSnapshot {
989 all := m.Snapshot()
990 result := all[:0]
991 for _, snapshot := range all {
992 if snapshot.Required && snapshot.State != ResidentIdle && snapshot.State != ResidentCompleted {
993 result = append(result, snapshot)
994 }
995 }
996 return result
997}
998
999// Interrupt cancels only a live child's running turn and drops its queued
1000// follow-ups. The child stays live with its transcript, so a later resume
1001// continues with full context. It reports false when no turn is running.
1002func (m *ResidentManager) Interrupt(ctx context.Context, childID string) (bool, error) {
1003 if m == nil {
1004 return false, errors.New("resident manager: unavailable")
1005 }
1006 m.lifecycleMu.Lock()
1007 defer m.lifecycleMu.Unlock()
1008 child := m.Get(strings.TrimSpace(childID))
1009 if child == nil {
1010 return false, errors.New("resident manager: child is not live")
1011 }
1012 interrupted, err := child.interruptTurn(ctx)
1013 if err != nil {
1014 return false, err
1015 }
1016 m.notifyUpdate(child.spec.ID, false)
1017 return interrupted, nil
1018}
1019
1020// Stop cancels one live child and waits for its control loop. A disk-only
1021// interrupted child must be explicitly resumed by host construction instead.
1022func (m *ResidentManager) Stop(ctx context.Context, childID string) error {
1023 if m == nil {
1024 return errors.New("resident manager: unavailable")
1025 }
1026 m.lifecycleMu.Lock()
1027 defer m.lifecycleMu.Unlock()
1028 childID = strings.TrimSpace(childID)
1029 m.mu.Lock()
1030 child := m.children[childID]
1031 m.mu.Unlock()
1032 if child == nil {
1033 return errors.New("resident manager: child is not live")
1034 }
1035 if err := child.Close(ctx); err != nil {
1036 return err
1037 }
1038 // A stopped child has released its resident agent and journal. Retain only
1039 // its durable spec so an explicit follow-up can rebuild a fresh resident
1040 // child; keeping the closed child live would make Resume append to a closed
1041 // journal.
1042 m.mu.Lock()
1043 if m.children[childID] == child {
1044 delete(m.children, childID)
1045 m.recovered[childID] = residentSnapshot(child)
1046 m.recoveredSpec[childID] = child.spec
1047 }
1048 m.mu.Unlock()
1049 m.notifyUpdate(childID, false)
1050 return nil
1051}
1052
1053// History reads bounded finalized resident history through the manager-owned
1054// state root. It accepts only a child ID, never a filesystem path.
1055func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryItem, error) {
1056 if m == nil {
1057 return nil, errors.New("resident manager: unavailable")
1058 }
1059 dir, release, err := m.readableResidentDir(childID)
1060 if err != nil {
1061 return nil, err
1062 }
1063 defer release()
1064 return ReadResidentHistory(dir, limit)
1065}
1066
1067// HistoryPage reads one recent-first, bounded page through the manager-owned
1068// state root. Public consumers identify only a child and receive an opaque
1069// cursor for older content.
1070func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (ResidentHistoryPage, error) {
1071 if m == nil {
1072 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
1073 }
1074 dir, release, err := m.readableResidentDir(childID)
1075 if err != nil {
1076 return ResidentHistoryPage{}, err
1077 }
1078 defer release()
1079 return ReadResidentHistoryPage(dir, olderCursor, limit)
1080}
1081
1082// Result returns the bounded latest-turn projection for a resident child.
1083func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
1084 if m == nil {
1085 return ResidentResult{}, errors.New("resident manager: unavailable")
1086 }
1087 dir, release, err := m.readableResidentDir(childID)
1088 if err != nil {
1089 return ResidentResult{}, err
1090 }
1091 defer release()
1092 return ReadResidentResult(filepath.Join(dir, residentResultName))
1093}
1094
1095func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
1096 dir, err := ResidentHistoryDir(m.root, childID)
1097 if err != nil {
1098 return "", nil, err
1099 }
1100 childID = strings.TrimSpace(childID)
1101 m.mu.Lock()
1102 live := m.children[childID] != nil
1103 m.mu.Unlock()
1104 if live {
1105 return dir, func() {}, nil
1106 }
1107 lease, err := acquireResidentLease(dir)
1108 if err != nil {
1109 if errors.Is(err, ErrResidentLeaseBusy) {
1110 return "", nil, errors.New("resident manager: child is owned by another zut process")
1111 }
1112 return "", nil, err
1113 }
1114 return dir, func() { _ = lease.Close() }, nil
1115}
1116
1117func (m *ResidentManager) Close(ctx context.Context) error {
1118 if m == nil {
1119 return nil
1120 }
1121 m.mu.Lock()
1122 m.closed = true
1123 children := make([]*ResidentChild, 0, len(m.children))
1124 for _, child := range m.children {
1125 children = append(children, child)
1126 }
1127 m.mu.Unlock()
1128 var errs []error
1129 for _, child := range children {
1130 if err := child.Close(ctx); err != nil {
1131 errs = append(errs, err)
1132 }
1133 }
1134 return errors.Join(errs...)
1135}