1package subagents
2
3import (
4 "bufio"
5 "context"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/bnema/zut/packages/core"
17 "github.com/bnema/zut/packages/provider"
18)
19
20func TestResidentManagerSpawnsJournaledInProcessChild(t *testing.T) {
21 run := make(chan string, 1)
22 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
23 return func(_ context.Context, prompt string) error { run <- prompt; return nil }, nil
24 })
25 defer manager.Close(context.Background())
26
27 child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "child-manager", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "inspect this")
28 if err != nil {
29 t.Fatalf("Spawn: %v", err)
30 }
31 if child == nil {
32 t.Fatal("Spawn returned nil child")
33 }
34 select {
35 case got := <-run:
36 if got != "inspect this" {
37 t.Fatalf("runner prompt = %q", got)
38 }
39 case <-time.After(time.Second):
40 t.Fatal("resident runner did not execute accepted task")
41 }
42 deadline := time.Now().Add(time.Second)
43 for {
44 result, readErr := manager.Result("child-manager")
45 if readErr == nil {
46 if result.State != ResidentIdle || result.TurnID == "" {
47 t.Fatalf("result = %#v", result)
48 }
49 break
50 }
51 if time.Now().After(deadline) {
52 t.Fatalf("result projection was not written: %v", readErr)
53 }
54 time.Sleep(time.Millisecond)
55 }
56}
57
58func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) {
59 completed := make(chan ResidentCompletion, 1)
60 manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, journal *ResidentJournal) (ResidentTurnRunner, error) {
61 return func(context.Context, string) error {
62 return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "the actual child answer"}}}})
63 }, nil
64 })
65 manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion })
66 t.Cleanup(func() { _ = manager.Close(context.Background()) })
67 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "summary-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil {
68 t.Fatal(err)
69 }
70 select {
71 case completion := <-completed:
72 if completion.TurnID == "" {
73 t.Fatal("completion turn ID is empty")
74 }
75 if completion.Summary != "the actual child answer" {
76 t.Fatalf("completion summary = %q", completion.Summary)
77 }
78 case <-time.After(time.Second):
79 t.Fatal("resident child did not complete")
80 }
81}
82
83func TestResidentManagerReportsQueuedTurnsWhenFinalResultCannotPersist(t *testing.T) {
84 started := make(chan struct{})
85 release := make(chan struct{})
86 completed := make(chan ResidentCompletion, 2)
87 var journal *ResidentJournal
88 manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, childJournal *ResidentJournal) (ResidentTurnRunner, error) {
89 journal = childJournal
90 return func(context.Context, string) error {
91 close(started)
92 <-release
93 return nil
94 }, nil
95 })
96 manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion })
97 t.Cleanup(func() { _ = manager.Close(context.Background()) })
98
99 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "persistence-failure-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "initial task"); err != nil {
100 t.Fatal(err)
101 }
102 select {
103 case <-started:
104 case <-time.After(time.Second):
105 t.Fatal("resident child did not start")
106 }
107 if err := manager.Resume(context.Background(), "persistence-failure-child", "queued follow-up"); err != nil {
108 t.Fatalf("Resume: %v", err)
109 }
110 if err := journal.Close(); err != nil {
111 t.Fatalf("Close journal: %v", err)
112 }
113 close(release)
114
115 seen := make(map[string]ResidentCompletion, 2)
116 for range 2 {
117 select {
118 case completion := <-completed:
119 seen[completion.Task] = completion
120 case <-time.After(time.Second):
121 t.Fatalf("terminal completions = %#v, want initial and queued turns", seen)
122 }
123 }
124 for _, task := range []string{"initial task", "queued follow-up"} {
125 completion, ok := seen[task]
126 if !ok || completion.TurnID == "" || completion.Err == nil || !strings.Contains(completion.Err.Error(), "persist resident child terminal state") {
127 t.Fatalf("completion for %q = %#v", task, completion)
128 }
129 }
130}
131
132func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) {
133 started := make(chan struct{})
134 completed := make(chan ResidentCompletion, 1)
135 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
136 return func(ctx context.Context, _ string) error {
137 close(started)
138 <-ctx.Done()
139 return ctx.Err()
140 }, nil
141 })
142 manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion })
143 t.Cleanup(func() { _ = manager.Close(context.Background()) })
144
145 child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "interrupted-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task")
146 if err != nil {
147 t.Fatal(err)
148 }
149 select {
150 case <-started:
151 case <-time.After(time.Second):
152 t.Fatal("resident child did not start")
153 }
154 if err := child.Close(context.Background()); err != nil {
155 t.Fatalf("Close: %v", err)
156 }
157 select {
158 case completion := <-completed:
159 if completion.ChildID != "interrupted-child" || completion.TurnID == "" || !errors.Is(completion.Err, context.Canceled) {
160 t.Fatalf("completion = %#v", completion)
161 }
162 case <-time.After(time.Second):
163 t.Fatal("interrupted turn did not report completion")
164 }
165}
166
167func TestResidentManagerLiveReturnsChildSnapshot(t *testing.T) {
168 started := make(chan struct{})
169 release := make(chan struct{})
170 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
171 return func(context.Context, string) error {
172 close(started)
173 <-release
174 return nil
175 }, nil
176 })
177 defer manager.Close(context.Background())
178 defer close(release)
179 _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "live-manager", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task")
180 if err != nil {
181 t.Fatal(err)
182 }
183 select {
184 case <-started:
185 case <-time.After(time.Second):
186 t.Fatal("resident child did not start")
187 }
188 live, ok := manager.Live("live-manager")
189 if !ok || live.TurnID == "" || live.State != ResidentRunning {
190 t.Fatalf("Live = (%#v, %t)", live, ok)
191 }
192 if _, ok := manager.Live("missing"); ok {
193 t.Fatal("Live returned missing child")
194 }
195}
196
197func TestResidentManagerHistoryObserverSkipsTransientStreamEvents(t *testing.T) {
198 started := make(chan struct{})
199 publishedDelta := make(chan struct{})
200 persist := make(chan struct{})
201 release := make(chan struct{})
202 historyUpdates := make(chan string, 1)
203 manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, journal *ResidentJournal) (ResidentTurnRunner, error) {
204 return func(context.Context, string) error {
205 close(started)
206 if err := journal.RecordAgentEvent(core.EvTextDelta{Delta: "streaming"}); err != nil {
207 return err
208 }
209 close(publishedDelta)
210 <-persist
211 if err := journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "finalized"}}}}); err != nil {
212 return err
213 }
214 <-release
215 return nil
216 }, nil
217 })
218 manager.SetHistoryUpdateObserver(func(childID string) { historyUpdates <- childID })
219 t.Cleanup(func() {
220 close(release)
221 if err := manager.Close(context.Background()); err != nil {
222 t.Error(err)
223 }
224 })
225 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "history-observer", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil {
226 t.Fatal(err)
227 }
228 <-started
229 <-publishedDelta
230 select {
231 case got := <-historyUpdates:
232 t.Fatalf("transient stream event requested history reload for %q", got)
233 default:
234 }
235 close(persist)
236 select {
237 case got := <-historyUpdates:
238 if got != "history-observer" {
239 t.Fatalf("history update child = %q", got)
240 }
241 case <-time.After(time.Second):
242 t.Fatal("finalized event did not request history reload")
243 }
244}
245
246func TestResidentManagerActivityObserverTracksRunningChildrenWithoutPolling(t *testing.T) {
247 started := make(chan struct{})
248 release := make(chan struct{})
249 activity := make(chan bool, 2)
250 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
251 return func(context.Context, string) error {
252 close(started)
253 <-release
254 return nil
255 }, nil
256 })
257 manager.SetActivityObserver(func(active bool) { activity <- active })
258 t.Cleanup(func() {
259 if err := manager.Close(context.Background()); err != nil {
260 t.Error(err)
261 }
262 })
263 select {
264 case got := <-activity:
265 if got {
266 t.Fatal("new manager reported active resident")
267 }
268 default:
269 t.Fatal("activity observer did not receive its initial state")
270 }
271 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "activity-observer", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil {
272 t.Fatal(err)
273 }
274 select {
275 case got := <-activity:
276 if !got {
277 t.Fatal("spawn marked resident activity inactive")
278 }
279 case <-time.After(time.Second):
280 t.Fatal("spawn did not report resident activity")
281 }
282 <-started
283 close(release)
284 select {
285 case got := <-activity:
286 if got {
287 t.Fatal("completed resident remained active")
288 }
289 case <-time.After(time.Second):
290 t.Fatal("completion did not clear resident activity")
291 }
292}
293
294func TestResidentManagerRecordsFactoryFailureAfterAcceptance(t *testing.T) {
295 root := t.TempDir()
296 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
297 return nil, errors.New("synthetic factory failure")
298 })
299 spec := ResidentChildSpec{ID: "factory-failure", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
300 if _, err := manager.Spawn(context.Background(), spec, "task"); err == nil {
301 t.Fatal("Spawn succeeded")
302 }
303 metadata, err := ReadResidentMetadata(filepath.Join(root, spec.ID, residentMetadataName))
304 if err != nil {
305 t.Fatal(err)
306 }
307 if metadata.State != ResidentFailed {
308 t.Fatalf("state = %q, want failed", metadata.State)
309 }
310}
311
312func TestResidentManagerReconcileIgnoresLegacyAgentsContainer(t *testing.T) {
313 root := t.TempDir()
314 legacyChild := filepath.Join(root, "agents", "old-child")
315 if err := os.MkdirAll(legacyChild, 0o700); err != nil {
316 t.Fatal(err)
317 }
318 if err := os.WriteFile(filepath.Join(legacyChild, "events.jsonl"), []byte("legacy\n"), 0o600); err != nil {
319 t.Fatal(err)
320 }
321 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
322 return nil, nil
323 })
324 if errs := manager.Reconcile(); len(errs) != 0 {
325 t.Fatalf("Reconcile errors = %v", errs)
326 }
327 if snapshots := manager.Snapshot(); len(snapshots) != 0 {
328 t.Fatalf("Reconcile recovered legacy child: %#v", snapshots)
329 }
330}
331
332func TestResidentManagerResumeRebuildsStoppedChild(t *testing.T) {
333 runs := make(chan string, 2)
334 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
335 return func(ctx context.Context, prompt string) error {
336 runs <- prompt
337 if prompt == "initial" {
338 <-ctx.Done()
339 return ctx.Err()
340 }
341 return nil
342 }, nil
343 })
344 defer manager.Close(context.Background())
345 spec := ResidentChildSpec{ID: "stopped-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5", Required: true}
346 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
347 t.Fatal(err)
348 }
349 if got := <-runs; got != "initial" {
350 t.Fatalf("initial prompt = %q", got)
351 }
352 if err := manager.Stop(context.Background(), spec.ID); err != nil {
353 t.Fatal(err)
354 }
355 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
356 t.Fatalf("Resume stopped child: %v", err)
357 }
358 if got := <-runs; got != "follow up" {
359 t.Fatalf("follow-up prompt = %q", got)
360 }
361 deadline := time.Now().Add(time.Second)
362 for len(manager.UnmetRequired()) != 0 {
363 if time.Now().After(deadline) {
364 t.Fatalf("required child remained unmet: %#v", manager.UnmetRequired())
365 }
366 time.Sleep(time.Millisecond)
367 }
368}
369
370func TestResidentManagerPreparesWorkspaceBeforeAcceptance(t *testing.T) {
371 root := t.TempDir()
372 prepared := false
373 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
374 prepared = true
375 if req.AgentID != "workspace-child" || req.StateDir != filepath.Join(root, "workspace-child") {
376 t.Fatalf("workspace request = %#v", req)
377 }
378 return SharedWorkspace{Root: req.RepositoryRoot}.Prepare(context.Background(), req)
379 }, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
380 if !prepared || spec.Workspace != "/repo" {
381 t.Fatalf("factory spec = %#v, prepared=%t", spec, prepared)
382 }
383 return func(context.Context, string) error { return nil }, nil
384 })
385 defer manager.Close(context.Background())
386 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "workspace-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo"}, "task"); err != nil {
387 t.Fatal(err)
388 }
389}
390
391func TestResidentManagerRejectsWorkspaceOutsideAllowedRootsBeforeAcceptance(t *testing.T) {
392 prepared := false
393 manager := newResidentManager(t.TempDir(), SubagentPolicy{AllowedRoots: []string{t.TempDir()}}, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) {
394 prepared = true
395 return nil, nil
396 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
397 return func(context.Context, string) error { return nil }, nil
398 })
399 outside := t.TempDir()
400 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "outside-root", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: outside}, "task"); err == nil {
401 t.Fatal("Spawn succeeded outside allowed roots")
402 }
403 if prepared {
404 t.Fatal("workspace preparation ran outside allowed roots")
405 }
406}
407
408func TestResidentManagerCapturesSuccessfulWorktreeBeforeCleanup(t *testing.T) {
409 root := t.TempDir()
410 workspace := &testResidentWorkspace{dir: "/isolated", mode: WorkspaceWorktree, capture: WorkspaceCapture{Patch: []byte("patch"), ChangedFiles: []string{"changed.go"}}}
411 manager := NewResidentManagerWithWorkspace(root, 0, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) {
412 return workspace, nil
413 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
414 return func(context.Context, string) error { return nil }, nil
415 })
416 defer manager.Close(context.Background())
417 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 {
418 t.Fatal(err)
419 }
420 deadline := time.Now().Add(time.Second)
421 for manager.Snapshot()[0].State != ResidentIdle {
422 if time.Now().After(deadline) {
423 t.Fatal("turn did not finish")
424 }
425 time.Sleep(time.Millisecond)
426 }
427 if workspace.cleaned {
428 t.Fatal("idle child worktree was removed before explicit close")
429 }
430 result, err := manager.Result("capture-child")
431 if err != nil {
432 t.Fatal(err)
433 }
434 if result.PatchRef != PatchRef("capture-child") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "changed.go" {
435 t.Fatalf("result = %#v", result)
436 }
437 patch, err := os.ReadFile(filepath.Join(root, "capture-child", residentPatchName))
438 if err != nil || string(patch) != "patch" {
439 t.Fatalf("patch = %q, err=%v", patch, err)
440 }
441 if err := manager.Stop(context.Background(), "capture-child"); err != nil {
442 t.Fatal(err)
443 }
444 if !workspace.cleaned {
445 t.Fatal("successful worktree was not cleaned on stop")
446 }
447}
448
449func TestResidentManagerRetainsFailedWorktree(t *testing.T) {
450 workspace := &testResidentWorkspace{dir: "/isolated", mode: WorkspaceWorktree}
451 manager := NewResidentManagerWithWorkspace(t.TempDir(), 0, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) { return workspace, nil }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
452 return func(context.Context, string) error { return errors.New("turn failed") }, nil
453 })
454 defer manager.Close(context.Background())
455 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 {
456 t.Fatal(err)
457 }
458 deadline := time.Now().Add(time.Second)
459 for manager.Snapshot()[0].State == ResidentRunning || manager.Snapshot()[0].State == ResidentQueued {
460 if time.Now().After(deadline) {
461 t.Fatal("turn did not finish")
462 }
463 time.Sleep(time.Millisecond)
464 }
465 if workspace.captured || workspace.cleaned {
466 t.Fatalf("failed worktree captured=%t cleaned=%t", workspace.captured, workspace.cleaned)
467 }
468}
469
470func TestResidentManagerCapturesRealWorktreePatch(t *testing.T) {
471 if _, err := exec.LookPath("git"); err != nil {
472 t.Skip("git is required")
473 }
474 root, repo := t.TempDir(), initTestRepo(t)
475 manager := NewResidentManager(root, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
476 return func(context.Context, string) error {
477 return os.WriteFile(filepath.Join(spec.Workspace, "README.md"), []byte("resident change\n"), 0o600)
478 }, nil
479 })
480 defer manager.Close(context.Background())
481 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")
482 if err != nil {
483 t.Fatal(err)
484 }
485 deadline := time.Now().Add(time.Second)
486 for child.State() != ResidentIdle {
487 if time.Now().After(deadline) {
488 t.Fatal("worktree turn did not finish")
489 }
490 time.Sleep(time.Millisecond)
491 }
492 result, err := manager.Result("git-worktree")
493 if err != nil {
494 t.Fatal(err)
495 }
496 if result.PatchRef != PatchRef("git-worktree") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "README.md" {
497 t.Fatalf("result = %#v", result)
498 }
499 patch, err := os.ReadFile(filepath.Join(root, "git-worktree", residentPatchName))
500 if err != nil || !strings.Contains(string(patch), "resident change") {
501 t.Fatalf("patch = %q, err=%v", patch, err)
502 }
503 if err := manager.Stop(context.Background(), "git-worktree"); err != nil {
504 t.Fatal(err)
505 }
506 if _, err := os.Stat(child.workspace.Dir()); !os.IsNotExist(err) {
507 t.Fatalf("worktree remains after stop: %v", err)
508 }
509}
510
511type testResidentWorkspace struct {
512 dir string
513 mode WorkspaceMode
514 capture WorkspaceCapture
515 captured, cleaned bool
516}
517
518func (w *testResidentWorkspace) Dir() string { return w.dir }
519func (w *testResidentWorkspace) RepositoryRoot() string { return "/repo" }
520func (w *testResidentWorkspace) Mode() WorkspaceMode { return w.mode }
521func (w *testResidentWorkspace) Capture(context.Context) (WorkspaceCapture, error) {
522 w.captured = true
523 return w.capture, nil
524}
525func (w *testResidentWorkspace) Cleanup(context.Context) error { w.cleaned = true; return nil }
526
527func TestResidentManagerRejectsDuplicateBeforeSecondJournalAcceptance(t *testing.T) {
528 root := t.TempDir()
529 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
530 return func(context.Context, string) error { return nil }, nil
531 })
532 defer manager.Close(context.Background())
533 spec := ResidentChildSpec{ID: "duplicate", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
534 if _, err := manager.Spawn(context.Background(), spec, "first"); err != nil {
535 t.Fatal(err)
536 }
537 if _, err := manager.Spawn(context.Background(), spec, "second"); err == nil {
538 t.Fatal("duplicate Spawn succeeded")
539 }
540 records, err := ReadResidentJournal(filepath.Join(root, spec.ID, residentTranscriptName))
541 if err != nil {
542 t.Fatal(err)
543 }
544 accepted := 0
545 for _, record := range records {
546 if record.Type == residentRecordAccepted {
547 accepted++
548 if record.Prompt != "first" {
549 t.Fatalf("accepted prompt = %q, want first", record.Prompt)
550 }
551 }
552 }
553 if accepted != 1 {
554 t.Fatalf("records = %#v, want one accepted spawn", records)
555 }
556}
557
558func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutReplay(t *testing.T) {
559 root := t.TempDir()
560 journal, err := OpenResidentJournal(root, "recovered")
561 if err != nil {
562 t.Fatal(err)
563 }
564 spec := ResidentChildSpec{ID: "recovered", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5", Required: true}
565 if err := journal.Accept(spec, "do not replay"); err != nil {
566 t.Fatal(err)
567 }
568 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
569 t.Fatal(err)
570 }
571 if err := journal.Close(); err != nil {
572 t.Fatal(err)
573 }
574 runs := 0
575 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
576 runs++
577 return func(context.Context, string) error { return nil }, nil
578 })
579 if errs := manager.Reconcile(); len(errs) != 0 {
580 t.Fatalf("Reconcile errors = %v", errs)
581 }
582 if runs != 0 {
583 t.Fatalf("factory ran %d times during reconciliation", runs)
584 }
585 snapshots := manager.Snapshot()
586 if len(snapshots) != 1 || snapshots[0].State != ResidentInterrupted || !snapshots[0].Required {
587 t.Fatalf("snapshots = %#v", snapshots)
588 }
589 if _, err := manager.Spawn(context.Background(), spec, "overwrite"); err == nil {
590 t.Fatal("Spawn overwrote recovered journal")
591 }
592}
593
594func TestResidentManagerReconcileDoesNotMutateForeignProcessJournal(t *testing.T) {
595 if os.Getenv("ZUT_RESIDENT_MANAGER_HELPER") == "1" {
596 root := os.Getenv("ZUT_RESIDENT_MANAGER_ROOT")
597 journal, err := OpenResidentJournal(root, "foreign-process")
598 if err != nil {
599 t.Fatal(err)
600 }
601 spec := ResidentChildSpec{ID: "foreign-process", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
602 if err := journal.Accept(spec, "active task"); err != nil {
603 t.Fatal(err)
604 }
605 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
606 t.Fatal(err)
607 }
608 if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolCall, Time: time.Now().UTC(), ToolID: "call-1", ToolName: "bash", ToolArgs: json.RawMessage(`{"command":"pwd"}`)}); err != nil {
609 t.Fatal(err)
610 }
611 fmt.Fprintln(os.Stdout, "ready")
612 if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
613 t.Fatal(err)
614 }
615 if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)}); err != nil {
616 t.Fatal(err)
617 }
618 if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
619 t.Fatal(err)
620 }
621 if err := journal.Close(); err != nil {
622 t.Fatal(err)
623 }
624 return
625 }
626
627 root := t.TempDir()
628 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
629 defer cancel()
630 cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentManagerReconcileDoesNotMutateForeignProcessJournal$")
631 defer func() {
632 if cmd.Process != nil {
633 _ = cmd.Process.Kill()
634 }
635 _ = cmd.Wait()
636 }()
637 cmd.Env = append(os.Environ(), "ZUT_RESIDENT_MANAGER_HELPER=1", "ZUT_RESIDENT_MANAGER_ROOT="+root)
638 stdout, err := cmd.StdoutPipe()
639 if err != nil {
640 t.Fatal(err)
641 }
642 stdin, err := cmd.StdinPipe()
643 if err != nil {
644 t.Fatal(err)
645 }
646 if err := cmd.Start(); err != nil {
647 t.Fatal(err)
648 }
649 ready, err := bufio.NewReader(stdout).ReadString('\n')
650 if err != nil {
651 t.Fatal(err)
652 }
653 if ready != "ready\n" {
654 t.Fatalf("helper readiness = %q", ready)
655 }
656 transcript := filepath.Join(root, "foreign-process", residentTranscriptName)
657 before, err := os.ReadFile(transcript)
658 if err != nil {
659 t.Fatal(err)
660 }
661
662 observer := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
663 return func(context.Context, string) error { return nil }, nil
664 })
665 if errs := observer.Reconcile(); len(errs) != 0 {
666 t.Fatalf("Reconcile errors = %v", errs)
667 }
668 after, err := os.ReadFile(transcript)
669 if err != nil {
670 t.Fatal(err)
671 }
672 if string(after) != string(before) {
673 t.Fatal("foreign reconciliation modified the transcript")
674 }
675 snapshot, ok := observer.SnapshotFor("foreign-process")
676 if !ok || !snapshot.OwnedElsewhere || snapshot.State != ResidentRunning {
677 t.Fatalf("foreign snapshot = %#v, found=%v", snapshot, ok)
678 }
679 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err == nil || !strings.Contains(err.Error(), "owned by another") {
680 t.Fatalf("foreign Resume error = %v", err)
681 }
682 if _, err := observer.History("foreign-process", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
683 t.Fatalf("foreign History error = %v", err)
684 }
685 if _, err := observer.HistoryPage("foreign-process", "", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
686 t.Fatalf("foreign HistoryPage error = %v", err)
687 }
688 if _, err := observer.Result("foreign-process"); err == nil || !strings.Contains(err.Error(), "owned by another") {
689 t.Fatalf("foreign Result error = %v", err)
690 }
691 if _, err := fmt.Fprintln(stdin); err != nil {
692 t.Fatal(err)
693 }
694 if err := stdin.Close(); err != nil {
695 t.Fatal(err)
696 }
697 if err := cmd.Wait(); err != nil {
698 t.Fatal(err)
699 }
700 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err != nil {
701 t.Fatalf("Resume after owner exit: %v", err)
702 }
703 if err := observer.Stop(context.Background(), "foreign-process"); err != nil {
704 t.Fatalf("Stop rebuilt child: %v", err)
705 }
706}
707
708func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
709 root := t.TempDir()
710 journal, err := OpenResidentJournal(root, "resume-recovered")
711 if err != nil {
712 t.Fatal(err)
713 }
714 spec := ResidentChildSpec{ID: "resume-recovered", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5"}
715 if err := journal.Accept(spec, "old task must not replay"); err != nil {
716 t.Fatal(err)
717 }
718 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
719 t.Fatal(err)
720 }
721 if err := journal.Close(); err != nil {
722 t.Fatal(err)
723 }
724 runs := make(chan string, 1)
725 manager := NewResidentManager(root, func(got ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
726 if got.SessionID != spec.SessionID {
727 t.Fatalf("session = %q, want %q", got.SessionID, spec.SessionID)
728 }
729 return func(_ context.Context, prompt string) error { runs <- prompt; return nil }, nil
730 })
731 defer manager.Close(context.Background())
732 if errs := manager.Reconcile(); len(errs) != 0 {
733 t.Fatalf("Reconcile = %v", errs)
734 }
735 if err := manager.Resume(context.Background(), spec.ID, "new explicit prompt"); err != nil {
736 t.Fatal(err)
737 }
738 select {
739 case got := <-runs:
740 if got != "new explicit prompt" {
741 t.Fatalf("runner prompt = %q", got)
742 }
743 case <-time.After(time.Second):
744 t.Fatal("explicit resume did not run")
745 }
746}
747
748func TestResidentManagerExplicitResumeReusesInterruptedWorktree(t *testing.T) {
749 root := t.TempDir()
750 journal, err := OpenResidentJournal(root, "resume-worktree")
751 if err != nil {
752 t.Fatal(err)
753 }
754 spec := ResidentChildSpec{ID: "resume-worktree", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", Workspace: "/prior-worktree", WorkspaceMode: WorkspaceWorktree}
755 if err := journal.Accept(spec, "old task"); err != nil {
756 t.Fatal(err)
757 }
758 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
759 t.Fatal(err)
760 }
761 if err := journal.Close(); err != nil {
762 t.Fatal(err)
763 }
764 workspace := &testResidentWorkspace{dir: "/prior-worktree", mode: WorkspaceWorktree}
765 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
766 if req.ExistingPath != "/prior-worktree" {
767 t.Fatalf("resume workspace request = %#v", req)
768 }
769 return workspace, nil
770 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
771 return func(context.Context, string) error { return nil }, nil
772 })
773 defer manager.Close(context.Background())
774 if errs := manager.Reconcile(); len(errs) != 0 {
775 t.Fatal(errs)
776 }
777 if err := manager.Resume(context.Background(), spec.ID, "resume"); err != nil {
778 t.Fatal(err)
779 }
780}
781
782func TestResidentManagerAppliesPositiveQueueTimeoutDurably(t *testing.T) {
783 started := make(chan string, 2)
784 release := make(chan struct{})
785 manager := NewResidentManagerWithPolicy(t.TempDir(), SubagentPolicy{MaxConcurrent: 1, QueueTimeout: 20 * time.Millisecond}, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
786 return func(_ context.Context, _ string) error {
787 started <- spec.ID
788 if spec.ID == "first" {
789 <-release
790 }
791 return nil
792 }, nil
793 })
794 defer manager.Close(context.Background())
795 for _, id := range []string{"first", "second"} {
796 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: id, SessionID: id, Provider: "openai", Model: "gpt-5"}, id); err != nil {
797 t.Fatalf("Spawn(%s): %v", id, err)
798 }
799 }
800 select {
801 case got := <-started:
802 if got != "first" {
803 t.Fatalf("first runner = %q, want first", got)
804 }
805 case <-time.After(time.Second):
806 t.Fatal("first runner did not start")
807 }
808 deadline := time.Now().Add(time.Second)
809 for {
810 result, err := manager.Result("second")
811 if err == nil && result.State == ResidentFailed {
812 break
813 }
814 if time.Now().After(deadline) {
815 t.Fatalf("queued child did not record timeout: result=%#v err=%v", result, err)
816 }
817 time.Sleep(time.Millisecond)
818 }
819 select {
820 case got := <-started:
821 t.Fatalf("timed out child ran as %q", got)
822 default:
823 }
824 close(release)
825}
826
827func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
828 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
829 return func(context.Context, string) error { return nil }, nil
830 })
831 defer manager.Close(context.Background())
832 accepted := make(chan string, 2)
833 manager.SetAcceptedObserver(func(_ ResidentChildSpec, _ string, prompt string) { accepted <- prompt })
834 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
835 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
836 t.Fatal(err)
837 }
838 if got := <-accepted; got != "initial" {
839 t.Fatalf("initial observer prompt = %q", got)
840 }
841 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
842 t.Fatal(err)
843 }
844 if got := <-accepted; got != "follow up" {
845 t.Fatalf("follow-up observer prompt = %q", got)
846 }
847}
848
849func TestResidentManagerReconcileReplacesStaleRecoveredSnapshots(t *testing.T) {
850 root := t.TempDir()
851 journal, err := OpenResidentJournal(root, "stale")
852 if err != nil {
853 t.Fatal(err)
854 }
855 spec := ResidentChildSpec{ID: "stale", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
856 if err := journal.Accept(spec, "task"); err != nil {
857 t.Fatal(err)
858 }
859 if err := journal.Close(); err != nil {
860 t.Fatal(err)
861 }
862 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) { return nil, nil })
863 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 1 {
864 t.Fatalf("first Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
865 }
866 if err := os.RemoveAll(filepath.Join(root, "stale")); err != nil {
867 t.Fatal(err)
868 }
869 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 0 {
870 t.Fatalf("second Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
871 }
872}
873
874func TestResidentManagerLimitsConcurrentTurnsGlobally(t *testing.T) {
875 started := make(chan string, DefaultResidentConcurrency+1)
876 release := make(chan struct{})
877 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
878 return func(ctx context.Context, prompt string) error {
879 started <- prompt
880 select {
881 case <-release:
882 return nil
883 case <-ctx.Done():
884 return ctx.Err()
885 }
886 }, nil
887 })
888 defer manager.Close(context.Background())
889
890 for i := 0; i < DefaultResidentConcurrency+1; i++ {
891 spec := ResidentChildSpec{ID: fmt.Sprintf("child-%d", i), SessionID: fmt.Sprintf("session-%d", i), Provider: "openai", Model: "gpt-5"}
892 if _, err := manager.Spawn(context.Background(), spec, spec.ID); err != nil {
893 t.Fatalf("Spawn(%d): %v", i, err)
894 }
895 }
896 for i := 0; i < DefaultResidentConcurrency; i++ {
897 select {
898 case <-started:
899 case <-time.After(time.Second):
900 t.Fatal("did not fill the default six slots")
901 }
902 }
903 select {
904 case extra := <-started:
905 t.Fatalf("seventh turn started before a slot released: %q", extra)
906 default:
907 }
908 close(release)
909 select {
910 case <-started:
911 case <-time.After(time.Second):
912 t.Fatal("queued seventh turn did not start after a slot released")
913 }
914}