openrtc/

native_protocol.rs

use base64::Engine as _;
use serde::{Deserialize, Serialize};

fn parse_application_key_agreement_public_key(
    json: &serde_json::Value,
) -> Option<[u8; crate::key_agreement::PUBLIC_KEY_BYTES]> {
    let payload = json.get("applicationKeyAgreement")?.as_object()?;
    if payload.get("algorithm").and_then(|value| value.as_str())
        != Some(crate::key_agreement::KEY_ALGORITHM)
    {
        return None;
    }
    let public_key = payload.get("publicKey")?.as_str()?.trim();
    if public_key.is_empty() {
        return None;
    }
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(public_key)
        .ok()?;
    if bytes.len() != crate::key_agreement::PUBLIC_KEY_BYTES {
        return None;
    }
    let mut out = [0u8; crate::key_agreement::PUBLIC_KEY_BYTES];
    out.copy_from_slice(&bytes);
    Some(out)
}

fn parse_transport_trust_device_id(json: &serde_json::Value) -> Option<String> {
    json.get("transportTrust")
        .and_then(|value| value.as_object())
        .and_then(|payload| payload.get("deviceId"))
        .and_then(|value| value.as_str())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum NativeMessageAction {
    Request,
    Response,
    Event,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeMainMessage {
    pub channel: String,
    pub action: NativeMessageAction,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub request_id: Option<String>,
    pub payload: serde_json::Value,
    pub timestamp: i64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub from: Option<String>,
}

/// Channel name used for SDK-owned session token presentation.
/// The connecting client sends this as the first native message after
/// connection when a compound ticket contained an embedded token.
pub const SESSION_TOKEN_CHANNEL: &str = "session-token";
pub const PEER_DATA_CHANNEL: &str = "peer-data";
pub const OFFLINE_PROOF_CHANNEL: &str = "offline-device-proof";
/// `send_peer` is the bounded, ordered message surface. Larger payloads use a
/// named stream or a protocol package so lifecycle/control frames cannot be
/// stalled behind bulk transfer data on the native-main stream.
pub const MAX_PEER_MESSAGE_BYTES: usize = 64 * 1024;

/// Lifetime contract for the stream carrying a session-token presentation.
///
/// This describes the stream, not the sender runtime or grant scope. Browser
/// bridges commonly use a bounded request/response exchange, while native
/// peers retain the admitted stream as their generation-bound control route.
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum TokenStreamContract {
    #[default]
    OneShotAdmission,
    PersistentControl,
}

/// Reverse-direction admission carried in the response to a typed reciprocal
/// request. Keeping it on the request's stream binds both proofs to the same
/// physical generation and avoids selecting a stale endpoint connection after
/// network replacement.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReciprocalSessionTokenPresentation {
    pub presentation_id: String,
    pub token: String,
    pub token_payload: String,
    pub device_id: String,
    pub stream_contract: TokenStreamContract,
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg(not(target_arch = "wasm32"))]
pub(crate) enum ReciprocalSessionTokenPresentationState {
    Absent,
    Malformed,
    Present(ReciprocalSessionTokenPresentation),
}

impl NativeMainMessage {
    pub fn is_handshake(&self) -> bool {
        self.channel == "handshake"
    }

    pub fn is_session_token_presentation(&self) -> bool {
        self.channel == SESSION_TOKEN_CHANNEL && matches!(self.action, NativeMessageAction::Request)
    }

    pub fn is_session_token_response(&self) -> bool {
        self.channel == SESSION_TOKEN_CHANNEL
            && matches!(self.action, NativeMessageAction::Response)
    }

    pub fn is_session_token_response_ack(&self, connection_id: &str) -> bool {
        self.channel == SESSION_TOKEN_CHANNEL
            && matches!(self.action, NativeMessageAction::Event)
            && self
                .payload
                .get("responseAck")
                .and_then(|value| value.as_bool())
                == Some(true)
            && self
                .payload
                .get("connectionId")
                .and_then(|value| value.as_str())
                == Some(connection_id)
    }

    pub fn is_peer_data(&self) -> bool {
        self.channel == PEER_DATA_CHANNEL
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn is_offline_proof_request(&self) -> bool {
        self.channel == OFFLINE_PROOF_CHANNEL && matches!(self.action, NativeMessageAction::Request)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn is_offline_proof_response(&self) -> bool {
        self.channel == OFFLINE_PROOF_CHANNEL
            && matches!(self.action, NativeMessageAction::Response)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn offline_proof_challenge(&self) -> Option<crate::offline::OfflineProofChallenge> {
        serde_json::from_value(self.payload.get("challenge")?.clone()).ok()
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn offline_connection_proof(
        &self,
    ) -> Option<crate::offline::OfflineConnectionProof> {
        serde_json::from_value(self.payload.get("proof")?.clone()).ok()
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn offline_proof_request(challenge: &crate::offline::OfflineProofChallenge) -> Self {
        Self {
            channel: OFFLINE_PROOF_CHANNEL.to_string(),
            action: NativeMessageAction::Request,
            request_id: Some(challenge.request_id.clone()),
            payload: serde_json::json!({ "challenge": challenge }),
            timestamp: crate::coordination::now_millis_u64().min(i64::MAX as u64) as i64,
            from: None,
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn offline_proof_response(
        request_id: &str,
        proof: &crate::offline::OfflineConnectionProof,
    ) -> Self {
        Self {
            channel: OFFLINE_PROOF_CHANNEL.to_string(),
            action: NativeMessageAction::Response,
            request_id: Some(request_id.to_string()),
            payload: serde_json::json!({ "proof": proof }),
            timestamp: crate::coordination::now_millis_u64().min(i64::MAX as u64) as i64,
            from: None,
        }
    }

    pub fn peer_data_payload(&self) -> Option<Vec<u8>> {
        let payload_base64 = self
            .payload
            .get("payloadBase64")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())?;
        base64::engine::general_purpose::STANDARD
            .decode(payload_base64.as_bytes())
            .ok()
    }

    pub fn claimed_device_id(&self) -> Option<String> {
        self.payload
            .get("deviceId")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    pub fn presented_session_token(&self) -> Option<String> {
        self.payload
            .get("token")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    pub fn presented_session_token_payload(&self) -> Option<String> {
        self.payload
            .get("tokenPayload")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    pub fn session_token_stream_contract(&self) -> TokenStreamContract {
        self.payload
            .get("streamContract")
            .cloned()
            .and_then(|value| serde_json::from_value(value).ok())
            .unwrap_or_default()
    }

    /// Whether the presenter has lost the reciprocal inbound proof for this
    /// physical generation and is asking the peer's admission actor to
    /// re-present its own token. Missing on older peers means `false`.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn session_token_reciprocal_requested(&self) -> bool {
        self.payload
            .get("reciprocalRequested")
            .and_then(|value| value.as_bool())
            .unwrap_or(false)
    }

    pub fn reciprocal_session_token_mode(&self) -> Option<&str> {
        self.payload
            .get("reciprocalMode")
            .and_then(|value| value.as_str())
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn reciprocal_session_token_presentation_state(
        &self,
    ) -> ReciprocalSessionTokenPresentationState {
        let Some(presentation) = self.payload.get("reciprocalPresentation").cloned() else {
            return ReciprocalSessionTokenPresentationState::Absent;
        };
        let Ok(parsed) = serde_json::from_value::<ReciprocalSessionTokenPresentation>(presentation)
        else {
            return ReciprocalSessionTokenPresentationState::Malformed;
        };
        if parsed.presentation_id.trim().is_empty()
            || parsed.token.trim().is_empty()
            || parsed.token_payload.trim().is_empty()
            || parsed.device_id.trim().is_empty()
        {
            return ReciprocalSessionTokenPresentationState::Malformed;
        }
        ReciprocalSessionTokenPresentationState::Present(parsed)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn reciprocal_session_token_presentation_id(&self) -> Option<String> {
        self.payload
            .get("reciprocalPresentation")
            .and_then(|value| value.get("presentationId"))
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    /// New native peers opt into an explicit approval-delivery ACK before a
    /// persistent control stream participates in canonical stream arbitration.
    pub(crate) fn session_token_response_ack_requested(&self) -> bool {
        self.payload
            .get("responseAckRequested")
            .and_then(|value| value.as_bool())
            .unwrap_or(false)
    }

    pub fn with_session_token_stream_contract(mut self, contract: TokenStreamContract) -> Self {
        if let Some(payload) = self.payload.as_object_mut() {
            payload.insert(
                "streamContract".to_string(),
                serde_json::to_value(contract)
                    .expect("session-token stream contract must serialize"),
            );
        }
        self
    }

    pub(crate) fn with_session_token_reciprocal_requested(mut self, requested: bool) -> Self {
        if requested {
            if let Some(payload) = self.payload.as_object_mut() {
                payload.insert(
                    "reciprocalRequested".to_string(),
                    serde_json::Value::Bool(true),
                );
                payload.insert(
                    "reciprocalMode".to_string(),
                    serde_json::Value::String("inline-v1".to_string()),
                );
            }
        }
        self
    }

    pub fn session_token_approved(&self) -> Option<bool> {
        self.payload
            .get("approved")
            .and_then(|value| value.as_bool())
    }

    pub fn approved_session_scope(&self) -> Option<String> {
        self.payload
            .get("scope")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    pub fn session_token_error(&self) -> Option<String> {
        self.payload
            .get("reason")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    /// Build a session-token presentation message for the connecting client
    /// to send to the host.
    pub fn session_token_presentation(token: &str) -> Self {
        Self::session_token_presentation_with_payload(token, None)
    }

    pub(crate) fn session_token_presentation_with_payload(
        token: &str,
        token_payload: Option<&str>,
    ) -> Self {
        Self::session_token_presentation_with_payload_and_device_id(token, token_payload, None)
    }

    pub(crate) fn session_token_presentation_with_payload_and_device_id(
        token: &str,
        token_payload: Option<&str>,
        device_id: Option<&str>,
    ) -> Self {
        #[cfg(target_arch = "wasm32")]
        let timestamp = js_sys::Date::now() as i64;
        #[cfg(not(target_arch = "wasm32"))]
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0);

        let mut payload = serde_json::json!({
            "token": token,
            "streamContract": TokenStreamContract::OneShotAdmission,
            "responseAckRequested": true,
        });
        if let Some(token_payload) = token_payload {
            if let Some(object) = payload.as_object_mut() {
                object.insert(
                    "tokenPayload".to_string(),
                    serde_json::Value::String(token_payload.to_string()),
                );
            }
        }
        if let Some(device_id) = device_id.map(str::trim).filter(|value| !value.is_empty()) {
            if let Some(object) = payload.as_object_mut() {
                object.insert(
                    "deviceId".to_string(),
                    serde_json::Value::String(device_id.to_string()),
                );
            }
        }

        Self {
            channel: SESSION_TOKEN_CHANNEL.to_string(),
            action: NativeMessageAction::Request,
            request_id: None,
            payload,
            timestamp,
            from: None,
        }
    }

    pub fn session_token_approval(scope: Option<&str>, connection_id: &str) -> Self {
        #[cfg(target_arch = "wasm32")]
        let timestamp = js_sys::Date::now() as i64;
        #[cfg(not(target_arch = "wasm32"))]
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0);

        Self {
            channel: SESSION_TOKEN_CHANNEL.to_string(),
            action: NativeMessageAction::Response,
            request_id: None,
            payload: serde_json::json!({
                "approved": true,
                "scope": scope,
                "connectionId": connection_id,
            }),
            timestamp,
            from: None,
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn with_reciprocal_session_token_presentation(
        mut self,
        presentation: ReciprocalSessionTokenPresentation,
    ) -> Self {
        if let Some(payload) = self.payload.as_object_mut() {
            payload.insert(
                "reciprocalMode".to_string(),
                serde_json::Value::String("inline-v1".to_string()),
            );
            payload.insert(
                "reciprocalPresentation".to_string(),
                serde_json::to_value(presentation)
                    .expect("reciprocal session-token presentation must serialize"),
            );
        }
        self
    }

    pub fn session_token_rejection(reason: &str, connection_id: &str) -> Self {
        #[cfg(target_arch = "wasm32")]
        let timestamp = js_sys::Date::now() as i64;
        #[cfg(not(target_arch = "wasm32"))]
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0);

        Self {
            channel: SESSION_TOKEN_CHANNEL.to_string(),
            action: NativeMessageAction::Response,
            request_id: None,
            payload: serde_json::json!({
                "approved": false,
                "reason": reason,
                "connectionId": connection_id,
            }),
            timestamp,
            from: None,
        }
    }

    pub fn session_token_response_ack(connection_id: &str) -> Self {
        #[cfg(target_arch = "wasm32")]
        let timestamp = js_sys::Date::now() as i64;
        #[cfg(not(target_arch = "wasm32"))]
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0);

        Self {
            channel: SESSION_TOKEN_CHANNEL.to_string(),
            action: NativeMessageAction::Event,
            request_id: None,
            payload: serde_json::json!({
                "responseAck": true,
                "connectionId": connection_id,
            }),
            timestamp,
            from: None,
        }
    }

    pub fn with_reciprocal_session_token_ack(
        mut self,
        presentation_id: &str,
        accepted: bool,
        scope: Option<&str>,
        reason: Option<&str>,
    ) -> Self {
        if let Some(payload) = self.payload.as_object_mut() {
            payload.insert(
                "reciprocalPresentationId".to_string(),
                serde_json::Value::String(presentation_id.to_string()),
            );
            payload.insert(
                "reciprocalAccepted".to_string(),
                serde_json::Value::Bool(accepted),
            );
            if let Some(scope) = scope {
                payload.insert(
                    "reciprocalScope".to_string(),
                    serde_json::Value::String(scope.to_string()),
                );
            }
            if let Some(reason) = reason {
                payload.insert(
                    "reciprocalReason".to_string(),
                    serde_json::Value::String(reason.to_string()),
                );
            }
        }
        self
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn reciprocal_session_token_ack_presentation_id(&self) -> Option<String> {
        self.payload
            .get("reciprocalPresentationId")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn reciprocal_session_token_ack_accepted(&self) -> Option<bool> {
        self.payload
            .get("reciprocalAccepted")
            .and_then(|value| value.as_bool())
    }

    pub fn reciprocal_session_token_ack_scope(&self) -> Option<String> {
        self.payload
            .get("reciprocalScope")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
    }

    pub fn peer_data(payload: &[u8]) -> Self {
        #[cfg(target_arch = "wasm32")]
        let timestamp = js_sys::Date::now() as i64;
        #[cfg(not(target_arch = "wasm32"))]
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0);

        Self {
            channel: PEER_DATA_CHANNEL.to_string(),
            action: NativeMessageAction::Event,
            request_id: None,
            payload: serde_json::json!({
                "payloadBase64": base64::engine::general_purpose::STANDARD.encode(payload),
            }),
            timestamp,
            from: None,
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn encode_native_main_payload(message: &NativeMainMessage) -> Result<Vec<u8>, String> {
    let payload = serde_json::to_vec(message)
        .map_err(|error| format!("encode native-main message: {error}"))?;
    let len = u32::try_from(payload.len())
        .map_err(|_| "native-main message exceeds the frame bound".to_string())?;
    let mut frame = Vec::with_capacity(4 + payload.len());
    frame.extend_from_slice(&len.to_be_bytes());
    frame.extend_from_slice(&payload);
    Ok(frame)
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn encode_native_main_opening(message: &NativeMainMessage) -> Result<Vec<u8>, String> {
    let payload = encode_native_main_payload(message)?;
    let label = b"main";
    let mut opening = Vec::with_capacity(1 + 4 + label.len() + payload.len());
    opening.push(0);
    opening.extend_from_slice(&(label.len() as u32).to_be_bytes());
    opening.extend_from_slice(label);
    opening.extend_from_slice(&payload);
    Ok(opening)
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HandshakeFeatures {
    pub carrier_schema: Option<u8>,
    pub webrtc: Option<bool>,
    pub moq: Option<bool>,
    pub dedicated_carrier_control: Option<bool>,
    pub dedicated_carrier_control_ready: Option<bool>,
    pub ble: Option<bool>,
    pub application_key_agreement: Option<bool>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TypeScriptHandshake {
    pub action: Option<String>,
    pub session_token: Option<String>,
    pub session_token_payload: Option<String>,
    pub claimed_device_id: Option<String>,
    pub capabilities: Option<HandshakeFeatures>,
    pub application_key_agreement_public_key: Option<[u8; crate::key_agreement::PUBLIC_KEY_BYTES]>,
}

#[derive(Debug, Clone)]
pub enum ParsedMainFrame {
    NativeMessage(NativeMainMessage),
    TypeScriptHandshake(TypeScriptHandshake),
    TypeScriptJson(serde_json::Value),
    Opaque,
}

#[derive(Debug, Clone)]
pub enum InspectedMainFrame {
    NativeMessage {
        message: NativeMainMessage,
        handshake: Option<HandshakeBinding>,
    },
    /// SDK-owned response for a session-token presentation received on an
    /// application-forwarded one-shot main stream. The host adapter writes
    /// this response verbatim, then asks the client to run post-admission
    /// lifecycle work only after the wire flush completes.
    SessionTokenResponse {
        response: NativeMainMessage,
        accepted: bool,
        is_first_presentation: bool,
    },
    ForwardOpaque,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct HandshakeBinding {
    pub known_device_id: Option<String>,
    pub claimed_device_id: Option<String>,
    pub admitted_device_id: Option<String>,
    pub authoritative_device_id_hint: Option<String>,
}

pub fn parse_main_frame(frame: &[u8]) -> ParsedMainFrame {
    if let Ok(message) = serde_json::from_slice::<NativeMainMessage>(frame) {
        return ParsedMainFrame::NativeMessage(message);
    }

    if frame.len() >= 2 && frame[0] == 0x00 {
        if let Ok(json) = serde_json::from_slice::<serde_json::Value>(&frame[1..]) {
            if json.get("type").and_then(|value| value.as_str()) == Some("handshake") {
                return ParsedMainFrame::TypeScriptHandshake(TypeScriptHandshake {
                    action: json
                        .get("action")
                        .and_then(|value| value.as_str())
                        .map(ToOwned::to_owned),
                    session_token: json
                        .get("sessionToken")
                        .and_then(|value| value.as_str())
                        .map(str::trim)
                        .filter(|value| !value.is_empty())
                        .map(ToOwned::to_owned),
                    session_token_payload: json
                        .get("sessionTokenPayload")
                        .and_then(|value| value.as_str())
                        .map(str::trim)
                        .filter(|value| !value.is_empty())
                        .map(ToOwned::to_owned),
                    claimed_device_id: parse_transport_trust_device_id(&json),
                    capabilities: json
                        .get("capabilities")
                        .and_then(|value| value.as_object())
                        .map(|value| HandshakeFeatures {
                            carrier_schema: value
                                .get("carrierSchema")
                                .and_then(|version| version.as_u64())
                                .and_then(|version| u8::try_from(version).ok()),
                            webrtc: value.get("webrtc").and_then(|flag| flag.as_bool()),
                            moq: value.get("moq").and_then(|flag| flag.as_bool()),
                            dedicated_carrier_control: value
                                .get("dedicatedCarrierControl")
                                .and_then(|flag| flag.as_bool()),
                            dedicated_carrier_control_ready: value
                                .get("dedicatedCarrierControlReady")
                                .and_then(|flag| flag.as_bool()),
                            ble: value.get("ble").and_then(|flag| flag.as_bool()),
                            application_key_agreement: value
                                .get("applicationKeyAgreement")
                                .and_then(|flag| flag.as_bool()),
                        }),
                    application_key_agreement_public_key:
                        parse_application_key_agreement_public_key(&json),
                });
            }
            return ParsedMainFrame::TypeScriptJson(json);
        }
    }

    ParsedMainFrame::Opaque
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_native_handshake_message() {
        let frame = serde_json::json!({
            "channel": "handshake",
            "action": "request",
            "payload": { "deviceId": "device-1" },
            "timestamp": 123,
        });

        match parse_main_frame(frame.to_string().as_bytes()) {
            ParsedMainFrame::NativeMessage(message) => {
                assert!(message.is_handshake());
                assert_eq!(message.claimed_device_id().as_deref(), Some("device-1"));
            }
            other => panic!("expected native message, got {:?}", other),
        }
    }

    #[test]
    fn parses_typescript_handshake_message() {
        let json = serde_json::json!({
            "type": "handshake",
            "action": "hello",
            "sessionToken": "token-1",
            "sessionTokenPayload": "payload-1",
            "transportTrust": {
                "version": 1,
                "deviceId": "device-1",
                "identityFingerprint": "fingerprint-1"
            },
            "capabilities": {
                "carrierSchema": 2,
                "webrtc": true,
                "moq": false,
                "dedicatedCarrierControl": true,
                "dedicatedCarrierControlReady": false,
                "ble": true,
            },
        });
        let mut frame = vec![0x00];
        frame.extend_from_slice(json.to_string().as_bytes());

        match parse_main_frame(&frame) {
            ParsedMainFrame::TypeScriptHandshake(handshake) => {
                assert_eq!(handshake.action.as_deref(), Some("hello"));
                assert_eq!(handshake.session_token.as_deref(), Some("token-1"));
                assert_eq!(
                    handshake.session_token_payload.as_deref(),
                    Some("payload-1")
                );
                assert_eq!(handshake.claimed_device_id.as_deref(), Some("device-1"));
                let capabilities = handshake.capabilities.expect("capabilities");
                assert_eq!(capabilities.carrier_schema, Some(2));
                assert_eq!(capabilities.webrtc, Some(true));
                assert_eq!(capabilities.moq, Some(false));
                assert_eq!(capabilities.dedicated_carrier_control, Some(true));
                assert_eq!(capabilities.dedicated_carrier_control_ready, Some(false));
                assert_eq!(capabilities.ble, Some(true));
            }
            other => panic!("expected typescript handshake, got {:?}", other),
        }
    }

    #[test]
    fn falls_back_to_opaque_for_unrecognized_bytes() {
        assert!(matches!(
            parse_main_frame(b"\x01\x02\x03"),
            ParsedMainFrame::Opaque
        ));
    }

    #[test]
    fn parses_peer_data_native_message() {
        let message = NativeMainMessage::peer_data(b"hello peer");
        let frame = serde_json::to_vec(&message).expect("peer data should serialize");

        match parse_main_frame(&frame) {
            ParsedMainFrame::NativeMessage(parsed) => {
                assert!(parsed.is_peer_data());
                assert_eq!(
                    parsed.peer_data_payload().as_deref(),
                    Some(b"hello peer".as_slice())
                );
            }
            other => panic!("expected native peer data message, got {:?}", other),
        }
    }

    #[test]
    fn session_token_stream_contract_round_trips_and_legacy_defaults_to_one_shot() {
        let persistent = NativeMainMessage::session_token_presentation("token")
            .with_session_token_stream_contract(TokenStreamContract::PersistentControl);
        let encoded = serde_json::to_vec(&persistent).expect("serialize persistent contract");
        let decoded: NativeMainMessage =
            serde_json::from_slice(&encoded).expect("deserialize persistent contract");
        assert_eq!(
            decoded.session_token_stream_contract(),
            TokenStreamContract::PersistentControl,
        );

        let mut legacy = NativeMainMessage::session_token_presentation("legacy-token");
        legacy
            .payload
            .as_object_mut()
            .expect("presentation payload")
            .remove("streamContract");
        assert_eq!(
            legacy.session_token_stream_contract(),
            TokenStreamContract::OneShotAdmission,
        );
        assert!(!legacy.session_token_reciprocal_requested());
    }

    #[test]
    fn session_token_reciprocal_request_round_trips_and_defaults_false() {
        let legacy = NativeMainMessage::session_token_presentation("legacy-token");
        assert!(!legacy.session_token_reciprocal_requested());
        assert_eq!(legacy.reciprocal_session_token_mode(), None);

        let requested = NativeMainMessage::session_token_presentation("token")
            .with_session_token_reciprocal_requested(true);
        let encoded = serde_json::to_vec(&requested).expect("serialize reciprocal request");
        let decoded: NativeMainMessage =
            serde_json::from_slice(&encoded).expect("deserialize reciprocal request");
        assert!(decoded.session_token_reciprocal_requested());
        assert_eq!(decoded.reciprocal_session_token_mode(), Some("inline-v1"));
    }

    #[test]
    fn reciprocal_session_token_presentation_round_trips_on_approval() {
        let response = NativeMainMessage::session_token_approval(Some("user-device"), "conn-1")
            .with_reciprocal_session_token_presentation(ReciprocalSessionTokenPresentation {
                presentation_id: "presentation-1".to_string(),
                token: "reverse-token".to_string(),
                token_payload: "reverse-payload".to_string(),
                device_id: "device-1".to_string(),
                stream_contract: TokenStreamContract::PersistentControl,
            });
        let encoded = serde_json::to_vec(&response).expect("serialize reciprocal response");
        let decoded: NativeMainMessage =
            serde_json::from_slice(&encoded).expect("deserialize reciprocal response");
        assert_eq!(decoded.reciprocal_session_token_mode(), Some("inline-v1"));
        assert_eq!(
            decoded.reciprocal_session_token_presentation_state(),
            ReciprocalSessionTokenPresentationState::Present(ReciprocalSessionTokenPresentation {
                presentation_id: "presentation-1".to_string(),
                token: "reverse-token".to_string(),
                token_payload: "reverse-payload".to_string(),
                device_id: "device-1".to_string(),
                stream_contract: TokenStreamContract::PersistentControl,
            })
        );

        assert_eq!(
            NativeMainMessage::session_token_approval(None, "conn-1")
                .reciprocal_session_token_presentation_state(),
            ReciprocalSessionTokenPresentationState::Absent,
        );

        let malformed = NativeMainMessage::session_token_approval(None, "conn-1")
            .with_reciprocal_session_token_presentation(ReciprocalSessionTokenPresentation {
                presentation_id: "presentation-2".to_string(),
                token: "".to_string(),
                token_payload: "reverse-payload".to_string(),
                device_id: "device-1".to_string(),
                stream_contract: TokenStreamContract::OneShotAdmission,
            });
        assert_eq!(
            malformed.reciprocal_session_token_presentation_state(),
            ReciprocalSessionTokenPresentationState::Malformed,
        );
    }

    #[test]
    fn parses_pluto_signal_envelope_with_ts_prefix() {
        let json = serde_json::json!({
            "type": "#pluto-signal",
            "content": {
                "transport": "webrtc",
                "type": "sdp",
                "sdp": { "type": "answer", "sdp": "v=0\r\n" }
            }
        });
        let mut frame = vec![0x00];
        frame.extend_from_slice(json.to_string().as_bytes());

        match parse_main_frame(&frame) {
            ParsedMainFrame::TypeScriptJson(parsed) => {
                assert_eq!(
                    parsed.get("type").and_then(|v| v.as_str()),
                    Some("#pluto-signal")
                );
                let content = parsed.get("content").unwrap();
                assert_eq!(
                    content.get("transport").and_then(|v| v.as_str()),
                    Some("webrtc")
                );
            }
            other => panic!("expected TypeScriptJson for #pluto-signal, got {:?}", other),
        }
    }

    #[test]
    fn raw_pluto_signal_without_prefix_is_opaque() {
        let json = serde_json::json!({
            "type": "#pluto-signal",
            "content": { "transport": "webrtc", "type": "sdp" }
        });
        let frame = json.to_string().into_bytes();

        match parse_main_frame(&frame) {
            ParsedMainFrame::NativeMessage(_) => {
                panic!("raw #pluto-signal should not parse as NativeMainMessage")
            }
            ParsedMainFrame::TypeScriptJson(_) => {
                panic!("raw #pluto-signal without 0x00 prefix should not be TypeScriptJson")
            }
            ParsedMainFrame::Opaque => {}
            other => panic!("expected Opaque, got {:?}", other),
        }
    }

    #[test]
    fn session_token_response_helpers_round_trip() {
        let presentation = NativeMainMessage::session_token_presentation_with_payload(
            "token-1",
            Some("payload-1"),
        );
        assert!(presentation.is_session_token_presentation());
        assert_eq!(
            presentation.presented_session_token().as_deref(),
            Some("token-1")
        );
        assert_eq!(
            presentation.presented_session_token_payload().as_deref(),
            Some("payload-1")
        );
        assert!(presentation.session_token_response_ack_requested());

        let presentation_with_device =
            NativeMainMessage::session_token_presentation_with_payload_and_device_id(
                "token-1",
                Some("payload-1"),
                Some("device-1"),
            );
        assert_eq!(
            presentation_with_device.claimed_device_id().as_deref(),
            Some("device-1")
        );

        let approved = NativeMainMessage::session_token_approval(Some("magic-link"), "conn-1");
        assert!(approved.is_session_token_response());
        assert_eq!(approved.session_token_approved(), Some(true));
        assert_eq!(
            approved.approved_session_scope().as_deref(),
            Some("magic-link")
        );

        let rejected = NativeMainMessage::session_token_rejection("invalid-token", "conn-1");
        assert!(rejected.is_session_token_response());
        assert_eq!(rejected.session_token_approved(), Some(false));
        assert_eq!(
            rejected.session_token_error().as_deref(),
            Some("invalid-token")
        );

        let ack = NativeMainMessage::session_token_response_ack("conn-1");
        assert!(!ack.is_session_token_presentation());
        assert!(ack.is_session_token_response_ack("conn-1"));
        assert!(!ack.is_session_token_response_ack("conn-2"));

        let reciprocal_ack = NativeMainMessage::session_token_response_ack("conn-1")
            .with_reciprocal_session_token_ack("presentation-1", true, Some("user-device"), None);
        assert_eq!(
            reciprocal_ack.payload.get("reciprocalPresentationId"),
            Some(&serde_json::Value::String("presentation-1".to_string())),
        );
        assert_eq!(
            reciprocal_ack.payload.get("reciprocalAccepted"),
            Some(&serde_json::Value::Bool(true)),
        );
    }
}