feat(channels): phoenix V2 JSON codec — v0.6 chunk 2

Json<T> newtype codec (src/channels/phoenix.rs), feature "phoenix":
speaks the Phoenix V2 wire format [join_ref, ref, topic, event, payload]
for any T: Serialize + DeserializeOwned. Replies encode as phx_reply
with the {"status", "response"} wrapper; payload-less control frames
encode payload as {}.

Deviation from roadmap sketch (documented in module docs): the roadmap
sketched a blanket Encode/Decode impl on bare P: Serialize. That is
coherence poison — with additive features, enabling "phoenix" anywhere
in the crate graph would forbid every user-written Encode impl. The
blanket therefore lives behind the Json<T> newtype: zero-boilerplate
(Deref/DerefMut/From), but opt-in per payload type.

Encode failure panics and rides the linked-teardown contract (channel
actor is linked to the conn; a payload that cannot serialize is a
programming error, not a protocol event). Decode strictness is exactly
T's Deserialize — use serde defaults or Json<serde_json::Value> for
lenient channels.

Integration (tests/integration.rs, cfg(feature = "phoenix")):
- channels_join_heartbeat_event_broadcast_leave: join ack ordering,
  conn-side heartbeat, unrouted-topic error, join rejection, broadcast
  incl. sender, correlated echo reply, leave -> ok + phx_close, and
  post-leave isolation.
- channels_codec_garbage_closes_1002: undecodable text frame closes
  the socket with protocol-error status.
- shutdown_with_open_channel_terminates: drain force-stop tears down
  the linked channel actor and the runtime reaches AllDone in budget.

Hammer: 35x lifecycle subset (+channels filter) + 3x full (phoenix) +
1x full (phoenix+smarm-trace) + featureless + channels-only, all green.
Suite: 84 unit + 45 integration + 2 doc under --features phoenix.

src/channels/phoenix.rs

