1package swarm
2
3import (
4 "path/filepath"
5 "testing"
6 "time"
7)
8
9// TestEventLogAppendAndRead is the simplest possible invariant:
10// what we write goes back through ReadEventLog exactly, including
11// the flat field shape and the timestamp.
12func TestEventLogAppendAndRead(t *testing.T) {
13 path := filepath.Join(t.TempDir(), "events.jsonl")
14 log, err := OpenEventLog(path)
15 if err != nil {
16 t.Fatalf("open: %v", err)
17 }
18 want := []Event{
19 NewEvent("turn_start", map[string]any{"step": 0}),
20 NewEvent("assistant_message", map[string]any{
21 "content": []any{map[string]any{"type": "text", "text": "hi"}},
22 }),
23 NewEvent("turn_end", map[string]any{"stop": "end"}),
24 }
25 for _, ev := range want {
26 if err := log.Append(ev); err != nil {
27 t.Fatalf("append: %v", err)
28 }
29 }
30 _ = log.Close()
31
32 got, err := ReadEventLog(path)
33 if err != nil {
34 t.Fatalf("read: %v", err)
35 }
36 if len(got) != len(want) {
37 t.Fatalf("got %d events, want %d", len(got), len(want))
38 }
39 for i := range want {
40 if got[i].Type != want[i].Type {
41 t.Errorf("event %d type = %q; want %q", i, got[i].Type, want[i].Type)
42 }
43 }
44}
45
46// TestReadEventLogIgnoresGarbage documents that one malformed line
47// doesn't break the whole replay. The runner's responsibility is to
48// only write well-formed events, but a partial crash could leave
49// a half-written line; we don't want that to prevent the dashboard
50// from showing the rest.
51func TestReadEventLogIgnoresGarbage(t *testing.T) {
52 path := filepath.Join(t.TempDir(), "events.jsonl")
53 log, err := OpenEventLog(path)
54 if err != nil {
55 t.Fatalf("open: %v", err)
56 }
57 if err := log.Append(NewEvent("first", nil)); err != nil {
58 t.Fatalf("append: %v", err)
59 }
60 // Sneak a garbage line in by writing directly.
61 if _, err := log.f.WriteString("not json at all\n"); err != nil {
62 t.Fatalf("raw write: %v", err)
63 }
64 if err := log.Append(NewEvent("second", nil)); err != nil {
65 t.Fatalf("append: %v", err)
66 }
67 _ = log.Close()
68
69 got, err := ReadEventLog(path)
70 if err != nil {
71 t.Fatalf("read: %v", err)
72 }
73 if len(got) != 2 || got[0].Type != "first" || got[1].Type != "second" {
74 t.Fatalf("garbage tripped parser: %+v", got)
75 }
76}
77
78// TestReadEventLogDeduplicatesDoubleWrites pins the read-time fix
79// for already-polluted events.jsonl files. The historical bug: the
80// supervisor parsed child stdout and called log.Append(ev), AND the
81// child's mirror appended the same event. Two writes per event
82// landed in the file in quick succession with near-identical
83// timestamps, and the next zot launch replayed everything twice —
84// the user saw every transcript line duplicated.
85//
86// The write-time fix makes the child's mirror dormant when the
87// supervisor is alive, but old files are still polluted. ReadEventLog
88// now folds back-to-back same-type identical-payload events whose
89// timestamps are within ~250ms of each other.
90func TestReadEventLogDeduplicatesDoubleWrites(t *testing.T) {
91 path := filepath.Join(t.TempDir(), "events.jsonl")
92 log, err := OpenEventLog(path)
93 if err != nil {
94 t.Fatalf("open: %v", err)
95 }
96
97 // Two writes per logical event, mimicking the supervisor +
98 // mirror double-write pattern. Timestamps differ by a few ms
99 // like they would in the wild.
100 now := time.Now()
101 writePair := func(typ string, data map[string]any) {
102 e1 := NewEvent(typ, data)
103 e1.Time = now
104 if err := log.Append(e1); err != nil {
105 t.Fatalf("append: %v", err)
106 }
107 e2 := NewEvent(typ, data)
108 e2.Time = now.Add(2 * time.Millisecond)
109 if err := log.Append(e2); err != nil {
110 t.Fatalf("append: %v", err)
111 }
112 now = now.Add(500 * time.Millisecond) // next event well outside the dedup window
113 }
114
115 writePair("assistant_message", map[string]any{
116 "content": []any{map[string]any{"type": "text", "text": "hi"}},
117 })
118 writePair("turn_end", map[string]any{"step": float64(1)})
119 _ = log.Close()
120
121 got, err := ReadEventLog(path)
122 if err != nil {
123 t.Fatalf("read: %v", err)
124 }
125 if len(got) != 2 {
126 t.Fatalf("expected double-writes to collapse to 2 events; got %d", len(got))
127 }
128 if got[0].Type != "assistant_message" || got[1].Type != "turn_end" {
129 t.Errorf("unexpected types after dedup: %q, %q", got[0].Type, got[1].Type)
130 }
131}
132
133// TestReadEventLogKeepsLegitimateAdjacentEvents pins the inverse:
134// two events of the same type with identical payloads but separated
135// by more than the dedup window (250ms) must both survive. The
136// agent legitimately can emit identical adjacent events (think: a
137// retry running the same tool seconds apart).
138func TestReadEventLogKeepsLegitimateAdjacentEvents(t *testing.T) {
139 path := filepath.Join(t.TempDir(), "events.jsonl")
140 log, err := OpenEventLog(path)
141 if err != nil {
142 t.Fatalf("open: %v", err)
143 }
144 now := time.Now()
145 for i := 0; i < 3; i++ {
146 e := NewEvent("tool_call", map[string]any{"name": "read"})
147 e.Time = now.Add(time.Duration(i) * time.Second)
148 if err := log.Append(e); err != nil {
149 t.Fatalf("append: %v", err)
150 }
151 }
152 _ = log.Close()
153
154 got, err := ReadEventLog(path)
155 if err != nil {
156 t.Fatalf("read: %v", err)
157 }
158 if len(got) != 3 {
159 t.Errorf("dedup collapsed legitimately separated events; got %d, want 3", len(got))
160 }
161}
162
163// TestFollowerEmitsNewEvents follows a live file and ensures events
164// appended after the follower starts make it to the channel.
165func TestFollowerEmitsNewEvents(t *testing.T) {
166 path := filepath.Join(t.TempDir(), "events.jsonl")
167 // Pre-seed one event so the file exists. The follower's
168 // contract is "emit events appended AFTER it starts", so this
169 // first event is not expected on the channel.
170 log, err := OpenEventLog(path)
171 if err != nil {
172 t.Fatalf("open: %v", err)
173 }
174 defer log.Close()
175 if err := log.Append(NewEvent("preexisting", nil)); err != nil {
176 t.Fatalf("append: %v", err)
177 }
178
179 fol := FollowEventLog(path, 10*time.Millisecond)
180 defer fol.Close()
181
182 // Give the follower one tick to read the pre-existing event so
183 // its offset advances past the first line. After that, any new
184 // append should show up.
185 time.Sleep(40 * time.Millisecond)
186 drain(fol.Events())
187
188 for i := 0; i < 3; i++ {
189 if err := log.Append(NewEvent("live", map[string]any{"i": i})); err != nil {
190 t.Fatalf("append %d: %v", i, err)
191 }
192 }
193
194 got := waitFor(t, fol.Events(), 3, time.Second)
195 if len(got) != 3 {
196 t.Fatalf("got %d live events, want 3", len(got))
197 }
198 for i, ev := range got {
199 if ev.Type != "live" {
200 t.Errorf("event %d type = %q; want live", i, ev.Type)
201 }
202 }
203}
204
205func drain(ch <-chan Event) {
206 for {
207 select {
208 case <-ch:
209 default:
210 return
211 }
212 }
213}
214
215func waitFor(t *testing.T, ch <-chan Event, n int, timeout time.Duration) []Event {
216 t.Helper()
217 var got []Event
218 deadline := time.After(timeout)
219 for len(got) < n {
220 select {
221 case ev := <-ch:
222 got = append(got, ev)
223 case <-deadline:
224 return got
225 }
226 }
227 return got
228}