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 TestResidentManagerReconcileSkipsIncompatibleBudgetJournal(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 for range 2 {
615 if errs := manager.Reconcile(); len(errs) != 0 {
616 t.Fatalf("Reconcile errors = %v", errs)
617 }
618 }
619 if _, err := ReconcileResidentJournal(filepath.Dir(transcript)); !errors.Is(err, ErrIncompatibleResidentBudget) {
620 t.Fatalf("explicit reconciliation error = %v, want incompatible budget", err)
621 }
622 if snapshots := manager.Snapshot(); len(snapshots) != 0 {
623 t.Fatalf("snapshots = %#v", snapshots)
624 }
625 unchanged, err := os.ReadFile(transcript)
626 if err != nil {
627 t.Fatal(err)
628 }
629 if string(unchanged) != legacy {
630 t.Fatal("Reconcile modified incompatible journal")
631 }
632
633 valid, err := OpenResidentJournal(root, "valid")
634 if err != nil {
635 t.Fatal(err)
636 }
637 spec.ID = "valid"
638 if err := valid.Accept(spec, "keep discoverable"); err != nil {
639 t.Fatal(err)
640 }
641 if err := valid.Close(); err != nil {
642 t.Fatal(err)
643 }
644 corrupt, err := OpenResidentJournal(root, "corrupt")
645 if err != nil {
646 t.Fatal(err)
647 }
648 if err := corrupt.Close(); err != nil {
649 t.Fatal(err)
650 }
651 for range 2 {
652 errs := manager.Reconcile()
653 if len(errs) != 1 || !strings.Contains(errs[0].Error(), "resident child corrupt:") {
654 t.Fatalf("Reconcile errors = %v, want corrupt child error only", errs)
655 }
656 if snapshots := manager.Snapshot(); len(snapshots) != 1 || snapshots[0].ID != "valid" {
657 t.Fatalf("snapshots = %#v, want valid sibling only", snapshots)
658 }
659 }
660}
661
662func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutReplay(t *testing.T) {
663 root := t.TempDir()
664 journal, err := OpenResidentJournal(root, "recovered")
665 if err != nil {
666 t.Fatal(err)
667 }
668 spec := ResidentChildSpec{ID: "recovered", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5", Required: true}
669 if err := journal.Accept(spec, "do not replay"); err != nil {
670 t.Fatal(err)
671 }
672 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
673 t.Fatal(err)
674 }
675 if err := journal.Close(); err != nil {
676 t.Fatal(err)
677 }
678 runs := 0
679 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
680 runs++
681 return func(context.Context, string) error { return nil }, nil
682 })
683 if errs := manager.Reconcile(); len(errs) != 0 {
684 t.Fatalf("Reconcile errors = %v", errs)
685 }
686 if runs != 0 {
687 t.Fatalf("factory ran %d times during reconciliation", runs)
688 }
689 snapshots := manager.Snapshot()
690 if len(snapshots) != 1 || snapshots[0].State != ResidentInterrupted || !snapshots[0].Required {
691 t.Fatalf("snapshots = %#v", snapshots)
692 }
693 if _, err := manager.Spawn(context.Background(), spec, "overwrite"); err == nil {
694 t.Fatal("Spawn overwrote recovered journal")
695 }
696}
697
698func TestResidentManagerReconcileDoesNotMutateForeignProcessJournal(t *testing.T) {
699 if os.Getenv("ZUT_RESIDENT_MANAGER_HELPER") == "1" {
700 root := os.Getenv("ZUT_RESIDENT_MANAGER_ROOT")
701 journal, err := OpenResidentJournal(root, "foreign-process")
702 if err != nil {
703 t.Fatal(err)
704 }
705 spec := ResidentChildSpec{ID: "foreign-process", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
706 if err := journal.Accept(spec, "active task"); err != nil {
707 t.Fatal(err)
708 }
709 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
710 t.Fatal(err)
711 }
712 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 {
713 t.Fatal(err)
714 }
715 fmt.Fprintln(os.Stdout, "ready")
716 if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
717 t.Fatal(err)
718 }
719 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 {
720 t.Fatal(err)
721 }
722 if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
723 t.Fatal(err)
724 }
725 if err := journal.Close(); err != nil {
726 t.Fatal(err)
727 }
728 return
729 }
730
731 root := t.TempDir()
732 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
733 defer cancel()
734 cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentManagerReconcileDoesNotMutateForeignProcessJournal$")
735 defer func() {
736 if cmd.Process != nil {
737 _ = cmd.Process.Kill()
738 }
739 _ = cmd.Wait()
740 }()
741 cmd.Env = append(os.Environ(), "ZUT_RESIDENT_MANAGER_HELPER=1", "ZUT_RESIDENT_MANAGER_ROOT="+root)
742 stdout, err := cmd.StdoutPipe()
743 if err != nil {
744 t.Fatal(err)
745 }
746 stdin, err := cmd.StdinPipe()
747 if err != nil {
748 t.Fatal(err)
749 }
750 if err := cmd.Start(); err != nil {
751 t.Fatal(err)
752 }
753 ready, err := bufio.NewReader(stdout).ReadString('\n')
754 if err != nil {
755 t.Fatal(err)
756 }
757 if ready != "ready\n" {
758 t.Fatalf("helper readiness = %q", ready)
759 }
760 transcript := filepath.Join(root, "foreign-process", residentTranscriptName)
761 before, err := os.ReadFile(transcript)
762 if err != nil {
763 t.Fatal(err)
764 }
765
766 observer := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
767 return func(context.Context, string) error { return nil }, nil
768 })
769 if errs := observer.Reconcile(); len(errs) != 0 {
770 t.Fatalf("Reconcile errors = %v", errs)
771 }
772 after, err := os.ReadFile(transcript)
773 if err != nil {
774 t.Fatal(err)
775 }
776 if string(after) != string(before) {
777 t.Fatal("foreign reconciliation modified the transcript")
778 }
779 snapshot, ok := observer.SnapshotFor("foreign-process")
780 if !ok || !snapshot.OwnedElsewhere || snapshot.State != ResidentRunning {
781 t.Fatalf("foreign snapshot = %#v, found=%v", snapshot, ok)
782 }
783 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err == nil || !strings.Contains(err.Error(), "owned by another") {
784 t.Fatalf("foreign Resume error = %v", err)
785 }
786 if _, err := observer.History("foreign-process", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
787 t.Fatalf("foreign History error = %v", err)
788 }
789 if _, err := observer.HistoryPage("foreign-process", "", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
790 t.Fatalf("foreign HistoryPage error = %v", err)
791 }
792 if _, err := observer.Result("foreign-process"); err == nil || !strings.Contains(err.Error(), "owned by another") {
793 t.Fatalf("foreign Result error = %v", err)
794 }
795 if _, err := fmt.Fprintln(stdin); err != nil {
796 t.Fatal(err)
797 }
798 if err := stdin.Close(); err != nil {
799 t.Fatal(err)
800 }
801 if err := cmd.Wait(); err != nil {
802 t.Fatal(err)
803 }
804 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err != nil {
805 t.Fatalf("Resume after owner exit: %v", err)
806 }
807 if err := observer.Stop(context.Background(), "foreign-process"); err != nil {
808 t.Fatalf("Stop rebuilt child: %v", err)
809 }
810}
811
812func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
813 root := t.TempDir()
814 journal, err := OpenResidentJournal(root, "resume-recovered")
815 if err != nil {
816 t.Fatal(err)
817 }
818 spec := ResidentChildSpec{ID: "resume-recovered", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5"}
819 if err := journal.Accept(spec, "old task must not replay"); err != nil {
820 t.Fatal(err)
821 }
822 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
823 t.Fatal(err)
824 }
825 if err := journal.Close(); err != nil {
826 t.Fatal(err)
827 }
828 runs := make(chan string, 1)
829 manager := NewResidentManager(root, func(got ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
830 if got.SessionID != spec.SessionID {
831 t.Fatalf("session = %q, want %q", got.SessionID, spec.SessionID)
832 }
833 return func(_ context.Context, prompt string) error { runs <- prompt; return nil }, nil
834 })
835 defer manager.Close(context.Background())
836 if errs := manager.Reconcile(); len(errs) != 0 {
837 t.Fatalf("Reconcile = %v", errs)
838 }
839 if err := manager.Resume(context.Background(), spec.ID, "new explicit prompt"); err != nil {
840 t.Fatal(err)
841 }
842 select {
843 case got := <-runs:
844 if got != "new explicit prompt" {
845 t.Fatalf("runner prompt = %q", got)
846 }
847 case <-time.After(time.Second):
848 t.Fatal("explicit resume did not run")
849 }
850}
851
852func TestResidentManagerExplicitResumeReusesInterruptedWorktree(t *testing.T) {
853 root := t.TempDir()
854 journal, err := OpenResidentJournal(root, "resume-worktree")
855 if err != nil {
856 t.Fatal(err)
857 }
858 spec := ResidentChildSpec{ID: "resume-worktree", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", Workspace: "/prior-worktree", WorkspaceMode: WorkspaceWorktree}
859 if err := journal.Accept(spec, "old task"); err != nil {
860 t.Fatal(err)
861 }
862 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
863 t.Fatal(err)
864 }
865 if err := journal.Close(); err != nil {
866 t.Fatal(err)
867 }
868 workspace := &testResidentWorkspace{dir: "/prior-worktree", mode: WorkspaceWorktree}
869 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
870 if req.ExistingPath != "/prior-worktree" {
871 t.Fatalf("resume workspace request = %#v", req)
872 }
873 return workspace, nil
874 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
875 return func(context.Context, string) error { return nil }, nil
876 })
877 defer manager.Close(context.Background())
878 if errs := manager.Reconcile(); len(errs) != 0 {
879 t.Fatal(errs)
880 }
881 if err := manager.Resume(context.Background(), spec.ID, "resume"); err != nil {
882 t.Fatal(err)
883 }
884}
885
886func TestResidentManagerAppliesPositiveQueueTimeoutDurably(t *testing.T) {
887 started := make(chan string, 2)
888 release := make(chan struct{})
889 manager := NewResidentManagerWithPolicy(t.TempDir(), SubagentPolicy{MaxConcurrent: 1, QueueTimeout: 20 * time.Millisecond}, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
890 return func(_ context.Context, _ string) error {
891 started <- spec.ID
892 if spec.ID == "first" {
893 <-release
894 }
895 return nil
896 }, nil
897 })
898 defer manager.Close(context.Background())
899 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "first", SessionID: "first", Provider: "openai", Model: "gpt-5"}, "first"); err != nil {
900 t.Fatalf("Spawn(first): %v", err)
901 }
902 select {
903 case got := <-started:
904 if got != "first" {
905 t.Fatalf("first runner = %q, want first", got)
906 }
907 case <-time.After(time.Second):
908 t.Fatal("first runner did not start")
909 }
910 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "second", SessionID: "second", Provider: "openai", Model: "gpt-5"}, "second"); err != nil {
911 t.Fatalf("Spawn(second): %v", err)
912 }
913 deadline := time.Now().Add(time.Second)
914 for {
915 result, err := manager.Result("second")
916 if err == nil && result.State == ResidentFailed {
917 break
918 }
919 if time.Now().After(deadline) {
920 t.Fatalf("queued child did not record timeout: result=%#v err=%v", result, err)
921 }
922 time.Sleep(time.Millisecond)
923 }
924 select {
925 case got := <-started:
926 t.Fatalf("timed out child ran as %q", got)
927 default:
928 }
929 close(release)
930}
931
932func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
933 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
934 return func(context.Context, string) error { return nil }, nil
935 })
936 defer manager.Close(context.Background())
937 accepted := make(chan string, 2)
938 manager.SetAcceptedObserver(func(_ ResidentChildSpec, _ string, prompt string) { accepted <- prompt })
939 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
940 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
941 t.Fatal(err)
942 }
943 if got := <-accepted; got != "initial" {
944 t.Fatalf("initial observer prompt = %q", got)
945 }
946 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
947 t.Fatal(err)
948 }
949 if got := <-accepted; got != "follow up" {
950 t.Fatalf("follow-up observer prompt = %q", got)
951 }
952}
953
954func TestResidentManagerReconcileReplacesStaleRecoveredSnapshots(t *testing.T) {
955 root := t.TempDir()
956 journal, err := OpenResidentJournal(root, "stale")
957 if err != nil {
958 t.Fatal(err)
959 }
960 spec := ResidentChildSpec{ID: "stale", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
961 if err := journal.Accept(spec, "task"); err != nil {
962 t.Fatal(err)
963 }
964 if err := journal.Close(); err != nil {
965 t.Fatal(err)
966 }
967 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) { return nil, nil })
968 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 1 {
969 t.Fatalf("first Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
970 }
971 if err := os.RemoveAll(filepath.Join(root, "stale")); err != nil {
972 t.Fatal(err)
973 }
974 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 0 {
975 t.Fatalf("second Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
976 }
977}
978
979func TestResidentManagerLimitsConcurrentTurnsGlobally(t *testing.T) {
980 started := make(chan string, DefaultResidentConcurrency+1)
981 release := make(chan struct{})
982 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
983 return func(ctx context.Context, prompt string) error {
984 started <- prompt
985 select {
986 case <-release:
987 return nil
988 case <-ctx.Done():
989 return ctx.Err()
990 }
991 }, nil
992 })
993 defer manager.Close(context.Background())
994
995 for i := 0; i < DefaultResidentConcurrency+1; i++ {
996 spec := ResidentChildSpec{ID: fmt.Sprintf("child-%d", i), SessionID: fmt.Sprintf("session-%d", i), Provider: "openai", Model: "gpt-5"}
997 if _, err := manager.Spawn(context.Background(), spec, spec.ID); err != nil {
998 t.Fatalf("Spawn(%d): %v", i, err)
999 }
1000 }
1001 for i := 0; i < DefaultResidentConcurrency; i++ {
1002 select {
1003 case <-started:
1004 case <-time.After(time.Second):
1005 t.Fatal("did not fill the default six slots")
1006 }
1007 }
1008 select {
1009 case extra := <-started:
1010 t.Fatalf("seventh turn started before a slot released: %q", extra)
1011 default:
1012 }
1013 close(release)
1014 select {
1015 case <-started:
1016 case <-time.After(time.Second):
1017 t.Fatal("queued seventh turn did not start after a slot released")
1018 }
1019}