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