websocket impl in progress

Cargo.lock

···
  "smol_str",
  "thiserror 2.0.17",
  "tokio",
+ "tokio-tungstenite-wasm",
  "tracing",
  "trait-variant",
  "url",
···
 ]
 
 [[package]]
+name = "sha1"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
+]
+
+[[package]]
 name = "sha1_smol"
 version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "tokio-tungstenite"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite",
+]
+
+[[package]]
+name = "tokio-tungstenite-wasm"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e21a5c399399c3db9f08d8297ac12b500e86bca82e930253fdc62eaf9c0de6ae"
+dependencies = [
+ "futures-channel",
+ "futures-util",
+ "http",
+ "httparse",
+ "js-sys",
+ "thiserror 1.0.69",
+ "tokio",
+ "tokio-tungstenite",
+ "wasm-bindgen",
+ "web-sys",
+]
+
+[[package]]
 name = "tokio-util"
 version = "0.7.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
 [[package]]
+name = "tungstenite"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a"
+dependencies = [
+ "byteorder",
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.8.5",
+ "sha1",
+ "thiserror 1.0.69",
+ "utf-8",
+]
+
+[[package]]
 name = "twoway"
 version = "0.1.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
+
+[[package]]
+name = "utf-8"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
 
 [[package]]
 name = "utf8_iter"

crates/jacquard-api/src/app_bsky/feed/post.rs

···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Entity<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Post<'a> {
···
 
 #[jacquard_derive::open_union]
 #[derive(
-    serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic,
+    serde::Serialize,
+    serde::Deserialize,
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    jacquard_derive::IntoStatic
 )]
 #[serde(tag = "$type")]
 #[serde(bound(deserialize = "'de: 'a"))]
···
 
 /// Typed wrapper for GetRecord response with this collection's record type.
 #[derive(
-    serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic,
+    serde::Serialize,
+    serde::Deserialize,
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    jacquard_derive::IntoStatic
 )]
 #[serde(rename_all = "camelCase")]
 pub struct PostGetRecordOutput<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct ReplyRef<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct TextSlice<'a> {
     pub end: i64,
     pub start: i64,
-}
+}

crates/jacquard-api/src/com_atproto/label/subscribe_labels.rs

···
 #[serde(bound(deserialize = "'de: 'a"))]
 pub enum SubscribeLabelsMessage<'a> {
     #[serde(rename = "#labels")]
-    Labels(Box<jacquard_common::types::value::Data<'a>>),
+    Labels(Box<crate::com_atproto::label::subscribe_labels::Labels<'a>>),
     #[serde(rename = "#info")]
-    Info(Box<jacquard_common::types::value::Data<'a>>),
+    Info(Box<crate::com_atproto::label::subscribe_labels::Info<'a>>),
 }
 
 #[jacquard_derive::open_union]
···
             Self::Unknown(err) => write!(f, "Unknown error: {:?}", err),
         }
     }
+}
+
+///Stream response type for
+///com.atproto.label.subscribeLabels
+pub struct SubscribeLabelsStream;
+impl jacquard_common::xrpc::SubscriptionResp for SubscribeLabelsStream {
+    const NSID: &'static str = "com.atproto.label.subscribeLabels";
+    const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
+    type Message<'de> = SubscribeLabelsMessage<'de>;
+    type Error<'de> = SubscribeLabelsError<'de>;
+}
+
+impl jacquard_common::xrpc::XrpcSubscription for SubscribeLabels {
+    const NSID: &'static str = "com.atproto.label.subscribeLabels";
+    const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
+    type Stream = SubscribeLabelsStream;
+}
+
+pub struct SubscribeLabelsEndpoint;
+impl jacquard_common::xrpc::SubscriptionEndpoint for SubscribeLabelsEndpoint {
+    const PATH: &'static str = "/xrpc/com.atproto.label.subscribeLabels";
+    const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
+    type Params<'de> = SubscribeLabels;
+    type Stream = SubscribeLabelsStream;
 }

crates/jacquard-api/src/com_atproto/sync/subscribe_repos.rs

···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Account<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Commit<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Identity<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    Default
+    Default,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Info<'a> {
···
     PartialEq,
     Eq,
     bon::Builder,
-    jacquard_derive::IntoStatic
+    jacquard_derive::IntoStatic,
 )]
 #[builder(start_fn = new)]
 #[serde(rename_all = "camelCase")]
···
 
 #[jacquard_derive::open_union]
 #[derive(
-    serde::Serialize,
-    serde::Deserialize,
-    Debug,
-    Clone,
-    PartialEq,
-    Eq,
-    jacquard_derive::IntoStatic
+    serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic,
 )]
 #[serde(tag = "$type")]
 #[serde(bound(deserialize = "'de: 'a"))]
 pub enum SubscribeReposMessage<'a> {
     #[serde(rename = "#commit")]
