// Protocol infrastructure — extracted from mesh.rs

#[cfg(test)]
use crate::mesh::NodeRole;
use crate::mesh::PeerAnnouncement;

pub(crate) mod convert;
use anyhow::Result;
pub(crate) use convert::*;
use iroh::endpoint::Connection;
use iroh::{Endpoint, EndpointAddr, EndpointId};
use prost::Message;
pub const ALPN_CONTROL_V1: &[u8] = b"mesh-llm-control/1";
pub const ALPN_V1: &[u8] = b"mesh-llm/1";
#[cfg(test)]
pub const ALPN: &[u8] = ALPN_V1;
pub(crate) const NODE_PROTOCOL_GENERATION: u32 = 1;
pub(crate) const MAX_CONTROL_FRAME_BYTES: usize = 8 * 1024 * 1024; // 8 MiB

pub(crate) const STREAM_GOSSIP: u8 = 0x01;
pub(crate) const STREAM_TUNNEL: u8 = 0x02;
pub(crate) const STREAM_TUNNEL_MAP: u8 = 0x03;
pub const STREAM_TUNNEL_HTTP: u8 = 0x04;
pub(crate) const STREAM_ROUTE_REQUEST: u8 = 0x05;
pub(crate) const STREAM_PEER_DOWN: u8 = 0x06;
pub(crate) const STREAM_PEER_LEAVING: u8 = 0x07;
pub(crate) const STREAM_PLUGIN_CHANNEL: u8 = 0x08;
pub(crate) const STREAM_PLUGIN_BULK_TRANSFER: u8 = 0x09;
/// Reserved legacy mesh-plane config subscription stream ID.
///
/// Config and inventory control now live exclusively on `mesh-llm-control/1`;
/// keep 0x0b reserved so old wire values are not accidentally reused.
pub(crate) const STREAM_CONFIG_SUBSCRIBE: u8 = 0x0b;
/// Reserved legacy mesh-plane config push stream ID.
///
/// Config and inventory control now live exclusively on `mesh-llm-control/1`;
/// keep 0x0c reserved so old wire values are not accidentally reused.
pub(crate) const STREAM_CONFIG_PUSH: u8 = 0x0c;
pub(crate) const STREAM_SUBPROTOCOL: u8 = 0x0d;
const _: () = {
    let _ = ALPN_CONTROL_V1;
    let _ = STREAM_CONFIG_SUBSCRIBE;
    let _ = STREAM_CONFIG_PUSH;
    let _ = STREAM_SUBPROTOCOL;
};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ControlProtocol {
    ProtoV1,
}

#[derive(Debug, PartialEq)]
pub(crate) enum ControlFrameError {
    #[cfg(test)]
    OversizeFrame {
        size: usize,
    },
    BadGeneration {
        got: u32,
    },
    InvalidEndpointId {
        got: usize,
    },
    InvalidSenderId {
        got: usize,
    },
    MissingHttpPort,
    MissingControlOwnerId,
    InvalidConfigHashLength {
        got: usize,
    },
    InvalidSubprotocol,
    InvalidPublicKeyLength {
        got: usize,
    },
    MissingSignature,
    InvalidSignatureLength {
        got: usize,
    },
    MissingConfig,
    MissingControlEnvelope,
    MissingControlCommand,
    MissingControlResult,
    MissingControlOwnership,
    MissingRequestId,
    InvalidOwnerControlErrorCode {
        got: i32,
    },
    #[cfg(test)]
    DecodeError(String),
    #[cfg(test)]
    WrongStreamType {
        expected: u8,
        got: u8,
    },
    ForgedSender,
}

impl std::fmt::Display for ControlFrameError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            #[cfg(test)]
            ControlFrameError::OversizeFrame { size } => write!(
                f,
                "control frame too large: {} bytes (max {})",
                size, MAX_CONTROL_FRAME_BYTES
            ),
            ControlFrameError::BadGeneration { got } => write!(
                f,
                "bad protocol generation: expected {}, got {}",
                NODE_PROTOCOL_GENERATION, got
            ),
            ControlFrameError::InvalidEndpointId { got } => {
                write!(f, "invalid endpoint_id length: expected 32, got {}", got)
            }
            ControlFrameError::InvalidSenderId { got } => {
                write!(f, "invalid sender_id length: expected 32, got {}", got)
            }
            ControlFrameError::MissingHttpPort => {
                write!(f, "HOST-role peer annotation missing http_port")
            }
            ControlFrameError::MissingControlOwnerId => {
                write!(f, "owner control handshake missing owner_id")
            }
            ControlFrameError::InvalidConfigHashLength { got } => {
                write!(f, "invalid config_hash length: expected 32, got {}", got)
            }
            ControlFrameError::InvalidSubprotocol => {
                write!(f, "subprotocol entries require a non-empty name and major")
            }
            ControlFrameError::InvalidPublicKeyLength { got } => {
                write!(f, "invalid public key length: expected 32, got {}", got)
            }
            ControlFrameError::MissingSignature => write!(f, "config push missing signature"),
            ControlFrameError::InvalidSignatureLength { got } => {
                write!(f, "invalid signature length: expected 64, got {got}")
            }
            ControlFrameError::MissingConfig => {
                write!(f, "config field is required but missing")
            }
            ControlFrameError::MissingControlEnvelope => {
                write!(f, "owner control envelope requires exactly one payload")
            }
            ControlFrameError::MissingControlCommand => {
                write!(
                    f,
                    "owner control request requires exactly one command variant"
                )
            }
            ControlFrameError::MissingControlResult => {
                write!(
                    f,
                    "owner control response requires exactly one result variant"
                )
            }
            ControlFrameError::MissingControlOwnership => {
                write!(f, "owner control handshake missing ownership attestation")
            }
            ControlFrameError::MissingRequestId => {
                write!(f, "owner control request_id must be non-zero")
            }
            ControlFrameError::InvalidOwnerControlErrorCode { got } => {
                write!(f, "invalid owner control error code: {got}")
            }
            #[cfg(test)]
            ControlFrameError::DecodeError(msg) => write!(f, "protobuf decode error: {}", msg),
            #[cfg(test)]
            ControlFrameError::WrongStreamType { expected, got } => write!(
                f,
                "wrong stream type: expected {:#04x}, got {:#04x}",
                expected, got
            ),
            ControlFrameError::ForgedSender => {
                write!(f, "frame peer_id does not match QUIC connection identity")
            }
        }
    }
}

impl std::error::Error for ControlFrameError {}

