1package swarm
2
3import (
4 "bufio"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 "sync"
12 "time"
13)
14
15// Event is one structured datum in an agent's durable event log.
16//
17// The shape is intentionally a thin map+timestamp wrapper rather
18// than a strict struct so the swarm package doesn't have to track
19// every field added to core.AgentEvent. The child writes
20// `modes.EventToJSON(ev)` plus a few swarm-internal types
21// (lifecycle, user_input echo, error). The supervisor doesn't
22// interpret most fields; it just keeps them for replay and lifts
23// a few well-known ones into AgentSnapshot (Activity, Status,
24// Tail) for the dashboard.
25//
26// The on-disk file (events.jsonl) is append-only, one event per
27// line, newline-terminated JSON. Reading is forward-only; readers
28// stat the file size and read from their last offset on every
29// poll.
30type Event struct {
31 Time time.Time `json:"time"`
32 Type string `json:"type"`
33 Data map[string]any `json:"-"`
34 Raw map[string]any `json:"-"` // includes type+time+data for replay
35}
36
37// MarshalJSON flattens Data into the top-level object so consumers
38// see {type, time, ...fields} rather than {type, time, data:{...}}.
39func (e Event) MarshalJSON() ([]byte, error) {
40 out := make(map[string]any, len(e.Data)+2)
41 for k, v := range e.Data {
42 out[k] = v
43 }
44 out["type"] = e.Type
45 out["time"] = e.Time
46 return json.Marshal(out)
47}
48
49// UnmarshalJSON accepts a flat object with at least type+time.
50// All other fields land in both Data and Raw.
51func (e *Event) UnmarshalJSON(b []byte) error {
52 m := map[string]any{}
53 if err := json.Unmarshal(b, &m); err != nil {
54 return err
55 }
56 e.Raw = m
57 e.Data = map[string]any{}
58 for k, v := range m {
59 if k == "type" || k == "time" {
60 continue
61 }
62 e.Data[k] = v
63 }
64 if t, ok := m["type"].(string); ok {
65 e.Type = t
66 } else {
67 return errors.New("swarm event: missing type")
68 }
69 if ts, ok := m["time"].(string); ok {
70 parsed, err := time.Parse(time.RFC3339Nano, ts)
71 if err == nil {
72 e.Time = parsed
73 }
74 }
75 return nil
76}
77
78// EventLog is an append-only writer for events.jsonl. Safe for
79// concurrent writers (the child only has one writer, but tests
80// occasionally fan out).
81type EventLog struct {
82 path string
83 mu sync.Mutex
84 f *os.File
85}
86
87// OpenEventLog opens (or creates) the events.jsonl file at path.
88// Parent directories are created as needed.
89func OpenEventLog(path string) (*EventLog, error) {
90 if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
91 return nil, fmt.Errorf("event log dir: %w", err)
92 }
93 f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
94 if err != nil {
95 return nil, fmt.Errorf("event log open: %w", err)
96 }
97 return &EventLog{path: path, f: f}, nil
98}
99
100// Path returns the absolute log path.
101func (l *EventLog) Path() string { return l.path }
102
103// Append writes one event. The encoding is `<json>\n`. Concurrent
104// callers are serialised; small enough events never need partial
105// writes since unix guarantees atomicity for writes ≤ PIPE_BUF on
106// regular files, and we hold a per-process mutex on top.
107func (l *EventLog) Append(ev Event) error {
108 l.mu.Lock()
109 defer l.mu.Unlock()
110 if ev.Time.IsZero() {
111 ev.Time = time.Now()
112 }
113 b, err := json.Marshal(ev)
114 if err != nil {
115 return err
116 }
117 b = append(b, '\n')
118 _, err = l.f.Write(b)
119 return err
120}
121
122// Close flushes and closes the underlying file.
123func (l *EventLog) Close() error {
124 l.mu.Lock()
125 defer l.mu.Unlock()
126 if l.f == nil {
127 return nil
128 }
129 err := l.f.Close()
130 l.f = nil
131 return err
132}
133
134// NewEvent is a convenience constructor that fills the timestamp.
135func NewEvent(typ string, data map[string]any) Event {
136 if data == nil {
137 data = map[string]any{}
138 }
139 return Event{Time: time.Now(), Type: typ, Data: data}
140}
141
142// ReadEventLog parses every event currently in the file. Used by
143// the dashboard on first open and by tests; live tailing uses
144// FollowEventLog below.
145func ReadEventLog(path string) ([]Event, error) {
146 f, err := os.Open(path)
147 if err != nil {
148 if os.IsNotExist(err) {
149 return nil, nil
150 }
151 return nil, err
152 }
153 defer f.Close()
154 return readEvents(f)
155}
156
157// ReadEventLogTail parses only the last maxBytes of the log.
158// A partial first line is skipped; a complete line on the boundary is kept.
159func ReadEventLogTail(path string, maxBytes int64) ([]Event, error) {
160 f, err := os.Open(path)
161 if err != nil {
162 if os.IsNotExist(err) {
163 return nil, nil
164 }
165 return nil, err
166 }
167 defer f.Close()
168 fi, err := f.Stat()
169 if err != nil {
170 return nil, err
171 }
172 if maxBytes <= 0 || fi.Size() <= maxBytes {
173 return readEvents(f)
174 }
175 start := fi.Size() - maxBytes
176 var prev [1]byte
177 if _, err := f.ReadAt(prev[:], start-1); err != nil {
178 return nil, err
179 }
180 if _, err := f.Seek(start, io.SeekStart); err != nil {
181 return nil, err
182 }
183 br := bufio.NewReader(f)
184 if prev[0] != '\n' {
185 if _, err := br.ReadBytes('\n'); err != nil {
186 if errors.Is(err, io.EOF) {
187 return nil, nil
188 }
189 return nil, err
190 }
191 }
192 return readEvents(br)
193}
194
195func readEvents(r io.Reader) ([]Event, error) {
196 br := bufio.NewReader(r)
197 var out []Event
198 for {
199 line, err := br.ReadBytes('\n')
200 if len(line) > 0 {
201 var ev Event
202 if jerr := json.Unmarshal(line, &ev); jerr == nil {
203 if !isLikelyDoubleWrite(ev, out) {
204 out = append(out, ev)
205 }
206 }
207 // Malformed lines are skipped silently; the dashboard
208 // renders only well-formed events and the child is the
209 // only writer.
210 }
211 if err != nil {
212 if errors.Is(err, io.EOF) {
213 return out, nil
214 }
215 return out, err
216 }
217 }
218}
219
220// isLikelyDoubleWrite reports whether ev is a back-to-back duplicate
221// of the most recent event in `tail` — same type and identical Data
222// payload, with timestamps within a small window (the supervisor and
223// the child's mirror used to BOTH write each event to disk, landing
224// the same content twice in quick succession in events.jsonl). The
225// historical behaviour was fixed at write time (the child's mirror
226// is now dormant unless the supervisor is gone), but on-disk files
227// from before the fix are still polluted, so we dedupe defensively
228// at read time too.
229//
230// We deliberately bound by time (250ms) so two genuinely identical
231// adjacent events that happen seconds apart — e.g. an agent that
232// runs the same tool twice in a row — still both render.
233func isLikelyDoubleWrite(ev Event, tail []Event) bool {
234 if len(tail) == 0 {
235 return false
236 }
237 prev := tail[len(tail)-1]
238 if prev.Type != ev.Type {
239 return false
240 }
241 if !ev.Time.IsZero() && !prev.Time.IsZero() {
242 dt := ev.Time.Sub(prev.Time)
243 if dt < 0 {
244 dt = -dt
245 }
246 if dt > 250*time.Millisecond {
247 return false
248 }
249 }
250 return sameEventData(prev.Data, ev.Data)
251}
252
253// sameEventData deep-compares two event payloads. Cheap because the
254// payloads are small map[string]any trees built from JSON, and only
255// called for adjacent same-type pairs.
256func sameEventData(a, b map[string]any) bool {
257 if len(a) != len(b) {
258 return false
259 }
260 ab, _ := json.Marshal(a)
261 bb, _ := json.Marshal(b)
262 return string(ab) == string(bb)
263}
264
265// EventFollower polls an events.jsonl file and emits new events as
266// they're appended. The supervisor uses one per agent so dashboard
267// snapshots stay current without reparsing the whole file on every
268// frame. Close stops the goroutine; the channel returned by Events
269// is closed when Close is called or the file is removed.
270type EventFollower struct {
271 path string
272 interval time.Duration
273 out chan Event
274 done chan struct{}
275 closed chan struct{}
276 once sync.Once
277}
278
279// FollowEventLog starts polling path every interval. interval ≤ 0
280// defaults to 50ms which is invisible to the user and cheap on
281// disk (one stat + one short read per agent per tick).
282func FollowEventLog(path string, interval time.Duration) *EventFollower {
283 if interval <= 0 {
284 interval = 50 * time.Millisecond
285 }
286 f := &EventFollower{
287 path: path,
288 interval: interval,
289 out: make(chan Event, 256),
290 done: make(chan struct{}),
291 closed: make(chan struct{}),
292 }
293 go f.loop()
294 return f
295}
296
297// Events returns the receive end of the event stream.
298func (f *EventFollower) Events() <-chan Event { return f.out }
299
300// Close stops polling and closes the events channel. It waits for
301// the polling goroutine to exit so tests on Windows can remove the
302// temp directory without racing an open file handle.
303func (f *EventFollower) Close() {
304 f.once.Do(func() {
305 close(f.done)
306 <-f.closed
307 })
308}
309
310func (f *EventFollower) loop() {
311 defer close(f.closed)
312 defer close(f.out)
313 var offset int64
314 tick := time.NewTicker(f.interval)
315 defer tick.Stop()
316 for {
317 select {
318 case <-f.done:
319 return
320 case <-tick.C:
321 }
322 fi, err := os.Stat(f.path)
323 if err != nil {
324 continue
325 }
326 if fi.Size() <= offset {
327 continue
328 }
329 fh, err := os.Open(f.path)
330 if err != nil {
331 continue
332 }
333 if _, err := fh.Seek(offset, io.SeekStart); err != nil {
334 _ = fh.Close()
335 continue
336 }
337 evs, _ := readEvents(fh)
338 newOffset, _ := fh.Seek(0, io.SeekCurrent)
339 _ = fh.Close()
340 offset = newOffset
341 for _, ev := range evs {
342 select {
343 case f.out <- ev:
344 case <-f.done:
345 return
346 }
347 }
348 }
349}