@@ -0,0 +1,223 @@
+//! Phoenix V2 wire codec — feature `"phoenix"` (implies `"channels"`,
+//! spends dependency slot #4 on `serde`/`serde_json`, ratified
+//! 2026-06-12).
+//!
+//! The wire format phoenix.js speaks: one JSON array per frame,
+//!
+//! ```text
+//! [join_ref, ref, topic, event, payload]
+//! ```
+//!
+//! refs are strings or `null`; replies are `event: "phx_reply"` with
+//! `payload: {"status": "ok"|"error", "response": <payload>}`.
+//!
+//! # `Json<T>`, not a blanket impl on `P`
+//!
+//! The ratified sketch said "JsonCodec blanket impl via serde". A
+//! literal `impl<P: Serialize + DeserializeOwned> Encode for P` is
+//! coherence poison: features are additive across a dependency graph,
+//! so the moment ANY crate enables `"phoenix"`, every hand-written
+//! `Encode` impl in every other crate stops compiling (the compiler
+//! cannot prove a foreign type will never gain `Serialize`). That
+//! breaks the design's own promise that `"channels"`-only users supply
+//! their own codecs. The blanket therefore lives one newtype away:
+//! [`Json<T>`] is V2-JSON for ANY serde-capable `T` — including
+//! `Json<serde_json::Value>` for schemaless payloads — while the
+//! `Encode`/`Decode` namespace stays open.
+//!
+//! # Strictness
+//!
+//! Decoding is exactly as strict as `T`'s `Deserialize`: a phoenix.js
+//! `{}` join payload fails against a `T` with required fields (the
+//! socket closes `1002`). Use `#[serde(default)]`, `Option` fields, or
+//! `Json<serde_json::Value>` where the client is loose. Encoding a `T`
+//! that fails `Serialize` (non-string map keys and friends) panics —
+//! that is a bug in the payload type, and the panic rides the
+//! documented channel-panic contract (linked teardown).
+
+use super::{
+    ChannelFrame, ChannelMessage, Decode, DecodeError, Encode, FrameRef, MessageRef, Status,
+    EV_REPLY,
+};
+use crate::ws::Message;
+
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use serde_json::{json, Map, Value};
+
+/// V2-JSON payload wrapper: `Json<T>` speaks the phoenix wire for any
+/// `T: Serialize + DeserializeOwned`. Derefs to `T`.
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct Json<T>(pub T);
+
+impl<T> std::ops::Deref for Json<T> {
+    type Target = T;
+    fn deref(&self) -> &T {
+        &self.0
+    }
+}
+impl<T> std::ops::DerefMut for Json<T> {
+    fn deref_mut(&mut self) -> &mut T {
+        &mut self.0
+    }
+}
+impl<T> From<T> for Json<T> {
+    fn from(t: T) -> Self {
+        Json(t)
+    }
+}
+
+fn empty_object() -> Value {
+    Value::Object(Map::new())
+}
+
+fn to_value<T: Serialize>(t: &T) -> Value {
+    serde_json::to_value(t).expect("phoenix codec: payload failed to serialize")
+}
+
+impl<T: Serialize + DeserializeOwned + Send + Sync + 'static> Encode for Json<T> {
+    fn encode(f: FrameRef<'_, Self>) -> Message {
+        let (event, payload) = match f.message {
+            MessageRef::Event { event, payload } => {
+                (event, payload.map(|p| to_value(&p.0)).unwrap_or_else(empty_object))
+            }
+            MessageRef::Reply { status, payload } => (
+                EV_REPLY,
+                json!({
+                    "status": match status {
+                        Status::Ok => "ok",
+                        Status::Error => "error",
+                    },
+                    "response": payload.map(|p| to_value(&p.0)).unwrap_or_else(empty_object),
+                }),
+            ),
+        };
+        let arr = json!([f.join_ref, f.reference, f.topic, event, payload]);
+        Message::Text(arr.to_string())
+    }
+}
+
+impl<T: Serialize + DeserializeOwned + Send + Sync + 'static> Decode for Json<T> {
+    fn decode(msg: &Message) -> Result<ChannelFrame<Self>, DecodeError> {
+        let Message::Text(s) = msg else {
+            return Err(DecodeError("phoenix V2 frames are text".into()));
+        };
+        let (join_ref, reference, topic, event, payload): (
+            Option<String>,
+            Option<String>,
+            String,
+            String,
+            Value,
+        ) = serde_json::from_str(s).map_err(|e| DecodeError(e.to_string()))?;
+
+        let message = if event == EV_REPLY {
+            // Clients don't reply in practice; decoded for symmetry
+            // (and ignored upstream).
+            let status = match payload.get("status").and_then(Value::as_str) {
+                Some("ok") => Status::Ok,
+                _ => Status::Error,
+            };
+            let response = payload.get("response").cloned().unwrap_or_else(empty_object);
+            let payload = serde_json::from_value::<T>(response)
+                .map_err(|e| DecodeError(format!("reply response: {e}")))?;
+            ChannelMessage::Reply { status, payload: Some(Json(payload)) }
+        } else {
+            let payload = serde_json::from_value::<T>(payload)
+                .map_err(|e| DecodeError(format!("payload for '{event}': {e}")))?;
+            ChannelMessage::Event { event, payload: Json(payload) }
+        };
+        Ok(ChannelFrame { join_ref, reference, topic, message })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    type V = Json<Value>;
+
+    fn parse(m: &Message) -> Value {
+        let Message::Text(s) = m else { panic!("not text") };
+        serde_json::from_str(s).unwrap()
+    }
+
+    #[test]
+    fn decodes_phoenix_js_join_frame() {
+        let m = Message::Text(r#"["3","3","room:lobby","phx_join",{"token":"t"}]"#.into());
+        let f = V::decode(&m).unwrap();
+        assert_eq!(f.join_ref.as_deref(), Some("3"));
+        assert_eq!(f.reference.as_deref(), Some("3"));
+        assert_eq!(f.topic, "room:lobby");
+        let ChannelMessage::Event { event, payload } = f.message else { panic!() };
+        assert_eq!(event, "phx_join");
+        assert_eq!(payload.0, json!({"token": "t"}));
+    }
+
+    #[test]
+    fn decodes_heartbeat_null_refs_allowed() {
+        let m = Message::Text(r#"[null,"7","phoenix","heartbeat",{}]"#.into());
+        let f = V::decode(&m).unwrap();
+        assert_eq!(f.join_ref, None);
+        assert_eq!(f.topic, "phoenix");
+    }
+
+    #[test]
+    fn encodes_ok_reply_with_status_response_wrapper() {
+        let p = Json(json!({"user_count": 2}));
+        let m = V::encode(FrameRef {
+            join_ref: Some("3"),
+            reference: Some("4"),
+            topic: "room:lobby",
+            message: MessageRef::Reply { status: Status::Ok, payload: Some(&p) },
+        });
+        assert_eq!(
+            parse(&m),
+            json!(["3", "4", "room:lobby", "phx_reply",
+                   {"status": "ok", "response": {"user_count": 2}}])
+        );
+    }
+
+    #[test]
+    fn encodes_payloadless_reply_and_event_as_empty_object() {
+        let m = V::encode(FrameRef {
+            join_ref: None,
+            reference: Some("9"),
+            topic: "phoenix",
+            message: MessageRef::Reply { status: Status::Ok, payload: None },
+        });
+        assert_eq!(
+            parse(&m),
+            json!([null, "9", "phoenix", "phx_reply", {"status": "ok", "response": {}}])
+        );
+        let m = V::encode(FrameRef {
+            join_ref: Some("3"),
+            reference: None,
+            topic: "room:lobby",
+            message: MessageRef::Event { event: "phx_close", payload: None },
+        });
+        assert_eq!(parse(&m), json!(["3", null, "room:lobby", "phx_close", {}]));
+    }
+
+    #[test]
+    fn typed_payload_round_trip_and_strictness() {
+        #[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq)]
+        struct Chat {
+            body: String,
+        }
+        let m = Message::Text(r#"["1","2","room:a","new_msg",{"body":"hi"}]"#.into());
+        let f = Json::<Chat>::decode(&m).unwrap();
+        let ChannelMessage::Event { payload, .. } = f.message else { panic!() };
+        assert_eq!(payload.0, Chat { body: "hi".into() });
+
+        // Missing required field: the codec is as strict as T.
+        let m = Message::Text(r#"["1","2","room:a","new_msg",{}]"#.into());
+        assert!(Json::<Chat>::decode(&m).is_err());
+    }
+
+    #[test]
+    fn garbage_and_binary_are_decode_errors() {
+        assert!(V::decode(&Message::Text("not json".into())).is_err());
+        assert!(V::decode(&Message::Text(r#"{"a":1}"#.into())).is_err());
+        assert!(V::decode(&Message::Binary(vec![1, 2, 3])).is_err());
+    }
+}

tests/integration.rs

@@ -1408,3 +1408,199 @@ fn shutdown_with_open_chat_terminates() {
     let _ = a.read_to_end(&mut rest);
     let _ = b.read_to_end(&mut rest);
 }
+
+// ===========================================================================
+// Channels (v0.6) — phoenix V2 wire over a real socket
+// ===========================================================================
+
+#[cfg(feature = "phoenix")]
+mod channels_wire {
+    use super::*;
+    use serde_json::{json, Value};
+    use urus::channels::phoenix::Json;
+    use urus::channels::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status};
+
+    type P = Json<Value>;
+
+    /// join rejects payloads carrying "deny"; "shout" broadcasts to the
+    /// room (sender included); anything else echoes as an ok reply.
+    struct Lobby;
+
+    impl Channel<P> for Lobby {
+        fn join(&mut self, topic: &str, payload: P, _s: &ChannelSocket<P>) -> Result<P, P> {
+            if payload.0.get("deny").is_some() {
+                Err(Json(json!({"reason": "denied"})))
+            } else {
+                Ok(Json(json!({"joined": topic})))
+            }
+        }
+        fn handle_in(&mut self, event: &str, payload: P, s: &ChannelSocket<P>) {
+            match event {
+                "shout" => s.broadcast("shouted", payload),
+                _ => s.reply(Status::Ok, Json(json!({"echo": event}))),
+            }
+        }
+    }
+
+    fn channels_pipeline() -> Pipeline {
+        let hub: std::sync::Arc<std::sync::OnceLock<ChannelHub<P>>> =
+            std::sync::Arc::new(std::sync::OnceLock::new());
+        Pipeline::new().plug(Router::new().get("/socket", move |c: Conn, _n: Next| {
+            // Hub construction is in-runtime only (it spawns the pubsub
+            // table); the NON-static OnceLock is what lets shutdown
+            // drain the table (the v0.5 pattern).
+            let hub = hub.clone();
+            let hub = hub.get_or_init(|| {
+                ChannelHub::new(PrefixRouter::new().channel("room:*", |_: &str| {
+                    Box::new(Lobby) as Box<dyn Channel<P>>
+                }))
+            });
+            hub.upgrade(c)
+        }))
+    }
+
+    const CHAN_HANDSHAKE: &[u8] =
+        b"GET /socket HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
+          Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n";
+
+    fn chan_connect(port: u16) -> TcpStream {
+        let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
+        s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+        s.write_all(CHAN_HANDSHAKE).unwrap();
+        let head = String::from_utf8(read_response_head(&mut s)).unwrap();
+        assert!(head.starts_with("HTTP/1.1 101"), "handshake: {head}");
+        s
+    }
+
+    fn chan_send(s: &mut TcpStream, frame: Value) {
+        ws_send(s, &Frame::new(Opcode::Text, frame.to_string()));
+    }
+
+    fn chan_read(s: &mut TcpStream, buf: &mut Vec<u8>) -> Value {
+        let f = ws_read_frame(s, buf);
+        assert_eq!(f.opcode, Opcode::Text, "expected V2 text frame");
+        serde_json::from_slice(&f.payload).expect("server sent non-JSON frame")
+    }
+
+    #[test]
+    fn channels_join_heartbeat_event_broadcast_leave() {
+        let port = spawn_server(channels_pipeline());
+
+        let mut a = chan_connect(port);
+        let mut a_buf = Vec::new();
+
+        // Join — the reply IS the subscription ack.
+        chan_send(&mut a, json!(["1", "1", "room:lobby", "phx_join", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["1", "1", "room:lobby", "phx_reply",
+                   {"status": "ok", "response": {"joined": "room:lobby"}}])
+        );
+
+        // Transport heartbeat.
+        chan_send(&mut a, json!([null, "2", "phoenix", "heartbeat", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!([null, "2", "phoenix", "phx_reply", {"status": "ok", "response": {}}])
+        );
+
+        // Unrouted topic: error reply.
+        chan_send(&mut a, json!(["9", "3", "hall:x", "phx_join", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["9", "3", "hall:x", "phx_reply", {"status": "error", "response": {}}])
+        );
+
+        // Rejected join: the channel's error payload comes back.
+        chan_send(&mut a, json!(["9", "4", "room:vip", "phx_join", {"deny": true}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["9", "4", "room:vip", "phx_reply",
+                   {"status": "error", "response": {"reason": "denied"}}])
+        );
+
+        // Second member; a's shout reaches both (sender included).
+        let mut b = chan_connect(port);
+        let mut b_buf = Vec::new();
+        chan_send(&mut b, json!(["1", "1", "room:lobby", "phx_join", {}]));
+        chan_read(&mut b, &mut b_buf);
+
+        chan_send(&mut a, json!(["1", "5", "room:lobby", "shout", {"body": "hi"}]));
+        let push = json!(["1", null, "room:lobby", "shouted", {"body": "hi"}]);
+        assert_eq!(chan_read(&mut a, &mut a_buf), push);
+        assert_eq!(chan_read(&mut b, &mut b_buf), push);
+
+        // Plain event echoes as a correlated reply.
+        chan_send(&mut a, json!(["1", "6", "room:lobby", "wave", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["1", "6", "room:lobby", "phx_reply",
+                   {"status": "ok", "response": {"echo": "wave"}}])
+        );
+
+        // Leave: ok reply, then phx_close; the topic is then unjoined.
+        chan_send(&mut a, json!(["1", "7", "room:lobby", "phx_leave", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["1", "7", "room:lobby", "phx_reply", {"status": "ok", "response": {}}])
+        );
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["1", null, "room:lobby", "phx_close", {}])
+        );
+        chan_send(&mut a, json!(["1", "8", "room:lobby", "wave", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!(["1", "8", "room:lobby", "phx_reply", {"status": "error", "response": {}}])
+        );
+
+        // a is gone from the room: b's shout comes back to b only, and
+        // a's connection stays a working transport (heartbeat).
+        chan_send(&mut b, json!(["1", "9", "room:lobby", "shout", {"n": 2}]));
+        assert_eq!(
+            chan_read(&mut b, &mut b_buf),
+            json!(["1", null, "room:lobby", "shouted", {"n": 2}])
+        );
+        chan_send(&mut a, json!([null, "10", "phoenix", "heartbeat", {}]));
+        assert_eq!(
+            chan_read(&mut a, &mut a_buf),
+            json!([null, "10", "phoenix", "phx_reply", {"status": "ok", "response": {}}])
+        );
+    }
+
+    #[test]
+    fn channels_codec_garbage_closes_1002() {
+        let port = spawn_server(channels_pipeline());
+        let mut s = chan_connect(port);
+        let mut buf = Vec::new();
+        ws_send(&mut s, &Frame::new(Opcode::Text, "not a v2 frame"));
+        let f = ws_read_frame(&mut s, &mut buf);
+        assert_eq!(f.opcode, Opcode::Close);
+        assert_eq!(u16::from_be_bytes([f.payload[0], f.payload[1]]), 1002);
+    }
+
+    #[test]
+    fn shutdown_with_open_channel_terminates() {
+        // The whole exit chain under one roof: drain force-stops the
+        // conn actor parked in its select -> linked channel actor dies
+        // (and/or wakes on the dropped inbound arm) -> its subscription
+        // is monitor-pruned -> the non-static hub's last handle drops
+        // with the drained pipeline -> pubsub table exits -> AllDone.
+        let (port, handle, done_rx) =
+            spawn_server_with_handle(channels_pipeline(), Duration::from_millis(300));
+
+        let mut a = chan_connect(port);
+        let mut a_buf = Vec::new();
+        chan_send(&mut a, json!(["1", "1", "room:lobby", "phx_join", {}]));
+        let reply = chan_read(&mut a, &mut a_buf);
+        assert_eq!(reply[3], "phx_reply"); // joined: channel actor live
+
+        handle.shutdown();
+        done_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("serve did not return: channel actor, hub, or table outlived the drain");
+
+        let mut rest = Vec::new();
+        let _ = a.read_to_end(&mut rest);
+    }
+}