pub(crate) trait ValidateControlFrame: prost::Message + Default + Sized {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::GossipFrame {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        if self.sender_id.len() != 32 {
            return Err(ControlFrameError::InvalidSenderId {
                got: self.sender_id.len(),
            });
        }
        for pa in &self.peers {
            validate_peer_announcement(pa)?;
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::TunnelMap {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.owner_peer_id.len() != 32 {
            return Err(ControlFrameError::InvalidEndpointId {
                got: self.owner_peer_id.len(),
            });
        }
        for entry in &self.entries {
            if entry.target_peer_id.len() != 32 {
                return Err(ControlFrameError::InvalidEndpointId {
                    got: entry.target_peer_id.len(),
                });
            }
        }
        Ok(())
    }
}
impl ValidateControlFrame for crate::proto::node::RouteTableRequest {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        if !self.requester_id.is_empty() && self.requester_id.len() != 32 {
            return Err(ControlFrameError::InvalidEndpointId {
                got: self.requester_id.len(),
            });
        }
        Ok(())
    }
}
impl ValidateControlFrame for crate::proto::node::RouteTable {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        for entry in &self.entries {
            if entry.endpoint_id.len() != 32 {
                return Err(ControlFrameError::InvalidEndpointId {
                    got: entry.endpoint_id.len(),
                });
            }
        }
        Ok(())
    }
}
impl ValidateControlFrame for crate::proto::node::PeerDown {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        if self.peer_id.len() != 32 {
            return Err(ControlFrameError::InvalidEndpointId {
                got: self.peer_id.len(),
            });
        }
        Ok(())
    }
}
impl ValidateControlFrame for crate::proto::node::PeerLeaving {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        if self.peer_id.len() != 32 {
            return Err(ControlFrameError::InvalidEndpointId {
                got: self.peer_id.len(),
            });
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlEnvelope {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        let payloads = [
            self.handshake.is_some(),
            self.request.is_some(),
            self.response.is_some(),
            self.error.is_some(),
        ];
        if payloads.into_iter().filter(|present| *present).count() != 1 {
            return Err(ControlFrameError::MissingControlEnvelope);
        }
        if let Some(handshake) = &self.handshake {
            handshake.validate_frame()?;
        }
        if let Some(request) = &self.request {
            request.validate_frame()?;
        }
        if let Some(response) = &self.response {
            response.validate_frame()?;
        }
        if let Some(error) = &self.error {
            error.validate_frame()?;
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlHandshake {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        let ownership = self
            .ownership
            .as_ref()
            .ok_or(ControlFrameError::MissingControlOwnership)?;
        if ownership.owner_id.trim().is_empty() {
            return Err(ControlFrameError::MissingControlOwnerId);
        }
        validate_public_key_length(ownership.owner_sign_public_key.len())?;
        validate_endpoint_id_length(ownership.node_endpoint_id.len())?;
        if ownership.signature.is_empty() {
            return Err(ControlFrameError::MissingSignature);
        }
        if ownership.signature.len() != 64 {
            return Err(ControlFrameError::InvalidSignatureLength {
                got: ownership.signature.len(),
            });
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlRequest {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.request_id == 0 {
            return Err(ControlFrameError::MissingRequestId);
        }
        let commands = [
            self.get_config.is_some(),
            self.watch_config.is_some(),
            self.apply_config.is_some(),
            self.refresh_inventory.is_some(),
        ];
        if commands.into_iter().filter(|present| *present).count() != 1 {
            return Err(ControlFrameError::MissingControlCommand);
        }
        if let Some(request) = &self.get_config {
            request.validate_frame()?;
        }
        if let Some(request) = &self.watch_config {
            request.validate_frame()?;
        }
        if let Some(request) = &self.apply_config {
            request.validate_frame()?;
        }
        if let Some(request) = &self.refresh_inventory {
            request.validate_frame()?;
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlResponse {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.request_id == 0 {
            return Err(ControlFrameError::MissingRequestId);
        }
        let results = [
            self.get_config.is_some(),
            self.watch_config.is_some(),
            self.apply_config.is_some(),
            self.refresh_inventory.is_some(),
        ];
        if results.into_iter().filter(|present| *present).count() != 1 {
            return Err(ControlFrameError::MissingControlResult);
        }
        if let Some(response) = &self.get_config {
            response.validate_frame()?;
        }
        if let Some(response) = &self.watch_config {
            response.validate_frame()?;
        }
        if let Some(response) = &self.apply_config {
            response.validate_frame()?;
        }
        if let Some(response) = &self.refresh_inventory {
            response.validate_frame()?;
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlError {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if matches!(
            crate::proto::node::OwnerControlErrorCode::try_from(self.code),
            Err(_) | Ok(crate::proto::node::OwnerControlErrorCode::Unspecified)
        ) {
            return Err(ControlFrameError::InvalidOwnerControlErrorCode { got: self.code });
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlGetConfigRequest {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.requester_node_id.len())?;
        validate_endpoint_id_length(self.target_node_id.len())?;
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlGetConfigResponse {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        self.snapshot
            .as_ref()
            .ok_or(ControlFrameError::MissingConfig)?
            .validate_frame()
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlWatchConfigRequest {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.requester_node_id.len())?;
        validate_endpoint_id_length(self.target_node_id.len())?;
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlWatchConfigResponse {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        let results = [
            self.accepted.is_some(),
            self.snapshot.is_some(),
            self.update.is_some(),
        ];
        if results.into_iter().filter(|present| *present).count() != 1 {
            return Err(ControlFrameError::MissingControlResult);
        }
        if let Some(accepted) = &self.accepted {
            accepted.validate_frame()?;
        }
        if let Some(snapshot) = &self.snapshot {
            snapshot.validate_frame()?;
        }
        if let Some(update) = &self.update {
            update.validate_frame()?;
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlWatchAccepted {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.target_node_id.len())?;
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlApplyConfigRequest {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.requester_node_id.len())?;
        validate_endpoint_id_length(self.target_node_id.len())?;
        if self.config.is_none() {
            return Err(ControlFrameError::MissingConfig);
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlApplyConfigResponse {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.success || !self.config_hash.is_empty() {
            validate_config_hash_length(self.config_hash.len())?;
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlRefreshInventoryRequest {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.requester_node_id.len())?;
        validate_endpoint_id_length(self.target_node_id.len())?;
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlRefreshInventoryResponse {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        self.snapshot
            .as_ref()
            .ok_or(ControlFrameError::MissingConfig)?
            .validate_frame()
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlConfigSnapshot {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.node_id.len())?;
        validate_config_hash_length(self.config_hash.len())?;
        if self.config.is_none() {
            return Err(ControlFrameError::MissingConfig);
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::OwnerControlConfigUpdate {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        validate_endpoint_id_length(self.node_id.len())?;
        validate_config_hash_length(self.config_hash.len())?;
        if self.config.is_none() {
            return Err(ControlFrameError::MissingConfig);
        }
        Ok(())
    }
}

impl ValidateControlFrame for crate::proto::node::MeshSubprotocolOpen {
    fn validate_frame(&self) -> Result<(), ControlFrameError> {
        if self.gen != NODE_PROTOCOL_GENERATION {
            return Err(ControlFrameError::BadGeneration { got: self.gen });
        }
        if self.name.trim().is_empty() || self.major == 0 {
            return Err(ControlFrameError::InvalidSubprotocol);
        }
        Ok(())
    }
}

pub(crate) fn validate_peer_announcement(
    pa: &crate::proto::node::PeerAnnouncement,
) -> Result<(), ControlFrameError> {
    if pa.endpoint_id.len() != 32 {
        return Err(ControlFrameError::InvalidEndpointId {
            got: pa.endpoint_id.len(),
        });
    }
    if pa.role == crate::proto::node::NodeRole::Host as i32 && pa.http_port.is_none() {
        return Err(ControlFrameError::MissingHttpPort);
    }
    for subprotocol in &pa.subprotocols {
        if subprotocol.name.trim().is_empty() || subprotocol.major == 0 {
            return Err(ControlFrameError::InvalidSubprotocol);
        }
    }
    Ok(())
}

fn validate_endpoint_id_length(len: usize) -> Result<(), ControlFrameError> {
    if len != 32 {
        return Err(ControlFrameError::InvalidEndpointId { got: len });
    }
    Ok(())
}

fn validate_config_hash_length(len: usize) -> Result<(), ControlFrameError> {
    if len != 32 {
        return Err(ControlFrameError::InvalidConfigHashLength { got: len });
    }
    Ok(())
}

fn validate_public_key_length(len: usize) -> Result<(), ControlFrameError> {
    if len != 32 {
        return Err(ControlFrameError::InvalidPublicKeyLength { got: len });
    }
    Ok(())
}

pub(crate) fn protocol_from_alpn(alpn: &[u8]) -> ControlProtocol {
    let _ = alpn;
    ControlProtocol::ProtoV1
}

pub(crate) fn connection_protocol(conn: &Connection) -> ControlProtocol {
    protocol_from_alpn(conn.alpn())
}

pub(crate) async fn connect_mesh(endpoint: &Endpoint, addr: EndpointAddr) -> Result<Connection> {
    let connecting = endpoint.connect(addr, ALPN_V1).await?;
    Ok(connecting)
}

pub(crate) async fn write_len_prefixed(
    send: &mut iroh::endpoint::SendStream,
    body: &[u8],
) -> Result<()> {
    send.write_all(&(body.len() as u32).to_le_bytes()).await?;
    send.write_all(body).await?;
    Ok(())
}

pub(crate) async fn read_len_prefixed(recv: &mut iroh::endpoint::RecvStream) -> Result<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    recv.read_exact(&mut len_buf).await?;
    let len = u32::from_le_bytes(len_buf) as usize;
    if len > MAX_CONTROL_FRAME_BYTES {
        anyhow::bail!("control frame too large: {} bytes", len);
    }
    let mut buf = vec![0u8; len];
    recv.read_exact(&mut buf).await?;
    Ok(buf)
}

pub(crate) async fn write_gossip_payload(
    send: &mut iroh::endpoint::SendStream,
    protocol: ControlProtocol,
    anns: &[PeerAnnouncement],
    sender_id: EndpointId,
) -> Result<()> {
    let _ = protocol;
    let frame = build_gossip_frame(anns, sender_id);
    write_len_prefixed(send, &frame.encode_to_vec()).await?;
    Ok(())
}

pub(crate) fn decode_gossip_payload(
    protocol: ControlProtocol,
    remote: EndpointId,
    buf: &[u8],
) -> Result<Vec<(EndpointAddr, PeerAnnouncement)>> {
    let _ = protocol;
    let frame = crate::proto::node::GossipFrame::decode(buf)
        .map_err(|e| anyhow::anyhow!("gossip decode from {}: {e}", remote.fmt_short()))?;
    frame
        .validate_frame()
        .map_err(|e| anyhow::anyhow!("invalid gossip frame from {}: {e}", remote.fmt_short()))?;
    if frame.sender_id.as_slice() != remote.as_bytes() {
        anyhow::bail!(
            "gossip sender_id mismatch from {}: connection identity does not match frame sender_id",
            remote.fmt_short()
        );
    }
    Ok(frame
        .peers
        .iter()
        .filter_map(proto_ann_to_local)
        .collect::<Vec<_>>())
}

#[cfg(test)]
pub(crate) fn encode_control_frame(stream_type: u8, msg: &impl prost::Message) -> Vec<u8> {
    let proto_bytes = msg.encode_to_vec();
    let len = proto_bytes.len() as u32;
    let mut buf = Vec::with_capacity(1 + 4 + proto_bytes.len());
    buf.push(stream_type);
    buf.extend_from_slice(&len.to_le_bytes());
    buf.extend_from_slice(&proto_bytes);
    buf
}

#[cfg(test)]
pub(crate) fn decode_control_frame<T: ValidateControlFrame>(
    expected_stream_type: u8,
    data: &[u8],
) -> Result<T, ControlFrameError> {
    const HEADER_LEN: usize = 5;
    if data.len() < HEADER_LEN {
        return Err(ControlFrameError::DecodeError(format!(
            "frame too short: {} bytes (minimum {})",
            data.len(),
            HEADER_LEN
        )));
    }
    let actual_type = data[0];
    if actual_type != expected_stream_type {
        return Err(ControlFrameError::WrongStreamType {
            expected: expected_stream_type,
            got: actual_type,
        });
    }
    let len = u32::from_le_bytes(data[1..5].try_into().unwrap()) as usize;
    if len > MAX_CONTROL_FRAME_BYTES {
        return Err(ControlFrameError::OversizeFrame { size: len });
    }
    let proto_bytes = data.get(5..5 + len).ok_or_else(|| {
        ControlFrameError::DecodeError(format!(
            "frame truncated: header says {} bytes but only {} available",
            len,
            data.len().saturating_sub(5)
        ))
    })?;
    let msg = T::decode(proto_bytes).map_err(|e| ControlFrameError::DecodeError(e.to_string()))?;
    msg.validate_frame()?;
    Ok(msg)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::crypto::OwnershipSummary;
    use crate::mesh::{resolve_peer_down, resolve_peer_leaving, PeerInfo};
    use crate::proto::node::{
        ConfiguredModelRef, GossipFrame, MeshSubprotocolOpen, NodeConfigSnapshot, NodeGpuConfig,
        NodeModelEntry, NodePluginEntry, NodeRole, OwnerControlError, OwnerControlErrorCode,
        OwnerControlHandshake, PeerAnnouncement, RouteTableRequest, SignedNodeOwnership,
    };
    use iroh::{EndpointAddr, EndpointId, SecretKey};
    use std::collections::{HashMap, HashSet};

    fn make_valid_gossip_frame() -> GossipFrame {
        GossipFrame {
            gen: NODE_PROTOCOL_GENERATION,
            sender_id: vec![0u8; 32],
            peers: vec![PeerAnnouncement {
                endpoint_id: vec![0u8; 32],
                role: NodeRole::Worker as i32,
                ..Default::default()
            }],
        }
    }

    fn make_config_snapshot() -> NodeConfigSnapshot {
        NodeConfigSnapshot {
            version: 1,
            gpu: Some(NodeGpuConfig {
                assignment: crate::proto::node::GpuAssignment::Auto as i32,
            }),
            models: vec![NodeModelEntry {
                model: "Qwen3-8B".to_string(),
                mmproj: Some("mmproj-cut".to_string()),
                ctx_size: Some(8192),
                gpu_id: Some("pci:0000:65:00.0".to_string()),
                model_ref: Some(ConfiguredModelRef {
                    declared_ref: "Qwen3-8B".to_string(),
                    source_kind: None,
                    revision: None,
                }),
                mmproj_ref: Some(ConfiguredModelRef {
                    declared_ref: "mmproj-cut".to_string(),
                    source_kind: None,
                    revision: None,
                }),
            }],
            plugins: vec![NodePluginEntry {
                name: "blackboard".to_string(),
                enabled: Some(true),
                command: Some("mesh-llm".to_string()),
                args: vec!["--plugin".to_string(), "blackboard".to_string()],
            }],
        }
    }

    fn make_valid_owner_control_handshake() -> OwnerControlHandshake {
        OwnerControlHandshake {
            ownership: Some(SignedNodeOwnership {
                version: 1,
                cert_id: "cert-1".to_string(),
                owner_id: "owner-1".to_string(),
                owner_sign_public_key: vec![0x11; 32],
                node_endpoint_id: vec![0x22; 32],
                issued_at_unix_ms: 1,
                expires_at_unix_ms: 2,
                node_label: Some("node-01".to_string()),
                hostname_hint: Some("node-01".to_string()),
                signature: vec![0x33; 64],
            }),
        }
    }

    #[test]
    fn owner_control_handshake_empty_owner_id_uses_handshake_error() {
        let mut handshake = make_valid_owner_control_handshake();
        handshake
            .ownership
            .as_mut()
            .expect("test handshake must include ownership")
            .owner_id = "   ".to_string();

        let err = handshake
            .validate_frame()
            .expect_err("handshake with blank owner_id must be rejected");
        assert!(matches!(err, ControlFrameError::MissingControlOwnerId));
        assert_eq!(err.to_string(), "owner control handshake missing owner_id");
    }

    #[test]
    fn owner_control_error_rejects_invalid_error_code() {
        for code in [OwnerControlErrorCode::Unspecified as i32, 9999] {
            let err = OwnerControlError {
                code,
                message: "invalid".to_string(),
                request_id: Some(1),
                current_revision: None,
            }
            .validate_frame()
            .expect_err("invalid owner-control error code must be rejected");
            assert!(matches!(
                err,
                ControlFrameError::InvalidOwnerControlErrorCode { got } if got == code
            ));
            assert_eq!(
                err.to_string(),
                format!("invalid owner control error code: {code}")
            );
        }
    }

    fn make_test_peer_info(peer_id: EndpointId) -> PeerInfo {
        PeerInfo {
            id: peer_id,
            addr: EndpointAddr {
                id: peer_id,
                addrs: Default::default(),
            },
            role: crate::mesh::NodeRole::Worker,
            first_joined_mesh_ts: None,
            models: vec![],
            vram_bytes: 0,
            rtt_ms: None,
            model_source: None,
            serving_models: vec![],
            hosted_models: vec![],
            hosted_models_known: false,
            available_models: vec![],
            requested_models: vec![],
            explicit_model_interests: vec![],
            last_seen: std::time::Instant::now(),
            last_mentioned: std::time::Instant::now(),
            version: None,
            gpu_name: None,
            hostname: None,
            is_soc: None,
            gpu_vram: None,
            gpu_reserved_bytes: None,
            gpu_mem_bandwidth_gbps: None,
            gpu_compute_tflops_fp32: None,
            gpu_compute_tflops_fp16: None,
            available_model_metadata: vec![],
            experts_summary: None,
            available_model_sizes: HashMap::new(),
            served_model_descriptors: vec![],
            served_model_runtime: vec![],
            owner_attestation: None,
            artifact_transfer_supported: false,
            stage_protocol_generation_supported: false,
            stage_status_list_supported: false,
            owner_summary: OwnershipSummary::default(),
            display_rtt: None,
            propagated_latency: None,
        }
    }

    #[test]
    fn protocol_from_alpn_defaults_to_v1() {
        assert_eq!(protocol_from_alpn(ALPN_V1), ControlProtocol::ProtoV1);
        assert_eq!(
            protocol_from_alpn(b"mesh-llm/999"),
            ControlProtocol::ProtoV1
        );
    }
    #[test]
    fn control_frame_roundtrip() {
        let frame = make_valid_gossip_frame();
        let encoded = encode_control_frame(STREAM_GOSSIP, &frame);
        let decoded: GossipFrame = decode_control_frame(STREAM_GOSSIP, &encoded)
            .expect("valid gossip frame must decode successfully");
        assert_eq!(decoded.gen, NODE_PROTOCOL_GENERATION);
        assert_eq!(decoded.peers.len(), 1);
        assert_eq!(decoded.peers[0].endpoint_id, vec![0u8; 32]);
        assert_eq!(decoded.peers[0].role, NodeRole::Worker as i32);
    }

    #[test]
    fn mesh_subprotocol_open_roundtrips_and_validates() {
        let open = MeshSubprotocolOpen {
            gen: NODE_PROTOCOL_GENERATION,
            name: skippy_protocol::STAGE_SUBPROTOCOL_NAME.to_string(),
            major: skippy_protocol::STAGE_SUBPROTOCOL_MAJOR,
        };
        let encoded = encode_control_frame(STREAM_SUBPROTOCOL, &open);
        let decoded: MeshSubprotocolOpen =
            decode_control_frame(STREAM_SUBPROTOCOL, &encoded).unwrap();
        assert_eq!(decoded.name, skippy_protocol::STAGE_SUBPROTOCOL_NAME);
        assert_eq!(decoded.major, skippy_protocol::STAGE_SUBPROTOCOL_MAJOR);

        let bad = MeshSubprotocolOpen {
            gen: NODE_PROTOCOL_GENERATION,
            name: String::new(),
            major: skippy_protocol::STAGE_SUBPROTOCOL_MAJOR,
        };
        let encoded = encode_control_frame(STREAM_SUBPROTOCOL, &bad);
        let err = decode_control_frame::<MeshSubprotocolOpen>(STREAM_SUBPROTOCOL, &encoded)
            .expect_err("empty subprotocol names must be rejected");
        assert!(matches!(err, ControlFrameError::InvalidSubprotocol));
    }

    #[test]
    fn proto_v1_route_table_rejects_bad_generation_or_legacy_payload() {
        use crate::proto::node::RouteTable;

        let zero_gen_req = RouteTableRequest {
            requester_id: vec![0u8; 32],
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_ROUTE_REQUEST, &zero_gen_req);
        let err = decode_control_frame::<RouteTableRequest>(STREAM_ROUTE_REQUEST, &encoded)
            .expect_err("request gen=0 must be rejected");
        assert!(
            matches!(err, ControlFrameError::BadGeneration { got: 0 }),
            "expected BadGeneration{{got:0}}, got {:?}",
            err
        );

        let wrong_gen_req = RouteTableRequest {
            requester_id: vec![0u8; 32],
            gen: 99,
        };
        let encoded = encode_control_frame(STREAM_ROUTE_REQUEST, &wrong_gen_req);
        let err = decode_control_frame::<RouteTableRequest>(STREAM_ROUTE_REQUEST, &encoded)
            .expect_err("request gen=99 must be rejected");
        assert!(
            matches!(err, ControlFrameError::BadGeneration { got: 99 }),
            "expected BadGeneration{{got:99}}, got {:?}",
            err
        );

        let bad_gen_response = RouteTable {
            entries: vec![],
            mesh_id: None,
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_ROUTE_REQUEST, &bad_gen_response);
        let err = decode_control_frame::<RouteTable>(STREAM_ROUTE_REQUEST, &encoded)
            .expect_err("response gen=0 must be rejected");
        assert!(
            matches!(err, ControlFrameError::BadGeneration { got: 0 }),
            "expected BadGeneration{{got:0}} for response, got {:?}",
            err
        );

        let wrong_gen_response = RouteTable {
            entries: vec![],
            mesh_id: None,
            gen: 42,
        };
        let encoded = encode_control_frame(STREAM_ROUTE_REQUEST, &wrong_gen_response);
        let err = decode_control_frame::<RouteTable>(STREAM_ROUTE_REQUEST, &encoded)
            .expect_err("response gen=42 must be rejected");
        assert!(
            matches!(err, ControlFrameError::BadGeneration { got: 42 }),
            "expected BadGeneration{{got:42}} for response, got {:?}",
            err
        );

        let legacy_json = b"{\"hosts\":[],\"mesh_id\":null}";
        let mut fake_frame = vec![STREAM_ROUTE_REQUEST];
        fake_frame.extend_from_slice(&(legacy_json.len() as u32).to_le_bytes());
        fake_frame.extend_from_slice(legacy_json);
        let err = decode_control_frame::<RouteTableRequest>(STREAM_ROUTE_REQUEST, &fake_frame)
            .expect_err("legacy JSON payload must be rejected");
        assert!(
            matches!(err, ControlFrameError::DecodeError(_)),
            "expected DecodeError for JSON payload, got {:?}",
            err
        );
    }

    #[test]
    fn peer_lifecycle_messages_roundtrip() {
        use crate::proto::node::{PeerDown, PeerLeaving};

        let leaving_id = EndpointId::from(SecretKey::from_bytes(&[0x55; 32]).public());

        let mut peers: HashMap<EndpointId, PeerInfo> = HashMap::new();
        peers.insert(leaving_id, make_test_peer_info(leaving_id));
        let mut connection_ids: HashSet<EndpointId> = HashSet::new();
        connection_ids.insert(leaving_id);

        let leaving_msg = PeerLeaving {
            peer_id: leaving_id.as_bytes().to_vec(),
            gen: NODE_PROTOCOL_GENERATION,
        };
        let encoded = encode_control_frame(STREAM_PEER_LEAVING, &leaving_msg);
        let decoded_leaving: PeerLeaving = decode_control_frame(STREAM_PEER_LEAVING, &encoded)
            .expect("valid PeerLeaving must decode");

        let accepted_id = resolve_peer_leaving(leaving_id, &decoded_leaving)
            .expect("PeerLeaving from sender itself must be accepted");

        peers.remove(&accepted_id);
        connection_ids.remove(&accepted_id);

        assert!(
            !peers.contains_key(&leaving_id),
            "leaving peer must be removed from peers after accepted PeerLeaving"
        );
        assert!(
            !connection_ids.contains(&leaving_id),
            "leaving peer must be removed from connections after accepted PeerLeaving"
        );

        let self_id = EndpointId::from(SecretKey::from_bytes(&[0xAA; 32]).public());
        let dead_id = EndpointId::from(SecretKey::from_bytes(&[0xBB; 32]).public());

        let mut peers: HashMap<EndpointId, PeerInfo> = HashMap::new();
        peers.insert(dead_id, make_test_peer_info(dead_id));
        let mut connection_ids: HashSet<EndpointId> = HashSet::new();
        connection_ids.insert(dead_id);

        let down_msg = PeerDown {
            peer_id: dead_id.as_bytes().to_vec(),
            gen: NODE_PROTOCOL_GENERATION,
        };
        let encoded = encode_control_frame(STREAM_PEER_DOWN, &down_msg);
        let decoded_down: PeerDown =
            decode_control_frame(STREAM_PEER_DOWN, &encoded).expect("valid PeerDown must decode");

        let result = resolve_peer_down(self_id, dead_id, true);
        assert_eq!(
            result,
            Some(dead_id),
            "confirmed-unreachable peer must be returned for removal"
        );

        if let Some(id) = result {
            peers.remove(&id);
            connection_ids.remove(&id);
        }

        assert!(
            !peers.contains_key(&dead_id),
            "dead peer must be removed from peers when confirmed unreachable"
        );
        assert!(
            !connection_ids.contains(&dead_id),
            "dead peer must be removed from connections when confirmed unreachable"
        );

        assert_eq!(decoded_down.gen, NODE_PROTOCOL_GENERATION);
    }

    #[test]
    fn peer_lifecycle_rejects_forged_sender_or_unverified_down() {
        use crate::proto::node::{PeerDown, PeerLeaving};

        let valid_peer_bytes = EndpointId::from(SecretKey::from_bytes(&[0x77; 32]).public())
            .as_bytes()
            .to_vec();

        let bad_gen_down = PeerDown {
            peer_id: valid_peer_bytes.clone(),
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_PEER_DOWN, &bad_gen_down);
        let err = decode_control_frame::<PeerDown>(STREAM_PEER_DOWN, &encoded)
            .expect_err("PeerDown gen=0 must be rejected");
        assert!(
            matches!(err, ControlFrameError::BadGeneration { got: 0 }),
            "expected BadGeneration{{got:0}} for PeerDown, got {:?}",
            err
        );

        let bad_gen_leaving = PeerLeaving {
            peer_id: valid_peer_bytes.clone(),
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_PEER_LEAVING, &bad_gen_leaving);
        let err = decode_control_frame::<PeerLeaving>(STREAM_PEER_LEAVING, &encoded)
            .expect_err("PeerLeaving gen=0 must be rejected");
        assert!(
            matches!(err, ControlFrameError::BadGeneration { got: 0 }),
            "expected BadGeneration{{got:0}} for PeerLeaving, got {:?}",
            err
        );

        let remote_id = EndpointId::from(SecretKey::from_bytes(&[0x11; 32]).public());
        let victim_id = EndpointId::from(SecretKey::from_bytes(&[0x22; 32]).public());

        let mut peers: HashMap<EndpointId, PeerInfo> = HashMap::new();
        peers.insert(victim_id, make_test_peer_info(victim_id));

        let forged = PeerLeaving {
            peer_id: victim_id.as_bytes().to_vec(),
            gen: NODE_PROTOCOL_GENERATION,
        };
        let encoded = encode_control_frame(STREAM_PEER_LEAVING, &forged);
        let decoded: PeerLeaving = decode_control_frame(STREAM_PEER_LEAVING, &encoded)
            .expect("structurally valid PeerLeaving must decode");

        let err = resolve_peer_leaving(remote_id, &decoded)
            .expect_err("forged PeerLeaving (peer_id != remote) must be rejected");
        assert!(
            matches!(err, crate::protocol::ControlFrameError::ForgedSender),
            "expected ForgedSender, got {:?}",
            err
        );

        assert!(
            peers.contains_key(&victim_id),
            "victim peer must NOT be removed when PeerLeaving is forged"
        );

        let self_id = EndpointId::from(SecretKey::from_bytes(&[0x33; 32]).public());
        let still_alive_id = EndpointId::from(SecretKey::from_bytes(&[0x44; 32]).public());

        let mut peers: HashMap<EndpointId, PeerInfo> = HashMap::new();
        peers.insert(still_alive_id, make_test_peer_info(still_alive_id));

        let result = resolve_peer_down(self_id, still_alive_id, false);
        assert!(
            result.is_none(),
            "PeerDown must not trigger removal when peer is still reachable"
        );

        assert!(
            peers.contains_key(&still_alive_id),
            "reachable peer must NOT be removed after PeerDown with should_remove=false"
        );
    }

    #[test]
    fn proto_v1_control_frames_reject_legacy_json_and_wrong_gen() {
        use crate::proto::node::{PeerDown, PeerLeaving};

        // JSON bytes that look plausible for the old wire format on each stream
        let json_gossip = b"[{\"addr\":{\"id\":\"aabbcc\",\"addrs\":[]}}]";
        let json_tunnel_map = b"{\"owner\":\"aabbcc\",\"entries\":[]}";
        let json_route = b"{\"hosts\":[],\"mesh_id\":null}";
        let json_peer_down = b"\"aabbccdd\"";
        let json_peer_leaving = b"\"aabbccdd\"";

        // All migrated streams must reject legacy JSON with DecodeError
        for (stream_type, json_bytes) in [
            (STREAM_GOSSIP, json_gossip.as_slice()),
            (STREAM_TUNNEL_MAP, json_tunnel_map.as_slice()),
            (STREAM_ROUTE_REQUEST, json_route.as_slice()),
            (STREAM_PEER_DOWN, json_peer_down.as_slice()),
            (STREAM_PEER_LEAVING, json_peer_leaving.as_slice()),
        ] {
            let mut frame = vec![stream_type];
            frame.extend_from_slice(&(json_bytes.len() as u32).to_le_bytes());
            frame.extend_from_slice(json_bytes);
            // Each stream uses its own message type for decode; we test gossip and route
            // request specifically since those carry gen validation too.
            if stream_type == STREAM_GOSSIP {
                let err = decode_control_frame::<GossipFrame>(stream_type, &frame).expect_err(
                    &format!("JSON must be rejected on stream {:#04x}", stream_type),
                );
                assert!(
                    matches!(err, ControlFrameError::DecodeError(_)),
                    "stream {:#04x}: expected DecodeError for JSON, got {:?}",
                    stream_type,
                    err
                );
            } else if stream_type == STREAM_ROUTE_REQUEST {
                let err =
                    decode_control_frame::<RouteTableRequest>(stream_type, &frame).expect_err(
                        &format!("JSON must be rejected on stream {:#04x}", stream_type),
                    );
                assert!(
                    matches!(err, ControlFrameError::DecodeError(_)),
                    "stream {:#04x}: expected DecodeError for JSON, got {:?}",
                    stream_type,
                    err
                );
            }
            // STREAM_TUNNEL_MAP, STREAM_PEER_DOWN, STREAM_PEER_LEAVING: JSON fails prost
            // decode which returns DecodeError — verified via the decode_control_frame
            // path used in the existing per-stream tests.
        }

        // All migrated streams must also reject gen=0 and gen=99 where gen is checked
        let bad_gen_gossip = GossipFrame {
            gen: 0,
            sender_id: vec![],
            peers: vec![PeerAnnouncement {
                endpoint_id: vec![0u8; 32],
                role: NodeRole::Worker as i32,
                ..Default::default()
            }],
        };
        let encoded = encode_control_frame(STREAM_GOSSIP, &bad_gen_gossip);
        let err = decode_control_frame::<GossipFrame>(STREAM_GOSSIP, &encoded)
            .expect_err("GossipFrame gen=0 must be rejected");
        assert!(matches!(err, ControlFrameError::BadGeneration { got: 0 }));

        let bad_gen_req = RouteTableRequest {
            requester_id: vec![0u8; 32],
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_ROUTE_REQUEST, &bad_gen_req);
        let err = decode_control_frame::<RouteTableRequest>(STREAM_ROUTE_REQUEST, &encoded)
            .expect_err("RouteTableRequest gen=0 must be rejected");
        assert!(matches!(err, ControlFrameError::BadGeneration { got: 0 }));

        let bad_gen_down = PeerDown {
            peer_id: vec![0u8; 32],
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_PEER_DOWN, &bad_gen_down);
        let err = decode_control_frame::<PeerDown>(STREAM_PEER_DOWN, &encoded)
            .expect_err("PeerDown gen=0 must be rejected");
        assert!(matches!(err, ControlFrameError::BadGeneration { got: 0 }));

        let bad_gen_leaving = PeerLeaving {
            peer_id: vec![0u8; 32],
            gen: 0,
        };
        let encoded = encode_control_frame(STREAM_PEER_LEAVING, &bad_gen_leaving);
        let err = decode_control_frame::<PeerLeaving>(STREAM_PEER_LEAVING, &encoded)
            .expect_err("PeerLeaving gen=0 must be rejected");
        assert!(matches!(err, ControlFrameError::BadGeneration { got: 0 }));

        // Wrong gen (e.g. 2) also rejected
        let wrong_gen_gossip = GossipFrame {
            gen: 2,
            sender_id: vec![0u8; 32],
            peers: vec![PeerAnnouncement {
                endpoint_id: vec![0u8; 32],
                role: NodeRole::Worker as i32,
                ..Default::default()
            }],
        };
        let encoded = encode_control_frame(STREAM_GOSSIP, &wrong_gen_gossip);
        let err = decode_control_frame::<GossipFrame>(STREAM_GOSSIP, &encoded)
            .expect_err("GossipFrame gen=2 (future version) must be rejected");
        assert!(matches!(err, ControlFrameError::BadGeneration { got: 2 }));
    }

    #[test]
    fn owner_fields_roundtrip_through_proto_announcement() {
        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xAB; 32]).public());
        let ann = super::PeerAnnouncement {
            addr: iroh::EndpointAddr {
                id: peer_id,
                addrs: Default::default(),
            },
            role: super::NodeRole::Worker,
            first_joined_mesh_ts: None,
            models: vec![],
            vram_bytes: 0,
            model_source: None,
            serving_models: vec![],
            hosted_models: None,
            available_models: vec![],
            requested_models: vec![],
            explicit_model_interests: vec![],
            version: None,
            model_demand: HashMap::new(),
            mesh_id: None,
            gpu_name: None,
            hostname: None,
            is_soc: None,
            gpu_vram: None,
            gpu_reserved_bytes: None,
            gpu_mem_bandwidth_gbps: None,
            gpu_compute_tflops_fp32: None,
            gpu_compute_tflops_fp16: None,
            available_model_metadata: vec![],
            experts_summary: None,
            available_model_sizes: HashMap::new(),
            served_model_descriptors: vec![],
            served_model_runtime: vec![],
            owner_attestation: Some(crate::crypto::SignedNodeOwnership {
                claim: crate::crypto::NodeOwnershipClaim {
                    version: 1,
                    cert_id: "cert-123".to_string(),
                    owner_id: "owner-abc".to_string(),
                    owner_sign_public_key: "11".repeat(32),
                    node_endpoint_id: "22".repeat(32),
                    issued_at_unix_ms: 10,
                    expires_at_unix_ms: 20,
                    node_label: Some("studio".to_string()),
                    hostname_hint: Some("worker-01".to_string()),
                },
                signature: "33".repeat(64),
            }),
            artifact_transfer_supported: true,
            stage_protocol_generation_supported: true,
            stage_status_list_supported: true,
            latency_ms: None,
            latency_source: None,
            latency_age_ms: None,
            latency_observer_id: None,
        };
        let proto_pa = local_ann_to_proto_ann(&ann);
        let skippy = proto_pa
            .subprotocols
            .iter()
            .find(|subprotocol| subprotocol.name == skippy_protocol::STAGE_SUBPROTOCOL_NAME)
            .expect("skippy-stage subprotocol should be advertised");
        assert_eq!(skippy.major, skippy_protocol::STAGE_SUBPROTOCOL_MAJOR);
        assert!(
            skippy
                .features
                .iter()
                .any(|feature| feature
                    == skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_ARTIFACT_TRANSFER)
        );
        assert!(skippy
            .features
            .iter()
            .any(|feature| feature == skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST));
        assert!(skippy.features.iter().any(|feature| feature
            == skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V2));
        assert_eq!(
            proto_pa
                .owner_attestation
                .as_ref()
                .map(|att| att.owner_id.as_str()),
            Some("owner-abc")
        );

        let (_, roundtripped) =
            proto_ann_to_local(&proto_pa).expect("proto_ann_to_local must succeed");
        assert!(roundtripped.artifact_transfer_supported);
        assert!(roundtripped.stage_status_list_supported);
        assert!(roundtripped.stage_protocol_generation_supported);
        let roundtripped = roundtripped
            .owner_attestation
            .expect("owner attestation must round-trip");
        assert_eq!(roundtripped.claim.owner_id, "owner-abc");
        assert_eq!(roundtripped.claim.cert_id, "cert-123");
        assert_eq!(roundtripped.claim.node_label.as_deref(), Some("studio"));
    }

    #[test]
    fn proto_announcement_without_current_stage_generation_is_not_stage_compatible() {
        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xCD; 32]).public());
        let proto_pa = crate::proto::node::PeerAnnouncement {
            endpoint_id: peer_id.as_bytes().to_vec(),
            role: crate::proto::node::NodeRole::Worker as i32,
            subprotocols: vec![crate::proto::node::MeshSubprotocol {
                name: skippy_protocol::STAGE_SUBPROTOCOL_NAME.to_string(),
                major: skippy_protocol::STAGE_SUBPROTOCOL_MAJOR,
                features: vec![
                    skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_CONTROL.to_string(),
                    skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST.to_string(),
                ],
            }],
            ..Default::default()
        };

        let (_, ann) = proto_ann_to_local(&proto_pa).expect("proto announcement should decode");

        assert!(!ann.stage_protocol_generation_supported);
        assert!(ann.stage_status_list_supported);
    }

    #[test]
    fn proto_announcement_without_stage_control_is_not_stage_compatible() {
        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xCE; 32]).public());
        let proto_pa = crate::proto::node::PeerAnnouncement {
            endpoint_id: peer_id.as_bytes().to_vec(),
            role: crate::proto::node::NodeRole::Worker as i32,
            subprotocols: vec![crate::proto::node::MeshSubprotocol {
                name: skippy_protocol::STAGE_SUBPROTOCOL_NAME.to_string(),
                major: skippy_protocol::STAGE_SUBPROTOCOL_MAJOR,
                features: vec![
                    skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V2
                        .to_string(),
                ],
            }],
            ..Default::default()
        };

        let (_, ann) = proto_ann_to_local(&proto_pa).expect("proto announcement should decode");

        assert!(!ann.stage_protocol_generation_supported);
    }

    #[test]
    fn test_proto_round_trip_with_bandwidth_and_tflops() {
        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xBC; 32]).public());
        let ann = super::PeerAnnouncement {
            addr: EndpointAddr {
                id: peer_id,
                addrs: Default::default(),
            },
            role: super::NodeRole::Host { http_port: 3131 },
            first_joined_mesh_ts: None,
            models: vec!["Qwen".to_string()],
            vram_bytes: 48_000_000_000,
            model_source: Some("Qwen.gguf".to_string()),
            serving_models: vec!["Qwen".to_string()],
            hosted_models: Some(vec!["Qwen".to_string()]),
            available_models: vec![],
            requested_models: vec![],
            explicit_model_interests: vec!["Qwen/Qwen3-Coder-Next-GGUF@main:Q4_K_M".to_string()],
            version: Some("0.52.0".to_string()),
            model_demand: HashMap::new(),
            mesh_id: Some("mesh-proto-roundtrip".to_string()),
            gpu_name: Some("NVIDIA A100".to_string()),
            hostname: Some("worker-01".to_string()),
            is_soc: Some(false),
            gpu_vram: Some("51539607552".to_string()),
            gpu_reserved_bytes: Some("1073741824".to_string()),
            gpu_mem_bandwidth_gbps: Some("1948.70".to_string()),
            gpu_compute_tflops_fp32: Some("19.50".to_string()),
            gpu_compute_tflops_fp16: Some("312.00".to_string()),
            available_model_metadata: vec![],
            experts_summary: None,
            available_model_sizes: HashMap::new(),
            served_model_descriptors: vec![],
            served_model_runtime: vec![],
            owner_attestation: None,
            artifact_transfer_supported: true,
            stage_protocol_generation_supported: true,
            stage_status_list_supported: true,
            latency_ms: None,
            latency_source: None,
            latency_age_ms: None,
            latency_observer_id: None,
        };

        let proto_pa = local_ann_to_proto_ann(&ann);
        let hardware = proto_pa
            .hardware
            .as_ref()
            .expect("hardware info must be present");
        assert_eq!(hardware.hostname.as_deref(), Some("worker-01"));
        assert_eq!(hardware.is_soc, Some(false));
        assert_eq!(hardware.gpus.len(), 1);
        assert_eq!(hardware.gpus[0].name.as_deref(), Some("NVIDIA A100"));
        assert_eq!(hardware.gpus[0].vram_bytes.as_deref(), Some("51539607552"));
        assert_eq!(
            hardware.gpus[0].reserved_bytes.as_deref(),
            Some("1073741824")
        );
        assert_eq!(
            hardware.gpus[0].mem_bandwidth_gbps.as_deref(),
            Some("1948.70")
        );
        assert_eq!(
            hardware.gpus[0].compute_tflops_fp32.as_deref(),
            Some("19.50")
        );
        assert_eq!(
            hardware.gpus[0].compute_tflops_fp16.as_deref(),
            Some("312.00")
        );

        let (_, roundtripped) =
            proto_ann_to_local(&proto_pa).expect("proto_ann_to_local must succeed");
        assert_eq!(
            roundtripped.gpu_reserved_bytes.as_deref(),
            Some("1073741824")
        );
        assert_eq!(
            roundtripped.gpu_mem_bandwidth_gbps.as_deref(),
            Some("1948.70")
        );
        assert_eq!(
            roundtripped.gpu_compute_tflops_fp32.as_deref(),
            Some("19.50")
        );
        assert_eq!(
            roundtripped.gpu_compute_tflops_fp16.as_deref(),
            Some("312.00")
        );
        assert_eq!(
            roundtripped.explicit_model_interests,
            vec!["Qwen/Qwen3-Coder-Next-GGUF@main:Q4_K_M".to_string()]
        );
    }

    #[test]
    fn test_proto_backward_compat_missing_tflops() {
        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xCD; 32]).public());
        let proto_pa = crate::proto::node::PeerAnnouncement {
            endpoint_id: peer_id.as_bytes().to_vec(),
            role: NodeRole::Worker as i32,
            gpu_name: Some("NVIDIA A100".to_string()),
            gpu_vram: Some("51539607552".to_string()),
            hardware: Some(crate::proto::node::HardwareInfo {
                is_soc: Some(false),
                hostname: None,
                gpus: vec![crate::proto::node::GpuInfo {
                    name: Some("NVIDIA A100".to_string()),
                    vram_bytes: Some("51539607552".to_string()),
                    reserved_bytes: None,
                    mem_bandwidth_gbps: Some("1948.70".to_string()),
                    compute_tflops_fp32: None,
                    compute_tflops_fp16: None,
                }],
            }),
            ..Default::default()
        };

        let (_, roundtripped) =
            proto_ann_to_local(&proto_pa).expect("proto_ann_to_local must succeed");
        assert_eq!(roundtripped.gpu_reserved_bytes, None);
        assert_eq!(
            roundtripped.gpu_mem_bandwidth_gbps.as_deref(),
            Some("1948.70")
        );
        assert_eq!(roundtripped.gpu_compute_tflops_fp32, None);
        assert_eq!(roundtripped.gpu_compute_tflops_fp16, None);
    }

    #[test]
    fn test_proto_gpu_info_preserves_legacy_fields_for_old_consumers() {
        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xCE; 32]).public());
        let proto_pa = crate::proto::node::PeerAnnouncement {
            endpoint_id: peer_id.as_bytes().to_vec(),
            role: NodeRole::Worker as i32,
            hardware: Some(crate::proto::node::HardwareInfo {
                is_soc: Some(false),
                hostname: Some("worker-01".to_string()),
                gpus: vec![
                    crate::proto::node::GpuInfo {
                        name: Some("NVIDIA A100".to_string()),
                        vram_bytes: Some("51539607552".to_string()),
                        reserved_bytes: Some("1073741824".to_string()),
                        mem_bandwidth_gbps: Some("1948.70".to_string()),
                        compute_tflops_fp32: Some("19.50".to_string()),
                        compute_tflops_fp16: Some("312.00".to_string()),
                    },
                    crate::proto::node::GpuInfo {
                        name: Some("NVIDIA A100".to_string()),
                        vram_bytes: Some("51539607552".to_string()),
                        reserved_bytes: None,
                        mem_bandwidth_gbps: Some("1948.70".to_string()),
                        compute_tflops_fp32: Some("19.50".to_string()),
                        compute_tflops_fp16: Some("312.00".to_string()),
                    },
                ],
            }),
            ..Default::default()
        };

