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
157func readEvents(r io.Reader) ([]Event, error) {
158 br := bufio.NewReader(r)
159 var out []Event
160 for {
161 line, err := br.ReadBytes('\n')
162 if len(line) > 0 {
163 var ev Event
164 if jerr := json.Unmarshal(line, &ev); jerr == nil {
165 if !isLikelyDoubleWrite(ev, out) {
166 out = append(out, ev)
167 }
168 }
169 // Malformed lines are skipped silently; the dashboard
170 // renders only well-formed events and the child is the
171 // only writer.
172 }
173 if err != nil {
174 if errors.Is(err, io.EOF) {
175 return out, nil
176 }
177 return out, err
178 }
179 }
180}
181
182// isLikelyDoubleWrite reports whether ev is a back-to-back duplicate
183// of the most recent event in `tail` — same type and identical Data
184// payload, with timestamps within a small window (the supervisor and
185// the child's mirror used to BOTH write each event to disk, landing
186// the same content twice in quick succession in events.jsonl). The
187// historical behaviour was fixed at write time (the child's mirror
188// is now dormant unless the supervisor is gone), but on-disk files
189// from before the fix are still polluted, so we dedupe defensively
190// at read time too.
191//
192// We deliberately bound by time (250ms) so two genuinely identical
193// adjacent events that happen seconds apart — e.g. an agent that
194// runs the same tool twice in a row — still both render.
195func isLikelyDoubleWrite(ev Event, tail []Event) bool {
196 if len(tail) == 0 {
197 return false
198 }
199 prev := tail[len(tail)-1]
200 if prev.Type != ev.Type {
201 return false
202 }
203 if !ev.Time.IsZero() && !prev.Time.IsZero() {
204 dt := ev.Time.Sub(prev.Time)
205 if dt < 0 {
206 dt = -dt
207 }
208 if dt > 250*time.Millisecond {
209 return false
210 }
211 }
212 return sameEventData(prev.Data, ev.Data)
213}
214
215// sameEventData deep-compares two event payloads. Cheap because the
216// payloads are small map[string]any trees built from JSON, and only
217// called for adjacent same-type pairs.
218func sameEventData(a, b map[string]any) bool {
219 if len(a) != len(b) {
220 return false
221 }
222 ab, _ := json.Marshal(a)
223 bb, _ := json.Marshal(b)
224 return string(ab) == string(bb)
225}
226
227// EventFollower polls an events.jsonl file and emits new events as
228// they're appended. The supervisor uses one per agent so dashboard
229// snapshots stay current without reparsing the whole file on every
230// frame. Close stops the goroutine; the channel returned by Events
231// is closed when Close is called or the file is removed.
232type EventFollower struct {
233 path string
234 interval time.Duration
235 out chan Event
236 done chan struct{}
237 closed chan struct{}
238 once sync.Once
239}
240
241// FollowEventLog starts polling path every interval. interval ≤ 0
242// defaults to 50ms which is invisible to the user and cheap on
243// disk (one stat + one short read per agent per tick).
244func FollowEventLog(path string, interval time.Duration) *EventFollower {
245 if interval <= 0 {
246 interval = 50 * time.Millisecond
247 }
248 f := &EventFollower{
249 path: path,
250 interval: interval,
251 out: make(chan Event, 256),
252 done: make(chan struct{}),
253 closed: make(chan struct{}),
254 }
255 go f.loop()
256 return f
257}
258
259// Events returns the receive end of the event stream.
260func (f *EventFollower) Events() <-chan Event { return f.out }
261
262// Close stops polling and closes the events channel. It waits for
263// the polling goroutine to exit so tests on Windows can remove the
264// temp directory without racing an open file handle.
265func (f *EventFollower) Close() {
266 f.once.Do(func() {
267 close(f.done)
268 <-f.closed
269 })
270}
271
272func (f *EventFollower) loop() {
273 defer close(f.closed)
274 defer close(f.out)
275 var offset int64
276 tick := time.NewTicker(f.interval)
277 defer tick.Stop()
278 for {
279 select {
280 case <-f.done:
281 return
282 case <-tick.C:
283 }
284 fi, err := os.Stat(f.path)
285 if err != nil {
286 continue
287 }
288 if fi.Size() <= offset {
289 continue
290 }
291 fh, err := os.Open(f.path)
292 if err != nil {
293 continue
294 }
295 if _, err := fh.Seek(offset, io.SeekStart); err != nil {
296 _ = fh.Close()
297 continue
298 }
299 evs, _ := readEvents(fh)
300 newOffset, _ := fh.Seek(0, io.SeekCurrent)
301 _ = fh.Close()
302 offset = newOffset
303 for _, ev := range evs {
304 select {
305 case f.out <- ev:
306 case <-f.done:
307 return
308 }
309 }
310 }
311}