Add reusable Rust SDK relay helpers

Author: willwashburn

packages/sdk-rust/CHANGELOG.md

@@ -6,6 +6,18 @@ The format is based on Keep a Changelog, and this project follows Semantic Versi
 
 ## [Unreleased]
 
+### Added
+- Added raw WebSocket event subscriptions with `WsClient::subscribe_raw_events()` and `RawEventReceiver`.
+- Added SDK-owned raw event normalization helpers:
+  - `normalize_inbound_event(...)`
+  - `normalize_command_invocation(...)`
+  - `normalize_sender_identity(...)`
+  - `NormalizedInboundEvent`, `NormalizedCommandInvocation`, `NormalizedEventKind`, `SenderKind`, and `RelayPriority`
+- Added identity helpers `agent_name_eq(...)` and `is_self_name(...)`.
+- Added `AgentRegistrationClient::registered_agent_client(...)` for cached/spawned token registration followed by agent-client construction.
+- Added idempotent channel startup helpers `AgentClient::ensure_joined_channel(...)` and `AgentClient::ensure_joined_channels(...)`.
+- Added `DmParticipantsCache` for bounded workspace DM participant lookup caching.
+
 ### Changed
 - Made `agent.spawn_requested` websocket parsing tolerant of missing or `null` `task`, `channel`, and `already_existed` fields.
 - Kept `AgentSpawnRequestedPayload.task` as a `String` for API compatibility by deserializing missing or `null` values to an empty string.

packages/sdk-rust/README.md

@@ -83,9 +83,17 @@ agent.dm("other-agent", "Private message", None).await?;
 agent.create_channel(CreateChannelRequest {
     name: "my-channel".to_string(),
     topic: Some("Channel topic".to_string()),
+    metadata: None,
 }).await?;
 
 agent.join_channel("my-channel").await?;
+
+// Idempotent startup helper: create if needed, then join if needed
+agent.ensure_joined_channel(CreateChannelRequest {
+    name: "general".to_string(),
+    topic: Some("General discussion".to_string()),
+    metadata: None,
+}).await?;
 ```
 
 ### Real-time Events
@@ -121,6 +129,36 @@ while let Ok(event) = events.recv().await {
 }
 ```
 