        let (_, roundtripped) =
            proto_ann_to_local(&proto_pa).expect("proto_ann_to_local must succeed");
        assert_eq!(roundtripped.hostname.as_deref(), Some("worker-01"));
        assert_eq!(roundtripped.gpu_name.as_deref(), Some("2× NVIDIA A100"));
        assert_eq!(
            roundtripped.gpu_vram.as_deref(),
            Some("51539607552,51539607552")
        );
        assert_eq!(
            roundtripped.gpu_reserved_bytes.as_deref(),
            Some("1073741824,")
        );
        assert_eq!(
            roundtripped.gpu_mem_bandwidth_gbps.as_deref(),
            Some("1948.70,1948.70")
        );
        assert_eq!(
            roundtripped.gpu_compute_tflops_fp32.as_deref(),
            Some("19.50,19.50")
        );
        assert_eq!(
            roundtripped.gpu_compute_tflops_fp16.as_deref(),
            Some("312.00,312.00")
        );
        assert_eq!(roundtripped.is_soc, Some(false));
    }

    #[test]
    fn mesh_config_proto_roundtrip() {
        let snapshot = make_config_snapshot();
        let config = proto_config_to_mesh(&snapshot);
        assert_mesh_config_from_proto(&config);

        let roundtripped = mesh_config_to_proto(&config);
        assert_proto_config_roundtrip_matches(&roundtripped, &snapshot);
    }

