//! Mesh membership via iroh QUIC connections.
//!
//! Mesh control traffic uses QUIC ALPN `mesh-llm/1` and multiplexes bi-streams
//! by first byte. Mesh-owned subsystem streams use `STREAM_SUBPROTOCOL` on the
//! admitted mesh connection; Skippy activation transport remains on the
//! latency-sensitive `skippy-stage/1` ALPN.

pub use mesh_llm_types::mesh::{
    infer_available_model_descriptors, infer_local_served_model_descriptor,
    infer_served_model_descriptors, merge_demand, ModelDemand, ModelRuntimeDescriptor,
    ModelSourceKind, ServedModelDescriptor, ServedModelIdentity, DEMAND_TTL_SECS, MAX_SPLIT_RTT_MS,
};

use anyhow::{Context, Result};
use base64::Engine;
use iroh::endpoint::Connection;
use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, TransportAddr};
use prost::Message;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;

use tokio::sync::{watch, Mutex};

use crate::crypto::{
    default_node_ownership_path, save_node_ownership, sign_node_ownership,
    verify_control_plane_target_node, verify_node_ownership, OwnershipStatus, OwnershipSummary,
    SignedNodeOwnership, TrustPolicy, TrustStore, DEFAULT_NODE_CERT_LIFETIME_SECS,
};
use crate::protocol::*;

use skippy_protocol::proto::stage as skippy_stage_proto;

const PRETTY_LOCAL_REQUEST_WINDOW_SECS: u64 = 24 * 60 * 60;
const EPHEMERAL_QUIC_PORT: u16 = 0;

fn emit_mesh_info(message: String) {
    let _ = crate::cli::output::emit_event(crate::cli::output::OutputEvent::Info {
        message,
        context: None,
    });
}

fn emit_mesh_warning(message: String) {
    let _ = crate::cli::output::emit_event(crate::cli::output::OutputEvent::Warning {
        message,
        context: None,
    });
}

fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

fn current_time_unix_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

pub(super) const PEER_CONNECT_AND_GOSSIP_TIMEOUT: std::time::Duration =
    std::time::Duration::from_secs(30);
const ARTIFACT_TRANSFER_OPEN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
const ARTIFACT_TRANSFER_READ_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
const ARTIFACT_TRANSFER_BUFFER_BYTES: usize = 1024 * 1024;

type MeshBiStream = (iroh::endpoint::SendStream, iroh::endpoint::RecvStream);

enum StageStreamAccept {
    Dispatch(MeshBiStream, u8),
    Continue,
    Closed,
}

struct NodeHardwareSnapshot {
    vram_bytes: u64,
    gpu_name: Option<String>,
    hostname: Option<String>,
    is_soc: Option<bool>,
    gpu_vram: Option<String>,
    gpu_reserved_bytes: Option<String>,
}

struct OwnerRuntimeInit {
    trust_store: TrustStore,
    trust_policy: TrustPolicy,
    owner_attestation: Option<SignedNodeOwnership>,
}

struct DetectedVramLog {
    detected_gb: f64,
    max_gb: Option<f64>,
    capped_bytes: Option<u64>,
}

struct AcceptedMeshStream {
    send: iroh::endpoint::SendStream,
    recv: iroh::endpoint::RecvStream,
    stream_type: u8,
}

enum ClosedConnectionRecovery {
    Reconnect(EndpointAddr),
    RemovePeer,
    AlreadyReplaced,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct QuicBindSelection {
    pub ip: Option<IpAddr>,
    pub port: Option<u16>,
}

fn quic_bind_addr(bind: QuicBindSelection) -> Option<SocketAddr> {
    if let Some(ip) = bind.ip {
        return Some(SocketAddr::new(
            ip,
            bind.port.unwrap_or(EPHEMERAL_QUIC_PORT),
        ));
    }

    if let Some(port) = bind.port {
        return Some(SocketAddr::from(([0, 0, 0, 0], port)));
    }

    #[cfg(target_os = "windows")]
    {
        Some(std::net::SocketAddr::from((
            [127, 0, 0, 1],
            EPHEMERAL_QUIC_PORT,
        )))
    }

    #[cfg(not(target_os = "windows"))]
    {
        None
    }
}

fn default_control_bind_addr() -> std::net::SocketAddr {
    std::net::SocketAddr::from(([127, 0, 0, 1], 0))
}

fn is_public_ipv4_candidate(socket: &SocketAddr) -> bool {
    match socket.ip() {
        IpAddr::V4(ip) => is_global_ipv4_candidate(ip),
        IpAddr::V6(_) => false,
    }
}

fn is_global_ipv4_candidate(ip: Ipv4Addr) -> bool {
    let [a, b, c, _] = ip.octets();
    !(ip.is_private()
        || ip.is_loopback()
        || ip.is_link_local()
        || ip.is_multicast()
        || ip.is_broadcast()
        || ip.is_unspecified()
        || (a == 100 && (64..=127).contains(&b))
        || (a == 192 && b == 0 && c == 0)
        || (a == 192 && b == 0 && c == 2)
        || (a == 198 && (b == 18 || b == 19))
        || (a == 198 && b == 51 && c == 100)
        || (a == 203 && b == 0 && c == 113)
        || a >= 240)
}

fn build_stun_binding_request() -> [u8; 20] {
    let mut req = [0u8; 20];
    req[1] = 0x01;
    req[4] = 0x21;
    req[5] = 0x12;
    req[6] = 0xA4;
    req[7] = 0x42;
    rand::fill(&mut req[8..20]);
    req
}

async fn resolve_stun_server(server: &str) -> Option<std::net::SocketAddr> {
    let mut addrs = tokio::net::lookup_host(server).await.ok()?;
    addrs.next()
}

fn parse_stun_mapped_ipv4(
    attr_type: u16,
    value: &[u8],
    magic: &[u8],
    advertised_port: u16,
) -> Option<std::net::SocketAddr> {
    use std::net::SocketAddrV4;

    if value.len() < 8 || value[1] != 0x01 {
        return None;
    }
    let ip = match attr_type {
        0x0020 => Ipv4Addr::new(
            value[4] ^ magic[0],
            value[5] ^ magic[1],
            value[6] ^ magic[2],
            value[7] ^ magic[3],
        ),
        0x0001 => Ipv4Addr::new(value[4], value[5], value[6], value[7]),
        _ => return None,
    };
    Some(std::net::SocketAddr::V4(SocketAddrV4::new(
        ip,
        advertised_port,
    )))
}

fn parse_stun_public_addr(
    response: &[u8],
    len: usize,
    magic: &[u8],
    advertised_port: u16,
) -> Option<std::net::SocketAddr> {
    let mut i = 20;
    while i + 4 <= len {
        let attr_type = u16::from_be_bytes([response[i], response[i + 1]]);
        let attr_len = u16::from_be_bytes([response[i + 2], response[i + 3]]) as usize;
        if i + 4 + attr_len > len {
            break;
        }
        let value = &response[i + 4..i + 4 + attr_len];
        if let Some(addr) = parse_stun_mapped_ipv4(attr_type, value, magic, advertised_port) {
            return Some(addr);
        }
        i += (4 + (attr_len + 3)) & !3;
    }
    None
}

fn endpoint_addr_has_public_ipv4(addr: &EndpointAddr) -> bool {
    addr.addrs.iter().any(|candidate| match candidate {
        TransportAddr::Ip(socket) => is_public_ipv4_candidate(socket),
        _ => false,
    })
}

// Host-network Docker and CNI bridges commonly reuse the same 172.* addresses
// on every host. When a node selects a bind IP, only advertise that direct IP
// while preserving relay and public candidates for non-LAN reachability.
fn filter_endpoint_addr_for_bind_ip(
    mut addr: EndpointAddr,
    bind_ip: Option<IpAddr>,
) -> EndpointAddr {
    let Some(bind_ip) = bind_ip else {
        return addr;
    };
    addr.addrs.retain(|candidate| match candidate {
        TransportAddr::Ip(socket) => socket.ip() == bind_ip || is_public_ipv4_candidate(socket),
        _ => true,
    });
    addr
}

fn effective_relay_urls(relay_urls: &[String]) -> Vec<String> {
    if relay_urls.is_empty() {
        vec![
            "https://usw1-2.relay.michaelneale.mesh-llm.iroh.link./".into(),
            "https://aps1-1.relay.michaelneale.mesh-llm.iroh.link./".into(),
        ]
    } else {
        relay_urls.to_vec()
    }
}

fn relay_map_from_urls(urls: &[String]) -> iroh::RelayMap {
    let configs = urls
        .iter()
        .map(|url| iroh::RelayConfig::new(url.parse().expect("invalid relay URL"), None));
    iroh::RelayMap::from_iter(configs)
}

fn encode_endpoint_addr_token(addr: &EndpointAddr) -> String {
    let json = serde_json::to_vec(addr).expect("endpoint addr should serialize");
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json)
}

fn control_endpoint_addr(
    endpoint: &Endpoint,
    advertise_addr: Option<std::net::SocketAddr>,
) -> EndpointAddr {
    let mut addr = endpoint.addr();
    if let Some(advertise_addr) = advertise_addr {
        addr.addrs
            .retain(|addr| matches!(addr, TransportAddr::Relay(_)));
        addr.addrs.insert(TransportAddr::Ip(advertise_addr));
    }
    addr
}

fn partial_artifact_path(destination: &std::path::Path) -> std::path::PathBuf {
    let file_name = destination
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("artifact");
    let unique = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| duration.as_nanos())
        .unwrap_or(0);
    destination.with_file_name(format!(
        ".{file_name}.{}.{}.part",
        std::process::id(),
        unique
    ))
}

struct PartialArtifactGuard {
    path: std::path::PathBuf,
    armed: bool,
}

impl PartialArtifactGuard {
    fn new(path: std::path::PathBuf) -> Self {
        Self { path, armed: true }
    }

    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for PartialArtifactGuard {
    fn drop(&mut self) {
        if self.armed {
            let _ = std::fs::remove_file(&self.path);
        }
    }
}

async fn read_artifact_transfer_chunk<R>(
    reader: &mut R,
    buffer: &mut [u8],
    idle_timeout: std::time::Duration,
) -> Result<usize>
where
    R: tokio::io::AsyncRead + Unpin,
{
    let read = tokio::time::timeout(idle_timeout, tokio::io::AsyncReadExt::read(reader, buffer))
        .await
        .map_err(|_| {
            anyhow::anyhow!("artifact transfer body read idle timeout after {idle_timeout:?}")
        })?
        .context("read artifact transfer bytes")?;
    anyhow::ensure!(
        read > 0,
        "artifact transfer ended before expected byte count"
    );
    Ok(read)
}

async fn write_artifact_transfer_response(
    send: &mut iroh::endpoint::SendStream,
    accepted: bool,
    total_size: u64,
    sha256: Option<&str>,
    error: Option<&str>,
) -> Result<()> {
    let response = skippy_stage_proto::StageArtifactTransferResponse {
        gen: skippy_protocol::STAGE_PROTOCOL_GENERATION,
        accepted,
        total_size,
        sha256: sha256.map(str::to_string),
        error: error.map(str::to_string),
    };
    skippy_protocol::validate_stage_artifact_transfer_response(&response)
        .map_err(|error| anyhow::anyhow!("invalid artifact transfer response: {error}"))?;
    write_len_prefixed(send, &response.encode_to_vec()).await?;
    if !accepted {
        let _ = send.finish();
    }
    Ok(())
}

fn artifact_transfer_allowed_by_topology(
    topologies: &[StageTopologyInstance],
    remote: EndpointId,
    package_dir: &std::path::Path,
    request: &skippy_stage_proto::StageArtifactTransferRequest,
) -> Result<bool> {
    let relative_path =
        crate::models::artifact_transfer::safe_relative_artifact_path(&request.relative_path)?;
    let manifest_path =
        std::path::PathBuf::from(crate::models::artifact_transfer::PACKAGE_MANIFEST_FILE);
    for topology in topologies {
        if topology.topology_id != request.topology_id
            || topology.run_id != request.run_id
            || topology.package_ref != request.package_ref
            || !topology
                .manifest_sha256
                .eq_ignore_ascii_case(&request.manifest_sha256)
        {
            continue;
        }
        let final_stage_index = topology.stages.iter().map(|stage| stage.stage_index).max();
        for assignment in topology
            .stages
            .iter()
            .filter(|stage| stage.node_id == remote && stage.stage_id == request.stage_id)
        {
            if relative_path == manifest_path {
                return Ok(true);
            }
            let include_output = final_stage_index == Some(assignment.stage_index);
            let allowed = crate::models::artifact_transfer::required_stage_package_artifacts(
                package_dir,
                &topology.package_ref,
                &topology.manifest_sha256,
                crate::models::artifact_transfer::StageArtifactSelection {
                    layer_start: assignment.layer_start,
                    layer_end: assignment.layer_end,
                    include_embeddings: assignment.layer_start == 0,
                    include_output,
                    include_projectors: assignment.layer_start == 0,
                },
            )?;
            if allowed.iter().any(|artifact| {
                artifact.relative_path == relative_path
                    && request
                        .expected_size
                        .is_none_or(|expected_size| Some(expected_size) == artifact.expected_size)
                    && request
                        .expected_sha256
                        .as_deref()
                        .is_none_or(|expected_sha| {
                            artifact
                                .expected_sha256
                                .as_deref()
                                .is_some_and(|sha| sha.eq_ignore_ascii_case(expected_sha))
                        })
            }) {
                return Ok(true);
            }
        }
    }
    Ok(false)
}

fn preflight_pushed_config_for_current_node(config: &crate::plugin::MeshConfig) -> Result<()> {
    let survey = crate::system::hardware::query(&[
        crate::system::hardware::Metric::GpuName,
        crate::system::hardware::Metric::GpuFacts,
    ]);
    preflight_pushed_config_for_current_node_with_gpus(config, &survey.gpus)
}

fn preflight_pushed_config_for_current_node_with_gpus(
    config: &crate::plugin::MeshConfig,
    gpus: &[crate::system::hardware::GpuFacts],
) -> Result<()> {
    if config.gpu.assignment != crate::plugin::GpuAssignment::Pinned {
        return Ok(());
    }

    for model in &config.models {
        let gpu = crate::system::hardware::resolve_pinned_gpu_strict(model.gpu_id.as_deref(), gpus)
            .map_err(anyhow::Error::new)
            .with_context(|| {
                format!(
                    "pushed config model '{}' failed pinned GPU preflight",
                    model.model
                )
            })?;

        let stable_id = gpu
            .stable_id
            .as_deref()
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "pushed config model '{}' resolved pinned GPU at index {} without a stable_id",
                    model.model,
                    gpu.index
                )
            })
            .with_context(|| {
                format!(
                    "pushed config model '{}' failed pinned GPU preflight",
                    model.model
                )
            })?;

        if gpu.backend_device.is_none() {
            return Err(anyhow::anyhow!(
                "pushed config model '{}' resolved pinned GPU '{}' at index {} without a backend_device",
                model.model,
                stable_id,
                gpu.index
            ))
            .with_context(|| {
                format!(
                    "pushed config model '{}' failed pinned GPU preflight",
                    model.model
                )
            });
        }
    }

    Ok(())
}

fn endpoint_id_hex(id: EndpointId) -> String {
    hex::encode(id.as_bytes())
}

fn new_plugin_message_id(source_peer_id: &str) -> String {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    format!("{source_peer_id}:{nanos}:{}", rand::random::<u64>())
}

fn node_role_label(role: &NodeRole) -> String {
    match role {
        NodeRole::Worker => "worker".into(),
        NodeRole::Host { .. } => "host".into(),
        NodeRole::Client => "client".into(),
    }
}

fn owner_control_error_envelope(
    code: crate::proto::node::OwnerControlErrorCode,
    request_id: Option<u64>,
    current_revision: Option<u64>,
    message: impl Into<String>,
) -> crate::proto::node::OwnerControlEnvelope {
    crate::proto::node::OwnerControlEnvelope {
        gen: NODE_PROTOCOL_GENERATION,
        handshake: None,
        request: None,
        response: None,
        error: Some(crate::proto::node::OwnerControlError {
            code: code as i32,
            message: message.into(),
            request_id,
            current_revision,
        }),
    }
}

fn owner_control_rejection_envelope(
    data: &[u8],
    request_id: Option<u64>,
    err: &ControlFrameError,
) -> crate::proto::node::OwnerControlEnvelope {
    let code = if matches!(err, ControlFrameError::MissingControlCommand) {
        crate::proto::node::OwnerControlErrorCode::UnknownCommand
    } else if serde_json::from_slice::<serde_json::Value>(data).is_ok() {
        crate::proto::node::OwnerControlErrorCode::LegacyJsonUnsupported
    } else {
        crate::proto::node::OwnerControlErrorCode::BadRequest
    };
    owner_control_error_envelope(code, request_id, None, err.to_string())
}

fn infer_remote_served_descriptors(
    primary_model_name: &str,
    serving_models: &[String],
    model_source: Option<&str>,
) -> Vec<ServedModelDescriptor> {
    let primary = model_source.and_then(identity_from_model_source);
    serving_models
        .iter()
        .enumerate()
        .map(|(idx, model_name)| {
            let identity = if idx == 0 || model_name == primary_model_name {
                let mut identity = primary
                    .clone()
                    .unwrap_or_else(|| unknown_identity(model_name));
                identity.model_name = model_name.clone();
                identity.is_primary = true;
                if identity.local_file_name.is_none() {
                    identity.local_file_name = Some(format!("{model_name}.gguf"));
                }
                identity
            } else {
                unknown_identity(model_name)
            };
            ServedModelDescriptor {
                identity,
                capabilities_known: false,
                capabilities: crate::models::ModelCapabilities::default(),
                topology: None,
            }
        })
        .collect()
}

fn unknown_identity(model_name: &str) -> ServedModelIdentity {
    ServedModelIdentity {
        model_name: model_name.to_string(),
        is_primary: false,
        source_kind: ModelSourceKind::Unknown,
        canonical_ref: None,
        repository: None,
        revision: None,
        artifact: None,
        local_file_name: Some(format!("{model_name}.gguf")),
        identity_hash: None,
    }
}

fn identity_from_model_source(source: &str) -> Option<ServedModelIdentity> {
    let trimmed = source.trim();
    if trimmed.is_empty() {
        return None;
    }

    if let Ok(model_ref) = model_ref::ModelRef::parse(trimmed) {
        let display_id = model_ref.display_id();
        return Some(ServedModelIdentity {
            model_name: String::new(),
            is_primary: false,
            source_kind: ModelSourceKind::HuggingFace,
            canonical_ref: Some(display_id.clone()),
            repository: Some(model_ref.repo),
            revision: model_ref.revision,
            artifact: model_ref.selector,
            local_file_name: None,
            identity_hash: Some(identity_hash_for(&display_id)),
        });
    }

    if trimmed.starts_with('/') || trimmed.starts_with("./") || trimmed.starts_with("../") {
        return Some(local_gguf_identity_from_source(trimmed));
    }

    if let Some((repo_id, revision, file)) = parse_hf_resolve_url_parts(trimmed) {
        let canonical_ref = format_hf_canonical_ref(&repo_id, revision.as_deref(), &file);
        return Some(ServedModelIdentity {
            model_name: String::new(),
            is_primary: false,
            source_kind: ModelSourceKind::HuggingFace,
            canonical_ref: Some(canonical_ref.clone()),
            repository: Some(repo_id),
            revision,
            artifact: Some(file.clone()),
            local_file_name: file.rsplit('/').next().map(str::to_string),
            identity_hash: Some(identity_hash_for(&canonical_ref)),
        });
    }

    if let Some((repo_id, revision, file)) = parse_hf_ref_parts(trimmed) {
        let canonical_ref = format_hf_canonical_ref(&repo_id, revision.as_deref(), &file);
        return Some(ServedModelIdentity {
            model_name: String::new(),
            is_primary: false,
            source_kind: ModelSourceKind::HuggingFace,
            canonical_ref: Some(canonical_ref.clone()),
            repository: Some(repo_id),
            revision,
            artifact: Some(file.clone()),
            local_file_name: file.rsplit('/').next().map(str::to_string),
            identity_hash: Some(identity_hash_for(&canonical_ref)),
        });
    }

    if trimmed.starts_with("http://") || trimmed.starts_with("https://") {
        return Some(ServedModelIdentity {
            model_name: String::new(),
            is_primary: false,
            source_kind: ModelSourceKind::DirectUrl,
            canonical_ref: Some(trimmed.to_string()),
            repository: None,
            revision: None,
            artifact: None,
            local_file_name: trimmed.rsplit('/').next().map(str::to_string),
            identity_hash: Some(identity_hash_for(trimmed)),
        });
    }

    if trimmed.ends_with(".gguf")
        || (trimmed.contains('/') && !trimmed.ends_with('/') && trimmed.split('/').count() != 2)
    {
        return Some(local_gguf_identity_from_source(trimmed));
    }

    Some(ServedModelIdentity {
        model_name: String::new(),
        is_primary: false,
        source_kind: ModelSourceKind::Catalog,
        canonical_ref: Some(trimmed.to_string()),
        repository: None,
        revision: None,
        artifact: None,
        local_file_name: None,
        identity_hash: Some(identity_hash_for(&format!("catalog:{trimmed}"))),
    })
}

fn local_gguf_identity_from_source(source: &str) -> ServedModelIdentity {
    let local_file_name = std::path::Path::new(source)
        .file_name()
        .and_then(|value| value.to_str())
        .map(str::to_string);
    ServedModelIdentity {
        model_name: String::new(),
        is_primary: false,
        source_kind: ModelSourceKind::LocalGguf,
        canonical_ref: None,
        repository: None,
        revision: None,
        artifact: None,
        local_file_name,
        identity_hash: None,
    }
}

fn identity_from_model_path(
    model_name: &str,
    path: &std::path::Path,
) -> Option<ServedModelIdentity> {
    if let Some(identity) = crate::models::huggingface_identity_for_path(path) {
        return Some(ServedModelIdentity {
            model_name: model_name.to_string(),
            is_primary: false,
            source_kind: ModelSourceKind::HuggingFace,
            canonical_ref: Some(identity.canonical_ref.clone()),
            repository: Some(identity.repo_id),
            revision: Some(identity.revision),
            artifact: Some(identity.file),
            local_file_name: Some(identity.local_file_name),
            identity_hash: Some(identity_hash_for(&identity.canonical_ref)),
        });
    }

    if path.exists() {
        let local_file_name = path
            .file_name()
            .and_then(|value| value.to_str())
            .map(str::to_string)
            .or_else(|| Some(format!("{model_name}.gguf")));
        return Some(ServedModelIdentity {
            model_name: model_name.to_string(),
            is_primary: false,
            source_kind: ModelSourceKind::LocalGguf,
            canonical_ref: None,
            repository: None,
            revision: None,
            artifact: None,
            local_file_name,
            identity_hash: None,
        });
    }

    None
}

#[allow(dead_code)]
fn descriptor_from_model_path(
    model_name: &str,
    path: &std::path::Path,
    is_primary: bool,
) -> Option<ServedModelDescriptor> {
    let mut identity = identity_from_model_path(model_name, path)?;
    identity.is_primary = is_primary;
    Some(descriptor_from_identity(model_name, identity))
}

#[allow(dead_code)]
fn descriptor_from_identity(
    model_name: &str,
    mut identity: ServedModelIdentity,
) -> ServedModelDescriptor {
    identity.model_name = model_name.to_string();
    let path = crate::models::find_model_path(model_name);
    let topology = crate::models::infer_local_model_topology(&path);
    let mut capabilities =
        crate::models::capabilities::infer_local_model_capabilities(model_name, &path);
    capabilities.moe = false;
    ServedModelDescriptor {
        identity,
        capabilities_known: true,
        capabilities,
        topology,
    }
}

fn parse_hf_ref_parts(input: &str) -> Option<(String, Option<String>, String)> {
    if input.starts_with('/') || input.starts_with("./") || input.starts_with("../") {
        return None;
    }
    let parts: Vec<&str> = input.splitn(3, '/').collect();
    if parts.len() != 3 {
        return None;
    }
    let (repo_tail, revision) = match parts[1].split_once('@') {
        Some((repo, revision)) => (repo, Some(revision.to_string())),
        None => (parts[1], None),
    };
    if parts[0].is_empty() || repo_tail.is_empty() || parts[2].is_empty() {
        return None;
    }
    Some((
        format!("{}/{}", parts[0], repo_tail),
        revision,
        parts[2].to_string(),
    ))
}

fn parse_hf_resolve_url_parts(url: &str) -> Option<(String, Option<String>, String)> {
    let path = url
        .strip_prefix("https://huggingface.co/")
        .or_else(|| url.strip_prefix("http://huggingface.co/"))?;
    let (repo, rest) = path.split_once("/resolve/")?;
    let (revision, file) = rest.split_once('/')?;
    let canonical = format!("{repo}@{revision}/{file}");
    parse_hf_ref_parts(&canonical)
}

fn format_hf_canonical_ref(repo: &str, revision: Option<&str>, file: &str) -> String {
    match revision {
        Some(revision) => format!("{repo}@{revision}/{file}"),
        None => format!("{repo}/{file}"),
    }
}

fn identity_hash_for(input: &str) -> String {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(input.as_bytes());
    hex::encode(hasher.finalize())
}

fn peer_info_to_mesh_peer(peer: &PeerInfo) -> crate::plugin::proto::MeshPeer {
    crate::plugin::proto::MeshPeer {
        peer_id: endpoint_id_hex(peer.id),
        version: peer.version.clone().unwrap_or_default(),
        capabilities: Vec::new(),
        role: node_role_label(&peer.role),
        vram_bytes: peer.vram_bytes,
        models: peer.models.clone(),
        serving_models: peer.serving_models.clone(),
        available_models: Vec::new(),
        requested_models: peer.requested_models.clone(),
        rtt_ms: peer.current_direct_rtt_ms(),
        model_source: peer.model_source.clone().unwrap_or_default(),
        hosted_models: peer.hosted_models.clone(),
        hosted_models_known: Some(peer.hosted_models_known),
    }
}

fn policy_accepts_peer(policy: TrustPolicy, owner_summary: &OwnershipSummary) -> bool {
    match policy {
        TrustPolicy::Off | TrustPolicy::PreferOwned => true,
        TrustPolicy::RequireOwned | TrustPolicy::Allowlist => {
            owner_summary.status == OwnershipStatus::Verified
        }
    }
}

fn load_or_refresh_owner_attestation(
    owner_keypair: &crate::crypto::OwnerKeypair,
    endpoint_id: EndpointId,
    node_label: Option<String>,
    hostname_hint: Option<String>,
) -> Result<SignedNodeOwnership> {
    // Always sign a fresh attestation on startup when the owner key is available.
    // This ensures that key rotation is always reflected immediately and no stale
    // certificate can persist across restarts.
    let path = default_node_ownership_path()?;
    let ownership = sign_node_ownership(
        owner_keypair,
        endpoint_id.as_bytes(),
        current_time_unix_ms() + DEFAULT_NODE_CERT_LIFETIME_SECS * 1000,
        node_label,
        hostname_hint,
    )?;
    save_node_ownership(&path, &ownership)?;
    Ok(ownership)
}

fn model_identity_score(identity: &ServedModelIdentity) -> u8 {
    let kind_score = match identity.source_kind {
        ModelSourceKind::HuggingFace => 4,
        ModelSourceKind::Catalog => 3,
        ModelSourceKind::DirectUrl => 2,
        ModelSourceKind::LocalGguf => 1,
        ModelSourceKind::Unknown => 0,
    };
    let canonical_bonus = if identity.canonical_ref.is_some() {
        2
    } else {
        0
    };
    let revision_bonus = if identity.revision.is_some() { 1 } else { 0 };
    kind_score + canonical_bonus + revision_bonus
}

fn model_descriptor_score(descriptor: &ServedModelDescriptor) -> u8 {
    let identity = &descriptor.identity;
    let capability_bonus = u8::from(descriptor.capabilities.multimodal)
        + u8::from(descriptor.capabilities.audio != crate::models::CapabilityLevel::None)
        + u8::from(descriptor.capabilities.vision != crate::models::CapabilityLevel::None)
        + u8::from(descriptor.capabilities.reasoning != crate::models::CapabilityLevel::None)
        + u8::from(descriptor.capabilities.tool_use != crate::models::CapabilityLevel::None);
    model_identity_score(identity) + capability_bonus
}

fn upsert_mesh_catalog_descriptor(
    descriptors: &mut HashMap<String, ServedModelDescriptor>,
    descriptor: ServedModelDescriptor,
) {
    if descriptor.identity.model_name.is_empty() {
        return;
    }
    let mut keys = vec![descriptor.identity.model_name.clone()];
    if let Some(public_id) = public_model_id_from_identity(&descriptor.identity) {
        keys.push(public_id);
    }
    keys.sort();
    keys.dedup();
    for key in keys {
        match descriptors.get(&key) {
            Some(existing)
                if model_descriptor_score(existing) >= model_descriptor_score(&descriptor) => {}
            _ => {
                descriptors.insert(key, descriptor.clone());
            }
        }
    }
}

/// Merge two demand maps. For each model, take max of last_active and request_count.
/// Role a node plays in the mesh.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub enum NodeRole {
    /// Provides staged GPU compute for a specific model.
    #[default]
    Worker,
    /// Runs the local serving runtime for a specific model and provides the HTTP API.
    Host { http_port: u16 },
    /// Lite client — no compute, accesses the API via tunnel.
    Client,
}

/// Gossip payload — extends EndpointAddr with role metadata.
/// Internal mesh gossip model. Legacy JSON v0 is adapted at the boundary.
#[derive(Debug, Clone)]
pub(crate) struct PeerAnnouncement {
    pub(crate) addr: EndpointAddr,
    pub(crate) role: NodeRole,
    pub(crate) first_joined_mesh_ts: Option<u64>,
    pub(crate) models: Vec<String>,
    pub(crate) vram_bytes: u64,
    pub(crate) model_source: Option<String>,
    pub(crate) serving_models: Vec<String>,
    pub(crate) hosted_models: Option<Vec<String>>,
    /// All GGUF filenames on disk in managed or legacy local storage (for mesh catalog)
    pub(crate) available_models: Vec<String>,
    pub(crate) requested_models: Vec<String>,
    /// Advisory canonical refs this node wants the mesh to consider.
    pub(crate) explicit_model_interests: Vec<String>,
    pub(crate) version: Option<String>,
    pub(crate) model_demand: HashMap<String, ModelDemand>,
    pub(crate) mesh_id: Option<String>,
    pub(crate) gpu_name: Option<String>,
    pub(crate) hostname: Option<String>,
    pub(crate) is_soc: Option<bool>,
    pub(crate) gpu_vram: Option<String>,
    pub(crate) gpu_reserved_bytes: Option<String>,
    pub(crate) gpu_mem_bandwidth_gbps: Option<String>,
    pub(crate) gpu_compute_tflops_fp32: Option<String>,
    pub(crate) gpu_compute_tflops_fp16: Option<String>,
    pub(crate) available_model_metadata: Vec<crate::proto::node::CompactModelMetadata>,
    pub(crate) experts_summary: Option<crate::proto::node::ExpertsSummary>,
    pub(crate) available_model_sizes: HashMap<String, u64>,
    pub(crate) served_model_descriptors: Vec<ServedModelDescriptor>,
    pub(crate) served_model_runtime: Vec<ModelRuntimeDescriptor>,
    pub(crate) owner_attestation: Option<SignedNodeOwnership>,
    pub(crate) artifact_transfer_supported: bool,
    pub(crate) stage_protocol_generation_supported: bool,
    pub(crate) stage_status_list_supported: bool,
    pub(crate) latency_ms: Option<u32>,
    pub(crate) latency_source: Option<crate::proto::node::LatencySource>,
    pub(crate) latency_age_ms: Option<u64>,
    pub(crate) latency_observer_id: Option<EndpointId>,
}

/// A single direct RTT measurement (e.g. from gossip exchange).
#[derive(Debug, Clone)]
pub struct DirectLatencyObservation {
    pub rtt_ms: u32,
    pub observed_at: std::time::Instant,
}

/// Latency propagated via transitive gossip (not measured directly).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PropagatedLatencyObservation {
    pub latency_ms: u32,
    pub age_ms_at_received: u64,
    pub received_at: std::time::Instant,
    pub observer_id: Option<EndpointId>,
}

/// Which source a display latency value came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DisplayLatencySource {
    Direct,
    Estimated,
    Unknown,
}

/// Computed display latency for UI/API consumption.
#[derive(Debug, Clone)]
pub struct DisplayLatency {
    pub latency_ms: Option<u32>,
    pub source: DisplayLatencySource,
    pub age_ms: u64,
    pub observer_id: Option<EndpointId>,
}

#[derive(Debug, Clone)]
pub struct PeerInfo {
    pub id: EndpointId,
    pub addr: EndpointAddr,
    pub role: NodeRole,
    pub first_joined_mesh_ts: Option<u64>,
    pub models: Vec<String>,
    pub vram_bytes: u64,
    pub rtt_ms: Option<u32>,
    pub model_source: Option<String>,
    /// All models assigned to this peer, even if not yet healthy.
    pub serving_models: Vec<String>,
    /// Models this node is actively routing inference for.
    pub hosted_models: Vec<String>,
    /// True when this peer explicitly advertised `hosted_models`.
    pub hosted_models_known: bool,
    /// All GGUFs on disk
    pub available_models: Vec<String>,
    /// Models this node has requested the mesh to serve
    pub requested_models: Vec<String>,
    /// Advisory canonical refs this peer wants the mesh to consider.
    pub explicit_model_interests: Vec<String>,
    /// Last time we directly communicated with this peer (gossip, heartbeat, tunnel).
    /// Only updated by direct bi-directional gossip exchanges, heartbeat probes,
    /// and inbound connections — never by transitive mentions.
    /// Used by PeerDown silencing to require independent proof-of-life.
    pub last_seen: std::time::Instant,
    /// Last time a bridge peer mentioned this peer in gossip.
    /// Updated on every transitive gossip update. Used together with `last_seen`
    /// for pruning and `collect_announcements`: a peer is included/kept as long
    /// as either timestamp is fresh.
    pub last_mentioned: std::time::Instant,
    /// mesh-llm version (e.g. "0.23.0")
    pub version: Option<String>,
    /// GPU name/model (e.g. "NVIDIA A100", "Apple M4 Max")
    pub gpu_name: Option<String>,
    /// Hostname of the node
    pub hostname: Option<String>,
    pub is_soc: Option<bool>,
    pub gpu_vram: Option<String>,
    pub gpu_reserved_bytes: Option<String>,
    pub gpu_mem_bandwidth_gbps: Option<String>,
    pub gpu_compute_tflops_fp32: Option<String>,
    pub gpu_compute_tflops_fp16: Option<String>,
    pub available_model_metadata: Vec<crate::proto::node::CompactModelMetadata>,
    pub experts_summary: Option<crate::proto::node::ExpertsSummary>,
    pub available_model_sizes: HashMap<String, u64>,
    pub served_model_descriptors: Vec<ServedModelDescriptor>,
    pub served_model_runtime: Vec<ModelRuntimeDescriptor>,
    pub owner_attestation: Option<SignedNodeOwnership>,
    pub artifact_transfer_supported: bool,
    pub stage_protocol_generation_supported: bool,
    pub stage_status_list_supported: bool,
    /// Most recent direct RTT sample for display purposes (refreshed periodically).
    pub display_rtt: Option<DirectLatencyObservation>,
    /// Latency propagated via transitive gossip.
    pub propagated_latency: Option<PropagatedLatencyObservation>,
    pub owner_summary: OwnershipSummary,
}

