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 TestResidentManagerCompletedAnswerCanResume(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: "completed-child", SessionID: "completed-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
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 != ResidentIdle {
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 TestResidentManagerCapturesFailedWorktreePatch(t *testing.T) {
476 root := t.TempDir()
477 workspace := &testResidentWorkspace{dir: "/isolated", mode: WorkspaceWorktree, capture: WorkspaceCapture{Patch: []byte("partial patch"), ChangedFiles: []string{"changed.go"}}}
478 manager := NewResidentManagerWithWorkspace(root, 0, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) {
479 return workspace, nil
480 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
481 return func(context.Context, string) error { return errors.New("turn failed") }, nil
482 })
483 defer manager.Close(context.Background())
484 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "failed-capture", SessionID: "child-session", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", WorkspaceMode: WorkspaceWorktree}, "task"); err != nil {
485 t.Fatal(err)
486 }
487 deadline := time.Now().Add(time.Second)
488 for manager.Snapshot()[0].State == ResidentRunning || manager.Snapshot()[0].State == ResidentQueued {
489 if time.Now().After(deadline) {
490 t.Fatal("turn did not finish")
491 }
492 time.Sleep(time.Millisecond)
493 }
494 // A non-canceled failure still captures partial work, but the turn
495 // stays failed: capture never converts failure to success.
496 if !workspace.captured {
497 t.Fatal("failed worktree was not captured")
498 }
499 snapshot := manager.Snapshot()[0]
500 if snapshot.State != ResidentFailed {
501 t.Fatalf("snapshot state = %q, want failed", snapshot.State)
502 }
503 result, err := manager.Result("failed-capture")
504 if err != nil {
505 t.Fatal(err)
506 }
507 if result.PatchRef != PatchRef("failed-capture") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "changed.go" {
508 t.Fatalf("result = %#v", result)
509 }
510 if result.Handoff != "" {
511 t.Fatalf("new failure result carries handoff %q, want empty", result.Handoff)
512 }
513}
514
515func TestResidentManagerRetainsFailedWorktree(t *testing.T) {
516 workspace := &testResidentWorkspace{dir: "/isolated", mode: WorkspaceWorktree}
517 manager := NewResidentManagerWithWorkspace(t.TempDir(), 0, func(context.Context, WorkspaceRequest) (WorkspaceHandle, error) { return workspace, nil }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
518 return func(context.Context, string) error { return errors.New("turn failed") }, nil
519 })
520 defer manager.Close(context.Background())
521 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 {
522 t.Fatal(err)
523 }
524 deadline := time.Now().Add(time.Second)
525 for manager.Snapshot()[0].State == ResidentRunning || manager.Snapshot()[0].State == ResidentQueued {
526 if time.Now().After(deadline) {
527 t.Fatal("turn did not finish")
528 }
529 time.Sleep(time.Millisecond)
530 }
531 // Non-canceled failures capture partial work for recovery, but the
532 // worktree is retained (never cleaned) until an explicit stop or close.
533 if !workspace.captured || workspace.cleaned {
534 t.Fatalf("failed worktree captured=%t cleaned=%t", workspace.captured, workspace.cleaned)
535 }
536}
537
538func TestResidentManagerCapturesRealWorktreePatch(t *testing.T) {
539 if _, err := exec.LookPath("git"); err != nil {
540 t.Skip("git is required")
541 }
542 root, repo := t.TempDir(), initTestRepo(t)
543 manager := NewResidentManager(root, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
544 return func(context.Context, string) error {
545 return os.WriteFile(filepath.Join(spec.Workspace, "README.md"), []byte("resident change\n"), 0o600)
546 }, nil
547 })
548 defer manager.Close(context.Background())
549 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")
550 if err != nil {
551 t.Fatal(err)
552 }
553 // Creating and diffing a real Git worktree can exceed a second on Windows,
554 // especially under filesystem scanning. Keep a bounded wait while allowing
555 // the actual asynchronous work to complete.
556 deadline := time.Now().Add(5 * time.Second)
557 for child.State() != ResidentIdle {
558 if time.Now().After(deadline) {
559 t.Fatal("worktree turn did not finish")
560 }
561 time.Sleep(time.Millisecond)
562 }
563 result, err := manager.Result("git-worktree")
564 if err != nil {
565 t.Fatal(err)
566 }
567 if result.PatchRef != PatchRef("git-worktree") || len(result.ChangedFiles) != 1 || result.ChangedFiles[0] != "README.md" {
568 t.Fatalf("result = %#v", result)
569 }
570 patch, err := os.ReadFile(filepath.Join(root, "git-worktree", residentPatchName))
571 if err != nil || !strings.Contains(string(patch), "resident change") {
572 t.Fatalf("patch = %q, err=%v", patch, err)
573 }
574 if err := manager.Stop(context.Background(), "git-worktree"); err != nil {
575 t.Fatal(err)
576 }
577 if _, err := os.Stat(child.workspace.Dir()); !os.IsNotExist(err) {
578 t.Fatalf("worktree remains after stop: %v", err)
579 }
580}
581
582type testResidentWorkspace struct {
583 dir string
584 mode WorkspaceMode
585 capture WorkspaceCapture
586 captured, cleaned bool
587}
588
589func (w *testResidentWorkspace) Dir() string { return w.dir }
590func (w *testResidentWorkspace) RepositoryRoot() string { return "/repo" }
591func (w *testResidentWorkspace) Mode() WorkspaceMode { return w.mode }
592func (w *testResidentWorkspace) Capture(context.Context) (WorkspaceCapture, error) {
593 w.captured = true
594 return w.capture, nil
595}
596func (w *testResidentWorkspace) Cleanup(context.Context) error { w.cleaned = true; return nil }
597
598func TestResidentManagerRejectsDuplicateBeforeSecondJournalAcceptance(t *testing.T) {
599 root := t.TempDir()
600 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
601 return func(context.Context, string) error { return nil }, nil
602 })
603 defer manager.Close(context.Background())
604 spec := ResidentChildSpec{ID: "duplicate", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
605 if _, err := manager.Spawn(context.Background(), spec, "first"); err != nil {
606 t.Fatal(err)
607 }
608 if _, err := manager.Spawn(context.Background(), spec, "second"); err == nil {
609 t.Fatal("duplicate Spawn succeeded")
610 }
611 records, err := ReadResidentJournal(filepath.Join(root, spec.ID, residentTranscriptName))
612 if err != nil {
613 t.Fatal(err)
614 }
615 accepted := 0
616 for _, record := range records {
617 if record.Type == residentRecordAccepted {
618 accepted++
619 if record.Prompt != "first" {
620 t.Fatalf("accepted prompt = %q, want first", record.Prompt)
621 }
622 }
623 }
624 if accepted != 1 {
625 t.Fatalf("records = %#v, want one accepted spawn", records)
626 }
627}
628
629func TestResidentManagerReconcileSkipsIncompatibleJournal(t *testing.T) {
630 root := t.TempDir()
631 journal, err := OpenResidentJournal(root, "legacy-version")
632 if err != nil {
633 t.Fatal(err)
634 }
635 spec := ResidentChildSpec{ID: "legacy-version", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
636 if err := journal.Accept(spec, "review"); err != nil {
637 t.Fatal(err)
638 }
639 if err := journal.Close(); err != nil {
640 t.Fatal(err)
641 }
642 transcript := filepath.Join(root, spec.ID, residentTranscriptName)
643 data, err := os.ReadFile(transcript)
644 if err != nil {
645 t.Fatal(err)
646 }
647 legacy := strings.Replace(string(data), `"version":3`, `"version":1`, 1)
648 if legacy == string(data) {
649 t.Fatal("accepted record has no journal version")
650 }
651 if err := os.WriteFile(transcript, []byte(legacy), 0o600); err != nil {
652 t.Fatal(err)
653 }
654 manager := NewResidentManager(root, nil)
655 t.Cleanup(func() { _ = manager.Close(context.Background()) })
656 for range 2 {
657 if errs := manager.Reconcile(); len(errs) != 0 {
658 t.Fatalf("Reconcile errors = %v", errs)
659 }
660 }
661 if _, err := ReconcileResidentJournal(filepath.Dir(transcript)); !errors.Is(err, ErrIncompatibleResidentJournal) {
662 t.Fatalf("explicit reconciliation error = %v, want incompatible version", err)
663 }
664 if snapshots := manager.Snapshot(); len(snapshots) != 0 {
665 t.Fatalf("snapshots = %#v", snapshots)
666 }
667 unchanged, err := os.ReadFile(transcript)
668 if err != nil {
669 t.Fatal(err)
670 }
671 if string(unchanged) != legacy {
672 t.Fatal("Reconcile modified incompatible journal")
673 }
674
675 valid, err := OpenResidentJournal(root, "valid")
676 if err != nil {
677 t.Fatal(err)
678 }
679 spec.ID = "valid"
680 if err := valid.Accept(spec, "keep discoverable"); err != nil {
681 t.Fatal(err)
682 }
683 if err := valid.Close(); err != nil {
684 t.Fatal(err)
685 }
686 corrupt, err := OpenResidentJournal(root, "corrupt")
687 if err != nil {
688 t.Fatal(err)
689 }
690 if err := corrupt.Close(); err != nil {
691 t.Fatal(err)
692 }
693 for range 2 {
694 errs := manager.Reconcile()
695 if len(errs) != 1 || !strings.Contains(errs[0].Error(), "resident child corrupt:") {
696 t.Fatalf("Reconcile errors = %v, want corrupt child error only", errs)
697 }
698 if snapshots := manager.Snapshot(); len(snapshots) != 1 || snapshots[0].ID != "valid" {
699 t.Fatalf("snapshots = %#v, want valid sibling only", snapshots)
700 }
701 }
702}
703
704func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutReplay(t *testing.T) {
705 root := t.TempDir()
706 journal, err := OpenResidentJournal(root, "recovered")
707 if err != nil {
708 t.Fatal(err)
709 }
710 spec := ResidentChildSpec{ID: "recovered", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5", Required: true}
711 if err := journal.Accept(spec, "do not replay"); err != nil {
712 t.Fatal(err)
713 }
714 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
715 t.Fatal(err)
716 }
717 if err := journal.Close(); err != nil {
718 t.Fatal(err)
719 }
720 runs := 0
721 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
722 runs++
723 return func(context.Context, string) error { return nil }, nil
724 })
725 if errs := manager.Reconcile(); len(errs) != 0 {
726 t.Fatalf("Reconcile errors = %v", errs)
727 }
728 if runs != 0 {
729 t.Fatalf("factory ran %d times during reconciliation", runs)
730 }
731 snapshots := manager.Snapshot()
732 if len(snapshots) != 1 || snapshots[0].State != ResidentInterrupted || !snapshots[0].Required {
733 t.Fatalf("snapshots = %#v", snapshots)
734 }
735 if _, err := manager.Spawn(context.Background(), spec, "overwrite"); err == nil {
736 t.Fatal("Spawn overwrote recovered journal")
737 }
738}
739
740func TestResidentManagerReconcileDoesNotMutateForeignProcessJournal(t *testing.T) {
741 if os.Getenv("ZUT_RESIDENT_MANAGER_HELPER") == "1" {
742 root := os.Getenv("ZUT_RESIDENT_MANAGER_ROOT")
743 journal, err := OpenResidentJournal(root, "foreign-process")
744 if err != nil {
745 t.Fatal(err)
746 }
747 spec := ResidentChildSpec{ID: "foreign-process", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
748 if err := journal.Accept(spec, "active task"); err != nil {
749 t.Fatal(err)
750 }
751 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
752 t.Fatal(err)
753 }
754 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 {
755 t.Fatal(err)
756 }
757 fmt.Fprintln(os.Stdout, "ready")
758 if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
759 t.Fatal(err)
760 }
761 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 {
762 t.Fatal(err)
763 }
764 if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
765 t.Fatal(err)
766 }
767 if err := journal.Close(); err != nil {
768 t.Fatal(err)
769 }
770 return
771 }
772
773 root := t.TempDir()
774 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
775 defer cancel()
776 cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentManagerReconcileDoesNotMutateForeignProcessJournal$")
777 defer func() {
778 if cmd.Process != nil {
779 _ = cmd.Process.Kill()
780 }
781 _ = cmd.Wait()
782 }()
783 cmd.Env = append(os.Environ(), "ZUT_RESIDENT_MANAGER_HELPER=1", "ZUT_RESIDENT_MANAGER_ROOT="+root)
784 stdout, err := cmd.StdoutPipe()
785 if err != nil {
786 t.Fatal(err)
787 }
788 stdin, err := cmd.StdinPipe()
789 if err != nil {
790 t.Fatal(err)
791 }
792 if err := cmd.Start(); err != nil {
793 t.Fatal(err)
794 }
795 ready, err := bufio.NewReader(stdout).ReadString('\n')
796 if err != nil {
797 t.Fatal(err)
798 }
799 if ready != "ready\n" {
800 t.Fatalf("helper readiness = %q", ready)
801 }
802 transcript := filepath.Join(root, "foreign-process", residentTranscriptName)
803 before, err := os.ReadFile(transcript)
804 if err != nil {
805 t.Fatal(err)
806 }
807
808 observer := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
809 return func(context.Context, string) error { return nil }, nil
810 })
811 if errs := observer.Reconcile(); len(errs) != 0 {
812 t.Fatalf("Reconcile errors = %v", errs)
813 }
814 after, err := os.ReadFile(transcript)
815 if err != nil {
816 t.Fatal(err)
817 }
818 if string(after) != string(before) {
819 t.Fatal("foreign reconciliation modified the transcript")
820 }
821 snapshot, ok := observer.SnapshotFor("foreign-process")
822 if !ok || !snapshot.OwnedElsewhere || snapshot.State != ResidentRunning {
823 t.Fatalf("foreign snapshot = %#v, found=%v", snapshot, ok)
824 }
825 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err == nil || !strings.Contains(err.Error(), "owned by another") {
826 t.Fatalf("foreign Resume error = %v", err)
827 }
828 if _, err := observer.History("foreign-process", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
829 t.Fatalf("foreign History error = %v", err)
830 }
831 if _, err := observer.HistoryPage("foreign-process", "", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
832 t.Fatalf("foreign HistoryPage error = %v", err)
833 }
834 if _, err := observer.Result("foreign-process"); err == nil || !strings.Contains(err.Error(), "owned by another") {
835 t.Fatalf("foreign Result error = %v", err)
836 }
837 if _, err := fmt.Fprintln(stdin); err != nil {
838 t.Fatal(err)
839 }
840 if err := stdin.Close(); err != nil {
841 t.Fatal(err)
842 }
843 if err := cmd.Wait(); err != nil {
844 t.Fatal(err)
845 }
846 if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err != nil {
847 t.Fatalf("Resume after owner exit: %v", err)
848 }
849 if err := observer.Stop(context.Background(), "foreign-process"); err != nil {
850 t.Fatalf("Stop rebuilt child: %v", err)
851 }
852}
853
854func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
855 root := t.TempDir()
856 journal, err := OpenResidentJournal(root, "resume-recovered")
857 if err != nil {
858 t.Fatal(err)
859 }
860 spec := ResidentChildSpec{ID: "resume-recovered", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5"}
861 if err := journal.Accept(spec, "old task must not replay"); err != nil {
862 t.Fatal(err)
863 }
864 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
865 t.Fatal(err)
866 }
867 if err := journal.Close(); err != nil {
868 t.Fatal(err)
869 }
870 runs := make(chan string, 1)
871 manager := NewResidentManager(root, func(got ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
872 if got.SessionID != spec.SessionID {
873 t.Fatalf("session = %q, want %q", got.SessionID, spec.SessionID)
874 }
875 return func(_ context.Context, prompt string) error { runs <- prompt; return nil }, nil
876 })
877 defer manager.Close(context.Background())
878 if errs := manager.Reconcile(); len(errs) != 0 {
879 t.Fatalf("Reconcile = %v", errs)
880 }
881 if err := manager.Resume(context.Background(), spec.ID, "new explicit prompt"); err != nil {
882 t.Fatal(err)
883 }
884 select {
885 case got := <-runs:
886 if got != "new explicit prompt" {
887 t.Fatalf("runner prompt = %q", got)
888 }
889 case <-time.After(time.Second):
890 t.Fatal("explicit resume did not run")
891 }
892}
893
894func TestResidentManagerExplicitResumeReusesInterruptedWorktree(t *testing.T) {
895 root := t.TempDir()
896 journal, err := OpenResidentJournal(root, "resume-worktree")
897 if err != nil {
898 t.Fatal(err)
899 }
900 spec := ResidentChildSpec{ID: "resume-worktree", SessionID: "child-session", InitialTurnID: "old-turn", Provider: "openai", Model: "gpt-5", RepositoryRoot: "/repo", Workspace: "/prior-worktree", WorkspaceMode: WorkspaceWorktree}
901 if err := journal.Accept(spec, "old task"); err != nil {
902 t.Fatal(err)
903 }
904 if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
905 t.Fatal(err)
906 }
907 if err := journal.Close(); err != nil {
908 t.Fatal(err)
909 }
910 workspace := &testResidentWorkspace{dir: "/prior-worktree", mode: WorkspaceWorktree}
911 manager := NewResidentManagerWithWorkspace(root, 0, func(_ context.Context, req WorkspaceRequest) (WorkspaceHandle, error) {
912 if req.ExistingPath != "/prior-worktree" {
913 t.Fatalf("resume workspace request = %#v", req)
914 }
915 return workspace, nil
916 }, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
917 return func(context.Context, string) error { return nil }, nil
918 })
919 defer manager.Close(context.Background())
920 if errs := manager.Reconcile(); len(errs) != 0 {
921 t.Fatal(errs)
922 }
923 if err := manager.Resume(context.Background(), spec.ID, "resume"); err != nil {
924 t.Fatal(err)
925 }
926}
927
928func TestResidentManagerAppliesPositiveQueueTimeoutDurably(t *testing.T) {
929 started := make(chan string, 2)
930 release := make(chan struct{})
931 manager := NewResidentManagerWithPolicy(t.TempDir(), SubagentPolicy{MaxConcurrent: 1, QueueTimeout: 20 * time.Millisecond}, func(spec ResidentChildSpec, _ *ResidentJournal) (ResidentTurnRunner, error) {
932 return func(_ context.Context, _ string) error {
933 started <- spec.ID
934 if spec.ID == "first" {
935 <-release
936 }
937 return nil
938 }, nil
939 })
940 defer manager.Close(context.Background())
941 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "first", SessionID: "first", Provider: "openai", Model: "gpt-5"}, "first"); err != nil {
942 t.Fatalf("Spawn(first): %v", err)
943 }
944 select {
945 case got := <-started:
946 if got != "first" {
947 t.Fatalf("first runner = %q, want first", got)
948 }
949 case <-time.After(time.Second):
950 t.Fatal("first runner did not start")
951 }
952 if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "second", SessionID: "second", Provider: "openai", Model: "gpt-5"}, "second"); err != nil {
953 t.Fatalf("Spawn(second): %v", err)
954 }
955 deadline := time.Now().Add(time.Second)
956 for {
957 result, err := manager.Result("second")
958 if err == nil && result.State == ResidentFailed {
959 break
960 }
961 if time.Now().After(deadline) {
962 t.Fatalf("queued child did not record timeout: result=%#v err=%v", result, err)
963 }
964 time.Sleep(time.Millisecond)
965 }
966 select {
967 case got := <-started:
968 t.Fatalf("timed out child ran as %q", got)
969 default:
970 }
971 close(release)
972}
973
974func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
975 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
976 return func(context.Context, string) error { return nil }, nil
977 })
978 defer manager.Close(context.Background())
979 accepted := make(chan string, 2)
980 manager.SetAcceptedObserver(func(_ ResidentChildSpec, _ string, prompt string) { accepted <- prompt })
981 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
982 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
983 t.Fatal(err)
984 }
985 if got := <-accepted; got != "initial" {
986 t.Fatalf("initial observer prompt = %q", got)
987 }
988 if err := manager.Resume(context.Background(), spec.ID, "follow up"); err != nil {
989 t.Fatal(err)
990 }
991 if got := <-accepted; got != "follow up" {
992 t.Fatalf("follow-up observer prompt = %q", got)
993 }
994}
995
996func TestResidentManagerReconcileReplacesStaleRecoveredSnapshots(t *testing.T) {
997 root := t.TempDir()
998 journal, err := OpenResidentJournal(root, "stale")
999 if err != nil {
1000 t.Fatal(err)
1001 }
1002 spec := ResidentChildSpec{ID: "stale", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}
1003 if err := journal.Accept(spec, "task"); err != nil {
1004 t.Fatal(err)
1005 }
1006 if err := journal.Close(); err != nil {
1007 t.Fatal(err)
1008 }
1009 manager := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) { return nil, nil })
1010 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 1 {
1011 t.Fatalf("first Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
1012 }
1013 if err := os.RemoveAll(filepath.Join(root, "stale")); err != nil {
1014 t.Fatal(err)
1015 }
1016 if errs := manager.Reconcile(); len(errs) != 0 || len(manager.Snapshot()) != 0 {
1017 t.Fatalf("second Reconcile = %v, snapshots=%#v", errs, manager.Snapshot())
1018 }
1019}
1020
1021func TestResidentManagerLimitsConcurrentTurnsGlobally(t *testing.T) {
1022 started := make(chan string, DefaultResidentConcurrency+1)
1023 release := make(chan struct{})
1024 manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
1025 return func(ctx context.Context, prompt string) error {
1026 started <- prompt
1027 select {
1028 case <-release:
1029 return nil
1030 case <-ctx.Done():
1031 return ctx.Err()
1032 }
1033 }, nil
1034 })
1035 defer manager.Close(context.Background())
1036
1037 for i := 0; i < DefaultResidentConcurrency+1; i++ {
1038 spec := ResidentChildSpec{ID: fmt.Sprintf("child-%d", i), SessionID: fmt.Sprintf("session-%d", i), Provider: "openai", Model: "gpt-5"}
1039 if _, err := manager.Spawn(context.Background(), spec, spec.ID); err != nil {
1040 t.Fatalf("Spawn(%d): %v", i, err)
1041 }
1042 }
1043 for i := 0; i < DefaultResidentConcurrency; i++ {
1044 select {
1045 case <-started:
1046 case <-time.After(time.Second):
1047 t.Fatal("did not fill the default six slots")
1048 }
1049 }
1050 select {
1051 case extra := <-started:
1052 t.Fatalf("seventh turn started before a slot released: %q", extra)
1053 default:
1054 }
1055 close(release)
1056 select {
1057 case <-started:
1058 case <-time.After(time.Second):
1059 t.Fatal("queued seventh turn did not start after a slot released")
1060 }
1061}