    fn assert_mesh_config_from_proto(config: &crate::plugin::MeshConfig) {
        assert_eq!(config.version, Some(1));
        assert_eq!(config.gpu.assignment, crate::plugin::GpuAssignment::Auto);
        assert_eq!(config.models.len(), 1);
        assert_eq!(config.models[0].model, "Qwen3-8B");
        assert_eq!(config.models[0].mmproj.as_deref(), Some("mmproj-cut"));
        assert_eq!(config.models[0].ctx_size, Some(8192));
        assert_eq!(config.models[0].gpu_id.as_deref(), Some("pci:0000:65:00.0"));
        assert_eq!(config.plugins.len(), 1);
        assert_eq!(config.plugins[0].name, "blackboard");
    }

    fn assert_proto_config_roundtrip_matches(
        roundtripped: &NodeConfigSnapshot,
        snapshot: &NodeConfigSnapshot,
    ) {
        assert_eq!(roundtripped.version, snapshot.version);
        assert_eq!(
            roundtripped.gpu.as_ref().map(|g| g.assignment),
            Some(crate::proto::node::GpuAssignment::Auto as i32)
        );
        assert_eq!(roundtripped.models.len(), snapshot.models.len());
        assert_eq!(roundtripped.models[0].model, snapshot.models[0].model);
        assert_eq!(roundtripped.models[0].mmproj, snapshot.models[0].mmproj);
        assert_eq!(roundtripped.models[0].ctx_size, snapshot.models[0].ctx_size);
        assert_eq!(roundtripped.models[0].gpu_id, snapshot.models[0].gpu_id);
        assert_eq!(
            roundtripped.models[0].model_ref,
            snapshot.models[0].model_ref
        );
        assert_eq!(
            roundtripped.models[0].mmproj_ref,
            snapshot.models[0].mmproj_ref
        );
        assert_eq!(roundtripped.plugins.len(), snapshot.plugins.len());
        assert_eq!(roundtripped.plugins[0].name, snapshot.plugins[0].name);
    }

