1package subagents
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 subagent package doesn't have to track
19// every field added to core.AgentEvent. The child writes
20// `modes.EventToJSON(ev)` plus a few subagent-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 Version int `json:"version,omitempty"`
34 MessageID string `json:"message_id,omitempty"`
35 AgentID string `json:"agent_id,omitempty"`
36 TurnID string `json:"turn_id,omitempty"`
37 Data map[string]any `json:"-"`
38 Raw map[string]any `json:"-"` // original canonical event for replay
39}
40
41// MarshalJSON writes the canonical versioned event shape used by the
42// worker protocol and durable event logs.
43func (e Event) MarshalJSON() ([]byte, error) {
44 version := e.Version
45 if version == 0 {
46 version = ProtocolVersion
47 }
48 when := e.Time
49 if when.IsZero() {
50 when = time.Now().UTC()
51 }
52 payload := e.Data
53 if payload == nil {
54 payload = map[string]any{}
55 }
56 out := map[string]any{
57 "version": version,
58 "type": e.Type,
59 "timestamp": when,
60 "payload": payload,
61 }
62 if e.MessageID != "" {
63 out["message_id"] = e.MessageID
64 }
65 if e.AgentID != "" {
66 out["agent_id"] = e.AgentID
67 }
68 if e.TurnID != "" {
69 out["turn_id"] = e.TurnID
70 }
71 return json.Marshal(out)
72}
73
74// UnmarshalJSON accepts the canonical versioned event shape and retains the
75// decoded payload for dashboard replay.
76func (e *Event) UnmarshalJSON(b []byte) error {
77 var wire struct {
78 Version int `json:"version"`
79 Type string `json:"type"`
80 MessageID string `json:"message_id,omitempty"`
81 AgentID string `json:"agent_id,omitempty"`
82 TurnID string `json:"turn_id,omitempty"`
83 Timestamp time.Time `json:"timestamp"`
84 Payload map[string]any `json:"payload"`
85 }
86 if err := json.Unmarshal(b, &wire); err != nil {
87 return err
88 }
89 if wire.Version != ProtocolVersion {
90 return fmt.Errorf("subagent event: unsupported version %d", wire.Version)
91 }
92 if wire.Type == "" {
93 return errors.New("subagent event: missing type")
94 }
95 if wire.Timestamp.IsZero() {
96 return errors.New("subagent event: missing timestamp")
97 }
98 if wire.Payload == nil {
99 wire.Payload = map[string]any{}
100 }
101 e.Raw = map[string]any{}
102 if err := json.Unmarshal(b, &e.Raw); err != nil {
103 return err
104 }
105 e.Time = wire.Timestamp
106 e.Type = wire.Type
107 e.Version = wire.Version
108 e.MessageID = wire.MessageID
109 e.AgentID = wire.AgentID
110 e.TurnID = wire.TurnID
111 e.Data = wire.Payload
112 return nil
113}
114
115const (
116 // Event log lines are produced by a child process and may contain model
117 // payloads. Bound both per-line decoding and replay memory so a corrupt or
118 // hostile log cannot force an unbounded allocation during Reload.
119 maxEventLogLineBytes = 2 * 1024 * 1024
120 maxEventLogEvents = 100_000
121)
122
123// EventLog is an append-only writer for events.jsonl. Safe for
124// concurrent writers (the child only has one writer, but tests
125// occasionally fan out).
126type EventLog struct {
127 path string
128 mu sync.Mutex
129 f *os.File
130}
131
132// OpenEventLog opens (or creates) the events.jsonl file at path.
133// Parent directories are created as needed.
134func OpenEventLog(path string) (*EventLog, error) {
135 if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
136 return nil, fmt.Errorf("event log dir: %w", err)
137 }
138 f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
139 if err != nil {
140 return nil, fmt.Errorf("event log open: %w", err)
141 }
142 _ = f.Chmod(0o600)
143 return &EventLog{path: path, f: f}, nil
144}
145
146// Path returns the absolute log path.
147func (l *EventLog) Path() string { return l.path }
148
149// Append writes one event. The encoding is `<json>\n`. Concurrent
150// callers are serialised; small enough events never need partial
151// writes since unix guarantees atomicity for writes ≤ PIPE_BUF on
152// regular files, and we hold a per-process mutex on top.
153func (l *EventLog) Append(ev Event) error {
154 l.mu.Lock()
155 defer l.mu.Unlock()
156 if l.f == nil {
157 return errors.New("event log: closed")
158 }
159 if ev.Time.IsZero() {
160 ev.Time = time.Now().UTC()
161 }
162 b, err := json.Marshal(ev)
163 if err != nil {
164 return err
165 }
166 if len(b) > maxEventLogLineBytes {
167 return fmt.Errorf("event log: event exceeds %d bytes", maxEventLogLineBytes)
168 }
169 b = append(b, '\n')
170 n, err := l.f.Write(b)
171 if err == nil && n != len(b) {
172 return io.ErrShortWrite
173 }
174 return err
175}
176
177// Close flushes and closes the underlying file.
178func (l *EventLog) Close() error {
179 l.mu.Lock()
180 defer l.mu.Unlock()
181 if l.f == nil {
182 return nil
183 }
184 err := l.f.Close()
185 l.f = nil
186 return err
187}
188
189// NewEvent is a convenience constructor that fills the timestamp.
190func NewEvent(typ string, data map[string]any) Event {
191 if data == nil {
192 data = map[string]any{}
193 }
194 return Event{Time: time.Now().UTC(), Type: typ, Version: ProtocolVersion, Data: data}
195}
196
197// ReadEventLog parses every event currently in the file. Used by
198// the dashboard on first open and by tests; live tailing uses
199// FollowEventLog below.
200func ReadEventLog(path string) ([]Event, error) {
201 f, err := os.Open(path)
202 if err != nil {
203 if os.IsNotExist(err) {
204 return nil, nil
205 }
206 return nil, err
207 }
208 defer f.Close()
209 return readEvents(f)
210}
211
212type eventRing struct {
213 events []Event
214 next int
215 count int
216}
217
218func newEventRing() eventRing {
219 return eventRing{events: make([]Event, 0)}
220}
221
222func (r *eventRing) append(ev Event) {
223 if r.count < maxEventLogEvents {
224 r.events = append(r.events, ev)
225 r.count++
226 if r.count == maxEventLogEvents {
227 r.next = 0
228 } else {
229 r.next = r.count
230 }
231 return
232 }
233
234 r.events[r.next] = ev
235 r.next = (r.next + 1) % len(r.events)
236}
237
238func (r *eventRing) last() (Event, bool) {
239 if r.count == 0 {
240 return Event{}, false
241 }
242 i := r.next - 1
243 if i < 0 {
244 i += len(r.events)
245 }
246 return r.events[i], true
247}
248
249func (r *eventRing) isLikelyDoubleWrite(ev Event) bool {
250 previous, ok := r.last()
251 return ok && isLikelyDoubleWrite(ev, previous)
252}
253
254func (r *eventRing) materialize() []Event {
255 out := make([]Event, r.count)
256 if r.count == 0 {
257 return out
258 }
259 start := r.next - r.count
260 if start < 0 {
261 start += len(r.events)
262 }
263 n := copy(out, r.events[start:])
264 copy(out[n:], r.events[:r.count-n])
265 return out
266}
267
268func readEvents(r io.Reader) ([]Event, error) {
269 br := bufio.NewReader(r)
270 ring := newEventRing()
271 for {
272 line, err, truncated := readBoundedLine(br, maxEventLogLineBytes)
273 if len(line) > 0 && !truncated {
274 var ev Event
275 if jerr := json.Unmarshal(line, &ev); jerr == nil && !ring.isLikelyDoubleWrite(ev) {
276 ring.append(ev)
277 }
278 // Malformed and oversized lines are skipped silently; the
279 // dashboard renders only well-formed events and the child is
280 // the only writer.
281 }
282 if err != nil {
283 if errors.Is(err, io.EOF) {
284 return ring.materialize(), nil
285 }
286 return ring.materialize(), err
287 }
288 }
289}
290
291// readFollowerEvents parses only newline-terminated records from a live log.
292// A follower must not advance past an incomplete final line: another process
293// may be in the middle of appending that record, and skipping its start would
294// make the completed event invisible on the next poll.
295func readFollowerEvents(r io.Reader, offset int64) ([]Event, int64) {
296 br := bufio.NewReader(r)
297 ring := newEventRing()
298 committed := offset
299 for {
300 line, truncated, consumed, complete, err := readBoundedLineStats(br, maxEventLogLineBytes)
301 if !complete {
302 break
303 }
304 committed += consumed
305 if len(line) > 0 && !truncated {
306 var ev Event
307 if jerr := json.Unmarshal(line, &ev); jerr == nil && !ring.isLikelyDoubleWrite(ev) {
308 ring.append(ev)
309 }
310 }
311 if err != nil {
312 break
313 }
314 }
315 return ring.materialize(), committed
316}
317
318// isLikelyDoubleWrite reports whether ev is a back-to-back duplicate
319// of previous — same type and identical Data payload, with timestamps
320// within a small window (the supervisor and the child's mirror used to
321// BOTH write each event to disk, landing the same content twice in quick
322// succession in events.jsonl). The historical behaviour was fixed at write
323// time (the child's mirror is now dormant unless the supervisor is gone),
324// but on-disk files from before the fix are still polluted, so we dedupe
325// defensively at read time too.
326//
327// We deliberately bound by time (250ms) so two genuinely identical adjacent
328// events that happen seconds apart — e.g. an agent that runs the same tool
329// twice in a row — still both render.
330func isLikelyDoubleWrite(ev, previous Event) bool {
331 if previous.Type != ev.Type {
332 return false
333 }
334 if !ev.Time.IsZero() && !previous.Time.IsZero() {
335 dt := ev.Time.Sub(previous.Time)
336 if dt < 0 {
337 dt = -dt
338 }
339 if dt > 250*time.Millisecond {
340 return false
341 }
342 }
343 return sameEventData(previous.Data, ev.Data)
344}
345
346// sameEventData deep-compares two event payloads. Cheap because the
347// payloads are small map[string]any trees built from JSON, and only
348// called for adjacent same-type pairs.
349func sameEventData(a, b map[string]any) bool {
350 if len(a) != len(b) {
351 return false
352 }
353 ab, _ := json.Marshal(a)
354 bb, _ := json.Marshal(b)
355 return string(ab) == string(bb)
356}
357
358// EventFollower polls an events.jsonl file and emits new events as
359// they're appended. The supervisor uses one per agent so dashboard
360// snapshots stay current without reparsing the whole file on every
361// frame. Close stops the goroutine; the channel returned by Events
362// is closed when Close is called or the file is removed.
363type EventFollower struct {
364 path string
365 interval time.Duration
366 out chan Event
367 done chan struct{}
368 closed chan struct{}
369 once sync.Once
370}
371
372// FollowEventLog starts polling path every interval. interval ≤ 0
373// defaults to 50ms which is invisible to the user and cheap on
374// disk (one stat + one short read per agent per tick).
375func FollowEventLog(path string, interval time.Duration) *EventFollower {
376 if interval <= 0 {
377 interval = 50 * time.Millisecond
378 }
379 f := &EventFollower{
380 path: path,
381 interval: interval,
382 out: make(chan Event, 256),
383 done: make(chan struct{}),
384 closed: make(chan struct{}),
385 }
386 go f.loop()
387 return f
388}
389
390// Events returns the receive end of the event stream.
391func (f *EventFollower) Events() <-chan Event { return f.out }
392
393// Close stops polling and closes the events channel. It waits for
394// the polling goroutine to exit so tests on Windows can remove the
395// temp directory without racing an open file handle.
396func (f *EventFollower) Close() {
397 f.once.Do(func() {
398 close(f.done)
399 <-f.closed
400 })
401}
402
403func (f *EventFollower) loop() {
404 defer close(f.closed)
405 defer close(f.out)
406 var offset int64
407 tick := time.NewTicker(f.interval)
408 defer tick.Stop()
409 for {
410 select {
411 case <-f.done:
412 return
413 case <-tick.C:
414 }
415 fi, err := os.Stat(f.path)
416 if err != nil {
417 continue
418 }
419 if fi.Size() <= offset {
420 continue
421 }
422 fh, err := os.Open(f.path)
423 if err != nil {
424 continue
425 }
426 if _, err := fh.Seek(offset, io.SeekStart); err != nil {
427 _ = fh.Close()
428 continue
429 }
430 evs, newOffset := readFollowerEvents(fh, offset)
431 _ = fh.Close()
432 offset = newOffset
433 for _, ev := range evs {
434 select {
435 case f.out <- ev:
436 case <-f.done:
437 return
438 }
439 }
440 }
441}