1package subagents
2
3import (
4 "context"
5 "path/filepath"
6 "runtime"
7 "testing"
8 "time"
9)
10
11func TestResumeWithPromptDeliversFollowUpToIdleWorker(t *testing.T) {
12 if runtime.GOOS == "windows" {
13 t.Skip("subagent inbox transport uses Unix-domain sockets")
14 }
15
16 root := shortSocketDir(t)
17 ready := make(chan struct{}, 1)
18 commands := make(chan Envelope, 2)
19 f := New(Config{
20 Root: root, RepoRoot: root,
21 NewRunner: func(a *Agent) Runner {
22 return RunnerFunc(func(ctx context.Context, _ Sink) error {
23 listener, err := Listen(a.InboxPath)
24 if err != nil {
25 return err
26 }
27 defer listener.Close()
28 a.setProcessState(ProcessAlive)
29 a.setTurnState(TurnIdle, "")
30 ready <- struct{}{}
31 for {
32 select {
33 case <-ctx.Done():
34 return ctx.Err()
35 case line, ok := <-listener.Lines():
36 if !ok {
37 return nil
38 }
39 command, err := ParseCommand(line)
40 if err != nil {
41 return err
42 }
43 if command.Type == CommandAgentShutdown {
44 return nil
45 }
46 commands <- command
47 }
48 }
49 })
50 },
51 })
52 defer f.StopAll()
53
54 a, err := f.Spawn(context.Background(), "review the implementation")
55 if err != nil {
56 t.Fatal(err)
57 }
58 select {
59 case <-ready:
60 case <-time.After(time.Second):
61 t.Fatal("idle worker did not open its inbox")
62 }
63
64 const followUp = "I applied your review. What do you think now?"
65 continued, err := f.ResumeWithPrompt(context.Background(), a.ID, followUp)
66 if err != nil {
67 t.Fatalf("continue idle worker: %v", err)
68 }
69 if continued != a {
70 t.Fatalf("continued agent = %p, want existing worker %p", continued, a)
71 }
72 if got := a.TurnState(); got != TurnQueued {
73 t.Fatalf("turn state after sending follow-up = %s, want %s", got, TurnQueued)
74 }
75 if _, err := f.ResumeWithPrompt(context.Background(), a.ID, "duplicate follow-up"); err == nil {
76 t.Fatal("second follow-up succeeded while the first was queued")
77 }
78 select {
79 case command := <-commands:
80 if command.Type != CommandTurnStart {
81 t.Fatalf("command type = %q, want %q", command.Type, CommandTurnStart)
82 }
83 var payload TurnStartPayload
84 if err := command.DecodePayload(&payload); err != nil {
85 t.Fatal(err)
86 }
87 if payload.Prompt != followUp {
88 t.Fatalf("follow-up prompt = %q, want %q", payload.Prompt, followUp)
89 }
90 case <-time.After(time.Second):
91 t.Fatal("idle worker did not receive the follow-up")
92 }
93
94 persisted, err := readAgentMeta(filepath.Join(root, "agents", a.ID))
95 if err != nil {
96 t.Fatal(err)
97 }
98 if persisted.ResumePrompt != followUp || persisted.ResumePromptAt.IsZero() {
99 t.Fatalf("queued follow-up was not persisted: prompt %q accepted %s", persisted.ResumePrompt, persisted.ResumePromptAt)
100 }
101 if err := f.Stop(a.ID); err != nil {
102 t.Fatal(err)
103 }
104 a.Wait()
105
106 restarted := make(chan *Agent, 1)
107 f2 := New(Config{
108 Root: root, RepoRoot: root,
109 NewRunner: func(agent *Agent) Runner {
110 return RunnerFunc(func(ctx context.Context, _ Sink) error {
111 restarted <- agent
112 <-ctx.Done()
113 return ctx.Err()
114 })
115 },
116 })
117 defer f2.StopAll()
118 if loaded, errs := f2.Reload(); loaded != 1 || len(errs) != 0 {
119 t.Fatalf("reload = (%d, %v), want (1, no errors)", loaded, errs)
120 }
121 resumed, err := f2.ResumeSession(context.Background(), a.ID)
122 if err != nil {
123 t.Fatal(err)
124 }
125 select {
126 case got := <-restarted:
127 if got != resumed || got.resumePrompt() != followUp {
128 t.Fatalf("unacknowledged follow-up was not replayed: agent %p prompt %q", got, got.resumePrompt())
129 }
130 case <-time.After(time.Second):
131 t.Fatal("restarted worker did not receive the follow-up")
132 }
133}