-    Commit(Box<jacquard_common::types::value::Data<'a>>),
+    Commit(Box<crate::com_atproto::sync::subscribe_repos::Commit<'a>>),
     #[serde(rename = "#sync")]
-    Sync(Box<jacquard_common::types::value::Data<'a>>),
+    Sync(Box<crate::com_atproto::sync::subscribe_repos::Sync<'a>>),
     #[serde(rename = "#identity")]
-    Identity(Box<jacquard_common::types::value::Data<'a>>),
+    Identity(Box<crate::com_atproto::sync::subscribe_repos::Identity<'a>>),
     #[serde(rename = "#account")]
-    Account(Box<jacquard_common::types::value::Data<'a>>),
+    Account(Box<crate::com_atproto::sync::subscribe_repos::Account<'a>>),
     #[serde(rename = "#info")]
-    Info(Box<jacquard_common::types::value::Data<'a>>),
+    Info(Box<crate::com_atproto::sync::subscribe_repos::Info<'a>>),
 }
 
 #[jacquard_derive::open_union]
···
     Eq,
     thiserror::Error,
     miette::Diagnostic,
-    jacquard_derive::IntoStatic
+    jacquard_derive::IntoStatic,
 )]
 #[serde(tag = "error", content = "message")]
 #[serde(bound(deserialize = "'de: 'a"))]
···
     }
 }
 
+///Stream response type for
+///com.atproto.sync.subscribeRepos
+pub struct SubscribeReposStream;
+impl jacquard_common::xrpc::SubscriptionResp for SubscribeReposStream {
+    const NSID: &'static str = "com.atproto.sync.subscribeRepos";
+    const ENCODING: jacquard_common::xrpc::MessageEncoding =
+        jacquard_common::xrpc::MessageEncoding::DagCbor;
+    type Message<'de> = SubscribeReposMessage<'de>;
+    type Error<'de> = SubscribeReposError<'de>;
+}
+
+impl jacquard_common::xrpc::XrpcSubscription for SubscribeRepos {
+    const NSID: &'static str = "com.atproto.sync.subscribeRepos";
+    const ENCODING: jacquard_common::xrpc::MessageEncoding =
+        jacquard_common::xrpc::MessageEncoding::DagCbor;
+    type Stream = SubscribeReposStream;
+}
+
+pub struct SubscribeReposEndpoint;
+impl jacquard_common::xrpc::SubscriptionEndpoint for SubscribeReposEndpoint {
+    const PATH: &'static str = "/xrpc/com.atproto.sync.subscribeRepos";
+    const ENCODING: jacquard_common::xrpc::MessageEncoding =
+        jacquard_common::xrpc::MessageEncoding::DagCbor;
+    type Params<'de> = SubscribeRepos;
+    type Stream = SubscribeReposStream;
+}
+
 /// A repo operation, ie a mutation of a single record.
 #[jacquard_derive::lexicon]
 #[derive(
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct RepoOp<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Sync<'a> {
···
     pub seq: i64,
     /// Timestamp of when this message was originally broadcast.
     pub time: jacquard_common::types::string::Datetime,
-}
+}

crates/jacquard-common/Cargo.toml

···
 # Streaming support (optional)
 n0-future = { version = "0.1", optional = true }
 futures = { version = "0.3", optional = true }
+tokio-tungstenite-wasm = { version = "0.4", optional = true }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
 getrandom = { version = "0.3.4", features = ["wasm_js"] }
···
 reqwest = { workspace = true, optional = true, features = [ "http2", "system-proxy", "rustls-tls"] }
 
 [features]
-default = ["service-auth", "reqwest-client", "crypto"]
+default = ["service-auth", "reqwest-client", "crypto", "websocket"]
 crypto = []
 crypto-ed25519 = ["crypto", "dep:ed25519-dalek"]
 crypto-k256 = ["crypto", "dep:k256", "k256/ecdsa"]
···
 reqwest-client = ["dep:reqwest"]
 tracing = ["dep:tracing"]
 streaming = ["n0-future", "futures"]
-websocket = ["streaming"]
+websocket = ["streaming", "tokio-tungstenite-wasm"]
 
 [dependencies.ed25519-dalek]
 version = "2"

crates/jacquard-common/src/error.rs

···
         #[source]
         serde_ipld_dagcbor::DecodeError<HttpError>,
     ),
+    /// DAG-CBOR deserialization failed (in-memory, e.g., WebSocket frames)
+    #[error("Failed to deserialize DAG-CBOR: {0}")]
+    DagCborInfallible(
+        #[from]
+        #[source]
+        serde_ipld_dagcbor::DecodeError<std::convert::Infallible>,
+    ),
 }
 
 /// HTTP error response (non-200 status codes outside of XRPC error handling)

crates/jacquard-common/src/lib.rs

···
 pub mod websocket;
 
 #[cfg(feature = "websocket")]
-pub use websocket::{WebSocketClient, WebSocketConnection};
+pub use websocket::{
+    tungstenite_client::TungsteniteClient, CloseCode, CloseFrame, WebSocketClient,
+    WebSocketConnection, WsMessage, WsSink, WsStream, WsText,
+};
 
 pub use types::value::*;
 

crates/jacquard-common/src/stream.rs

···
     Closed,
     /// Protocol violation or framing error
     Protocol,
+    /// Message deserialization failed
+    Decode,
+    /// Wrong message format (e.g., text frame when expecting binary)
+    WrongMessageFormat,
 }
 
 impl StreamError {
···
             source: Some(msg.into().into()),
         }
     }
+
+    /// Create a decode error with source
+    pub fn decode(source: impl Error + Send + Sync + 'static) -> Self {
+        Self {
+            kind: StreamErrorKind::Decode,
+            source: Some(Box::new(source)),
+        }
+    }
+
+    /// Create a wrong message format error
+    pub fn wrong_message_format(msg: impl Into<String>) -> Self {
+        Self {
+            kind: StreamErrorKind::WrongMessageFormat,
+            source: Some(msg.into().into()),
+        }
+    }
 }
 
 impl fmt::Display for StreamError {
···
             StreamErrorKind::Transport => write!(f, "Transport error"),
             StreamErrorKind::Closed => write!(f, "Stream closed"),
             StreamErrorKind::Protocol => write!(f, "Protocol error"),
+            StreamErrorKind::Decode => write!(f, "Decode error"),
+            StreamErrorKind::WrongMessageFormat => write!(f, "Wrong message format"),
         }?;
 
         if let Some(source) = &self.source {
···
 
 impl Error for StreamError {
     fn source(&self) -> Option<&(dyn Error + 'static)> {
-        self.source.as_ref().map(|e| e.as_ref() as &(dyn Error + 'static))
+        self.source
+            .as_ref()
+            .map(|e| e.as_ref() as &(dyn Error + 'static))
     }
 }
 
···
     }
 
     /// Convert into the inner boxed stream
-    pub fn into_inner(self) -> Box<dyn n0_future::Stream<Item = Result<Bytes, StreamError>> + Unpin> {
+    pub fn into_inner(
+        self,
+    ) -> Box<dyn n0_future::Stream<Item = Result<Bytes, StreamError>> + Unpin> {
         self.inner
     }
 }
···
     async fn byte_stream_can_be_created() {
         use futures::stream;
 
-        let data = vec![
-            Ok(Bytes::from("hello")),
-            Ok(Bytes::from(" world")),
-        ];
+        let data = vec![Ok(Bytes::from("hello")), Ok(Bytes::from(" world"))];
         let stream = stream::iter(data);
 
         let byte_stream = ByteStream::new(stream);

crates/jacquard-common/src/types/uri.rs

···
 };
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use smol_str::ToSmolStr;
-use std::{fmt::Display, marker::PhantomData, str::FromStr};
+use std::{fmt::Display, marker::PhantomData, ops::Deref, str::FromStr};
 use url::Url;
 
 /// Generic URI with type-specific parsing
···
 impl<R: Collection> Display for RecordUri<'_, R> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         self.0.fmt(f)
+    }
+}
+
+impl<'a, R: Collection> AsRef<AtUri<'a>> for RecordUri<'a, R> {
+    fn as_ref(&self) -> &AtUri<'a> {
+        &self.0
+    }
+}
+
+impl<'a, R: Collection> Deref for RecordUri<'a, R> {
+    type Target = AtUri<'a>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
     }
 }
 

