1package agent
2
3import (
4 "bytes"
5 "encoding/json"
6 "io"
7 "os"
8 "path/filepath"
9 "strings"
10 "testing"
11
12 "github.com/bnema/zut/packages/agent/subagents"
13)
14
15// TestWorkerOutputLimitsUseSupervisorPolicy verifies that the child reads
16// the effective output caps propagated by the supervisor and retains safe
17// defaults for standalone workers.
18func TestWorkerOutputLimitsUseSupervisorPolicy(t *testing.T) {
19 t.Setenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES", "123")
20 t.Setenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES", "7")
21 bytesLimit, linesLimit := workerOutputLimits()
22 if bytesLimit != 123 || linesLimit != 7 {
23 t.Fatalf("worker output limits = (%d, %d), want (123, 7)", bytesLimit, linesLimit)
24 }
25 t.Setenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES", "invalid")
26 t.Setenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES", "0")
27 bytesLimit, linesLimit = workerOutputLimits()
28 if bytesLimit != 500_000 || linesLimit != 5_000 {
29 t.Fatalf("invalid worker output limits = (%d, %d), want defaults", bytesLimit, linesLimit)
30 }
31}
32
33// TestSupervisorEmitterMirrorDormantUntilStdoutBreaks regresses the
34// "everything is doubled after reopening a subagent agent" bug.
35//
36// Symptom: events.jsonl held two copies of every event because the
37// child mirrored each event to disk AND the supervisor parsed the
38// child's stdout and appended each event to disk too. On next zut
39// launch the replay produced two transcript lines per real one.
40//
41// Fix invariant: while stdout writes succeed (i.e. the supervisor is
42// alive on the other end of the pipe), the child's mirror writes
43// NOTHING. Only when a stdout write fails (broken pipe → orphan)
44// does the mirror take over so events still get persisted.
45func TestSupervisorEmitterMirrorDormantUntilStdoutBreaks(t *testing.T) {
46 // Real *os.File for the emitter's stdout-equivalent so the
47 // emitter's write() path (which expects *os.File) actually runs.
48 stdoutPath := filepath.Join(t.TempDir(), "stdout.fifo")
49 stdoutFile, err := os.Create(stdoutPath)
50 if err != nil {
51 t.Fatalf("create stdout file: %v", err)
52 }
53 defer stdoutFile.Close()
54
55 // Mirror writes go to a separate events.jsonl that we can read
56 // at the end to assert how many events the mirror emitted.
57 mirrorPath := filepath.Join(t.TempDir(), "events.jsonl")
58 mirror, err := subagents.OpenEventLog(mirrorPath)
59 if err != nil {
60 t.Fatalf("open mirror: %v", err)
61 }
62 defer mirror.Close()
63
64 em := newSubagentEmitter(stdoutFile, mirror)
65
66 // Healthy stdout: emit three events. Mirror must stay empty.
67 em.emit("turn_start", map[string]any{"step": 1})
68 em.emit("assistant_message", map[string]any{"text": "hi"})
69 em.emit("turn_end", map[string]any{"step": 1})
70
71 got, err := subagents.ReadEventLog(mirrorPath)
72 if err != nil {
73 t.Fatalf("read mirror: %v", err)
74 }
75 if len(got) != 0 {
76 t.Fatalf("mirror wrote %d events while supervisor was alive; want 0 (every event would otherwise double on the next reload)\n%+v",
77 len(got), got)
78 }
79
80 // Simulate supervisor death: close stdout so the next Write
81 // returns EBADF / broken pipe. The emitter must flip into
82 // orphan mode and start writing through the mirror.
83 if err := stdoutFile.Close(); err != nil {
84 t.Fatalf("close stdout: %v", err)
85 }
86
87 em.emit("assistant_message", map[string]any{"text": "after orphan"})
88 em.emit("turn_end", map[string]any{"step": 2})
89
90 got, err = subagents.ReadEventLog(mirrorPath)
91 if err != nil {
92 t.Fatalf("read mirror post-orphan: %v", err)
93 }
94 if len(got) < 2 {
95 t.Fatalf("mirror failed to take over after stdout died: got %d events", len(got))
96 }
97 if got[len(got)-1].Type != "turn_end" {
98 t.Errorf("last mirrored event type = %q; want turn_end", got[len(got)-1].Type)
99 }
100}
101
102// TestSupervisorEmitterLargeResultIsNotDuplicatedOnWire regresses the
103// large-result wire duplication bug.
104func TestSupervisorEmitterLargeResultIsNotDuplicatedOnWire(t *testing.T) {
105 path := filepath.Join(t.TempDir(), "events.jsonl")
106 file, err := os.Create(path)
107 if err != nil {
108 t.Fatal(err)
109 }
110 defer file.Close()
111
112 em := newSubagentEmitter(file, nil)
113 em.setProtocolIdentity("agent-1")
114 em.emit("turn.result", map[string]any{
115 "status": "succeeded",
116 "turn_id": "turn-1",
117 "output": strings.Repeat("x", 500_000),
118 })
119 if err := file.Close(); err != nil {
120 t.Fatal(err)
121 }
122 wire, err := os.ReadFile(path)
123 if err != nil {
124 t.Fatal(err)
125 }
126 if len(wire) >= 900_000 {
127 t.Fatalf("large result appears duplicated on wire: %d bytes", len(wire))
128 }
129 var object map[string]any
130 if err := json.Unmarshal(bytes.TrimSpace(wire), &object); err != nil {
131 t.Fatal(err)
132 }
133 if _, exists := object["output"]; exists {
134 t.Fatal("large output was flattened a second time")
135 }
136 payload, ok := object["payload"].(map[string]any)
137 if !ok || len(payload["output"].(string)) != 500_000 {
138 t.Fatal("canonical payload lost the complete bounded output")
139 }
140 events, err := subagents.ReadEventLog(path)
141 if err != nil || len(events) != 1 {
142 t.Fatalf("supervisor event parser could not recover the large result: events=%d err=%v", len(events), err)
143 }
144 if output, ok := events[0].Data["output"].(string); !ok || len(output) != 500_000 {
145 t.Fatal("supervisor parser lost the large result payload")
146 }
147}
148
149// TestSupervisorEmitterStdoutShapeMatchesSupervisorParser pins the
150// wire-format contract: each emitted event lands on stdout as one
151// JSON object per line with type+time at top level alongside the
152// data fields. The supervisor's runner parses this exact shape.
153func TestSupervisorEmitterStdoutShapeMatchesSupervisorParser(t *testing.T) {
154 // Pipe so we can read what the emitter wrote.
155 r, w, err := os.Pipe()
156 if err != nil {
157 t.Fatalf("pipe: %v", err)
158 }
159 defer r.Close()
160
161 em := newSubagentEmitter(w, nil)
162 em.emit("turn_start", map[string]any{"step": 1})
163 _ = w.Close()
164
165 body, err := io.ReadAll(r)
166 if err != nil {
167 t.Fatalf("read pipe: %v", err)
168 }
169 // One trailing newline => one event line.
170 lines := bytes.Split(bytes.TrimRight(body, "\n"), []byte("\n"))
171 if len(lines) != 1 {
172 t.Fatalf("expected 1 event line, got %d: %q", len(lines), body)
173 }
174 var object map[string]any
175 if err := json.Unmarshal(lines[0], &object); err != nil {
176 t.Fatalf("not valid json: %v\n%s", err, lines[0])
177 }
178 if object["type"] != "turn.started" {
179 t.Errorf("type field missing or wrong: %v", object["type"])
180 }
181 if _, ok := object["timestamp"].(string); !ok {
182 t.Errorf("timestamp field missing: %v", object["timestamp"])
183 }
184 payload, ok := object["payload"].(map[string]any)
185 if !ok || payload["step"] != float64(1) {
186 t.Errorf("payload step field missing or wrong: %v", object["payload"])
187 }
188}