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 iv.TrackSubagentWorker(first, first.Task, false)
290 const followUp = "queued follow-up"
291 cancel := iv.PrepareResumedSubagentWorker(first, followUp, false)
292 if cancel == nil {
293 t.Fatal("resume registration returned nil cleanup")
294 }
295 if first.OnTurnEnd == nil {
296 t.Fatal("worker has no completion watcher")
297 }
298 first.OnTurnEnd(1, "")
299 first.OnTurnEnd(2, "")
300
301 update := waitForQueuedPrompt(t, iv)
302 if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {
303 t.Fatalf("active follow-up was not batched with the original turn: %q", update)
304 }
305 if !strings.Contains(update, "task: "+followUp) {
306 t.Fatalf("active follow-up task missing: %q", update)
307 }
308}
309
310func TestCompletionDeliveryHoldBatchesFastAndLateRegistrations(t *testing.T) {
311 mgr := subagents.New(subagents.Config{
312 Root: t.TempDir(),
313 RepoRoot: t.TempDir(),
314 NewRunner: func(*subagents.Agent) subagents.Runner {
315 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
316 <-ctx.Done()
317 return ctx.Err()
318 })
319 },
320 })
321 first, err := mgr.Spawn(context.Background(), "first task")
322 if err != nil {
323 t.Fatal(err)
324 }
325 second, err := mgr.Spawn(context.Background(), "second task")
326 if err != nil {
327 t.Fatal(err)
328 }
329 t.Cleanup(func() {
330 _ = mgr.Stop(first.ID)
331 _ = mgr.Stop(second.ID)
332 first.Wait()
333 second.Wait()
334 })
335
336 iv := newQueuedAutoSubagentsInteractive()
337 release := iv.beginCompletionDeliveryHold()
338 iv.TrackSubagentWorker(first, first.Task, false)
339 first.OnTurnEnd(1, "")
340 iv.TrackSubagentWorker(second, second.Task, false)
341 second.OnTurnEnd(1, "")
342 release()
343
344 update := waitForQueuedPrompt(t, iv)
345 if !strings.Contains(update, "[auto-subagents update] 2 sub-agent(s) finished:") {
346 t.Fatalf("fast and late registrations were not batched: %q", update)
347 }
348 if !strings.Contains(update, "task: first task") || !strings.Contains(update, "task: second task") {
349 t.Fatalf("batched update omitted one task: %q", update)
350 }
351 iv.mu.Lock()
352 queued := len(iv.queued)
353 iv.mu.Unlock()
354 if queued != 1 {
355 t.Fatalf("queued updates = %d; want exactly one batched update", queued)
356 }
357}
358
359func TestCompleteSupervisorWatchReportsTurnOutcomeOnce(t *testing.T) {
360 started := make(chan struct{})
361 mgr := subagents.New(subagents.Config{
362 Root: t.TempDir(),
363 RepoRoot: t.TempDir(),
364 NewRunner: func(*subagents.Agent) subagents.Runner {
365 return subagents.RunnerFunc(func(ctx context.Context, _ subagents.Sink) error {
366 close(started)
367 <-ctx.Done()
368 return ctx.Err()
369 })
370 },
371 })
372 a, err := mgr.Spawn(context.Background(), "report the date")
373 if err != nil {
374 t.Fatal(err)
375 }
376 <-started
377 t.Cleanup(func() {
378 _ = mgr.Stop(a.ID)
379 a.Wait()
380 })
381
382 iv := newQueuedAutoSubagentsInteractive()
383 iv.TrackSubagentWorker(a, a.Task, false)
384 if a.OnTurnEnd == nil {
385 t.Fatal("worker has no completion watcher")
386 }
387 a.OnTurnEnd(1, "")
388
389 update := waitForQueuedPrompt(t, iv)
390 if !strings.Contains(update, "status: completed") {
391 t.Fatalf("update uses daemon status instead of turn outcome: %q", update)
392 }
393 iv.mu.Lock()
394 queued := len(iv.queued)
395 iv.mu.Unlock()
396 if queued != 1 {
397 t.Fatalf("queued updates = %d; want exactly one", queued)
398 }
399}
400
401type autoSubagentsQueueClient struct {
402 requests chan provider.Request
403}
404
405func (c *autoSubagentsQueueClient) Name() string { return "auto-subagents-queue-test" }
406
407func (c *autoSubagentsQueueClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
408 c.requests <- req
409 out := make(chan provider.Event, 1)
410 go func() {
411 defer close(out)
412 out <- provider.EventDone{
413 Stop: provider.StopEnd,
414 Message: provider.Message{
415 Role: provider.RoleAssistant,
416 Content: []provider.Content{provider.TextBlock{Text: "done"}},
417 },
418 }
419 }()
420 return out, nil
421}
422
423func TestAutoSubagentsUpdateQueuedAtTurnEndStartsFollowUp(t *testing.T) {
424 client := &autoSubagentsQueueClient{requests: make(chan provider.Request, 2)}
425 agent := core.NewAgent(client, "test-model", "", nil)
426 interactive := NewInteractive(InteractiveConfig{Agent: agent})
427 interactive.runCtx = context.Background()
428
429 var once sync.Once
430 agent.OnEvent = func(ev core.AgentEvent) {
431 if _, ok := ev.(core.EvDone); !ok {
432 return
433 }
434 once.Do(func() {
435 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."
436 update := subagents.FormatCompletionUpdate([]subagents.Completion{{Task: "smoke test"}}, instruction)
437 interactive.SubmitOrQueue(update, nil)
438 })
439 }
440
441 interactive.startTurn(context.Background(), "initial")
442 select {
443 case <-client.requests:
444 case <-time.After(2 * time.Second):
445 t.Fatal("initial request did not start")
446 }
447
448 var followUp provider.Request
449 select {
450 case followUp = <-client.requests:
451 case <-time.After(2 * time.Second):
452 t.Fatal("queued auto-subagents update did not start a follow-up request")
453 }
454 foundUpdate := false
455 for _, message := range followUp.Messages {
456 for _, content := range message.Content {
457 if text, ok := content.(provider.TextBlock); ok && strings.Contains(text.Text, "[auto-subagents update]") {
458 foundUpdate = true
459 }
460 }
461 }
462 if !foundUpdate {
463 t.Fatalf("follow-up request did not contain the auto-subagents update: %#v", followUp.Messages)
464 }
465
466 deadline := time.NewTimer(2 * time.Second)
467 defer deadline.Stop()
468 for {
469 interactive.mu.Lock()
470 busy := interactive.busy
471 interactive.mu.Unlock()
472 if !busy && agent.QueuedMessageCount() == 0 {
473 return
474 }
475 select {
476 case <-deadline.C:
477 t.Fatal("interactive turn remained busy or retained a queued update")
478 case <-time.After(time.Millisecond):
479 }
480 }
481}
482
483func newQueuedAutoSubagentsInteractive() *Interactive {
484 return &Interactive{
485 agent: &core.Agent{},
486 busy: true,
487 compacting: true,
488 dirty: make(chan struct{}, 1),
489 }
490}
491
492func waitForQueuedPrompt(t *testing.T, iv *Interactive) string {
493 t.Helper()
494 deadline := time.NewTimer(autoSubagentsTestTimeout)
495 defer deadline.Stop()
496 ticker := time.NewTicker(time.Millisecond)
497 defer ticker.Stop()
498 for {
499 iv.mu.Lock()
500 if len(iv.queued) > 0 {
501 prompt := iv.queued[0]
502 iv.mu.Unlock()
503 return prompt
504 }
505 iv.mu.Unlock()
506 select {
507 case <-deadline.C:
508 t.Fatal("timed out waiting for auto-subagents update")
509 case <-ticker.C:
510 }
511 }
512}