1package subagents
2
3import (
4 "context"
5 "errors"
6 "path/filepath"
7 "runtime"
8 "testing"
9 "time"
10)
11
12func TestResumeWithPromptQueuesFollowUpsForActiveWorker(t *testing.T) {
13 if runtime.GOOS == "windows" {
14 t.Skip("subagent inbox transport uses Unix-domain sockets")
15 }
16
17 root := shortSocketDir(t)
18 ready := make(chan struct{}, 1)
19 commands := make(chan Envelope, 2)
20 f := New(Config{
21 Root: root, RepoRoot: root,
22 NewRunner: func(a *Agent) Runner {
23 return RunnerFunc(func(ctx context.Context, _ Sink) error {
24 listener, err := Listen(a.InboxPath)
25 if err != nil {
26 return err
27 }
28 defer listener.Close()
29 a.setProcessState(ProcessAlive)
30 a.setTurnState(TurnRunning, "turn-1")
31 ready <- struct{}{}
32 for {
33 select {
34 case <-ctx.Done():
35 return ctx.Err()
36 case line, ok := <-listener.Lines():
37 if !ok {
38 return nil
39 }
40 command, err := ParseCommand(line)
41 if err != nil {
42 return err
43 }
44 if command.Type == CommandAgentShutdown {
45 return nil
46 }
47 commands <- command
48 }
49 }
50 })
51 },
52 })
53 defer f.StopAll()
54
55 a, err := f.Spawn(context.Background(), "review the implementation")
56 if err != nil {
57 t.Fatal(err)
58 }
59 select {
60 case <-ready:
61 case <-time.After(time.Second):
62 t.Fatal("active worker did not open its inbox")
63 }
64 canceled, cancel := context.WithCancel(context.Background())
65 cancel()
66 if _, err := f.ResumeWithPrompt(canceled, a.ID, "must not be accepted"); !errors.Is(err, context.Canceled) {
67 t.Fatalf("canceled active follow-up error = %v, want context canceled", err)
68 }
69
70 const (
71 firstFollowUp = "Check the parser behavior."
72 secondFollowUp = "Then check the error handling."
73 )
74 if _, err := f.ResumeWithPrompt(context.Background(), a.ID, firstFollowUp); err != nil {
75 t.Fatalf("queue first active follow-up: %v", err)
76 }
77 if _, err := f.ResumeWithPrompt(context.Background(), a.ID, secondFollowUp); err != nil {
78 t.Fatalf("queue second active follow-up: %v", err)
79 }
80 select {
81 case command := <-commands:
82 t.Fatalf("active follow-up was delivered before idle: %v", command.Type)
83 default:
84 }
85
86 persisted, err := readAgentMeta(filepath.Join(root, "agents", a.ID))
87 if err != nil {
88 t.Fatal(err)
89 }
90 if persisted.ResumePrompt != "" || len(persisted.ResumeQueue) != 2 {
91 t.Fatalf("active follow-ups persisted as prompt %q queue %d, want empty prompt and queue 2", persisted.ResumePrompt, len(persisted.ResumeQueue))
92 }
93 if firstID, secondID := persisted.ResumeQueue[0].CommandID, persisted.ResumeQueue[1].CommandID; firstID == "" || secondID == "" || firstID == secondID {
94 t.Fatalf("queued follow-up command IDs = %q, %q; want distinct non-empty IDs", firstID, secondID)
95 }
96
97 if err := updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-1"})); err != nil {
98 t.Fatalf("apply first idle event: %v", err)
99 }
100 select {
101 case command := <-commands:
102 var payload TurnStartPayload
103 if command.Type != CommandTurnStart || command.DecodePayload(&payload) != nil || payload.Prompt != firstFollowUp {
104 t.Fatalf("first idle delivery = type %q prompt %q, want turn start %q", command.Type, payload.Prompt, firstFollowUp)
105 }
106 if !payload.NewRun {
107 t.Fatal("first queued follow-up did not request a fresh run budget")
108 }
109 case <-time.After(time.Second):
110 t.Fatal("first active follow-up was not delivered at idle")
111 }
112 if got := a.TurnState(); got != TurnQueued {
113 t.Fatalf("turn state after first queued delivery = %s, want %s", got, TurnQueued)
114 }
115
116 if err := updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
117 "turn_id": "turn-2", "step": float64(2), "lifetime_turns": 2, "current_run_turns": 1,
118 })); err != nil {
119 t.Fatalf("apply second turn started event: %v", err)
120 }
121 if err := updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-2"})); err != nil {
122 t.Fatalf("apply second idle event: %v", err)
123 }
124 select {
125 case command := <-commands:
126 var payload TurnStartPayload
127 if command.Type != CommandTurnStart || command.DecodePayload(&payload) != nil || payload.Prompt != secondFollowUp {
128 t.Fatalf("second idle delivery = type %q prompt %q, want turn start %q", command.Type, payload.Prompt, secondFollowUp)
129 }
130 case <-time.After(time.Second):
131 t.Fatal("second active follow-up was not delivered at the next idle turn")
132 }
133
134 persisted, err = readAgentMeta(filepath.Join(root, "agents", a.ID))
135 if err != nil {
136 t.Fatal(err)
137 }
138 if persisted.ResumePrompt != secondFollowUp || len(persisted.ResumeQueue) != 0 {
139 t.Fatalf("second follow-up persistence = prompt %q queue %d, want prompt %q and empty queue", persisted.ResumePrompt, len(persisted.ResumeQueue), secondFollowUp)
140 }
141 if persisted.ResumePromptID == "" {
142 t.Fatal("promoted follow-up lost its command ID")
143 }
144 if err := f.Stop(a.ID); err != nil {
145 t.Fatal(err)
146 }
147 a.Wait()
148}
149
150func TestResumeWithPromptDeliversFollowUpToIdleWorker(t *testing.T) {
151 if runtime.GOOS == "windows" {
152 t.Skip("subagent inbox transport uses Unix-domain sockets")
153 }
154
155 root := shortSocketDir(t)
156 ready := make(chan struct{}, 1)
157 commands := make(chan Envelope, 2)
158 f := New(Config{
159 Root: root, RepoRoot: root,
160 NewRunner: func(a *Agent) Runner {
161 return RunnerFunc(func(ctx context.Context, _ Sink) error {
162 listener, err := Listen(a.InboxPath)
163 if err != nil {
164 return err
165 }
166 defer listener.Close()
167 a.setProcessState(ProcessAlive)
168 a.setTurnState(TurnIdle, "")
169 ready <- struct{}{}
170 for {
171 select {
172 case <-ctx.Done():
173 return ctx.Err()
174 case line, ok := <-listener.Lines():
175 if !ok {
176 return nil
177 }
178 command, err := ParseCommand(line)
179 if err != nil {
180 return err
181 }
182 if command.Type == CommandAgentShutdown {
183 return nil
184 }
185 commands <- command
186 }
187 }
188 })
189 },
190 })
191 defer f.StopAll()
192
193 a, err := f.Spawn(context.Background(), "review the implementation")
194 if err != nil {
195 t.Fatal(err)
196 }
197 select {
198 case <-ready:
199 case <-time.After(time.Second):
200 t.Fatal("idle worker did not open its inbox")
201 }
202 a.setTurnCounts(4, 2)
203
204 const followUp = "I applied your review. What do you think now?"
205 continued, err := f.ResumeWithPrompt(context.Background(), a.ID, followUp)
206 if err != nil {
207 t.Fatalf("continue idle worker: %v", err)
208 }
209 if continued != a {
210 t.Fatalf("continued agent = %p, want existing worker %p", continued, a)
211 }
212 if got := a.TurnState(); got != TurnQueued {
213 t.Fatalf("turn state after sending follow-up = %s, want %s", got, TurnQueued)
214 }
215 if got := a.CurrentRunTurnsValue(); got != 0 {
216 t.Fatalf("current-run turns after explicit resume = %d, want 0 before turn starts", got)
217 }
218 if got := a.LifetimeTurnsValue(); got != 4 {
219 t.Fatalf("lifetime turns after explicit resume = %d, want 4", got)
220 }
221 if _, err := f.ResumeWithPrompt(context.Background(), a.ID, "duplicate follow-up"); err == nil {
222 t.Fatal("second follow-up succeeded while the first was queued")
223 }
224 select {
225 case command := <-commands:
226 if command.Type != CommandTurnStart {
227 t.Fatalf("command type = %q, want %q", command.Type, CommandTurnStart)
228 }
229 var payload TurnStartPayload
230 if err := command.DecodePayload(&payload); err != nil {
231 t.Fatal(err)
232 }
233 if payload.Prompt != followUp {
234 t.Fatalf("follow-up prompt = %q, want %q", payload.Prompt, followUp)
235 }
236 if !payload.NewRun {
237 t.Fatal("idle follow-up did not request a fresh run budget")
238 }
239 case <-time.After(time.Second):
240 t.Fatal("idle worker did not receive the follow-up")
241 }
242
243 persisted, err := readAgentMeta(filepath.Join(root, "agents", a.ID))
244 if err != nil {
245 t.Fatal(err)
246 }
247 if persisted.ResumePrompt != followUp || persisted.ResumePromptAt.IsZero() {
248 t.Fatalf("queued follow-up was not persisted: prompt %q accepted %s", persisted.ResumePrompt, persisted.ResumePromptAt)
249 }
250 if err := f.Stop(a.ID); err != nil {
251 t.Fatal(err)
252 }
253 a.Wait()
254
255 restarted := make(chan *Agent, 1)
256 f2 := New(Config{
257 Root: root, RepoRoot: root,
258 NewRunner: func(agent *Agent) Runner {
259 return RunnerFunc(func(ctx context.Context, _ Sink) error {
260 restarted <- agent
261 <-ctx.Done()
262 return ctx.Err()
263 })
264 },
265 })
266 defer f2.StopAll()
267 if loaded, errs := f2.Reload(); loaded != 1 || len(errs) != 0 {
268 t.Fatalf("reload = (%d, %v), want (1, no errors)", loaded, errs)
269 }
270 resumed, err := f2.ResumeSession(context.Background(), a.ID)
271 if err != nil {
272 t.Fatal(err)
273 }
274 select {
275 case got := <-restarted:
276 if got != resumed || got.resumePrompt() != followUp {
277 t.Fatalf("unacknowledged follow-up was not replayed: agent %p prompt %q", got, got.resumePrompt())
278 }
279 case <-time.After(time.Second):
280 t.Fatal("restarted worker did not receive the follow-up")
281 }
282}