amadan

eleven / ocho Public

server: internal/client learns typed-event delivery + ephemeral send/receive (#1453)

The transport slice of the agents design (docs/superpowers/specs/
2026-08-05-eleven-claude-agents-design.md): OnEvent delivers decrypted typed
payloads (live + catch-up), OnEphemeral receives relay-only frames, and
SendEphemeral is the Go end of the web's ctx.emitEphemeral. All opt-in —
a client that sets neither hook behaves exactly as before. This is the
seam the code comments already pointed at (internal/client as the shared
member client layer); the bridge subcommand builds on it next.


Claude-Session: https://claude.ai/code/session_01Xxn758bWfMnEe9xZZuk5Mx
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by codex a106585346a9d1f8edb0f73a213f9d4d03463da1 parent 5c86c2b
3 files changed, +356 −5
  • agents_transport_test.go +235 −0
  • internal/client/client.go +20 −5
  • internal/client/liveevents.go +101 −0
diff --git a/agents_transport_test.go b/agents_transport_test.go
new file mode 100644
index 0000000..c6d223f
--- /dev/null
+++ b/agents_transport_test.go
@@ -0,0 +1,235 @@
+package main
+
+// 🤖 End-to-end tests for internal/client's live typed-event + ephemeral
+// transport (liveevents.go) — the plumbing the agents bridge rides
+// (docs/superpowers/specs/2026-08-05-eleven-claude-agents-design.md). Same
+// harness as eleven_client_test.go: real router via newTestServer, two real
+// E2EE members.
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net/http"
+ "testing"
+ "time"
+
+ lclient "lchat/internal/client"
+ "lchat/internal/e2ee"
+)
+
+// groupForTest creates a named group containing both members, the way the web
+// composer does, and returns its thread id.
+func groupForTest(t *testing.T, tsURL string, a, b *lclient.Client) string {
+ t.Helper()
+ aUID, err := a.Me()
+ if err != nil {
+ t.Fatal(err)
+ }
+ bUID, err := b.Me()
+ if err != nil {
+ t.Fatal(err)
+ }
+ tk, err := e2ee.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ tkB64 := e2ee.B64uEncode(tk)
+ wrapTo := func(pub string) string {
+ w, err := e2ee.WrapThreadKey(tkB64, pub)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return w
+ }
+ nameBlob, err := e2ee.EncryptJSON(tk, map[string]string{"name": "Agents Test"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, _ := json.Marshal(map[string]any{
+ "token": a.Session.Token, "kind": "group", "name_blob": nameBlob,
+ "members": []map[string]string{
+ {"user_id": aUID, "wrapped": wrapTo(a.Session.UserPub)},
+ {"user_id": bUID, "wrapped": wrapTo(b.Session.UserPub)},
+ },
+ })
+ resp, err := http.Post(tsURL+"/api/threads", "application/json", bytes.NewReader(body))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer resp.Body.Close()
+ var created struct {
+ ThreadID string `json:"thread_id"`
+ }
+ if err := json.NewDecoder(resp.Body).Decode(&created); err != nil || created.ThreadID == "" {
+ t.Fatalf("create thread: %v (id %q)", err, created.ThreadID)
+ }
+ return created.ThreadID
+}
+
+func TestClientTypedEventDelivery(t *testing.T) {
+ ts, _ := newTestServer(t)
+ bridge := joinForTest(t, ts.URL, "Claude")
+ member := joinForTest(t, ts.URL, "Paul")
+ thread := groupForTest(t, ts.URL, bridge, member)
+
+ events := make(chan lclient.Event, 8)
+ msgs := make(chan lclient.Message, 8)
+ ready := make(chan struct{})
+ bridge.OnEvent = func(e lclient.Event) { events <- e }
+ bridge.OnMessage = func(m lclient.Message) { msgs <- m }
+ bridge.OnReady = func() { ready <- struct{}{} }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ go bridge.Run(ctx)
+ select {
+ case <-ready:
+ case <-time.After(5 * time.Second):
+ t.Fatal("bridge never connected")
+ }
+
+ // A member emits a typed event (a steering decision): it must arrive on
+ // OnEvent with the transport fields injected, and never on OnMessage.
+ if err := member.SendEvent(thread, map[string]any{
+ "type": "session.approval", "id": "s1", "rid": "a1", "state": "allowed",
+ }); err != nil {
+ t.Fatalf("SendEvent: %v", err)
+ }
+ select {
+ case e := <-events:
+ if e.Type != "session.approval" || e.ThreadID != thread || e.ID == 0 {
+ t.Errorf("event: %+v", e)
+ }
+ if e.Payload["name"] != "Paul" || e.Payload["rid"] != "a1" {
+ t.Errorf("payload: %+v", e.Payload)
+ }
+ if _, ok := e.Payload["uid"].(string); !ok {
+ t.Errorf("uid not injected: %+v", e.Payload)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("typed event never arrived on OnEvent")
+ }
+ select {
+ case m := <-msgs:
+ t.Errorf("typed event leaked to OnMessage: %+v", m)
+ default:
+ }
+
+ // A plain chat message still lands on OnMessage, not OnEvent.
+ if err := member.Send(thread, "hello"); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case m := <-msgs:
+ if m.Text != "hello" {
+ t.Errorf("message: %+v", m)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("chat message never arrived on OnMessage")
+ }
+ select {
+ case e := <-events:
+ t.Errorf("chat message leaked to OnEvent: %+v", e)
+ default:
+ }
+}
+
+func TestClientTypedEventCatchUp(t *testing.T) {
+ ts, _ := newTestServer(t)
+ bridge := joinForTest(t, ts.URL, "Claude")
+ member := joinForTest(t, ts.URL, "Paul")
+ thread := groupForTest(t, ts.URL, bridge, member)
+
+ // Baseline the bridge's cursor on the thread (first connect delivers
+ // nothing historical), then drop the connection.
+ ready := make(chan struct{})
+ events := make(chan lclient.Event, 8)
+ bridge.OnEvent = func(e lclient.Event) { events <- e }
+ bridge.OnReady = func() { ready <- struct{}{} }
+ ctx1, cancel1 := context.WithCancel(context.Background())
+ go bridge.Run(ctx1)
+ select {
+ case <-ready:
+ case <-time.After(5 * time.Second):
+ t.Fatal("bridge never connected")
+ }
+ cancel1()
+
+ // An event lands while the bridge is away; the reconnect's catch-up must
+ // deliver it to OnEvent.
+ if err := member.SendEvent(thread, map[string]any{"type": "session.prompt", "id": "s1", "text": "fix it"}); err != nil {
+ t.Fatal(err)
+ }
+ ctx2, cancel2 := context.WithCancel(context.Background())
+ defer cancel2()
+ go bridge.Run(ctx2)
+ select {
+ case e := <-events:
+ if e.Type != "session.prompt" || e.Payload["text"] != "fix it" {
+ t.Errorf("catch-up event: %+v", e)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("catch-up never delivered the typed event")
+ }
+}
+
+func TestClientEphemeralRoundTrip(t *testing.T) {
+ ts, _ := newTestServer(t)
+ bridge := joinForTest(t, ts.URL, "Claude")
+ member := joinForTest(t, ts.URL, "Paul")
+ thread := groupForTest(t, ts.URL, bridge, member)
+
+ // Both ends live: the bridge sends the tail, the member's client receives
+ // it on OnEphemeral.
+ run := func(c *lclient.Client) (chan lclient.Event, context.CancelFunc) {
+ ev := make(chan lclient.Event, 8)
+ ready := make(chan struct{})
+ c.OnEphemeral = func(e lclient.Event) { ev <- e }
+ c.OnReady = func() { ready <- struct{}{} }
+ ctx, cancel := context.WithCancel(context.Background())
+ go c.Run(ctx)
+ select {
+ case <-ready:
+ case <-time.After(5 * time.Second):
+ t.Fatal("client never connected")
+ }
+ return ev, cancel
+ }
+ memberEv, cancelM := run(member)
+ defer cancelM()
+ _, cancelB := run(bridge)
+ defer cancelB()
+
+ if err := bridge.SendEphemeral(thread, map[string]any{"type": "session.stream", "id": "s1", "text": "partial…"}); err != nil {
+ t.Fatalf("SendEphemeral: %v", err)
+ }
+ select {
+ case e := <-memberEv:
+ if e.Type != "session.stream" || e.ID != 0 || e.Payload["name"] != "Claude" || e.Payload["text"] != "partial…" {
+ t.Errorf("ephemeral: %+v", e)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("ephemeral frame never arrived")
+ }
+
+ // Relay-only means relay-only: nothing was stored.
+ evs, err := member.Events(thread, "session.")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(evs) != 0 {
+ t.Errorf("ephemeral frame was stored: %+v", evs)
+ }
+}
+
+func TestClientSendEphemeralRequiresConnection(t *testing.T) {
+ ts, _ := newTestServer(t)
+ bridge := joinForTest(t, ts.URL, "Claude")
+ member := joinForTest(t, ts.URL, "Paul")
+ thread := groupForTest(t, ts.URL, bridge, member)
+ // No Run: an ephemeral frame has no store to land in, so without a live
+ // socket the send must fail loudly rather than vanish.
+ if err := bridge.SendEphemeral(thread, map[string]any{"type": "session.stream"}); err == nil {
+ t.Fatal("SendEphemeral without a connection should error")
+ }
+}
diff --git a/internal/client/client.go b/internal/client/client.go
index c352ecd..527d81d 100644
--- a/internal/client/client.go
+++ b/internal/client/client.go
@@ -34,7 +34,14 @@ type Client struct {
OnMessage func(Message)
// OnReady, if set, is called once each time the client finishes connecting.
OnReady func()
- Logf func(format string, args ...any)
+ // OnEvent, if set, receives every decrypted TYPED payload (mini-app
+ // events), live or via catch-up — traffic OnMessage deliberately never
+ // sees. Unset, typed payloads are consumed silently, as before.
+ OnEvent func(Event)
+ // OnEphemeral, if set, receives relay-only frames (typing beats, live
+ // tails — id-less, never stored). Unset, they're dropped, as before.
+ OnEphemeral func(Event)
+ Logf func(format string, args ...any)
hc *http.Client
uid string // this member's user id (lazily fetched from /api/me)
@@ -187,9 +194,14 @@ func (c *Client) onFrame(data []byte) {
if json.Unmarshal(data, &f) != nil {
return
}
- // v1 renders new messages only: reactions/edits/deletes and ephemeral
- // presence traffic (typing beats ride as id-less type:"msg") are skipped.
- if f.Type != "msg" || f.ThreadID == "" || f.Ephemeral {
+ // Chat renders new messages only: reactions/edits/deletes are skipped.
+ // Ephemeral presence traffic (id-less type:"msg" frames) goes to the
+ // OnEphemeral hook when a caller wants it, and nowhere else.
+ if f.Type != "msg" || f.ThreadID == "" {
+ return
+ }
+ if f.Ephemeral {
+ c.deliverEphemeral(f.ThreadID, f.Blob)
return
}
c.mu.Lock()
@@ -243,7 +255,10 @@ func (c *Client) deliver(threadID string, id int64, blob string) {
return // e.g. a message from before we held the key
}
if p.Type != "" {
- return // a typed payload (mini-app event, connect traffic), not chat text
+ // A typed payload (mini-app event, connect traffic), not chat text —
+ // delivered through the event hook, never OnMessage.
+ c.deliverEvent(threadID, id, entry.key, blob)
+ return
}
if c.OnMessage != nil {
if p.Name == c.Session.Name {
diff --git a/internal/client/liveevents.go b/internal/client/liveevents.go
new file mode 100644
index 0000000..ae0d99e
--- /dev/null
+++ b/internal/client/liveevents.go
@@ -0,0 +1,101 @@
+package client
+
+// 🤖 Live typed-event + ephemeral transport: the client-side plumbing the
+// agents bridge (docs/superpowers/specs/2026-08-05-eleven-claude-agents-design.md)
+// needs and Events()/SendEvent() don't cover — DELIVERY of typed payloads as
+// they arrive (OnEvent), receipt of relay-only presence frames (OnEphemeral),
+// and the ephemeral send (SendEphemeral, the Go end of the web's
+// ctx.emitEphemeral). The hooks are opt-in: a client that sets neither behaves
+// exactly as before (typed payloads consumed silently, ephemeral frames
+// dropped — the right default for a person's chat surface, and the bot
+// runtime's deliberate policy for a DM bot that must not answer typing beats).
+
+import (
+ "time"
+
+ "lchat/internal/e2ee"
+)
+
+// Event is one decrypted typed payload (a mini-app event — web/static/apps.js
+// routes these by their `type` string). ID is 0 for an ephemeral frame, which
+// the server relays without storing (hub.go: no insert, no id).
+type Event struct {
+ ThreadID string
+ ID int64
+ Type string
+ Payload map[string]any // the full decrypted payload, `type` included
+}
+
+// deliverEvent decrypts a typed payload and hands it to OnEvent. The chat
+// path (deliver) has already advanced the cursor and decided this blob is
+// typed traffic; this decrypts again into the generic shape — a second
+// AES-GCM pass over ≤64KB, cheap and simpler than threading a raw decode
+// through the chat path.
+func (c *Client) deliverEvent(threadID string, id int64, key []byte, blob string) {
+ if c.OnEvent == nil {
+ return
+ }
+ var p map[string]any
+ if e2ee.DecryptJSON(key, blob, &p) != nil {
+ return
+ }
+ t, _ := p["type"].(string)
+ if t == "" {
+ return
+ }
+ c.OnEvent(Event{ThreadID: threadID, ID: id, Type: t, Payload: p})
+}
+
+// deliverEphemeral handles a relay-only frame: decrypt, require a type (the
+// web's ephemeral transport only ever carries typed app events), no cursor —
+// ephemeral frames are id-less by design so they can never be confused with
+// stored messages. Only threads whose key is already cached are considered:
+// ephemeral chatter must not trigger a thread reload (any thread worth
+// steering has durable traffic that loads it the normal way).
+func (c *Client) deliverEphemeral(threadID, blob string) {
+ if c.OnEphemeral == nil || threadID == "" {
+ return
+ }
+ c.mu.Lock()
+ entry, ok := c.keys[threadID]
+ c.mu.Unlock()
+ if !ok {
+ return
+ }
+ var p map[string]any
+ if e2ee.DecryptJSON(entry.key, blob, &p) != nil {
+ return
+ }
+ t, _ := p["type"].(string)
+ if t == "" {
+ return
+ }
+ c.OnEphemeral(Event{ThreadID: threadID, Type: t, Payload: p})
+}
+
+// SendEphemeral relays one typed event to the thread's connected members and
+// nothing else — never stored, never pushed, never folded (the Go end of the
+// web's ctx.emitEphemeral; hub.go re-broadcasts and forgets). The transport
+// fields — name, ts, uid — are injected exactly as SendEvent injects them, so
+// receivers attribute both kinds the same way. Requires a live Run connection:
+// there is no REST fallback because an ephemeral frame has no store to land in.
+func (c *Client) SendEphemeral(threadID string, event map[string]any) error {
+ key, err := c.threadKey(threadID)
+ if err != nil {
+ return err
+ }
+ payload := make(map[string]any, len(event)+3)
+ for k, v := range event {
+ payload[k] = v
+ }
+ payload["name"] = c.Session.Name
+ payload["ts"] = time.Now().UnixMilli()
+ if uid, err := c.Me(); err == nil && uid != "" {
+ payload["uid"] = uid
+ }
+ blob, err := e2ee.EncryptJSON(key, payload)
+ if err != nil {
+ return err
+ }
+ return c.send(map[string]any{"type": "msg", "thread_id": threadID, "blob": blob, "ephemeral": true})
+}