#[derive(Debug)]
pub struct OwnerRuntimeConfig {
    pub keypair: Option<crate::crypto::OwnerKeypair>,
    pub control_bind: Option<std::net::SocketAddr>,
    pub control_advertise_addr: Option<std::net::SocketAddr>,
    pub node_label: Option<String>,
    pub trust_store: TrustStore,
    pub trust_policy: TrustPolicy,
}

struct ControlListenerLifecycle {
    endpoint: Endpoint,
    token: String,
    shutdown_requested: Arc<std::sync::atomic::AtomicBool>,
    shutdown: Arc<tokio::sync::Notify>,
    task: tokio::task::JoinHandle<()>,
}
#[derive(Debug, Clone)]
pub struct MeshCatalogEntry {
    pub model_name: String,
    pub descriptor: Option<ServedModelDescriptor>,
}

impl PeerInfo {
    fn from_announcement(
        id: EndpointId,
        addr: EndpointAddr,
        ann: &PeerAnnouncement,
        owner_summary: OwnershipSummary,
    ) -> Self {
        Self {
            id,
            addr,
            role: ann.role.clone(),
            first_joined_mesh_ts: ann.first_joined_mesh_ts,
            models: ann.models.clone(),
            vram_bytes: ann.vram_bytes,
            rtt_ms: None,
            model_source: ann.model_source.clone(),
            serving_models: ann.serving_models.clone(),
            hosted_models: ann.hosted_models.clone().unwrap_or_default(),
            hosted_models_known: ann.hosted_models.is_some(),
            available_models: ann.available_models.clone(),
            requested_models: ann.requested_models.clone(),
            explicit_model_interests: ann.explicit_model_interests.clone(),
            last_seen: std::time::Instant::now(),
            last_mentioned: std::time::Instant::now(),
            version: ann.version.clone(),
            gpu_name: ann.gpu_name.clone(),
            hostname: ann.hostname.clone(),
            is_soc: ann.is_soc,
            gpu_vram: ann.gpu_vram.clone(),
            gpu_reserved_bytes: ann.gpu_reserved_bytes.clone(),
            gpu_mem_bandwidth_gbps: ann.gpu_mem_bandwidth_gbps.clone(),
            gpu_compute_tflops_fp32: ann.gpu_compute_tflops_fp32.clone(),
            gpu_compute_tflops_fp16: ann.gpu_compute_tflops_fp16.clone(),
            available_model_metadata: ann.available_model_metadata.clone(),
            experts_summary: ann.experts_summary.clone(),
            available_model_sizes: ann.available_model_sizes.clone(),
            served_model_descriptors: ann.served_model_descriptors.clone(),
            served_model_runtime: ann.served_model_runtime.clone(),
            owner_attestation: ann.owner_attestation.clone(),
            artifact_transfer_supported: ann.artifact_transfer_supported,
            stage_protocol_generation_supported: ann.stage_protocol_generation_supported,
            stage_status_list_supported: ann.stage_status_list_supported,
            display_rtt: None,
            propagated_latency: None,
            owner_summary,
        }
    }

    /// Return the most recent direct RTT sample for display, falling back to best-seen RTT.
    pub fn current_direct_rtt_ms(&self) -> Option<u32> {
        self.display_rtt.as_ref().map(|d| d.rtt_ms).or(self.rtt_ms)
    }

    /// Compute display latency from direct sample or propagated data.
    pub fn display_latency(&self) -> DisplayLatency {
        if let Some(ref direct) = self.display_rtt {
            return DisplayLatency {
                latency_ms: Some(direct.rtt_ms),
                source: DisplayLatencySource::Direct,
                age_ms: direct.observed_at.elapsed().as_millis() as u64,
                observer_id: None,
            };
        }
        if let Some(ref propagated) = self.propagated_latency {
            return DisplayLatency {
                latency_ms: Some(propagated.latency_ms),
                source: DisplayLatencySource::Estimated,
                age_ms: propagated.age_ms_at_received
                    + propagated.received_at.elapsed().as_millis() as u64,
                observer_id: propagated.observer_id,
            };
        }
        DisplayLatency {
            latency_ms: self.rtt_ms,
            source: DisplayLatencySource::Unknown,
            age_ms: 0,
            observer_id: None,
        }
    }

    #[cfg(test)]
    pub fn is_assigned_model(&self, model: &str) -> bool {
        self.serving_models.iter().any(|m| m == model)
    }

    pub fn routable_models(&self) -> Vec<String> {
        let raw = if self.hosted_models_known {
            &self.hosted_models
        } else {
            &self.serving_models
        };
        let mut models = raw
            .iter()
            .map(|model| self.public_model_id_for_routable_model(model))
            .collect::<Vec<_>>();
        models.sort();
        models.dedup();
        models
    }

    pub fn routes_model(&self, model: &str) -> bool {
        let raw = if self.hosted_models_known {
            &self.hosted_models
        } else {
            &self.serving_models
        };
        raw.iter().any(|candidate| {
            candidate == model || self.public_model_id_for_routable_model(candidate) == model
        })
    }

    pub fn accepts_http_inference(&self) -> bool {
        matches!(self.role, NodeRole::Host { .. })
    }

    pub fn http_routable_models(&self) -> Vec<String> {
        if self.accepts_http_inference() {
            self.routable_models()
        } else {
            Vec::new()
        }
    }

    pub fn routes_http_model(&self, model: &str) -> bool {
        self.accepts_http_inference() && self.routes_model(model)
    }

    fn public_model_id_for_routable_model(&self, model: &str) -> String {
        self.served_model_descriptors
            .iter()
            .find(|descriptor| descriptor.identity.model_name == model)
            .and_then(|descriptor| public_model_id_from_identity(&descriptor.identity))
            .unwrap_or_else(|| canonical_demand_model_ref(model))
    }

    pub fn advertised_context_length(&self, model: &str) -> Option<u32> {
        self.served_model_runtime
            .iter()
            .find(|runtime| runtime.model_name == model)
            .and_then(ModelRuntimeDescriptor::advertised_context_length)
    }
}

fn public_model_id_from_identity(identity: &ServedModelIdentity) -> Option<String> {
    match identity.source_kind {
        ModelSourceKind::HuggingFace => identity
            .repository
            .as_deref()
            .map(|repo| {
                let selector = identity
                    .artifact
                    .as_deref()
                    .and_then(model_ref::quant_selector_from_gguf_file)
                    .or_else(|| identity.artifact.clone());
                model_ref::format_model_ref(repo, None, selector.as_deref())
            })
            .or_else(|| {
                identity
                    .canonical_ref
                    .as_deref()
                    .and_then(|model_ref| model_ref::ModelRef::parse(model_ref).ok())
                    .map(|model_ref| model_ref.display_id())
            }),
        ModelSourceKind::Catalog => identity
            .canonical_ref
            .as_deref()
            .and_then(|model_ref| model_ref::ModelRef::parse(model_ref).ok())
            .map(|model_ref| model_ref.display_id()),
        ModelSourceKind::LocalGguf | ModelSourceKind::DirectUrl | ModelSourceKind::Unknown => None,
    }
}

fn canonical_demand_model_ref(model: &str) -> String {
    if let Ok(model_ref) = model_ref::ModelRef::parse(model) {
        return model_ref.display_id();
    }
    crate::models::find_loaded_remote_catalog_model_exact(model)
        .map(|remote_model| crate::models::remote_catalog_model_ref(&remote_model))
        .unwrap_or_else(|| model.to_string())
}

/// Peers not directly verified within this window are considered stale
/// and excluded from gossip propagation. After 2x this duration they're removed entirely.
const PEER_STALE_SECS: u64 = 180; // 3 minutes

/// How long a dead-peer entry blocks transitive re-learning and outbound
/// reconnection. After this period the entry expires silently and the peer
/// can be re-discovered through normal gossip propagation. If the peer is
/// genuinely gone, no bridge peer will mention it and it stays forgotten.
const DEAD_PEER_TTL: std::time::Duration = std::time::Duration::from_secs(300); // 5 minutes
/// Detect available VRAM. On Apple Silicon, uses ~75% of system RAM
/// (the rest is reserved for OS/apps on unified memory).
/// Detect available memory for model loading, capped by max_vram_gb if set.
/// "VRAM" is a misnomer — on macOS unified memory and Linux CPU-only, this
/// is system RAM. On Linux with a GPU, it's actual GPU VRAM.
pub fn detect_vram_bytes_capped(max_vram_gb: Option<f64>) -> u64 {
    let mut detected = crate::system::hardware::survey().vram_bytes;
    if let Some(cap) = max_vram_gb {
        let cap_bytes = (cap * 1e9) as u64;
        if cap_bytes < detected {
            detected = cap_bytes;
        }
    }
    detected
}

/// Lightweight routing table for passive nodes (clients + standby GPU).
/// Contains just enough info to route requests to the right host.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RoutingTable {
    pub hosts: Vec<RouteEntry>,
    /// Stable mesh identity — shared by all nodes in the same mesh.
    #[serde(default)]
    pub mesh_id: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RouteEntry {
    pub model: String,
    pub node_id: String,
    pub endpoint_id: EndpointId,
    pub vram_gb: f64,
}

/// Discover our public IP via STUN, then pair it with the given port.
/// We can't send STUN from the bound port (iroh owns it), but we only need
/// the public IP — the port is known from --bind-port + router forwarding.
async fn stun_public_addr(advertised_port: u16) -> Option<std::net::SocketAddr> {
    let stun_servers = [
        "stun.l.google.com:19302",
        "stun.cloudflare.com:3478",
        "stun.stunprotocol.org:3478",
    ];

    // Bind to ephemeral port — we only care about the IP, not the mapped port.
    let sock = tokio::net::UdpSocket::bind("0.0.0.0:0").await.ok()?;

    for server in &stun_servers {
        if let Some(addr) = probe_stun_server(&sock, server, advertised_port).await {
            tracing::info!("STUN discovered public address: {addr}");
            return Some(addr);
        }
    }

    tracing::warn!("STUN: could not discover public address");
    None
}

async fn probe_stun_server(
    sock: &tokio::net::UdpSocket,
    server: &str,
    advertised_port: u16,
) -> Option<std::net::SocketAddr> {
    let req = build_stun_binding_request();
    let dest = resolve_stun_server(server).await?;
    sock.send_to(&req, dest).await.ok()?;

    let mut buf = [0u8; 256];
    let (len, _) =
        tokio::time::timeout(std::time::Duration::from_secs(2), sock.recv_from(&mut buf))
            .await
            .ok()?
            .ok()?;
    if len < 20 {
        return None;
    }

    parse_stun_public_addr(&buf, len, &req[4..8], advertised_port)
}

async fn startup_secret_key(role: &NodeRole) -> Result<SecretKey> {
    if matches!(role, NodeRole::Client) || std::env::var("MESH_LLM_EPHEMERAL_KEY").is_ok() {
        let key = SecretKey::generate();
        tracing::info!("Using ephemeral key (unique identity)");
        Ok(key)
    } else {
        load_or_create_key().await
    }
}

fn startup_transport_config() -> iroh::endpoint::QuicTransportConfig {
    // Keep QUIC connections alive during long inference calls.
    //
    // noq-proto's default `max_idle_timeout` is ~30s and `keep_alive_interval`
    // is `None`. A non-streaming inference request (e.g. MoA reducer or any
    // `stream:false` call) sends nothing on the wire while the remote model is
    // generating tokens. Under concurrent load (multiple in-flight model
    // requests + gossip + heartbeats) noq's multipath bookkeeping will close
    // an idle path, and if it is the last open path the whole connection
    // drops mid-stream. The in-flight stream errors with `connection lost`
    // and the caller has to retry from scratch.
    //
    // A 10s keep-alive sends a small PING every 10s on each path, keeping
    // paths and the connection healthy during long compute. The 5-minute idle
    // timeout is defense in depth for truly silent connections (paused
    // agents, suspended laptops); short-term silence is handled by
    // keep-alive.
    let max_idle = iroh::endpoint::IdleTimeout::try_from(std::time::Duration::from_secs(300))
        .expect("5-minute idle timeout fits in a VarInt");
    let keep_alive = std::time::Duration::from_secs(10);
    let path_idle = std::time::Duration::from_secs(300);
    iroh::endpoint::QuicTransportConfig::builder()
        .max_concurrent_bidi_streams(1024u32.into())
        .keep_alive_interval(keep_alive)
        .max_idle_timeout(Some(max_idle))
        // noq-proto's multipath uses per-path idle timers independent of the
        // connection-level idle. Without these, a path can be torn down while
        // the connection idle timer is fine, and when the last path closes the
        // connection dies with `LastOpenPath`. Mirror connection-level
        // settings onto the default per-path config.
        .default_path_max_idle_timeout(path_idle)
        .default_path_keep_alive_interval(keep_alive)
        .build()
}

async fn bind_mesh_endpoint(
    secret_key: SecretKey,
    relay_urls: &[String],
    quic_bind: QuicBindSelection,
) -> Result<Endpoint> {
    let mut builder = Endpoint::builder(iroh::endpoint::presets::Minimal)
        .secret_key(secret_key)
        .alpns(vec![
            ALPN_V1.to_vec(),
            skippy_protocol::STAGE_ALPN_V1.to_vec(),
        ])
        .transport_config(startup_transport_config());

    let urls = effective_relay_urls(relay_urls);
    tracing::info!("Relay: {:?}", urls);
    builder = builder.relay_mode(iroh::endpoint::RelayMode::Custom(relay_map_from_urls(
        &urls,
    )));

    if let Some(addr) = quic_bind_addr(quic_bind) {
        tracing::info!("Binding QUIC to {addr}");
        builder = builder.bind_addr(addr)?;
    }

    builder.bind().await.map_err(Into::into)
}

async fn wait_for_endpoint_online(endpoint: &Endpoint, connected_log: &str, timeout_log: &str) {
    match tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.online()).await {
        Ok(()) => tracing::info!("{connected_log}"),
        Err(_) => tracing::warn!("{timeout_log}"),
    }
}

fn hardware_snapshot_for_start(
    hw: crate::system::hardware::HardwareSurvey,
    role: &NodeRole,
    max_vram_gb: Option<f64>,
) -> NodeHardwareSnapshot {
    let mut vram_bytes = hw.vram_bytes;
    let gpu_name = if matches!(role, NodeRole::Client) {
        None
    } else {
        hw.gpu_name
    };
    let hostname = hw.hostname;
    let is_soc = Some(hw.is_soc);
    let gpu_vram = (!hw.gpu_vram.is_empty()).then(|| {
        hw.gpu_vram
            .iter()
            .map(|b| b.to_string())
            .collect::<Vec<_>>()
            .join(",")
    });
    let gpu_reserved_bytes = if hw.gpu_reserved.iter().all(Option::is_none) {
        None
    } else {
        Some(
            hw.gpu_reserved
                .iter()
                .map(|value| value.map(|v| v.to_string()).unwrap_or_default())
                .collect::<Vec<_>>()
                .join(","),
        )
    };

    log_detected_vram(&mut vram_bytes, max_vram_gb);

    NodeHardwareSnapshot {
        vram_bytes,
        gpu_name,
        hostname,
        is_soc,
        gpu_vram,
        gpu_reserved_bytes,
    }
}

fn detected_vram_log(vram_bytes: u64, max_vram_gb: Option<f64>) -> DetectedVramLog {
    let detected_gb = vram_bytes as f64 / 1e9;
    let capped_bytes = max_vram_gb
        .map(|max_gb| ((max_gb * 1e9) as u64, max_gb))
        .and_then(|(max_bytes, _)| (max_bytes < vram_bytes).then_some(max_bytes));
    DetectedVramLog {
        detected_gb,
        max_gb: max_vram_gb,
        capped_bytes,
    }
}

fn log_detected_vram(vram_bytes: &mut u64, max_vram_gb: Option<f64>) {
    let log = detected_vram_log(*vram_bytes, max_vram_gb);
    if let Some(max_gb) = log.max_gb {
        log_detected_vram_with_cap(vram_bytes, log.detected_gb, max_gb, log.capped_bytes);
    } else {
        tracing::info!("Detected VRAM: {:.1} GB", log.detected_gb);
    }
}

fn log_detected_vram_with_cap(
    vram_bytes: &mut u64,
    detected_gb: f64,
    max_gb: f64,
    capped_bytes: Option<u64>,
) {
    if let Some(capped_bytes) = capped_bytes {
        tracing::info!(
            "Detected VRAM: {:.1} GB, capped to {:.1} GB (--max-vram)",
            detected_gb,
            max_gb
        );
        *vram_bytes = capped_bytes;
    } else {
        tracing::info!(
            "Detected VRAM: {:.1} GB (--max-vram {:.1} has no effect)",
            detected_gb,
            max_gb
        );
    }
}

fn init_owner_runtime(
    owner_config: Option<&OwnerRuntimeConfig>,
    endpoint_id: EndpointId,
    hostname: Option<String>,
) -> Result<OwnerRuntimeInit> {
    let trust_store = owner_config
        .map(|config| config.trust_store.clone())
        .unwrap_or_default();
    let trust_policy = owner_config
        .map(|config| config.trust_policy)
        .unwrap_or_default();
    let owner_attestation = match owner_config.and_then(|config| config.keypair.as_ref()) {
        Some(keypair) => Some(load_or_refresh_owner_attestation(
            keypair,
            endpoint_id,
            owner_config.and_then(|config| config.node_label.clone()),
            hostname,
        )?),
        None => None,
    };

    Ok(OwnerRuntimeInit {
        trust_store,
        trust_policy,
        owner_attestation,
    })
}

fn configure_control_relay(
    mut builder: iroh::endpoint::Builder,
    relay_urls: Option<&[String]>,
) -> iroh::endpoint::Builder {
    if let Some(relay_urls) = relay_urls {
        let urls = effective_relay_urls(relay_urls);
        tracing::info!("Owner-control relay: {:?}", urls);
        builder = builder.relay_mode(iroh::endpoint::RelayMode::Custom(relay_map_from_urls(
            &urls,
        )));
    } else {
        builder = builder.relay_mode(iroh::endpoint::RelayMode::Disabled);
    }
    builder
}

fn default_plugin_event_source(endpoint_id: EndpointId, source_peer_id: &mut String) {
    if source_peer_id.is_empty() {
        *source_peer_id = endpoint_id_hex(endpoint_id);
    }
}

#[derive(Clone)]
pub struct Node {
    endpoint: Endpoint,
    public_addr: Option<std::net::SocketAddr>,
    quic_bind: QuicBindSelection,
    state: Arc<Mutex<MeshState>>,
    role: Arc<Mutex<NodeRole>>,
    models: Arc<Mutex<Vec<String>>>,
    model_source: Arc<Mutex<Option<String>>>,
    serving_models: Arc<Mutex<Vec<String>>>,
    served_model_descriptors: Arc<Mutex<Vec<ServedModelDescriptor>>>,
    model_runtime_descriptors: Arc<Mutex<Vec<ModelRuntimeDescriptor>>>,
    hosted_models: Arc<Mutex<Vec<String>>>,
    llama_ready: Arc<Mutex<bool>>,
    available_models: Arc<Mutex<Vec<String>>>,
    requested_models: Arc<Mutex<Vec<String>>>,
    explicit_model_interests: Arc<Mutex<Vec<String>>>,
    /// Mesh-wide demand map — merged from gossip + local API requests.
    /// This is the single source of truth for "what does the mesh want?"
    model_demand: Arc<std::sync::Mutex<HashMap<String, ModelDemand>>>,
    mesh_id: Arc<Mutex<Option<String>>>,
    first_joined_mesh_ts: Arc<Mutex<Option<u64>>>,
    accepting: Arc<(tokio::sync::Notify, std::sync::atomic::AtomicBool)>,
    vram_bytes: u64,
    peer_change_tx: watch::Sender<usize>,
    pub peer_change_rx: watch::Receiver<usize>,
    inflight_requests: Arc<std::sync::atomic::AtomicUsize>,
    inflight_change_tx: watch::Sender<u64>,
    routing_metrics: crate::network::metrics::RoutingMetrics,
    routing_telemetry:
        Arc<std::sync::Mutex<Option<Arc<dyn crate::network::metrics::RoutingTelemetrySink>>>>,
    local_request_metrics: Arc<LocalRequestMetricsSampler>,
    runtime_data_producer: crate::runtime_data::RuntimeDataProducer,
    tunnel_tx: tokio::sync::mpsc::Sender<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)>,
    tunnel_http_tx:
        tokio::sync::mpsc::Sender<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)>,
    stage_transport_tx: tokio::sync::mpsc::Sender<(
        EndpointId,
        iroh::endpoint::SendStream,
        iroh::endpoint::RecvStream,
    )>,
    stage_control_tx: Arc<
        Mutex<
            Option<
                tokio::sync::mpsc::UnboundedSender<crate::inference::skippy::StageControlCommand>,
            >,
        >,
    >,
    stage_transport_bridges: Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>,
    stage_topologies: Arc<Mutex<StageTopologyState>>,
    plugin_manager: Arc<Mutex<Option<crate::plugin::PluginManager>>>,
    display_name: Arc<Mutex<Option<String>>>,
    owner_attestation: Arc<Mutex<Option<SignedNodeOwnership>>>,
    owner_summary: Arc<Mutex<OwnershipSummary>>,
    control_listener: Arc<Mutex<Option<ControlListenerLifecycle>>>,
    trust_store: Arc<Mutex<TrustStore>>,
    trust_policy: TrustPolicy,
    pub enumerate_host: bool,
    pub gpu_name: Option<String>,
    pub hostname: Option<String>,
    pub is_soc: Option<bool>,
    pub gpu_vram: Option<String>,
    pub gpu_reserved_bytes: Option<String>,
    pub gpu_mem_bandwidth_gbps: Arc<tokio::sync::Mutex<Option<Vec<f64>>>>,
    pub gpu_compute_tflops_fp32: Arc<tokio::sync::Mutex<Option<Vec<f64>>>>,
    pub gpu_compute_tflops_fp16: Arc<tokio::sync::Mutex<Option<Vec<f64>>>>,
    config_state: Arc<tokio::sync::Mutex<crate::runtime::config_state::ConfigState>>,
    config_revision_tx: Arc<tokio::sync::watch::Sender<u64>>,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct LocalRequestMetricsSnapshot {
    pub accepted_request_counts: Vec<u64>,
    pub latency_samples_ms: Vec<u64>,
}

#[derive(Default)]
struct LocalRequestMetricsSampler {
    inner: std::sync::Mutex<LocalRequestMetricsWindow>,
}

#[derive(Default)]
struct LocalRequestMetricsWindow {
    accepted_by_second: VecDeque<(u64, u64)>,
    completed_latencies_ms: VecDeque<(u64, u64)>,
}

struct PeerDownReport {
    conn_opt: Option<Connection>,
    peer_addr: Option<EndpointAddr>,
    recently_seen: bool,
    reporter_cooled: bool,
}

fn peer_down_endpoint_id(frame: &crate::proto::node::PeerDown) -> Option<EndpointId> {
    let peer_id_arr: [u8; 32] = match frame.peer_id.as_slice().try_into() {
        Ok(bytes) => bytes,
        Err(_) => {
            tracing::warn!("PeerDown: peer_id is not 32 bytes — rejecting");
            return None;
        }
    };
    match iroh::PublicKey::from_bytes(&peer_id_arr) {
        Ok(key) => Some(EndpointId::from(key)),
        Err(_) => {
            tracing::warn!("PeerDown: peer_id is not a valid public key — rejecting");
            None
        }
    }
}

impl LocalRequestMetricsSampler {
    fn record_request_accepted(&self) {
        let now_sec = now_secs();
        let mut guard = self
            .inner
            .lock()
            .expect("pretty request metrics mutex poisoned");
        guard.prune(now_sec);
        if let Some((second, count)) = guard.accepted_by_second.back_mut() {
            if *second == now_sec {
                *count += 1;
                return;
            }
        }
        guard.accepted_by_second.push_back((now_sec, 1));
    }

    fn record_request_completed(&self, started_at: std::time::Instant) {
        let now_sec = now_secs();
        let latency_ms = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        let mut guard = self
            .inner
            .lock()
            .expect("pretty request metrics mutex poisoned");
        guard.prune(now_sec);
        guard
            .completed_latencies_ms
            .push_back((now_sec, latency_ms));
    }

    fn snapshot(&self) -> LocalRequestMetricsSnapshot {
        let now_sec = now_secs();
        let window_start = now_sec.saturating_sub(PRETTY_LOCAL_REQUEST_WINDOW_SECS - 1);
        let mut guard = self
            .inner
            .lock()
            .expect("pretty request metrics mutex poisoned");
        guard.prune(now_sec);

        let accepted_by_second = guard
            .accepted_by_second
            .iter()
            .copied()
            .collect::<HashMap<_, _>>();
        let accepted_request_counts = (window_start..=now_sec)
            .map(|second| accepted_by_second.get(&second).copied().unwrap_or(0))
            .collect();
        let latency_samples_ms = guard
            .completed_latencies_ms
            .iter()
            .filter_map(|(second, latency_ms)| (*second >= window_start).then_some(*latency_ms))
            .collect();

        LocalRequestMetricsSnapshot {
            accepted_request_counts,
            latency_samples_ms,
        }
    }
}

impl LocalRequestMetricsWindow {
    fn prune(&mut self, now_sec: u64) {
        let oldest_kept_second = now_sec.saturating_sub(PRETTY_LOCAL_REQUEST_WINDOW_SECS - 1);
        while let Some((second, _)) = self.accepted_by_second.front() {
            if *second < oldest_kept_second {
                self.accepted_by_second.pop_front();
            } else {
                break;
            }
        }
        while let Some((second, _)) = self.completed_latencies_ms.front() {
            if *second < oldest_kept_second {
                self.completed_latencies_ms.pop_front();
            } else {
                break;
            }
        }
    }
}

/// Cooldown period after a reporter's death claim is rejected. During this
/// window, the same reporter cannot trigger a probe for the same target.
const PEER_DOWN_REPORTER_COOLDOWN_SECS: u64 = 600; // 10 minutes

struct MeshState {
    peers: HashMap<EndpointId, PeerInfo>,
    connections: HashMap<EndpointId, Connection>,
    /// Remote peers' tunnel maps: peer_endpoint_id → { target_endpoint_id → tunnel_port_on_that_peer }
    remote_tunnel_maps: HashMap<EndpointId, HashMap<EndpointId, u16>>,
    /// Peers confirmed dead — don't reconnect from gossip discovery.
    /// Cleared when the peer successfully reconnects via rejoin/join.
    /// Entries expire after [`DEAD_PEER_TTL`] so that peers recovered
    /// on other paths can be re-learned transitively through gossip.
    dead_peers: HashMap<EndpointId, std::time::Instant>,
    /// Tracks (reporter, target) pairs where a PeerDown claim was rejected
    /// (target was still reachable). Used to suppress repeated false reports
    /// from unreliable reporters (e.g. relay-partitioned nodes).
    peer_down_rejections: HashMap<(EndpointId, EndpointId), std::time::Instant>,
    seen_plugin_messages: HashMap<String, std::time::Instant>,
    seen_plugin_message_order: VecDeque<(std::time::Instant, String)>,
    /// Last policy-rejection status per peer — used to suppress duplicate log lines.
    /// Only logs when the status transitions (first rejection or status change).
    policy_rejected_peers: HashMap<EndpointId, OwnershipStatus>,
}

/// Returns `true` if the given peer has completed gossip validation and is
/// a full mesh member. Unadmitted peers are in `state.connections` but not
/// in `state.peers` — they are quarantined until gossip succeeds.
#[cfg(test)]
pub(crate) fn is_peer_admitted(peers: &HashMap<EndpointId, PeerInfo>, id: &EndpointId) -> bool {
    peers.contains_key(id)
}

/// Returns `true` if the given stream type is permitted before a peer has
/// been admitted through gossip.
///
/// Only two streams bypass the quarantine gate:
/// - `STREAM_GOSSIP (0x01)`: the admission handshake itself.
/// - `STREAM_ROUTE_REQUEST (0x05)`: passive/client request-only path — caller
///   is NEVER promoted to `state.peers`.
///
/// Every other stream — including tunnel (0x02 / 0x04) — requires the
/// remote to have completed gossip first.
pub(crate) fn stream_allowed_before_admission(stream_type: u8) -> bool {
    stream_type == STREAM_GOSSIP || stream_type == STREAM_ROUTE_REQUEST
}

pub(crate) fn ingest_tunnel_map(
    remote: EndpointId,
    frame: &crate::proto::node::TunnelMap,
    remote_tunnel_maps: &mut HashMap<EndpointId, HashMap<EndpointId, u16>>,
) -> Result<()> {
    if frame.owner_peer_id.as_slice() != remote.as_bytes() {
        anyhow::bail!(
            "TunnelMap owner_peer_id mismatch: frame claims owner {}, but connected peer is {}",
            hex::encode(&frame.owner_peer_id),
            remote.fmt_short()
        );
    }

    let mut tunnel_map: HashMap<EndpointId, u16> = HashMap::new();
    for entry in &frame.entries {
        if entry.target_peer_id.len() != 32 {
            anyhow::bail!(
                "TunnelMap entry has invalid target_peer_id length: {} (expected 32)",
                entry.target_peer_id.len()
            );
        }
        if entry.tunnel_port > u16::MAX as u32 {
            anyhow::bail!(
                "TunnelMap entry has out-of-range tunnel_port: {} (max {})",
                entry.tunnel_port,
                u16::MAX
            );
        }
        let arr: [u8; 32] = entry.target_peer_id.as_slice().try_into().unwrap();
        let eid = EndpointId::from(
            iroh::PublicKey::from_bytes(&arr)
                .map_err(|e| anyhow::anyhow!("Invalid target_peer_id bytes: {e}"))?,
        );
        tunnel_map.insert(eid, entry.tunnel_port as u16);
    }

    remote_tunnel_maps.insert(remote, tunnel_map);
    Ok(())
}

/// Validates the sender-identity rule for a validated `PeerLeaving` frame.
/// Returns `Ok(leaving_id)` if `frame.peer_id == remote` (sender is announcing its own departure).
/// Returns `Err(ForgedSender)` if `frame.peer_id != remote` — no peer should be removed.
pub(crate) fn resolve_peer_leaving(
    remote: EndpointId,
    frame: &crate::proto::node::PeerLeaving,
) -> Result<EndpointId, ControlFrameError> {
    if frame.peer_id.as_slice() != remote.as_bytes() {
        return Err(ControlFrameError::ForgedSender);
    }
    let arr: [u8; 32] =
        frame
            .peer_id
            .as_slice()
            .try_into()
            .map_err(|_| ControlFrameError::InvalidEndpointId {
                got: frame.peer_id.len(),
            })?;
    let pk =
        iroh::PublicKey::from_bytes(&arr).map_err(|_| ControlFrameError::InvalidEndpointId {
            got: frame.peer_id.len(),
        })?;
    Ok(EndpointId::from(pk))
}