crates/jacquard-common/src/websocket.rs

···
 //! WebSocket client abstraction
 
-use crate::stream::{ByteStream, ByteSink};
+use bytes::Bytes;
+use n0_future::Stream;
+use n0_future::stream::Boxed;
+use std::borrow::Borrow;
+use std::fmt::{self, Display};
 use std::future::Future;
+use std::ops::Deref;
 use url::Url;
 
+use crate::CowStr;
+use crate::stream::StreamError;
+
+/// UTF-8 validated bytes for WebSocket text messages
+#[repr(transparent)]
+#[derive(Debug, Clone, Eq, PartialEq, Hash, PartialOrd, Ord)]
+pub struct WsText(Bytes);
+
+impl WsText {
+    /// Create from static string
+    pub const fn from_static(s: &'static str) -> Self {
+        Self(Bytes::from_static(s.as_bytes()))
+    }
+
+    /// Get as string slice
+    pub fn as_str(&self) -> &str {
+        unsafe { std::str::from_utf8_unchecked(&self.0) }
+    }
+
+    /// Create from bytes without validation (caller must ensure UTF-8)
+    ///
+    /// # Safety
+    /// Bytes must be valid UTF-8
+    pub unsafe fn from_bytes_unchecked(bytes: Bytes) -> Self {
+        Self(bytes)
+    }
+
+    /// Convert into underlying bytes
+    pub fn into_bytes(self) -> Bytes {
+        self.0
+    }
+}
+
+impl Deref for WsText {
+    type Target = str;
+    fn deref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl AsRef<str> for WsText {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl AsRef<[u8]> for WsText {
+    fn as_ref(&self) -> &[u8] {
+        &self.0
+    }
+}
+
+impl AsRef<Bytes> for WsText {
+    fn as_ref(&self) -> &Bytes {
+        &self.0
+    }
+}
+
+impl Borrow<str> for WsText {
+    fn borrow(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl Display for WsText {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        Display::fmt(self.as_str(), f)
+    }
+}
+
+impl From<String> for WsText {
+    fn from(s: String) -> Self {
+        Self(Bytes::from(s))
+    }
+}
+
+impl From<&str> for WsText {
+    fn from(s: &str) -> Self {
+        Self(Bytes::copy_from_slice(s.as_bytes()))
+    }
+}
+
+impl From<&String> for WsText {
+    fn from(s: &String) -> Self {
+        Self::from(s.as_str())
+    }
+}
+
+impl TryFrom<Bytes> for WsText {
+    type Error = std::str::Utf8Error;
+    fn try_from(bytes: Bytes) -> Result<Self, Self::Error> {
+        std::str::from_utf8(&bytes)?;
+        Ok(Self(bytes))
+    }
+}
+
+impl TryFrom<Vec<u8>> for WsText {
+    type Error = std::str::Utf8Error;
+    fn try_from(vec: Vec<u8>) -> Result<Self, Self::Error> {
+        Self::try_from(Bytes::from(vec))
+    }
+}
+
+impl From<WsText> for Bytes {
+    fn from(t: WsText) -> Bytes {
+        t.0
+    }
+}
+
+impl Default for WsText {
+    fn default() -> Self {
+        Self(Bytes::new())
+    }
+}
+
+/// WebSocket close code
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[repr(u16)]
+pub enum CloseCode {
+    /// Normal closure
+    Normal = 1000,
+    /// Endpoint going away
+    Away = 1001,
+    /// Protocol error
+    Protocol = 1002,
+    /// Unsupported data
+    Unsupported = 1003,
+    /// Invalid frame payload data
+    Invalid = 1007,
+    /// Policy violation
+    Policy = 1008,
+    /// Message too big
+    Size = 1009,
+    /// Extension negotiation failure
+    Extension = 1010,
+    /// Unexpected condition
+    Error = 1011,
+    /// TLS handshake failure
+    Tls = 1015,
+    /// Other code
+    Other(u16),
+}
+
+impl From<u16> for CloseCode {
+    fn from(code: u16) -> Self {
+        match code {
+            1000 => CloseCode::Normal,
+            1001 => CloseCode::Away,
+            1002 => CloseCode::Protocol,
+            1003 => CloseCode::Unsupported,
+            1007 => CloseCode::Invalid,
+            1008 => CloseCode::Policy,
+            1009 => CloseCode::Size,
+            1010 => CloseCode::Extension,
+            1011 => CloseCode::Error,
+            1015 => CloseCode::Tls,
+            other => CloseCode::Other(other),
+        }
+    }
+}
+
+impl From<CloseCode> for u16 {
+    fn from(code: CloseCode) -> u16 {
+        match code {
+            CloseCode::Normal => 1000,
+            CloseCode::Away => 1001,
+            CloseCode::Protocol => 1002,
+            CloseCode::Unsupported => 1003,
+            CloseCode::Invalid => 1007,
+            CloseCode::Policy => 1008,
+            CloseCode::Size => 1009,
+            CloseCode::Extension => 1010,
+            CloseCode::Error => 1011,
+            CloseCode::Tls => 1015,
+            CloseCode::Other(code) => code,
+        }
+    }
+}
+
+/// WebSocket close frame
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CloseFrame<'a> {
+    /// Close code
+    pub code: CloseCode,
+    /// Close reason text
+    pub reason: CowStr<'a>,
+}
+
+impl<'a> CloseFrame<'a> {
+    /// Create a new close frame
+    pub fn new(code: CloseCode, reason: impl Into<CowStr<'a>>) -> Self {
+        Self {
+            code,
+            reason: reason.into(),
+        }
+    }
+}
+
+/// WebSocket message
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum WsMessage {
+    /// Text message (UTF-8)
+    Text(WsText),
+    /// Binary message
+    Binary(Bytes),
+    /// Close frame
+    Close(Option<CloseFrame<'static>>),
+}
+
+impl WsMessage {
+    /// Check if this is a text message
+    pub fn is_text(&self) -> bool {
+        matches!(self, WsMessage::Text(_))
+    }
+
+    /// Check if this is a binary message
+    pub fn is_binary(&self) -> bool {
+        matches!(self, WsMessage::Binary(_))
+    }
+
+    /// Check if this is a close message
+    pub fn is_close(&self) -> bool {
+        matches!(self, WsMessage::Close(_))
+    }
+
+    /// Get as text, if this is a text message
+    pub fn as_text(&self) -> Option<&str> {
+        match self {
+            WsMessage::Text(t) => Some(t.as_str()),
+            _ => None,
+        }
+    }
+
+    /// Get as bytes
+    pub fn as_bytes(&self) -> Option<&[u8]> {
+        match self {
+            WsMessage::Text(t) => Some(t.as_ref()),
+            WsMessage::Binary(b) => Some(b),
+            WsMessage::Close(_) => None,
+        }
+    }
+}
+
+impl From<WsText> for WsMessage {
+    fn from(text: WsText) -> Self {
+        WsMessage::Text(text)
+    }
+}
+
+impl From<String> for WsMessage {
+    fn from(s: String) -> Self {
+        WsMessage::Text(WsText::from(s))
+    }
+}
+
+impl From<&str> for WsMessage {
+    fn from(s: &str) -> Self {
+        WsMessage::Text(WsText::from(s))
+    }
+}
+
+impl From<Bytes> for WsMessage {
+    fn from(bytes: Bytes) -> Self {
+        WsMessage::Binary(bytes)
+    }
+}
+
+impl From<Vec<u8>> for WsMessage {
+    fn from(vec: Vec<u8>) -> Self {
+        WsMessage::Binary(Bytes::from(vec))
+    }
+}
+
+/// WebSocket message stream
+pub struct WsStream(Boxed<Result<WsMessage, StreamError>>);
+
+impl WsStream {
+    /// Create a new message stream
+    #[cfg(not(target_arch = "wasm32"))]
+    pub fn new<S>(stream: S) -> Self
+    where
+        S: Stream<Item = Result<WsMessage, StreamError>> + Send + 'static,
+    {
+        Self(Box::pin(stream))
+    }
+
+    /// Create a new message stream
+    #[cfg(target_arch = "wasm32")]
+    pub fn new<S>(stream: S) -> Self
+    where
+        S: Stream<Item = Result<WsMessage, StreamError>> + 'static,
+    {
+        Self(Box::pin(stream))
+    }
+
+    /// Convert into the inner pinned boxed stream
+    pub fn into_inner(self) -> Boxed<Result<WsMessage, StreamError>> {
+        self.0
+    }
+}
+
+/// Extension trait for decoding typed messages from WebSocket streams
+pub trait WsStreamExt: Sized {
+    /// Decode JSON text/binary frames into typed messages
+    ///
+    /// Deserializes messages but does not automatically convert to owned.
+    /// The caller is responsible for calling `.into_static()` if needed.
+    fn decode_json<T>(self) -> impl Stream<Item = Result<T, StreamError>>
+    where
+        T: for<'de> serde::Deserialize<'de>;
+
+    /// Decode DAG-CBOR binary frames into typed messages
+    ///
+    /// Deserializes messages but does not automatically convert to owned.
+    /// The caller is responsible for calling `.into_static()` if needed.
+    fn decode_cbor<T>(self) -> impl Stream<Item = Result<T, StreamError>>
+    where
+        T: for<'de> serde::Deserialize<'de>;
+}
+
+impl WsStreamExt for WsStream {
+    fn decode_json<T>(self) -> impl Stream<Item = Result<T, StreamError>>
+    where
+        T: for<'de> serde::Deserialize<'de>,
+    {
+        use n0_future::StreamExt as _;
+
+        Box::pin(self.into_inner().filter_map(|msg_result| match msg_result {
+            Ok(WsMessage::Text(text)) => {
+                Some(serde_json::from_slice(text.as_ref()).map_err(StreamError::decode))
+            }
+            Ok(WsMessage::Binary(bytes)) => {
+                Some(serde_json::from_slice(&bytes).map_err(StreamError::decode))
+            }
+            Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())),
+            Err(e) => Some(Err(e)),
+        }))
+    }
+
+    fn decode_cbor<T>(self) -> impl Stream<Item = Result<T, StreamError>>
+    where
+        T: for<'de> serde::Deserialize<'de>,
+    {
+        use n0_future::StreamExt as _;
+
+        Box::pin(self.into_inner().filter_map(|msg_result| {
+            match msg_result {
+                Ok(WsMessage::Binary(bytes)) => Some(
+                    serde_ipld_dagcbor::from_slice(&bytes)
+                        .map_err(|e| StreamError::decode(crate::error::DecodeError::from(e))),
+                ),
+                Ok(WsMessage::Text(_)) => Some(Err(StreamError::wrong_message_format(
+                    "expected binary frame for CBOR, got text",
+                ))),
+                Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())),
+                Err(e) => Some(Err(e)),
+            }
+        }))
+    }
+}
+
+impl fmt::Debug for WsStream {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("WsStream").finish_non_exhaustive()
+    }
+}
+
+/// WebSocket message sink
+pub struct WsSink(Box<dyn n0_future::Sink<WsMessage, Error = StreamError>>);
+
+impl WsSink {
+    /// Create a new message sink
+    pub fn new<S>(sink: S) -> Self
+    where
+        S: n0_future::Sink<WsMessage, Error = StreamError> + 'static,
+    {
+        Self(Box::new(sink))
+    }
+
+    /// Convert into the inner boxed sink
+    pub fn into_inner(self) -> Box<dyn n0_future::Sink<WsMessage, Error = StreamError>> {
+        self.0
+    }
+}
+
+impl fmt::Debug for WsSink {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("WsSink").finish_non_exhaustive()
+    }
+}
+
 /// WebSocket client trait
 #[cfg_attr(not(target_arch = "wasm32"), trait_variant::make(Send))]
 pub trait WebSocketClient {
···
     type Error: std::error::Error + Send + Sync + 'static;
 
     /// Connect to a WebSocket endpoint
-    fn connect(
-        &self,
-        url: Url,
-    ) -> impl Future<Output = Result<WebSocketConnection, Self::Error>>;
+    fn connect(&self, url: Url) -> impl Future<Output = Result<WebSocketConnection, Self::Error>>;
 }
 
 /// WebSocket connection with bidirectional streams
 pub struct WebSocketConnection {
-    tx: ByteSink,
-    rx: ByteStream,
+    tx: WsSink,
+    rx: WsStream,
 }
 
 impl WebSocketConnection {
     /// Create a new WebSocket connection
-    pub fn new(tx: ByteSink, rx: ByteStream) -> Self {
+    pub fn new(tx: WsSink, rx: WsStream) -> Self {
         Self { tx, rx }
     }
 
     /// Get mutable access to the sender
-    pub fn sender_mut(&mut self) -> &mut ByteSink {
+    pub fn sender_mut(&mut self) -> &mut WsSink {
         &mut self.tx
     }
 
     /// Get mutable access to the receiver
-    pub fn receiver_mut(&mut self) -> &mut ByteStream {
+    pub fn receiver_mut(&mut self) -> &mut WsStream {
         &mut self.rx
     }
 
+    /// Get a reference to the receiver
+    pub fn receiver(&self) -> &WsStream {
+        &self.rx
+    }
+
+    /// Get a reference to the sender
+    pub fn sender(&self) -> &WsSink {
+        &self.tx
+    }
+
     /// Split into sender and receiver
-    pub fn split(self) -> (ByteSink, ByteStream) {
+    pub fn split(self) -> (WsSink, WsStream) {
         (self.tx, self.rx)
     }
 
···
     }
 }
 
-impl std::fmt::Debug for WebSocketConnection {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl fmt::Debug for WebSocketConnection {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("WebSocketConnection")
             .finish_non_exhaustive()
     }
 }
 
+/// Concrete WebSocket client implementation using tokio-tungstenite-wasm
+pub mod tungstenite_client {
+    use super::*;
+    use crate::IntoStatic;
+    use futures::{SinkExt, StreamExt};
+
+    /// WebSocket client backed by tokio-tungstenite-wasm
+    #[derive(Debug, Clone, Default)]
+    pub struct TungsteniteClient;
+
+    impl TungsteniteClient {
+        /// Create a new tungstenite WebSocket client
+        pub fn new() -> Self {
+            Self
+        }
+    }
+
+    impl WebSocketClient for TungsteniteClient {
+        type Error = tokio_tungstenite_wasm::Error;
+
+        async fn connect(&self, url: Url) -> Result<WebSocketConnection, Self::Error> {
+            let ws_stream = tokio_tungstenite_wasm::connect(url.as_str()).await?;
+
+            let (sink, stream) = ws_stream.split();
+
+            // Convert tungstenite messages to our WsMessage
+            let rx_stream = stream.filter_map(|result| async move {
+                match result {
+                    Ok(msg) => match convert_message(msg) {
+                        Some(ws_msg) => Some(Ok(ws_msg)),
+                        None => None, // Skip ping/pong
+                    },
+                    Err(e) => Some(Err(StreamError::transport(e))),
+                }
+            });
+
+            let rx = WsStream::new(rx_stream);
+
+            // Convert our WsMessage to tungstenite messages
+            let tx_sink = sink.with(|msg: WsMessage| async move {
+                Ok::<_, tokio_tungstenite_wasm::Error>(msg.into())
+            });
+
+            let tx_sink_mapped = tx_sink.sink_map_err(|e| StreamError::transport(e));
+            let tx = WsSink::new(tx_sink_mapped);
+
+            Ok(WebSocketConnection::new(tx, rx))
+        }
+    }
+
+    /// Convert tokio-tungstenite-wasm Message to our WsMessage
+    /// Returns None for Ping/Pong which we auto-handle
+    fn convert_message(msg: tokio_tungstenite_wasm::Message) -> Option<WsMessage> {
+        use tokio_tungstenite_wasm::Message;
+
+        match msg {
+            Message::Text(vec) => {
+                // tokio-tungstenite-wasm Text contains Vec<u8> (UTF-8 validated)
+                let bytes = Bytes::from(vec);
+                Some(WsMessage::Text(unsafe {
+                    WsText::from_bytes_unchecked(bytes)
+                }))
+            }
+            Message::Binary(vec) => Some(WsMessage::Binary(Bytes::from(vec))),
+            Message::Close(frame) => {
+                let close_frame = frame.map(|f| {
+                    let code = convert_close_code(f.code);
+                    CloseFrame::new(code, CowStr::from(f.reason.into_owned()))
+                });
+                Some(WsMessage::Close(close_frame))
+            }
+        }
+    }
+
+    /// Convert tokio-tungstenite-wasm CloseCode to our CloseCode
+    fn convert_close_code(code: tokio_tungstenite_wasm::CloseCode) -> CloseCode {
+        use tokio_tungstenite_wasm::CloseCode as TungsteniteCode;
+
+        match code {
+            TungsteniteCode::Normal => CloseCode::Normal,
+            TungsteniteCode::Away => CloseCode::Away,
+            TungsteniteCode::Protocol => CloseCode::Protocol,
+            TungsteniteCode::Unsupported => CloseCode::Unsupported,
+            TungsteniteCode::Invalid => CloseCode::Invalid,
+            TungsteniteCode::Policy => CloseCode::Policy,
+            TungsteniteCode::Size => CloseCode::Size,
+            TungsteniteCode::Extension => CloseCode::Extension,
+            TungsteniteCode::Error => CloseCode::Error,
+            TungsteniteCode::Tls => CloseCode::Tls,
+            // For other variants, extract raw code
+            other => {
+                let raw: u16 = other.into();
+                CloseCode::from(raw)
+            }
+        }
+    }
+
+    impl From<WsMessage> for tokio_tungstenite_wasm::Message {
+        fn from(msg: WsMessage) -> Self {
+            use tokio_tungstenite_wasm::Message;
+
+            match msg {
+                WsMessage::Text(text) => {
+                    // tokio-tungstenite-wasm Text expects String
+                    let bytes = text.into_bytes();
+                    // Safe: WsText is already UTF-8 validated
+                    let string = unsafe { String::from_utf8_unchecked(bytes.to_vec()) };
+                    Message::Text(string)
+                }
+                WsMessage::Binary(bytes) => Message::Binary(bytes.to_vec()),
+                WsMessage::Close(frame) => {
+                    let close_frame = frame.map(|f| {
+                        let code = u16::from(f.code).into();
+                        tokio_tungstenite_wasm::CloseFrame {
+                            code,
+                            reason: f.reason.into_static().to_string().into(),
+                        }
+                    });
+                    Message::Close(close_frame)
+                }
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::stream::StreamError;
+
+    #[test]
+    fn ws_text_from_string() {
+        let text = WsText::from("hello");
+        assert_eq!(text.as_str(), "hello");
+    }
+
+    #[test]
+    fn ws_text_deref() {
+        let text = WsText::from(String::from("world"));
+        assert_eq!(&*text, "world");
+    }
+
+    #[test]
+    fn ws_text_try_from_bytes() {
+        let bytes = Bytes::from("test");
+        let text = WsText::try_from(bytes).unwrap();
+        assert_eq!(text.as_str(), "test");
+    }
+
+    #[test]
+    fn ws_text_invalid_utf8() {
+        let bytes = Bytes::from(vec![0xFF, 0xFE]);
+        assert!(WsText::try_from(bytes).is_err());
+    }
+
+    #[test]
+    fn ws_message_text() {
+        let msg = WsMessage::from("hello");
+        assert!(msg.is_text());
+        assert_eq!(msg.as_text(), Some("hello"));
+    }
+
+    #[test]
+    fn ws_message_binary() {
+        let msg = WsMessage::from(vec![1, 2, 3]);
+        assert!(msg.is_binary());
+        assert_eq!(msg.as_bytes(), Some(&[1u8, 2, 3][..]));
+    }
+
+    #[test]
+    fn close_code_conversion() {
+        assert_eq!(u16::from(CloseCode::Normal), 1000);
+        assert_eq!(CloseCode::from(1000), CloseCode::Normal);
+        assert_eq!(CloseCode::from(9999), CloseCode::Other(9999));
+    }
 
     #[test]
     fn websocket_connection_has_tx_and_rx() {
-        use futures::stream;
         use futures::sink::SinkExt;
-        use bytes::Bytes;
+        use futures::stream;
 
-        let rx_stream = stream::iter(vec![Ok(Bytes::from("test"))]);
-        let rx = ByteStream::new(rx_stream);
+        let rx_stream = stream::iter(vec![Ok(WsMessage::from("test"))]);
+        let rx = WsStream::new(rx_stream);
 
-        // Create a sink that converts Infallible to StreamError
-        let drain_sink = futures::sink::drain().sink_map_err(|_: std::convert::Infallible| {
-            StreamError::closed()
-        });
-        let tx = ByteSink::new(drain_sink);
+        let drain_sink = futures::sink::drain()
+            .sink_map_err(|_: std::convert::Infallible| StreamError::closed());
+        let tx = WsSink::new(drain_sink);
 
         let conn = WebSocketConnection::new(tx, rx);
         assert!(conn.is_open());

crates/jacquard-common/src/xrpc.rs

···
 #[cfg(feature = "streaming")]
 pub use streaming::StreamingResponse;
 
+#[cfg(feature = "websocket")]
+pub mod subscription;
+
+#[cfg(feature = "websocket")]
+pub use subscription::{MessageEncoding, SubscriptionEndpoint, SubscriptionResp, XrpcSubscription};
+
 use bytes::Bytes;
 use http::{
     HeaderName, HeaderValue, Request, StatusCode,

crates/jacquard-common/src/xrpc/subscription.rs

···
+//! WebSocket subscription support for XRPC
+//!
+//! This module defines traits and types for typed WebSocket subscriptions,
+//! mirroring the request/response pattern used for HTTP XRPC endpoints.
+
+use serde::{Deserialize, Serialize};
+use std::error::Error;
+
+use crate::IntoStatic;
+
+/// Encoding format for subscription messages
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MessageEncoding {
+    /// JSON text frames
+    Json,
+    /// DAG-CBOR binary frames
+    DagCbor,
+}
+
+/// XRPC subscription stream response trait
+///
+/// Analogous to `XrpcResp` but for WebSocket subscriptions.
+/// Defines the message and error types for a subscription stream.
+///
+/// This trait is implemented on a marker struct to keep it lifetime-free
+/// while using GATs for the message/error types.
+pub trait SubscriptionResp {
+    /// The NSID for this subscription
+    const NSID: &'static str;
+
+    /// Message encoding (JSON or DAG-CBOR)
+    const ENCODING: MessageEncoding;
+
+    /// Message union type
+    type Message<'de>: Deserialize<'de> + IntoStatic;
+
+    /// Error union type
+    type Error<'de>: Error + Deserialize<'de> + IntoStatic;
+}
+
+/// XRPC subscription (WebSocket)
+///
+/// This trait is analogous to `XrpcRequest` but for WebSocket subscriptions.
+/// It defines the NSID and associated stream response type.
+///
+/// The trait is implemented on the subscription parameters type.
+pub trait XrpcSubscription: Serialize {
+    /// The NSID for this XRPC subscription
+    const NSID: &'static str;
+
+    /// Message encoding (JSON or DAG-CBOR)
+    const ENCODING: MessageEncoding;
+
+    /// Stream response type (marker struct)
+    type Stream: SubscriptionResp;
+
+    /// Encode query params for WebSocket URL
+    ///
+    /// Default implementation uses serde_html_form to encode the struct as query parameters.
+    fn query_params(&self) -> Vec<(String, String)> {
+        // Default: use serde_html_form to encode self
+        serde_html_form::to_string(self)
+            .ok()
+            .map(|s| {
+                s.split('&')
+                    .filter_map(|pair| {
+                        let mut parts = pair.splitn(2, '=');
+                        Some((parts.next()?.to_string(), parts.next()?.to_string()))
+                    })
+                    .collect()
+            })
+            .unwrap_or_default()
+    }
+}
+
+/// XRPC subscription endpoint trait (server-side)
+///
+/// Analogous to `XrpcEndpoint` but for WebSocket subscriptions.
+/// Defines the fully-qualified path and associated parameter/stream types.
+///
+/// This exists primarily for server-side frameworks (like Axum) to extract
+/// typed subscription parameters without lifetime issues.
+pub trait SubscriptionEndpoint {
+    /// Fully-qualified path ('/xrpc/[nsid]') where this subscription endpoint lives
+    const PATH: &'static str;
+
+    /// Message encoding (JSON or DAG-CBOR)
+    const ENCODING: MessageEncoding;
+
+    /// Subscription parameters type
+    type Params<'de>: XrpcSubscription + Deserialize<'de> + IntoStatic;
+
+    /// Stream response type
+    type Stream: SubscriptionResp;
+}

crates/jacquard-lexicon/src/codegen/xrpc.rs

···
 use proc_macro2::TokenStream;
 use quote::quote;
 
-use super::utils::make_ident;
 use super::CodeGenerator;
+use super::utils::make_ident;
 
 impl<'c> CodeGenerator<'c> {
     /// Generate query type
···
             output.push(error_enum);
         }
 
+        // Generate XrpcSubscription trait impl
+        let params_has_lifetime = sub
+            .parameters
+            .as_ref()
+            .map(|p| match p {
+                crate::lexicon::LexXrpcSubscriptionParameter::Params(params) => {
+                    self.params_need_lifetime(params)
+                }
+            })
+            .unwrap_or(false);
+
+        let has_params = sub.parameters.is_some();
+        let has_message = sub.message.is_some();
+        let has_errors = sub.errors.is_some();
+
+        let subscription_impl = self.generate_xrpc_subscription_impl(
+            nsid,
+            &type_base,
+            has_params,
+            params_has_lifetime,
+            has_message,
+            has_errors,
+        )?;
+        output.push(subscription_impl);
+
         Ok(quote! {
             #(#output)*
         })
···
                 let mut variants = Vec::new();
                 for ref_str in &union.refs {
                     let ref_str_s = ref_str.as_ref();
+
+                    // Normalize local refs (starting with #) by prepending current NSID
+                    let normalized_ref = if ref_str.starts_with('#') {
+                        format!("{}{}", nsid, ref_str)
+                    } else {
+                        ref_str.to_string()
+                    };
+
                     // Parse ref to get NSID and def name
                     let (ref_nsid, ref_def) =
-                        if let Some((nsid, fragment)) = ref_str.split_once('#') {
-                            (nsid, fragment)
+                        if let Some((nsid_part, fragment)) = normalized_ref.split_once('#') {
+                            (nsid_part, fragment)
                         } else {
-                            (ref_str.as_ref(), "main")
+                            (normalized_ref.as_str(), "main")
                         };
 
                     let variant_name = if ref_def == "main" {
···
                     };
                     let variant_ident =
                         syn::Ident::new(&variant_name, proc_macro2::Span::call_site());
-                    let type_path = self.ref_to_rust_type(ref_str)?;
+                    let type_path = self.ref_to_rust_type(&normalized_ref)?;
 
                     variants.push(quote! {
                         #[serde(rename = #ref_str_s)]
···
                     match field_type {
                         LexObjectProperty::Union(union) => {
                             // Skip empty, single-variant unions unless they're self-referential
-                            if !union.refs.is_empty() && (union.refs.len() > 1 || self.is_self_referential_union(nsid, &struct_name, union)) {
-                                let union_name = self.generate_field_type_name(nsid, &struct_name, field_name, "");
+                            if !union.refs.is_empty()
+                                && (union.refs.len() > 1
+                                    || self.is_self_referential_union(nsid, &struct_name, union))
+                            {
+                                let union_name = self.generate_field_type_name(
+                                    nsid,
+                                    &struct_name,
+                                    field_name,
+                                    "",
+                                );
                                 let refs: Vec<_> = union.refs.iter().cloned().collect();
-                                let union_def =
-                                    self.generate_union(nsid, &union_name, &refs, None, union.closed)?;
+                                let union_def = self.generate_union(
+                                    nsid,
+                                    &union_name,
+                                    &refs,
+                                    None,
+                                    union.closed,
+                                )?;
                                 unions.push(union_def);
                             }
                         }
···
                             if let LexArrayItem::Union(union) = &array.items {
                                 // Skip single-variant array unions
                                 if union.refs.len() > 1 {
-                                    let union_name = self.generate_field_type_name(nsid, &struct_name, field_name, "Item");
+                                    let union_name = self.generate_field_type_name(
+                                        nsid,
+                                        &struct_name,
+                                        field_name,
+                                        "Item",
+                                    );
                                     let refs: Vec<_> = union.refs.iter().cloned().collect();
-                                    let union_def = self.generate_union(nsid, &union_name, &refs, None, union.closed)?;
+                                    let union_def = self.generate_union(
+                                        nsid,
+                                        &union_name,
+                                        &refs,
+                                        None,
+                                        union.closed,
+                                    )?;
                                     unions.push(union_def);
                                 }
                             }
···
         let (has_default, has_builder) = if is_binary_body {
             (false, true)
         } else if let Some(crate::lexicon::LexXrpcBodySchema::Object(obj)) = &body.schema {
-            use crate::codegen::structs::{count_required_fields, all_required_are_defaultable_strings, conflicts_with_builder_macro};
+            use crate::codegen::structs::{
+                all_required_are_defaultable_strings, conflicts_with_builder_macro,
+                count_required_fields,
+            };
             let required_count = count_required_fields(obj);
             let can_default = required_count == 0 || all_required_are_defaultable_strings(obj);
-            let can_builder = required_count >= 1 && !can_default && !conflicts_with_builder_macro(type_base);
+            let can_builder =
+                required_count >= 1 && !can_default && !conflicts_with_builder_macro(type_base);
             (can_default, can_builder)
         } else {
             (false, false)
···
                 match field_type {
                     LexObjectProperty::Union(union) => {
                         // Skip empty, single-variant unions unless they're self-referential
-                        if !union.refs.is_empty() && (union.refs.len() > 1 || self.is_self_referential_union(nsid, type_base, union)) {
-                            let union_name = self.generate_field_type_name(nsid, type_base, field_name, "");
+                        if !union.refs.is_empty()
+                            && (union.refs.len() > 1
+                                || self.is_self_referential_union(nsid, type_base, union))
+                        {
+                            let union_name =
+                                self.generate_field_type_name(nsid, type_base, field_name, "");
                             let refs: Vec<_> = union.refs.iter().cloned().collect();
                             let union_def =
                                 self.generate_union(nsid, &union_name, &refs, None, union.closed)?;
···
                         if let LexArrayItem::Union(union) = &array.items {
                             // Skip single-variant array unions
                             if union.refs.len() > 1 {
-                                let union_name = self.generate_field_type_name(nsid, type_base, field_name, "Item");
+                                let union_name = self
+                                    .generate_field_type_name(nsid, type_base, field_name, "Item");
                                 let refs: Vec<_> = union.refs.iter().cloned().collect();
-                                let union_def = self.generate_union(nsid, &union_name, &refs, None, union.closed)?;
+                                let union_def = self.generate_union(
+                                    nsid,
+                                    &union_name,
+                                    &refs,
+                                    None,
+                                    union.closed,
+                                )?;
                                 unions.push(union_def);
                             }
                         }
···
 
         // Determine if we should derive Default
         // Check if schema is an Object and apply heuristics
-        let has_default = if let Some(crate::lexicon::LexXrpcBodySchema::Object(obj)) = &body.schema {
-            use crate::codegen::structs::{count_required_fields, all_required_are_defaultable_strings};
+        let has_default = if let Some(crate::lexicon::LexXrpcBodySchema::Object(obj)) = &body.schema
+        {
+            use crate::codegen::structs::{
+                all_required_are_defaultable_strings, count_required_fields,
+            };
             let required_count = count_required_fields(obj);
             required_count == 0 || all_required_are_defaultable_strings(obj)
         } else {
···
                 match field_type {
                     LexObjectProperty::Union(union) => {
                         // Skip single-variant unions unless they're self-referential
-                        if union.refs.len() > 1 || self.is_self_referential_union(nsid, &struct_name, union) {
-                            let union_name = self.generate_field_type_name(nsid, &struct_name, field_name, "");
+                        if union.refs.len() > 1
+                            || self.is_self_referential_union(nsid, &struct_name, union)
+                        {
+                            let union_name =
+                                self.generate_field_type_name(nsid, &struct_name, field_name, "");
                             let refs: Vec<_> = union.refs.iter().cloned().collect();
                             let union_def =
                                 self.generate_union(nsid, &union_name, &refs, None, union.closed)?;
···
                         if let LexArrayItem::Union(union) = &array.items {
                             // Skip single-variant array unions
                             if union.refs.len() > 1 {
-                                let union_name = self.generate_field_type_name(nsid, &struct_name, field_name, "Item");
+                                let union_name = self.generate_field_type_name(
+                                    nsid,
+                                    &struct_name,
+                                    field_name,
+                                    "Item",
+                                );
                                 let refs: Vec<_> = union.refs.iter().cloned().collect();
-                                let union_def = self.generate_union(nsid, &union_name, &refs, None, union.closed)?;
+                                let union_def = self.generate_union(
+                                    nsid,
+                                    &union_name,
+                                    &refs,
+                                    None,
+                                    union.closed,
+                                )?;
                                 unions.push(union_def);
                             }
                         }
···
         );
 
         let response_type = quote! {
-            #[doc = "Response type for "]
+            #[doc = " Response type for "]
             #[doc = #nsid]
             pub struct #response_ident;
 
···
                     #decode_body_method
                 }
 
-                #[doc = "Endpoint type for "]
+                #[doc = " Endpoint type for "]
                 #[doc = #nsid]
                 pub struct #endpoint_ident;
 
···
                     type Response = #response_ident;
                 }
 
-                #[doc = "Endpoint type for "]
+                #[doc = " Endpoint type for "]
                 #[doc = #nsid]
                 pub struct #endpoint_ident;
 
···
                 }
             })
         }
+    }
+
+    /// Generate XrpcSubscription trait impl for a subscription endpoint
+    pub(super) fn generate_xrpc_subscription_impl(
+        &self,
+        nsid: &str,
+        type_base: &str,
+        has_params: bool,
+        params_has_lifetime: bool,
+        has_message: bool,
+        has_errors: bool,
+    ) -> Result<TokenStream> {
+        // Generate stream response marker struct
+        let stream_ident = syn::Ident::new(
+            &format!("{}Stream", type_base),
+            proc_macro2::Span::call_site(),
+        );
+
+        let message_type = if has_message {
+            let msg_ident = syn::Ident::new(
+                &format!("{}Message", type_base),
+                proc_macro2::Span::call_site(),
+            );
+            quote! { #msg_ident<'de> }
+        } else {
+            quote! { () }
+        };
+
+        let error_type = if has_errors {
+            let err_ident = syn::Ident::new(
+                &format!("{}Error", type_base),
+                proc_macro2::Span::call_site(),
+            );
+            quote! { #err_ident<'de> }
+        } else {
+            quote! { jacquard_common::xrpc::GenericError<'de> }
+        };
+
+        // Determine encoding from nsid convention
+        // ATProto subscriptions use DAG-CBOR, community ones might use JSON
+        let encoding = if nsid.starts_with("com.atproto") {
+            quote! { jacquard_common::xrpc::MessageEncoding::DagCbor }
+        } else {
+            quote! { jacquard_common::xrpc::MessageEncoding::Json }
+        };
+
+        // Generate SubscriptionResp impl
+        let stream_resp_impl = quote! {
+            #[doc = "Stream response type for "]
+            #[doc = #nsid]
+            pub struct #stream_ident;
+
+            impl jacquard_common::xrpc::SubscriptionResp for #stream_ident {
+                const NSID: &'static str = #nsid;
+                const ENCODING: jacquard_common::xrpc::MessageEncoding = #encoding;
+
+                type Message<'de> = #message_type;
+                type Error<'de> = #error_type;
+            }
+        };
+
+        let params_ident = if has_params {
+            syn::Ident::new(type_base, proc_macro2::Span::call_site())
+        } else {
+            // Generate marker struct if no params
+            let marker = syn::Ident::new(type_base, proc_macro2::Span::call_site());
+            let endpoint_ident = syn::Ident::new(
+                &format!("{}Endpoint", type_base),
+                proc_macro2::Span::call_site(),
+            );
+            let endpoint_path = format!("/xrpc/{}", nsid);
+
+            return Ok(quote! {
+                #stream_resp_impl
+
+                #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
+                pub struct #marker;
+
+                impl jacquard_common::xrpc::XrpcSubscription for #marker {
+                    const NSID: &'static str = #nsid;
+                    const ENCODING: jacquard_common::xrpc::MessageEncoding = #encoding;
+
+                    type Stream = #stream_ident;
+                }
+
+                pub struct #endpoint_ident;
+
+                impl jacquard_common::xrpc::SubscriptionEndpoint for #endpoint_ident {
+                    const PATH: &'static str = #endpoint_path;
+                    const ENCODING: jacquard_common::xrpc::MessageEncoding = #encoding;
+
+                    type Params<'de> = #marker;
+                    type Stream = #stream_ident;
+                }
+            });
+        };
+
+        let (impl_generics, impl_target, endpoint_params_type) =
+            if has_params && params_has_lifetime {
+                (
+                    quote! { <'a> },
+                    quote! { #params_ident<'a> },
+                    quote! { #params_ident<'de> },
+                )
+            } else {
+                (
+                    quote! {},
+                    quote! { #params_ident },
+                    quote! { #params_ident },
+                )
+            };
+
+        let endpoint_ident = syn::Ident::new(
+            &format!("{}Endpoint", type_base),
+            proc_macro2::Span::call_site(),
+        );
+
+        let endpoint_path = format!("/xrpc/{}", nsid);
+
+        Ok(quote! {
+            #stream_resp_impl
+
+            impl #impl_generics jacquard_common::xrpc::XrpcSubscription for #impl_target {
+                const NSID: &'static str = #nsid;
+                const ENCODING: jacquard_common::xrpc::MessageEncoding = #encoding;
+
+                type Stream = #stream_ident;
+            }
+
+            pub struct #endpoint_ident;
+
+            impl jacquard_common::xrpc::SubscriptionEndpoint for #endpoint_ident {
+                const PATH: &'static str = #endpoint_path;
+                const ENCODING: jacquard_common::xrpc::MessageEncoding = #encoding;
+
+                type Params<'de> = #endpoint_params_type;
+                type Stream = #stream_ident;
+            }
+        })
     }
 }