    #[test]
    fn config_sync_prefers_structured_model_refs() {
        let snapshot = NodeConfigSnapshot {
            version: 1,
            gpu: Some(NodeGpuConfig {
                assignment: crate::proto::node::GpuAssignment::Auto as i32,
            }),
            models: vec![NodeModelEntry {
                model: "legacy.gguf".to_string(),
                mmproj: Some("legacy-mmproj.gguf".to_string()),
                ctx_size: Some(4096),
                gpu_id: None,
                model_ref: Some(ConfiguredModelRef {
                    declared_ref: "structured.gguf".to_string(),
                    source_kind: Some("huggingface".to_string()),
                    revision: Some("main".to_string()),
                }),
                mmproj_ref: Some(ConfiguredModelRef {
                    declared_ref: "structured-mmproj.gguf".to_string(),
                    source_kind: Some("huggingface".to_string()),
                    revision: Some("main".to_string()),
                }),
            }],
            plugins: vec![],
        };

        let restored = proto_config_to_mesh(&snapshot);

        assert_eq!(restored.models[0].model, "structured.gguf");
        assert_eq!(
            restored.models[0].mmproj.as_deref(),
            Some("structured-mmproj.gguf")
        );
    }

    #[test]
    fn config_sync_empty_structured_refs_fall_back_to_legacy_strings() {
        let snapshot = NodeConfigSnapshot {
            version: 1,
            gpu: Some(NodeGpuConfig {
                assignment: crate::proto::node::GpuAssignment::Auto as i32,
            }),
            models: vec![NodeModelEntry {
                model: "legacy.gguf".to_string(),
                mmproj: Some("legacy-mmproj.gguf".to_string()),
                ctx_size: None,
                gpu_id: None,
                model_ref: Some(ConfiguredModelRef {
                    declared_ref: "   ".to_string(),
                    source_kind: Some("huggingface".to_string()),
                    revision: Some("main".to_string()),
                }),
                mmproj_ref: Some(ConfiguredModelRef {
                    declared_ref: "".to_string(),
                    source_kind: Some("huggingface".to_string()),
                    revision: Some("main".to_string()),
                }),
            }],
            plugins: vec![],
        };

        let restored = proto_config_to_mesh(&snapshot);

        assert_eq!(restored.models[0].model, "legacy.gguf");
        assert_eq!(
            restored.models[0].mmproj.as_deref(),
            Some("legacy-mmproj.gguf")
        );
    }