/// Channels returned by Node::start for inbound tunnel streams.
pub struct TunnelChannels {
    pub rpc: tokio::sync::mpsc::Receiver<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)>,
    pub http: tokio::sync::mpsc::Receiver<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)>,
    pub stage: tokio::sync::mpsc::Receiver<(
        EndpointId,
        iroh::endpoint::SendStream,
        iroh::endpoint::RecvStream,
    )>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StageTopologyInstance {
    pub topology_id: String,
    pub run_id: String,
    pub model_id: String,
    pub package_ref: String,
    pub manifest_sha256: String,
    pub stages: Vec<StageAssignment>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StageAssignment {
    pub stage_id: String,
    pub stage_index: u32,
    pub node_id: EndpointId,
    pub layer_start: u32,
    pub layer_end: u32,
    pub endpoint: StageEndpoint,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StageEndpoint {
    pub bind_addr: String,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StageRuntimeStatus {
    pub topology_id: String,
    pub run_id: String,
    pub model_id: String,
    pub backend: String,
    pub package_ref: Option<String>,
    pub manifest_sha256: Option<String>,
    pub source_model_path: Option<String>,
    pub source_model_sha256: Option<String>,
    pub source_model_bytes: Option<u64>,
    pub materialized_path: Option<String>,
    pub materialized_pinned: bool,
    pub projector_path: Option<String>,
    pub stage_id: String,
    pub stage_index: u32,
    pub node_id: Option<EndpointId>,
    pub layer_start: u32,
    pub layer_end: u32,
    pub state: crate::inference::skippy::StageRuntimeState,
    pub bind_addr: String,
    pub activation_width: u32,
    pub wire_dtype: crate::inference::skippy::StageWireDType,
    pub selected_device: Option<skippy_protocol::StageDevice>,
    pub ctx_size: u32,
    pub lane_count: u32,
    pub n_batch: Option<u32>,
    pub n_ubatch: Option<u32>,
    pub flash_attn_type: skippy_protocol::FlashAttentionType,
    pub error: Option<String>,
    pub shutdown_generation: u64,
}

#[derive(Clone, Debug, Default)]
struct StageTopologyState {
    topologies: HashMap<String, StageTopologyInstance>,
    statuses: HashMap<String, StageRuntimeStatus>,
}

impl StageTopologyState {
    fn record_topology(&mut self, topology: StageTopologyInstance) {
        self.topologies.insert(
            stage_topology_key(&topology.topology_id, &topology.run_id),
            topology,
        );
    }

    fn activate_topology(&mut self, topology: StageTopologyInstance) {
        let active_key = stage_topology_key(&topology.topology_id, &topology.run_id);
        let model_id = topology.model_id.clone();
        self.topologies
            .retain(|key, existing| existing.model_id != model_id || key == &active_key);
        self.statuses.retain(|_, status| {
            status.model_id != model_id
                || (status.topology_id == topology.topology_id && status.run_id == topology.run_id)
        });
        self.record_topology(topology);
    }

    fn visible_topologies(&self) -> Vec<StageTopologyInstance> {
        self.topologies
            .values()
            .filter(|topology| {
                topology.stages.len() > 1
                    || !self.statuses.values().any(|status| {
                        status.topology_id == topology.topology_id
                            && status.run_id == topology.run_id
                    })
            })
            .cloned()
            .collect()
    }

    fn runtime_statuses(&self) -> Vec<StageRuntimeStatus> {
        self.statuses
            .values()
            .filter(|status| {
                !status.topology_id.is_empty()
                    && !status.run_id.is_empty()
                    && !status.stage_id.is_empty()
            })
            .cloned()
            .collect()
    }

    fn record_status(&mut self, runtime_status: StageRuntimeStatus) {
        if runtime_status.topology_id.is_empty()
            || runtime_status.run_id.is_empty()
            || runtime_status.stage_id.is_empty()
        {
            return;
        }
        if !runtime_status.bind_addr.is_empty() && !runtime_status.bind_addr.ends_with(":0") {
            let topology_key =
                stage_topology_key(&runtime_status.topology_id, &runtime_status.run_id);
            if let Some(topology) = self.topologies.get_mut(&topology_key) {
                if let Some(stage) = topology
                    .stages
                    .iter_mut()
                    .find(|stage| stage.stage_id == runtime_status.stage_id)
                {
                    stage.endpoint.bind_addr = runtime_status.bind_addr.clone();
                }
            }
        }
        self.statuses.insert(
            stage_runtime_status_key(
                &runtime_status.topology_id,
                &runtime_status.run_id,
                &runtime_status.stage_id,
            ),
            runtime_status,
        );
    }

    fn active_statuses(&self) -> Vec<StageRuntimeStatus> {
        self.statuses
            .values()
            .filter(|status| {
                matches!(
                    status.state,
                    crate::inference::skippy::StageRuntimeState::Starting
                        | crate::inference::skippy::StageRuntimeState::Ready
                )
            })
            .cloned()
            .collect()
    }
}

pub struct InflightRequestGuard {
    inflight_requests: Arc<std::sync::atomic::AtomicUsize>,
    inflight_change_tx: watch::Sender<u64>,
    local_request_metrics: Arc<LocalRequestMetricsSampler>,
    started_at: std::time::Instant,
    routing_metrics: crate::network::metrics::RoutingMetrics,
    routing_telemetry: Option<Arc<dyn crate::network::metrics::RoutingTelemetrySink>>,
    runtime_data_producer: crate::runtime_data::RuntimeDataProducer,
}

impl Drop for InflightRequestGuard {
    fn drop(&mut self) {
        let _ = self.inflight_requests.fetch_update(
            std::sync::atomic::Ordering::Relaxed,
            std::sync::atomic::Ordering::Relaxed,
            |current| current.checked_sub(1),
        );
        let _ = self.inflight_change_tx.send(
            self.inflight_requests
                .load(std::sync::atomic::Ordering::Relaxed) as u64,
        );
        self.local_request_metrics
            .record_request_completed(self.started_at);
        let current_inflight_requests =
            self.inflight_requests
                .load(std::sync::atomic::Ordering::Relaxed) as u64;
        if let Some(routing_telemetry) = &self.routing_telemetry {
            routing_telemetry.observe_inflight_requests(current_inflight_requests);
        }
        self.runtime_data_producer.publish_routing_snapshot(
            self.routing_metrics
                .collector_snapshot(current_inflight_requests),
        );
    }
}

#[async_trait::async_trait]
impl crate::inference::skippy::StagePackagePrefetcher for Node {
    async fn prefetch_stage_package(
        &self,
        request: &crate::inference::skippy::StagePrepareRequest,
    ) -> Result<()> {
        self.prefetch_stage_package_from_coordinator(request).await
    }
}

impl Node {
    pub(crate) fn set_routing_telemetry_sink(
        &self,
        sink: Option<Arc<dyn crate::network::metrics::RoutingTelemetrySink>>,
    ) {
        *self
            .routing_telemetry
            .lock()
            .expect("routing telemetry sink lock poisoned") = sink;
    }

    fn routing_telemetry_sink(
        &self,
    ) -> Option<Arc<dyn crate::network::metrics::RoutingTelemetrySink>> {
        self.routing_telemetry
            .lock()
            .expect("routing telemetry sink lock poisoned")
            .clone()
    }

    fn publish_routing_runtime_snapshot(&self) {
        self.runtime_data_producer.publish_routing_snapshot(
            self.routing_metrics
                .collector_snapshot(self.inflight_requests()),
        );
    }

    pub fn begin_inflight_request(&self) -> InflightRequestGuard {
        self.local_request_metrics.record_request_accepted();
        self.inflight_requests
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let current = self
            .inflight_requests
            .load(std::sync::atomic::Ordering::Relaxed) as u64;
        let _ = self.inflight_change_tx.send(current);
        self.routing_metrics.observe_inflight(current);
        let routing_telemetry = self.routing_telemetry_sink();
        if let Some(sink) = &routing_telemetry {
            sink.observe_inflight_requests(current);
        }
        self.publish_routing_runtime_snapshot();
        InflightRequestGuard {
            inflight_requests: self.inflight_requests.clone(),
            inflight_change_tx: self.inflight_change_tx.clone(),
            local_request_metrics: self.local_request_metrics.clone(),
            started_at: std::time::Instant::now(),
            routing_metrics: self.routing_metrics.clone(),
            routing_telemetry,
            runtime_data_producer: self.runtime_data_producer.clone(),
        }
    }

    pub fn inflight_requests(&self) -> u64 {
        self.inflight_requests
            .load(std::sync::atomic::Ordering::Relaxed) as u64
    }

    /// Locally observed routing metrics, used by the auto-router to score
    /// models by their measured throughput from this node's perspective.
    pub fn routing_metrics(&self) -> &crate::network::metrics::RoutingMetrics {
        &self.routing_metrics
    }

    pub fn inflight_change_rx(&self) -> watch::Receiver<u64> {
        self.inflight_change_tx.subscribe()
    }

    pub(crate) async fn set_stage_control_sender(
        &self,
        tx: tokio::sync::mpsc::UnboundedSender<crate::inference::skippy::StageControlCommand>,
    ) {
        *self.stage_control_tx.lock().await = Some(tx);
    }

    pub async fn record_stage_topology(&self, topology: StageTopologyInstance) {
        self.stage_topologies.lock().await.record_topology(topology);
    }

    pub async fn activate_stage_topology(&self, topology: StageTopologyInstance) {
        self.stage_topologies
            .lock()
            .await
            .activate_topology(topology);
    }

    pub async fn stage_topologies(&self) -> Vec<StageTopologyInstance> {
        self.stage_topologies.lock().await.visible_topologies()
    }

    pub async fn stage_runtime_statuses(&self) -> Vec<StageRuntimeStatus> {
        self.stage_topologies.lock().await.runtime_statuses()
    }

    pub async fn refresh_stage_runtime_statuses(&self, timeout: std::time::Duration) {
        let active_statuses = self.stage_topologies.lock().await.active_statuses();
        for status in active_statuses {
            if status.stage_index == 0 {
                continue;
            }
            let Some(peer_id) = status.node_id else {
                continue;
            };
            let filter = crate::inference::skippy::StageStatusFilter {
                topology_id: Some(status.topology_id.clone()),
                run_id: Some(status.run_id.clone()),
                stage_id: Some(status.stage_id.clone()),
            };
            let refresh = async {
                if peer_id == self.endpoint.id() {
                    self.query_local_stage_status(filter)
                        .await
                        .map(crate::inference::skippy::StageControlResponse::Status)
                } else {
                    self.send_stage_control(
                        peer_id,
                        crate::inference::skippy::StageControlRequest::Status(filter),
                    )
                    .await
                }
            };
            match tokio::time::timeout(timeout, refresh).await {
                Ok(Ok(crate::inference::skippy::StageControlResponse::Status(statuses))) => {
                    if statuses.is_empty() {
                        self.record_stage_status(
                            Some(peer_id),
                            stage_snapshot_from_runtime_status(
                                &status,
                                crate::inference::skippy::StageRuntimeState::Failed,
                                Some("stage status missing from runtime".to_string()),
                            ),
                        )
                        .await;
                    } else {
                        for status in statuses {
                            self.record_stage_status(Some(peer_id), status).await;
                        }
                    }
                }
                Ok(Ok(crate::inference::skippy::StageControlResponse::Ready(ready))) => {
                    self.record_stage_status(Some(peer_id), ready.status).await;
                }
                Ok(Ok(_)) => {}
                Ok(Err(error)) => {
                    self.record_stage_status(
                        Some(peer_id),
                        stage_snapshot_from_runtime_status(
                            &status,
                            crate::inference::skippy::StageRuntimeState::Failed,
                            Some(error.to_string()),
                        ),
                    )
                    .await;
                }
                Err(_) => {
                    tracing::debug!(
                        topology_id = %status.topology_id,
                        run_id = %status.run_id,
                        stage_id = %status.stage_id,
                        peer = %peer_id.fmt_short(),
                        "stage status refresh timed out; preserving last known status"
                    );
                }
            }
        }
    }

    pub(crate) async fn record_stage_status(
        &self,
        node_id: Option<EndpointId>,
        status: crate::inference::skippy::StageStatusSnapshot,
    ) {
        let runtime_status = stage_runtime_status_from_snapshot(node_id, status);
        self.stage_topologies
            .lock()
            .await
            .record_status(runtime_status);
    }

    pub(crate) async fn query_local_stage_status(
        &self,
        filter: crate::inference::skippy::StageStatusFilter,
    ) -> Result<Vec<crate::inference::skippy::StageStatusSnapshot>> {
        let control_tx = self.stage_control_tx.lock().await.clone();
        let Some(tx) = control_tx else {
            anyhow::bail!("stage control is not available");
        };
        let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
        tx.send(crate::inference::skippy::StageControlCommand {
            request: crate::inference::skippy::StageControlRequest::Status(filter),
            resp: resp_tx,
        })
        .map_err(|_| anyhow::anyhow!("stage control loop is unavailable"))?;
        match resp_rx
            .await
            .map_err(|_| anyhow::anyhow!("stage control response dropped"))??
        {
            crate::inference::skippy::StageControlResponse::Status(statuses) => Ok(statuses),
            crate::inference::skippy::StageControlResponse::Ready(_) => {
                anyhow::bail!("unexpected ready response for stage status request")
            }
            _ => anyhow::bail!("unexpected response for stage status request"),
        }
    }

    pub(crate) async fn send_local_stage_control(
        &self,
        mut request: crate::inference::skippy::StageControlRequest,
    ) -> Result<crate::inference::skippy::StageControlResponse> {
        self.prepare_stage_control_request(&mut request).await?;
        if let crate::inference::skippy::StageControlRequest::Load(load) = &request {
            self.record_stage_topology(stage_topology_from_load(self.endpoint.id(), load))
                .await;
        }
        let control_tx = self.stage_control_tx.lock().await.clone();
        let Some(tx) = control_tx else {
            anyhow::bail!("stage control is not available");
        };
        let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
        tx.send(crate::inference::skippy::StageControlCommand {
            request,
            resp: resp_tx,
        })
        .map_err(|_| anyhow::anyhow!("stage control loop is unavailable"))?;
        let response = resp_rx
            .await
            .map_err(|_| anyhow::anyhow!("stage control response dropped"))??;
        match &response {
            crate::inference::skippy::StageControlResponse::Ready(ready) => {
                self.record_stage_status(Some(self.endpoint.id()), ready.status.clone())
                    .await;
            }
            crate::inference::skippy::StageControlResponse::Status(statuses) => {
                for status in statuses {
                    self.record_stage_status(Some(self.endpoint.id()), status.clone())
                        .await;
                }
            }
            _ => {}
        }
        Ok(response)
    }

    pub async fn send_stage_control(
        &self,
        peer_id: EndpointId,
        request: crate::inference::skippy::StageControlRequest,
    ) -> Result<crate::inference::skippy::StageControlResponse> {
        use prost::Message as _;

        let timeout = Self::stage_control_request_timeout(&request);
        if let crate::inference::skippy::StageControlRequest::Load(load) = &request {
            self.record_stage_topology(stage_topology_from_load(peer_id, load))
                .await;
        }
        let frame = stage_control_request_to_proto(self.endpoint.id(), request);
        let response = tokio::time::timeout(timeout, async {
            let (mut send, mut recv) = if self
                .peer_supports_skippy_subprotocol_feature(
                    peer_id,
                    skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_CONTROL,
                )
                .await
            {
                self.open_skippy_stage_mesh_stream(peer_id, skippy_protocol::STAGE_STREAM_CONTROL)
                    .await?
            } else {
                let conn = self.stage_connection_to_peer(peer_id).await?;
                let (mut send, recv) = conn.open_bi().await?;
                send.write_all(&[skippy_protocol::STAGE_STREAM_CONTROL])
                    .await?;
                (send, recv)
            };
            write_len_prefixed(&mut send, &frame.encode_to_vec()).await?;
            let buf = read_len_prefixed(&mut recv).await?;
            let response =
                skippy_protocol::proto::stage::StageControlResponse::decode(buf.as_slice())
                    .map_err(|e| anyhow::anyhow!("StageControlResponse decode error: {e}"))?;
            skippy_protocol::validate_stage_control_response(&response)
                .map_err(|e| anyhow::anyhow!("StageControlResponse validation error: {e}"))?;
            let _ = send.finish();
            stage_control_response_from_proto(response)
        })
        .await
        .map_err(|_| {
            anyhow::anyhow!("timeout waiting for stage control response after {timeout:?}")
        })??;

        match &response {
            crate::inference::skippy::StageControlResponse::Ready(ready) => {
                self.record_stage_status(Some(peer_id), ready.status.clone())
                    .await;
            }
            crate::inference::skippy::StageControlResponse::Status(statuses) => {
                for status in statuses {
                    self.record_stage_status(Some(peer_id), status.clone())
                        .await;
                }
            }
            _ => {}
        }
        Ok(response)
    }

    fn stage_control_request_timeout(
        request: &crate::inference::skippy::StageControlRequest,
    ) -> std::time::Duration {
        match request {
            crate::inference::skippy::StageControlRequest::Claim(_)
            | crate::inference::skippy::StageControlRequest::Stop(_)
            | crate::inference::skippy::StageControlRequest::Status(_)
            | crate::inference::skippy::StageControlRequest::Inventory(_)
            | crate::inference::skippy::StageControlRequest::CancelPrepare(_)
            | crate::inference::skippy::StageControlRequest::StatusUpdate(_) => {
                std::time::Duration::from_secs(30)
            }
            crate::inference::skippy::StageControlRequest::Load(load) => {
                crate::inference::skippy::stage_load_timeout(load)
            }
            crate::inference::skippy::StageControlRequest::Prepare(prepare) => {
                crate::inference::skippy::stage_load_timeout(&prepare.load)
            }
        }
    }

    pub async fn open_stage_transport_stream(
        &self,
        peer_id: EndpointId,
        topology_id: impl Into<String>,
        run_id: impl Into<String>,
        stage_id: impl Into<String>,
    ) -> Result<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)> {
        use prost::Message as _;

        let open = skippy_protocol::proto::stage::StageTransportOpen {
            gen: skippy_protocol::STAGE_PROTOCOL_GENERATION,
            requester_id: self.endpoint.id().as_bytes().to_vec(),
            topology_id: topology_id.into(),
            run_id: run_id.into(),
            stage_id: stage_id.into(),
        };
        skippy_protocol::validate_stage_transport_open(&open)
            .map_err(|e| anyhow::anyhow!("StageTransportOpen validation error: {e}"))?;
        let conn = self.stage_connection_to_peer(peer_id).await?;
        let (mut send, recv) = conn.open_bi().await?;
        send.write_all(&[skippy_protocol::STAGE_STREAM_TRANSPORT])
            .await?;
        write_len_prefixed(&mut send, &open.encode_to_vec()).await?;
        Ok((send, recv))
    }

    pub async fn ensure_stage_transport_bridge(
        &self,
        peer_id: EndpointId,
        topology_id: impl Into<String>,
        run_id: impl Into<String>,
        stage_id: impl Into<String>,
    ) -> Result<String> {
        let topology_id = topology_id.into();
        let run_id = run_id.into();
        let stage_id = stage_id.into();
        let key = stage_runtime_status_key(&topology_id, &run_id, &stage_id);
        if self.stage_transport_bridges.lock().await.contains_key(&key) {
            anyhow::bail!(
                "stage transport bridge already exists for {topology_id}/{run_id}/{stage_id}"
            );
        }

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let bind_addr = listener.local_addr()?.to_string();
        let node = self.clone();
        let topology_for_task = topology_id.clone();
        let run_for_task = run_id.clone();
        let stage_for_task = stage_id.clone();
        let handle = tokio::spawn(async move {
            loop {
                let Ok((tcp_stream, _)) = listener.accept().await else {
                    break;
                };
                let node = node.clone();
                let topology_id = topology_for_task.clone();
                let run_id = run_for_task.clone();
                let stage_id = stage_for_task.clone();
                tokio::spawn(async move {
                    if let Err(err) = async {
                        tcp_stream.set_nodelay(true)?;
                        let (send, recv) = node
                            .open_stage_transport_stream(peer_id, topology_id, run_id, stage_id)
                            .await?;
                        let (tcp_read, tcp_write) = tokio::io::split(tcp_stream);
                        crate::network::tunnel::relay_bidirectional(tcp_read, tcp_write, send, recv)
                            .await
                    }
                    .await
                    {
                        tracing::warn!(
                            "stage transport bridge to {} ended: {err}",
                            peer_id.fmt_short()
                        );
                    }
                });
            }
        });
        self.stage_transport_bridges
            .lock()
            .await
            .insert(key, handle);
        Ok(bind_addr)
    }

    pub(crate) async fn stop_stage_transport_bridge(
        &self,
        topology_id: &str,
        run_id: &str,
        stage_id: &str,
    ) {
        let key = stage_runtime_status_key(topology_id, run_id, stage_id);
        if let Some(handle) = self.stage_transport_bridges.lock().await.remove(&key) {
            handle.abort();
        }
    }

    pub fn record_inference_attempt(
        &self,
        model: Option<&str>,
        target: &crate::inference::election::InferenceTarget,
        queue_wait: std::time::Duration,
        attempt_time: std::time::Duration,
        outcome: crate::network::metrics::AttemptOutcome,
        completion_tokens: Option<u64>,
    ) {
        let attempt_target = match target {
            crate::inference::election::InferenceTarget::Local(port) => {
                crate::network::metrics::AttemptTarget::Local(format!("127.0.0.1:{port}"))
            }
            crate::inference::election::InferenceTarget::Remote(peer_id) => {
                crate::network::metrics::AttemptTarget::Remote(peer_id.fmt_short().to_string())
            }
            crate::inference::election::InferenceTarget::None => return,
        };
        self.routing_metrics.record_attempt(
            model,
            attempt_target.clone(),
            queue_wait,
            attempt_time,
            outcome,
            completion_tokens,
        );
        if let Some(sink) = self.routing_telemetry_sink() {
            sink.record_route_attempt(model, &attempt_target, outcome);
        }
        self.publish_routing_runtime_snapshot();
    }

    pub fn record_endpoint_attempt(
        &self,
        model: Option<&str>,
        endpoint: &str,
        queue_wait: std::time::Duration,
        attempt_time: std::time::Duration,
        outcome: crate::network::metrics::AttemptOutcome,
        completion_tokens: Option<u64>,
    ) {
        let model_ref = model.map(canonical_demand_model_ref);
        let attempt_target = crate::network::metrics::AttemptTarget::Endpoint(endpoint.to_string());
        self.routing_metrics.record_attempt(
            model_ref.as_deref(),
            attempt_target.clone(),
            queue_wait,
            attempt_time,
            outcome,
            completion_tokens,
        );
        if let Some(sink) = self.routing_telemetry_sink() {
            sink.record_route_attempt(model_ref.as_deref(), &attempt_target, outcome);
        }
        self.publish_routing_runtime_snapshot();
    }

    pub fn record_routed_request(
        &self,
        model: Option<&str>,
        attempts: usize,
        outcome: crate::network::metrics::RequestOutcome,
    ) {
        let model_ref = model.map(canonical_demand_model_ref);
        self.routing_metrics
            .record_request(model_ref.as_deref(), attempts, outcome);
        if let Some(sink) = self.routing_telemetry_sink() {
            sink.record_model_request(model_ref.as_deref(), attempts, outcome);
        }
        self.publish_routing_runtime_snapshot();
    }

    pub fn local_request_metrics_snapshot(&self) -> LocalRequestMetricsSnapshot {
        self.local_request_metrics.snapshot()
    }

    pub(crate) fn runtime_data_collector(&self) -> crate::runtime_data::RuntimeDataCollector {
        self.runtime_data_producer.collector()
    }

    pub async fn owner_summary(&self) -> OwnershipSummary {
        self.owner_summary.lock().await.clone()
    }

    pub async fn control_endpoint(&self) -> Option<String> {
        let guard = self.control_listener.lock().await;
        guard.as_ref().map(|listener| listener.token.clone())
    }

    pub async fn shutdown_control_listener(&self) {
        let lifecycle = self.control_listener.lock().await.take();
        if let Some(lifecycle) = lifecycle {
            lifecycle
                .shutdown_requested
                .store(true, std::sync::atomic::Ordering::Release);
            lifecycle.shutdown.notify_waiters();
            let _ = lifecycle.task.await;
            drop(lifecycle.endpoint);
        }
    }

    pub async fn start(
        role: NodeRole,
        relay_urls: &[String],
        quic_bind: QuicBindSelection,
        max_vram_gb: Option<f64>,
        enumerate_host: bool,
        owner_config: Option<OwnerRuntimeConfig>,
        config_path: Option<&std::path::Path>,
    ) -> Result<(Self, TunnelChannels)> {
        let secret_key = startup_secret_key(&role).await?;
        let endpoint = bind_mesh_endpoint(secret_key.clone(), relay_urls, quic_bind).await?;
        // Wait briefly for relay connection so the invite token includes the relay URL.
        // On sinkholed networks this times out and we proceed without relay (direct UDP only).
        wait_for_endpoint_online(
            &endpoint,
            "Relay connected",
            "Relay connection timed out (5s) — proceeding without relay",
        )
        .await;

        // Discover public IP via STUN so the invite token includes it.
        // With --bind-port, the advertised port is the bound port (for port forwarding).
        // Without --bind-port, port 0 is intentional: it asks the OS for a conflict-free
        // ephemeral port. The IP is still useful for hole-punching.
        // Relay STUN may not work on sinkholed networks, so we use raw STUN to Google/Cloudflare.
        let stun_port = quic_bind.port.unwrap_or(EPHEMERAL_QUIC_PORT);
        let public_addr = stun_public_addr(stun_port).await;

        let (peer_change_tx, peer_change_rx) = watch::channel(0usize);
        let (inflight_change_tx, _inflight_change_rx) = watch::channel(0u64);
        let (tunnel_tx, tunnel_rx) = tokio::sync::mpsc::channel(256);
        let (tunnel_http_tx, tunnel_http_rx) = tokio::sync::mpsc::channel(256);
        let (stage_transport_tx, stage_transport_rx) = tokio::sync::mpsc::channel(256);

        let hardware =
            hardware_snapshot_for_start(crate::system::hardware::survey(), &role, max_vram_gb);
        let owner_runtime = init_owner_runtime(
            owner_config.as_ref(),
            endpoint.id(),
            hardware.hostname.clone(),
        )?;
        let owner_summary = verify_node_ownership(
            owner_runtime.owner_attestation.as_ref(),
            endpoint.id().as_bytes(),
            &owner_runtime.trust_store,
            TrustPolicy::Off,
            current_time_unix_ms(),
        );
        let config_state_init = {
            let path = crate::plugin::config_path(config_path)
                .unwrap_or_else(|_| std::path::PathBuf::from("config.toml"));
            crate::runtime::config_state::ConfigState::load(&path)?
        };
        let config_revision_init = config_state_init.revision();
        let runtime_data_collector = crate::runtime_data::RuntimeDataCollector::new();
        let runtime_data_producer =
            runtime_data_collector.producer(crate::runtime_data::RuntimeDataSource {
                scope: "routing",
                plugin_data_key: None,
                plugin_endpoint_key: None,
            });

        let node = Node {
            endpoint,
            public_addr,
            quic_bind,
            state: Arc::new(Mutex::new(MeshState {
                peers: HashMap::new(),
                connections: HashMap::new(),
                remote_tunnel_maps: HashMap::new(),
                dead_peers: HashMap::new(),
                peer_down_rejections: HashMap::new(),
                seen_plugin_messages: HashMap::new(),
                seen_plugin_message_order: VecDeque::new(),
                policy_rejected_peers: HashMap::new(),
            })),
            role: Arc::new(Mutex::new(role)),
            models: Arc::new(Mutex::new(Vec::new())),
            model_source: Arc::new(Mutex::new(None)),
            serving_models: Arc::new(Mutex::new(Vec::new())),
            served_model_descriptors: Arc::new(Mutex::new(Vec::new())),
            model_runtime_descriptors: Arc::new(Mutex::new(Vec::new())),
            hosted_models: Arc::new(Mutex::new(Vec::new())),
            llama_ready: Arc::new(Mutex::new(false)),
            available_models: Arc::new(Mutex::new(Vec::new())),
            requested_models: Arc::new(Mutex::new(Vec::new())),
            explicit_model_interests: Arc::new(Mutex::new(Vec::new())),
            model_demand: Arc::new(std::sync::Mutex::new(HashMap::new())),
            mesh_id: Arc::new(Mutex::new(None)),
            first_joined_mesh_ts: Arc::new(Mutex::new(None)),
            accepting: Arc::new((
                tokio::sync::Notify::new(),
                std::sync::atomic::AtomicBool::new(false),
            )),
            vram_bytes: hardware.vram_bytes,
            peer_change_tx,
            peer_change_rx,
            inflight_requests: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            inflight_change_tx,
            routing_metrics: crate::network::metrics::RoutingMetrics::default(),
            routing_telemetry: Arc::new(std::sync::Mutex::new(None)),
            local_request_metrics: Arc::new(LocalRequestMetricsSampler::default()),
            runtime_data_producer,
            tunnel_tx,
            tunnel_http_tx,
            stage_transport_tx,
            stage_control_tx: Arc::new(Mutex::new(None)),
            stage_transport_bridges: Arc::new(Mutex::new(HashMap::new())),
            stage_topologies: Arc::new(Mutex::new(StageTopologyState::default())),
            plugin_manager: Arc::new(Mutex::new(None)),
            display_name: Arc::new(Mutex::new(None)),
            owner_attestation: Arc::new(Mutex::new(owner_runtime.owner_attestation)),
            owner_summary: Arc::new(Mutex::new(owner_summary)),
            control_listener: Arc::new(Mutex::new(None)),
            trust_store: Arc::new(Mutex::new(owner_runtime.trust_store)),
            trust_policy: owner_runtime.trust_policy,
            enumerate_host,
            gpu_name: hardware.gpu_name,
            hostname: hardware.hostname,
            is_soc: hardware.is_soc,
            gpu_vram: hardware.gpu_vram,
            gpu_reserved_bytes: hardware.gpu_reserved_bytes,
            gpu_mem_bandwidth_gbps: Arc::new(tokio::sync::Mutex::new(None)),
            gpu_compute_tflops_fp32: Arc::new(tokio::sync::Mutex::new(None)),
            gpu_compute_tflops_fp16: Arc::new(tokio::sync::Mutex::new(None)),
            config_state: Arc::new(tokio::sync::Mutex::new(config_state_init)),
            config_revision_tx: {
                let (tx, _rx) = tokio::sync::watch::channel(config_revision_init);
                Arc::new(tx)
            },
        };

        node.maybe_start_control_listener(
            secret_key,
            owner_config.as_ref().and_then(|config| config.control_bind),
            owner_config
                .as_ref()
                .and_then(|config| config.control_advertise_addr),
            Some(relay_urls),
        )
        .await?;

        // Accept loop starts but waits for start_accepting() before processing connections.
        // This lets a node exist before it is ready to accept mesh traffic.
        let node2 = node.clone();
        tokio::spawn(async move {
            node2.accept_loop().await;
        });

        Ok((
            node,
            TunnelChannels {
                rpc: tunnel_rx,
                http: tunnel_http_rx,
                stage: stage_transport_rx,
            },
        ))
    }

    #[cfg(test)]
    pub async fn new_for_tests(role: NodeRole) -> Result<Self> {
        let (node, _) = Self::new_for_tests_with_secret(role).await?;
        Ok(node)
    }

    #[cfg(test)]
    pub(crate) async fn new_for_tests_with_secret(role: NodeRole) -> Result<(Self, SecretKey)> {
        let (node, secret_key) = {
            let secret_key = SecretKey::generate();
            let transport_config = iroh::endpoint::QuicTransportConfig::builder()
                .max_concurrent_bidi_streams(1024u32.into())
                .build();
            let endpoint = Endpoint::builder(iroh::endpoint::presets::Minimal)
                .secret_key(secret_key.clone())
                .alpns(vec![ALPN.to_vec(), skippy_protocol::STAGE_ALPN_V1.to_vec()])
                .relay_mode(iroh::endpoint::RelayMode::Disabled)
                .transport_config(transport_config)
                .bind_addr(std::net::SocketAddr::from(([127, 0, 0, 1], 0)))?
                .bind()
                .await?;
            (
                Self::new_test_node_from_endpoint(role, endpoint),
                secret_key,
            )
        };
        Ok((node, secret_key))
    }

    #[cfg(test)]
    fn new_test_node_from_endpoint(role: NodeRole, endpoint: Endpoint) -> Self {
        let (peer_change_tx, peer_change_rx) = watch::channel(0usize);
        let (inflight_change_tx, _inflight_change_rx) = watch::channel(0u64);
        let (tunnel_tx, _tunnel_rx) = tokio::sync::mpsc::channel(256);
        let (tunnel_http_tx, _tunnel_http_rx) = tokio::sync::mpsc::channel(256);
        let (stage_transport_tx, _stage_transport_rx) = tokio::sync::mpsc::channel(256);
        let runtime_data_collector = crate::runtime_data::RuntimeDataCollector::new();
        let runtime_data_producer =
            runtime_data_collector.producer(crate::runtime_data::RuntimeDataSource {
                scope: "routing",
                plugin_data_key: None,
                plugin_endpoint_key: None,
            });

        Node {
            endpoint,
            public_addr: None,
            quic_bind: QuicBindSelection::default(),
            state: Arc::new(Mutex::new(MeshState {
                peers: HashMap::new(),
                connections: HashMap::new(),
                remote_tunnel_maps: HashMap::new(),
                dead_peers: HashMap::new(),
                peer_down_rejections: HashMap::new(),
                seen_plugin_messages: HashMap::new(),
                seen_plugin_message_order: VecDeque::new(),
                policy_rejected_peers: HashMap::new(),
            })),
            role: Arc::new(Mutex::new(role)),
            models: Arc::new(Mutex::new(Vec::new())),
            model_source: Arc::new(Mutex::new(None)),
            serving_models: Arc::new(Mutex::new(Vec::new())),
            served_model_descriptors: Arc::new(Mutex::new(Vec::new())),
            model_runtime_descriptors: Arc::new(Mutex::new(Vec::new())),
            hosted_models: Arc::new(Mutex::new(Vec::new())),
            llama_ready: Arc::new(Mutex::new(false)),
            available_models: Arc::new(Mutex::new(Vec::new())),
            requested_models: Arc::new(Mutex::new(Vec::new())),
            explicit_model_interests: Arc::new(Mutex::new(Vec::new())),
            model_demand: Arc::new(std::sync::Mutex::new(HashMap::new())),
            mesh_id: Arc::new(Mutex::new(None)),
            first_joined_mesh_ts: Arc::new(Mutex::new(None)),
            accepting: Arc::new((
                tokio::sync::Notify::new(),
                std::sync::atomic::AtomicBool::new(false),
            )),
            vram_bytes: 0,
            peer_change_tx,
            peer_change_rx,
            inflight_requests: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            inflight_change_tx,
            routing_metrics: crate::network::metrics::RoutingMetrics::default(),
            routing_telemetry: Arc::new(std::sync::Mutex::new(None)),
            local_request_metrics: Arc::new(LocalRequestMetricsSampler::default()),
            runtime_data_producer,
            tunnel_tx,
            tunnel_http_tx,
            stage_transport_tx,
            stage_control_tx: Arc::new(Mutex::new(None)),
            stage_transport_bridges: Arc::new(Mutex::new(HashMap::new())),
            stage_topologies: Arc::new(Mutex::new(StageTopologyState::default())),
            plugin_manager: Arc::new(Mutex::new(None)),
            display_name: Arc::new(Mutex::new(None)),
            owner_attestation: Arc::new(Mutex::new(None)),
            owner_summary: Arc::new(Mutex::new(OwnershipSummary::default())),
            control_listener: Arc::new(Mutex::new(None)),
            trust_store: Arc::new(Mutex::new(TrustStore::default())),
            trust_policy: TrustPolicy::Off,
            enumerate_host: false,
            gpu_name: None,
            hostname: None,
            is_soc: None,
            gpu_vram: None,
            gpu_reserved_bytes: None,
            gpu_mem_bandwidth_gbps: Arc::new(tokio::sync::Mutex::new(None)),
            gpu_compute_tflops_fp32: Arc::new(tokio::sync::Mutex::new(None)),
            gpu_compute_tflops_fp16: Arc::new(tokio::sync::Mutex::new(None)),
            config_state: Arc::new(tokio::sync::Mutex::new(
                crate::runtime::config_state::ConfigState::default(),
            )),
            config_revision_tx: {
                let (tx, _rx) = tokio::sync::watch::channel(0);
                Arc::new(tx)
            },
        }
    }

    async fn maybe_start_control_listener(
        &self,
        secret_key: SecretKey,
        bind_addr: Option<std::net::SocketAddr>,
        advertise_addr: Option<std::net::SocketAddr>,
        relay_urls: Option<&[String]>,
    ) -> Result<()> {
        if self.local_verified_owner_id().await.is_none() {
            return Ok(());
        }

        let mut builder = Endpoint::builder(iroh::endpoint::presets::Minimal)
            .secret_key(secret_key)
            .alpns(vec![ALPN_CONTROL_V1.to_vec()])
            .bind_addr(bind_addr.unwrap_or_else(default_control_bind_addr))?;
        builder = configure_control_relay(builder, relay_urls);
        let endpoint = builder.bind().await?;
        if relay_urls.is_some() {
            wait_for_endpoint_online(
                &endpoint,
                "Owner-control relay connected",
                "Owner-control relay connection timed out (5s) — proceeding with direct endpoint addresses only",
            )
            .await;
        }
        let token = encode_endpoint_addr_token(&control_endpoint_addr(&endpoint, advertise_addr));
        let shutdown_requested = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let shutdown = Arc::new(tokio::sync::Notify::new());
        let task_endpoint = endpoint.clone();
        let task_shutdown_requested = shutdown_requested.clone();
        let task_shutdown = shutdown.clone();
        let node = self.clone();
        let task = tokio::spawn(Box::pin(async move {
            node.control_accept_loop(task_endpoint, task_shutdown_requested, task_shutdown)
                .await;
        }));
        *self.control_listener.lock().await = Some(ControlListenerLifecycle {
            endpoint,
            token,
            shutdown_requested,
            shutdown,
            task,
        });
        Ok(())
    }

    fn plugin_manager_local_kind(&self) -> crate::plugin::proto::mesh_event::Kind {
        if self.accepting.1.load(std::sync::atomic::Ordering::Acquire) {
            crate::plugin::proto::mesh_event::Kind::LocalAccepting
        } else {
            crate::plugin::proto::mesh_event::Kind::LocalStandby
        }
    }

    async fn broadcast_existing_mesh_snapshot(
        &self,
        plugin_manager: &crate::plugin::PluginManager,
        peers: Vec<PeerInfo>,
    ) {
        let _ = plugin_manager
            .broadcast_mesh_event(
                self.build_mesh_event(self.plugin_manager_local_kind(), None, String::new())
                    .await,
            )
            .await;
        if self.mesh_id.lock().await.is_some() {
            let _ = plugin_manager
                .broadcast_mesh_event(
                    self.build_mesh_event(
                        crate::plugin::proto::mesh_event::Kind::MeshIdUpdated,
                        None,
                        String::new(),
                    )
                    .await,
                )
                .await;
        }
        for peer in peers {
            if let Err(err) = plugin_manager
                .broadcast_mesh_event(
                    self.build_mesh_event(
                        crate::plugin::proto::mesh_event::Kind::PeerUp,
                        Some(peer_info_to_mesh_peer(&peer)),
                        String::new(),
                    )
                    .await,
                )
                .await
            {
                tracing::debug!(
                    "Failed to send existing peer snapshot to plugins for {}: {err}",
                    peer.id.fmt_short()
                );
            }
        }
    }

    #[cfg(test)]
    pub async fn insert_test_peer(&self, peer: PeerInfo) {
        self.state.lock().await.peers.insert(peer.id, peer);
    }

    pub fn invite_token(&self) -> String {
        let mut addr = self.endpoint_addr_for_advertisement();
        // Inject STUN-discovered public address if relay STUN didn't provide one.
        if let Some(pub_addr) = self.public_addr {
            if !endpoint_addr_has_public_ipv4(&addr) {
                addr.addrs.insert(TransportAddr::Ip(pub_addr));
            }
        }
        addr = filter_endpoint_addr_for_bind_ip(addr, self.quic_bind.ip);
        let json = serde_json::to_vec(&addr).expect("serializable");
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&json)
    }

    fn endpoint_addr_for_advertisement(&self) -> EndpointAddr {
        let mut addr = self.endpoint.addr();
        if self.quic_bind.ip.is_some() {
            addr = filter_endpoint_addr_for_bind_ip(addr, self.quic_bind.ip);
        }
        addr
    }

    /// Decode an invite token into an [`EndpointAddr`] without connecting.
    /// Returns `Err` if the token is not valid base64 or not valid JSON.
    pub fn decode_invite_token(invite_token: &str) -> Result<EndpointAddr> {
        let json = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(invite_token)
            .context("invalid invite token encoding")?;
        serde_json::from_slice(&json).context("invalid invite token JSON")
    }

    #[cfg(test)]
    pub async fn sync_from_peer_for_tests(&self, remote: &Self) {
        let remote_id = remote.endpoint.id();
        let their_announcements = remote.collect_announcements().await;
        for ann in &their_announcements {
            if ann.addr.id == self.endpoint.id() {
                continue;
            }
            if ann.addr.id == remote_id {
                if let Some(ref their_id) = ann.mesh_id {
                    self.set_mesh_id(their_id.clone()).await;
                }
                self.merge_remote_demand(&ann.model_demand);
                self.add_peer(remote_id, ann.addr.clone(), ann).await;
            } else {
                self.update_transitive_peer(ann.addr.id, &ann.addr, ann, remote_id)
                    .await;
            }
        }
    }

    async fn build_mesh_event(
        &self,
        kind: crate::plugin::proto::mesh_event::Kind,
        peer: Option<crate::plugin::proto::MeshPeer>,
        detail_json: String,
    ) -> crate::plugin::proto::MeshEvent {
        crate::plugin::proto::MeshEvent {
            kind: kind as i32,
            peer,
            local_peer_id: endpoint_id_hex(self.endpoint.id()),
            mesh_id: self.mesh_id.lock().await.clone().unwrap_or_default(),
            detail_json,
        }
    }

    /// Enable accepting inbound connections. Call before join() or when ready to participate.
    /// Until this is called, the accept loop blocks waiting.
    pub fn start_accepting(&self) {
        self.accepting
            .1
            .store(true, std::sync::atomic::Ordering::Release);
        self.accepting.0.notify_waiters();
        let node = self.clone();
        tokio::spawn(async move {
            let plugin_manager = node.plugin_manager.lock().await.clone();
            if let Some(plugin_manager) = plugin_manager {
                let _ = plugin_manager
                    .broadcast_mesh_event(
                        node.build_mesh_event(
                            crate::plugin::proto::mesh_event::Kind::LocalAccepting,
                            None,
                            String::new(),
                        )
                        .await,
                    )
                    .await;
            }
        });
    }

    pub async fn join(&self, invite_token: &str) -> Result<()> {
        let json = base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(invite_token)?;
        let addr: EndpointAddr = serde_json::from_slice(&json)?;
        // Clear dead status — explicit join should always attempt connection
        self.state.lock().await.dead_peers.remove(&addr.id);
        self.connect_to_peer(addr).await
    }

    /// Like [`join`], but retries once after a delay on transient (connect/timeout)
    /// errors.  Decode errors (invalid base64/JSON) fail immediately.
    pub async fn join_with_retry(&self, invite_token: &str) -> Result<()> {
        // Decode first — bail immediately on bad tokens.
        let json = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(invite_token)
            .context("invalid invite token encoding")?;
        let addr: EndpointAddr =
            serde_json::from_slice(&json).context("invalid invite token JSON")?;

        // Three attempts with increasing backoff.  Relay-only joins need
        // WebSocket setup + QUIC handshake at high RTT — two attempts at
        // 15s were not enough.  Three at 30s with 5s/10s gaps give ~105s
        // total budget which covers all but the worst relay conditions.
        let backoffs = [5, 10];
        self.state.lock().await.dead_peers.remove(&addr.id);
        let mut last_err = match self.connect_to_peer(addr.clone()).await {
            Ok(()) => return Ok(()),
            Err(e) => e,
        };
        for (attempt, delay_secs) in backoffs.iter().enumerate() {
            tracing::info!(
                "Join attempt {} failed ({last_err:#}), retrying in {delay_secs}s...",
                attempt + 1
            );
            tokio::time::sleep(std::time::Duration::from_secs(*delay_secs)).await;
            self.state.lock().await.dead_peers.remove(&addr.id);
            match self.connect_to_peer(addr.clone()).await {
                Ok(()) => return Ok(()),
                Err(e) => last_err = e,
            }
        }
        Err(last_err)
    }

    /// Connect to a peer without gossip exchange — for passive nodes (clients/standby).
    pub fn id(&self) -> EndpointId {
        self.endpoint.id()
    }

    pub async fn role(&self) -> NodeRole {
        self.role.lock().await.clone()
    }

    pub async fn set_role(&self, role: NodeRole) {
        *self.role.lock().await = role;
    }

    pub async fn set_models(&self, models: Vec<String>) {
        *self.models.lock().await = models;
    }

    pub async fn models(&self) -> Vec<String> {
        self.models.lock().await.clone()
    }

    pub async fn set_model_source(&self, source: String) {
        *self.model_source.lock().await = Some(source);
        self.refresh_served_model_descriptors().await;
    }

    pub async fn set_serving_models(&self, models: Vec<String>) {
        *self.serving_models.lock().await = models;
        self.refresh_served_model_descriptors().await;
    }

    pub async fn set_served_model_descriptors(&self, descriptors: Vec<ServedModelDescriptor>) {
        let model_names: std::collections::HashSet<_> = descriptors
            .iter()
            .map(|descriptor| descriptor.identity.model_name.clone())
            .collect();
        *self.served_model_descriptors.lock().await = descriptors;
        self.model_runtime_descriptors
            .lock()
            .await
            .retain(|runtime| model_names.contains(&runtime.model_name));
    }

    pub async fn upsert_served_model_descriptor(&self, descriptor: ServedModelDescriptor) {
        let mut descriptors = self.served_model_descriptors.lock().await;
        if let Some(existing) = descriptors
            .iter_mut()
            .find(|existing| existing.identity.model_name == descriptor.identity.model_name)
        {
            *existing = descriptor;
        } else {
            descriptors.push(descriptor);
        }
    }

    pub async fn remove_served_model_descriptor(&self, model_name: &str) {
        self.served_model_descriptors
            .lock()
            .await
            .retain(|descriptor| descriptor.identity.model_name != model_name);
        self.model_runtime_descriptors
            .lock()
            .await
            .retain(|runtime| runtime.model_name != model_name);
    }

    pub async fn set_model_runtime_context_length(
        &self,
        model_name: &str,
        context_length: Option<u32>,
    ) {
        let identity_hash = self
            .served_model_descriptors
            .lock()
            .await
            .iter()
            .find(|descriptor| descriptor.identity.model_name == model_name)
            .and_then(|descriptor| descriptor.identity.identity_hash.clone());
        let mut runtimes = self.model_runtime_descriptors.lock().await;
        if let Some(context_length) = context_length {
            if let Some(runtime) = runtimes
                .iter_mut()
                .find(|runtime| runtime.model_name == model_name)
            {
                runtime.identity_hash = identity_hash.or_else(|| runtime.identity_hash.clone());
                runtime.context_length = Some(context_length);
                runtime.ready = true;
            } else {
                runtimes.push(ModelRuntimeDescriptor {
                    model_name: model_name.to_string(),
                    identity_hash,
                    context_length: Some(context_length),
                    ready: true,
                });
            }
        } else {
            runtimes.retain(|runtime| runtime.model_name != model_name);
        }
    }

    pub async fn local_model_context_length(&self, model_name: &str) -> Option<u32> {
        self.model_runtime_descriptors
            .lock()
            .await
            .iter()
            .find(|runtime| runtime.model_name == model_name)
            .and_then(ModelRuntimeDescriptor::advertised_context_length)
    }

    pub async fn peer_model_context_length(
        &self,
        peer_id: EndpointId,
        model_name: &str,
    ) -> Option<u32> {
        self.state
            .lock()
            .await
            .peers
            .get(&peer_id)
            .and_then(|peer| peer.advertised_context_length(model_name))
    }

    pub async fn served_model_descriptors(&self) -> Vec<ServedModelDescriptor> {
        self.served_model_descriptors.lock().await.clone()
    }

    pub async fn serving_models(&self) -> Vec<String> {
        self.serving_models.lock().await.clone()
    }

    pub async fn set_hosted_models(&self, models: Vec<String>) {
        *self.hosted_models.lock().await = models;
    }

    pub async fn hosted_models(&self) -> Vec<String> {
        self.hosted_models.lock().await.clone()
    }

    async fn refresh_served_model_descriptors(&self) {
        let serving_models = self.serving_models.lock().await.clone();
        let existing_by_name: HashMap<_, _> = self
            .served_model_descriptors
            .lock()
            .await
            .iter()
            .map(|descriptor| (descriptor.identity.model_name.clone(), descriptor.clone()))
            .collect();
        let mut descriptors = if let Some(primary_model_name) = serving_models.first() {
            let model_source = self.model_source.lock().await.clone();
            let primary_model_path = crate::models::find_model_path(primary_model_name);
            infer_served_model_descriptors(
                primary_model_name,
                &serving_models,
                model_source.as_deref(),
                Some(primary_model_path.as_path()),
            )
        } else {
            Vec::new()
        };
        for descriptor in &mut descriptors {
            if let Some(existing) = existing_by_name.get(&descriptor.identity.model_name) {
                descriptor.capabilities = existing.capabilities;
                descriptor.capabilities_known = existing.capabilities_known;
                if existing.topology.is_some() {
                    descriptor.topology = existing.topology.clone();
                }
            }
        }
        self.set_served_model_descriptors(descriptors).await;
    }

    /// Set the operator-facing display name for this node.
    pub async fn set_display_name(&self, name: String) {
        *self.display_name.lock().await = Some(name);
    }

    pub async fn set_plugin_manager(&self, plugin_manager: crate::plugin::PluginManager) {
        let peers = {
            let state = self.state.lock().await;
            state.peers.values().cloned().collect::<Vec<_>>()
        };
        *self.plugin_manager.lock().await = Some(plugin_manager.clone());
        self.broadcast_existing_mesh_snapshot(&plugin_manager, peers)
            .await;
    }

    pub async fn plugin_manager(&self) -> Option<crate::plugin::PluginManager> {
        self.plugin_manager.lock().await.clone()
    }

    pub fn start_plugin_channel_forwarder(
        &self,
        mut rx: tokio::sync::mpsc::Receiver<crate::plugin::PluginMeshEvent>,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            while let Some(event) = rx.recv().await {
                if let Err(err) = node.forward_plugin_event(event).await {
                    tracing::debug!("Plugin mesh forward failed: {err}");
                }
            }
        });
    }

    async fn emit_plugin_mesh_event(
        &self,
        kind: crate::plugin::proto::mesh_event::Kind,
        peer: Option<&PeerInfo>,
        detail_json: String,
    ) {
        let plugin_manager = self.plugin_manager.lock().await.clone();
        if let Some(plugin_manager) = plugin_manager {
            if let Err(err) = plugin_manager
                .broadcast_mesh_event(
                    self.build_mesh_event(kind, peer.map(peer_info_to_mesh_peer), detail_json)
                        .await,
                )
                .await
            {
                tracing::debug!(
                    "Failed to deliver plugin mesh event {:?} for {}: {err}",
                    kind,
                    peer.map(|p| p.id.fmt_short().to_string())
                        .unwrap_or_else(|| self.endpoint.id().fmt_short().to_string())
                );
            }
        }
    }

    async fn update_peer_rtt(&self, id: EndpointId, rtt_ms: u32) {
        // 0ms is not a valid network RTT — it indicates a measurement artifact
        // (e.g. local buffer time before the actual network round-trip).
        if rtt_ms == 0 {
            return;
        }
        let (updated_peer, old_rtt) = {
            let mut state = self.state.lock().await;
            if let Some(peer) = state.peers.get_mut(&id) {
                let prev = peer.rtt_ms;
                // Only accept equal-or-lower RTT. Gossip round-trip timing
                // can inflate the value when routed via relay, overwriting a
                // good direct-path measurement. The RTT gate only cares about
                // "fast enough for split", so keeping the best-seen value is
                // correct — if the path truly degrades the peer will be
                // unreachable and removed via the normal liveness path.
                if prev.is_some_and(|p| rtt_ms > p) {
                    // Store display_rtt regardless (for UI refresh), but don't update best RTT.
                    peer.display_rtt = Some(DirectLatencyObservation {
                        rtt_ms,
                        observed_at: std::time::Instant::now(),
                    });
                    return;
                }
                peer.rtt_ms = Some(rtt_ms);
                peer.display_rtt = Some(DirectLatencyObservation {
                    rtt_ms,
                    observed_at: std::time::Instant::now(),
                });
                (Some(peer.clone()), prev)
            } else {
                (None, None)
            }
        };
        if let Some(peer) = updated_peer {
            tracing::info!("Peer {} RTT: {}ms", id.fmt_short(), rtt_ms);
            // If RTT dropped from above the split threshold (80ms) to below it
            // (e.g. relay → direct), trigger a re-election so the peer can now
            // be included in split mode.
            let was_above = old_rtt.is_some_and(|r| r > MAX_SPLIT_RTT_MS);
            if was_above && rtt_ms <= MAX_SPLIT_RTT_MS {
                emit_mesh_info(format!(
                    "📡 Peer {} RTT improved ({}ms → {}ms) — re-electing for split",
                    id.fmt_short(),
                    old_rtt.unwrap_or(0),
                    rtt_ms
                ));
                let count = self.state.lock().await.peers.len();
                let _ = self.peer_change_tx.send(count);
            }
            self.emit_plugin_mesh_event(
                crate::plugin::proto::mesh_event::Kind::PeerUpdated,
                Some(&peer),
                String::new(),
            )
            .await;
        }
    }

    /// Re-gossip our state to all connected peers.
    /// Call after changing assigned/hosted state, role, or configured models.
    pub async fn regossip(&self) {
        let conns: Vec<(EndpointId, Connection)> = {
            let state = self.state.lock().await;
            state
                .connections
                .iter()
                .map(|(id, c)| (*id, c.clone()))
                .collect()
        };
        for (peer_id, conn) in conns {
            let node = self.clone();
            tokio::spawn(async move {
                if let Err(e) = node.initiate_gossip(conn, peer_id).await {
                    tracing::debug!("Regossip to {} failed: {e}", peer_id.fmt_short());
                }
            });
        }
    }

    /// Gossip with one connected peer to update routing table.
    /// Used by: (1) passive nodes' periodic 60s heartbeat, (2) background
    /// refresh on tunnel failure so future requests have fresh routing.
    pub async fn gossip_one_peer(&self) {
        let conn = {
            let state = self.state.lock().await;
            state
                .connections
                .iter()
                .next()
                .map(|(id, c)| (*id, c.clone()))
        };
        if let Some((peer_id, conn)) = conn {
            let _ = self.initiate_gossip_inner(conn, peer_id, false).await;
        }
    }

    pub async fn is_llama_ready(&self) -> bool {
        *self.llama_ready.lock().await
    }

    pub async fn mesh_id(&self) -> Option<String> {
        self.mesh_id.lock().await.clone()
    }

    pub async fn first_joined_mesh_ts(&self) -> Option<u64> {
        *self.first_joined_mesh_ts.lock().await
    }

    pub async fn set_first_joined_mesh_ts_if_absent(&self, ts: u64) -> bool {
        let mut current = self.first_joined_mesh_ts.lock().await;
        if current.is_none() {
            *current = Some(ts);
            true
        } else {
            false
        }
    }

    /// Set the mesh identity. If None was set, adopts the given ID (from gossip).
    /// If already set, ignores (originator's ID wins).
    pub async fn set_mesh_id(&self, id: String) {
        let mut current = self.mesh_id.lock().await;
        if current.is_none() {
            *current = Some(id);
            drop(current);
            self.emit_plugin_mesh_event(
                crate::plugin::proto::mesh_event::Kind::MeshIdUpdated,
                None,
                String::new(),
            )
            .await;
        }
    }

    /// Set mesh ID unconditionally (for originator).
    pub async fn set_mesh_id_force(&self, id: String) {
        *self.mesh_id.lock().await = Some(id);
        self.emit_plugin_mesh_event(
            crate::plugin::proto::mesh_event::Kind::MeshIdUpdated,
            None,
            String::new(),
        )
        .await;
    }

    pub async fn set_available_models(&self, models: Vec<String>) {
        *self.available_models.lock().await = models;
    }

    pub async fn available_models(&self) -> Vec<String> {
        self.available_models.lock().await.clone()
    }

    /// Record a request for a model — updates the demand map.
    /// Called from API proxy on every request (including misses for unserved models).
    /// Uses std::sync::Mutex (not tokio) so it can be called from sync context too.
    pub fn record_request(&self, model: &str) {
        // "auto" is a routing directive, not a real model — don't pollute demand
        if model == "auto" || model.is_empty() {
            return;
        }
        let model_ref = canonical_demand_model_ref(model);
        let mut demand = self.model_demand.lock().unwrap();
        let entry = demand.entry(model_ref).or_default();
        entry.last_active = now_secs();
        entry.request_count += 1;
    }

    /// Get the current demand map (for gossip and assignment decisions).
    pub fn get_demand(&self) -> HashMap<String, ModelDemand> {
        self.model_demand.lock().unwrap().clone()
    }

    /// Merge incoming demand from gossip into our local map.
    pub fn merge_remote_demand(&self, remote: &HashMap<String, ModelDemand>) {
        let mut demand = self.model_demand.lock().unwrap();
        merge_demand(&mut demand, remote);
    }

    /// Remove demand entries that have expired (past TTL and not pinned).
    /// Call periodically to prevent unbounded map growth.
    pub async fn gc_demand(&self) {
        let now = now_secs();
        let my_requested = self.requested_models.lock().await;
        let peers = self.state.lock().await;
        let mut pinned: std::collections::HashSet<String> = my_requested.iter().cloned().collect();
        for p in peers.peers.values() {
            for m in &p.requested_models {
                pinned.insert(m.clone());
            }
        }
        drop(peers);
        drop(my_requested);

        let mut demand = self.model_demand.lock().unwrap();
        demand.retain(|model, d| pinned.contains(model) || (now - d.last_active) < DEMAND_TTL_SECS);
    }

    /// Get active demand entries (within TTL or pinned by a live node).
    /// This replaces mesh_wanted_models().
    pub async fn active_demand(&self) -> HashMap<String, ModelDemand> {
        let now = now_secs();
        let demand = self.model_demand.lock().unwrap().clone();

        // Check which models are pinned (declared via --model by self or a live peer)
        let my_requested = self.requested_models.lock().await;
        let peers = self.state.lock().await;
        let mut pinned: std::collections::HashSet<String> = my_requested.iter().cloned().collect();
        for p in peers.peers.values() {
            for m in &p.requested_models {
                pinned.insert(m.clone());
            }
        }
        drop(peers);
        drop(my_requested);

        demand
            .into_iter()
            .filter(|(model, d)| pinned.contains(model) || (now - d.last_active) < DEMAND_TTL_SECS)
            .collect()
    }

    pub async fn set_requested_models(&self, models: Vec<String>) {
        let models = models
            .into_iter()
            .map(|model| canonical_demand_model_ref(&model))
            .collect::<Vec<_>>();
        // Seed demand entries for --model declarations
        {
            let mut demand = self.model_demand.lock().unwrap();
            let now = now_secs();
            for m in &models {
                let entry = demand.entry(m.clone()).or_default();
                entry.last_active = entry.last_active.max(now);
            }
        }
        *self.requested_models.lock().await = models;
    }

    pub async fn requested_models(&self) -> Vec<String> {
        self.requested_models.lock().await.clone()
    }

    pub async fn set_explicit_model_interests(&self, mut model_refs: Vec<String>) {
        model_refs.retain(|model_ref| !model_ref.trim().is_empty());
        model_refs.sort();
        model_refs.dedup();
        *self.explicit_model_interests.lock().await = model_refs;
    }

    pub async fn explicit_model_interests(&self) -> Vec<String> {
        self.explicit_model_interests.lock().await.clone()
    }

    async fn forward_plugin_event(&self, event: crate::plugin::PluginMeshEvent) -> Result<()> {
        match event {
            crate::plugin::PluginMeshEvent::Channel {
                plugin_id,
                mut message,
            } => {
                if !self
                    .plugin_event_channel_declared(&plugin_id, &message.channel, "message")
                    .await
                {
                    return Ok(());
                }
                default_plugin_event_source(self.endpoint.id(), &mut message.source_peer_id);
                let frame = crate::plugin::proto::MeshChannelFrame {
                    plugin_id,
                    message_id: new_plugin_message_id(&message.source_peer_id),
                    message: Some(message),
                };
                if !self.remember_plugin_message(frame.message_id.clone()).await {
                    return Ok(());
                }
                self.broadcast_plugin_channel_frame(&frame, None).await
            }
            crate::plugin::PluginMeshEvent::BulkTransfer {
                plugin_id,
                mut message,
            } => {
                if !self
                    .plugin_event_channel_declared(&plugin_id, &message.channel, "bulk transfer")
                    .await
                {
                    return Ok(());
                }
                default_plugin_event_source(self.endpoint.id(), &mut message.source_peer_id);
                let frame = crate::plugin::proto::MeshBulkFrame {
                    plugin_id,
                    message_id: new_plugin_message_id(&message.source_peer_id),
                    message: Some(message),
                };
                if !self.remember_plugin_message(frame.message_id.clone()).await {
                    return Ok(());
                }
                self.broadcast_plugin_bulk_frame(&frame, None).await
            }
        }
    }

    async fn plugin_event_channel_declared(
        &self,
        plugin_id: &str,
        channel: &str,
        noun: &str,
    ) -> bool {
        let plugin_manager = self.plugin_manager.lock().await.clone();
        if let Some(plugin_manager) = plugin_manager {
            if !plugin_manager
                .plugin_declares_mesh_channel(plugin_id, channel)
                .await
            {
                tracing::debug!(
                    plugin = %plugin_id,
                    channel = %channel,
                    "Dropping outbound {noun} for undeclared mesh channel"
                );
                return false;
            }
        }
        true
    }

    async fn remember_plugin_message(&self, message_id: String) -> bool {
        /// How long to remember a message ID. Any duplicate arriving within
        /// this window is suppressed. This must be longer than the worst-case
        /// propagation delay across alternate mesh paths — 120s is generous.
        const DEDUP_TTL: std::time::Duration = std::time::Duration::from_secs(120);
        /// Hard cap to bound memory even if message volume is extreme.
        const DEDUP_HARD_CAP: usize = 100_000;

        let now = std::time::Instant::now();
        let mut state = self.state.lock().await;

        // Evict entries older than the TTL
        while let Some((ts, _)) = state.seen_plugin_message_order.front() {
            if now.duration_since(*ts) >= DEDUP_TTL {
                if let Some((_, id)) = state.seen_plugin_message_order.pop_front() {
                    state.seen_plugin_messages.remove(&id);
                }
            } else {
                break;
            }
        }

        // Already seen?
        if state.seen_plugin_messages.contains_key(&message_id) {
            return false;
        }

        // Hard cap: if under extreme load we still accumulate too many,
        // evict the oldest regardless of TTL.
        while state.seen_plugin_message_order.len() >= DEDUP_HARD_CAP {
            if let Some((_, id)) = state.seen_plugin_message_order.pop_front() {
                state.seen_plugin_messages.remove(&id);
            }
        }

        state.seen_plugin_messages.insert(message_id.clone(), now);
        state.seen_plugin_message_order.push_back((now, message_id));
        true
    }

    async fn broadcast_plugin_channel_frame(
        &self,
        frame: &crate::plugin::proto::MeshChannelFrame,
        skip_peer: Option<EndpointId>,
    ) -> Result<()> {
        let data = frame.encode_to_vec();
        let conns: Vec<(EndpointId, Connection)> = {
            let state = self.state.lock().await;
            state
                .connections
                .iter()
                .filter(|(peer_id, _)| Some(**peer_id) != skip_peer)
                .map(|(peer_id, conn)| (*peer_id, conn.clone()))
                .collect()
        };
        for (peer_id, conn) in conns {
            let bytes = data.clone();
            tokio::spawn(async move {
                let result = async {
                    let (mut send, _recv) = conn.open_bi().await?;
                    send.write_all(&[STREAM_PLUGIN_CHANNEL]).await?;
                    send.write_all(&(bytes.len() as u32).to_le_bytes()).await?;
                    send.write_all(&bytes).await?;
                    send.finish()?;
                    Ok::<_, anyhow::Error>(())
                }
                .await;
                if let Err(e) = result {
                    tracing::debug!(
                        "Failed to broadcast plugin frame to {}: {e}",
                        peer_id.fmt_short()
                    );
                }
            });
        }
        Ok(())
    }

    async fn broadcast_plugin_bulk_frame(
        &self,
        frame: &crate::plugin::proto::MeshBulkFrame,
        skip_peer: Option<EndpointId>,
    ) -> Result<()> {
        let data = frame.encode_to_vec();
        let conns: Vec<(EndpointId, Connection)> = {
            let state = self.state.lock().await;
            state
                .connections
                .iter()
                .filter(|(peer_id, _)| Some(**peer_id) != skip_peer)
                .map(|(peer_id, conn)| (*peer_id, conn.clone()))
                .collect()
        };
        for (peer_id, conn) in conns {
            let bytes = data.clone();
            tokio::spawn(async move {
                let result = async {
                    let (mut send, _recv) = conn.open_bi().await?;
                    send.write_all(&[STREAM_PLUGIN_BULK_TRANSFER]).await?;
                    send.write_all(&(bytes.len() as u32).to_le_bytes()).await?;
                    send.write_all(&bytes).await?;
                    send.finish()?;
                    Ok::<_, anyhow::Error>(())
                }
                .await;
                if let Err(e) = result {
                    tracing::debug!(
                        "Failed to broadcast plugin bulk frame to {}: {e}",
                        peer_id.fmt_short()
                    );
                }
            });
        }
        Ok(())
    }

    async fn handle_plugin_channel_stream(
        &self,
        _remote: EndpointId,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        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 > 10_000_000 {
            anyhow::bail!("Plugin channel frame too large");
        }
        let mut buf = vec![0u8; len];
        recv.read_exact(&mut buf).await?;
        send.finish()?;

        let frame = crate::plugin::proto::MeshChannelFrame::decode(buf.as_slice())?;
        if frame.plugin_id.is_empty() || frame.message_id.is_empty() {
            return Ok(());
        }
        if !self.remember_plugin_message(frame.message_id.clone()).await {
            return Ok(());
        }

        let Some(message) = frame.message.clone() else {
            return Ok(());
        };
        let local_peer_id = endpoint_id_hex(self.endpoint.id());
        let deliver_local =
            message.target_peer_id.is_empty() || message.target_peer_id == local_peer_id;

        if deliver_local {
            let plugin_manager = self.plugin_manager.lock().await.clone();
            if let Some(plugin_manager) = plugin_manager {
                plugin_manager
                    .dispatch_channel_message(crate::plugin::PluginMeshEvent::Channel {
                        plugin_id: frame.plugin_id.clone(),
                        message: message.clone(),
                    })
                    .await?;
            }
        }

        // Targeted messages: forward only to the specific target peer if we
        // have a direct connection.  Do NOT flood-broadcast targeted messages
        // to all connections — that causes O(N²) amplification across the mesh.
        // Untargeted broadcasts: deliver locally only.  The originator already
        // sent to all their direct connections.
        if !message.target_peer_id.is_empty() && message.target_peer_id != local_peer_id {
            // Look up connection to the target peer by hex ID
            let target_conn = {
                let state = self.state.lock().await;
                state
                    .connections
                    .iter()
                    .find(|(id, _)| endpoint_id_hex(**id) == message.target_peer_id)
                    .map(|(id, conn)| (*id, conn.clone()))
            };
            if let Some((_target_id, conn)) = target_conn {
                let data = frame.encode_to_vec();
                tokio::spawn(async move {
                    let result = async {
                        let (mut send, _recv) = conn.open_bi().await?;
                        send.write_all(&[STREAM_PLUGIN_CHANNEL]).await?;
                        send.write_all(&(data.len() as u32).to_le_bytes()).await?;
                        send.write_all(&data).await?;
                        send.finish()?;
                        Ok::<_, anyhow::Error>(())
                    }
                    .await;
                    if let Err(e) = result {
                        tracing::debug!("Failed to forward targeted plugin frame: {e}");
                    }
                });
            }
        }

        Ok(())
    }

    async fn handle_plugin_bulk_stream(
        &self,
        _remote: EndpointId,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        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 > 64_000_000 {
            anyhow::bail!("Plugin bulk frame too large");
        }
        let mut buf = vec![0u8; len];
        recv.read_exact(&mut buf).await?;
        send.finish()?;

        let frame = crate::plugin::proto::MeshBulkFrame::decode(buf.as_slice())?;
        if frame.plugin_id.is_empty() || frame.message_id.is_empty() {
            return Ok(());
        }
        if !self.remember_plugin_message(frame.message_id.clone()).await {
            return Ok(());
        }

        let Some(message) = frame.message.clone() else {
            return Ok(());
        };
        let local_peer_id = endpoint_id_hex(self.endpoint.id());
        let deliver_local =
            message.target_peer_id.is_empty() || message.target_peer_id == local_peer_id;

        if deliver_local {
            let plugin_manager = self.plugin_manager.lock().await.clone();
            if let Some(plugin_manager) = plugin_manager {
                plugin_manager
                    .dispatch_bulk_transfer_message(crate::plugin::PluginMeshEvent::BulkTransfer {
                        plugin_id: frame.plugin_id.clone(),
                        message: message.clone(),
                    })
                    .await?;
            }
        }

        // Same policy as channel frames: targeted → forward to target only,
        // broadcast → deliver locally only (originator already sent to their
        // direct connections).
        if !message.target_peer_id.is_empty() && message.target_peer_id != local_peer_id {
            let target_conn = {
                let state = self.state.lock().await;
                state
                    .connections
                    .iter()
                    .find(|(id, _)| endpoint_id_hex(**id) == message.target_peer_id)
                    .map(|(id, conn)| (*id, conn.clone()))
            };
            if let Some((_target_id, conn)) = target_conn {
                let data = frame.encode_to_vec();
                tokio::spawn(async move {
                    let result = async {
                        let (mut send, _recv) = conn.open_bi().await?;
                        send.write_all(&[STREAM_PLUGIN_BULK_TRANSFER]).await?;
                        send.write_all(&(data.len() as u32).to_le_bytes()).await?;
                        send.write_all(&data).await?;
                        send.finish()?;
                        Ok::<_, anyhow::Error>(())
                    }
                    .await;
                    if let Err(e) = result {
                        tracing::debug!("Failed to forward targeted plugin bulk frame: {e}");
                    }
                });
            }
        }

        Ok(())
    }

    /// Get the mesh catalog: local installed models plus mesh served/requested models.
    /// Returns deduplicated canonical model refs.
    pub async fn mesh_catalog(&self) -> Vec<String> {
        // Snapshot each lock independently to avoid holding multiple locks.
        let my_available = self.available_models.lock().await.clone();
        let my_requested = self.requested_models.lock().await.clone();
        let my_serving_models = self.serving_models.lock().await.clone();
        let peer_data: Vec<_> = {
            let state = self.state.lock().await;
            state
                .peers
                .values()
                .map(|p| {
                    (
                        p.available_models.clone(),
                        p.requested_models.clone(),
                        p.serving_models.clone(),
                    )
                })
                .collect()
        };
        let mut all = std::collections::HashSet::new();
        for m in &my_available {
            all.insert(m.clone());
        }
        for m in &my_requested {
            all.insert(m.clone());
        }
        for m in &my_serving_models {
            all.insert(m.clone());
        }
        for (avail, req, serving_models) in &peer_data {
            for m in avail {
                all.insert(m.clone());
            }
            for m in req {
                all.insert(m.clone());
            }
            for m in serving_models {
                all.insert(m.clone());
            }
        }
        let mut result: Vec<String> = all.into_iter().collect();
        result.sort();
        result
    }

    pub async fn mesh_catalog_entries(&self) -> Vec<MeshCatalogEntry> {
        let names = self.mesh_catalog().await;
        let my_available = self.available_models.lock().await.clone();
        let my_served_descriptors = self.served_model_descriptors.lock().await.clone();
        let peer_descriptors: Vec<_> = {
            let state = self.state.lock().await;
            state
                .peers
                .values()
                .map(|p| p.served_model_descriptors.clone())
                .collect()
        };

        let mut by_name: HashMap<String, ServedModelDescriptor> = HashMap::new();
        for descriptor in infer_available_model_descriptors(&my_available)
            .into_iter()
            .chain(my_served_descriptors)
        {
            upsert_mesh_catalog_descriptor(&mut by_name, descriptor);
        }
        for served in peer_descriptors {
            for descriptor in served {
                upsert_mesh_catalog_descriptor(&mut by_name, descriptor);
            }
        }

        names
            .into_iter()
            .map(|model_name| MeshCatalogEntry {
                descriptor: by_name.get(&model_name).cloned(),
                model_name,
            })
            .collect()
    }

    /// Get all models currently reachable via the mesh HTTP/API ingress.
    ///
    /// This is intentionally stricter than "loaded in VRAM somewhere": split
    /// workers may contribute compute for a model but cannot accept chat
    /// requests directly.
    pub async fn models_being_served(&self) -> Vec<String> {
        let my_hosted_models = self.hosted_models.lock().await.clone();
        let peer_data: Vec<_> = {
            let state = self.state.lock().await;
            state.peers.values().cloned().collect()
        };
        let mut served = std::collections::HashSet::new();
        for s in &my_hosted_models {
            served.insert(s.clone());
        }
        for peer in &peer_data {
            for m in peer.http_routable_models() {
                served.insert(m.clone());
            }
        }
        let mut result: Vec<String> = served.into_iter().collect();
        result.sort();
        result
    }

    /// Find a host for a specific model, using hash-based selection for load distribution.
    /// When multiple hosts serve the same model, picks one based on our node ID hash.
    /// All host IDs serving a model, with hash-preferred host first.
    /// Used for retry: if the first host fails, try the next.
    pub async fn hosts_for_model(&self, model: &str) -> Vec<EndpointId> {
        let state = self.state.lock().await;
        let mut hosts: Vec<EndpointId> = state
            .peers
            .values()
            .filter(|p| p.routes_http_model(model))
            .map(|p| p.id)
            .collect();
        hosts.sort();
        // Put the hash-preferred host first so normal path tries it first
        if !hosts.is_empty() {
            let my_id = self.endpoint.id();
            let id_bytes = my_id.as_bytes();
            let hash = id_bytes
                .iter()
                .fold(0u64, |acc, &b| acc.wrapping_mul(31).wrapping_add(b as u64));
            let idx = (hash as usize) % hosts.len();
            hosts.rotate_left(idx);
        }
        hosts
    }

    /// Find ANY host in the mesh (fallback when no model match).
    pub async fn any_host(&self) -> Option<PeerInfo> {
        let state = self.state.lock().await;
        state
            .peers
            .values()
            .find(|p| !p.http_routable_models().is_empty())
            .cloned()
    }

    /// Build the current routing table from this node's view of the mesh.
    pub async fn routing_table(&self) -> RoutingTable {
        let my_hosted_models = self.hosted_models.lock().await.clone();
        let my_role = self.role.lock().await.clone();
        let peer_data: Vec<_> = {
            let state = self.state.lock().await;
            state.peers.values().cloned().collect()
        };
        let mut hosts = Vec::new();

        // Include self if we're serving through the local API proxy
        if !matches!(my_role, NodeRole::Client) {
            for model in my_hosted_models {
                hosts.push(RouteEntry {
                    model,
                    node_id: format!("{}", self.endpoint.id().fmt_short()),
                    endpoint_id: self.endpoint.id(),
                    vram_gb: self.vram_bytes as f64 / 1e9,
                });
            }
        }

        // Include peers that are serving through their local API proxies
        for peer in &peer_data {
            for model in peer.http_routable_models() {
                hosts.push(RouteEntry {
                    model,
                    node_id: format!("{}", peer.id.fmt_short()),
                    endpoint_id: peer.id,
                    vram_gb: peer.vram_bytes as f64 / 1e9,
                });
            }
        }

        let mesh_id = self.mesh_id.lock().await.clone();
        RoutingTable { hosts, mesh_id }
    }

    pub fn vram_bytes(&self) -> u64 {
        self.vram_bytes
    }

    pub async fn peers(&self) -> Vec<PeerInfo> {
        self.state.lock().await.peers.values().cloned().collect()
    }

    async fn connection_to_peer(&self, peer_id: EndpointId) -> Result<Connection> {
        let state = self.state.lock().await;
        match state.connections.get(&peer_id).cloned() {
            Some(conn) => Ok(conn),
            None => {
                let addr = state.peers.get(&peer_id).map(|p| p.addr.clone());
                drop(state);
                let Some(addr) = addr else {
                    anyhow::bail!("No connection or address for {}", peer_id.fmt_short());
                };
                let conn = tokio::time::timeout(
                    std::time::Duration::from_secs(10),
                    connect_mesh(&self.endpoint, addr),
                )
                .await
                .map_err(|_| anyhow::anyhow!("Timeout connecting to {}", peer_id.fmt_short()))?
                .map_err(|e| {
                    anyhow::anyhow!("Failed to connect to {}: {e}", peer_id.fmt_short())
                })?;
                self.state
                    .lock()
                    .await
                    .connections
                    .insert(peer_id, conn.clone());
                let node_for_dispatch = self.clone();
                let conn_for_dispatch = conn.clone();
                tokio::spawn(async move {
                    node_for_dispatch
                        .dispatch_streams(conn_for_dispatch, peer_id)
                        .await;
                });
                if let Err(error) = self
                    .initiate_gossip_inner(conn.clone(), peer_id, false)
                    .await
                {
                    self.state.lock().await.connections.remove(&peer_id);
                    anyhow::bail!(
                        "Failed to complete gossip with {} before opening mesh stream: {error}",
                        peer_id.fmt_short()
                    );
                }
                Ok(conn)
            }
        }
    }

    async fn open_mesh_subprotocol_stream(
        &self,
        peer_id: EndpointId,
        name: &str,
        major: u32,
    ) -> Result<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)> {
        use prost::Message as _;

        let conn = self.connection_to_peer(peer_id).await?;
        let (mut send, recv) = conn.open_bi().await?;
        send.write_all(&[STREAM_SUBPROTOCOL]).await?;
        let open = crate::proto::node::MeshSubprotocolOpen {
            gen: NODE_PROTOCOL_GENERATION,
            name: name.to_string(),
            major,
        };
        open.validate_frame()
            .map_err(|error| anyhow::anyhow!("invalid mesh subprotocol open: {error}"))?;
        write_len_prefixed(&mut send, &open.encode_to_vec()).await?;
        Ok((send, recv))
    }

    async fn open_skippy_stage_mesh_stream(
        &self,
        peer_id: EndpointId,
        stream_kind: u8,
    ) -> Result<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)> {
        let (mut send, recv) = self
            .open_mesh_subprotocol_stream(
                peer_id,
                skippy_protocol::STAGE_SUBPROTOCOL_NAME,
                skippy_protocol::STAGE_SUBPROTOCOL_MAJOR,
            )
            .await?;
        send.write_all(&[stream_kind]).await?;
        Ok((send, recv))
    }

    async fn stage_connection_to_peer(&self, peer_id: EndpointId) -> Result<Connection> {
        let addr = {
            let state = self.state.lock().await;
            state.peers.get(&peer_id).map(|p| p.addr.clone())
        };
        let Some(addr) = addr else {
            anyhow::bail!("No address for stage peer {}", peer_id.fmt_short());
        };
        let conn = tokio::time::timeout(std::time::Duration::from_secs(10), async {
            self.endpoint
                .connect(addr, skippy_protocol::STAGE_ALPN_V1)
                .await
        })
        .await
        .map_err(|_| anyhow::anyhow!("Timeout connecting to stage peer {}", peer_id.fmt_short()))?
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to connect to stage peer {}: {e}",
                peer_id.fmt_short()
            )
        })?;
        Ok(conn)
    }

    /// Open an HTTP tunnel bi-stream to a peer (tagged STREAM_TUNNEL_HTTP).
    /// If no connection exists, tries to connect on-demand (for passive nodes
    /// that learned about hosts from routing table but aren't directly connected).
    pub async fn open_http_tunnel(
        &self,
        peer_id: EndpointId,
    ) -> Result<(iroh::endpoint::SendStream, iroh::endpoint::RecvStream)> {
        let conn = self.connection_to_peer(peer_id).await?;
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            let (mut send, recv) = conn.open_bi().await?;
            send.write_all(&[STREAM_TUNNEL_HTTP]).await?;
            Ok::<_, anyhow::Error>((send, recv))
        })
        .await
        .map_err(|_| anyhow::anyhow!("Timeout opening tunnel to {}", peer_id.fmt_short()))?;

        if result.is_err() {
            // Connection failed — peer is likely dead, broadcast it
            tracing::info!(
                "Tunnel to {} failed, broadcasting death",
                peer_id.fmt_short()
            );
            self.handle_peer_death(peer_id).await;
        }

        result
    }

    // --- Connection handling ---

    async fn accept_loop(&self) {
        // Wait until start_accepting() is called before processing any connections.
        // Check flag first to handle the case where start_accepting() was called before we got here.
        if !self.accepting.1.load(std::sync::atomic::Ordering::Acquire) {
            self.accepting.0.notified().await;
        }
        tracing::info!("Accept loop: now accepting inbound connections");

        loop {
            let incoming = match self.endpoint.accept().await {
                Some(i) => i,
                None => break,
            };
            let node = self.clone();
            tokio::spawn(async move {
                if let Err(e) = node.handle_incoming(incoming).await {
                    tracing::warn!("Incoming connection error: {e}");
                }
            });
        }
    }

    async fn control_accept_loop(
        &self,
        endpoint: Endpoint,
        shutdown_requested: Arc<std::sync::atomic::AtomicBool>,
        shutdown: Arc<tokio::sync::Notify>,
    ) {
        loop {
            if shutdown_requested.load(std::sync::atomic::Ordering::Acquire) {
                break;
            }
            tokio::select! {
                _ = shutdown.notified() => break,
                incoming = endpoint.accept() => {
                    let Some(incoming) = incoming else {
                        break;
                    };
                    let node = self.clone();
                    tokio::spawn(Box::pin(async move {
                        if let Err(error) = node.handle_control_incoming(incoming).await {
                            tracing::debug!("Control-plane incoming connection error: {error}");
                        }
                    }));
                }
            }
        }
    }

    async fn remember_incoming_connection(&self, remote: EndpointId, conn: &Connection) -> bool {
        let mut state = self.state.lock().await;
        let was_dead = state.dead_peers.remove(&remote).is_some();
        if was_dead {
            emit_mesh_info(format!(
                "🔄 Previously dead peer {} reconnected",
                remote.fmt_short()
            ));
        }
        state.connections.insert(remote, conn.clone());
        was_dead
    }

    fn spawn_reconnect_gossip(&self, conn: Connection, remote: EndpointId) {
        let node = self.clone();
        tokio::spawn(async move {
            if let Err(e) = node.initiate_gossip_inner(conn, remote, false).await {
                tracing::debug!("Reconnect gossip with {} failed: {e}", remote.fmt_short());
            }
        });
    }

    async fn handle_incoming(&self, incoming: iroh::endpoint::Incoming) -> Result<()> {
        let mut accepting = incoming.accept()?;
        let alpn = accepting.alpn().await?;
        let conn = accepting.await?;
        let remote = conn.remote_id();
        if self.handle_stage_alpn(&alpn, conn.clone(), remote).await {
            return Ok(());
        }
        tracing::info!("Inbound connection from {}", remote.fmt_short());

        // Store connection for stream dispatch (tunneling, route requests, etc.)
        // Don't add to peer list yet — only gossip exchange promotes to peer.
        let was_dead = self.remember_incoming_connection(remote, &conn).await;

        // If this peer was previously dead, immediately gossip to restore their
        // assigned/routable state in our peer list. Without this, models served by the
        // reconnecting peer stay invisible until the next heartbeat (up to 60s).
        if was_dead {
            self.spawn_reconnect_gossip(conn.clone(), remote);
        }

        self.dispatch_streams(conn, remote).await;
        Ok(())
    }

    async fn handle_stage_alpn(&self, alpn: &[u8], conn: Connection, remote: EndpointId) -> bool {
        if alpn != skippy_protocol::STAGE_ALPN_V1 {
            return false;
        }
        tracing::info!(
            "Inbound skippy stage connection from {}",
            remote.fmt_short()
        );
        self.dispatch_stage_streams(conn, remote).await;
        true
    }

    async fn handle_control_incoming(&self, incoming: iroh::endpoint::Incoming) -> Result<()> {
        let mut accepting = incoming.accept()?;
        let alpn = accepting.alpn().await?;
        anyhow::ensure!(
            alpn.as_slice() == ALPN_CONTROL_V1,
            "unexpected control-plane ALPN {:?}",
            String::from_utf8_lossy(&alpn)
        );
        let conn = accepting.await?;
        let remote = conn.remote_id();
        loop {
            let (mut send, mut recv) = match conn.accept_bi().await {
                Ok(streams) => streams,
                Err(error) => {
                    tracing::debug!(
                        "Control-plane connection from {} closed: {error}",
                        remote.fmt_short()
                    );
                    break;
                }
            };
            let node = self.clone();
            tokio::spawn(Box::pin(async move {
                if let Err(error) = node
                    .handle_control_stream(remote, &mut send, &mut recv)
                    .await
                {
                    tracing::debug!(
                        "Control-plane stream from {} failed: {error}",
                        remote.fmt_short()
                    );
                }
            }));
        }
        Ok(())
    }

    async fn read_owner_control_handshake(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Result<Option<crate::proto::node::OwnerControlHandshake>> {
        let handshake_bytes = match read_len_prefixed(recv).await {
            Ok(bytes) => bytes,
            Err(error) => {
                tracing::debug!(
                    "control handshake read failed from {}: {error}",
                    remote.fmt_short()
                );
                return Ok(None);
            }
        };

        let handshake_envelope =
            match crate::proto::node::OwnerControlEnvelope::decode(handshake_bytes.as_slice()) {
                Ok(envelope) => envelope,
                Err(error) => {
                    let code =
                        if serde_json::from_slice::<serde_json::Value>(&handshake_bytes).is_ok() {
                            crate::proto::node::OwnerControlErrorCode::LegacyJsonUnsupported
                        } else {
                            crate::proto::node::OwnerControlErrorCode::InvalidHandshake
                        };
                    let _ = self
                        .send_owner_control_terminal_envelope(
                            send,
                            owner_control_error_envelope(code, None, None, error.to_string()),
                        )
                        .await;
                    return Ok(None);
                }
            };
        if let Err(error) = handshake_envelope.validate_frame() {
            let _ = self
                .send_owner_control_terminal_envelope(
                    send,
                    owner_control_error_envelope(
                        crate::proto::node::OwnerControlErrorCode::InvalidHandshake,
                        None,
                        None,
                        error.to_string(),
                    ),
                )
                .await;
            return Ok(None);
        }
        let Some(handshake) = handshake_envelope.handshake else {
            let _ = self
                .send_owner_control_terminal_envelope(
                    send,
                    owner_control_error_envelope(
                        crate::proto::node::OwnerControlErrorCode::InvalidHandshake,
                        None,
                        None,
                        "first owner-control envelope must be a handshake",
                    ),
                )
                .await;
            return Ok(None);
        };
        Ok(Some(handshake))
    }

    async fn read_owner_control_request(
        &self,
        send: &mut iroh::endpoint::SendStream,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Result<Option<crate::proto::node::OwnerControlRequest>> {
        let request_bytes = match read_len_prefixed(recv).await {
            Ok(bytes) => bytes,
            Err(_) => return Ok(None),
        };
        let envelope =
            match crate::proto::node::OwnerControlEnvelope::decode(request_bytes.as_slice()) {
                Ok(envelope) => envelope,
                Err(error) => {
                    let code =
                        if serde_json::from_slice::<serde_json::Value>(&request_bytes).is_ok() {
                            crate::proto::node::OwnerControlErrorCode::LegacyJsonUnsupported
                        } else {
                            crate::proto::node::OwnerControlErrorCode::BadRequest
                        };
                    let _ = self
                        .send_owner_control_terminal_envelope(
                            send,
                            owner_control_error_envelope(code, None, None, error.to_string()),
                        )
                        .await;
                    return Ok(None);
                }
            };
        if let Err(error) = envelope.validate_frame() {
            let request_id = envelope.request.as_ref().map(|request| request.request_id);
            let _ = self
                .send_owner_control_terminal_envelope(
                    send,
                    owner_control_rejection_envelope(&request_bytes, request_id, &error),
                )
                .await;
            return Ok(None);
        }
        let Some(request) = envelope.request else {
            let _ = self
                .send_owner_control_terminal_envelope(
                    send,
                    owner_control_error_envelope(
                        crate::proto::node::OwnerControlErrorCode::BadRequest,
                        None,
                        None,
                        "owner-control envelope must contain a request after handshake",
                    ),
                )
                .await;
            return Ok(None);
        };
        Ok(Some(request))
    }

    async fn handle_control_stream(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Result<()> {
        let Some(handshake) = self
            .read_owner_control_handshake(remote, send, recv)
            .await?
        else {
            return Ok(());
        };

        let local_owner = self.owner_summary.lock().await.clone();
        let trust_store = self.trust_store.lock().await.clone();
        if let Err(error) = crate::crypto::verify_control_plane_peer_ownership(
            &local_owner,
            handshake.ownership.as_ref(),
            remote.as_bytes(),
            &trust_store,
            self.trust_policy,
            current_time_unix_ms(),
        ) {
            let _ = self
                .send_owner_control_terminal_envelope(
                    send,
                    self.owner_control_auth_error_envelope(&error),
                )
                .await;
            return Ok(());
        }

        loop {
            let Some(request) = self.read_owner_control_request(send, recv).await? else {
                break;
            };
            let watch_request = request.watch_config.is_some();
            self.handle_owner_control_request(remote, send, recv, request)
                .await?;
            if watch_request {
                break;
            }
        }
        Ok(())
    }

    async fn stage_stream_admitted(&self, remote: EndpointId) -> bool {
        let state = self.state.lock().await;
        state.peers.contains_key(&remote)
    }

    async fn dispatch_stage_stream_kind(
        &self,
        remote: EndpointId,
        stream_type: u8,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) {
        match stream_type {
            skippy_protocol::STAGE_STREAM_CONTROL => {
                let node = self.clone();
                tokio::spawn(async move {
                    if let Err(e) = node.handle_stage_control(remote, send, recv).await {
                        tracing::warn!("stage control error from {}: {e}", remote.fmt_short());
                    }
                });
            }
            skippy_protocol::STAGE_STREAM_TRANSPORT => {
                if self
                    .stage_transport_tx
                    .send((remote, send, recv))
                    .await
                    .is_err()
                {
                    tracing::warn!("Stage transport channel closed, dropping stream");
                }
            }
            skippy_protocol::STAGE_STREAM_ARTIFACT_TRANSFER => {
                let node = self.clone();
                tokio::spawn(async move {
                    if let Err(e) = node
                        .handle_artifact_transfer_stream(remote, send, recv)
                        .await
                    {
                        tracing::debug!(
                            "legacy artifact transfer stream error from {}: {e}",
                            remote.fmt_short()
                        );
                    }
                });
            }
            other => {
                tracing::warn!(
                    "Unknown skippy stage stream type {other:#04x} from {}",
                    remote.fmt_short()
                );
            }
        }
    }

    async fn dispatch_stage_streams(&self, conn: Connection, remote: EndpointId) {
        loop {
            match self.accept_stage_stream(&conn, remote).await {
                StageStreamAccept::Dispatch((send, recv), stream_type) => {
                    self.dispatch_stage_stream_kind(remote, stream_type, send, recv)
                        .await;
                }
                StageStreamAccept::Continue => continue,
                StageStreamAccept::Closed => break,
            }
        }
    }

    async fn accept_stage_stream(
        &self,
        conn: &Connection,
        remote: EndpointId,
    ) -> StageStreamAccept {
        let (send, mut recv) = match conn.accept_bi().await {
            Ok(streams) => streams,
            Err(e) => {
                tracing::info!(
                    "Skippy stage connection to {} closed: {e}",
                    remote.fmt_short()
                );
                return StageStreamAccept::Closed;
            }
        };
        if !self.stage_stream_admitted(remote).await {
            tracing::warn!(
                "Quarantine: skippy stage stream from unadmitted peer {} rejected",
                remote.fmt_short()
            );
            drop((send, recv));
            return StageStreamAccept::Continue;
        }
        let mut type_buf = [0u8; 1];
        if recv.read_exact(&mut type_buf).await.is_err() {
            return StageStreamAccept::Continue;
        }
        StageStreamAccept::Dispatch((send, recv), type_buf[0])
    }

    async fn accept_mesh_stream(
        &self,
        conn: &Connection,
        remote: EndpointId,
    ) -> Result<AcceptedMeshStream, ()> {
        let (send, mut recv) = conn.accept_bi().await.map_err(|error| {
            tracing::info!("Connection to {} closed: {error}", remote.fmt_short());
        })?;
        let mut type_buf = [0u8; 1];
        if recv.read_exact(&mut type_buf).await.is_err() {
            return Err(());
        }
        Ok(AcceptedMeshStream {
            send,
            recv,
            stream_type: type_buf[0],
        })
    }

    async fn admitted_mesh_stream(
        &self,
        remote: EndpointId,
        stream_type: u8,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) -> Option<MeshBiStream> {
        if stream_allowed_before_admission(stream_type) {
            return Some((send, recv));
        }
        let admitted = {
            let state = self.state.lock().await;
            state.peers.contains_key(&remote)
        };
        if admitted {
            Some((send, recv))
        } else {
            tracing::warn!(
                "Quarantine: stream {:#04x} from unadmitted peer {} rejected — peer must complete gossip first",
                stream_type,
                remote.fmt_short()
            );
            drop((send, recv));
            None
        }
    }

    async fn recover_closed_connection(&self, remote: EndpointId, closing_stable_id: usize) {
        match self
            .remove_closed_connection(remote, closing_stable_id)
            .await
        {
            ClosedConnectionRecovery::Reconnect(addr) => {
                self.reconnect_closed_connection_or_remove(remote, addr)
                    .await;
            }
            ClosedConnectionRecovery::RemovePeer => {
                self.remove_peer(remote).await;
            }
            ClosedConnectionRecovery::AlreadyReplaced => {}
        }
    }

    async fn reconnect_closed_connection_or_remove(&self, remote: EndpointId, addr: EndpointAddr) {
        tracing::info!("Attempting reconnect to {}...", remote.fmt_short());
        if let Some(new_conn) = self.reconnect_closed_peer(remote, addr).await {
            self.complete_recovered_connection(remote, new_conn).await;
        } else {
            tracing::info!("Reconnect to {} failed — removing peer", remote.fmt_short());
            self.remove_peer(remote).await;
        }
    }

    async fn remove_closed_connection(
        &self,
        remote: EndpointId,
        closing_stable_id: usize,
    ) -> ClosedConnectionRecovery {
        let mut state = self.state.lock().await;
        if !heartbeat::should_remove_connection(
            state.connections.get(&remote).map(|conn| conn.stable_id()),
            closing_stable_id,
        ) {
            tracing::debug!(
                "Connection dispatcher for {} closed after the tracked connection was replaced",
                remote.fmt_short()
            );
            return ClosedConnectionRecovery::AlreadyReplaced;
        }
        state.connections.remove(&remote);
        match state.peers.get(&remote).map(|peer| peer.addr.clone()) {
            Some(addr) => ClosedConnectionRecovery::Reconnect(addr),
            None => ClosedConnectionRecovery::RemovePeer,
        }
    }

    async fn reconnect_closed_peer(
        &self,
        remote: EndpointId,
        addr: EndpointAddr,
    ) -> Option<Connection> {
        match tokio::time::timeout(
            std::time::Duration::from_secs(10),
            connect_mesh(&self.endpoint, addr),
        )
        .await
        {
            Ok(Ok(new_conn)) => {
                tracing::info!("Reconnected to {}", remote.fmt_short());
                Some(new_conn)
            }
            _ => None,
        }
    }

    async fn complete_recovered_connection(&self, remote: EndpointId, new_conn: Connection) {
        {
            let mut state = self.state.lock().await;
            state.connections.insert(remote, new_conn.clone());
        }
        if self
            .recovered_connection_gossip_ok(remote, new_conn.clone())
            .await
        {
            let node = self.clone();
            tokio::spawn(async move {
                node.dispatch_streams(new_conn, remote).await;
            });
        } else {
            tracing::info!(
                "Reconnect gossip to {} failed — peer is dead, removing",
                remote.fmt_short()
            );
            self.remove_peer(remote).await;
        }
    }

    async fn recovered_connection_gossip_ok(
        &self,
        remote: EndpointId,
        new_conn: Connection,
    ) -> bool {
        tokio::time::timeout(
            std::time::Duration::from_secs(10),
            self.initiate_gossip(new_conn, remote),
        )
        .await
        .map(|result| result.is_ok())
        .unwrap_or(false)
    }

    /// Dispatch bi-streams on a connection by type byte
    fn dispatch_streams(
        &self,
        conn: Connection,
        remote: EndpointId,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + '_>> {
        Box::pin(self._dispatch_streams(conn, remote))
    }

    fn spawn_gossip_stream(
        &self,
        remote: EndpointId,
        protocol: ControlProtocol,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            if let Err(error) = node
                .handle_gossip_stream(remote, protocol, send, recv)
                .await
            {
                tracing::warn!("Gossip stream error from {}: {error}", remote.fmt_short());
            }
        });
    }

    fn spawn_tunnel_map_stream(
        &self,
        remote: EndpointId,
        protocol: ControlProtocol,
        recv: iroh::endpoint::RecvStream,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            if let Err(error) = node.handle_tunnel_map_stream(remote, protocol, recv).await {
                tracing::warn!(
                    "Tunnel map stream error from {}: {error}",
                    remote.fmt_short()
                );
            }
        });
    }

    fn spawn_route_request_stream(
        &self,
        protocol: ControlProtocol,
        send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            if protocol == ControlProtocol::ProtoV1 {
                let proto_buf = match read_len_prefixed(&mut recv).await {
                    Ok(buf) => buf,
                    Err(error) => {
                        tracing::warn!(
                            "Route request: failed to read proto body — rejecting: {error}"
                        );
                        return;
                    }
                };
                let req = match crate::proto::node::RouteTableRequest::decode(proto_buf.as_slice())
                {
                    Ok(request) => request,
                    Err(error) => {
                        tracing::warn!("Route request: invalid protobuf — rejecting: {error}");
                        return;
                    }
                };
                if let Err(error) = req.validate_frame() {
                    tracing::warn!("Route request: frame validation failed — rejecting: {error}");
                    return;
                }
            }
            use prost::Message as _;
            let mut send = send;
            let table = node.routing_table().await;
            let proto_table = routing_table_to_proto(&table);
            let _ = write_len_prefixed(&mut send, &proto_table.encode_to_vec()).await;
            let _ = send.finish();
        });
    }

    fn spawn_plugin_channel_stream(
        &self,
        remote: EndpointId,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            if let Err(error) = node.handle_plugin_channel_stream(remote, send, recv).await {
                tracing::debug!(
                    "Plugin channel stream error from {}: {error}",
                    remote.fmt_short()
                );
            }
        });
    }

    fn spawn_plugin_bulk_stream(
        &self,
        remote: EndpointId,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            if let Err(error) = node.handle_plugin_bulk_stream(remote, send, recv).await {
                tracing::debug!(
                    "Plugin bulk stream error from {}: {error}",
                    remote.fmt_short()
                );
            }
        });
    }

    fn spawn_subprotocol_stream(
        &self,
        remote: EndpointId,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) {
        let node = self.clone();
        tokio::spawn(async move {
            if let Err(error) = node
                .handle_mesh_subprotocol_stream(remote, send, recv)
                .await
            {
                tracing::debug!(
                    "subprotocol stream error from {}: {error}",
                    remote.fmt_short()
                );
            }
        });
    }

    fn spawn_peer_down_stream(&self, remote: EndpointId, recv: iroh::endpoint::RecvStream) {
        let node = self.clone();
        tokio::spawn(async move {
            node.handle_peer_down_stream(remote, recv).await;
        });
    }

    async fn handle_peer_down_stream(
        &self,
        remote: EndpointId,
        mut recv: iroh::endpoint::RecvStream,
    ) {
        let Some(dead_id) = self.decode_peer_down_frame(&mut recv).await else {
            return;
        };
        let report = self.peer_down_report(remote, dead_id).await;
        self.apply_peer_down_report(remote, dead_id, report).await;
    }

    async fn decode_peer_down_frame(
        &self,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Option<EndpointId> {
        let frame = self.read_peer_down_frame(recv).await?;
        peer_down_endpoint_id(&frame)
    }

    async fn read_peer_down_frame(
        &self,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Option<crate::proto::node::PeerDown> {
        let proto_buf = match read_len_prefixed(recv).await {
            Ok(buf) => buf,
            Err(e) => {
                tracing::warn!("PeerDown: failed to read proto body — rejecting: {e}");
                return None;
            }
        };
        self.decode_peer_down_proto(&proto_buf)
    }

    fn decode_peer_down_proto(&self, proto_buf: &[u8]) -> Option<crate::proto::node::PeerDown> {
        let frame = match crate::proto::node::PeerDown::decode(proto_buf) {
            Ok(f) => f,
            Err(e) => {
                tracing::warn!("PeerDown: invalid protobuf — rejecting: {e}");
                return None;
            }
        };
        if let Err(e) = frame.validate_frame() {
            tracing::warn!("PeerDown: frame validation failed — rejecting: {e}");
            return None;
        }
        Some(frame)
    }

    async fn peer_down_report(&self, remote: EndpointId, dead_id: EndpointId) -> PeerDownReport {
        let state = self.state.lock().await;
        let conn_opt = state.connections.get(&dead_id).cloned();
        let peer = state.peers.get(&dead_id);
        let peer_addr = peer.map(|p| p.addr.clone());
        let recently_seen = peer
            .map(|p| p.last_seen.elapsed().as_secs() < PEER_STALE_SECS)
            .unwrap_or(false);
        let reporter_cooled = state
            .peer_down_rejections
            .get(&(remote, dead_id))
            .is_some_and(|t| t.elapsed().as_secs() < PEER_DOWN_REPORTER_COOLDOWN_SECS);
        PeerDownReport {
            conn_opt,
            peer_addr,
            recently_seen,
            reporter_cooled,
        }
    }

    async fn apply_peer_down_report(
        &self,
        remote: EndpointId,
        dead_id: EndpointId,
        report: PeerDownReport,
    ) {
        match peer_down_report_disposition(report.reporter_cooled, report.recently_seen) {
            PeerDownReportDisposition::SuppressReporterCooldown => tracing::debug!(
                "PeerDown: {} reported {} dead but reporter is in cooldown, ignoring",
                remote.fmt_short(),
                dead_id.fmt_short()
            ),
            PeerDownReportDisposition::RejectRecentlySeen => {
                self.reject_recent_peer_down_report(remote, dead_id).await;
            }
            PeerDownReportDisposition::ProbeReachability => {
                self.probe_and_apply_peer_down(remote, dead_id, report)
                    .await;
            }
        }
    }

    async fn reject_recent_peer_down_report(&self, remote: EndpointId, dead_id: EndpointId) {
        emit_mesh_info(format!(
            "ℹ️  Peer {} reported dead by {} but seen recently (direct alive), ignoring",
            dead_id.fmt_short(),
            remote.fmt_short()
        ));
        self.record_peer_down_rejection(remote, dead_id).await;
    }

    async fn probe_and_apply_peer_down(
        &self,
        remote: EndpointId,
        dead_id: EndpointId,
        report: PeerDownReport,
    ) {
        let should_remove = self
            .peer_down_probe_should_remove(dead_id, report.conn_opt, report.peer_addr)
            .await;
        if let Some(id) = resolve_peer_down(self.endpoint.id(), dead_id, should_remove) {
            self.remove_confirmed_peer_down(remote, id).await;
        } else if dead_id != self.endpoint.id() {
            emit_mesh_info(format!(
                "ℹ️  Peer {} reported dead by {} but still reachable, ignoring",
                dead_id.fmt_short(),
                remote.fmt_short()
            ));
            self.record_peer_down_rejection(remote, dead_id).await;
        }
    }

    async fn peer_down_probe_should_remove(
        &self,
        dead_id: EndpointId,
        conn_opt: Option<Connection>,
        peer_addr: Option<EndpointAddr>,
    ) -> bool {
        if let Some(conn) = conn_opt {
            return !matches!(
                tokio::time::timeout(std::time::Duration::from_secs(5), conn.open_bi()).await,
                Ok(Ok(_))
            );
        }
        let Some(addr) = peer_addr else {
            return true;
        };
        match tokio::time::timeout(
            std::time::Duration::from_secs(8),
            connect_mesh(&self.endpoint, addr),
        )
        .await
        {
            Ok(Ok(new_conn)) => {
                self.keep_reachable_peer_down_connection(dead_id, new_conn)
                    .await;
                false
            }
            _ => true,
        }
    }

    async fn keep_reachable_peer_down_connection(&self, dead_id: EndpointId, new_conn: Connection) {
        emit_mesh_info(format!(
            "ℹ️  Peer {} reported dead but we reached them, keeping",
            dead_id.fmt_short()
        ));
        let mut state = self.state.lock().await;
        if state.connections.contains_key(&dead_id) {
            return;
        }
        state.connections.insert(dead_id, new_conn.clone());
        drop(state);
        let node = self.clone();
        tokio::spawn(async move {
            node.dispatch_streams(new_conn, dead_id).await;
        });
    }

    async fn remove_confirmed_peer_down(&self, remote: EndpointId, id: EndpointId) {
        emit_mesh_warning(format!(
            "⚠️  Peer {} reported dead by {}, confirmed, removing",
            id.fmt_short(),
            remote.fmt_short()
        ));
        let mut state = self.state.lock().await;
        state.dead_peers.insert(id, std::time::Instant::now());
        state.connections.remove(&id);
        drop(state);
        self.remove_peer(id).await;
    }

    async fn record_peer_down_rejection(&self, remote: EndpointId, dead_id: EndpointId) {
        self.state
            .lock()
            .await
            .peer_down_rejections
            .insert((remote, dead_id), std::time::Instant::now());
    }

    async fn dispatch_mesh_stream(
        &self,
        remote: EndpointId,
        protocol: ControlProtocol,
        stream_type: u8,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) -> bool {
        if stream_type == STREAM_TUNNEL {
            return self.forward_tunnel_stream(send, recv).await;
        }
        if stream_type == STREAM_TUNNEL_HTTP {
            return self.forward_tunnel_http_stream(send, recv).await;
        }

        self.spawn_non_tunnel_mesh_stream(remote, protocol, stream_type, send, recv);
        true
    }

    async fn forward_tunnel_stream(
        &self,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) -> bool {
        if self.tunnel_tx.send((send, recv)).await.is_err() {
            tracing::warn!("Tunnel receiver dropped");
            return false;
        }
        true
    }

    async fn forward_tunnel_http_stream(
        &self,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) -> bool {
        if self.tunnel_http_tx.send((send, recv)).await.is_err() {
            tracing::warn!("HTTP tunnel receiver dropped");
            return false;
        }
        true
    }

    fn spawn_non_tunnel_mesh_stream(
        &self,
        remote: EndpointId,
        protocol: ControlProtocol,
        stream_type: u8,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) {
        match stream_type {
            STREAM_GOSSIP => self.spawn_gossip_stream(remote, protocol, send, recv),
            STREAM_TUNNEL_MAP => self.spawn_tunnel_map_stream(remote, protocol, recv),
            STREAM_ROUTE_REQUEST => self.spawn_route_request_stream(protocol, send, recv),
            STREAM_PEER_DOWN => self.spawn_peer_down_stream(remote, recv),
            STREAM_PEER_LEAVING => self.spawn_peer_leaving_stream(remote, recv),
            STREAM_PLUGIN_CHANNEL => self.spawn_plugin_channel_stream(remote, send, recv),
            STREAM_PLUGIN_BULK_TRANSFER => self.spawn_plugin_bulk_stream(remote, send, recv),
            STREAM_SUBPROTOCOL => self.spawn_subprotocol_stream(remote, send, recv),
            other => tracing::warn!("Unknown stream type {other} from {}", remote.fmt_short()),
        }
    }

    fn spawn_peer_leaving_stream(&self, remote: EndpointId, recv: iroh::endpoint::RecvStream) {
        let node = self.clone();
        tokio::spawn(async move {
            node.handle_peer_leaving_stream(remote, recv).await;
        });
    }

    async fn handle_peer_leaving_stream(
        &self,
        remote: EndpointId,
        mut recv: iroh::endpoint::RecvStream,
    ) {
        let Some(leaving_id) = self.decode_peer_leaving(remote, &mut recv).await else {
            return;
        };
        emit_mesh_info(format!(
            "👋 Peer {} announced clean shutdown",
            leaving_id.fmt_short()
        ));
        let mut state = self.state.lock().await;
        state
            .dead_peers
            .insert(leaving_id, std::time::Instant::now());
        state.connections.remove(&leaving_id);
        drop(state);
        self.remove_peer(leaving_id).await;
    }

    async fn decode_peer_leaving(
        &self,
        remote: EndpointId,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Option<EndpointId> {
        let frame = self.read_peer_leaving_frame(recv).await?;
        self.resolve_peer_leaving_frame(remote, &frame)
    }

    async fn read_peer_leaving_frame(
        &self,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Option<crate::proto::node::PeerLeaving> {
        let proto_buf = match read_len_prefixed(recv).await {
            Ok(buf) => buf,
            Err(e) => {
                tracing::warn!("PeerLeaving: failed to read proto body — rejecting: {e}");
                return None;
            }
        };
        self.decode_peer_leaving_proto(&proto_buf)
    }

    fn decode_peer_leaving_proto(
        &self,
        proto_buf: &[u8],
    ) -> Option<crate::proto::node::PeerLeaving> {
        let frame = match crate::proto::node::PeerLeaving::decode(proto_buf) {
            Ok(f) => f,
            Err(e) => {
                tracing::warn!("PeerLeaving: invalid protobuf — rejecting: {e}");
                return None;
            }
        };
        if let Err(e) = frame.validate_frame() {
            tracing::warn!("PeerLeaving: frame validation failed — rejecting: {e}");
            return None;
        }
        Some(frame)
    }

    fn resolve_peer_leaving_frame(
        &self,
        remote: EndpointId,
        frame: &crate::proto::node::PeerLeaving,
    ) -> Option<EndpointId> {
        match resolve_peer_leaving(remote, frame) {
            Ok(id) => Some(id),
            Err(e) => {
                tracing::warn!("PeerLeaving from {}: rejected ({})", remote.fmt_short(), e);
                None
            }
        }
    }

    async fn _dispatch_streams(&self, conn: Connection, remote: EndpointId) {
        let protocol = connection_protocol(&conn);
        let dispatcher_stable_id = conn.stable_id();
        loop {
            let accepted = match self.accept_mesh_stream(&conn, remote).await {
                Ok(accepted) => accepted,
                Err(()) => {
                    self.recover_closed_connection(remote, dispatcher_stable_id)
                        .await;
                    break;
                }
            };
            let Some((send, recv)) = self
                .admitted_mesh_stream(remote, accepted.stream_type, accepted.send, accepted.recv)
                .await
            else {
                continue;
            };
            if !self
                .dispatch_mesh_stream(remote, protocol, accepted.stream_type, send, recv)
                .await
            {
                break;
            }
        }
    }

    async fn handle_mesh_subprotocol_stream(
        &self,
        remote: EndpointId,
        send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        use prost::Message as _;

        let buf = read_len_prefixed(&mut recv).await?;
        let open = crate::proto::node::MeshSubprotocolOpen::decode(buf.as_slice())
            .map_err(|error| anyhow::anyhow!("MeshSubprotocolOpen decode error: {error}"))?;
        open.validate_frame()
            .map_err(|error| anyhow::anyhow!("MeshSubprotocolOpen validation error: {error}"))?;
        match (open.name.as_str(), open.major) {
            (skippy_protocol::STAGE_SUBPROTOCOL_NAME, skippy_protocol::STAGE_SUBPROTOCOL_MAJOR) => {
                self.handle_skippy_stage_subprotocol_stream(remote, send, recv)
                    .await
            }
            _ => anyhow::bail!(
                "unsupported mesh subprotocol {}/{} from {}",
                open.name,
                open.major,
                remote.fmt_short()
            ),
        }
    }

    async fn handle_skippy_stage_subprotocol_stream(
        &self,
        remote: EndpointId,
        send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        let mut type_buf = [0u8; 1];
        recv.read_exact(&mut type_buf).await?;
        match type_buf[0] {
            skippy_protocol::STAGE_STREAM_CONTROL => {
                self.handle_stage_control(remote, send, recv).await
            }
            skippy_protocol::STAGE_STREAM_ARTIFACT_TRANSFER => {
                self.handle_artifact_transfer_stream(remote, send, recv)
                    .await
            }
            skippy_protocol::STAGE_STREAM_TRANSPORT => {
                anyhow::bail!("skippy activation transport stays on skippy-stage/1")
            }
            other => anyhow::bail!("unknown skippy stage subprotocol stream kind {other:#04x}"),
        }
    }

    async fn decode_stage_control_request(
        &self,
        remote: EndpointId,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> anyhow::Result<skippy_stage_proto::StageControlRequest> {
        let buf = read_len_prefixed(recv).await.map_err(|e| {
            tracing::warn!(
                "handle_stage_control: read_len_prefixed failed from {}: {e}",
                remote.fmt_short()
            );
            e
        })?;
        let frame = skippy_protocol::proto::stage::StageControlRequest::decode(buf.as_slice())
            .map_err(|e| {
                tracing::warn!(
                    "handle_stage_control: decode failed from {}: {e}",
                    remote.fmt_short()
                );
                anyhow::anyhow!("StageControlRequest decode error: {e}")
            })?;
        skippy_protocol::validate_stage_control_request(&frame).map_err(|e| {
            tracing::warn!(
                "handle_stage_control: validation failed from {}: {e}",
                remote.fmt_short()
            );
            anyhow::anyhow!("StageControlRequest validation error: {e}")
        })?;
        anyhow::ensure!(
            frame.requester_id.as_slice() == remote.as_bytes(),
            "stage control requester_id does not match QUIC peer identity"
        );
        Ok(frame)
    }

    fn stage_control_request_kind(frame: &skippy_stage_proto::StageControlRequest) -> &'static str {
        match &frame.command {
            Some(skippy_stage_proto::stage_control_request::Command::ClaimCoordinator(_)) => {
                "claim"
            }
            Some(skippy_stage_proto::stage_control_request::Command::LoadStage(_)) => "load",
            Some(skippy_stage_proto::stage_control_request::Command::StopStage(_)) => "stop",
            Some(skippy_stage_proto::stage_control_request::Command::PrepareStage(_)) => "prepare",
            _ => "other",
        }
    }

    async fn record_stage_control_response(
        &self,
        response: &crate::inference::skippy::StageControlResponse,
    ) {
        match response {
            crate::inference::skippy::StageControlResponse::Ready(ready) => {
                self.record_stage_status(Some(self.endpoint.id()), ready.status.clone())
                    .await;
            }
            crate::inference::skippy::StageControlResponse::Status(statuses) => {
                for status in statuses {
                    self.record_stage_status(Some(self.endpoint.id()), status.clone())
                        .await;
                }
            }
            _ => {}
        }
    }

    async fn execute_stage_control_request(
        &self,
        request: crate::inference::skippy::StageControlRequest,
    ) -> anyhow::Result<crate::inference::skippy::StageControlResponse> {
        let control_tx = self.stage_control_tx.lock().await.clone();
        match control_tx {
            Some(tx) => {
                let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
                tx.send(crate::inference::skippy::StageControlCommand {
                    request,
                    resp: resp_tx,
                })
                .map_err(|_| anyhow::anyhow!("stage control loop is unavailable"))?;
                resp_rx
                    .await
                    .map_err(|_| anyhow::anyhow!("stage control response dropped"))?
            }
            None => Ok(stage_control_unavailable_response(request)),
        }
    }

    async fn handle_stage_control(
        &self,
        remote: EndpointId,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> anyhow::Result<()> {
        use prost::Message as _;

        let frame = self.decode_stage_control_request(remote, &mut recv).await?;
        let request_kind = Self::stage_control_request_kind(&frame);
        tracing::debug!(
            "handle_stage_control: received {request_kind} from {}",
            remote.fmt_short()
        );

        let mut request = stage_control_request_from_proto(frame)?;
        self.prepare_stage_control_request(&mut request)
            .await
            .map_err(|e| {
                tracing::warn!(
                    "handle_stage_control: prepare failed for {request_kind} from {}: {e}",
                    remote.fmt_short()
                );
                e
            })?;
        if let crate::inference::skippy::StageControlRequest::Load(load) = &request {
            self.record_stage_topology(stage_topology_from_load(self.endpoint.id(), load))
                .await;
        }
        let response = self.execute_stage_control_request(request).await?;
        self.record_stage_control_response(&response).await;
        let status_list_supported = self
            .peer_supports_skippy_subprotocol_feature(
                remote,
                skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST,
            )
            .await;
        let proto_response = stage_control_response_to_proto(response, status_list_supported);
        write_len_prefixed(&mut send, &proto_response.encode_to_vec()).await?;
        let _ = send.finish();
        Ok(())
    }

    async fn prepare_stage_control_request(
        &self,
        request: &mut crate::inference::skippy::StageControlRequest,
    ) -> anyhow::Result<()> {
        match request {
            crate::inference::skippy::StageControlRequest::Claim(_) => {}
            crate::inference::skippy::StageControlRequest::Load(load) => {
                if load.load_mode == skippy_protocol::LoadMode::RuntimeSlice
                    && load
                        .model_path
                        .as_deref()
                        .is_none_or(|path| !std::path::Path::new(path).exists())
                {
                    for candidate in [
                        load.model_id.as_str(),
                        load.package_ref.strip_prefix("gguf://").unwrap_or_default(),
                    ]
                    .into_iter()
                    .filter(|candidate| !candidate.is_empty())
                    {
                        if let Ok(path) =
                            crate::models::resolve_model_spec(std::path::Path::new(candidate)).await
                        {
                            if path.exists() {
                                load.model_path = Some(path.to_string_lossy().to_string());
                                break;
                            }
                        }
                    }
                }
                let Some(downstream) = load.downstream.as_mut() else {
                    return Ok(());
                };
                let Some(downstream_node) = downstream.node_id else {
                    return Ok(());
                };
                if downstream_node == self.endpoint.id() {
                    return Ok(());
                }
                let bridge_addr = self
                    .ensure_stage_transport_bridge(
                        downstream_node,
                        load.topology_id.clone(),
                        load.run_id.clone(),
                        downstream.stage_id.clone(),
                    )
                    .await?;
                downstream.endpoint = bridge_addr;
            }
            crate::inference::skippy::StageControlRequest::Prepare(_) => {}
            crate::inference::skippy::StageControlRequest::Stop(stop) => {
                self.stop_stage_transport_bridge(&stop.topology_id, &stop.run_id, &stop.stage_id)
                    .await;
            }
            crate::inference::skippy::StageControlRequest::Status(_)
            | crate::inference::skippy::StageControlRequest::Inventory(_)
            | crate::inference::skippy::StageControlRequest::CancelPrepare(_)
            | crate::inference::skippy::StageControlRequest::StatusUpdate(_) => {}
        }
        Ok(())
    }

    async fn prefetch_stage_package_from_coordinator(
        &self,
        prepare: &crate::inference::skippy::StagePrepareRequest,
    ) -> Result<()> {
        let load = &prepare.load;
        if load.load_mode != skippy_protocol::LoadMode::LayerPackage {
            return Ok(());
        }
        if !crate::models::artifact_transfer::artifact_transfer_enabled() {
            return Ok(());
        }
        let Some(coordinator_id) = prepare.coordinator_id else {
            return Ok(());
        };
        if coordinator_id == self.endpoint.id() {
            return Ok(());
        }
        if !self
            .peer_supports_skippy_subprotocol_feature(
                coordinator_id,
                skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_ARTIFACT_TRANSFER,
            )
            .await
        {
            return Ok(());
        }
        self.fetch_stage_package_artifacts_from_peer(coordinator_id, load)
            .await
    }

    async fn peer_supports_skippy_subprotocol_feature(
        &self,
        peer_id: EndpointId,
        feature: &str,
    ) -> bool {
        let peer = {
            let state = self.state.lock().await;
            state.peers.get(&peer_id).cloned()
        };
        let Some(peer) = peer else {
            return false;
        };
        match feature {
            skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_CONTROL => {
                peer.stage_protocol_generation_supported
            }
            skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_ARTIFACT_TRANSFER => {
                self.artifact_transfer_allowed_for_peer(&peer).await
            }
            skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST => {
                peer.stage_status_list_supported
            }
            _ => false,
        }
    }

    async fn fetch_stage_package_artifacts_from_peer(
        &self,
        peer_id: EndpointId,
        load: &crate::inference::skippy::StageLoadRequest,
    ) -> Result<()> {
        let package_dir =
            crate::models::artifact_transfer::package_cache_dir_for_ref(&load.package_ref)?;
        let manifest_request = crate::models::artifact_transfer::manifest_artifact_request(
            &load.package_ref,
            &load.manifest_sha256,
        )?;
        let manifest_path =
            crate::models::artifact_transfer::local_artifact_path(&package_dir, &manifest_request);
        if !crate::models::artifact_transfer::local_artifact_satisfies(
            &package_dir,
            &manifest_request,
            true,
        )? {
            self.fetch_artifact_from_peer(peer_id, load, &manifest_request, &manifest_path)
                .await
                .context("fetch package manifest from peer")?;
        }

        let artifacts = crate::models::artifact_transfer::required_stage_package_artifacts(
            &package_dir,
            &load.package_ref,
            &load.manifest_sha256,
            crate::models::artifact_transfer::StageArtifactSelection {
                layer_start: load.layer_start,
                layer_end: load.layer_end,
                include_embeddings: load.layer_start == 0,
                include_output: load.downstream.is_none(),
                include_projectors: load.layer_start == 0,
            },
        )?;
        for artifact in artifacts {
            if crate::models::artifact_transfer::local_artifact_satisfies(
                &package_dir,
                &artifact,
                true,
            )? {
                continue;
            }
            let destination =
                crate::models::artifact_transfer::local_artifact_path(&package_dir, &artifact);
            self.fetch_artifact_from_peer(peer_id, load, &artifact, &destination)
                .await
                .with_context(|| {
                    format!(
                        "fetch package artifact {} from peer",
                        artifact.relative_path.display()
                    )
                })?;
        }
        Ok(())
    }

    async fn fetch_artifact_from_peer(
        &self,
        peer_id: EndpointId,
        load: &crate::inference::skippy::StageLoadRequest,
        artifact: &crate::models::artifact_transfer::PackageArtifactRequest,
        destination: &std::path::Path,
    ) -> Result<()> {
        use tokio::io::AsyncWriteExt;

        if let Some(parent) = destination.parent() {
            tokio::fs::create_dir_all(parent)
                .await
                .context("create package artifact directory")?;
        }
        crate::models::artifact_transfer::ensure_local_artifact_install_parent(
            &artifact.package_ref,
            destination,
        )?;
        let temp_path = partial_artifact_path(destination);
        let mut partial_guard = PartialArtifactGuard::new(temp_path.clone());
        let offset = 0;

        let frame = skippy_stage_proto::StageArtifactTransferRequest {
            gen: skippy_protocol::STAGE_PROTOCOL_GENERATION,
            requester_id: self.endpoint.id().as_bytes().to_vec(),
            topology_id: load.topology_id.clone(),
            run_id: load.run_id.clone(),
            stage_id: load.stage_id.clone(),
            package_ref: artifact.package_ref.clone(),
            manifest_sha256: artifact.manifest_sha256.clone(),
            relative_path: artifact.relative_path.to_string_lossy().to_string(),
            offset,
            expected_size: artifact.expected_size,
            expected_sha256: artifact.expected_sha256.clone(),
        };
        skippy_protocol::validate_stage_artifact_transfer_request(&frame)
            .map_err(|error| anyhow::anyhow!("invalid artifact transfer request: {error}"))?;

        let response = tokio::time::timeout(ARTIFACT_TRANSFER_OPEN_TIMEOUT, async {
            let (mut send, mut recv) = self
                .open_skippy_stage_mesh_stream(
                    peer_id,
                    skippy_protocol::STAGE_STREAM_ARTIFACT_TRANSFER,
                )
                .await?;
            write_len_prefixed(&mut send, &frame.encode_to_vec()).await?;
            let _ = send.finish();
            let response_buf = read_len_prefixed(&mut recv).await?;
            let response =
                skippy_stage_proto::StageArtifactTransferResponse::decode(response_buf.as_slice())
                    .map_err(|error| {
                        anyhow::anyhow!("StageArtifactTransferResponse decode error: {error}")
                    })?;
            skippy_protocol::validate_stage_artifact_transfer_response(&response).map_err(
                |error| anyhow::anyhow!("StageArtifactTransferResponse validation error: {error}"),
            )?;
            Ok::<_, anyhow::Error>((recv, response))
        })
        .await
        .map_err(|_| anyhow::anyhow!("timeout opening artifact transfer stream"))??;
        let (mut recv, response) = response;
        if !response.accepted {
            anyhow::bail!(
                "peer artifact transfer rejected: {}",
                response
                    .error
                    .unwrap_or_else(|| "artifact unavailable".to_string())
            );
        }
        if let Some(expected_size) = artifact.expected_size {
            anyhow::ensure!(
                response.total_size == expected_size,
                "peer artifact size mismatch"
            );
        } else if artifact.relative_path.as_path()
            == std::path::Path::new(crate::models::artifact_transfer::PACKAGE_MANIFEST_FILE)
        {
            anyhow::ensure!(
                response.total_size <= crate::models::artifact_transfer::MAX_PACKAGE_MANIFEST_BYTES,
                "peer package manifest exceeds transfer limit"
            );
        } else {
            anyhow::bail!("peer artifact response missing expected size");
        }
        if let Some(expected_sha) = artifact.expected_sha256.as_deref() {
            anyhow::ensure!(
                response
                    .sha256
                    .as_deref()
                    .is_some_and(|sha| sha.eq_ignore_ascii_case(expected_sha)),
                "peer artifact sha256 mismatch"
            );
        }
        anyhow::ensure!(
            offset <= response.total_size,
            "peer artifact response is smaller than resume offset"
        );

        let transfer_result = async {
            let mut file = tokio::fs::OpenOptions::new()
                .create_new(true)
                .write(true)
                .open(&temp_path)
                .await
                .context("open partial artifact")?;
            let mut remaining = response.total_size.saturating_sub(offset);
            let mut buffer = vec![0u8; ARTIFACT_TRANSFER_BUFFER_BYTES];
            while remaining > 0 {
                let limit = buffer.len().min(remaining as usize);
                let read = read_artifact_transfer_chunk(
                    &mut recv,
                    &mut buffer[..limit],
                    ARTIFACT_TRANSFER_READ_IDLE_TIMEOUT,
                )
                .await?;
                file.write_all(&buffer[..read])
                    .await
                    .context("write partial artifact")?;
                remaining -= read as u64;
            }
            file.flush().await.context("flush partial artifact")?;
            drop(file);

            let actual_size = tokio::fs::metadata(&temp_path)
                .await
                .context("stat partial artifact")?
                .len();
            anyhow::ensure!(
                actual_size == response.total_size,
                "partial artifact size mismatch after transfer"
            );
            let temp_for_hash = temp_path.clone();
            let actual_sha = tokio::task::spawn_blocking(move || {
                crate::models::artifact_transfer::file_sha256_hex(&temp_for_hash)
            })
            .await
            .context("join artifact sha256 task")??;
            let expected_sha = artifact
                .expected_sha256
                .as_deref()
                .or(response.sha256.as_deref())
                .context("peer artifact response missing sha256")?;
            anyhow::ensure!(
                actual_sha.eq_ignore_ascii_case(expected_sha),
                "transferred artifact sha256 mismatch"
            );
            if destination.exists() {
                let _ = tokio::fs::remove_file(destination).await;
            }
            tokio::fs::rename(&temp_path, destination)
                .await
                .context("install transferred artifact")?;
            Ok::<_, anyhow::Error>(())
        }
        .await;
        if let Err(error) = transfer_result {
            let _ = tokio::fs::remove_file(&temp_path).await;
            return Err(error);
        }
        partial_guard.disarm();
        Ok(())
    }

    async fn artifact_transfer_rejected(
        send: &mut iroh::endpoint::SendStream,
        total_size: u64,
        sha256: Option<&str>,
        error: &'static str,
    ) -> anyhow::Result<()> {
        write_artifact_transfer_response(send, false, total_size, sha256, Some(error)).await
    }

    async fn authorize_artifact_transfer_request(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request: &skippy_stage_proto::StageArtifactTransferRequest,
    ) -> anyhow::Result<Option<std::path::PathBuf>> {
        if !self
            .artifact_transfer_serving_allowed_for_remote(remote)
            .await
        {
            Self::artifact_transfer_rejected(send, 0, None, "artifact transfer disabled").await?;
            return Ok(None);
        }
        let Some(package_dir) = Self::artifact_transfer_package_dir(remote, send, request).await?
        else {
            return Ok(None);
        };
        let topologies = self
            .stage_topologies
            .lock()
            .await
            .topologies
            .values()
            .cloned()
            .collect::<Vec<_>>();
        if !Self::artifact_transfer_topology_allows(
            remote,
            send,
            request,
            &package_dir,
            &topologies,
        )
        .await?
        {
            return Ok(None);
        }
        Ok(Some(package_dir))
    }

    async fn artifact_transfer_package_dir(
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request: &skippy_stage_proto::StageArtifactTransferRequest,
    ) -> anyhow::Result<Option<std::path::PathBuf>> {
        match crate::models::artifact_transfer::package_cache_dir_for_ref(&request.package_ref) {
            Ok(path) => Ok(Some(path)),
            Err(error) => {
                tracing::debug!(
                    peer = %remote.fmt_short(),
                    "artifact transfer request has unsupported package ref: {error}"
                );
                Self::artifact_transfer_rejected(send, 0, None, "artifact unavailable").await?;
                Ok(None)
            }
        }
    }

    async fn artifact_transfer_topology_allows(
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request: &skippy_stage_proto::StageArtifactTransferRequest,
        package_dir: &std::path::Path,
        topologies: &[StageTopologyInstance],
    ) -> anyhow::Result<bool> {
        let allowed =
            match artifact_transfer_allowed_by_topology(topologies, remote, package_dir, request) {
                Ok(allowed) => allowed,
                Err(error) => {
                    tracing::debug!(
                        peer = %remote.fmt_short(),
                        path = %request.relative_path,
                        "artifact transfer authorization failed: {error}"
                    );
                    Self::artifact_transfer_rejected(send, 0, None, "artifact unavailable").await?;
                    return Ok(false);
                }
            };
        if !allowed {
            tracing::debug!(
                peer = %remote.fmt_short(),
                path = %request.relative_path,
                "artifact transfer request is not authorized for this stage assignment"
            );
            Self::artifact_transfer_rejected(send, 0, None, "artifact unavailable").await?;
        }
        Ok(allowed)
    }

    async fn resolve_artifact_transfer_request(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request: &skippy_stage_proto::StageArtifactTransferRequest,
    ) -> anyhow::Result<Option<crate::models::artifact_transfer::ServableArtifact>> {
        let artifact =
            match crate::models::artifact_transfer::servable_artifact_from_request(request) {
                Ok(artifact) => artifact,
                Err(error) => {
                    tracing::debug!(
                        peer = %remote.fmt_short(),
                        path = %request.relative_path,
                        "artifact transfer request cannot be served: {error}"
                    );
                    Self::artifact_transfer_rejected(send, 0, None, "artifact unavailable").await?;
                    return Ok(None);
                }
            };
        if request.offset > artifact.size {
            Self::artifact_transfer_rejected(
                send,
                artifact.size,
                Some(&artifact.sha256),
                "invalid transfer offset",
            )
            .await?;
            return Ok(None);
        }
        Ok(Some(artifact))
    }

    async fn handle_artifact_transfer_stream(
        &self,
        remote: EndpointId,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> anyhow::Result<()> {
        use tokio::io::{AsyncReadExt, AsyncSeekExt};

        let buf = read_len_prefixed(&mut recv).await?;
        let request = skippy_stage_proto::StageArtifactTransferRequest::decode(buf.as_slice())
            .map_err(|error| {
                anyhow::anyhow!("StageArtifactTransferRequest decode error: {error}")
            })?;
        skippy_protocol::validate_stage_artifact_transfer_request(&request).map_err(|error| {
            anyhow::anyhow!("StageArtifactTransferRequest validation error: {error}")
        })?;
        if request.requester_id.as_slice() != remote.as_bytes() {
            anyhow::bail!("artifact transfer requester_id does not match QUIC peer identity");
        }
        let Some(_package_dir) = self
            .authorize_artifact_transfer_request(remote, &mut send, &request)
            .await?
        else {
            return Ok(());
        };
        let Some(artifact) = self
            .resolve_artifact_transfer_request(remote, &mut send, &request)
            .await?
        else {
            return Ok(());
        };

        write_artifact_transfer_response(
            &mut send,
            true,
            artifact.size,
            Some(&artifact.sha256),
            None,
        )
        .await?;
        let mut file = tokio::fs::File::open(&artifact.path)
            .await
            .context("open artifact for transfer")?;
        file.seek(std::io::SeekFrom::Start(request.offset))
            .await
            .context("seek artifact for transfer")?;
        let mut buffer = vec![0u8; ARTIFACT_TRANSFER_BUFFER_BYTES];
        let mut remaining = artifact.size.saturating_sub(request.offset);
        while remaining > 0 {
            let limit = buffer.len().min(remaining as usize);
            let read = file
                .read(&mut buffer[..limit])
                .await
                .context("read artifact for transfer")?;
            anyhow::ensure!(read > 0, "artifact file ended before expected byte count");
            send.write_all(&buffer[..read])
                .await
                .context("write artifact transfer bytes")?;
            remaining -= read as u64;
        }
        let _ = send.finish();
        Ok(())
    }

    async fn local_verified_owner_id(&self) -> Option<String> {
        let summary = self.owner_summary.lock().await.clone();
        if summary.status == OwnershipStatus::Verified {
            summary.owner_id
        } else {
            None
        }
    }

    pub(crate) async fn artifact_transfer_allowed_for_peer(&self, peer: &PeerInfo) -> bool {
        peer.artifact_transfer_supported
            && self
                .artifact_transfer_policy_allows_peer_owner(&peer.owner_summary)
                .await
    }

    async fn artifact_transfer_serving_allowed_for_remote(&self, remote: EndpointId) -> bool {
        let peer_owner = {
            let state = self.state.lock().await;
            state
                .peers
                .get(&remote)
                .map(|peer| peer.owner_summary.clone())
        };
        let Some(peer_owner) = peer_owner else {
            return false;
        };
        self.artifact_transfer_policy_allows_peer_owner(&peer_owner)
            .await
    }

    async fn artifact_transfer_policy_allows_peer_owner(
        &self,
        peer_owner: &OwnershipSummary,
    ) -> bool {
        let local_owner = self.owner_summary.lock().await.clone();
        let trust_store = self.trust_store.lock().await.clone();
        crate::models::artifact_transfer::artifact_transfer_allowed_between(
            &local_owner,
            peer_owner,
            &trust_store,
        )
    }

    fn owner_control_snapshot_from_state(
        &self,
        state: &crate::runtime::config_state::ConfigState,
    ) -> crate::proto::node::OwnerControlConfigSnapshot {
        crate::proto::node::OwnerControlConfigSnapshot {
            node_id: self.endpoint.id().as_bytes().to_vec(),
            revision: state.revision(),
            config_hash: state.config_hash().to_vec(),
            config: Some(crate::protocol::convert::mesh_config_to_proto(
                state.config(),
            )),
            hostname: self.hostname.clone(),
        }
    }

    fn owner_control_update_from_state(
        &self,
        state: &crate::runtime::config_state::ConfigState,
    ) -> crate::proto::node::OwnerControlConfigUpdate {
        crate::proto::node::OwnerControlConfigUpdate {
            node_id: self.endpoint.id().as_bytes().to_vec(),
            revision: state.revision(),
            config_hash: state.config_hash().to_vec(),
            config: Some(crate::protocol::convert::mesh_config_to_proto(
                state.config(),
            )),
        }
    }

    async fn send_owner_control_envelope(
        &self,
        send: &mut iroh::endpoint::SendStream,
        envelope: crate::proto::node::OwnerControlEnvelope,
    ) -> anyhow::Result<()> {
        write_len_prefixed(send, &envelope.encode_to_vec()).await?;
        Ok(())
    }

    async fn send_owner_control_terminal_envelope(
        &self,
        send: &mut iroh::endpoint::SendStream,
        envelope: crate::proto::node::OwnerControlEnvelope,
    ) -> anyhow::Result<()> {
        self.send_owner_control_envelope(send, envelope).await?;
        let _ = send.finish();
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        Ok(())
    }

    async fn refresh_local_inventory_snapshot(&self) -> crate::models::LocalModelInventorySnapshot {
        let collector = self.runtime_data_collector();
        let snapshot = collector
            .coalesce_local_inventory_scan(|| {
                crate::models::scan_local_inventory_snapshot_with_progress(|_| {})
            })
            .await;
        self.set_available_models(crate::models::scan_local_models())
            .await;
        snapshot
    }

    fn owner_control_auth_error_envelope(
        &self,
        err: &crate::crypto::ControlPlaneAuthError,
    ) -> crate::proto::node::OwnerControlEnvelope {
        let code = match err {
            crate::crypto::ControlPlaneAuthError::MissingRemoteOwnerAttestation
            | crate::crypto::ControlPlaneAuthError::RemoteOwnershipInvalid { .. } => {
                crate::proto::node::OwnerControlErrorCode::InvalidHandshake
            }
            crate::crypto::ControlPlaneAuthError::TargetNodeMismatch { .. } => {
                crate::proto::node::OwnerControlErrorCode::TargetNodeMismatch
            }
            crate::crypto::ControlPlaneAuthError::MissingLocalOwnerIdentity { .. }
            | crate::crypto::ControlPlaneAuthError::RemoteOwnerMismatch { .. }
            | crate::crypto::ControlPlaneAuthError::UnsupportedTrustPolicy { .. } => {
                crate::proto::node::OwnerControlErrorCode::Unauthorized
            }
        };
        owner_control_error_envelope(code, None, None, err.to_string())
    }

    fn verify_owner_control_request_ids(
        &self,
        remote: EndpointId,
        requester_node_id: &[u8],
        target_node_id: &[u8],
        request_id: u64,
    ) -> Result<(), Box<crate::proto::node::OwnerControlEnvelope>> {
        if requester_node_id != remote.as_bytes() {
            return Err(Box::new(owner_control_error_envelope(
                crate::proto::node::OwnerControlErrorCode::BadRequest,
                Some(request_id),
                None,
                "requester_node_id does not match connection identity",
            )));
        }
        if let Err(err) =
            verify_control_plane_target_node(target_node_id, self.endpoint.id().as_bytes())
        {
            return Err(Box::new(owner_control_error_envelope(
                crate::proto::node::OwnerControlErrorCode::TargetNodeMismatch,
                Some(request_id),
                None,
                err.to_string(),
            )));
        }
        Ok(())
    }

    async fn send_owner_control_request_id_error(
        &self,
        send: &mut iroh::endpoint::SendStream,
        verification: Result<(), Box<crate::proto::node::OwnerControlEnvelope>>,
    ) -> Option<anyhow::Result<()>> {
        match verification {
            Ok(()) => None,
            Err(envelope) => Some(self.send_owner_control_envelope(send, *envelope).await),
        }
    }

    async fn current_owner_control_snapshot(
        &self,
    ) -> crate::proto::node::OwnerControlConfigSnapshot {
        let state = self.config_state.lock().await;
        self.owner_control_snapshot_from_state(&state)
    }

    async fn current_owner_control_update(&self) -> crate::proto::node::OwnerControlConfigUpdate {
        let state = self.config_state.lock().await;
        self.owner_control_update_from_state(&state)
    }

    fn owner_control_watch_response(
        &self,
        include_snapshot: bool,
        snapshot: Option<crate::proto::node::OwnerControlConfigSnapshot>,
        update: Option<crate::proto::node::OwnerControlConfigUpdate>,
    ) -> crate::proto::node::OwnerControlWatchConfigResponse {
        crate::proto::node::OwnerControlWatchConfigResponse {
            accepted: (!include_snapshot && update.is_none()).then(|| {
                crate::proto::node::OwnerControlWatchAccepted {
                    target_node_id: self.endpoint.id().as_bytes().to_vec(),
                }
            }),
            snapshot,
            update,
        }
    }

    fn owner_control_watch_envelope(
        &self,
        request_id: u64,
        watch_response: crate::proto::node::OwnerControlWatchConfigResponse,
    ) -> crate::proto::node::OwnerControlEnvelope {
        crate::proto::node::OwnerControlEnvelope {
            gen: NODE_PROTOCOL_GENERATION,
            handshake: None,
            request: None,
            response: Some(crate::proto::node::OwnerControlResponse {
                request_id,
                get_config: None,
                watch_config: Some(watch_response),
                apply_config: None,
                refresh_inventory: None,
            }),
            error: None,
        }
    }

    async fn send_owner_control_watch_update(
        &self,
        send: &mut iroh::endpoint::SendStream,
        request_id: u64,
        update: crate::proto::node::OwnerControlConfigUpdate,
    ) -> anyhow::Result<()> {
        self.send_owner_control_envelope(
            send,
            self.owner_control_watch_envelope(
                request_id,
                self.owner_control_watch_response(false, None, Some(update)),
            ),
        )
        .await
    }

    async fn handle_owner_control_get_config(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request_id: u64,
        get: crate::proto::node::OwnerControlGetConfigRequest,
    ) -> anyhow::Result<()> {
        if let Some(result) = self
            .send_owner_control_request_id_error(
                send,
                self.verify_owner_control_request_ids(
                    remote,
                    &get.requester_node_id,
                    &get.target_node_id,
                    request_id,
                ),
            )
            .await
        {
            return result;
        }
        let snapshot = self.current_owner_control_snapshot().await;
        self.send_owner_control_envelope(
            send,
            crate::proto::node::OwnerControlEnvelope {
                gen: NODE_PROTOCOL_GENERATION,
                handshake: None,
                request: None,
                response: Some(crate::proto::node::OwnerControlResponse {
                    request_id,
                    get_config: Some(crate::proto::node::OwnerControlGetConfigResponse {
                        snapshot: Some(snapshot),
                    }),
                    watch_config: None,
                    apply_config: None,
                    refresh_inventory: None,
                }),
                error: None,
            },
        )
        .await
    }

    async fn handle_owner_control_watch_config(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        recv: &mut iroh::endpoint::RecvStream,
        request_id: u64,
        watch: crate::proto::node::OwnerControlWatchConfigRequest,
    ) -> anyhow::Result<()> {
        let mut rev_rx = self.config_revision_tx.subscribe();
        if let Some(result) = self
            .send_owner_control_request_id_error(
                send,
                self.verify_owner_control_request_ids(
                    remote,
                    &watch.requester_node_id,
                    &watch.target_node_id,
                    request_id,
                ),
            )
            .await
        {
            return result;
        }

        self.send_owner_control_watch_start(send, request_id, watch.include_snapshot)
            .await?;

        self.stream_owner_control_watch_updates(send, recv, remote, request_id, &mut rev_rx)
            .await;

        Ok(())
    }

    async fn send_owner_control_watch_start(
        &self,
        send: &mut iroh::endpoint::SendStream,
        request_id: u64,
        include_snapshot: bool,
    ) -> anyhow::Result<()> {
        let watch_response = self.owner_control_watch_response(
            include_snapshot,
            if include_snapshot {
                Some(self.current_owner_control_snapshot().await)
            } else {
                None
            },
            None,
        );
        self.send_owner_control_envelope(
            send,
            self.owner_control_watch_envelope(request_id, watch_response),
        )
        .await?;

        Ok(())
    }

    async fn stream_owner_control_watch_updates(
        &self,
        send: &mut iroh::endpoint::SendStream,
        recv: &mut iroh::endpoint::RecvStream,
        remote: EndpointId,
        request_id: u64,
        rev_rx: &mut tokio::sync::watch::Receiver<u64>,
    ) {
        loop {
            tokio::select! {
                changed = rev_rx.changed() => {
                    if changed.is_err() {
                        break;
                    }
                    let update = self.current_owner_control_update().await;
                    if self
                        .send_owner_control_watch_update(send, request_id, update)
                        .await
                        .is_err()
                    {
                        break;
                    }
                }
                inbound = read_len_prefixed(recv) => {
                    if inbound.is_ok() {
                        tracing::debug!(
                            "owner-control watch from {} sent unexpected extra frame; closing stream",
                            remote.fmt_short()
                        );
                    }
                    break;
                }
            }
        }
    }

    async fn handle_owner_control_apply_config(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request_id: u64,
        apply: crate::proto::node::OwnerControlApplyConfigRequest,
    ) -> anyhow::Result<()> {
        use crate::runtime::config_state::{ApplyResult, ConfigApplyMode};

        if let Some(result) = self
            .send_owner_control_request_id_error(
                send,
                self.verify_owner_control_request_ids(
                    remote,
                    &apply.requester_node_id,
                    &apply.target_node_id,
                    request_id,
                ),
            )
            .await
        {
            return result;
        }
        let Some(config_snapshot) = apply.config.clone() else {
            return self
                .send_owner_control_envelope(
                    send,
                    owner_control_error_envelope(
                        crate::proto::node::OwnerControlErrorCode::BadRequest,
                        Some(request_id),
                        None,
                        "missing config payload",
                    ),
                )
                .await;
        };

        let mesh_config = crate::protocol::convert::proto_config_to_mesh(&config_snapshot);
        let config_state = Arc::clone(&self.config_state);
        let expected_revision = apply.expected_revision;
        let apply_result = tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
            preflight_pushed_config_for_current_node(&mesh_config)?;
            let mut state = config_state.blocking_lock();
            let result = state.apply(mesh_config, expected_revision);
            let current_revision = state.revision();
            let current_hash = *state.config_hash();
            Ok((result, current_revision, current_hash))
        })
        .await
        .map_err(|e| anyhow::anyhow!("config apply task panicked: {e}"))?;

        let (result, current_revision, current_hash) = match apply_result {
            Ok(values) => values,
            Err(error) => {
                return self
                    .send_owner_control_envelope(
                        send,
                        owner_control_error_envelope(
                            crate::proto::node::OwnerControlErrorCode::BadRequest,
                            Some(request_id),
                            None,
                            error.to_string(),
                        ),
                    )
                    .await;
            }
        };

        let envelope = match result {
            ApplyResult::Applied {
                revision,
                hash,
                apply_mode,
            } => {
                if apply_mode == ConfigApplyMode::Staged {
                    let _ = self.config_revision_tx.send(revision);
                }
                crate::proto::node::OwnerControlEnvelope {
                    gen: NODE_PROTOCOL_GENERATION,
                    handshake: None,
                    request: None,
                    response: Some(crate::proto::node::OwnerControlResponse {
                        request_id,
                        get_config: None,
                        watch_config: None,
                        apply_config: Some(crate::proto::node::OwnerControlApplyConfigResponse {
                            success: true,
                            current_revision: revision,
                            config_hash: hash.to_vec(),
                            error: None,
                            apply_mode: match apply_mode {
                                ConfigApplyMode::Staged => {
                                    crate::proto::node::ConfigApplyMode::Staged as i32
                                }
                                ConfigApplyMode::Noop => {
                                    crate::proto::node::ConfigApplyMode::Noop as i32
                                }
                            },
                        }),
                        refresh_inventory: None,
                    }),
                    error: None,
                }
            }
            ApplyResult::RevisionConflict { current_revision } => owner_control_error_envelope(
                crate::proto::node::OwnerControlErrorCode::RevisionConflict,
                Some(request_id),
                Some(current_revision),
                "revision conflict: expected_revision does not match current",
            ),
            ApplyResult::PersistedWithRevisionTrackingError {
                revision,
                hash,
                error,
            } => {
                let _ = self.config_revision_tx.send(revision);
                crate::proto::node::OwnerControlEnvelope {
                    gen: NODE_PROTOCOL_GENERATION,
                    handshake: None,
                    request: None,
                    response: Some(crate::proto::node::OwnerControlResponse {
                        request_id,
                        get_config: None,
                        watch_config: None,
                        apply_config: Some(crate::proto::node::OwnerControlApplyConfigResponse {
                            success: false,
                            current_revision: revision,
                            config_hash: hash.to_vec(),
                            error: Some(error),
                            apply_mode: crate::proto::node::ConfigApplyMode::Staged as i32,
                        }),
                        refresh_inventory: None,
                    }),
                    error: None,
                }
            }
            ApplyResult::ValidationError(error) | ApplyResult::PersistError(error) => {
                crate::proto::node::OwnerControlEnvelope {
                    gen: NODE_PROTOCOL_GENERATION,
                    handshake: None,
                    request: None,
                    response: Some(crate::proto::node::OwnerControlResponse {
                        request_id,
                        get_config: None,
                        watch_config: None,
                        apply_config: Some(crate::proto::node::OwnerControlApplyConfigResponse {
                            success: false,
                            current_revision,
                            config_hash: current_hash.to_vec(),
                            error: Some(error),
                            apply_mode: crate::proto::node::ConfigApplyMode::Unspecified as i32,
                        }),
                        refresh_inventory: None,
                    }),
                    error: None,
                }
            }
        };
        self.send_owner_control_envelope(send, envelope).await
    }

    async fn handle_owner_control_refresh_inventory(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        request_id: u64,
        refresh: crate::proto::node::OwnerControlRefreshInventoryRequest,
    ) -> anyhow::Result<()> {
        if let Some(result) = self
            .send_owner_control_request_id_error(
                send,
                self.verify_owner_control_request_ids(
                    remote,
                    &refresh.requester_node_id,
                    &refresh.target_node_id,
                    request_id,
                ),
            )
            .await
        {
            return result;
        }
        let _ = self.refresh_local_inventory_snapshot().await;
        let snapshot = self.current_owner_control_snapshot().await;
        self.send_owner_control_envelope(
            send,
            crate::proto::node::OwnerControlEnvelope {
                gen: NODE_PROTOCOL_GENERATION,
                handshake: None,
                request: None,
                response: Some(crate::proto::node::OwnerControlResponse {
                    request_id,
                    get_config: None,
                    watch_config: None,
                    apply_config: None,
                    refresh_inventory: Some(
                        crate::proto::node::OwnerControlRefreshInventoryResponse {
                            snapshot: Some(snapshot),
                        },
                    ),
                }),
                error: None,
            },
        )
        .await
    }

    async fn handle_owner_control_request(
        &self,
        remote: EndpointId,
        send: &mut iroh::endpoint::SendStream,
        recv: &mut iroh::endpoint::RecvStream,
        request: crate::proto::node::OwnerControlRequest,
    ) -> anyhow::Result<()> {
        let request_id = request.request_id;

        if let Some(get) = request.get_config {
            return self
                .handle_owner_control_get_config(remote, send, request_id, get)
                .await;
        }

        if let Some(watch) = request.watch_config {
            return self
                .handle_owner_control_watch_config(remote, send, recv, request_id, watch)
                .await;
        }

        if let Some(apply) = request.apply_config {
            return self
                .handle_owner_control_apply_config(remote, send, request_id, apply)
                .await;
        }

        if let Some(refresh) = request.refresh_inventory {
            return self
                .handle_owner_control_refresh_inventory(remote, send, request_id, refresh)
                .await;
        }

        self.send_owner_control_envelope(
            send,
            owner_control_error_envelope(
                crate::proto::node::OwnerControlErrorCode::UnknownCommand,
                Some(request_id),
                None,
                "unknown owner-control command",
            ),
        )
        .await
    }

    // --- Gossip ---

    async fn connect_to_peer(&self, addr: EndpointAddr) -> Result<()> {
        let peer_id = addr.id;
        if peer_id == self.endpoint.id() {
            return Ok(());
        }

        {
            let state = self.state.lock().await;
            if state.peers.contains_key(&peer_id) {
                return Ok(());
            }
            if state
                .dead_peers
                .get(&peer_id)
                .is_some_and(|t| t.elapsed() < DEAD_PEER_TTL)
            {
                tracing::debug!("Skipping connection to dead peer {}", peer_id.fmt_short());
                return Ok(());
            }
        }

        tracing::info!("Connecting to peer {}...", peer_id.fmt_short());
        let conn = match tokio::time::timeout(
            PEER_CONNECT_AND_GOSSIP_TIMEOUT,
            connect_mesh(&self.endpoint, addr.clone()),
        )
        .await
        {
            Ok(Ok(c)) => c,
            Ok(Err(e)) => {
                anyhow::bail!("Failed to connect to {}: {e}", peer_id.fmt_short());
            }
            Err(_) => {
                anyhow::bail!(
                    "Timeout connecting to {} ({}s)",
                    peer_id.fmt_short(),
                    PEER_CONNECT_AND_GOSSIP_TIMEOUT.as_secs()
                );
            }
        };

        // Store connection and start dispatcher for inbound streams from this peer
        {
            let mut state = self.state.lock().await;
            state.connections.insert(peer_id, conn.clone());
        }
        let node_for_dispatch = self.clone();
        let conn_for_dispatch = conn.clone();
        tokio::spawn(async move {
            node_for_dispatch
                .dispatch_streams(conn_for_dispatch, peer_id)
                .await;
        });

        // Gossip exchange to learn peer's role/VRAM and announce ourselves
        self.initiate_gossip(conn.clone(), peer_id).await?;

        // Schedule a delayed RTT recheck: the first gossip often goes via relay
        // (high RTT) because direct holepunch hasn't completed yet. After a few
        // seconds the direct path is usually ready, so re-check path info to get
        // the real RTT and potentially trigger a re-election for split mode.
        let node_for_recheck = self.clone();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
            let conn = node_for_recheck
                .state
                .lock()
                .await
                .connections
                .get(&peer_id)
                .cloned();
            if let Some(conn) = conn {
                let path_list = conn.paths();
                for path_info in &path_list {
                    if path_info.is_selected() {
                        let rtt_ms = path_info.rtt().as_millis() as u32;
                        if rtt_ms == 0 {
                            continue;
                        }
                        let path_type = if path_info.is_ip() { "direct" } else { "relay" };
                        if rtt_ms > 0 {
                            emit_mesh_info(format!(
                                "📡 Peer {} RTT recheck: {}ms ({})",
                                peer_id.fmt_short(),
                                rtt_ms,
                                path_type
                            ));
                            node_for_recheck.update_peer_rtt(peer_id, rtt_ms).await;
                        }
                        break;
                    }
                }
            }
        });
        Ok(())
    }

    async fn handle_tunnel_map_stream(
        &self,
        remote: EndpointId,
        protocol: ControlProtocol,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        use prost::Message as _;

        let buf = read_len_prefixed(&mut recv).await?;
        let _ = protocol;
        let frame = crate::proto::node::TunnelMap::decode(buf.as_slice())
            .map_err(|e| anyhow::anyhow!("TunnelMap decode error: {e}"))?;

        frame
            .validate_frame()
            .map_err(|e| anyhow::anyhow!("TunnelMap validation failed: {e}"))?;

        let entry_count = frame.entries.len();
        {
            let mut state = self.state.lock().await;
            ingest_tunnel_map(remote, &frame, &mut state.remote_tunnel_maps)?;
        }

        tracing::info!(
            "Received tunnel map from {} ({} entries)",
            remote.fmt_short(),
            entry_count
        );

        Ok(())
    }
}

fn stage_topology_key(topology_id: &str, run_id: &str) -> String {
    format!("{topology_id}\n{run_id}")
}

fn stage_runtime_status_key(topology_id: &str, run_id: &str, stage_id: &str) -> String {
    format!("{topology_id}\n{run_id}\n{stage_id}")
}

fn endpoint_id_from_bytes(bytes: Vec<u8>) -> anyhow::Result<EndpointId> {
    let arr: [u8; 32] = bytes.as_slice().try_into().map_err(|_| {
        anyhow::anyhow!(
            "invalid endpoint id length: expected 32, got {}",
            bytes.len()
        )
    })?;
    let public_key = iroh::PublicKey::from_bytes(&arr)
        .map_err(|error| anyhow::anyhow!("invalid endpoint id bytes: {error}"))?;
    Ok(EndpointId::from(public_key))
}

fn stage_runtime_status_from_snapshot(
    node_id: Option<EndpointId>,
    status: crate::inference::skippy::StageStatusSnapshot,
) -> StageRuntimeStatus {
    StageRuntimeStatus {
        topology_id: status.topology_id,
        run_id: status.run_id,
        model_id: status.model_id,
        backend: status.backend,
        package_ref: status.package_ref,
        manifest_sha256: status.manifest_sha256,
        source_model_path: status.source_model_path,
        source_model_sha256: status.source_model_sha256,
        source_model_bytes: status.source_model_bytes,
        materialized_path: status.materialized_path,
        materialized_pinned: status.materialized_pinned,
        projector_path: status.projector_path,
        stage_id: status.stage_id,
        stage_index: status.stage_index,
        node_id,
        layer_start: status.layer_start,
        layer_end: status.layer_end,
        state: status.state,
        bind_addr: status.bind_addr,
        activation_width: status.activation_width,
        wire_dtype: status.wire_dtype,
        selected_device: status.selected_device,
        ctx_size: status.ctx_size,
        lane_count: status.lane_count,
        n_batch: status.n_batch,
        n_ubatch: status.n_ubatch,
        flash_attn_type: status.flash_attn_type,
        error: status.error,
        shutdown_generation: status.shutdown_generation,
    }
}

fn stage_snapshot_from_runtime_status(
    status: &StageRuntimeStatus,
    state: crate::inference::skippy::StageRuntimeState,
    error: Option<String>,
) -> crate::inference::skippy::StageStatusSnapshot {
    crate::inference::skippy::StageStatusSnapshot {
        topology_id: status.topology_id.clone(),
        run_id: status.run_id.clone(),
        model_id: status.model_id.clone(),
        backend: status.backend.clone(),
        package_ref: status.package_ref.clone(),
        manifest_sha256: status.manifest_sha256.clone(),
        source_model_path: status.source_model_path.clone(),
        source_model_sha256: status.source_model_sha256.clone(),
        source_model_bytes: status.source_model_bytes,
        materialized_path: status.materialized_path.clone(),
        materialized_pinned: status.materialized_pinned,
        projector_path: status.projector_path.clone(),
        stage_id: status.stage_id.clone(),
        stage_index: status.stage_index,
        layer_start: status.layer_start,
        layer_end: status.layer_end,
        state,
        bind_addr: status.bind_addr.clone(),
        activation_width: status.activation_width,
        wire_dtype: status.wire_dtype,
        selected_device: status.selected_device.clone(),
        ctx_size: status.ctx_size,
        lane_count: status.lane_count,
        n_batch: status.n_batch,
        n_ubatch: status.n_ubatch,
        flash_attn_type: status.flash_attn_type,
        error,
        shutdown_generation: status.shutdown_generation,
        coordinator_term: 0,
        coordinator_id: None,
        lease_until_unix_ms: 0,
    }
}

fn stage_topology_from_load(
    node_id: EndpointId,
    load: &crate::inference::skippy::StageLoadRequest,
) -> StageTopologyInstance {
    StageTopologyInstance {
        topology_id: load.topology_id.clone(),
        run_id: load.run_id.clone(),
        model_id: load.model_id.clone(),
        package_ref: load.package_ref.clone(),
        manifest_sha256: load.manifest_sha256.clone(),
        stages: vec![StageAssignment {
            stage_id: load.stage_id.clone(),
            stage_index: load.stage_index,
            node_id,
            layer_start: load.layer_start,
            layer_end: load.layer_end,
            endpoint: StageEndpoint {
                bind_addr: load.bind_addr.clone(),
            },
        }],
    }
}

fn stage_control_request_to_proto(
    requester_id: EndpointId,
    request: crate::inference::skippy::StageControlRequest,
) -> skippy_stage_proto::StageControlRequest {
    use skippy_stage_proto::stage_control_request::Command;

    let command = match request {
        crate::inference::skippy::StageControlRequest::Claim(claim) => {
            Command::ClaimCoordinator(stage_coordinator_claim_to_proto(claim))
        }
        crate::inference::skippy::StageControlRequest::Load(load) => {
            Command::LoadStage(stage_load_to_proto(load))
        }
        crate::inference::skippy::StageControlRequest::Stop(stop) => {
            Command::StopStage(skippy_stage_proto::StopStage {
                topology_id: stop.topology_id,
                run_id: stop.run_id,
                stage_id: stop.stage_id,
                shutdown_generation: stop.shutdown_generation,
                coordinator_term: stop.coordinator_term,
            })
        }
        crate::inference::skippy::StageControlRequest::Status(status) => {
            Command::GetStageStatus(skippy_stage_proto::GetStageStatus {
                topology_id: status.topology_id,
                run_id: status.run_id,
                stage_id: status.stage_id,
            })
        }
        crate::inference::skippy::StageControlRequest::Inventory(inventory) => {
            Command::GetLayerInventory(skippy_stage_proto::GetLayerInventory {
                model_id: inventory.model_id,
                package_ref: inventory.package_ref,
                manifest_sha256: inventory.manifest_sha256,
            })
        }
        crate::inference::skippy::StageControlRequest::Prepare(prepare) => {
            Command::PrepareStage(skippy_stage_proto::PrepareStage {
                load_stage: Some(stage_load_to_proto(prepare.load)),
                coordinator_id: prepare.coordinator_id.map(|id| id.as_bytes().to_vec()),
            })
        }
        crate::inference::skippy::StageControlRequest::CancelPrepare(cancel) => {
            Command::CancelPrepareStage(skippy_stage_proto::CancelPrepareStage {
                topology_id: cancel.topology_id,
                run_id: cancel.run_id,
                stage_id: cancel.stage_id,
                shutdown_generation: cancel.shutdown_generation,
            })
        }
        crate::inference::skippy::StageControlRequest::StatusUpdate(status) => {
            Command::StageStatusUpdate(skippy_stage_proto::StageStatusUpdate {
                status: Some(stage_preparation_status_to_proto(status)),
            })
        }
    };

    skippy_stage_proto::StageControlRequest {
        gen: skippy_protocol::STAGE_PROTOCOL_GENERATION,
        requester_id: requester_id.as_bytes().to_vec(),
        command: Some(command),
    }
}

fn stage_load_to_proto(
    load: crate::inference::skippy::StageLoadRequest,
) -> skippy_stage_proto::LoadStage {
    skippy_stage_proto::LoadStage {
        topology_id: load.topology_id,
        run_id: load.run_id,
        model_id: load.model_id,
        backend: load.backend,
        package_ref: load.package_ref,
        manifest_sha256: load.manifest_sha256,
        stage_id: load.stage_id,
        stage_index: load.stage_index,
        layer_start: load.layer_start,
        layer_end: load.layer_end,
        model_path: load.model_path,
        source_model_bytes: load.source_model_bytes,
        projector_path: load.projector_path,
        selected_device: load.selected_device.map(stage_device_to_proto),
        bind_addr: load.bind_addr,
        activation_width: load.activation_width.max(0) as u32,
        wire_dtype: stage_wire_dtype_to_proto(load.wire_dtype) as i32,
        ctx_size: load.ctx_size,
        lane_count: load.lane_count,
        n_batch: load.n_batch,
        n_ubatch: load.n_ubatch,
        n_gpu_layers: load.n_gpu_layers,
        cache_type_k: load.cache_type_k,
        cache_type_v: load.cache_type_v,
        flash_attn_type: stage_flash_attn_type_to_proto(load.flash_attn_type) as i32,
        shutdown_generation: load.shutdown_generation,
        coordinator_term: load.coordinator_term,
        coordinator_id: load.coordinator_id.map(|id| id.to_string()),
        lease_until_unix_ms: load.lease_until_unix_ms,
        load_mode: match load.load_mode {
            skippy_protocol::LoadMode::RuntimeSlice => {
                skippy_stage_proto::StageLoadMode::RuntimeSlice as i32
            }
            skippy_protocol::LoadMode::LayerPackage => {
                skippy_stage_proto::StageLoadMode::LayerPackage as i32
            }
            skippy_protocol::LoadMode::ArtifactSlice => {
                skippy_stage_proto::StageLoadMode::ArtifactSlice as i32
            }
        },
        upstream: load.upstream.map(stage_peer_to_proto),
        downstream: load.downstream.map(stage_peer_to_proto),
    }
}

fn stage_coordinator_claim_to_proto(
    claim: crate::inference::skippy::StageCoordinatorClaim,
) -> skippy_stage_proto::ClaimCoordinator {
    skippy_stage_proto::ClaimCoordinator {
        model_id: claim.model_id,
        package_ref: claim.package_ref,
        manifest_sha256: claim.manifest_sha256,
        topology_id: claim.topology_id,
        run_id: claim.run_id,
        coordinator_id: claim.coordinator_id,
        coordinator_term: claim.coordinator_term,
        participant_set_hash: claim.participant_set_hash,
        topology_hash: claim.topology_hash,
        lease_until_unix_ms: claim.lease_until_unix_ms,
    }
}

fn stage_peer_to_proto(
    peer: crate::inference::skippy::StagePeerDescriptor,
) -> skippy_stage_proto::StagePeer {
    skippy_stage_proto::StagePeer {
        stage_id: peer.stage_id,
        stage_index: peer.stage_index,
        endpoint: peer.endpoint,
        node_id: peer.node_id.map(|id| id.as_bytes().to_vec()),
    }
}

fn stage_device_to_proto(device: skippy_protocol::StageDevice) -> skippy_stage_proto::StageDevice {
    skippy_stage_proto::StageDevice {
        backend_device: device.backend_device,
        stable_id: device.stable_id,
        index: device.index.map(|value| value as u64),
        vram_bytes: device.vram_bytes,
    }
}

fn stage_control_request_from_proto(
    frame: skippy_stage_proto::StageControlRequest,
) -> anyhow::Result<crate::inference::skippy::StageControlRequest> {
    use skippy_stage_proto::stage_control_request::Command;

    match frame
        .command
        .ok_or_else(|| anyhow::anyhow!("missing stage control command"))?
    {
        Command::ClaimCoordinator(claim) => {
            Ok(crate::inference::skippy::StageControlRequest::Claim(
                stage_coordinator_claim_from_proto(claim)?,
            ))
        }
        Command::LoadStage(load) => Ok(crate::inference::skippy::StageControlRequest::Load(
            stage_load_from_proto(load)?,
        )),
        Command::StopStage(stop) => Ok(crate::inference::skippy::StageControlRequest::Stop(
            crate::inference::skippy::StageStopRequest {
                topology_id: stop.topology_id,
                run_id: stop.run_id,
                stage_id: stop.stage_id,
                shutdown_generation: stop.shutdown_generation,
                coordinator_term: stop.coordinator_term,
            },
        )),
        Command::GetStageStatus(status) => {
            Ok(crate::inference::skippy::StageControlRequest::Status(
                crate::inference::skippy::StageStatusFilter {
                    topology_id: status.topology_id,
                    run_id: status.run_id,
                    stage_id: status.stage_id,
                },
            ))
        }
        Command::GetLayerInventory(inventory) => {
            Ok(crate::inference::skippy::StageControlRequest::Inventory(
                crate::inference::skippy::StageInventoryRequest {
                    model_id: inventory.model_id,
                    package_ref: inventory.package_ref,
                    manifest_sha256: inventory.manifest_sha256,
                },
            ))
        }
        Command::PrepareStage(prepare) => {
            let load = prepare
                .load_stage
                .ok_or_else(|| anyhow::anyhow!("prepare stage missing load_stage"))?;
            Ok(crate::inference::skippy::StageControlRequest::Prepare(
                crate::inference::skippy::StagePrepareRequest {
                    load: stage_load_from_proto(load)?,
                    coordinator_id: prepare
                        .coordinator_id
                        .map(endpoint_id_from_bytes)
                        .transpose()
                        .context("invalid prepare stage coordinator_id")?,
                },
            ))
        }
        Command::CancelPrepareStage(cancel) => Ok(
            crate::inference::skippy::StageControlRequest::CancelPrepare(
                crate::inference::skippy::StageCancelPrepareRequest {
                    topology_id: cancel.topology_id,
                    run_id: cancel.run_id,
                    stage_id: cancel.stage_id,
                    shutdown_generation: cancel.shutdown_generation,
                },
            ),
        ),
        Command::StageStatusUpdate(update) => {
            let status = update
                .status
                .ok_or_else(|| anyhow::anyhow!("stage status update missing status"))?;
            Ok(crate::inference::skippy::StageControlRequest::StatusUpdate(
                stage_preparation_status_from_proto(status),
            ))
        }
    }
}

fn stage_load_from_proto(
    load: skippy_stage_proto::LoadStage,
) -> anyhow::Result<crate::inference::skippy::StageLoadRequest> {
    Ok(crate::inference::skippy::StageLoadRequest {
        topology_id: load.topology_id,
        run_id: load.run_id,
        model_id: load.model_id,
        backend: load.backend,
        package_ref: load.package_ref,
        manifest_sha256: load.manifest_sha256,
        stage_id: load.stage_id,
        stage_index: load.stage_index,
        layer_start: load.layer_start,
        layer_end: load.layer_end,
        model_path: load.model_path,
        source_model_bytes: load.source_model_bytes,
        projector_path: load.projector_path,
        selected_device: load
            .selected_device
            .map(stage_device_from_proto)
            .transpose()?,
        bind_addr: load.bind_addr,
        activation_width: i32::try_from(load.activation_width)
            .context("stage activation_width exceeds i32")?,
        wire_dtype: stage_wire_dtype_from_proto(load.wire_dtype),
        ctx_size: load.ctx_size,
        lane_count: if load.lane_count == 0 {
            4
        } else {
            load.lane_count
        },
        n_batch: load.n_batch,
        n_ubatch: load.n_ubatch,
        n_gpu_layers: load.n_gpu_layers,
        cache_type_k: load.cache_type_k,
        cache_type_v: load.cache_type_v,
        flash_attn_type: stage_flash_attn_type_from_proto(load.flash_attn_type),
        shutdown_generation: load.shutdown_generation,
        coordinator_term: load.coordinator_term,
        coordinator_id: load
            .coordinator_id
            .map(|id| id.parse())
            .transpose()
            .context("invalid stage load coordinator_id")?,
        lease_until_unix_ms: load.lease_until_unix_ms,
        load_mode: stage_load_mode_from_proto(load.load_mode),
        upstream: load.upstream.map(stage_peer_from_proto).transpose()?,
        downstream: load.downstream.map(stage_peer_from_proto).transpose()?,
    })
}

fn stage_coordinator_claim_from_proto(
    claim: skippy_stage_proto::ClaimCoordinator,
) -> anyhow::Result<crate::inference::skippy::StageCoordinatorClaim> {
    Ok(crate::inference::skippy::StageCoordinatorClaim {
        model_id: claim.model_id,
        package_ref: claim.package_ref,
        manifest_sha256: claim.manifest_sha256,
        topology_id: claim.topology_id,
        run_id: claim.run_id,
        coordinator_id: claim.coordinator_id,
        coordinator_term: claim.coordinator_term,
        participant_set_hash: claim.participant_set_hash,
        topology_hash: claim.topology_hash,
        lease_until_unix_ms: claim.lease_until_unix_ms,
    })
}

fn stage_device_from_proto(
    device: skippy_stage_proto::StageDevice,
) -> anyhow::Result<skippy_protocol::StageDevice> {
    Ok(skippy_protocol::StageDevice {
        backend_device: device.backend_device,
        stable_id: device.stable_id,
        index: device
            .index
            .map(usize::try_from)
            .transpose()
            .context("stage selected_device.index exceeds usize")?,
        vram_bytes: device.vram_bytes,
    })
}

fn stage_peer_from_proto(
    peer: skippy_stage_proto::StagePeer,
) -> anyhow::Result<crate::inference::skippy::StagePeerDescriptor> {
    Ok(crate::inference::skippy::StagePeerDescriptor {
        stage_id: peer.stage_id,
        stage_index: peer.stage_index,
        endpoint: peer.endpoint,
        node_id: peer
            .node_id
            .map(endpoint_id_from_bytes)
            .transpose()
            .context("invalid stage peer node_id")?,
    })
}

fn stage_load_mode_from_proto(value: i32) -> skippy_protocol::LoadMode {
    match skippy_stage_proto::StageLoadMode::try_from(value)
        .unwrap_or(skippy_stage_proto::StageLoadMode::Unspecified)
    {
        skippy_stage_proto::StageLoadMode::Unspecified
        | skippy_stage_proto::StageLoadMode::RuntimeSlice => {
            skippy_protocol::LoadMode::RuntimeSlice
        }
        skippy_stage_proto::StageLoadMode::LayerPackage => skippy_protocol::LoadMode::LayerPackage,
        skippy_stage_proto::StageLoadMode::ArtifactSlice => {
            skippy_protocol::LoadMode::ArtifactSlice
        }
    }
}

fn stage_wire_dtype_from_proto(value: i32) -> crate::inference::skippy::StageWireDType {
    match skippy_stage_proto::StageWireDType::try_from(value)
        .unwrap_or(skippy_stage_proto::StageWireDType::StageWireDtypeUnspecified)
    {
        skippy_stage_proto::StageWireDType::StageWireDtypeUnspecified
        | skippy_stage_proto::StageWireDType::StageWireDtypeF16 => {
            crate::inference::skippy::StageWireDType::F16
        }
        skippy_stage_proto::StageWireDType::StageWireDtypeF32 => {
            crate::inference::skippy::StageWireDType::F32
        }
        skippy_stage_proto::StageWireDType::StageWireDtypeQ8 => {
            crate::inference::skippy::StageWireDType::Q8
        }
    }
}

fn stage_control_unavailable_response(
    request: crate::inference::skippy::StageControlRequest,
) -> crate::inference::skippy::StageControlResponse {
    let status = match request {
        crate::inference::skippy::StageControlRequest::Claim(claim) => {
            return crate::inference::skippy::StageControlResponse::ClaimAccepted(
                crate::inference::skippy::StageCoordinatorClaimAck {
                    accepted: false,
                    claim,
                    error: Some("stage control is not available".to_string()),
                },
            );
        }
        crate::inference::skippy::StageControlRequest::Load(load) => {
            stage_status_from_load(&load, crate::inference::skippy::StageRuntimeState::Failed)
        }
        crate::inference::skippy::StageControlRequest::Stop(stop) => {
            crate::inference::skippy::StageStatusSnapshot {
                topology_id: stop.topology_id,
                run_id: stop.run_id,
                model_id: String::new(),
                backend: "skippy".to_string(),
                package_ref: None,
                manifest_sha256: None,
                source_model_path: None,
                source_model_sha256: None,
                source_model_bytes: None,
                materialized_path: None,
                materialized_pinned: false,
                projector_path: None,
                stage_id: stop.stage_id,
                stage_index: 0,
                layer_start: 0,
                layer_end: 0,
                state: crate::inference::skippy::StageRuntimeState::Failed,
                bind_addr: String::new(),
                activation_width: 0,
                wire_dtype: crate::inference::skippy::StageWireDType::F16,
                selected_device: None,
                ctx_size: 0,
                lane_count: 0,
                n_batch: None,
                n_ubatch: None,
                flash_attn_type: skippy_protocol::FlashAttentionType::Auto,
                error: Some("stage control is not available".to_string()),
                shutdown_generation: stop.shutdown_generation,
                coordinator_term: stop.coordinator_term,
                coordinator_id: None,
                lease_until_unix_ms: 0,
            }
        }
        crate::inference::skippy::StageControlRequest::Status(_) => {
            return crate::inference::skippy::StageControlResponse::Status(Vec::new());
        }
        crate::inference::skippy::StageControlRequest::Inventory(inventory) => {
            return crate::inference::skippy::StageControlResponse::Inventory(
                crate::inference::skippy::StageLayerInventory {
                    model_id: inventory.model_id,
                    package_ref: inventory.package_ref,
                    manifest_sha256: inventory.manifest_sha256,
                    layer_count: 0,
                    ready_ranges: Vec::new(),
                    available_ranges: Vec::new(),
                    missing_ranges: Vec::new(),
                    preparing_ranges: Vec::new(),
                    source_model_path: None,
                    source_model_bytes: None,
                    source_model_kind: crate::inference::skippy::SourceModelKind::Unknown,
                },
            );
        }
        crate::inference::skippy::StageControlRequest::Prepare(prepare) => {
            return crate::inference::skippy::StageControlResponse::PrepareAccepted(
                crate::inference::skippy::StagePrepareAcceptedResponse {
                    accepted: false,
                    status: stage_preparation_status_from_load(
                        &prepare.load,
                        crate::inference::skippy::StagePreparationState::Failed,
                        Some("stage control is not available".to_string()),
                    ),
                    error: Some("stage control is not available".to_string()),
                },
            );
        }
        crate::inference::skippy::StageControlRequest::CancelPrepare(cancel) => {
            return crate::inference::skippy::StageControlResponse::PreparationStatus(
                stage_preparation_status_from_cancel(
                    cancel,
                    crate::inference::skippy::StagePreparationState::Failed,
                    Some("stage control is not available".to_string()),
                ),
            );
        }
        crate::inference::skippy::StageControlRequest::StatusUpdate(_) => {
            return crate::inference::skippy::StageControlResponse::StatusAck(
                crate::inference::skippy::StageStatusAck {
                    accepted: false,
                    error: Some("stage control is not available".to_string()),
                },
            );
        }
    };
    crate::inference::skippy::StageControlResponse::Ready(
        crate::inference::skippy::StageReadyResponse {
            accepted: false,
            status,
            error: Some("stage control is not available".to_string()),
        },
    )
}

fn stage_status_from_load(
    load: &crate::inference::skippy::StageLoadRequest,
    state: crate::inference::skippy::StageRuntimeState,
) -> crate::inference::skippy::StageStatusSnapshot {
    crate::inference::skippy::StageStatusSnapshot {
        topology_id: load.topology_id.clone(),
        run_id: load.run_id.clone(),
        model_id: load.model_id.clone(),
        backend: load.backend.clone(),
        package_ref: Some(load.package_ref.clone()),
        manifest_sha256: Some(load.manifest_sha256.clone()),
        source_model_path: load.model_path.clone(),
        source_model_sha256: None,
        source_model_bytes: load.source_model_bytes,
        materialized_path: None,
        materialized_pinned: false,
        projector_path: load.projector_path.clone(),
        stage_id: load.stage_id.clone(),
        stage_index: load.stage_index,
        layer_start: load.layer_start,
        layer_end: load.layer_end,
        state,
        bind_addr: load.bind_addr.clone(),
        activation_width: load.activation_width.max(0) as u32,
        wire_dtype: load.wire_dtype,
        selected_device: load.selected_device.clone(),
        ctx_size: load.ctx_size,
        lane_count: load.lane_count,
        n_batch: load.n_batch,
        n_ubatch: load.n_ubatch,
        flash_attn_type: load.flash_attn_type,
        error: Some("stage control is not available".to_string()),
        shutdown_generation: load.shutdown_generation,
        coordinator_term: load.coordinator_term,
        coordinator_id: load.coordinator_id,
        lease_until_unix_ms: load.lease_until_unix_ms,
    }
}

fn stage_preparation_status_from_load(
    load: &crate::inference::skippy::StageLoadRequest,
    state: crate::inference::skippy::StagePreparationState,
    error: Option<String>,
) -> crate::inference::skippy::StagePreparationStatus {
    crate::inference::skippy::StagePreparationStatus {
        topology_id: load.topology_id.clone(),
        run_id: load.run_id.clone(),
        model_id: load.model_id.clone(),
        backend: load.backend.clone(),
        package_ref: load.package_ref.clone(),
        manifest_sha256: load.manifest_sha256.clone(),
        stage_id: load.stage_id.clone(),
        stage_index: load.stage_index,
        layer_start: load.layer_start,
        layer_end: load.layer_end,
        state,
        bytes_done: None,
        bytes_total: None,
        bind_addr: None,
        error,
        shutdown_generation: load.shutdown_generation,
        coordinator_term: load.coordinator_term,
        coordinator_id: load.coordinator_id,
        lease_until_unix_ms: load.lease_until_unix_ms,
    }
}

fn stage_preparation_status_from_cancel(
    cancel: crate::inference::skippy::StageCancelPrepareRequest,
    state: crate::inference::skippy::StagePreparationState,
    error: Option<String>,
) -> crate::inference::skippy::StagePreparationStatus {
    crate::inference::skippy::StagePreparationStatus {
        topology_id: cancel.topology_id,
        run_id: cancel.run_id,
        model_id: String::new(),
        backend: "skippy".to_string(),
        package_ref: String::new(),
        manifest_sha256: String::new(),
        stage_id: cancel.stage_id,
        stage_index: 0,
        layer_start: 0,
        layer_end: 0,
        state,
        bytes_done: None,
        bytes_total: None,
        bind_addr: None,
        error,
        shutdown_generation: cancel.shutdown_generation,
        coordinator_term: 0,
        coordinator_id: None,
        lease_until_unix_ms: 0,
    }
}

fn stage_control_response_to_proto(
    response: crate::inference::skippy::StageControlResponse,
    status_list_supported: bool,
) -> skippy_stage_proto::StageControlResponse {
    use skippy_stage_proto::stage_control_response::Response;

    let response = match response {
        crate::inference::skippy::StageControlResponse::ClaimAccepted(accepted) => {
            Response::CoordinatorClaimAccepted(skippy_stage_proto::CoordinatorClaimAccepted {
                accepted: accepted.accepted,
                claim: Some(stage_coordinator_claim_to_proto(accepted.claim)),
                error: accepted.error,
            })
        }
        crate::inference::skippy::StageControlResponse::Ready(ready) => {
            Response::StageReady(skippy_stage_proto::StageReady {
                accepted: ready.accepted,
                status: Some(stage_status_to_proto(ready.status)),
                error: ready.error,
            })
        }
        crate::inference::skippy::StageControlResponse::Status(statuses) => {
            if status_list_supported {
                Response::StageStatuses(skippy_stage_proto::StageStatusList {
                    statuses: statuses.into_iter().map(stage_status_to_proto).collect(),
                })
            } else {
                Response::StageStatus(statuses.into_iter().next().map_or_else(
                    || skippy_stage_proto::StageStatus {
                        state: skippy_stage_proto::StageRuntimeState::Stopped as i32,
                        ..Default::default()
                    },
                    stage_status_to_proto,
                ))
            }
        }
        crate::inference::skippy::StageControlResponse::Inventory(inventory) => {
            Response::LayerInventory(layer_inventory_to_proto(inventory))
        }
        crate::inference::skippy::StageControlResponse::PrepareAccepted(accepted) => {
            Response::PrepareStageAccepted(skippy_stage_proto::PrepareStageAccepted {
                accepted: accepted.accepted,
                status: Some(stage_preparation_status_to_proto(accepted.status)),
                error: accepted.error,
            })
        }
        crate::inference::skippy::StageControlResponse::PreparationStatus(status) => {
            Response::StagePreparationStatus(stage_preparation_status_to_proto(status))
        }
        crate::inference::skippy::StageControlResponse::StatusAck(ack) => {
            Response::StageStatusAck(skippy_stage_proto::StageStatusAck {
                accepted: ack.accepted,
                error: ack.error,
            })
        }
    };

    skippy_stage_proto::StageControlResponse {
        gen: skippy_protocol::STAGE_PROTOCOL_GENERATION,
        response: Some(response),
    }
}

fn stage_control_response_from_proto(
    frame: skippy_stage_proto::StageControlResponse,
) -> anyhow::Result<crate::inference::skippy::StageControlResponse> {
    use skippy_stage_proto::stage_control_response::Response;

    match frame
        .response
        .ok_or_else(|| anyhow::anyhow!("missing stage control response"))?
    {
        Response::CoordinatorClaimAccepted(accepted) => {
            let claim = accepted
                .claim
                .ok_or_else(|| anyhow::anyhow!("coordinator claim accepted missing claim"))?;
            Ok(
                crate::inference::skippy::StageControlResponse::ClaimAccepted(
                    crate::inference::skippy::StageCoordinatorClaimAck {
                        accepted: accepted.accepted,
                        claim: stage_coordinator_claim_from_proto(claim)?,
                        error: accepted.error,
                    },
                ),
            )
        }
        Response::StageReady(ready) => {
            let status = ready
                .status
                .ok_or_else(|| anyhow::anyhow!("stage ready missing status"))?;
            Ok(crate::inference::skippy::StageControlResponse::Ready(
                crate::inference::skippy::StageReadyResponse {
                    accepted: ready.accepted,
                    status: stage_status_from_proto(status)?,
                    error: ready.error,
                },
            ))
        }
        Response::StageStatus(status) => {
            Ok(crate::inference::skippy::StageControlResponse::Status(
                vec![stage_status_from_proto(status)?],
            ))
        }
        Response::StageStatuses(statuses) => {
            Ok(crate::inference::skippy::StageControlResponse::Status(
                statuses
                    .statuses
                    .into_iter()
                    .map(stage_status_from_proto)
                    .collect::<anyhow::Result<Vec<_>>>()?,
            ))
        }
        Response::LayerInventory(inventory) => {
            Ok(crate::inference::skippy::StageControlResponse::Inventory(
                layer_inventory_from_proto(inventory),
            ))
        }
        Response::PrepareStageAccepted(accepted) => {
            let status = accepted
                .status
                .ok_or_else(|| anyhow::anyhow!("prepare stage accepted missing status"))?;
            Ok(
                crate::inference::skippy::StageControlResponse::PrepareAccepted(
                    crate::inference::skippy::StagePrepareAcceptedResponse {
                        accepted: accepted.accepted,
                        status: stage_preparation_status_from_proto(status),
                        error: accepted.error,
                    },
                ),
            )
        }
        Response::StagePreparationStatus(status) => Ok(
            crate::inference::skippy::StageControlResponse::PreparationStatus(
                stage_preparation_status_from_proto(status),
            ),
        ),
        Response::StageStatusAck(ack) => {
            Ok(crate::inference::skippy::StageControlResponse::StatusAck(
                crate::inference::skippy::StageStatusAck {
                    accepted: ack.accepted,
                    error: ack.error,
                },
            ))
        }
    }
}

fn layer_inventory_to_proto(
    inventory: crate::inference::skippy::StageLayerInventory,
) -> skippy_stage_proto::LayerInventory {
    skippy_stage_proto::LayerInventory {
        model_id: inventory.model_id,
        package_ref: inventory.package_ref,
        manifest_sha256: inventory.manifest_sha256,
        layer_count: inventory.layer_count,
        ready_ranges: inventory
            .ready_ranges
            .into_iter()
            .map(layer_range_to_proto)
            .collect(),
        available_ranges: inventory
            .available_ranges
            .into_iter()
            .map(layer_range_to_proto)
            .collect(),
        missing_ranges: inventory
            .missing_ranges
            .into_iter()
            .map(layer_range_to_proto)
            .collect(),
        preparing_ranges: inventory
            .preparing_ranges
            .into_iter()
            .map(stage_preparation_status_to_proto)
            .collect(),
        source_model_path: inventory.source_model_path,
        source_model_bytes: inventory.source_model_bytes,
        source_model_kind: source_model_kind_to_proto(inventory.source_model_kind) as i32,
    }
}

fn layer_inventory_from_proto(
    inventory: skippy_stage_proto::LayerInventory,
) -> crate::inference::skippy::StageLayerInventory {
    crate::inference::skippy::StageLayerInventory {
        model_id: inventory.model_id,
        package_ref: inventory.package_ref,
        manifest_sha256: inventory.manifest_sha256,
        layer_count: inventory.layer_count,
        ready_ranges: inventory
            .ready_ranges
            .into_iter()
            .map(layer_range_from_proto)
            .collect(),
        available_ranges: inventory
            .available_ranges
            .into_iter()
            .map(layer_range_from_proto)
            .collect(),
        missing_ranges: inventory
            .missing_ranges
            .into_iter()
            .map(layer_range_from_proto)
            .collect(),
        preparing_ranges: inventory
            .preparing_ranges
            .into_iter()
            .map(stage_preparation_status_from_proto)
            .collect(),
        source_model_path: inventory.source_model_path,
        source_model_bytes: inventory.source_model_bytes,
        source_model_kind: source_model_kind_from_proto(inventory.source_model_kind),
    }
}

fn layer_range_to_proto(
    range: crate::inference::skippy::LayerRange,
) -> skippy_stage_proto::LayerRange {
    skippy_stage_proto::LayerRange {
        layer_start: range.layer_start,
        layer_end: range.layer_end,
    }
}

fn layer_range_from_proto(
    range: skippy_stage_proto::LayerRange,
) -> crate::inference::skippy::LayerRange {
    crate::inference::skippy::LayerRange {
        layer_start: range.layer_start,
        layer_end: range.layer_end,
    }
}

fn source_model_kind_to_proto(
    kind: crate::inference::skippy::SourceModelKind,
) -> skippy_stage_proto::SourceModelKind {
    match kind {
        crate::inference::skippy::SourceModelKind::Unknown => {
            skippy_stage_proto::SourceModelKind::Unspecified
        }
        crate::inference::skippy::SourceModelKind::LayerPackage => {
            skippy_stage_proto::SourceModelKind::LayerPackage
        }
        crate::inference::skippy::SourceModelKind::PlainGguf => {
            skippy_stage_proto::SourceModelKind::PlainGguf
        }
        crate::inference::skippy::SourceModelKind::SplitGguf => {
            skippy_stage_proto::SourceModelKind::SplitGguf
        }
    }
}

fn source_model_kind_from_proto(value: i32) -> crate::inference::skippy::SourceModelKind {
    match skippy_stage_proto::SourceModelKind::try_from(value)
        .unwrap_or(skippy_stage_proto::SourceModelKind::Unspecified)
    {
        skippy_stage_proto::SourceModelKind::Unspecified => {
            crate::inference::skippy::SourceModelKind::Unknown
        }
        skippy_stage_proto::SourceModelKind::LayerPackage => {
            crate::inference::skippy::SourceModelKind::LayerPackage
        }
        skippy_stage_proto::SourceModelKind::PlainGguf => {
            crate::inference::skippy::SourceModelKind::PlainGguf
        }
        skippy_stage_proto::SourceModelKind::SplitGguf => {
            crate::inference::skippy::SourceModelKind::SplitGguf
        }
    }
}

fn stage_preparation_status_to_proto(
    status: crate::inference::skippy::StagePreparationStatus,
) -> skippy_stage_proto::StagePreparationStatus {
    skippy_stage_proto::StagePreparationStatus {
        topology_id: status.topology_id,
        run_id: status.run_id,
        model_id: status.model_id,
        backend: status.backend,
        package_ref: status.package_ref,
        manifest_sha256: status.manifest_sha256,
        stage_id: status.stage_id,
        stage_index: status.stage_index,
        layer_start: status.layer_start,
        layer_end: status.layer_end,
        state: stage_preparation_state_to_proto(status.state) as i32,
        bytes_done: status.bytes_done,
        bytes_total: status.bytes_total,
        bind_addr: status.bind_addr,
        error: status.error,
        shutdown_generation: status.shutdown_generation,
        coordinator_term: status.coordinator_term,
        coordinator_id: status.coordinator_id.map(|id| id.to_string()),
        lease_until_unix_ms: status.lease_until_unix_ms,
    }
}

fn stage_preparation_status_from_proto(
    status: skippy_stage_proto::StagePreparationStatus,
) -> crate::inference::skippy::StagePreparationStatus {
    let coordinator_id = status.coordinator_id.and_then(|id| match id.parse() {
        Ok(id) => Some(id),
        Err(error) => {
            tracing::warn!(
                coordinator_id = %id,
                error = %error,
                "invalid stage preparation coordinator_id"
            );
            None
        }
    });
    crate::inference::skippy::StagePreparationStatus {
        topology_id: status.topology_id,
        run_id: status.run_id,
        model_id: status.model_id,
        backend: status.backend,
        package_ref: status.package_ref,
        manifest_sha256: status.manifest_sha256,
        stage_id: status.stage_id,
        stage_index: status.stage_index,
        layer_start: status.layer_start,
        layer_end: status.layer_end,
        state: stage_preparation_state_from_proto(status.state),
        bytes_done: status.bytes_done,
        bytes_total: status.bytes_total,
        bind_addr: status.bind_addr,
        error: status.error,
        shutdown_generation: status.shutdown_generation,
        coordinator_term: status.coordinator_term,
        coordinator_id,
        lease_until_unix_ms: status.lease_until_unix_ms,
    }
}

fn stage_status_to_proto(
    status: crate::inference::skippy::StageStatusSnapshot,
) -> skippy_stage_proto::StageStatus {
    skippy_stage_proto::StageStatus {
        topology_id: status.topology_id,
        run_id: status.run_id,
        model_id: status.model_id,
        backend: status.backend,
        stage_id: status.stage_id,
        stage_index: status.stage_index,
        layer_start: status.layer_start,
        layer_end: status.layer_end,
        state: stage_runtime_state_to_proto(status.state) as i32,
        bind_addr: status.bind_addr,
        activation_width: status.activation_width,
        wire_dtype: stage_wire_dtype_to_proto(status.wire_dtype) as i32,
        error: status.error,
        shutdown_generation: status.shutdown_generation,
        selected_device: status.selected_device.map(stage_device_to_proto),
        ctx_size: status.ctx_size,
        lane_count: status.lane_count,
        n_batch: status.n_batch,
        n_ubatch: status.n_ubatch,
        package_ref: status.package_ref,
        manifest_sha256: status.manifest_sha256,
        source_model_path: status.source_model_path,
        source_model_sha256: status.source_model_sha256,
        source_model_bytes: status.source_model_bytes,
        materialized_path: status.materialized_path,
        materialized_pinned: Some(status.materialized_pinned),
        projector_path: status.projector_path,
        flash_attn_type: stage_flash_attn_type_to_proto(status.flash_attn_type) as i32,
        coordinator_term: status.coordinator_term,
        coordinator_id: status.coordinator_id.map(|id| id.to_string()),
        lease_until_unix_ms: status.lease_until_unix_ms,
    }
}

fn stage_status_from_proto(
    status: skippy_stage_proto::StageStatus,
) -> anyhow::Result<crate::inference::skippy::StageStatusSnapshot> {
    Ok(crate::inference::skippy::StageStatusSnapshot {
        topology_id: status.topology_id,
        run_id: status.run_id,
        model_id: status.model_id,
        backend: status.backend,
        stage_id: status.stage_id,
        stage_index: status.stage_index,
        layer_start: status.layer_start,
        layer_end: status.layer_end,
        state: stage_runtime_state_from_proto(status.state),
        bind_addr: status.bind_addr,
        activation_width: status.activation_width,
        wire_dtype: stage_wire_dtype_from_proto(status.wire_dtype),
        selected_device: status
            .selected_device
            .map(stage_device_from_proto)
            .transpose()?,
        ctx_size: status.ctx_size,
        lane_count: if status.lane_count == 0 {
            4
        } else {
            status.lane_count
        },
        n_batch: status.n_batch,
        n_ubatch: status.n_ubatch,
        package_ref: status.package_ref,
        manifest_sha256: status.manifest_sha256,
        source_model_path: status.source_model_path,
        source_model_sha256: status.source_model_sha256,
        source_model_bytes: status.source_model_bytes,
        materialized_path: status.materialized_path,
        materialized_pinned: status.materialized_pinned.unwrap_or(false),
        projector_path: status.projector_path,
        flash_attn_type: stage_flash_attn_type_from_proto(status.flash_attn_type),
        error: status.error,
        shutdown_generation: status.shutdown_generation,
        coordinator_term: status.coordinator_term,
        coordinator_id: status
            .coordinator_id
            .map(|id| id.parse())
            .transpose()
            .context("invalid stage status coordinator_id")?,
        lease_until_unix_ms: status.lease_until_unix_ms,
    })
}

fn stage_flash_attn_type_to_proto(
    value: skippy_protocol::FlashAttentionType,
) -> skippy_stage_proto::StageFlashAttnType {
    match value {
        skippy_protocol::FlashAttentionType::Auto => skippy_stage_proto::StageFlashAttnType::Auto,
        skippy_protocol::FlashAttentionType::Disabled => {
            skippy_stage_proto::StageFlashAttnType::Disabled
        }
        skippy_protocol::FlashAttentionType::Enabled => {
            skippy_stage_proto::StageFlashAttnType::Enabled
        }
    }
}

fn stage_flash_attn_type_from_proto(value: i32) -> skippy_protocol::FlashAttentionType {
    match skippy_stage_proto::StageFlashAttnType::try_from(value)
        .unwrap_or(skippy_stage_proto::StageFlashAttnType::Unspecified)
    {
        skippy_stage_proto::StageFlashAttnType::Unspecified
        | skippy_stage_proto::StageFlashAttnType::Auto => skippy_protocol::FlashAttentionType::Auto,
        skippy_stage_proto::StageFlashAttnType::Disabled => {
            skippy_protocol::FlashAttentionType::Disabled
        }
        skippy_stage_proto::StageFlashAttnType::Enabled => {
            skippy_protocol::FlashAttentionType::Enabled
        }
    }
}

fn stage_runtime_state_from_proto(value: i32) -> crate::inference::skippy::StageRuntimeState {
    match skippy_stage_proto::StageRuntimeState::try_from(value)
        .unwrap_or(skippy_stage_proto::StageRuntimeState::Failed)
    {
        skippy_stage_proto::StageRuntimeState::Starting => {
            crate::inference::skippy::StageRuntimeState::Starting
        }
        skippy_stage_proto::StageRuntimeState::Ready => {
            crate::inference::skippy::StageRuntimeState::Ready
        }
        skippy_stage_proto::StageRuntimeState::Stopping => {
            crate::inference::skippy::StageRuntimeState::Stopping
        }
        skippy_stage_proto::StageRuntimeState::Stopped
        | skippy_stage_proto::StageRuntimeState::Unspecified => {
            crate::inference::skippy::StageRuntimeState::Stopped
        }
        skippy_stage_proto::StageRuntimeState::Failed => {
            crate::inference::skippy::StageRuntimeState::Failed
        }
    }
}

fn stage_runtime_state_to_proto(
    state: crate::inference::skippy::StageRuntimeState,
) -> skippy_stage_proto::StageRuntimeState {
    match state {
        crate::inference::skippy::StageRuntimeState::Starting => {
            skippy_stage_proto::StageRuntimeState::Starting
        }
        crate::inference::skippy::StageRuntimeState::Ready => {
            skippy_stage_proto::StageRuntimeState::Ready
        }
        crate::inference::skippy::StageRuntimeState::Stopping => {
            skippy_stage_proto::StageRuntimeState::Stopping
        }
        crate::inference::skippy::StageRuntimeState::Stopped => {
            skippy_stage_proto::StageRuntimeState::Stopped
        }
        crate::inference::skippy::StageRuntimeState::Failed => {
            skippy_stage_proto::StageRuntimeState::Failed
        }
    }
}

fn stage_preparation_state_from_proto(
    value: i32,
) -> crate::inference::skippy::StagePreparationState {
    match skippy_stage_proto::StagePreparationState::try_from(value)
        .unwrap_or(skippy_stage_proto::StagePreparationState::Unspecified)
    {
        skippy_stage_proto::StagePreparationState::Assigned
        | skippy_stage_proto::StagePreparationState::Unspecified => {
            crate::inference::skippy::StagePreparationState::Assigned
        }
        skippy_stage_proto::StagePreparationState::Downloading => {
            crate::inference::skippy::StagePreparationState::Downloading
        }
        skippy_stage_proto::StagePreparationState::Available => {
            crate::inference::skippy::StagePreparationState::Available
        }
        skippy_stage_proto::StagePreparationState::Resolving => {
            crate::inference::skippy::StagePreparationState::Resolving
        }
        skippy_stage_proto::StagePreparationState::Loading => {
            crate::inference::skippy::StagePreparationState::Loading
        }
        skippy_stage_proto::StagePreparationState::Ready => {
            crate::inference::skippy::StagePreparationState::Ready
        }
        skippy_stage_proto::StagePreparationState::Failed => {
            crate::inference::skippy::StagePreparationState::Failed
        }
        skippy_stage_proto::StagePreparationState::Cancelled => {
            crate::inference::skippy::StagePreparationState::Cancelled
        }
    }
}

fn stage_preparation_state_to_proto(
    state: crate::inference::skippy::StagePreparationState,
) -> skippy_stage_proto::StagePreparationState {
    match state {
        crate::inference::skippy::StagePreparationState::Assigned => {
            skippy_stage_proto::StagePreparationState::Assigned
        }
        crate::inference::skippy::StagePreparationState::Downloading => {
            skippy_stage_proto::StagePreparationState::Downloading
        }
        crate::inference::skippy::StagePreparationState::Available => {
            skippy_stage_proto::StagePreparationState::Available
        }
        crate::inference::skippy::StagePreparationState::Resolving => {
            skippy_stage_proto::StagePreparationState::Resolving
        }
        crate::inference::skippy::StagePreparationState::Loading => {
            skippy_stage_proto::StagePreparationState::Loading
        }
        crate::inference::skippy::StagePreparationState::Ready => {
            skippy_stage_proto::StagePreparationState::Ready
        }
        crate::inference::skippy::StagePreparationState::Failed => {
            skippy_stage_proto::StagePreparationState::Failed
        }
        crate::inference::skippy::StagePreparationState::Cancelled => {
            skippy_stage_proto::StagePreparationState::Cancelled
        }
    }
}

fn stage_wire_dtype_to_proto(
    dtype: crate::inference::skippy::StageWireDType,
) -> skippy_stage_proto::StageWireDType {
    match dtype {
        crate::inference::skippy::StageWireDType::F32 => {
            skippy_stage_proto::StageWireDType::StageWireDtypeF32
        }
        crate::inference::skippy::StageWireDType::F16 => {
            skippy_stage_proto::StageWireDType::StageWireDtypeF16
        }
        crate::inference::skippy::StageWireDType::Q8 => {
            skippy_stage_proto::StageWireDType::StageWireDtypeQ8
        }
    }
}

/// Generate a mesh ID for a new mesh.
/// Named meshes: `sha256("mesh-llm:" + name + ":" + nostr_pubkey)` — deterministic, unique per creator.
/// Unnamed meshes: random UUID, persisted to `~/.mesh-llm/mesh-id`.
pub fn generate_mesh_id(name: Option<&str>, nostr_pubkey: Option<&str>) -> String {
    if let Some(name) = name {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        "mesh-llm:".hash(&mut hasher);
        name.hash(&mut hasher);
        if let Some(pk) = nostr_pubkey {
            pk.hash(&mut hasher);
        }
        format!("{:016x}", hasher.finish())
    } else {
        // Try to load persisted mesh-id
        let path = mesh_id_path();
        if let Ok(id) = std::fs::read_to_string(&path) {
            let id = id.trim().to_string();
            if !id.is_empty() {
                return id;
            }
        }
        // Generate new random ID and persist
        let id = format!(
            "{:016x}{:016x}",
            rand::random::<u64>(),
            rand::random::<u64>()
        );
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let _ = std::fs::write(&path, &id);
        id
    }
}

fn mesh_id_path() -> std::path::PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| std::path::PathBuf::from("."))
        .join(".mesh-llm")
        .join("mesh-id")
}

