feat(sdk): multiplexed subscribe, @self DM routing, stable event ids (#128)

* feat(sdk): multiplexed subscribe, @self DM routing, stable event ids

- AgentClient.subscribe(channels, onMessage): Subscription multiplexes a
  single websocket per agent, auto-resubscribes on reconnect, and filters
  the @self sentinel to DM-only delivery in the SDK.
- Server resolves to: "@self" on DM send from the authenticated agentId
  rather than trusting client-side substitution.
- WebSocket message events now carry a top-level UUID id derived
  deterministically from the underlying message id, so clients can
  dedupe across reconnects.
- RelayError distinguishes rate_limited and backpressure from generic
  transport failures.

Originally implemented as Track A of the proactive-runtime M3 workflow;
the worker reported COMPLETE and ran the @relaycast/types, /sdk, and
/server vitest suites green but never opened a PR. Re-verified locally:
38 + 294 + 437 tests pass on this branch.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(sdk): address subscription review comments

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: khaliqgant

README.md

@@ -38,7 +38,7 @@ await alice.channels.create({ name: 'general', topic: 'Team chat' });
 await bob.channels.join('general');
 await carol.channels.join('general');
 
-// 6) Realtime listeners (on.messageCreated is the onMessage-style hook)
+// 6) Realtime listeners on one multiplexed websocket per agent
 const agents = [
   { name: 'Alice', client: alice },
   { name: 'Bob', client: bob },
@@ -49,18 +49,15 @@ await Promise.all(
   agents.map(
     ({ name, client }) =>
       new Promise<void>((resolve) => {
-        client.connect();
+        client.subscribe(['general', '@self'], (event) => {
+          console.log(`[${name} stream] ${event.message.agentName}: ${event.message.text}`);
+        });
 
         const stopConnected = client.on.connected(() => {
-          client.subscribe(['general']);
           console.log(`${name} websocket connected`);
           stopConnected();
           resolve();
         });
-
-        client.on.messageCreated((event) => {
-          console.log(`[${name} stream] ${event.message.agentName}: ${event.message.text}`);
-        });
       }),
   ),
 );
@@ -75,7 +72,6 @@ await new Promise((resolve) => setTimeout(resolve, 1500));
 
 // 8) Cleanup
 for (const { client } of agents) {
-  client.unsubscribe(['general']);
   await client.disconnect();
 }
 ```
@@ -140,8 +136,7 @@ const { token } = await relay.agents.register({ name: 'Reviewer', type: 'agent'
 const me = relay.as(token);
 
 me.connect();
-me.on.connected(() => me.subscribe(['general']));
-me.on.messageCreated((event) => {
+me.subscribe(['general', '@self'], (event) => {
   console.log(`${event.message.agentName}: ${event.message.text}`);
 });
 
@@ -172,19 +167,12 @@ const relay = new RelayCast({ apiKey, baseUrl: localBaseUrl });
 Realtime example:
 
 ```typescript
-me.connect();
-const stopConnected = me.on.connected(() => {
-  me.subscribe(['general']);
-  stopConnected();
-});
-
-const unsub = me.on.messageCreated((event) => {
+const sub = me.subscribe(['general', '@self'], (event) => {
   console.log(`${event.message.agentName}: ${event.message.text}`);
 });
 
 // later
-unsub();
-me.unsubscribe(['general']);
+sub.unsubscribe();
 await me.disconnect();
 ```
 

openapi.yaml

@@ -1545,7 +1545,12 @@ paths:
   /dm:
     post:
       summary: Send DM
-      description: Send a direct message to another agent
+      description: |
+        Send a direct message to another agent.
+
+        The `to` field also accepts the `@self` sentinel, which is resolved
+        on the server to the authenticated agent identity so callers do not
+        need to guess their own routed name.
       tags:
         - Direct Messages
       security:
@@ -1567,7 +1572,7 @@ paths:
               properties:
                 to:
                   type: string
-                  description: Recipient agent name
+                  description: Recipient agent name or `@self`
                 text:
                   type: string
                 attachments:

packages/sdk-typescript/src/__tests__/agent-messaging.test.ts

@@ -104,6 +104,20 @@ describe('AgentClient', () => {
     });
   });
 
+  describe('post()', () => {
+    it('aliases send() for channel messages', async () => {
+      mockFetch.mockImplementation(() => mockResponse({ id: 'm_1' }));
+
+      await me.post('#general', 'hello');
+
+      expect(mockFetch).toHaveBeenCalledTimes(1);
+      const [url, init] = mockFetch.mock.calls[0]!;
+      expect(url).toBe('https://api.relaycast.dev/v1/channels/general/messages');
+      expect(init.method).toBe('POST');
+      expect(init.body).toBe(JSON.stringify({ text: 'hello', mode: 'wait' }));
+    });
+  });
+
   describe('messages()', () => {
     it('gets from /v1/channels/:name/messages', async () => {
       mockFetch.mockImplementation(() => mockResponse([{ id: 'm_1' }]));

packages/sdk-typescript/src/__tests__/agent-ws.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { AgentClient, type AgentClientOptions } from '../agent.js';
+import { stableRelaycastEventId } from '../event-id.js';
 import { HttpClient } from '../client.js';
 
 class MockWebSocket {
@@ -142,6 +143,30 @@ describe('AgentClient WebSocket integration', () => {
     expect(MockWebSocket.instances).toHaveLength(2);
   });
 
+  it('disconnect() clears manual and managed subscriptions before future reconnect', async () => {
+    const agent = createAgent();
+    agent.connect();
+    const ws1 = MockWebSocket.instances[0]!;
+    ws1.simulateOpen();
+    agent.subscribe(['general']);
+    agent.subscribe(['dev'], vi.fn());
+
+    const p = agent.disconnect();
+    await vi.advanceTimersByTimeAsync(200);
+    await p;
+
+    agent.subscribe(['random'], vi.fn());
+    const ws2 = MockWebSocket.instances[1]!;
+    ws2.simulateOpen();
+
+    const subscribePayloads = ws2.send.mock.calls
+      .map(([payload]) => JSON.parse(String(payload)))
+      .filter((payload) => payload.type === 'subscribe');
+    expect(subscribePayloads).toEqual([
+      { type: 'subscribe', channels: ['random'] },
+    ]);
+  });
+
   // --- typed on.* handlers ---
 
   it('on.messageCreated fires with message.created event', () => {
@@ -248,6 +273,8 @@ describe('AgentClient WebSocket integration', () => {
     agent.connect();
     const ws = MockWebSocket.instances[0]!;
     ws.simulateOpen();
+    agent.subscribe(['dev']);
+    ws.send.mockClear();
 
     agent.unsubscribe(['dev']);
 
@@ -268,6 +295,93 @@ describe('AgentClient WebSocket integration', () => {
     agent.unsubscribe(['general']);
   });
 
+  it('subscribe(channels, handler) multiplexes channel and self DM events on one socket', async () => {
+    const agent = createAgent();
+    const handler = vi.fn();
+
+    const subscription = agent.subscribe(['#general', '@self'], handler);
+
+    expect(MockWebSocket.instances).toHaveLength(1);
+    const ws = MockWebSocket.instances[0]!;
+    ws.simulateOpen();
+
+    expect(ws.send).toHaveBeenCalledWith(
+      JSON.stringify({ type: 'subscribe', channels: ['general'] }),
+    );
+
+    ws.simulateMessage({
+      id: stableRelaycastEventId('m_1'),
+      type: 'message.created',
+      channel: 'general',
+      message: { id: 'm_1', agent_name: 'Bot', text: 'hi', attachments: [] },
+    });
+    ws.simulateMessage({
+      id: stableRelaycastEventId('dm_1'),
+      type: 'dm.received',
+      conversation_id: 'conv_1',
+      message: { id: 'dm_1', agent_name: 'Alice', text: 'hello' },
+    });
+    ws.simulateMessage({
+      id: stableRelaycastEventId('m_2'),
+      type: 'message.created',
+      channel: 'random',
+      message: { id: 'm_2', agent_name: 'Bot', text: 'skip', attachments: [] },
+    });
+    await Promise.resolve();
+
+    expect(handler).toHaveBeenCalledTimes(2);
+    expect(handler.mock.calls[0]![0]).toMatchObject({ id: stableRelaycastEventId('m_1'), type: 'message.created' });
+    expect(handler.mock.calls[1]![0]).toMatchObject({ id: stableRelaycastEventId('dm_1'), type: 'dm.received' });
+
+    subscription.unsubscribe();
+    expect(ws.send).toHaveBeenCalledWith(
+      JSON.stringify({ type: 'unsubscribe', channels: ['general'] }),
+    );
+  });
+
+  it('subscribe(channels, handler) logs rejected handler promises', async () => {
+    const agent = createAgent();
+    const err = new Error('handler failed');
+    const handler = vi.fn().mockRejectedValue(err);
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined);
+
+    agent.subscribe(['general'], handler);
+    const ws = MockWebSocket.instances[0]!;
+    ws.simulateOpen();
+    ws.simulateMessage({
+      type: 'message.created',
+      channel: 'general',
+      message: { id: 'm_1', agent_name: 'Bot', text: 'hi', attachments: [] },
+    });
+
+    for (let i = 0; i < 5; i += 1) {
+      await Promise.resolve();
+    }
+
+    expect(handler).toHaveBeenCalledTimes(1);
+    expect(errorSpy).toHaveBeenCalledWith('[relaycast] Subscription handler failed', err);
+    errorSpy.mockRestore();
+  });
+
+  it('resubscribes managed channels after reconnect', () => {
+    const agent = createAgent();
+    const subscription = agent.subscribe(['general'], vi.fn());
+
+    const ws1 = MockWebSocket.instances[0]!;
+    ws1.simulateOpen();
+    ws1.simulateClose();
+
+    vi.advanceTimersByTime(1000);
+
+    const ws2 = MockWebSocket.instances[1]!;
+    ws2.simulateOpen();
+    expect(ws2.send).toHaveBeenCalledWith(
+      JSON.stringify({ type: 'subscribe', channels: ['general'] }),
+    );
+
+    subscription.unsubscribe();
+  });
+
   // --- lifecycle events ---
 
   it('on.connected fires on WebSocket open', () => {

packages/sdk-typescript/src/__tests__/errors.test.ts

@@ -64,6 +64,22 @@ describe('normalizeRelayErrorCode', () => {
     expect(normalizeRelayErrorCode('agent_not_found')).toBe('not_found');
   });
 
+  it('maps rate_limit_exceeded to rate_limited', () => {
+    expect(normalizeRelayErrorCode('rate_limit_exceeded')).toBe('rate_limited');
+  });
+
+  it('maps backpressure to backpressure', () => {
+    expect(normalizeRelayErrorCode('backpressure')).toBe('backpressure');
+  });
+
+  it('maps queue_overloaded to backpressure', () => {
+    expect(normalizeRelayErrorCode('queue_overloaded')).toBe('backpressure');
+  });
+
+  it('maps workspace_stream_backpressure to backpressure', () => {
+    expect(normalizeRelayErrorCode('workspace_stream_backpressure')).toBe('backpressure');
+  });
+
   it('maps not_found to not_found', () => {
     expect(normalizeRelayErrorCode('not_found')).toBe('not_found');
   });
@@ -92,6 +108,10 @@ describe('normalizeRelayErrorCode', () => {
     expect(normalizeRelayErrorCode('unknown_code', 404)).toBe('not_found');
   });
 
+  it('falls back to statusCode-based mapping for 429', () => {
+    expect(normalizeRelayErrorCode('unknown_code', 429)).toBe('rate_limited');
+  });
+
   it('falls back to statusCode-based mapping for 409', () => {
     expect(normalizeRelayErrorCode('unknown_code', 409)).toBe('name_conflict');
   });
@@ -126,6 +146,14 @@ describe('relayErrorRetryable', () => {
     expect(relayErrorRetryable('unauthorized', 401)).toBe(false);
   });
 
+  it('returns true for rate_limited', () => {
+    expect(relayErrorRetryable('rate_limited', 429)).toBe(true);
+  });
+
+  it('returns true for backpressure', () => {
+    expect(relayErrorRetryable('backpressure', 429)).toBe(true);
+  });
+
   it('returns false for workspace_mismatch', () => {
     expect(relayErrorRetryable('workspace_mismatch', 403)).toBe(false);
   });
@@ -167,6 +195,12 @@ describe('relayErrorFromApi', () => {
     expect(err.code).toBe('not_found');
   });
 
+  it('maps 429 api errors to rate_limited', () => {
+    const err = relayErrorFromApi('rate_limit_exceeded', 'slow down', 429);
+    expect(err.code).toBe('rate_limited');
+    expect(err.retryable).toBe(true);
+  });
+
   it('defaults to transport_error for completely unknown', () => {
     const err = relayErrorFromApi(undefined, 'oops', 500);
     expect(err.code).toBe('transport_error');

packages/sdk-typescript/src/__tests__/event-id.test.ts

@@ -0,0 +1,13 @@
+import { describe, expect, it } from 'vitest';
+import { stableRelaycastEventId } from '../event-id.js';
+
+describe('stableRelaycastEventId', () => {
+  it('preserves part boundaries when hashing', () => {
+    expect(stableRelaycastEventId('a', 'b:c')).not.toBe(stableRelaycastEventId('a:b', 'c'));
+  });
+
+  it('uses the stable empty-event fallback for blank input', () => {
+    expect(stableRelaycastEventId()).toBe(stableRelaycastEventId('empty-event'));
+    expect(stableRelaycastEventId('  ', undefined, null)).toBe(stableRelaycastEventId('empty-event'));
+  });
+});