| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "strings" |
| 6 | "testing" |
| 7 | |
| 8 | "reasonix/internal/event" |
| 9 | "reasonix/internal/eventwire" |
| 10 | ) |
| 11 | |
| 12 | func TestBroadcasterFiltersSessions(t *testing.T) { |
| 13 | b := NewBroadcaster() |
| 14 | b.SetCurrentSession("/sessions/current.jsonl") |
| 15 | current, stopCurrent := b.Subscribe() |
| 16 | all, stopAll := b.SubscribeAll() |
| 17 | defer stopCurrent() |
| 18 | defer stopAll() |
| 19 | |
| 20 | b.Emit(event.Event{Kind: event.Text, Text: "current", SessionPath: "/sessions/current.jsonl"}) |
| 21 | b.Emit(event.Event{Kind: event.Text, Text: "background", SessionPath: "/sessions/background.jsonl"}) |
| 22 | b.Emit(event.Event{Kind: event.Text, Text: "legacy"}) |
| 23 | |
| 24 | drain := func(ch <-chan []byte) []string { |
| 25 | var frames []string |
| 26 | for { |
| 27 | select { |
| 28 | case frame := <-ch: |
| 29 | frames = append(frames, string(frame)) |
| 30 | default: |
| 31 | return frames |
| 32 | } |
| 33 | } |
| 34 | } |
| 35 | if got := len(drain(current)); got != 2 { |
| 36 | t.Fatalf("current subscription received %d frames, want 2", got) |
| 37 | } |
| 38 | if got := len(drain(all)); got != 3 { |
| 39 | t.Fatalf("all-session subscription received %d frames, want 3", got) |
| 40 | } |
| 41 | } |
| 42 | |
| 43 | func TestBroadcasterMarksForegroundFramesAtPublication(t *testing.T) { |
| 44 | b := NewBroadcaster() |
| 45 | b.SetCurrentSession("/sessions/current.jsonl") |
| 46 | all, stop := b.SubscribeAll() |
| 47 | defer stop() |
| 48 | |
| 49 | b.Emit(event.Event{Kind: event.Text, Text: "current", SessionPath: "/sessions/current.jsonl"}) |
| 50 | b.Emit(event.Event{Kind: event.Text, Text: "background", SessionPath: "/sessions/background.jsonl"}) |
| 51 | |
| 52 | var current, background eventwire.Event |
| 53 | if err := json.Unmarshal(<-all, ¤t); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | if err := json.Unmarshal(<-all, &background); err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | if !current.SessionCurrent { |
| 60 | t.Fatalf("foreground frame was not marked current: %+v", current) |
| 61 | } |
| 62 | if background.SessionCurrent { |
| 63 | t.Fatalf("background frame was marked current: %+v", background) |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | func TestBroadcasterFanOut(t *testing.T) { |
| 68 | b := NewBroadcaster() |
| 69 | a, ca := b.Subscribe() |
| 70 | d, cd := b.Subscribe() |
| 71 | defer ca() |
| 72 | defer cd() |
| 73 | |
| 74 | if got := b.Subscribers(); got != 2 { |
| 75 | t.Fatalf("subscribers = %d, want 2", got) |
| 76 | } |
| 77 | |
| 78 | b.Emit(event.Event{Kind: event.Text, Text: "hi"}) |
| 79 | |
| 80 | for i, ch := range []<-chan []byte{a, d} { |
| 81 | var w eventwire.Event |
| 82 | if err := json.Unmarshal(<-ch, &w); err != nil { |
| 83 | t.Fatalf("subscriber %d: %v", i, err) |
| 84 | } |
| 85 | if w.Kind != "text" || w.Text != "hi" { |
| 86 | t.Errorf("subscriber %d got %+v", i, w) |
| 87 | } |
| 88 | } |
| 89 | } |
| 90 | |
| 91 | func TestBroadcasterEmitToHonorsCurrentSession(t *testing.T) { |
| 92 | b := NewBroadcaster() |
| 93 | b.SetCurrentSession("/sessions/b.jsonl") |
| 94 | current, stopCurrent := b.Subscribe() |
| 95 | all, stopAll := b.SubscribeAll() |
| 96 | defer stopCurrent() |
| 97 | defer stopAll() |
| 98 | b.EmitTo(current, event.Event{Kind: event.ApprovalRequest, SessionPath: "/sessions/a.jsonl"}) |
| 99 | b.EmitTo(all, event.Event{Kind: event.ApprovalRequest, SessionPath: "/sessions/a.jsonl"}) |
| 100 | if len(current) != 0 { |
| 101 | t.Fatal("current-only subscriber received a stale session replay") |
| 102 | } |
| 103 | if len(all) != 1 { |
| 104 | t.Fatal("all-session subscriber lost a tagged background replay") |
| 105 | } |
| 106 | b.EmitTo(current, event.Event{Kind: event.ApprovalRequest, SessionPath: "/sessions/b.jsonl"}) |
| 107 | if len(current) != 1 { |
| 108 | t.Fatal("current-only subscriber lost the current session replay") |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | func TestBroadcasterEmitsRetryingJSON(t *testing.T) { |
| 113 | b := NewBroadcaster() |
| 114 | ch, cancel := b.Subscribe() |
| 115 | defer cancel() |
| 116 | |
| 117 | b.Emit(event.Event{Kind: event.Retrying, RetryAttempt: 3, RetryMax: 10}) |
| 118 | |
| 119 | s := string(<-ch) |
| 120 | for _, want := range []string{`"kind":"retrying"`, `"retryAttempt":3`, `"retryMax":10`} { |
| 121 | if !strings.Contains(s, want) { |
| 122 | t.Fatalf("retrying broadcast JSON = %s, want it to contain %s", s, want) |
| 123 | } |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | func TestBroadcasterUnsubscribe(t *testing.T) { |
| 128 | b := NewBroadcaster() |
| 129 | _, cancel := b.Subscribe() |
| 130 | if b.Subscribers() != 1 { |
| 131 | t.Fatalf("want 1 subscriber") |
| 132 | } |
| 133 | cancel() |
| 134 | if b.Subscribers() != 0 { |
| 135 | t.Fatalf("unsubscribe should drop to 0, got %d", b.Subscribers()) |
| 136 | } |
| 137 | // Emitting with no subscribers must not panic. |
| 138 | b.Emit(event.Event{Kind: event.TurnDone}) |
| 139 | } |
| 140 | |
| 141 | func TestBroadcasterDropsSlowSubscriber(t *testing.T) { |
| 142 | b := NewBroadcaster() |
| 143 | ch, cancel := b.Subscribe() |
| 144 | defer cancel() |
| 145 | // Overfill far past the subscriber buffer without reading; Emit must not block. |
| 146 | for range 1000 { |
| 147 | b.Emit(event.Event{Kind: event.Text, Text: "x"}) |
| 148 | } |
| 149 | if len(ch) == 0 { |
| 150 | t.Error("expected some buffered frames") |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | func TestBroadcasterReservesCapacityForTerminalFrames(t *testing.T) { |
| 155 | b := NewBroadcaster() |
| 156 | ch, cancel := b.SubscribeAll() |
| 157 | defer cancel() |
| 158 | for range subscriberBufferSize * 10 { |
| 159 | b.Emit(event.Event{Kind: event.Text, Text: "delta"}) |
| 160 | } |
| 161 | b.Emit(event.Event{Kind: event.TurnDone}) |
| 162 | |
| 163 | found := false |
| 164 | for len(ch) > 0 { |
| 165 | var frame eventwire.Event |
| 166 | if err := json.Unmarshal(<-ch, &frame); err != nil { |
| 167 | t.Fatal(err) |
| 168 | } |
| 169 | if frame.Kind == "turn_done" { |
| 170 | found = true |
| 171 | } |
| 172 | } |
| 173 | if !found { |
| 174 | t.Fatal("slow subscriber lost the terminal frame after a delta flood") |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | func TestBroadcasterEvictsRecoverableFramesForTerminalEvents(t *testing.T) { |
| 179 | b := NewBroadcaster() |
| 180 | ch, cancel := b.SubscribeAll() |
| 181 | defer cancel() |
| 182 | for range subscriberBufferSize - subscriberPriorityReserve { |
| 183 | b.Emit(event.Event{Kind: event.Text, Text: "delta"}) |
| 184 | } |
| 185 | for range subscriberPriorityReserve { |
| 186 | b.Emit(event.Event{Kind: event.Notice, Text: "priority"}) |
| 187 | } |
| 188 | if got := len(ch); got != subscriberBufferSize { |
| 189 | t.Fatalf("saturated subscriber length = %d, want %d", got, subscriberBufferSize) |
| 190 | } |
| 191 | |
| 192 | b.Emit(event.Event{Kind: event.TurnDone}) |
| 193 | b.Emit(event.Event{Kind: event.SessionChanged, SessionPath: "/sessions/next.jsonl"}) |
| 194 | |
| 195 | found := map[string]bool{} |
| 196 | for len(ch) > 0 { |
| 197 | var frame eventwire.Event |
| 198 | if err := json.Unmarshal(<-ch, &frame); err != nil { |
| 199 | t.Fatal(err) |
| 200 | } |
| 201 | found[frame.Kind] = true |
| 202 | } |
| 203 | for _, kind := range []string{"turn_done", "session_changed"} { |
| 204 | if !found[kind] { |
| 205 | t.Fatalf("slow subscriber lost %s after priority reserve saturation", kind) |
| 206 | } |
| 207 | } |
| 208 | } |
| 209 | |
| 210 | func TestBroadcasterPreservesBackgroundJobCompletionNotice(t *testing.T) { |
| 211 | b := NewBroadcaster() |
| 212 | ch, cancel := b.SubscribeAll() |
| 213 | defer cancel() |
| 214 | for range subscriberBufferSize - subscriberPriorityReserve { |
| 215 | b.Emit(event.Event{Kind: event.Text, Text: "delta"}) |
| 216 | } |
| 217 | for range subscriberPriorityReserve { |
| 218 | b.Emit(event.Event{Kind: event.Notice, Text: "priority"}) |
| 219 | } |
| 220 | b.Emit(event.Event{Kind: event.Notice, Code: event.NoticeCodeBackgroundJobFinished, Text: "background task finished"}) |
| 221 | |
| 222 | found := false |
| 223 | for len(ch) > 0 { |
| 224 | var frame eventwire.Event |
| 225 | if err := json.Unmarshal(<-ch, &frame); err != nil { |
| 226 | t.Fatal(err) |
| 227 | } |
| 228 | if frame.Kind == "notice" && frame.Code == event.NoticeCodeBackgroundJobFinished { |
| 229 | found = true |
| 230 | } |
| 231 | } |
| 232 | if !found { |
| 233 | t.Fatal("slow subscriber lost background-job completion after priority reserve saturation") |
| 234 | } |
| 235 | } |
| 236 | |
| 237 | // Notices about final-format identities (taken over, reclaim requested, |
| 238 | // reclaimed, adopted) route by "session-id:<id>", which is not a filesystem |
| 239 | // path. Every emit path must carry it verbatim and current-only subscribers |
| 240 | // must still receive it: a canonicalized "<cwd>/session-id:x" matched no |
| 241 | // subscriber and was dropped, so a browser tab never saw the takeover banner. |
| 242 | func TestBroadcasterKeepsIdentityRoutesVerbatim(t *testing.T) { |
| 243 | b := NewBroadcaster() |
| 244 | b.SetCurrentSession("/sessions/current.jsonl") |
| 245 | current, stopCurrent := b.Subscribe() |
| 246 | all, stopAll := b.SubscribeAll() |
| 247 | defer stopCurrent() |
| 248 | defer stopAll() |
| 249 | const route = "session-id:abc" |
| 250 | |
| 251 | b.Emit(event.Event{Kind: event.Notice, Code: event.NoticeCodeSessionTakenOver, SessionPath: route}) |
| 252 | b.EmitTo(current, event.Event{Kind: event.Notice, Code: event.NoticeCodeSessionReclaimed, SessionPath: route}) |
| 253 | b.EmitWire(eventwire.Event{Kind: "text", Text: "mirrored", SessionPath: route}) |
| 254 | |
| 255 | decode := func(name string, ch <-chan []byte) eventwire.Event { |
| 256 | t.Helper() |
| 257 | var frame eventwire.Event |
| 258 | select { |
| 259 | case raw := <-ch: |
| 260 | if err := json.Unmarshal(raw, &frame); err != nil { |
| 261 | t.Fatal(err) |
| 262 | } |
| 263 | default: |
| 264 | t.Fatalf("%s subscriber did not receive the identity-routed frame", name) |
| 265 | } |
| 266 | return frame |
| 267 | } |
| 268 | for range 3 { |
| 269 | if frame := decode("current-only", current); frame.SessionPath != route { |
| 270 | t.Fatalf("current-only frame route = %q, want %q (%+v)", frame.SessionPath, route, frame) |
| 271 | } |
| 272 | } |
| 273 | for range 2 { |
| 274 | if frame := decode("all-session", all); frame.SessionPath != route { |
| 275 | t.Fatalf("all-session frame route = %q, want %q (%+v)", frame.SessionPath, route, frame) |
| 276 | } |
| 277 | } |
| 278 | // Legacy paths keep the path rule: a background transcript is still hidden |
| 279 | // from current-only subscribers. |
| 280 | b.Emit(event.Event{Kind: event.Text, Text: "background", SessionPath: "/sessions/background.jsonl"}) |
| 281 | select { |
| 282 | case raw := <-current: |
| 283 | t.Fatalf("current-only subscriber received a background legacy frame: %s", raw) |
| 284 | default: |
| 285 | } |
| 286 | if frame := decode("all-session", all); frame.SessionPath == "" || strings.Contains(frame.SessionPath, route) { |
| 287 | t.Fatalf("legacy frame lost its path: %+v", frame) |
| 288 | } |
| 289 | } |
| 290 |