    #[test]
    fn canonical_config_hash_is_stable() {
        let snapshot = make_config_snapshot();
        let hash1 = canonical_config_hash(&snapshot);
        let hash2 = canonical_config_hash(&snapshot);
        assert_eq!(hash1, hash2, "same config must produce the same hash");
        assert_eq!(hash1.len(), 32);

        let mut different = snapshot.clone();
        different.version = 2;
        let hash3 = canonical_config_hash(&different);
        assert_ne!(hash1, hash3, "different config must produce different hash");
    }

    #[test]
    fn canonical_config_hash_changes_when_structured_refs_change_encoding() {
        let mut legacy_only = make_config_snapshot();
        legacy_only.models[0].model_ref = None;
        legacy_only.models[0].mmproj_ref = None;

        let dual_encoded = make_config_snapshot();

        assert_ne!(
            canonical_config_hash(&legacy_only),
            canonical_config_hash(&dual_encoded),
            "legacy-only and dual-encoded snapshots currently have distinct hashes"
        );
    }
    #[test]
    fn config_sync_full_config_roundtrip() {
        use crate::plugin::{GpuAssignment, GpuConfig, ModelConfigEntry, PluginConfigEntry};
        let config = crate::plugin::MeshConfig {
            version: Some(1),
            gpu: GpuConfig {
                assignment: GpuAssignment::Auto,
                parallel: None,
            },
            owner_control: Default::default(),
            telemetry: Default::default(),
            models: vec![ModelConfigEntry {
                model: "Qwen3-8B.gguf".to_string(),
                mmproj: Some("mm.gguf".to_string()),
                ctx_size: Some(8192),
                gpu_id: Some("pci:0000:65:00.0".to_string()),
                parallel: None,
                cache_type_k: None,
                cache_type_v: None,
                batch: None,
                ubatch: None,
                flash_attention: None,
            }],
            plugins: vec![PluginConfigEntry {
                name: "blackboard".to_string(),
                enabled: Some(true),
                command: Some("mesh-llm".to_string()),
                args: vec!["--plugin".to_string()],
                url: None,
            }],
        };
        let snapshot = mesh_config_to_proto(&config);
        let restored = proto_config_to_mesh(&snapshot);
        assert_eq!(restored.version, config.version);
        assert_eq!(restored.models.len(), 1);
        assert_eq!(restored.models[0].model, "Qwen3-8B.gguf");
        assert_eq!(restored.models[0].mmproj.as_deref(), Some("mm.gguf"));
        assert_eq!(restored.models[0].ctx_size, Some(8192));
        assert_eq!(
            restored.models[0].gpu_id.as_deref(),
            Some("pci:0000:65:00.0")
        );
        assert_eq!(restored.plugins.len(), 1);
        assert_eq!(restored.plugins[0].name, "blackboard");
        assert_eq!(restored.plugins[0].enabled, Some(true));
        assert_eq!(restored.plugins[0].command.as_deref(), Some("mesh-llm"));
        assert_eq!(restored.plugins[0].args, vec!["--plugin"]);
    }