/// Save the mesh ID of the last mesh we successfully joined.
pub fn save_last_mesh_id(mesh_id: &str) {
    let path = dirs::home_dir()
        .unwrap_or_else(|| std::path::PathBuf::from("."))
        .join(".mesh-llm")
        .join("last-mesh");
    if let Some(parent) = path.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    let _ = std::fs::write(&path, mesh_id);
}

/// Load the mesh ID of the last mesh we successfully joined.
pub fn load_last_mesh_id() -> Option<String> {
    let path = dirs::home_dir()
        .unwrap_or_else(|| std::path::PathBuf::from("."))
        .join(".mesh-llm")
        .join("last-mesh");
    std::fs::read_to_string(&path)
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

// ---------------------------------------------------------------------------
// Public-to-private identity transition
// ---------------------------------------------------------------------------

fn was_public_path() -> std::path::PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| std::path::PathBuf::from("."))
        .join(".mesh-llm")
        .join("was-public")
}

fn clear_public_identity_file(path: &std::path::Path) -> bool {
    if !path.exists() {
        return true;
    }
    match std::fs::remove_file(path) {
        Ok(()) => {
            tracing::info!("Cleared {}", path.display());
            true
        }
        Err(_) => {
            tracing::warn!("Failed to clear {}", path.display());
            false
        }
    }
}

