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