    #[test]
    fn config_sync_empty_config_roundtrip() {
        let config = crate::plugin::MeshConfig::default();
        let snapshot = mesh_config_to_proto(&config);
        let restored = proto_config_to_mesh(&snapshot);
        assert!(restored.models.is_empty());
        assert!(restored.plugins.is_empty());
    }

    #[test]
    fn config_sync_config_hash_determinism() {
        use crate::plugin::{GpuAssignment, GpuConfig, ModelConfigEntry};
        let config = crate::plugin::MeshConfig {
            version: Some(1),
            gpu: GpuConfig {
                assignment: GpuAssignment::Auto,
                parallel: None,
            },
            owner_control: Default::default(),
            telemetry: Default::default(),
            models: vec![ModelConfigEntry {
                model: "test.gguf".to_string(),
                mmproj: None,
                ctx_size: None,
                gpu_id: None,
                parallel: None,
                cache_type_k: None,
                cache_type_v: None,
                batch: None,
                ubatch: None,
                flash_attention: None,
            }],
            plugins: vec![],
        };
        let snap1 = mesh_config_to_proto(&config);
        let snap2 = mesh_config_to_proto(&config);
        let h1 = canonical_config_hash(&snap1);
        let h2 = canonical_config_hash(&snap2);
        assert_eq!(h1, h2, "same config must produce same hash");

        let config2 = crate::plugin::MeshConfig {
            version: Some(1),
            gpu: GpuConfig {
                assignment: GpuAssignment::Auto,
                parallel: None,
            },
            owner_control: Default::default(),
            telemetry: Default::default(),
            models: vec![ModelConfigEntry {
                model: "other.gguf".to_string(),
                mmproj: None,
                ctx_size: None,
                gpu_id: None,
                parallel: None,
                cache_type_k: None,
                cache_type_v: None,
                batch: None,
                ubatch: None,
                flash_attention: None,
            }],
            plugins: vec![],
        };
        let snap3 = mesh_config_to_proto(&config2);
        let h3 = canonical_config_hash(&snap3);
        assert_ne!(h1, h3, "different config must produce different hash");
    }

