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 // Creating and diffing a real Git worktree can exceed a second on Windows,
486 // especially under filesystem scanning. Keep a bounded wait while allowing
487 // the actual asynchronous work to complete.
488 deadline := time.Now().Add(5 * time.Second)
489 for child.State() != ResidentIdle {
490 if time.Now().After(deadline) {
491 t.Fatal("worktree turn did not finish")
492 }
493 time.Sleep(time.Millisecond)
494 }
495 result, err := manager.Result("git-worktree")
496 if err != nil {
497 t.Fatal(err)
498 }
499 if result.PatchRef != PatchRef("git-worktree") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "README.md" {
500 t.Fatalf("result = %#v", result)
501 }
502 patch, err := os.ReadFile(filepath.Join(root, "git-worktree", residentPatchName))
503 if err != nil || !strings.Contains(string(patch), "resident change") {
504 t.Fatalf("patch = %q, err=%v", patch, err)
505 }
506 if err := manager.Stop(context.Background(), "git-worktree"); err != nil {
507 t.Fatal(err)
508 }
509 if _, err := os.Stat(child.workspace.Dir()); !os.IsNotExist(err) {
510 t.Fatalf("worktree remains after stop: %v", err)
511 }
512}
513
514type testResidentWorkspace struct {
515 dir string
516 mode WorkspaceMode
517 capture WorkspaceCapture
518 captured, cleaned bool
519}
520
521func (w *testResidentWorkspace) Dir() string { return w.dir }
522func (w *testResidentWorkspace) RepositoryRoot() string { return "/repo" }
523func (w *testResidentWorkspace) Mode() WorkspaceMode { return w.mode }
524func (w *testResidentWorkspace) Capture(context.Context) (WorkspaceCapture, error) {
525 w.captured = true
526 return w.capture, nil
527}
528func (w *testResidentWorkspace) Cleanup(context.Context) error { w.cleaned = true; return nil }
529
530func TestResidentManagerRejectsDuplicateBeforeSecondJournalAcceptance(t *testing.T) {
531 root := t.TempDir()
532 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
533 return func(context.Context, string) error { return nil }, nil
534 })
535 defer manager.Close(context.Background())
536 spec := ResidentChildSpec{ID: "duplicate", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
537 if _, err := manager.Spawn(context.Background(), spec, "first"); err != nil {
538 t.Fatal(err)
539 }
540 if _, err := manager.Spawn(context.Background(), spec, "second"); err == nil {
541 t.Fatal("duplicate Spawn succeeded")
542 }
543 records, err := ReadResidentJournal(filepath.Join(root, spec.ID, residentTranscriptName))
544 if err != nil {
545 t.Fatal(err)
546 }
547 accepted := 0
548 for _, record := range records {
549 if record.Type == residentRecordAccepted {
550 accepted++
551 if record.Prompt != "first" {
552 t.Fatalf("accepted prompt = %q, want first", record.Prompt)
553 }
554 }
555 }
556 if accepted != 1 {
557 t.Fatalf("records = %#v, want one accepted spawn", records)
558 }
559}
560
561func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutReplay(t *testing.T) {
562 root := t.TempDir()
563 journal, err := OpenResidentJournal(root, "recovered")
564 if err != nil {
565 t.Fatal(err)
566 }
567 spec := ResidentChildSpec{ID: "recovered", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5", Required: true}
568 if err := journal.Accept(spec, "do not replay"); err != nil {
569 t.Fatal(err)
570 }
571 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
572 t.Fatal(err)
573 }
574 if err := journal.Close(); err != nil {
575 t.Fatal(err)
576 }
577 runs := 0
578 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
579 runs++
580 return func(context.Context, string) error { return nil }, nil
581 })
582 if errs := manager.Reconcile(); len(errs) != 0 {
583 t.Fatalf("Reconcile errors = %v", errs)
584 }
585 if runs != 0 {
586 t.Fatalf("factory ran %d times during reconciliation", runs)
587 }
588 snapshots := manager.Snapshot()
589 if len(snapshots) != 1 || snapshots[0].State != ResidentInterrupted || !snapshots[0].Required {
590 t.Fatalf("snapshots = %#v", snapshots)
591 }
592 if _, err := manager.Spawn(context.Background(), spec, "overwrite"); err == nil {
593 t.Fatal("Spawn overwrote recovered journal")
594 }
595}
596
597func TestResidentManagerReconcileDoesNotMutateForeignProcessJournal(t *testing.T) {
598 if os.Getenv("ZUT_RESIDENT_MANAGER_HELPER") == "1" {
599 root := os.Getenv("ZUT_RESIDENT_MANAGER_ROOT")
600 journal, err := OpenResidentJournal(root, "foreign-process")
601 if err != nil {
602 t.Fatal(err)
603 }
604 spec := ResidentChildSpec{ID: "foreign-process", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
605 if err := journal.Accept(spec, "active task"); err != nil {
606 t.Fatal(err)
607 }
608 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
609 t.Fatal(err)
610 }
611 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 {
612 t.Fatal(err)
613 }
614 fmt.Fprintln(os.Stdout, "ready")
615 if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
616 t.Fatal(err)
617 }
618 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 {
619 t.Fatal(err)
620 }
621 if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
622 t.Fatal(err)
623 }
624 if err := journal.Close(); err != nil {
625 t.Fatal(err)
626 }
627 return
628 }
629
630 root := t.TempDir()
631 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
632 defer cancel()
633 cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentManagerReconcileDoesNotMutateForeignProcessJournal$")
634 defer func() {
635 if cmd.Process != nil {
636 _ = cmd.Process.Kill()
637 }
638 _ = cmd.Wait()
639 }()
640 cmd.Env = append(os.Environ(), "ZUT_RESIDENT_MANAGER_HELPER=1", "ZUT_RESIDENT_MANAGER_ROOT="+root)
641 stdout, err := cmd.StdoutPipe()
642 if err != nil {
643 t.Fatal(err)
644 }
645 stdin, err := cmd.StdinPipe()
646 if err != nil {
647 t.Fatal(err)
648 }
649 if err := cmd.Start(); err != nil {
650 t.Fatal(err)
651 }
652 ready, err := bufio.NewReader(stdout).ReadString('\n')
653 if err != nil {
654 t.Fatal(err)
655 }
656 if ready != "ready\n" {
657 t.Fatalf("helper readiness = %q", ready)
658 }
659 transcript := filepath.Join(root, "foreign-process", residentTranscriptName)
660 before, err := os.ReadFile(transcript)
661 if err != nil {
662 t.Fatal(err)
663 }
664
665 observer := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
666 return func(context.Context, string) error { return nil }, nil
667 })
668 if errs := observer.Reconcile(); len(errs) != 0 {
669 t.Fatalf("Reconcile errors = %v", errs)
670 }
671 after, err := os.ReadFile(transcript)
672 if err != nil {
673 t.Fatal(err)
674 }
675 if string(after) != string(before) {
676 t.Fatal("foreign reconciliation modified the transcript")
677 }
678 snapshot, ok := observer.SnapshotFor("foreign-process")
679 if !ok || !snapshot.OwnedElsewhere || snapshot.State != ResidentRunning {
680 t.Fatalf("foreign snapshot = %#v, found=%v", snapshot, ok)
681 }
682 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err == nil || !strings.Contains(err.Error(), "owned by another") {
683 t.Fatalf("foreign Resume error = %v", err)
684 }
685 if _, err := observer.History("foreign-process", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
686 t.Fatalf("foreign History error = %v", err)
687 }
688 if _, err := observer.HistoryPage("foreign-process", "", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
689 t.Fatalf("foreign HistoryPage error = %v", err)
690 }
691 if _, err := observer.Result("foreign-process"); err == nil || !strings.Contains(err.Error(), "owned by another") {
692 t.Fatalf("foreign Result error = %v", err)
693 }
694 if _, err := fmt.Fprintln(stdin); err != nil {
695 t.Fatal(err)
696 }
697 if err := stdin.Close(); err != nil {
698 t.Fatal(err)
699 }
700 if err := cmd.Wait(); err != nil {
701 t.Fatal(err)
702 }
703 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err != nil {
704 t.Fatalf("Resume after owner exit: %v", err)
705 }
706 if err := observer.Stop(context.Background(), "foreign-process"); err != nil {
707 t.Fatalf("Stop rebuilt child: %v", err)
708 }
709}
710
711func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
712 root := t.TempDir()
713 journal, err := OpenResidentJournal(root, "resume-recovered")
714 if err != nil {
715 t.Fatal(err)
716 }
717 spec := ResidentChildSpec{ID: "resume-recovered", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5"}
718 if err := journal.Accept(spec, "old task must not replay"); err != nil {
719 t.Fatal(err)
720 }
721 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
722 t.Fatal(err)
723 }
724 if err := journal.Close(); err != nil {
725 t.Fatal(err)
726 }
727 runs := make(chan string, 1)
728 manager := NewResidentManager(root, func(got ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
729 if got.SessionID != spec.SessionID {
730 t.Fatalf("session = %q, want %q", got.SessionID, spec.SessionID)
731 }
732 return func(_ context.Context, prompt string) error { runs <- prompt; return nil }, nil
733 })
734 defer manager.Close(context.Background())
735 if errs := manager.Reconcile(); len(errs) != 0 {
736 t.Fatalf("Reconcile = %v", errs)
737 }
738 if err := manager.Resume(context.Background(), spec.ID, "new explicit prompt"); err != nil {
739 t.Fatal(err)
740 }
741 select {
742 case got := <-runs:
743 if got != "new explicit prompt" {
744 t.Fatalf("runner prompt = %q", got)
745 }
746 case <-time.After(time.Second):
747 t.Fatal("explicit resume did not run")
748 }
749}
750
751func TestResidentManagerExplicitResumeReusesInterruptedWorktree(t *testing.T) {
752 root := t.TempDir()
753 journal, err := OpenResidentJournal(root, "resume-worktree")
754 if err != nil {
755 t.Fatal(err)
756 }
757 spec := ResidentChildSpec{ID: "resume-worktree", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", Workspace: "/prior-worktree", WorkspaceMode: WorkspaceWorktree}
758 if err := journal.Accept(spec, "old task"); err != nil {
759 t.Fatal(err)
760 }
761 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
762 t.Fatal(err)
763 }
764 if err := journal.Close(); err != nil {
765 t.Fatal(err)
766 }
767 workspace := &testResidentWorkspace{dir: "/prior-worktree", mode: WorkspaceWorktree}
768 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
769 if req.ExistingPath != "/prior-worktree" {
770 t.Fatalf("resume workspace request = %#v", req)
771 }
772 return workspace, nil
773 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
774 return func(context.Context, string) error { return nil }, nil
775 })
776 defer manager.Close(context.Background())
777 if errs := manager.Reconcile(); len(errs) != 0 {
778 t.Fatal(errs)
779 }
780 if err := manager.Resume(context.Background(), spec.ID, "resume"); err != nil {
781 t.Fatal(err)
782 }
783}
784
785func TestResidentManagerAppliesPositiveQueueTimeoutDurably(t *testing.T) {
786 started := make(chan string, 2)
787 release := make(chan struct{})
788 manager := NewResidentManagerWithPolicy(t.TempDir(), SubagentPolicy{MaxConcurrent: 1, QueueTimeout: 20 * time.Millisecond}, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
789 return func(_ context.Context, _ string) error {
790 started <- spec.ID
791 if spec.ID == "first" {
792 <-release
793 }
794 return nil
795 }, nil
796 })
797 defer manager.Close(context.Background())
798 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "first", SessionID: "first", Provider: "openai", Model: "gpt-5"}, "first"); err != nil {
799 t.Fatalf("Spawn(first): %v", err)
800 }
801 select {
802 case got := <-started:
803 if got != "first" {
804 t.Fatalf("first runner = %q, want first", got)
805 }
806 case <-time.After(time.Second):
807 t.Fatal("first runner did not start")
808 }
809 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "second", SessionID: "second", Provider: "openai", Model: "gpt-5"}, "second"); err != nil {
810 t.Fatalf("Spawn(second): %v", err)
811 }
812 deadline := time.Now().Add(time.Second)
813 for {
814 result, err := manager.Result("second")
815 if err == nil && result.State == ResidentFailed {
816 break
817 }
818 if time.Now().After(deadline) {
819 t.Fatalf("queued child did not record timeout: result=%#v err=%v", result, err)
820 }
821 time.Sleep(time.Millisecond)
822 }
823 select {
824 case got := <-started:
825 t.Fatalf("timed out child ran as %q", got)
826 default:
827 }
828 close(release)
829}
830
831func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
832 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
833 return func(context.Context, string) error { return nil }, nil
834 })
835 defer manager.Close(context.Background())
836 accepted := make(chan string, 2)
837 manager.SetAcceptedObserver(func(_ ResidentChildSpec, _ string, prompt string) { accepted <- prompt })
838 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
839 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
840 t.Fatal(err)
841 }
842 if got := <-accepted; got != "initial" {
843 t.Fatalf("initial observer prompt = %q", got)
844 }
845 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
846 t.Fatal(err)
847 }
848 if got := <-accepted; got != "follow up" {
849 t.Fatalf("follow-up observer prompt = %q", got)
850 }
851}
852
853func TestResidentManagerReconcileReplacesStaleRecoveredSnapshots(t *testing.T) {
854 root := t.TempDir()
855 journal, err := OpenResidentJournal(root, "stale")
856 if err != nil {
857 t.Fatal(err)
858 }
859 spec := ResidentChildSpec{ID: "stale", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
860 if err := journal.Accept(spec, "task"); err != nil {
861 t.Fatal(err)
862 }
863 if err := journal.Close(); err != nil {
864 t.Fatal(err)
865 }
866 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) { return nil, nil })
867 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 1 {
868 t.Fatalf("first Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
869 }
870 if err := os.RemoveAll(filepath.Join(root, "stale")); err != nil {
871 t.Fatal(err)
872 }
873 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 0 {
874 t.Fatalf("second Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
875 }
876}
877
878func TestResidentManagerLimitsConcurrentTurnsGlobally(t *testing.T) {
879 started := make(chan string, DefaultResidentConcurrency+1)
880 release := make(chan struct{})
881 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
882 return func(ctx context.Context, prompt string) error {
883 started <- prompt
884 select {
885 case <-release:
886 return nil
887 case <-ctx.Done():
888 return ctx.Err()
889 }
890 }, nil
891 })
892 defer manager.Close(context.Background())
893
894 for i := 0; i < DefaultResidentConcurrency+1; i++ {
895 spec := ResidentChildSpec{ID: fmt.Sprintf("child-%d", i), SessionID: fmt.Sprintf("session-%d", i), Provider: "openai", Model: "gpt-5"}
896 if _, err := manager.Spawn(context.Background(), spec, spec.ID); err != nil {
897 t.Fatalf("Spawn(%d): %v", i, err)
898 }
899 }
900 for i := 0; i < DefaultResidentConcurrency; i++ {
901 select {
902 case <-started:
903 case <-time.After(time.Second):
904 t.Fatal("did not fill the default six slots")
905 }
906 }
907 select {
908 case extra := <-started:
909 t.Fatalf("seventh turn started before a slot released: %q", extra)
910 default:
911 }
912 close(release)
913 select {
914 case <-started:
915 case <-time.After(time.Second):
916 t.Fatal("queued seventh turn did not start after a slot released")
917 }
918}