1package agent
2
3import (
4 "bytes"
5 "encoding/json"
6 "fmt"
7 "io"
8 "net/http"
9 "net/http/httptest"
10 "os"
11 "os/exec"
12 "path/filepath"
13 "runtime"
14 "strings"
15 "sync"
16 "syscall"
17 "testing"
18 "time"
19)
20
21var (
22 orchestrationBuildOnce sync.Once
23 orchestrationBuildDir string
24 orchestrationBuildBinary string
25 orchestrationBuildOutput []byte
26 orchestrationBuildErr error
27)
28
29func TestMain(m *testing.M) {
30 code := m.Run()
31 if orchestrationBuildDir != "" {
32 _ = os.RemoveAll(orchestrationBuildDir)
33 }
34 os.Exit(code)
35}
36
37func TestOrchestratedCLITwoDelegationWavesAcrossHeadlessModes(t *testing.T) {
38 if runtime.GOOS == "windows" {
39 t.Skip("signal and subprocess semantics differ on Windows")
40 }
41 if testing.Short() {
42 t.Skip("skip CLI orchestration end-to-end test in -short mode")
43 }
44 binary := buildOrchestrationCLI(t)
45 for _, mode := range []string{"print", "stream", "json"} {
46 t.Run(mode, func(t *testing.T) {
47 server := newTwoWaveProvider(t)
48 defer server.Close()
49 home := t.TempDir()
50 args := []string{"--orchestrate", "--provider", "openai", "--model", "gpt-5", "--base-url", server.URL(), "--api-key", "test-key"}
51 switch mode {
52 case "print":
53 args = append(args, "--print")
54 case "stream":
55 args = append(args, "--stream")
56 case "json":
57 args = append(args, "--json")
58 }
59 args = append(args, "delegate two waves")
60 cmd := exec.Command(binary, args...)
61 cmd.Env = orchestrationCLIEnv(binary, home)
62 var stdout, stderr bytes.Buffer
63 cmd.Stdout = &stdout
64 cmd.Stderr = &stderr
65 err := cmd.Run()
66 output := append(append([]byte(nil), stdout.Bytes()...), stderr.Bytes()...)
67 if err != nil {
68 t.Fatalf("zut %s: %v\n%s", mode, err, output)
69 }
70 if !bytes.Contains(output, []byte("final synthesis")) {
71 t.Fatalf("%s output omitted final synthesis: %s", mode, output)
72 }
73 if server.ParentCalls() != 5 || server.WorkerCalls() < 2 {
74 t.Fatalf("%s calls = parent %d, worker %d; want two delegation waves and two workers", mode, server.ParentCalls(), server.WorkerCalls())
75 }
76 if mode == "json" {
77 events := assertJSONL(t, stdout.Bytes())
78 assertJSONOrchestrationWaves(t, events)
79 }
80 if mode == "stream" && !bytes.Contains(stdout.Bytes(), []byte("wave one synthesis")) {
81 t.Fatalf("stream output omitted first parent turn: %s", output)
82 }
83 if mode == "stream" && bytes.Contains(stdout.Bytes(), []byte("[auto-subagents update]")) {
84 t.Fatalf("stream stdout leaked internal completion prompt: %s", stdout.Bytes())
85 }
86 if mode == "print" && bytes.Contains(stdout.Bytes(), []byte("wave one synthesis")) {
87 t.Fatalf("print output leaked intermediate parent turn: %s", output)
88 }
89 })
90 }
91}
92
93func TestOrchestratedCLICancellation(t *testing.T) {
94 if runtime.GOOS == "windows" {
95 t.Skip("signal and subprocess semantics differ on Windows")
96 }
97 if testing.Short() {
98 t.Skip("skip CLI orchestration end-to-end test in -short mode")
99 }
100 binary := buildOrchestrationCLI(t)
101 server, workerReady, releaseWorker := newCancellationProvider()
102 defer server.Close()
103 defer close(releaseWorker)
104
105 cmd := exec.Command(binary, "--orchestrate", "--json", "--provider", "openai", "--model", "gpt-5", "--base-url", server.URL, "--api-key", "test-key", "cancel")
106 cmd.Env = orchestrationCLIEnv(binary, t.TempDir())
107 var stdout, stderr bytes.Buffer
108 cmd.Stdout = &stdout
109 cmd.Stderr = &stderr
110 if err := cmd.Start(); err != nil {
111 t.Fatal(err)
112 }
113 waitForWorker := time.NewTimer(10 * time.Second)
114 select {
115 case <-workerReady:
116 if !waitForWorker.Stop() {
117 <-waitForWorker.C
118 }
119 case <-waitForWorker.C:
120 _ = cmd.Process.Kill()
121 _ = cmd.Wait()
122 t.Fatal("worker did not reach the blocking provider request")
123 }
124 if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
125 t.Fatalf("send cancellation: %v", err)
126 }
127 wait := make(chan error, 1)
128 go func() { wait <- cmd.Wait() }()
129 select {
130 case err := <-wait:
131 if err == nil {
132 t.Fatal("canceled orchestration unexpectedly succeeded")
133 }
134 case <-time.After(10 * time.Second):
135 _ = cmd.Process.Kill()
136 t.Fatal("canceled orchestration did not exit")
137 }
138
139 events := assertJSONL(t, stdout.Bytes())
140 errorCount := 0
141 for _, event := range events {
142 if event["type"] == "error" {
143 errorCount++
144 }
145 }
146 if errorCount != 1 {
147 t.Fatalf("cancellation error objects = %d, want one; stdout=%s stderr=%s", errorCount, stdout.Bytes(), stderr.Bytes())
148 }
149}
150
151func TestOrchestratedCLIParentFailure(t *testing.T) {
152 if runtime.GOOS == "windows" {
153 t.Skip("signal and subprocess semantics differ on Windows")
154 }
155 if testing.Short() {
156 t.Skip("skip CLI orchestration end-to-end test in -short mode")
157 }
158 binary := buildOrchestrationCLI(t)
159
160 t.Run("failure-jsonl", func(t *testing.T) {
161 server := newFailingProvider()
162 defer server.Close()
163 cmd := exec.Command(binary, "--orchestrate", "--json", "--provider", "openai", "--model", "gpt-5", "--base-url", server.URL, "--api-key", "test-key", "fail")
164 cmd.Env = orchestrationCLIEnv(binary, t.TempDir())
165 var stdout, stderr bytes.Buffer
166 cmd.Stdout = &stdout
167 cmd.Stderr = &stderr
168 err := cmd.Run()
169 if err == nil {
170 t.Fatal("failing orchestration unexpectedly succeeded")
171 }
172 events := assertJSONL(t, stdout.Bytes())
173 errorCount := 0
174 for _, event := range events {
175 if event["type"] == "error" {
176 errorCount++
177 }
178 }
179 if errorCount != 1 {
180 output := append(append([]byte(nil), stdout.Bytes()...), stderr.Bytes()...)
181 t.Fatalf("error objects = %d, want one: %s", errorCount, output)
182 }
183 })
184}
185
186type twoWaveProvider struct {
187 t *testing.T
188 server *httptest.Server
189 mu sync.Mutex
190 parentCalls int
191 workerCalls int
192}
193
194func newTwoWaveProvider(t *testing.T) *twoWaveProvider {
195 p := &twoWaveProvider{t: t}
196 p.server = httptest.NewServer(http.HandlerFunc(p.serveHTTP))
197 return p
198}
199
200func (p *twoWaveProvider) URL() string { return p.server.URL }
201func (p *twoWaveProvider) Close() { p.server.Close() }
202
203func (p *twoWaveProvider) ParentCalls() int {
204 p.mu.Lock()
205 defer p.mu.Unlock()
206 return p.parentCalls
207}
208
209func (p *twoWaveProvider) WorkerCalls() int {
210 p.mu.Lock()
211 defer p.mu.Unlock()
212 return p.workerCalls
213}
214
215func (p *twoWaveProvider) serveHTTP(w http.ResponseWriter, r *http.Request) {
216 body, err := io.ReadAll(r.Body)
217 if err != nil {
218 p.t.Errorf("read provider request: %v", err)
219 return
220 }
221 p.mu.Lock()
222 isParent := requestIsParent(body)
223 if isParent {
224 p.parentCalls++
225 call := p.parentCalls
226 p.mu.Unlock()
227 switch call {
228 case 1:
229 writeToolCall(w, "wave-one", "wave one")
230 case 2:
231 writeText(w, "wave one synthesis")
232 case 3:
233 writeToolCall(w, "wave-two", "wave two")
234 default:
235 writeText(w, "final synthesis")
236 }
237 return
238 }
239 p.workerCalls++
240 worker := p.workerCalls
241 p.mu.Unlock()
242 writeText(w, fmt.Sprintf("worker result %d", worker))
243}
244
245func newCancellationProvider() (*httptest.Server, <-chan struct{}, chan<- struct{}) {
246 workerReady := make(chan struct{})
247 releaseWorker := make(chan struct{})
248 var once sync.Once
249 var mu sync.Mutex
250 requests := 0
251 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
252 mu.Lock()
253 requests++
254 parent := requests == 1
255 mu.Unlock()
256 if parent {
257 writeToolCall(w, "cancel-worker", "long running worker")
258 return
259 }
260 once.Do(func() { close(workerReady) })
261 select {
262 case <-r.Context().Done():
263 case <-releaseWorker:
264 }
265 }))
266 return server, workerReady, releaseWorker
267}
268
269func newFailingProvider() *httptest.Server {
270 var mu sync.Mutex
271 parentCalls := 0
272 return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
273 body, _ := io.ReadAll(r.Body)
274 if requestIsParent(body) {
275 mu.Lock()
276 parentCalls++
277 call := parentCalls
278 mu.Unlock()
279 if call == 1 {
280 writeToolCall(w, "failure-worker", "failure wave")
281 return
282 }
283 w.WriteHeader(http.StatusBadGateway)
284 _, _ = io.WriteString(w, `{"error":{"message":"synthetic provider failure","type":"server_error"}}`)
285 return
286 }
287 writeText(w, "worker completed before parent failure")
288 }))
289}
290
291func requestIsParent(body []byte) bool {
292 var request struct {
293 Messages []struct {
294 Role string `json:"role"`
295 Content json.RawMessage `json:"content"`
296 } `json:"messages"`
297 }
298 if json.Unmarshal(body, &request) != nil {
299 return false
300 }
301 for _, message := range request.Messages {
302 if message.Role != "user" {
303 continue
304 }
305 var text string
306 if json.Unmarshal(message.Content, &text) == nil {
307 if text == "delegate two waves" || text == "fail" || text == "cancel" {
308 return true
309 }
310 }
311 if bytes.Contains(message.Content, []byte(`delegate two waves`)) || bytes.Contains(message.Content, []byte(`"text":"fail"`)) || bytes.Contains(message.Content, []byte(`"text":"cancel"`)) {
312 return true
313 }
314 }
315 return false
316}
317
318func writeToolCall(w http.ResponseWriter, id, task string) {
319 w.Header().Set("content-type", "text/event-stream")
320 arguments, _ := json.Marshal(map[string]any{"action": "spawn", "task": task})
321 chunk := map[string]any{
322 "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"tool_calls": []any{map[string]any{"index": 0, "id": id, "type": "function", "function": map[string]string{"name": "subagent", "arguments": string(arguments)}}}}, "finish_reason": "tool_calls"}},
323 "usage": map[string]int{"prompt_tokens": 8, "completion_tokens": 2},
324 }
325 data, _ := json.Marshal(chunk)
326 _, _ = fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", data)
327}
328
329func writeText(w http.ResponseWriter, text string) {
330 w.Header().Set("content-type", "text/event-stream")
331 chunk := map[string]any{
332 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": text}, "finish_reason": "stop"}},
333 "usage": map[string]int{"prompt_tokens": 8, "completion_tokens": 2},
334 }
335 data, _ := json.Marshal(chunk)
336 _, _ = fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", data)
337}
338
339func assertJSONL(t *testing.T, output []byte) []map[string]any {
340 t.Helper()
341 var events []map[string]any
342 for _, line := range strings.Split(strings.TrimSpace(string(output)), "\n") {
343 if strings.TrimSpace(line) == "" || strings.HasPrefix(line, "zut:") {
344 continue
345 }
346 var event map[string]any
347 if err := json.Unmarshal([]byte(line), &event); err != nil {
348 t.Fatalf("invalid JSONL line %q: %v", line, err)
349 }
350 events = append(events, event)
351 }
352 return events
353}
354
355func assertJSONOrchestrationWaves(t *testing.T, events []map[string]any) {
356 t.Helper()
357 var userMessages []string
358 seenTypes := make(map[string]bool)
359 for _, event := range events {
360 if typ, ok := event["type"].(string); ok {
361 seenTypes[typ] = true
362 }
363 if event["type"] != "user_message" {
364 continue
365 }
366 content, ok := event["content"].([]any)
367 if !ok {
368 t.Fatalf("user_message content = %T, want JSON array", event["content"])
369 }
370 var text strings.Builder
371 for _, block := range content {
372 object, ok := block.(map[string]any)
373 if !ok || object["type"] != "text" {
374 continue
375 }
376 if value, ok := object["text"].(string); ok {
377 text.WriteString(value)
378 }
379 }
380 userMessages = append(userMessages, text.String())
381 }
382 if len(userMessages) < 3 {
383 t.Fatalf("user messages = %d, want initial prompt and two parent completion turns", len(userMessages))
384 }
385 completionMessages := 0
386 for _, message := range userMessages {
387 if strings.HasPrefix(message, "[auto-subagents update]") {
388 completionMessages++
389 }
390 }
391 if completionMessages != 2 {
392 t.Fatalf("completion user messages = %d, want two: %#v", completionMessages, userMessages)
393 }
394 for _, typ := range []string{"assistant_message", "tool_call", "tool_execution_started", "tool_result", "usage", "turn_end"} {
395 if !seenTypes[typ] {
396 t.Fatalf("orchestrated JSONL omitted %s event; seen types: %v", typ, seenTypes)
397 }
398 }
399 for _, event := range events {
400 if event["type"] == "error" {
401 t.Fatalf("successful two-wave orchestration emitted an error event: %#v", event)
402 }
403 }
404}
405
406func orchestrationCLIEnv(binary, home string) []string {
407 // Restrict PATH so spawned workers resolve the same freshly built test
408 // binary instead of any installed zut executable on the host.
409 return append(os.Environ(), "ZUT_HOME="+home, "PATH="+filepath.Dir(binary))
410}
411
412func buildOrchestrationCLI(t *testing.T) string {
413 t.Helper()
414 orchestrationBuildOnce.Do(func() {
415 orchestrationBuildDir, orchestrationBuildErr = os.MkdirTemp("", "zut-orchestration-cli-")
416 if orchestrationBuildErr != nil {
417 return
418 }
419 orchestrationBuildBinary = filepath.Join(orchestrationBuildDir, "zut")
420 root := filepath.Join("..", "..")
421 cmd := exec.Command("go", "build", "-o", orchestrationBuildBinary, "./cmd/zut")
422 cmd.Dir = root
423 cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
424 orchestrationBuildOutput, orchestrationBuildErr = cmd.CombinedOutput()
425 })
426 if orchestrationBuildErr != nil {
427 t.Fatalf("build zut: %v\n%s", orchestrationBuildErr, orchestrationBuildOutput)
428 }
429 return orchestrationBuildBinary
430}