/// Record that this node was started in public mode (--auto / --publish / --mesh-name).
/// Called at startup so we can detect a public→private transition next time.
pub fn mark_was_public() {
    let path = was_public_path();
    if let Some(parent) = path.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    let _ = std::fs::write(&path, "1");
}

/// Returns true if the previous run was public (marker file exists).
pub fn was_previously_public() -> bool {
    was_public_path().exists()
}

/// Clear identity files (key, nostr.nsec, mesh-id, last-mesh, was-public) so the
/// next start gets a completely fresh identity. Called when transitioning from
/// public → private to avoid reusing a publicly-known identity in a private mesh.
pub fn clear_public_identity() {
    let home = dirs::home_dir().unwrap_or_else(|| std::path::PathBuf::from("."));
    let dir = home.join(".mesh-llm");
    let mut ok = true;
    for name in &["key", "nostr.nsec", "mesh-id", "last-mesh"] {
        ok &= clear_public_identity_file(&dir.join(name));
    }
    // Only remove the marker after identity files are gone, so a failed
    // cleanup is retried on the next private start.
    let marker = dir.join("was-public");
    if ok {
        let _ = std::fs::remove_file(&marker);
    } else {
        tracing::warn!("Keeping was-public marker — will retry cleanup next start");
    }
}

