1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "os/exec"
9 "path/filepath"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/bnema/zut/packages/core"
15 "github.com/bnema/zut/packages/provider"
16)
17
18func TestResidentManagerSpawnsJournaledInProcessChild(t *testing.T) {
19 run := make(chan string, 1)
20 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
21 return func(_ context.Context, prompt string) error { run <- prompt; return nil }, nil
22 })
23 defer manager.Close(context.Background())
24
25 child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "child-manager", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "inspect this")
26 if err != nil {
27 t.Fatalf("Spawn: %v", err)
28 }
29 if child == nil {
30 t.Fatal("Spawn returned nil child")
31 }
32 select {
33 case got := <-run:
34 if got != "inspect this" {
35 t.Fatalf("runner prompt = %q", got)
36 }
37 case <-time.After(time.Second):
38 t.Fatal("resident runner did not execute accepted task")
39 }
40 deadline := time.Now().Add(time.Second)
41 for {
42 result, readErr := manager.Result("child-manager")
43 if readErr == nil {
44 if result.State != ResidentIdle || result.TurnID == "" {
45 t.Fatalf("result = %#v", result)
46 }
47 break
48 }
49 if time.Now().After(deadline) {
50 t.Fatalf("result projection was not written: %v", readErr)
51 }
52 time.Sleep(time.Millisecond)
53 }
54}
55
56func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) {
57 completed := make(chan ResidentCompletion, 1)
58 manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, journal *ResidentJournal) (ResidentTurnRunner, error) {
59 return func(context.Context, string) error {
60 return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "the actual child answer"}}}})
61 }, nil
62 })
63 manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion })
64 t.Cleanup(func() { _ = manager.Close(context.Background()) })
65 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "summary-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil {
66 t.Fatal(err)
67 }
68 select {
69 case completion := <-completed:
70 if completion.Summary != "the actual child answer" {
71 t.Fatalf("completion summary = %q", completion.Summary)
72 }
73 case <-time.After(time.Second):
74 t.Fatal("resident child did not complete")
75 }
76}
77
78func TestResidentManagerLiveReturnsChildSnapshot(t *testing.T) {
79 started := make(chan struct{})
80 release := make(chan struct{})
81 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
82 return func(context.Context, string) error {
83 close(started)
84 <-release
85 return nil
86 }, nil
87 })
88 defer manager.Close(context.Background())
89 defer close(release)
90 _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "live-manager", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task")
91 if err != nil {
92 t.Fatal(err)
93 }
94 select {
95 case <-started:
96 case <-time.After(time.Second):
97 t.Fatal("resident child did not start")
98 }
99 live, ok := manager.Live("live-manager")
100 if !ok || live.TurnID == "" || live.State != ResidentRunning {
101 t.Fatalf("Live = (%#v, %t)", live, ok)
102 }
103 if _, ok := manager.Live("missing"); ok {
104 t.Fatal("Live returned missing child")
105 }
106}
107
108func TestResidentManagerHistoryObserverSkipsTransientStreamEvents(t *testing.T) {
109 started := make(chan struct{})
110 publishedDelta := make(chan struct{})
111 persist := make(chan struct{})
112 release := make(chan struct{})
113 historyUpdates := make(chan string, 1)
114 manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, journal *ResidentJournal) (ResidentTurnRunner, error) {
115 return func(context.Context, string) error {
116 close(started)
117 if err := journal.RecordAgentEvent(core.EvTextDelta{Delta: "streaming"}); err != nil {
118 return err
119 }
120 close(publishedDelta)
121 <-persist
122 if err := journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "finalized"}}}}); err != nil {
123 return err
124 }
125 <-release
126 return nil
127 }, nil
128 })
129 manager.SetHistoryUpdateObserver(func(childID string) { historyUpdates <- childID })
130 t.Cleanup(func() {
131 close(release)
132 if err := manager.Close(context.Background()); err != nil {
133 t.Error(err)
134 }
135 })
136 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "history-observer", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil {
137 t.Fatal(err)
138 }
139 <-started
140 <-publishedDelta
141 select {
142 case got := <-historyUpdates:
143 t.Fatalf("transient stream event requested history reload for %q", got)
144 default:
145 }
146 close(persist)
147 select {
148 case got := <-historyUpdates:
149 if got != "history-observer" {
150 t.Fatalf("history update child = %q", got)
151 }
152 case <-time.After(time.Second):
153 t.Fatal("finalized event did not request history reload")
154 }
155}
156
157func TestResidentManagerActivityObserverTracksRunningChildrenWithoutPolling(t *testing.T) {
158 started := make(chan struct{})
159 release := make(chan struct{})
160 activity := make(chan bool, 2)
161 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
162 return func(context.Context, string) error {
163 close(started)
164 <-release
165 return nil
166 }, nil
167 })
168 manager.SetActivityObserver(func(active bool) { activity <- active })
169 t.Cleanup(func() {
170 if err := manager.Close(context.Background()); err != nil {
171 t.Error(err)
172 }
173 })
174 select {
175 case got := <-activity:
176 if got {
177 t.Fatal("new manager reported active resident")
178 }
179 default:
180 t.Fatal("activity observer did not receive its initial state")
181 }
182 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "activity-observer", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil {
183 t.Fatal(err)
184 }
185 select {
186 case got := <-activity:
187 if !got {
188 t.Fatal("spawn marked resident activity inactive")
189 }
190 case <-time.After(time.Second):
191 t.Fatal("spawn did not report resident activity")
192 }
193 <-started
194 close(release)
195 select {
196 case got := <-activity:
197 if got {
198 t.Fatal("completed resident remained active")
199 }
200 case <-time.After(time.Second):
201 t.Fatal("completion did not clear resident activity")
202 }
203}
204
205func TestResidentManagerRecordsFactoryFailureAfterAcceptance(t *testing.T) {
206 root := t.TempDir()
207 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
208 return nil, errors.New("synthetic factory failure")
209 })
210 spec := ResidentChildSpec{ID: "factory-failure", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
211 if _, err := manager.Spawn(context.Background(), spec, "task"); err == nil {
212 t.Fatal("Spawn succeeded")
213 }
214 metadata, err := ReadResidentMetadata(filepath.Join(root, spec.ID, residentMetadataName))
215 if err != nil {
216 t.Fatal(err)
217 }
218 if metadata.State != ResidentFailed {
219 t.Fatalf("state = %q, want failed", metadata.State)
220 }
221}
222
223func TestResidentManagerReconcileIgnoresLegacyAgentsContainer(t *testing.T) {
224 root := t.TempDir()
225 legacyChild := filepath.Join(root, "agents", "old-child")
226 if err := os.MkdirAll(legacyChild, 0o700); err != nil {
227 t.Fatal(err)
228 }
229 if err := os.WriteFile(filepath.Join(legacyChild, "events.jsonl"), []byte("legacy\n"), 0o600); err != nil {
230 t.Fatal(err)
231 }
232 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
233 return nil, nil
234 })
235 if errs := manager.Reconcile(); len(errs) != 0 {
236 t.Fatalf("Reconcile errors = %v", errs)
237 }
238 if snapshots := manager.Snapshot(); len(snapshots) != 0 {
239 t.Fatalf("Reconcile recovered legacy child: %#v", snapshots)
240 }
241}
242
243func TestResidentManagerResumeRebuildsStoppedChild(t *testing.T) {
244 runs := make(chan string, 2)
245 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
246 return func(ctx context.Context, prompt string) error {
247 runs <- prompt
248 if prompt == "initial" {
249 <-ctx.Done()
250 return ctx.Err()
251 }
252 return nil
253 }, nil
254 })
255 defer manager.Close(context.Background())
256 spec := ResidentChildSpec{ID: "stopped-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5", Required: true}
257 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
258 t.Fatal(err)
259 }
260 if got := <-runs; got != "initial" {
261 t.Fatalf("initial prompt = %q", got)
262 }
263 if err := manager.Stop(context.Background(), spec.ID); err != nil {
264 t.Fatal(err)
265 }
266 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
267 t.Fatalf("Resume stopped child: %v", err)
268 }
269 if got := <-runs; got != "follow up" {
270 t.Fatalf("follow-up prompt = %q", got)
271 }
272 deadline := time.Now().Add(time.Second)
273 for len(manager.UnmetRequired()) != 0 {
274 if time.Now().After(deadline) {
275 t.Fatalf("required child remained unmet: %#v", manager.UnmetRequired())
276 }
277 time.Sleep(time.Millisecond)
278 }
279}
280
281func TestResidentManagerPreparesWorkspaceBeforeAcceptance(t *testing.T) {
282 root := t.TempDir()
283 prepared := false
284 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
285 prepared = true
286 if req.AgentID != "workspace-child" || req.StateDir != filepath.Join(root, "workspace-child") {
287 t.Fatalf("workspace request = %#v", req)
288 }
289 return SharedWorkspace{Root: req.RepositoryRoot}.Prepare(context.Background(), req)
290 }, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
291 if !prepared || spec.Workspace != "/repo" {
292 t.Fatalf("factory spec = %#v, prepared=%t", spec, prepared)
293 }
294 return func(context.Context, string) error { return nil }, nil
295 })
296 defer manager.Close(context.Background())
297 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "workspace-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo"}, "task"); err != nil {
298 t.Fatal(err)
299 }
300}
301
302func TestResidentManagerRejectsWorkspaceOutsideAllowedRootsBeforeAcceptance(t *testing.T) {
303 prepared := false
304 manager := newResidentManager(t.TempDir(), SubagentPolicy{AllowedRoots: []string{t.TempDir()}}, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) {
305 prepared = true
306 return nil, nil
307 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
308 return func(context.Context, string) error { return nil }, nil
309 })
310 outside := t.TempDir()
311 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "outside-root", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: outside}, "task"); err == nil {
312 t.Fatal("Spawn succeeded outside allowed roots")
313 }
314 if prepared {
315 t.Fatal("workspace preparation ran outside allowed roots")
316 }
317}
318
319func TestResidentManagerCapturesSuccessfulWorktreeBeforeCleanup(t *testing.T) {
320 root := t.TempDir()
321 workspace := &testResidentWorkspace{dir: "/isolated", mode: WorkspaceWorktree, capture: WorkspaceCapture{Patch: []byte("patch"), ChangedFiles: []string{"changed.go"}}}
322 manager := NewResidentManagerWithWorkspace(root, 0, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) {
323 return workspace, nil
324 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
325 return func(context.Context, string) error { return nil }, nil
326 })
327 defer manager.Close(context.Background())
328 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "capture-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", WorkspaceMode: WorkspaceWorktree}, "task"); err != nil {
329 t.Fatal(err)
330 }
331 deadline := time.Now().Add(time.Second)
332 for manager.Snapshot()[0].State != ResidentIdle {
333 if time.Now().After(deadline) {
334 t.Fatal("turn did not finish")
335 }
336 time.Sleep(time.Millisecond)
337 }
338 if workspace.cleaned {
339 t.Fatal("idle child worktree was removed before explicit close")
340 }
341 result, err := manager.Result("capture-child")
342 if err != nil {
343 t.Fatal(err)
344 }
345 if result.PatchRef != PatchRef("capture-child") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "changed.go" {
346 t.Fatalf("result = %#v", result)
347 }
348 patch, err := os.ReadFile(filepath.Join(root, "capture-child", residentPatchName))
349 if err != nil || string(patch) != "patch" {
350 t.Fatalf("patch = %q, err=%v", patch, err)
351 }
352 if err := manager.Stop(context.Background(), "capture-child"); err != nil {
353 t.Fatal(err)
354 }
355 if !workspace.cleaned {
356 t.Fatal("successful worktree was not cleaned on stop")
357 }
358}
359
360func TestResidentManagerRetainsFailedWorktree(t *testing.T) {
361 workspace := &testResidentWorkspace{dir: "/isolated", mode: WorkspaceWorktree}
362 manager := NewResidentManagerWithWorkspace(t.TempDir(), 0, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) { return workspace, nil }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
363 return func(context.Context, string) error { return errors.New("turn failed") }, nil
364 })
365 defer manager.Close(context.Background())
366 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "failed-worktree", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", WorkspaceMode: WorkspaceWorktree}, "task"); err != nil {
367 t.Fatal(err)
368 }
369 deadline := time.Now().Add(time.Second)
370 for manager.Snapshot()[0].State == ResidentRunning || manager.Snapshot()[0].State == ResidentQueued {
371 if time.Now().After(deadline) {
372 t.Fatal("turn did not finish")
373 }
374 time.Sleep(time.Millisecond)
375 }
376 if workspace.captured || workspace.cleaned {
377 t.Fatalf("failed worktree captured=%t cleaned=%t", workspace.captured, workspace.cleaned)
378 }
379}
380
381func TestResidentManagerCapturesRealWorktreePatch(t *testing.T) {
382 if _, err := exec.LookPath("git"); err != nil {
383 t.Skip("git is required")
384 }
385 root, repo := t.TempDir(), initTestRepo(t)
386 manager := NewResidentManager(root, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
387 return func(context.Context, string) error {
388 return os.WriteFile(filepath.Join(spec.Workspace, "README.md"), []byte("resident change\n"), 0o600)
389 }, nil
390 })
391 defer manager.Close(context.Background())
392 child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "git-worktree", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: repo, WorkspaceMode: WorkspaceWorktree, WorkspaceBase: "HEAD", WorkspaceCapture: CapturePatch}, "task")
393 if err != nil {
394 t.Fatal(err)
395 }
396 deadline := time.Now().Add(time.Second)
397 for child.State() != ResidentIdle {
398 if time.Now().After(deadline) {
399 t.Fatal("worktree turn did not finish")
400 }
401 time.Sleep(time.Millisecond)
402 }
403 result, err := manager.Result("git-worktree")
404 if err != nil {
405 t.Fatal(err)
406 }
407 if result.PatchRef != PatchRef("git-worktree") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "README.md" {
408 t.Fatalf("result = %#v", result)
409 }
410 patch, err := os.ReadFile(filepath.Join(root, "git-worktree", residentPatchName))
411 if err != nil || !strings.Contains(string(patch), "resident change") {
412 t.Fatalf("patch = %q, err=%v", patch, err)
413 }
414 if err := manager.Stop(context.Background(), "git-worktree"); err != nil {
415 t.Fatal(err)
416 }
417 if _, err := os.Stat(child.workspace.Dir()); !os.IsNotExist(err) {
418 t.Fatalf("worktree remains after stop: %v", err)
419 }
420}
421
422type testResidentWorkspace struct {
423 dir string
424 mode WorkspaceMode
425 capture WorkspaceCapture
426 captured, cleaned bool
427}
428
429func (w *testResidentWorkspace) Dir() string { return w.dir }
430func (w *testResidentWorkspace) RepositoryRoot() string { return "/repo" }
431func (w *testResidentWorkspace) Mode() WorkspaceMode { return w.mode }
432func (w *testResidentWorkspace) Capture(context.Context) (WorkspaceCapture, error) {
433 w.captured = true
434 return w.capture, nil
435}
436func (w *testResidentWorkspace) Cleanup(context.Context) error { w.cleaned = true; return nil }
437
438func TestResidentManagerRejectsDuplicateBeforeSecondJournalAcceptance(t *testing.T) {
439 root := t.TempDir()
440 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
441 return func(context.Context, string) error { return nil }, nil
442 })
443 defer manager.Close(context.Background())
444 spec := ResidentChildSpec{ID: "duplicate", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
445 if _, err := manager.Spawn(context.Background(), spec, "first"); err != nil {
446 t.Fatal(err)
447 }
448 if _, err := manager.Spawn(context.Background(), spec, "second"); err == nil {
449 t.Fatal("duplicate Spawn succeeded")
450 }
451 records, err := ReadResidentJournal(filepath.Join(root, spec.ID, residentTranscriptName))
452 if err != nil {
453 t.Fatal(err)
454 }
455 accepted := 0
456 for _, record := range records {
457 if record.Type == residentRecordAccepted {
458 accepted++
459 if record.Prompt != "first" {
460 t.Fatalf("accepted prompt = %q, want first", record.Prompt)
461 }
462 }
463 }
464 if accepted != 1 {
465 t.Fatalf("records = %#v, want one accepted spawn", records)
466 }
467}
468
469func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutReplay(t *testing.T) {
470 root := t.TempDir()
471 journal, err := OpenResidentJournal(root, "recovered")
472 if err != nil {
473 t.Fatal(err)
474 }
475 spec := ResidentChildSpec{ID: "recovered", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5", Required: true}
476 if err := journal.Accept(spec, "do not replay"); err != nil {
477 t.Fatal(err)
478 }
479 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
480 t.Fatal(err)
481 }
482 if err := journal.Close(); err != nil {
483 t.Fatal(err)
484 }
485 runs := 0
486 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
487 runs++
488 return func(context.Context, string) error { return nil }, nil
489 })
490 if errs := manager.Reconcile(); len(errs) != 0 {
491 t.Fatalf("Reconcile errors = %v", errs)
492 }
493 if runs != 0 {
494 t.Fatalf("factory ran %d times during reconciliation", runs)
495 }
496 snapshots := manager.Snapshot()
497 if len(snapshots) != 1 || snapshots[0].State != ResidentInterrupted || !snapshots[0].Required {
498 t.Fatalf("snapshots = %#v", snapshots)
499 }
500 if _, err := manager.Spawn(context.Background(), spec, "overwrite"); err == nil {
501 t.Fatal("Spawn overwrote recovered journal")
502 }
503}
504
505func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
506 root := t.TempDir()
507 journal, err := OpenResidentJournal(root, "resume-recovered")
508 if err != nil {
509 t.Fatal(err)
510 }
511 spec := ResidentChildSpec{ID: "resume-recovered", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5"}
512 if err := journal.Accept(spec, "old task must not replay"); err != nil {
513 t.Fatal(err)
514 }
515 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
516 t.Fatal(err)
517 }
518 if err := journal.Close(); err != nil {
519 t.Fatal(err)
520 }
521 runs := make(chan string, 1)
522 manager := NewResidentManager(root, func(got ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
523 if got.SessionID != spec.SessionID {
524 t.Fatalf("session = %q, want %q", got.SessionID, spec.SessionID)
525 }
526 return func(_ context.Context, prompt string) error { runs <- prompt; return nil }, nil
527 })
528 defer manager.Close(context.Background())
529 if errs := manager.Reconcile(); len(errs) != 0 {
530 t.Fatalf("Reconcile = %v", errs)
531 }
532 if err := manager.Resume(context.Background(), spec.ID, "new explicit prompt"); err != nil {
533 t.Fatal(err)
534 }
535 select {
536 case got := <-runs:
537 if got != "new explicit prompt" {
538 t.Fatalf("runner prompt = %q", got)
539 }
540 case <-time.After(time.Second):
541 t.Fatal("explicit resume did not run")
542 }
543}
544
545func TestResidentManagerExplicitResumeReusesInterruptedWorktree(t *testing.T) {
546 root := t.TempDir()
547 journal, err := OpenResidentJournal(root, "resume-worktree")
548 if err != nil {
549 t.Fatal(err)
550 }
551 spec := ResidentChildSpec{ID: "resume-worktree", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", Workspace: "/prior-worktree", WorkspaceMode: WorkspaceWorktree}
552 if err := journal.Accept(spec, "old task"); err != nil {
553 t.Fatal(err)
554 }
555 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
556 t.Fatal(err)
557 }
558 if err := journal.Close(); err != nil {
559 t.Fatal(err)
560 }
561 workspace := &testResidentWorkspace{dir: "/prior-worktree", mode: WorkspaceWorktree}
562 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
563 if req.ExistingPath != "/prior-worktree" {
564 t.Fatalf("resume workspace request = %#v", req)
565 }
566 return workspace, nil
567 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
568 return func(context.Context, string) error { return nil }, nil
569 })
570 defer manager.Close(context.Background())
571 if errs := manager.Reconcile(); len(errs) != 0 {
572 t.Fatal(errs)
573 }
574 if err := manager.Resume(context.Background(), spec.ID, "resume"); err != nil {
575 t.Fatal(err)
576 }
577}
578
579func TestResidentManagerAppliesPositiveQueueTimeoutDurably(t *testing.T) {
580 started := make(chan string, 2)
581 release := make(chan struct{})
582 manager := NewResidentManagerWithPolicy(t.TempDir(), SubagentPolicy{MaxConcurrent: 1, QueueTimeout: 20 * time.Millisecond}, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
583 return func(_ context.Context, _ string) error {
584 started <- spec.ID
585 if spec.ID == "first" {
586 <-release
587 }
588 return nil
589 }, nil
590 })
591 defer manager.Close(context.Background())
592 for _, id := range []string{"first", "second"} {
593 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: id, SessionID: id, Provider: "openai", Model: "gpt-5"}, id); err != nil {
594 t.Fatalf("Spawn(%s): %v", id, err)
595 }
596 }
597 select {
598 case got := <-started:
599 if got != "first" {
600 t.Fatalf("first runner = %q, want first", got)
601 }
602 case <-time.After(time.Second):
603 t.Fatal("first runner did not start")
604 }
605 deadline := time.Now().Add(time.Second)
606 for {
607 result, err := manager.Result("second")
608 if err == nil && result.State == ResidentFailed {
609 break
610 }
611 if time.Now().After(deadline) {
612 t.Fatalf("queued child did not record timeout: result=%#v err=%v", result, err)
613 }
614 time.Sleep(time.Millisecond)
615 }
616 select {
617 case got := <-started:
618 t.Fatalf("timed out child ran as %q", got)
619 default:
620 }
621 close(release)
622}
623
624func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
625 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
626 return func(context.Context, string) error { return nil }, nil
627 })
628 defer manager.Close(context.Background())
629 accepted := make(chan string, 2)
630 manager.SetAcceptedObserver(func(_ ResidentChildSpec, prompt string) { accepted <- prompt })
631 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
632 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
633 t.Fatal(err)
634 }
635 if got := <-accepted; got != "initial" {
636 t.Fatalf("initial observer prompt = %q", got)
637 }
638 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
639 t.Fatal(err)
640 }
641 if got := <-accepted; got != "follow up" {
642 t.Fatalf("follow-up observer prompt = %q", got)
643 }
644}
645
646func TestResidentManagerReconcileReplacesStaleRecoveredSnapshots(t *testing.T) {
647 root := t.TempDir()
648 journal, err := OpenResidentJournal(root, "stale")
649 if err != nil {
650 t.Fatal(err)
651 }
652 spec := ResidentChildSpec{ID: "stale", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
653 if err := journal.Accept(spec, "task"); err != nil {
654 t.Fatal(err)
655 }
656 if err := journal.Close(); err != nil {
657 t.Fatal(err)
658 }
659 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) { return nil, nil })
660 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 1 {
661 t.Fatalf("first Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
662 }
663 if err := os.RemoveAll(filepath.Join(root, "stale")); err != nil {
664 t.Fatal(err)
665 }
666 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 0 {
667 t.Fatalf("second Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
668 }
669}
670
671func TestResidentManagerLimitsConcurrentTurnsGlobally(t *testing.T) {
672 started := make(chan string, DefaultResidentConcurrency+1)
673 release := make(chan struct{})
674 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
675 return func(ctx context.Context, prompt string) error {
676 started <- prompt
677 select {
678 case <-release:
679 return nil
680 case <-ctx.Done():
681 return ctx.Err()
682 }
683 }, nil
684 })
685 defer manager.Close(context.Background())
686
687 for i := 0; i < DefaultResidentConcurrency+1; i++ {
688 spec := ResidentChildSpec{ID: fmt.Sprintf("child-%d", i), SessionID: fmt.Sprintf("session-%d", i), Provider: "openai", Model: "gpt-5"}
689 if _, err := manager.Spawn(context.Background(), spec, spec.ID); err != nil {
690 t.Fatalf("Spawn(%d): %v", i, err)
691 }
692 }
693 for i := 0; i < DefaultResidentConcurrency; i++ {
694 select {
695 case <-started:
696 case <-time.After(time.Second):
697 t.Fatal("did not fill the default six slots")
698 }
699 }
700 select {
701 case extra := <-started:
702 t.Fatalf("seventh turn started before a slot released: %q", extra)
703 default:
704 }
705 close(release)
706 select {
707 case <-started:
708 case <-time.After(time.Second):
709 t.Fatal("queued seventh turn did not start after a slot released")
710 }
711}