| 1 | package main |
| 2 | |
| 3 | // đŸ¤– End-to-end tests for internal/client's live typed-event + ephemeral |
| 4 | // transport (liveevents.go) — the plumbing the agents bridge rides |
| 5 | // (docs/superpowers/specs/2026-08-05-eleven-claude-agents-design.md). Same |
| 6 | // harness as eleven_client_test.go: real router via newTestServer, two real |
| 7 | // E2EE members. |
| 8 | |
| 9 | import ( |
| 10 | "bytes" |
| 11 | "context" |
| 12 | "encoding/json" |
| 13 | "net/http" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | lclient "lchat/internal/client" |
| 18 | "lchat/internal/e2ee" |
| 19 | ) |
| 20 | |
| 21 | // groupForTest creates a named group containing both members, the way the web |
| 22 | // composer does, and returns its thread id. |
| 23 | func groupForTest(t *testing.T, tsURL string, a, b *lclient.Client) string { |
| 24 | t.Helper() |
| 25 | aUID, err := a.Me() |
| 26 | if err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | bUID, err := b.Me() |
| 30 | if err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | tk, err := e2ee.GenerateKey() |
| 34 | if err != nil { |
| 35 | t.Fatal(err) |
| 36 | } |
| 37 | tkB64 := e2ee.B64uEncode(tk) |
| 38 | wrapTo := func(pub string) string { |
| 39 | w, err := e2ee.WrapThreadKey(tkB64, pub) |
| 40 | if err != nil { |
| 41 | t.Fatal(err) |
| 42 | } |
| 43 | return w |
| 44 | } |
| 45 | nameBlob, err := e2ee.EncryptJSON(tk, map[string]string{"name": "Agents Test"}) |
| 46 | if err != nil { |
| 47 | t.Fatal(err) |
| 48 | } |
| 49 | body, _ := json.Marshal(map[string]any{ |
| 50 | "token": a.Session.Token, "kind": "group", "name_blob": nameBlob, |
| 51 | "members": []map[string]string{ |
| 52 | {"user_id": aUID, "wrapped": wrapTo(a.Session.UserPub)}, |
| 53 | {"user_id": bUID, "wrapped": wrapTo(b.Session.UserPub)}, |
| 54 | }, |
| 55 | }) |
| 56 | resp, err := http.Post(tsURL+"/api/threads", "application/json", bytes.NewReader(body)) |
| 57 | if err != nil { |
| 58 | t.Fatal(err) |
| 59 | } |
| 60 | defer resp.Body.Close() |
| 61 | var created struct { |
| 62 | ThreadID string `json:"thread_id"` |
| 63 | } |
| 64 | if err := json.NewDecoder(resp.Body).Decode(&created); err != nil || created.ThreadID == "" { |
| 65 | t.Fatalf("create thread: %v (id %q)", err, created.ThreadID) |
| 66 | } |
| 67 | return created.ThreadID |
| 68 | } |
| 69 | |
| 70 | func TestClientTypedEventDelivery(t *testing.T) { |
| 71 | ts, _ := newTestServer(t) |
| 72 | bridge := joinForTest(t, ts.URL, "Claude") |
| 73 | member := joinForTest(t, ts.URL, "Paul") |
| 74 | thread := groupForTest(t, ts.URL, bridge, member) |
| 75 | |
| 76 | events := make(chan lclient.Event, 8) |
| 77 | msgs := make(chan lclient.Message, 8) |
| 78 | ready := make(chan struct{}) |
| 79 | bridge.OnEvent = func(e lclient.Event) { events <- e } |
| 80 | bridge.OnMessage = func(m lclient.Message) { msgs <- m } |
| 81 | bridge.OnReady = func() { ready <- struct{}{} } |
| 82 | ctx, cancel := context.WithCancel(context.Background()) |
| 83 | defer cancel() |
| 84 | go bridge.Run(ctx) |
| 85 | select { |
| 86 | case <-ready: |
| 87 | case <-time.After(5 * time.Second): |
| 88 | t.Fatal("bridge never connected") |
| 89 | } |
| 90 | |
| 91 | // A member emits a typed event (a steering decision): it must arrive on |
| 92 | // OnEvent with the transport fields injected, and never on OnMessage. |
| 93 | if err := member.SendEvent(thread, map[string]any{ |
| 94 | "type": "session.approval", "id": "s1", "rid": "a1", "state": "allowed", |
| 95 | }); err != nil { |
| 96 | t.Fatalf("SendEvent: %v", err) |
| 97 | } |
| 98 | select { |
| 99 | case e := <-events: |
| 100 | if e.Type != "session.approval" || e.ThreadID != thread || e.ID == 0 { |
| 101 | t.Errorf("event: %+v", e) |
| 102 | } |
| 103 | if e.Payload["name"] != "Paul" || e.Payload["rid"] != "a1" { |
| 104 | t.Errorf("payload: %+v", e.Payload) |
| 105 | } |
| 106 | if _, ok := e.Payload["uid"].(string); !ok { |
| 107 | t.Errorf("uid not injected: %+v", e.Payload) |
| 108 | } |
| 109 | case <-time.After(5 * time.Second): |
| 110 | t.Fatal("typed event never arrived on OnEvent") |
| 111 | } |
| 112 | select { |
| 113 | case m := <-msgs: |
| 114 | t.Errorf("typed event leaked to OnMessage: %+v", m) |
| 115 | default: |
| 116 | } |
| 117 | |
| 118 | // A plain chat message still lands on OnMessage, not OnEvent. |
| 119 | if err := member.Send(thread, "hello"); err != nil { |
| 120 | t.Fatal(err) |
| 121 | } |
| 122 | select { |
| 123 | case m := <-msgs: |
| 124 | if m.Text != "hello" { |
| 125 | t.Errorf("message: %+v", m) |
| 126 | } |
| 127 | case <-time.After(5 * time.Second): |
| 128 | t.Fatal("chat message never arrived on OnMessage") |
| 129 | } |
| 130 | select { |
| 131 | case e := <-events: |
| 132 | t.Errorf("chat message leaked to OnEvent: %+v", e) |
| 133 | default: |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | func TestClientTypedEventCatchUp(t *testing.T) { |
| 138 | ts, _ := newTestServer(t) |
| 139 | bridge := joinForTest(t, ts.URL, "Claude") |
| 140 | member := joinForTest(t, ts.URL, "Paul") |
| 141 | thread := groupForTest(t, ts.URL, bridge, member) |
| 142 | |
| 143 | // Baseline the bridge's cursor on the thread (first connect delivers |
| 144 | // nothing historical), then drop the connection. |
| 145 | ready := make(chan struct{}) |
| 146 | events := make(chan lclient.Event, 8) |
| 147 | bridge.OnEvent = func(e lclient.Event) { events <- e } |
| 148 | bridge.OnReady = func() { ready <- struct{}{} } |
| 149 | ctx1, cancel1 := context.WithCancel(context.Background()) |
| 150 | go bridge.Run(ctx1) |
| 151 | select { |
| 152 | case <-ready: |
| 153 | case <-time.After(5 * time.Second): |
| 154 | t.Fatal("bridge never connected") |
| 155 | } |
| 156 | cancel1() |
| 157 | |
| 158 | // An event lands while the bridge is away; the reconnect's catch-up must |
| 159 | // deliver it to OnEvent. |
| 160 | if err := member.SendEvent(thread, map[string]any{"type": "session.prompt", "id": "s1", "text": "fix it"}); err != nil { |
| 161 | t.Fatal(err) |
| 162 | } |
| 163 | ctx2, cancel2 := context.WithCancel(context.Background()) |
| 164 | defer cancel2() |
| 165 | go bridge.Run(ctx2) |
| 166 | select { |
| 167 | case e := <-events: |
| 168 | if e.Type != "session.prompt" || e.Payload["text"] != "fix it" { |
| 169 | t.Errorf("catch-up event: %+v", e) |
| 170 | } |
| 171 | case <-time.After(5 * time.Second): |
| 172 | t.Fatal("catch-up never delivered the typed event") |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | func TestClientEphemeralRoundTrip(t *testing.T) { |
| 177 | ts, _ := newTestServer(t) |
| 178 | bridge := joinForTest(t, ts.URL, "Claude") |
| 179 | member := joinForTest(t, ts.URL, "Paul") |
| 180 | thread := groupForTest(t, ts.URL, bridge, member) |
| 181 | |
| 182 | // Both ends live: the bridge sends the tail, the member's client receives |
| 183 | // it on OnEphemeral. |
| 184 | run := func(c *lclient.Client) (chan lclient.Event, context.CancelFunc) { |
| 185 | ev := make(chan lclient.Event, 8) |
| 186 | ready := make(chan struct{}) |
| 187 | c.OnEphemeral = func(e lclient.Event) { ev <- e } |
| 188 | c.OnReady = func() { ready <- struct{}{} } |
| 189 | ctx, cancel := context.WithCancel(context.Background()) |
| 190 | go c.Run(ctx) |
| 191 | select { |
| 192 | case <-ready: |
| 193 | case <-time.After(5 * time.Second): |
| 194 | t.Fatal("client never connected") |
| 195 | } |
| 196 | return ev, cancel |
| 197 | } |
| 198 | memberEv, cancelM := run(member) |
| 199 | defer cancelM() |
| 200 | _, cancelB := run(bridge) |
| 201 | defer cancelB() |
| 202 | |
| 203 | if err := bridge.SendEphemeral(thread, map[string]any{"type": "session.stream", "id": "s1", "text": "partial…"}); err != nil { |
| 204 | t.Fatalf("SendEphemeral: %v", err) |
| 205 | } |
| 206 | select { |
| 207 | case e := <-memberEv: |
| 208 | if e.Type != "session.stream" || e.ID != 0 || e.Payload["name"] != "Claude" || e.Payload["text"] != "partial…" { |
| 209 | t.Errorf("ephemeral: %+v", e) |
| 210 | } |
| 211 | case <-time.After(5 * time.Second): |
| 212 | t.Fatal("ephemeral frame never arrived") |
| 213 | } |
| 214 | |
| 215 | // Relay-only means relay-only: nothing was stored. |
| 216 | evs, err := member.Events(thread, "session.") |
| 217 | if err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | if len(evs) != 0 { |
| 221 | t.Errorf("ephemeral frame was stored: %+v", evs) |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | func TestClientSendEphemeralRequiresConnection(t *testing.T) { |
| 226 | ts, _ := newTestServer(t) |
| 227 | bridge := joinForTest(t, ts.URL, "Claude") |
| 228 | member := joinForTest(t, ts.URL, "Paul") |
| 229 | thread := groupForTest(t, ts.URL, bridge, member) |
| 230 | // No Run: an ephemeral frame has no store to land in, so without a live |
| 231 | // socket the send must fail loudly rather than vanish. |
| 232 | if err := bridge.SendEphemeral(thread, map[string]any{"type": "session.stream"}); err == nil { |
| 233 | t.Fatal("SendEphemeral without a connection should error") |
| 234 | } |
| 235 | } |
| 236 | |
amadan