/// Load secret key from ~/.mesh-llm/key, or create a new one and save it.
async fn load_or_create_key() -> Result<SecretKey> {
    let key_path = default_node_key_path()?;
    let dir = key_path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("Invalid node key path {}", key_path.display()))?;
    ensure_private_node_key_dir(dir)?;

    if key_path.exists() {
        ensure_private_node_key_file(&key_path)?;
        let hex = tokio::fs::read_to_string(&key_path).await?;
        let bytes = hex::decode(hex.trim())?;
        if bytes.len() != 32 {
            anyhow::bail!("Invalid key length in {}", key_path.display());
        }
        let key = SecretKey::from_bytes(&bytes.try_into().unwrap());
        tracing::info!("Loaded key from {}", key_path.display());
        return Ok(key);
    }

    let key = SecretKey::generate();
    save_node_key_to_path(&key_path, &key)?;
    tracing::info!("Generated new key, saved to {}", key_path.display());
    Ok(key)
}

pub fn default_node_key_path() -> Result<std::path::PathBuf> {
    let home =
        dirs::home_dir().ok_or_else(|| anyhow::anyhow!("Cannot determine home directory"))?;
    Ok(home.join(".mesh-llm").join("key"))
}

pub fn load_node_key_from_path(path: &std::path::Path) -> Result<SecretKey> {
    let hex = std::fs::read_to_string(path)?;
    let bytes = hex::decode(hex.trim())?;
    if bytes.len() != 32 {
        anyhow::bail!("Invalid key length in {}", path.display());
    }
    Ok(SecretKey::from_bytes(&bytes.try_into().unwrap()))
}

