1package agent
2
3import (
4 "bytes"
5 "encoding/json"
6 "io"
7 "os"
8 "path/filepath"
9 "testing"
10
11 "github.com/patriceckhart/zot/internal/swarm"
12)
13
14// TestSwarmEmitterMirrorDormantUntilStdoutBreaks regresses the
15// "everything is doubled after reopening a swarm agent" bug.
16//
17// Symptom: events.jsonl held two copies of every event because the
18// child mirrored each event to disk AND the supervisor parsed the
19// child's stdout and appended each event to disk too. On next zot
20// launch the replay produced two transcript lines per real one.
21//
22// Fix invariant: while stdout writes succeed (i.e. the supervisor is
23// alive on the other end of the pipe), the child's mirror writes
24// NOTHING. Only when a stdout write fails (broken pipe → orphan)
25// does the mirror take over so events still get persisted.
26func TestSwarmEmitterMirrorDormantUntilStdoutBreaks(t *testing.T) {
27 // Real *os.File for the emitter's stdout-equivalent so the
28 // emitter's write() path (which expects *os.File) actually runs.
29 stdoutPath := filepath.Join(t.TempDir(), "stdout.fifo")
30 stdoutFile, err := os.Create(stdoutPath)
31 if err != nil {
32 t.Fatalf("create stdout file: %v", err)
33 }
34 defer stdoutFile.Close()
35
36 // Mirror writes go to a separate events.jsonl that we can read
37 // at the end to assert how many events the mirror emitted.
38 mirrorPath := filepath.Join(t.TempDir(), "events.jsonl")
39 mirror, err := swarm.OpenEventLog(mirrorPath)
40 if err != nil {
41 t.Fatalf("open mirror: %v", err)
42 }
43 defer mirror.Close()
44
45 em := newSwarmEmitter(stdoutFile, mirror)
46
47 // Healthy stdout: emit three events. Mirror must stay empty.
48 em.emit("turn_start", map[string]any{"step": 1})
49 em.emit("assistant_message", map[string]any{"text": "hi"})
50 em.emit("turn_end", map[string]any{"step": 1})
51
52 got, err := swarm.ReadEventLog(mirrorPath)
53 if err != nil {
54 t.Fatalf("read mirror: %v", err)
55 }
56 if len(got) != 0 {
57 t.Fatalf("mirror wrote %d events while supervisor was alive; want 0 (every event would otherwise double on the next reload)\n%+v",
58 len(got), got)
59 }
60
61 // Simulate supervisor death: close stdout so the next Write
62 // returns EBADF / broken pipe. The emitter must flip into
63 // orphan mode and start writing through the mirror.
64 if err := stdoutFile.Close(); err != nil {
65 t.Fatalf("close stdout: %v", err)
66 }
67
68 em.emit("assistant_message", map[string]any{"text": "after orphan"})
69 em.emit("turn_end", map[string]any{"step": 2})
70
71 got, err = swarm.ReadEventLog(mirrorPath)
72 if err != nil {
73 t.Fatalf("read mirror post-orphan: %v", err)
74 }
75 if len(got) < 2 {
76 t.Fatalf("mirror failed to take over after stdout died: got %d events", len(got))
77 }
78 if got[len(got)-1].Type != "turn_end" {
79 t.Errorf("last mirrored event type = %q; want turn_end", got[len(got)-1].Type)
80 }
81}
82
83// TestSwarmEmitterStdoutShapeMatchesSupervisorParser pins the
84// wire-format contract: each emitted event lands on stdout as one
85// JSON object per line with type+time at top level alongside the
86// data fields. The supervisor's runner parses this exact shape.
87func TestSwarmEmitterStdoutShapeMatchesSupervisorParser(t *testing.T) {
88 // Pipe so we can read what the emitter wrote.
89 r, w, err := os.Pipe()
90 if err != nil {
91 t.Fatalf("pipe: %v", err)
92 }
93 defer r.Close()
94
95 em := newSwarmEmitter(w, nil)
96 em.emit("turn_start", map[string]any{"step": 1})
97 _ = w.Close()
98
99 body, err := io.ReadAll(r)
100 if err != nil {
101 t.Fatalf("read pipe: %v", err)
102 }
103 // One trailing newline => one event line.
104 lines := bytes.Split(bytes.TrimRight(body, "\n"), []byte("\n"))
105 if len(lines) != 1 {
106 t.Fatalf("expected 1 event line, got %d: %q", len(lines), body)
107 }
108 var flat map[string]any
109 if err := json.Unmarshal(lines[0], &flat); err != nil {
110 t.Fatalf("not valid json: %v\n%s", err, lines[0])
111 }
112 if flat["type"] != "turn_start" {
113 t.Errorf("type field missing or wrong: %v", flat["type"])
114 }
115 if _, ok := flat["time"].(string); !ok {
116 t.Errorf("time field missing: %v", flat["time"])
117 }
118 if flat["step"] != float64(1) {
119 t.Errorf("step field missing or wrong: %v", flat["step"])
120 }
121}