+For consumers that need to handle evolving event shapes before converting to typed SDK events, subscribe to raw JSON events and normalize them:
+
+```rust
+use relaycast::{normalize_inbound_event, WsClient, WsClientOptions};
+
+let mut ws = WsClient::new(WsClientOptions::new("at_live_xxx"));
+let mut raw_events = ws.subscribe_raw_events();
+ws.connect().await?;
+
+while let Ok(raw) = raw_events.recv().await {
+    if let Some(event) = normalize_inbound_event(&raw) {
+        println!("{} -> {}: {}", event.from, event.target, event.text);
+    }
+}
+```
+
+### Registration Helpers
+
+```rust
+use relaycast::{AgentRegistrationClient, RelayCast, RelayCastOptions};
+
+let relay = RelayCast::new(RelayCastOptions::new("rk_live_xxx"))?;
+let registration = AgentRegistrationClient::new(relay, "codex");
+
+let agent = registration
+    .registered_agent_client("worker-a", Some("codex"))
+    .await?;
+agent.send("#general", "ready", None, None, None).await?;
+```
+
 ### Files
 
 ```rust

packages/sdk-rust/src/agent.rs

@@ -1,7 +1,7 @@
 //! Agent client for message and channel operations.
 
 use crate::client::{ClientOptions, HttpClient, RequestOptions};
-use crate::error::Result;
+use crate::error::{RelayError, Result};
 use crate::types::*;
 use crate::ws::{EventReceiver, LifecycleReceiver, WsClient, WsClientOptions};
 
@@ -34,6 +34,17 @@ impl Default for DmOptions {
     }
 }
 
+/// Result of ensuring a channel exists and is joined.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct EnsureChannelOutcome {
+    /// Channel name that was ensured.
+    pub name: String,
+    /// Whether the channel was created by this call.
+    pub created: bool,
+    /// Whether the agent joined the channel during this call.
+    pub joined: bool,
+}
+
 impl AgentClient {
     /// Create a new agent client with the given token.
     pub fn new(token: impl Into<String>, base_url: Option<String>) -> Result<Self> {
@@ -314,28 +325,42 @@ impl AgentClient {
     // === DMs ===
 
     /// Send a direct message to another agent.
-    pub async fn dm(&self, agent: &str, text: &str, opts: Option<DmOptions>) -> Result<serde_json::Value> {
+    pub async fn dm(
+        &self,
+        agent: &str,
+        text: &str,
+        opts: Option<DmOptions>,
+    ) -> Result<serde_json::Value> {
         let opts = opts.unwrap_or_default();
         let body = SendDmRequest {
             to: agent.to_string(),
             text: text.to_string(),
             attachments: opts.attachments,
             mode: Some(opts.mode),
         };
-        let options = opts.idempotency_key.map(RequestOptions::with_idempotency_key);
+        let options = opts
+            .idempotency_key
+            .map(RequestOptions::with_idempotency_key);
         self.client.post("/v1/dm", Some(body), options).await
     }
 
     /// Send a direct message to another agent (typed response).
-    pub async fn dm_typed(&self, agent: &str, text: &str, opts: Option<DmOptions>) -> Result<DmSendResponse> {
+    pub async fn dm_typed(
+        &self,
+        agent: &str,
+        text: &str,
+        opts: Option<DmOptions>,
+    ) -> Result<DmSendResponse> {
         let opts = opts.unwrap_or_default();
         let body = SendDmRequest {
             to: agent.to_string(),
             text: text.to_string(),
             attachments: opts.attachments,
             mode: Some(opts.mode),
         };
-        let options = opts.idempotency_key.map(RequestOptions::with_idempotency_key);
+        let options = opts
+            .idempotency_key
+            .map(RequestOptions::with_idempotency_key);
         self.client.post("/v1/dm", Some(body), options).await
     }
 
@@ -434,7 +459,10 @@ impl AgentClient {
     ) -> Result<serde_json::Value> {
         let opts = opts.unwrap_or_default();
         let mut body = serde_json::Map::new();
-        body.insert("text".to_string(), serde_json::Value::String(text.to_string()));
+        body.insert(
+            "text".to_string(),
+            serde_json::Value::String(text.to_string()),
+        );
         body.insert(
             "mode".to_string(),
             serde_json::Value::String(match opts.mode {
@@ -443,9 +471,14 @@ impl AgentClient {
             }),
         );
         if let Some(attachments) = opts.attachments {
-            body.insert("attachments".to_string(), serde_json::to_value(attachments)?);
+            body.insert(
+                "attachments".to_string(),
+                serde_json::to_value(attachments)?,
+            );
         }
-        let options = opts.idempotency_key.map(RequestOptions::with_idempotency_key);
+        let options = opts
+            .idempotency_key
+            .map(RequestOptions::with_idempotency_key);
         self.client
             .post(
                 &format!("/v1/dm/{}/messages", urlencoding::encode(conversation_id)),
@@ -464,7 +497,10 @@ impl AgentClient {
     ) -> Result<GroupDmMessageResponse> {
         let opts = opts.unwrap_or_default();
         let mut body = serde_json::Map::new();
-        body.insert("text".to_string(), serde_json::Value::String(text.to_string()));
+        body.insert(
+            "text".to_string(),
+            serde_json::Value::String(text.to_string()),
+        );
         body.insert(
             "mode".to_string(),
             serde_json::Value::String(match opts.mode {
@@ -473,9 +509,14 @@ impl AgentClient {
             }),
         );
         if let Some(attachments) = opts.attachments {
-            body.insert("attachments".to_string(), serde_json::to_value(attachments)?);
+            body.insert(
+                "attachments".to_string(),
+                serde_json::to_value(attachments)?,
+            );
         }
-        let options = opts.idempotency_key.map(RequestOptions::with_idempotency_key);
+        let options = opts
+            .idempotency_key
+            .map(RequestOptions::with_idempotency_key);
         self.client
             .post(
                 &format!("/v1/dm/{}/messages", urlencoding::encode(conversation_id)),
@@ -544,6 +585,46 @@ impl AgentClient {
         self.client.post("/v1/channels", Some(request), None).await
     }
 
+    /// Ensure a channel exists and this agent is a member.
+    ///
+    /// `409 Conflict` from either create or join is treated as success, which
+    /// makes this safe to call during startup.
+    pub async fn ensure_joined_channel(
+        &self,
+        request: CreateChannelRequest,
+    ) -> Result<EnsureChannelOutcome> {
+        let name = request.name.clone();
+        let created = match self.create_channel(request).await {
+            Ok(_) => true,
+            Err(RelayError::Api { status: 409, .. }) => false,
+            Err(error) => return Err(error),
+        };
+
+        let joined = match self.join_channel(&name).await {
+            Ok(_) => true,
+            Err(RelayError::Api { status: 409, .. }) => false,
+            Err(error) => return Err(error),
+        };
+
+        Ok(EnsureChannelOutcome {
+            name,
+            created,
+            joined,
+        })
+    }
+
+    /// Ensure several channels exist and this agent is a member of each.
+    pub async fn ensure_joined_channels<I>(&self, requests: I) -> Result<Vec<EnsureChannelOutcome>>
+    where
+        I: IntoIterator<Item = CreateChannelRequest>,
+    {
+        let mut outcomes = Vec::new();
+        for request in requests {
+            outcomes.push(self.ensure_joined_channel(request).await?);
+        }
+        Ok(outcomes)
+    }
+
     /// List channels.
     pub async fn list_channels(&self, include_archived: bool) -> Result<Vec<Channel>> {
         let query = if include_archived {