pub fn save_node_key_to_path(path: &std::path::Path, key: &SecretKey) -> Result<()> {
    let parent = path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("Invalid node key path {}", path.display()))?;
    ensure_private_node_key_dir(parent)?;
    if path.exists() {
        ensure_private_node_key_file(path)?;
    }
    crate::crypto::write_keystore_bytes_atomically(path, hex::encode(key.to_bytes()).as_bytes())?;
    ensure_private_node_key_file(path)?;
    Ok(())
}

#[cfg(unix)]
fn ensure_private_node_key_dir(dir: &std::path::Path) -> Result<()> {
    use std::os::unix::fs::PermissionsExt;

    std::fs::create_dir_all(dir)?;
    let metadata = std::fs::metadata(dir)?;
    let mut perms = metadata.permissions();
    if perms.mode() & 0o077 != 0 {
        perms.set_mode(0o700);
        std::fs::set_permissions(dir, perms)?;
    }
    Ok(())
}

#[cfg(not(unix))]
fn ensure_private_node_key_dir(dir: &std::path::Path) -> Result<()> {
    std::fs::create_dir_all(dir)?;
    Ok(())
}

#[cfg(unix)]
fn ensure_private_node_key_file(path: &std::path::Path) -> Result<()> {
    use std::os::unix::fs::PermissionsExt;

    let metadata = std::fs::symlink_metadata(path)?;
    if !metadata.file_type().is_file() {
        anyhow::bail!("Node key path is not a regular file");
    }
    let mut perms = metadata.permissions();
    if perms.mode() & 0o077 != 0 {
        perms.set_mode(0o600);
        std::fs::set_permissions(path, perms)?;
    }
    Ok(())
}

#[cfg(not(unix))]
fn ensure_private_node_key_file(path: &std::path::Path) -> Result<()> {
    let metadata = std::fs::symlink_metadata(path)?;
    if !metadata.file_type().is_file() {
        anyhow::bail!("Node key path is not a regular file");
    }
    Ok(())
}

mod gossip;
mod heartbeat;
pub use gossip::backfill_legacy_descriptors;
#[allow(unused_imports)]
use gossip::{apply_transitive_ann, peer_meaningfully_changed};
#[allow(unused_imports)]
use heartbeat::{heartbeat_failure_policy_for_peer, HeartbeatFailurePolicy};
pub(crate) use heartbeat::{
    peer_down_report_disposition, resolve_peer_down, PeerDownReportDisposition,
};

#[cfg(test)]
mod tests;

#[cfg(test)]
mod public_identity_tests;
