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