1package subagents
2
3import (
4 "bytes"
5 "encoding/json"
6 "fmt"
7 "math"
8 "os"
9 "path/filepath"
10 "strings"
11 "testing"
12 "time"
13)
14
15// TestEventLogAppendAndRead is the simplest possible invariant:
16// what we write goes back through ReadEventLog exactly, including
17// the payload fields and the timestamp.
18func TestEventLogAppendAndRead(t *testing.T) {
19 path := filepath.Join(t.TempDir(), "events.jsonl")
20 log, err := OpenEventLog(path)
21 if err != nil {
22 t.Fatalf("open: %v", err)
23 }
24 want := []Event{
25 NewEvent("turn_start", map[string]any{"step": 0}),
26 NewEvent("assistant_message", map[string]any{
27 "content": []any{map[string]any{"type": "text", "text": "hi"}},
28 }),
29 NewEvent("turn_end", map[string]any{"stop": "end"}),
30 }
31 for _, ev := range want {
32 if err := log.Append(ev); err != nil {
33 t.Fatalf("append: %v", err)
34 }
35 }
36 _ = log.Close()
37
38 got, err := ReadEventLog(path)
39 if err != nil {
40 t.Fatalf("read: %v", err)
41 }
42 if len(got) != len(want) {
43 t.Fatalf("got %d events, want %d", len(got), len(want))
44 }
45 for i := range want {
46 if got[i].Type != want[i].Type {
47 t.Errorf("event %d type = %q; want %q", i, got[i].Type, want[i].Type)
48 }
49 }
50}
51
52func TestEventLogAppendUsesUTCForZeroTimestamp(t *testing.T) {
53 path := filepath.Join(t.TempDir(), "events.jsonl")
54 log, err := OpenEventLog(path)
55 if err != nil {
56 t.Fatal(err)
57 }
58 if err := log.Append(Event{Type: "zero_time", Version: ProtocolVersion, Data: map[string]any{}}); err != nil {
59 t.Fatal(err)
60 }
61 if err := log.Close(); err != nil {
62 t.Fatal(err)
63 }
64
65 got, err := ReadEventLog(path)
66 if err != nil {
67 t.Fatal(err)
68 }
69 if len(got) != 1 {
70 t.Fatalf("got %d events, want 1", len(got))
71 }
72 if got[0].Time.Location() != time.UTC {
73 t.Fatalf("zero timestamp location = %v, want UTC", got[0].Time.Location())
74 }
75}
76
77// TestReadEventLogIgnoresGarbage documents that one malformed line
78// doesn't break the whole replay. The runner's responsibility is to
79// only write well-formed events, but a partial crash could leave
80// a half-written line; we don't want that to prevent the dashboard
81// from showing the rest.
82func TestReadVersionedEventMergesPayloadFields(t *testing.T) {
83 path := filepath.Join(t.TempDir(), "events.jsonl")
84 wire := `{"version":1,"type":"turn.progress","message_id":"m1","agent_id":"a1","turn_id":"t1","timestamp":"2026-08-04T12:00:00Z","payload":{"text":"still working","future":{"step":2}}}` + "\n"
85 if err := os.WriteFile(path, []byte(wire), 0o600); err != nil {
86 t.Fatal(err)
87 }
88 got, err := ReadEventLog(path)
89 if err != nil {
90 t.Fatal(err)
91 }
92 if len(got) != 1 || got[0].Data["text"] != "still working" {
93 t.Fatalf("versioned payload was not replayed: %+v", got)
94 }
95 if future, ok := got[0].Data["future"].(map[string]any); !ok || future["step"] != float64(2) {
96 t.Fatalf("future payload field was lost: %+v", got[0].Data)
97 }
98}
99
100func TestReadEventLogSkipsOversizedLines(t *testing.T) {
101 path := filepath.Join(t.TempDir(), "events.jsonl")
102 oversized := fmt.Sprintf("%s\n", strings.Repeat("x", maxEventLogLineBytes+100))
103 valid := NewEvent("valid", map[string]any{"ok": true})
104 data, err := json.Marshal(valid)
105 if err != nil {
106 t.Fatal(err)
107 }
108 if err := os.WriteFile(path, append([]byte(oversized), append(data, '\n')...), 0o600); err != nil {
109 t.Fatal(err)
110 }
111 got, err := ReadEventLog(path)
112 if err != nil {
113 t.Fatal(err)
114 }
115 if len(got) != 1 || got[0].Type != "valid" {
116 t.Fatalf("oversized line affected replay: %+v", got)
117 }
118}
119
120func TestEventReadersRetainLatestEventsInOrder(t *testing.T) {
121 total := maxEventLogEvents + 2
122 var input bytes.Buffer
123 for i := 0; i < total; i++ {
124 ev := Event{
125 Time: time.Unix(int64(i), 0).UTC(),
126 Type: "indexed",
127 Version: ProtocolVersion,
128 Data: map[string]any{"index": i},
129 }
130 line, err := json.Marshal(ev)
131 if err != nil {
132 t.Fatal(err)
133 }
134 input.Write(line)
135 input.WriteByte('\n')
136 }
137
138 check := func(name string, got []Event) {
139 t.Helper()
140 if len(got) != maxEventLogEvents {
141 t.Fatalf("%s retained %d events, want %d", name, len(got), maxEventLogEvents)
142 }
143 for i, ev := range got {
144 index, ok := ev.Data["index"].(float64)
145 if !ok || int(index) != i+2 {
146 t.Fatalf("%s event %d index = %v, want %d", name, i, ev.Data["index"], i+2)
147 }
148 }
149 }
150
151 got, err := readEvents(bytes.NewReader(input.Bytes()))
152 if err != nil {
153 t.Fatal(err)
154 }
155 check("readEvents", got)
156
157 got, offset := readFollowerEvents(bytes.NewReader(input.Bytes()), 0)
158 check("readFollowerEvents", got)
159 if offset != int64(input.Len()) {
160 t.Fatalf("follower offset = %d, want %d", offset, input.Len())
161 }
162}
163
164func TestReadEventLogIgnoresGarbage(t *testing.T) {
165 path := filepath.Join(t.TempDir(), "events.jsonl")
166 log, err := OpenEventLog(path)
167 if err != nil {
168 t.Fatalf("open: %v", err)
169 }
170 if err := log.Append(NewEvent("first", nil)); err != nil {
171 t.Fatalf("append: %v", err)
172 }
173 // Sneak a garbage line in by writing directly.
174 if _, err := log.f.WriteString("not json at all\n"); err != nil {
175 t.Fatalf("raw write: %v", err)
176 }
177 if err := log.Append(NewEvent("second", nil)); err != nil {
178 t.Fatalf("append: %v", err)
179 }
180 _ = log.Close()
181
182 got, err := ReadEventLog(path)
183 if err != nil {
184 t.Fatalf("read: %v", err)
185 }
186 if len(got) != 2 || got[0].Type != "first" || got[1].Type != "second" {
187 t.Fatalf("garbage tripped parser: %+v", got)
188 }
189}
190
191// TestReadEventLogDeduplicatesDoubleWrites pins the read-time fix
192// for already-polluted events.jsonl files. The historical bug: the
193// supervisor parsed child stdout and called log.Append(ev), AND the
194// child's mirror appended the same event. Two writes per event
195// landed in the file in quick succession with near-identical
196// timestamps, and the next zut launch replayed everything twice —
197// the user saw every transcript line duplicated.
198//
199// The write-time fix makes the child's mirror dormant when the
200// supervisor is alive, but old files are still polluted. ReadEventLog
201// now folds back-to-back same-type identical-payload events whose
202// timestamps are within ~250ms of each other.
203func TestReadEventLogDeduplicatesDoubleWrites(t *testing.T) {
204 path := filepath.Join(t.TempDir(), "events.jsonl")
205 log, err := OpenEventLog(path)
206 if err != nil {
207 t.Fatalf("open: %v", err)
208 }
209
210 // Two writes per logical event, mimicking the supervisor +
211 // mirror double-write pattern. Timestamps differ by a few ms
212 // like they would in the wild.
213 now := time.Now()
214 writePair := func(typ string, data map[string]any) {
215 e1 := NewEvent(typ, data)
216 e1.Time = now
217 if err := log.Append(e1); err != nil {
218 t.Fatalf("append: %v", err)
219 }
220 e2 := NewEvent(typ, data)
221 e2.Time = now.Add(2 * time.Millisecond)
222 if err := log.Append(e2); err != nil {
223 t.Fatalf("append: %v", err)
224 }
225 now = now.Add(500 * time.Millisecond) // next event well outside the dedup window
226 }
227
228 writePair("assistant_message", map[string]any{
229 "content": []any{map[string]any{"type": "text", "text": "hi"}},
230 })
231 writePair("turn_end", map[string]any{"step": float64(1)})
232 _ = log.Close()
233
234 got, err := ReadEventLog(path)
235 if err != nil {
236 t.Fatalf("read: %v", err)
237 }
238 if len(got) != 2 {
239 t.Fatalf("expected double-writes to collapse to 2 events; got %d", len(got))
240 }
241 if got[0].Type != "assistant_message" || got[1].Type != "turn_end" {
242 t.Errorf("unexpected types after dedup: %q, %q", got[0].Type, got[1].Type)
243 }
244}
245
246// TestReadEventLogKeepsLegitimateAdjacentEvents pins the inverse:
247// two events of the same type with identical payloads but separated
248// by more than the dedup window (250ms) must both survive. The
249// agent legitimately can emit identical adjacent events (think: a
250// retry running the same tool seconds apart).
251func TestSameEventDataDoesNotConflateFailedEncodings(t *testing.T) {
252 payload := math.NaN()
253 if sameEventData(map[string]any{"value": payload}, map[string]any{"value": payload}) {
254 t.Fatal("payloads with non-equal values must not be treated as equal when encoding fails")
255 }
256}
257
258func TestReadEventLogKeepsLegitimateAdjacentEvents(t *testing.T) {
259 path := filepath.Join(t.TempDir(), "events.jsonl")
260 log, err := OpenEventLog(path)
261 if err != nil {
262 t.Fatalf("open: %v", err)
263 }
264 now := time.Now()
265 for i := 0; i < 3; i++ {
266 e := NewEvent("tool_call", map[string]any{"name": "read"})
267 e.Time = now.Add(time.Duration(i) * time.Second)
268 if err := log.Append(e); err != nil {
269 t.Fatalf("append: %v", err)
270 }
271 }
272 _ = log.Close()
273
274 got, err := ReadEventLog(path)
275 if err != nil {
276 t.Fatalf("read: %v", err)
277 }
278 if len(got) != 3 {
279 t.Errorf("dedup collapsed legitimately separated events; got %d, want 3", len(got))
280 }
281}
282
283// TestFollowerEmitsNewEvents follows a live file and ensures events
284// appended after the follower starts make it to the channel.
285func TestFollowerEmitsNewEvents(t *testing.T) {
286 path := filepath.Join(t.TempDir(), "events.jsonl")
287 // Pre-seed one event so the file exists. The follower's
288 // contract is "emit events appended AFTER it starts", so this
289 // first event is not expected on the channel.
290 log, err := OpenEventLog(path)
291 if err != nil {
292 t.Fatalf("open: %v", err)
293 }
294 defer log.Close()
295 if err := log.Append(NewEvent("preexisting", nil)); err != nil {
296 t.Fatalf("append: %v", err)
297 }
298
299 fol := FollowEventLog(path, 10*time.Millisecond)
300 defer fol.Close()
301
302 // Give the follower one tick to read the pre-existing event so
303 // its offset advances past the first line. After that, any new
304 // append should show up.
305 time.Sleep(40 * time.Millisecond)
306 drain(fol.Events())
307
308 for i := 0; i < 3; i++ {
309 if err := log.Append(NewEvent("live", map[string]any{"i": i})); err != nil {
310 t.Fatalf("append %d: %v", i, err)
311 }
312 }
313
314 got := waitFor(t, fol.Events(), 3, time.Second)
315 if len(got) != 3 {
316 t.Fatalf("got %d live events, want 3", len(got))
317 }
318 for i, ev := range got {
319 if ev.Type != "live" {
320 t.Errorf("event %d type = %q; want live", i, ev.Type)
321 }
322 }
323}
324
325func TestFollowerDoesNotSkipPartialFinalEvent(t *testing.T) {
326 seed := NewEvent("preexisting", nil)
327 seedBytes, err := json.Marshal(seed)
328 if err != nil {
329 t.Fatal(err)
330 }
331 seedBytes = append(seedBytes, '\n')
332 partial := NewEvent("partial", map[string]any{"ok": true})
333 partialBytes, err := json.Marshal(partial)
334 if err != nil {
335 t.Fatal(err)
336 }
337 mid := len(partialBytes) / 2
338 incomplete := append(append([]byte(nil), seedBytes...), partialBytes[:mid]...)
339 events, offset := readFollowerEvents(bytes.NewReader(incomplete), 0)
340 if len(events) != 1 || events[0].Type != "preexisting" {
341 t.Fatalf("complete prefix = %+v", events)
342 }
343 if offset != int64(len(seedBytes)) {
344 t.Fatalf("offset = %d, want complete prefix %d", offset, len(seedBytes))
345 }
346
347 complete := append(append([]byte(nil), seedBytes...), append(partialBytes, '\n')...)
348 events, finalOffset := readFollowerEvents(bytes.NewReader(complete[offset:]), offset)
349 if len(events) != 1 || events[0].Type != "partial" {
350 t.Fatalf("completed event = %+v", events)
351 }
352 if finalOffset != int64(len(complete)) {
353 t.Fatalf("final offset = %d, want %d", finalOffset, len(complete))
354 }
355}
356
357func drain(ch <-chan Event) {
358 for {
359 select {
360 case <-ch:
361 default:
362 return
363 }
364 }
365}
366
367func waitFor(t *testing.T, ch <-chan Event, n int, timeout time.Duration) []Event {
368 t.Helper()
369 var got []Event
370 deadline := time.After(timeout)
371 for len(got) < n {
372 select {
373 case ev := <-ch:
374 got = append(got, ev)
375 case <-deadline:
376 return got
377 }
378 }
379 return got
380}