1package modes
2
3import (
4 "context"
5 "errors"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/zut/packages/agent/subagents"
12 "github.com/bnema/zut/packages/core"
13 "github.com/bnema/zut/packages/provider"
14)
15
16const autoSubagentsTestTimeout = 10 * time.Second
17
18func TestAutoSubagentsSystemPromptKeepsProfileManifestWhileTogglingDelegationGuidance(t *testing.T) {
19 const onDemand = "delegate only on an explicit user request"
20 iv := &Interactive{
21 agent: &core.Agent{System: "base system"},
22 cfg: InteractiveConfig{
23 AutoSubagentsSystemAddendum: "auto-subagents guidance",
24 OnDemandSubagentsSystemAddendum: onDemand,
25 SubagentsSystemAddendum: "[subagents_list]\n- reviewer\n[/subagents_list]",
26 Supervisor: &subagents.Supervisor{},
27 },
28 }
29
30 iv.applyAutoSubagentsSystemPrompt(true)
31 if !strings.Contains(iv.agent.System, "[subagents_list]") || !strings.Contains(iv.agent.System, "auto-subagents guidance") {
32 t.Fatalf("enabled system prompt missing subagent blocks: %q", iv.agent.System)
33 }
34 before := iv.agent.System
35 iv.applyAutoSubagentsSystemPrompt(true)
36 if iv.agent.System != before {
37 t.Fatalf("enabling twice changed system prompt: %q -> %q", before, iv.agent.System)
38 }
39
40 iv.applyAutoSubagentsSystemPrompt(false)
41 if !strings.Contains(iv.agent.System, "[subagents_list]") {
42 t.Fatalf("disabled system prompt omitted profile manifest: %q", iv.agent.System)
43 }
44 if strings.Contains(iv.agent.System, "auto-subagents guidance") {
45 t.Fatalf("disabled system prompt retained orchestrator guidance: %q", iv.agent.System)
46 }
47 if !strings.Contains(iv.agent.System, onDemand) {
48 t.Fatalf("disabled system prompt omitted on-demand delegation guidance: %q", iv.agent.System)
49 }
50}
51
52func TestAutoSubagentsTogglePreservesMatchingBasePromptBlocks(t *testing.T) {
53 manifest := "[subagents_list]\n- reviewer\n[/subagents_list]"
54 guidance := "auto-subagents guidance"
55 base := "base system\n\n" + manifest + "\n\n" + guidance
56 iv := &Interactive{
57 agent: &core.Agent{System: base},
58 cfg: InteractiveConfig{
59 AutoSubagentsSystemAddendum: guidance,
60 SubagentsSystemAddendum: manifest,
61 Supervisor: &subagents.Supervisor{},
62 },
63 }
64
65 iv.applyAutoSubagentsSystemPrompt(true)
66 if strings.Count(iv.agent.System, manifest) != 2 || strings.Count(iv.agent.System, guidance) != 2 {
67 t.Fatalf("enabling should append owned duplicate blocks: %q", iv.agent.System)
68 }
69 iv.applyAutoSubagentsSystemPrompt(false)
70 want := base + "\n\n" + manifest
71 if got := strings.TrimSpace(iv.agent.System); got != want {
72 t.Fatalf("disabling changed matching base prompt blocks: %q; want %q", got, want)
73 }
74}
75
76func TestAutoSubagentsSummaryIncludesCompleteFinalResponseAfterLongTask(t *testing.T) {
77 response := "first response line\n" + strings.Repeat("result ", 150) + "final marker"
78 mgr := subagents.New(subagents.Config{
79 Root: t.TempDir(),
80 RepoRoot: t.TempDir(),
81 NewRunner: func(*subagents.Agent) subagents.Runner {
82 return subagents.RunnerFunc(func(_ context.Context, sink subagents.Sink) error {
83 sink.Transcript(response)
84 return nil
85 })
86 },
87 })
88 a, err := mgr.Spawn(context.Background(), strings.Repeat("long task ", 150))
89 if err != nil {
90 t.Fatal(err)
91 }
92
93 iv := newQueuedAutoSubagentsInteractive()
94 iv.TrackSubagentWorker(a, a.Task, false)
95 update := waitForQueuedPrompt(t, iv)
96 if !strings.Contains(update, "final response:\n"+response) {
97 t.Fatalf("update missing complete final response: %q", update)
98 }
99}
100
101func TestTrackSubagentWorkerReportsStartupFailure(t *testing.T) {
102 mgr := subagents.New(subagents.Config{
103 Root: t.TempDir(),
104 RepoRoot: t.TempDir(),
105 NewRunner: func(*subagents.Agent) subagents.Runner {
106 return subagents.RunnerFunc(func(context.Context, subagents.Sink) error {
107 return errors.New("listener startup failed")
108 })
109 },
110 })
111 a, err := mgr.Spawn(context.Background(), "report the date")
112 if err != nil {
113 t.Fatal(err)
114 }
115
116 iv := newQueuedAutoSubagentsInteractive()
117 iv.TrackSubagentWorker(a, a.Task, false)
118
119 update := waitForQueuedPrompt(t, iv)
120 if !strings.Contains(update, "status: failed") {
121 t.Fatalf("update missing failed status: %q", update)
122 }
123 if !strings.Contains(update, "listener startup failed") {
124 t.Fatalf("update missing startup error: %q", update)
125 }
126}
127
128func TestTrackRequiredSubagentWorkerReportsFailureAsynchronously(t *testing.T) {
129 mgr := subagents.New(subagents.Config{
130 Root: t.TempDir(),
131 RepoRoot: t.TempDir(),
132 NewRunner: func(*subagents.Agent) subagents.Runner {
133 return subagents.RunnerFunc(func(context.Context, subagents.Sink) error {
134 return errors.New("required reviewer failed")
135 })
136 },
137 })
138 a, err := mgr.SpawnReq(context.Background(), subagents.SpawnRequest{Task: "review the release", Required: true})
139 if err != nil {
140 t.Fatal(err)
141 }
142
143 iv := newQueuedAutoSubagentsInteractive()
144 iv.TrackSubagentWorker(a, a.Task, true)
145 update := waitForQueuedPrompt(t, iv)
146 if !strings.Contains(update, "status: failed") || !strings.Contains(update, "required reviewer failed") {
147 t.Fatalf("required worker failure was not delivered: %q", update)
148 }
149 if !a.Snapshot().Requirement.Unmet() {
150 t.Fatal("delivered failure did not remain recoverable")
151 }
152}
153
154func TestTrackStoppedSubagentWorkerDeliversTerminalUpdate(t *testing.T) {
155 started := make(chan struct{})
156 mgr := subagents.New(subagents.Config{
157 Root: t.TempDir(),
158 RepoRoot: t.TempDir(),
159 NewRunner: func(*subagents.Agent) subagents.Runner {
160 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
161 close(started)
162 <-ctx.Done()
163 return ctx.Err()
164 })
165 },
166 })
167 a, err := mgr.Spawn(context.Background(), "investigate the stuck worker")
168 if err != nil {
169 t.Fatal(err)
170 }
171 <-started
172 t.Cleanup(func() {
173 _ = mgr.Stop(a.ID)
174 a.Wait()
175 })
176
177 iv := newQueuedAutoSubagentsInteractive()
178 iv.TrackStoppedSubagentWorker(a)
179 if err := mgr.Stop(a.ID); err != nil {
180 t.Fatal(err)
181 }
182
183 update := waitForQueuedPrompt(t, iv)
184 if !strings.Contains(update, "task: investigate the stuck worker") {
185 t.Fatalf("stop update missing task: %q", update)
186 }
187 if !strings.Contains(update, "status: cancelled") {
188 t.Fatalf("stop update missing cancellation status: %q", update)
189 }
190 if !strings.Contains(update, "subagent stopped by request") || strings.Contains(update, "context canceled") {
191 t.Fatalf("stop update missing attributed cancellation: %q", update)
192 }
193}
194
195func TestTrackDeadlineFailureUsesAttributedResultError(t *testing.T) {
196 mgr := subagents.New(subagents.Config{
197 Root: t.TempDir(),
198 RepoRoot: t.TempDir(),
199 NewRunner: func(*subagents.Agent) subagents.Runner {
200 return subagents.RunnerFunc(func(context.Context, subagents.Sink) error {
201 return context.DeadlineExceeded
202 })
203 },
204 })
205 a, err := mgr.Spawn(context.Background(), "run until deadline")
206 if err != nil {
207 t.Fatal(err)
208 }
209
210 iv := newQueuedAutoSubagentsInteractive()
211 iv.TrackSubagentWorker(a, a.Task, false)
212 update := waitForQueuedPrompt(t, iv)
213 if !strings.Contains(update, "status: failed") {
214 t.Fatalf("deadline update changed failed status: %q", update)
215 }
216 want := subagents.ShutdownResultError(subagents.ShutdownOriginDeadline).Message
217 if !strings.Contains(update, want) {
218 t.Fatalf("deadline update missing attributed result error: %q", update)
219 }
220}
221
222func TestTrackResumedSubagentWorkerDeliversFollowUp(t *testing.T) {
223 started := make(chan struct{})
224 mgr := subagents.New(subagents.Config{
225 Root: t.TempDir(),
226 RepoRoot: t.TempDir(),
227 NewRunner: func(*subagents.Agent) subagents.Runner {
228 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
229 close(started)
230 <-ctx.Done()
231 return ctx.Err()
232 })
233 },
234 })
235 a, err := mgr.Spawn(context.Background(), "review the implementation")
236 if err != nil {
237 t.Fatal(err)
238 }
239 <-started
240 t.Cleanup(func() {
241 _ = mgr.Stop(a.ID)
242 a.Wait()
243 })
244
245 iv := newQueuedAutoSubagentsInteractive()
246 const followUp = "I applied your review. What do you think now?"
247 iv.TrackResumedSubagentWorker(a, followUp, false)
248 if a.OnTurnEnd == nil {
249 t.Fatal("resumed worker has no completion watcher")
250 }
251 a.OnTurnEnd(1, "")
252
253 update := waitForQueuedPrompt(t, iv)
254 if !strings.Contains(update, "task: "+followUp) {
255 t.Fatalf("follow-up update uses the original task instead of the new prompt: %q", update)
256 }
257 if !strings.Contains(update, "status: completed") {
258 t.Fatalf("follow-up update missing completed status: %q", update)
259 }
260}
261
262func TestTrackResumedSubagentWorkerQueuesActiveFollowUpOnSameWorker(t *testing.T) {
263 started := make(chan struct{})
264 mgr := subagents.New(subagents.Config{
265 Root: t.TempDir(),
266 RepoRoot: t.TempDir(),
267 NewRunner: func(*subagents.Agent) subagents.Runner {
268 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
269 select {
270 case started <- struct{}{}:
271 default:
272 }
273 <-ctx.Done()
274 return ctx.Err()
275 })
276 },
277 })
278 first, err := mgr.Spawn(context.Background(), "original task")
279 if err != nil {
280 t.Fatal(err)
281 }
282 <-started
283 t.Cleanup(func() {
284 _ = mgr.Stop(first.ID)
285 first.Wait()
286 })
287
288 iv := newQueuedAutoSubagentsInteractive()
289 release := iv.beginCompletionDeliveryHold()
290 iv.TrackSubagentWorker(first, first.Task, false)
291 const followUp = "queued follow-up"
292 cancel := iv.PrepareResumedSubagentWorker(first, followUp, false)
293 if cancel == nil {
294 t.Fatal("resume registration returned nil cleanup")
295 }
296 if first.OnTurnEnd == nil {
297 t.Fatal("worker has no completion watcher")
298 }
299 first.OnTurnEnd(1, "")
300 first.OnTurnEnd(2, "")
301 release()
302
303 update := waitForQueuedPrompt(t, iv)
304 if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {
305 t.Fatalf("active follow-up was not batched with the original turn: %q", update)
306 }
307 if !strings.Contains(update, "task: "+followUp) {
308 t.Fatalf("active follow-up task missing: %q", update)
309 }
310}
311
312func TestCompletionDeliveryHoldBatchesFastAndLateRegistrations(t *testing.T) {
313 mgr := subagents.New(subagents.Config{
314 Root: t.TempDir(),
315 RepoRoot: t.TempDir(),
316 NewRunner: func(*subagents.Agent) subagents.Runner {
317 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
318 <-ctx.Done()
319 return ctx.Err()
320 })
321 },
322 })
323 first, err := mgr.Spawn(context.Background(), "first task")
324 if err != nil {
325 t.Fatal(err)
326 }
327 second, err := mgr.Spawn(context.Background(), "second task")
328 if err != nil {
329 t.Fatal(err)
330 }
331 t.Cleanup(func() {
332 _ = mgr.Stop(first.ID)
333 _ = mgr.Stop(second.ID)
334 first.Wait()
335 second.Wait()
336 })
337
338 iv := newQueuedAutoSubagentsInteractive()
339 release := iv.beginCompletionDeliveryHold()
340 iv.TrackSubagentWorker(first, first.Task, false)
341 first.OnTurnEnd(1, "")
342 iv.TrackSubagentWorker(second, second.Task, false)
343 second.OnTurnEnd(1, "")
344 release()
345
346 update := waitForQueuedPrompt(t, iv)
347 if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {
348 t.Fatalf("fast and late registrations were not batched: %q", update)
349 }
350 if !strings.Contains(update, "task: first task") || !strings.Contains(update, "task: second task") {
351 t.Fatalf("batched update omitted one task: %q", update)
352 }
353 iv.mu.Lock()
354 queued := len(iv.queued)
355 iv.mu.Unlock()
356 if queued != 1 {
357 t.Fatalf("queued updates = %d; want exactly one batched update", queued)
358 }
359}
360
361func TestCompleteSupervisorWatchReportsTurnOutcomeOnce(t *testing.T) {
362 started := make(chan struct{})
363 mgr := subagents.New(subagents.Config{
364 Root: t.TempDir(),
365 RepoRoot: t.TempDir(),
366 NewRunner: func(*subagents.Agent) subagents.Runner {
367 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
368 close(started)
369 <-ctx.Done()
370 return ctx.Err()
371 })
372 },
373 })
374 a, err := mgr.Spawn(context.Background(), "report the date")
375 if err != nil {
376 t.Fatal(err)
377 }
378 <-started
379 t.Cleanup(func() {
380 _ = mgr.Stop(a.ID)
381 a.Wait()
382 })
383
384 iv := newQueuedAutoSubagentsInteractive()
385 iv.TrackSubagentWorker(a, a.Task, false)
386 if a.OnTurnEnd == nil {
387 t.Fatal("worker has no completion watcher")
388 }
389 a.OnTurnEnd(1, "")
390
391 update := waitForQueuedPrompt(t, iv)
392 if !strings.Contains(update, "status: completed") {
393 t.Fatalf("update uses daemon status instead of turn outcome: %q", update)
394 }
395 iv.mu.Lock()
396 queued := len(iv.queued)
397 iv.mu.Unlock()
398 if queued != 1 {
399 t.Fatalf("queued updates = %d; want exactly one", queued)
400 }
401}
402
403type autoSubagentsQueueClient struct {
404 requests chan provider.Request
405}
406
407func (c *autoSubagentsQueueClient) Name() string { return "auto-subagents-queue-test" }
408
409func (c *autoSubagentsQueueClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
410 c.requests <- req
411 out := make(chan provider.Event, 1)
412 go func() {
413 defer close(out)
414 out <- provider.EventDone{
415 Stop: provider.StopEnd,
416 Message: provider.Message{
417 Role: provider.RoleAssistant,
418 Content: []provider.Content{provider.TextBlock{Text: "done"}},
419 },
420 }
421 }()
422 return out, nil
423}
424
425func TestAutoSubagentsUpdateQueuedAtTurnEndStartsFollowUp(t *testing.T) {
426 client := &autoSubagentsQueueClient{requests: make(chan provider.Request, 2)}
427 agent := core.NewAgent(client, "test-model", "", nil)
428 interactive := NewInteractive(InteractiveConfig{Agent: agent})
429 interactive.runCtx = context.Background()
430
431 var once sync.Once
432 agent.OnEvent = func(ev core.AgentEvent) {
433 if _, ok := ev.(core.EvDone); !ok {
434 return
435 }
436 once.Do(func() {
437 instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks."
438 update := subagents.FormatCompletionUpdate([]subagents.Completion{{Task: "smoke test"}}, instruction)
439 interactive.SubmitOrQueue(update, nil)
440 })
441 }
442
443 interactive.startTurn(context.Background(), "initial")
444 select {
445 case <-client.requests:
446 case <-time.After(2 * time.Second):
447 t.Fatal("initial request did not start")
448 }
449
450 var followUp provider.Request
451 select {
452 case followUp = <-client.requests:
453 case <-time.After(2 * time.Second):
454 t.Fatal("queued auto-subagents update did not start a follow-up request")
455 }
456 foundUpdate := false
457 for _, message := range followUp.Messages {
458 for _, content := range message.Content {
459 if text, ok := content.(provider.TextBlock); ok && strings.Contains(text.Text, "[auto-subagents update]") {
460 foundUpdate = true
461 }
462 }
463 }
464 if !foundUpdate {
465 t.Fatalf("follow-up request did not contain the auto-subagents update: %#v", followUp.Messages)
466 }
467
468 deadline := time.NewTimer(2 * time.Second)
469 defer deadline.Stop()
470 for {
471 interactive.mu.Lock()
472 busy := interactive.busy
473 interactive.mu.Unlock()
474 if !busy && agent.QueuedMessageCount() == 0 {
475 return
476 }
477 select {
478 case <-deadline.C:
479 t.Fatal("interactive turn remained busy or retained a queued update")
480 case <-time.After(time.Millisecond):
481 }
482 }
483}
484
485func newQueuedAutoSubagentsInteractive() *Interactive {
486 return &Interactive{
487 agent: &core.Agent{},
488 busy: true,
489 compacting: true,
490 dirty: make(chan struct{}, 1),
491 }
492}
493
494func waitForQueuedPrompt(t *testing.T, iv *Interactive) string {
495 t.Helper()
496 deadline := time.NewTimer(autoSubagentsTestTimeout)
497 defer deadline.Stop()
498 ticker := time.NewTicker(time.Millisecond)
499 defer ticker.Stop()
500 for {
501 iv.mu.Lock()
502 if len(iv.queued) > 0 {
503 prompt := iv.queued[0]
504 iv.mu.Unlock()
505 return prompt
506 }
507 iv.mu.Unlock()
508 select {
509 case <-deadline.C:
510 t.Fatal("timed out waiting for auto-subagents update")
511 case <-ticker.C:
512 }
513 }
514}