    #[test]
    fn pinned_gpu_proto_roundtrip() {
        use crate::plugin::{GpuAssignment, GpuConfig, ModelConfigEntry};

        let config = crate::plugin::MeshConfig {
            version: Some(1),
            gpu: GpuConfig {
                assignment: GpuAssignment::Pinned,
                parallel: None,
            },
            owner_control: Default::default(),
            telemetry: Default::default(),
            models: vec![ModelConfigEntry {
                model: "Qwen3-8B-Q4_K_M".to_string(),
                mmproj: Some("mmproj-f16.gguf".to_string()),
                ctx_size: Some(8192),
                gpu_id: Some("pci:0000:65:00.0".to_string()),
                parallel: None,
                cache_type_k: None,
                cache_type_v: None,
                batch: None,
                ubatch: None,
                flash_attention: None,
            }],
            plugins: vec![],
        };

        let snapshot = mesh_config_to_proto(&config);
        assert_eq!(
            snapshot.gpu.as_ref().map(|gpu| gpu.assignment),
            Some(crate::proto::node::GpuAssignment::Pinned as i32),
            "pinned snapshots must not be downgraded to auto"
        );
        assert_eq!(
            snapshot.models[0].gpu_id.as_deref(),
            Some("pci:0000:65:00.0"),
            "proto snapshot must carry per-model gpu_id"
        );

        let restored = proto_config_to_mesh(&snapshot);
        assert_eq!(restored.gpu.assignment, GpuAssignment::Pinned);
        assert_eq!(
            restored.models[0].gpu_id.as_deref(),
            Some("pci:0000:65:00.0")
        );

        let roundtripped = mesh_config_to_proto(&restored);
        assert_eq!(
            roundtripped.gpu.as_ref().map(|gpu| gpu.assignment),
            Some(crate::proto::node::GpuAssignment::Pinned as i32),
            "re-encoded snapshot must keep pinned assignment"
        );
        assert_eq!(
            roundtripped.models[0].gpu_id.as_deref(),
            Some("pci:0000:65:00.0"),
            "re-encoded snapshot must keep gpu_id presence and value"
        );
    }

    #[test]
    fn pinned_gpu_proto_hash_changes_when_gpu_id_changes() {
        let mut snapshot_a = make_config_snapshot();
        snapshot_a.models[0].gpu_id = Some("pci:0000:65:00.0".to_string());

        let mut snapshot_b = snapshot_a.clone();
        snapshot_b.models[0].gpu_id = Some("pci:0000:66:00.0".to_string());

        assert_ne!(
            canonical_config_hash(&snapshot_a),
            canonical_config_hash(&snapshot_b),
            "changing only gpu_id must change the canonical config hash"
        );
    }

    #[test]
    fn pinned_gpu_proto_missing_gpu_id_decodes_as_none() {
        let snapshot = NodeConfigSnapshot {
            version: 1,
            gpu: Some(NodeGpuConfig {
                assignment: crate::proto::node::GpuAssignment::Pinned as i32,
            }),
            models: vec![NodeModelEntry {
                model: "Qwen3-8B-Q4_K_M".to_string(),
                mmproj: None,
                ctx_size: Some(4096),
                gpu_id: None,
                model_ref: Some(ConfiguredModelRef {
                    declared_ref: "Qwen3-8B-Q4_K_M".to_string(),
                    source_kind: None,
                    revision: None,
                }),
                mmproj_ref: None,
            }],
            plugins: vec![],
        };

        let encoded = snapshot.encode_to_vec();
        let decoded = NodeConfigSnapshot::decode(encoded.as_slice())
            .expect("payload without gpu_id must still decode");
        let restored = proto_config_to_mesh(&decoded);

        assert_eq!(
            restored.gpu.assignment,
            crate::plugin::GpuAssignment::Pinned
        );
        assert_eq!(restored.models.len(), 1);
        assert_eq!(restored.models[0].gpu_id, None);
        assert_eq!(restored.models[0].ctx_size, Some(4096));
    }

    #[test]
    fn test_peer_announcement_first_joined_mesh_ts_roundtrip() {
        use iroh::SecretKey;
        use std::collections::HashMap;

        let peer_id = EndpointId::from(SecretKey::from_bytes(&[0xEF; 32]).public());

        let ann_with_timestamp = super::PeerAnnouncement {
            addr: iroh::EndpointAddr {
                id: peer_id,
                addrs: Default::default(),
            },
            role: super::NodeRole::Worker,
            first_joined_mesh_ts: Some(1_700_000_000_000u64),
            models: vec![],
            vram_bytes: 0,
            model_source: None,
            serving_models: vec![],
            hosted_models: None,
            available_models: vec![],
            requested_models: vec![],
            explicit_model_interests: vec![],
            version: None,
            model_demand: HashMap::new(),
            mesh_id: None,
            gpu_name: None,
            hostname: None,
            is_soc: None,
            gpu_vram: None,
            gpu_reserved_bytes: None,
            gpu_mem_bandwidth_gbps: None,
            gpu_compute_tflops_fp32: None,
            gpu_compute_tflops_fp16: None,
            available_model_metadata: vec![],
            experts_summary: None,
            available_model_sizes: HashMap::new(),
            served_model_descriptors: vec![],
            served_model_runtime: vec![],
            owner_attestation: None,
            artifact_transfer_supported: true,
            stage_protocol_generation_supported: true,
            stage_status_list_supported: true,
            latency_ms: None,
            latency_source: None,
            latency_age_ms: None,
            latency_observer_id: None,
        };

        let proto_pa = local_ann_to_proto_ann(&ann_with_timestamp);
        assert_eq!(proto_pa.first_joined_mesh_ts, Some(1_700_000_000_000u64));

        let (_, roundtripped) =
            proto_ann_to_local(&proto_pa).expect("proto_ann_to_local must succeed");
        assert_eq!(
            roundtripped.first_joined_mesh_ts,
            Some(1_700_000_000_000u64)
        );

        let ann_without_timestamp = super::PeerAnnouncement {
            addr: iroh::EndpointAddr {
                id: peer_id,
                addrs: Default::default(),
            },
            role: super::NodeRole::Worker,
            first_joined_mesh_ts: None,
            models: vec![],
            vram_bytes: 0,
            model_source: None,
            serving_models: vec![],
            hosted_models: None,
            available_models: vec![],
            requested_models: vec![],
            explicit_model_interests: vec![],
            version: None,
            model_demand: HashMap::new(),
            mesh_id: None,
            gpu_name: None,
            hostname: None,
            is_soc: None,
            gpu_vram: None,
            gpu_reserved_bytes: None,
            gpu_mem_bandwidth_gbps: None,
            gpu_compute_tflops_fp32: None,
            gpu_compute_tflops_fp16: None,
            available_model_metadata: vec![],
            experts_summary: None,
            available_model_sizes: HashMap::new(),
            served_model_descriptors: vec![],
            served_model_runtime: vec![],
            owner_attestation: None,
            artifact_transfer_supported: false,
            stage_protocol_generation_supported: false,
            stage_status_list_supported: false,
            latency_ms: None,
            latency_source: None,
            latency_age_ms: None,
            latency_observer_id: None,
        };

        let proto_pa = local_ann_to_proto_ann(&ann_without_timestamp);
        assert_eq!(proto_pa.first_joined_mesh_ts, None);

        let (_, roundtripped) =
            proto_ann_to_local(&proto_pa).expect("proto_ann_to_local must succeed");
        assert_eq!(roundtripped.first_joined_mesh_ts, None);
    }
}
