geth/crates/geth-node/src/lib.rs

pub mod service;

use base64::Engine;
use geth_auth::{AuthExplanation, AuthOp, AuthOpKind};
use geth_cas::{
    BlobInfoSummary, FileConflict, FileConflictKind, FileConflictResolution, FileConflictStatus,
    FileRoot, FileRootScan, LocalCas, hash_path,
};
use geth_config::{GethConfig, GethPaths, RelayMode};
use geth_control::{
    CasBlob, ControlRequest, ControlResponse, KeychainStatusResponse, NodeIdResponse,
    PeerControlRequest, PeerControlResponse, StatusResponse,
};
use geth_crypto::AgentKey;
use geth_db::DbResource;
use geth_discovery::{
    DiscoveredPeer, DiscoverySource, EndpointCandidate, PEER_CARD_LAN_DISCOVERY_SERVICE, PeerCard,
    discovery_is_untrusted_note, peer_card_from_txt_attributes, peer_card_txt_attributes,
};
use geth_document::{DocumentResource, DocumentState};
use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode};
use geth_keychain::{KeychainOp, KeychainOpKind};
use geth_kv::{KvEntry, KvResource};
use geth_pipe::{PipeConnection, PipeListener};
use geth_pubsub::PubsubMessage;
use geth_resource::ResourceDescriptor;
use geth_secrets::{BearerAccess, ResourceMasterSecret};
use geth_ssh_identity::{
    SshCertApproval, SshCertKind, SshCertRequest, SshCertRequestStatus, SshCertificateRecord,
    SshRevocationEntry, SshRevocationExportFormat, SshRevocationKind, build_ssh_cert_sign_command,
    cert_request_id, certificate_id, openssh_krl_spec, parse_openssh_krl_spec, revocation_id,
    ssh_public_key_fingerprint, write_openssh_krl,
};
use geth_store::{
    Store, StoredAuthOp, StoredDbResource, StoredDocumentResource, StoredFileConflict,
    StoredFileRoot, StoredKeychainOp, StoredKvEntry, StoredKvStore, StoredModuleState,
    StoredPeerCard, StoredResource, StoredResourceSecret, StoredSshCertRequest,
    StoredSshCertificate, StoredSshRevocation,
};
use geth_types::{
    AuthOpId, BlobHash, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind,
    ResourceName, SshCertId, SshCertRequestId, UnixMillis,
};
use std::collections::{BTreeMap, VecDeque};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};

#[derive(Debug, thiserror::Error)]
pub enum NodeError {
    #[error("config error: {0}")]
    Config(#[from] geth_config::ConfigError),
    #[error("crypto error: {0}")]
    Crypto(#[from] geth_crypto::CryptoError),
    #[error("store error: {0}")]
    Store(#[from] geth_store::StoreError),
    #[error("cas error: {0}")]
    Cas(#[from] geth_cas::CasError),
    #[error("db error: {0}")]
    Db(#[from] geth_db::DbError),
    #[error("control error: {0}")]
    Control(#[from] geth_control::ControlError),
    #[error("json error: {0}")]
    Json(#[from] serde_json::Error),
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("invalid resource kind: {0}")]
    InvalidResourceKind(String),
    #[error("invalid db resource name: {0}")]
    InvalidDbName(String),
    #[error("db path does not exist or is not a file: {0}")]
    InvalidDbPath(String),
    #[error("db resource not found: {0}")]
    DbNotFound(String),
    #[error("invalid kv store name: {0}")]
    InvalidKvName(String),
    #[error("invalid kv key: {0}")]
    InvalidKvKey(String),
    #[error("kv store not found: {0}")]
    KvNotFound(String),
    #[error("invalid document name: {0}")]
    InvalidDocumentName(String),
    #[error("document not found: {0}")]
    DocumentNotFound(String),
    #[error("document error: {0}")]
    Document(#[from] geth_document::DocumentError),
    #[error("resource not found: {0}")]
    ResourceNotFound(String),
    #[error("unauthorized: {0}")]
    Unauthorized(String),
    #[error("secrets error: {0}")]
    Secrets(#[from] geth_secrets::SecretsError),
    #[error("pubsub error: {0}")]
    Pubsub(#[from] geth_pubsub::PubsubError),
    #[error("pipe error: {0}")]
    Pipe(#[from] geth_pipe::PipeError),
    #[error("runtime state lock poisoned")]
    RuntimeLockPoisoned,
    #[error("invalid ssh certificate kind: {0}")]
    InvalidSshCertKind(String),
    #[error("invalid ssh certificate request status: {0}")]
    InvalidSshCertStatus(String),
    #[error("invalid ssh revocation kind: {0}")]
    InvalidSshRevocationKind(String),
    #[error("ssh certificate request not found: {0}")]
    SshCertRequestNotFound(String),
    #[error("ssh certificate request must include at least one principal")]
    MissingSshCertPrincipal,
    #[error("ssh certificate flow error: {0}")]
    SshCertFlow(#[from] geth_ssh_identity::SshCertFlowError),
    #[error("ssh identity error: {0}")]
    SshIdentity(#[from] geth_ssh_identity::SshIdentityError),
    #[error("discovery error: {0}")]
    Discovery(#[from] geth_discovery::DiscoveryError),
    #[error("cannot export peer card before the daemon has an Iroh EndpointID")]
    IrohEndpointUnavailable,
    #[error("peer candidate not found: {0}")]
    PeerNotFound(String),
    #[error("iroh peer error: {0}")]
    IrohPeer(String),
}

#[derive(Clone, Debug)]
pub struct LocalNode {
    pub paths: GethPaths,
    pub agent_id: String,
    pub node_id: String,
    pub iroh_status: EndpointStatus,
    iroh_endpoint: Arc<Mutex<Option<GethIrohEndpoint>>>,
    runtime: Arc<NodeRuntime>,
}

#[derive(Debug)]
struct NodeRuntime {
    pubsub: Mutex<PubsubRuntime>,
    pipes: Mutex<PipeRuntime>,
}

#[derive(Debug, Default)]
struct PubsubRuntime {
    messages: VecDeque<PubsubMessage>,
}

#[derive(Debug, Default)]
struct PipeRuntime {
    listeners: BTreeMap<String, PipeListener>,
    connections: VecDeque<PipeConnection>,
}

const PUBSUB_RING_LIMIT: usize = 256;
const LIVE_SYNC_INTERVAL: Duration = Duration::from_secs(30);
const PIPE_CONNECTION_RING_LIMIT: usize = 256;

#[derive(Debug, serde::Deserialize, serde::Serialize)]
struct LiveSyncCursor {
    cursor_ms: i64,
}

pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
    paths.ensure_base_dirs()?;
    if !paths.config_file().exists() {
        std::fs::write(paths.config_file(), GethConfig::default_toml())?;
    }
    let key = AgentKey::load_or_create(&paths.agent_key())?;
    let agent_id = key.agent_id().to_string();
    let node_id = stable_node_id(&agent_id);
    let store = Store::open(&paths.metadata_db())?;
    store.upsert_agent(&agent_id, &key.public_key_hex())?;
    store.upsert_node(&node_id, "local", &agent_id)?;
    store.insert_resource(&StoredResource {
        resource_id: "resource:cas:local".to_owned(),
        kind: ResourceKind::Cas.to_string(),
        name: "local-cas".to_owned(),
        status: "active".to_owned(),
    })?;
    Ok(LocalNode {
        paths: paths.clone(),
        agent_id,
        node_id,
        iroh_status: EndpointStatus::scaffolded(),
        iroh_endpoint: Arc::new(Mutex::new(None)),
        runtime: Arc::new(NodeRuntime {
            pubsub: Mutex::new(PubsubRuntime::default()),
            pipes: Mutex::new(PipeRuntime::default()),
        }),
    })
}

pub fn open_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
    init_node(paths)
}

pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> {
    let mut node = init_node(&paths)?;
    let _iroh_endpoint = start_daemon_iroh_endpoint(&mut node).await?;
    if let Some(endpoint) = node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .clone()
    {
        spawn_iroh_control_accept_loop(node.clone(), endpoint);
        spawn_background_live_sync(node.clone());
    }
    let _peer_card_lan_discovery = start_peer_card_lan_discovery(&node).await;
    if Path::new(&paths.socket_path()).exists() {
        std::fs::remove_file(paths.socket_path())?;
    }
    let listener = UnixListener::bind(paths.socket_path())?;
    tracing::info!(socket = %paths.socket_path().display(), "geth daemon listening");

    loop {
        let (stream, _) = listener.accept().await?;
        let node = node.clone();
        tokio::spawn(async move {
            if let Err(error) = handle_stream(node, stream).await {
                tracing::warn!(%error, "control request failed");
            }
        });
    }
}

pub async fn send_control(
    paths: &GethPaths,
    request: ControlRequest,
) -> Result<ControlResponse, NodeError> {
    let mut stream = UnixStream::connect(paths.socket_path()).await?;
    stream
        .write_all(geth_control::encode_request(&request)?.as_bytes())
        .await?;
    stream.shutdown().await?;
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    Ok(geth_control::decode_response(&line)?)
}

pub async fn handle_request_async(
    node: &LocalNode,
    request: ControlRequest,
) -> Result<ControlResponse, NodeError> {
    match request {
        ControlRequest::PeerCardExport { out } => export_peer_card(node, out, true).await,
        ControlRequest::PeerPing { node: peer_node } => peer_ping(node, &peer_node).await,
        ControlRequest::PeerAuthCheck {
            node: peer_node,
            resource,
            capability,
        } => peer_auth_check(node, &peer_node, resource, capability).await,
        ControlRequest::CasFetch {
            node: peer_node,
            hash,
        } => cas_fetch_from_peer(node, &peer_node, hash).await,
        ControlRequest::SshCertSync { node: peer_node } => {
            ssh_cert_sync_from_peer(node, &peer_node).await
        }
        ControlRequest::SshRevocationSync { node: peer_node } => {
            ssh_revocation_sync_from_peer(node, &peer_node).await
        }
        other => handle_request(node, other),
    }
}

async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeError> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    let request = geth_control::decode_request(&line)?;
    let response = match handle_request_async(&node, request).await {
        Ok(response) => response,
        Err(error) => ControlResponse::Error {
            message: error.to_string(),
        },
    };
    let mut stream = reader.into_inner();
    stream
        .write_all(geth_control::encode_response(&response)?.as_bytes())
        .await?;
    Ok(())
}

async fn start_peer_card_lan_discovery(node: &LocalNode) -> Option<swarm_discovery::DropGuard> {
    if !node.iroh_status.local_discovery || !node.iroh_status.enabled {
        return None;
    }

    let card = match local_peer_card(node, DiscoverySource::Mdns, true).await {
        Ok(card) => card,
        Err(error) => {
            tracing::warn!(%error, "signed peer-card LAN discovery disabled");
            return None;
        }
    };
    let attributes = match peer_card_txt_attributes(&card) {
        Ok(attributes) => attributes,
        Err(error) => {
            tracing::warn!(%error, "could not encode peer-card LAN TXT payload");
            return None;
        }
    };
    let Some((port, addrs)) = lan_discovery_addrs(&card) else {
        tracing::warn!(
            "signed peer-card LAN discovery disabled because no direct Iroh address is known"
        );
        return None;
    };

    let paths = node.paths.clone();
    let self_node_id = node.node_id.clone();
    let discoverer = match swarm_discovery::Discoverer::new_interactive(
        PEER_CARD_LAN_DISCOVERY_SERVICE.to_owned(),
        lan_discovery_peer_id(&node.agent_id),
    )
    .with_addrs(port, addrs)
    .with_txt_attributes(attributes)
    {
        Ok(discoverer) => discoverer,
        Err(error) => {
            tracing::warn!(%error, "could not build peer-card LAN discovery");
            return None;
        }
    }
    .with_callback(move |_peer_id, peer| {
        if peer.is_expiry() {
            return;
        }
        let card = match peer_card_from_txt_attributes(peer.txt_attributes()) {
            Ok(card) => card,
            Err(error) => {
                tracing::debug!(%error, "ignoring invalid peer-card LAN discovery payload");
                return;
            }
        };
        if card.node_id.as_str() == self_node_id {
            return;
        }
        let discovered = match DiscoveredPeer::candidate(
            card.clone(),
            UnixMillis(geth_store::now_ms()),
            DiscoverySource::Mdns,
        ) {
            Ok(discovered) => discovered,
            Err(error) => {
                tracing::debug!(%error, "ignoring invalid LAN peer-card candidate");
                return;
            }
        };
        match Store::open(&paths.metadata_db()).and_then(|store| {
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: card.node_id.to_string(),
                card_json: serde_json::to_string(&card).map_err(geth_store::StoreError::from)?,
                updated_at_ms: discovered.discovered_at.0,
            })
        }) {
            Ok(()) => tracing::debug!(node = %card.node_id, "stored LAN peer-card candidate"),
            Err(error) => tracing::warn!(%error, "could not store LAN peer-card candidate"),
        }
    });

    match discoverer.spawn(&tokio::runtime::Handle::current()) {
        Ok(guard) => {
            tracing::info!(
                service = PEER_CARD_LAN_DISCOVERY_SERVICE,
                "signed peer-card LAN discovery running"
            );
            Some(guard)
        }
        Err(error) => {
            tracing::warn!(%error, "could not start signed peer-card LAN discovery");
            None
        }
    }
}

fn lan_discovery_peer_id(agent_id: &str) -> String {
    format!("geth-{}", geth_crypto::blake3_hex(agent_id.as_bytes()))
}

fn lan_discovery_addrs(card: &PeerCard) -> Option<(u16, Vec<std::net::IpAddr>)> {
    let mut parsed = card
        .endpoints
        .iter()
        .flat_map(|endpoint| endpoint.direct_addresses.iter())
        .filter_map(|addr| addr.parse::<std::net::SocketAddr>().ok())
        .collect::<Vec<_>>();
    parsed.sort_unstable();
    parsed.dedup();
    let port = parsed.first()?.port();
    let addrs = parsed
        .into_iter()
        .filter(|addr| addr.port() == port)
        .map(|addr| addr.ip())
        .collect::<Vec<_>>();
    (!addrs.is_empty()).then_some((port, addrs))
}

async fn export_peer_card(
    node: &LocalNode,
    out: Option<std::path::PathBuf>,
    include_node_addr: bool,
) -> Result<ControlResponse, NodeError> {
    let card = local_peer_card(node, DiscoverySource::Manual, include_node_addr).await?;
    if let Some(path) = &out {
        std::fs::write(path, serde_json::to_string_pretty(&card)?)?;
    }
    Ok(ControlResponse::PeerCardExported {
        card,
        out,
        note: discovery_is_untrusted_note().to_owned(),
    })
}

async fn local_peer_card(
    node: &LocalNode,
    source: DiscoverySource,
    include_node_addr: bool,
) -> Result<PeerCard, NodeError> {
    let endpoint_id = node
        .iroh_status
        .endpoint_id
        .clone()
        .ok_or(NodeError::IrohEndpointUnavailable)?;
    let endpoint = node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .clone();
    let node_addr = if include_node_addr {
        match endpoint {
            Some(endpoint) => endpoint.node_addr_snapshot().await.ok(),
            None => None,
        }
    } else {
        None
    };
    let candidate = EndpointCandidate {
        endpoint_id,
        relay_url: node_addr.as_ref().and_then(|addr| addr.relay_url.clone()),
        direct_addresses: node_addr
            .map(|addr| addr.direct_addresses)
            .unwrap_or_default(),
        source,
    };
    let key = AgentKey::load(&node.paths.agent_key())?;
    PeerCard::signed(
        NodeId::new(node.node_id.clone()),
        &key,
        vec![candidate],
        UnixMillis(geth_store::now_ms()),
    )
    .map_err(NodeError::from)
}

async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result<ControlResponse, NodeError> {
    let store = Store::open(&node.paths.metadata_db())?;
    let stored = store
        .get_peer_card(peer_node)?
        .ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?;
    let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
    peer_card.validate_candidate()?;
    let candidate = peer_card
        .endpoints
        .first()
        .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
    ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
    let node_addr = iroh_node_addr_from_candidate(candidate)?;
    let endpoint = node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .clone()
        .ok_or(NodeError::IrohEndpointUnavailable)?;
    let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?;
    let nonce = geth_crypto::blake3_hex(
        format!("{}\0{}\0{}", node.node_id, peer_node, geth_store::now_ms()).as_bytes(),
    );
    let request = PeerControlRequest::Ping {
        peer_card: self_card,
        nonce: nonce.clone(),
    };

    let conn = endpoint
        .endpoint()
        .connect(node_addr, geth_iroh::ALPN_CONTROL)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let alpn = conn
        .alpn()
        .map(display_alpn)
        .unwrap_or_else(|| "unknown".to_owned());
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.write_all(geth_control::encode_peer_request(&request)?.as_bytes())
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.finish()
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let bytes = recv
        .read_to_end(64 * 1024)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let text =
        std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    match geth_control::decode_peer_response(text)? {
        PeerControlResponse::Pong {
            node_id,
            agent_id,
            endpoint_id,
            alpn: remote_alpn,
            nonce: response_nonce,
            note,
            ..
        } if response_nonce == nonce => Ok(ControlResponse::PeerPinged {
            peer_node_id: node_id,
            peer_agent_id: agent_id,
            endpoint_id,
            alpn: if remote_alpn == "unknown" {
                alpn
            } else {
                remote_alpn
            },
            note,
        }),
        PeerControlResponse::Pong { .. } => Err(NodeError::IrohPeer(
            "peer ping response nonce did not match request".to_owned(),
        )),
        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
        PeerControlResponse::AuthChecked { .. } => Err(NodeError::IrohPeer(
            "peer returned auth-check response to ping request".to_owned(),
        )),
        PeerControlResponse::CasFetched { .. }
        | PeerControlResponse::SshCertSynced { .. }
        | PeerControlResponse::SshRevocationSynced { .. } => Err(NodeError::IrohPeer(
            "peer returned wrong response type to ping request".to_owned(),
        )),
    }
}

async fn peer_auth_check(
    node: &LocalNode,
    peer_node: &str,
    resource: String,
    capability: String,
) -> Result<ControlResponse, NodeError> {
    let store = Store::open(&node.paths.metadata_db())?;
    let stored = store
        .get_peer_card(peer_node)?
        .ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?;
    let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
    peer_card.validate_candidate()?;
    let candidate = peer_card
        .endpoints
        .first()
        .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
    ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
    let node_addr = iroh_node_addr_from_candidate(candidate)?;
    let endpoint = node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .clone()
        .ok_or(NodeError::IrohEndpointUnavailable)?;
    let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?;
    let nonce = geth_crypto::blake3_hex(
        format!(
            "{}\0{}\0{}\0{}\0{}",
            node.node_id,
            peer_node,
            resource,
            capability,
            geth_store::now_ms()
        )
        .as_bytes(),
    );
    let request = PeerControlRequest::AuthCheck {
        peer_card: self_card,
        resource: resource.clone(),
        capability: capability.clone(),
        nonce: nonce.clone(),
    };

    let conn = endpoint
        .endpoint()
        .connect(node_addr, geth_iroh::ALPN_CONTROL)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.write_all(geth_control::encode_peer_request(&request)?.as_bytes())
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.finish()
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let bytes = recv
        .read_to_end(64 * 1024)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let text =
        std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    match geth_control::decode_peer_response(text)? {
        PeerControlResponse::AuthChecked {
            node_id,
            agent_id,
            endpoint_id,
            resource: response_resource,
            capability: response_capability,
            allowed,
            reason,
            evaluated_ops,
            nonce: response_nonce,
            note,
            ..
        } if response_nonce == nonce
            && response_resource == resource
            && response_capability == capability =>
        {
            Ok(ControlResponse::PeerAuthChecked {
                peer_node_id: node_id,
                peer_agent_id: agent_id,
                endpoint_id,
                resource: response_resource,
                capability: response_capability,
                allowed,
                reason,
                evaluated_ops,
                note,
            })
        }
        PeerControlResponse::AuthChecked { .. } => Err(NodeError::IrohPeer(
            "peer auth-check response did not match request".to_owned(),
        )),
        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
        PeerControlResponse::Pong { .. } => Err(NodeError::IrohPeer(
            "peer returned pong to auth-check request".to_owned(),
        )),
        PeerControlResponse::CasFetched { .. }
        | PeerControlResponse::SshCertSynced { .. }
        | PeerControlResponse::SshRevocationSynced { .. } => Err(NodeError::IrohPeer(
            "peer returned wrong response type to auth-check request".to_owned(),
        )),
    }
}

async fn cas_fetch_from_peer(
    node: &LocalNode,
    peer_node: &str,
    hash: BlobHash,
) -> Result<ControlResponse, NodeError> {
    let store = Store::open(&node.paths.metadata_db())?;
    let stored = store
        .get_peer_card(peer_node)?
        .ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?;
    let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
    peer_card.validate_candidate()?;
    let candidate = peer_card
        .endpoints
        .first()
        .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
    ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
    let node_addr = iroh_node_addr_from_candidate(candidate)?;
    let endpoint = node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .clone()
        .ok_or(NodeError::IrohEndpointUnavailable)?;
    let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?;
    let nonce = geth_crypto::blake3_hex(
        format!(
            "{}\0{}\0{}\0{}",
            node.node_id,
            peer_node,
            hash.as_str(),
            geth_store::now_ms()
        )
        .as_bytes(),
    );
    let request = PeerControlRequest::CasFetch {
        peer_card: self_card,
        hash: hash.clone(),
        nonce: nonce.clone(),
    };

    let conn = endpoint
        .endpoint()
        .connect(node_addr, geth_iroh::ALPN_CONTROL)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.write_all(geth_control::encode_peer_request(&request)?.as_bytes())
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.finish()
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let bytes = recv
        .read_to_end(64 * 1024 * 1024)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let text =
        std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    match geth_control::decode_peer_response(text)? {
        PeerControlResponse::CasFetched {
            node_id,
            agent_id,
            endpoint_id,
            hash: response_hash,
            size_bytes,
            content_base64,
            allowed,
            reason,
            nonce: response_nonce,
            note,
            ..
        } if response_nonce == nonce && response_hash == hash => {
            if !allowed {
                return Ok(ControlResponse::CasFetched {
                    peer_node_id: node_id,
                    peer_agent_id: agent_id,
                    endpoint_id,
                    hash: response_hash,
                    size_bytes: 0,
                    allowed,
                    reason,
                    note,
                });
            }
            let content = content_base64
                .ok_or_else(|| NodeError::IrohPeer("peer omitted CAS content".to_owned()))
                .and_then(|content| {
                    base64::engine::general_purpose::STANDARD
                        .decode(content)
                        .map_err(|error| NodeError::IrohPeer(error.to_string()))
                })?;
            if content.len() as u64 != size_bytes {
                return Err(NodeError::IrohPeer(format!(
                    "peer announced {size_bytes} CAS bytes but returned {} bytes",
                    content.len()
                )));
            }
            let cas = LocalCas::new(node.paths.cas_dir());
            let info = cas.add_bytes(&content)?;
            if info.hash != response_hash {
                return Err(NodeError::IrohPeer(format!(
                    "peer returned content hash {} for requested {}",
                    info.hash, response_hash
                )));
            }
            store.record_cas_object(
                info.hash.as_str(),
                info.size_bytes,
                &info.path.to_string_lossy(),
            )?;
            Ok(ControlResponse::CasFetched {
                peer_node_id: node_id,
                peer_agent_id: agent_id,
                endpoint_id,
                hash: info.hash,
                size_bytes: info.size_bytes,
                allowed,
                reason,
                note,
            })
        }
        PeerControlResponse::CasFetched { .. } => Err(NodeError::IrohPeer(
            "peer CAS fetch response did not match request".to_owned(),
        )),
        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
        PeerControlResponse::Pong { .. }
        | PeerControlResponse::AuthChecked { .. }
        | PeerControlResponse::SshCertSynced { .. }
        | PeerControlResponse::SshRevocationSynced { .. } => Err(NodeError::IrohPeer(
            "peer returned wrong response type to CAS fetch".to_owned(),
        )),
    }
}

async fn ssh_cert_sync_from_peer(
    node: &LocalNode,
    peer_node: &str,
) -> Result<ControlResponse, NodeError> {
    let since_ms = load_live_sync_cursor(
        &Store::open(&node.paths.metadata_db())?,
        peer_node,
        "ssh-certs",
    )?;
    let response = request_peer_control(node, peer_node, "ssh-cert-sync", |peer_card, nonce| {
        PeerControlRequest::SshCertSync {
            peer_card,
            since_ms,
            nonce,
        }
    })
    .await?;
    match response {
        PeerControlResponse::SshCertSynced {
            node_id,
            agent_id,
            endpoint_id,
            requests,
            certificates,
            high_water_ms,
            allowed,
            reason,
            note,
            ..
        } => {
            if !allowed {
                return Ok(ControlResponse::SshCertSynced {
                    peer_node_id: node_id,
                    peer_agent_id: agent_id,
                    endpoint_id,
                    requests_imported: 0,
                    certificates_imported: 0,
                    allowed,
                    reason,
                    note,
                });
            }
            let store = Store::open(&node.paths.metadata_db())?;
            let requests_imported = requests.len();
            let certificates_imported = certificates.len();
            for request in &requests {
                store.insert_ssh_cert_request(&stored_from_ssh_cert_request(request))?;
            }
            for certificate in &certificates {
                store.insert_ssh_certificate(&stored_from_ssh_certificate(certificate))?;
            }
            store_live_sync_cursor(&store, peer_node, "ssh-certs", high_water_ms)?;
            Ok(ControlResponse::SshCertSynced {
                peer_node_id: node_id,
                peer_agent_id: agent_id,
                endpoint_id,
                requests_imported,
                certificates_imported,
                allowed,
                reason,
                note,
            })
        }
        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
        _ => Err(NodeError::IrohPeer(
            "peer returned wrong response type to SSH cert sync".to_owned(),
        )),
    }
}

async fn ssh_revocation_sync_from_peer(
    node: &LocalNode,
    peer_node: &str,
) -> Result<ControlResponse, NodeError> {
    let since_ms = load_live_sync_cursor(
        &Store::open(&node.paths.metadata_db())?,
        peer_node,
        "ssh-revocations",
    )?;
    let response = request_peer_control(
        node,
        peer_node,
        "ssh-revocation-sync",
        |peer_card, nonce| PeerControlRequest::SshRevocationSync {
            peer_card,
            since_ms,
            nonce,
        },
    )
    .await?;
    match response {
        PeerControlResponse::SshRevocationSynced {
            node_id,
            agent_id,
            endpoint_id,
            revocations,
            high_water_ms,
            allowed,
            reason,
            note,
            ..
        } => {
            if !allowed {
                return Ok(ControlResponse::SshRevocationSynced {
                    peer_node_id: node_id,
                    peer_agent_id: agent_id,
                    endpoint_id,
                    revocations_imported: 0,
                    allowed,
                    reason,
                    note,
                });
            }
            let store = Store::open(&node.paths.metadata_db())?;
            let revocations_imported = revocations.len();
            for revocation in &revocations {
                store.insert_ssh_revocation(&stored_from_ssh_revocation(revocation))?;
            }
            store_live_sync_cursor(&store, peer_node, "ssh-revocations", high_water_ms)?;
            Ok(ControlResponse::SshRevocationSynced {
                peer_node_id: node_id,
                peer_agent_id: agent_id,
                endpoint_id,
                revocations_imported,
                allowed,
                reason,
                note,
            })
        }
        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
        _ => Err(NodeError::IrohPeer(
            "peer returned wrong response type to SSH revocation sync".to_owned(),
        )),
    }
}

fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String {
    format!("live-sync:{peer_node}:{stream}")
}

fn load_live_sync_cursor(store: &Store, peer_node: &str, stream: &str) -> Result<i64, NodeError> {
    let key = live_sync_cursor_key(peer_node, stream);
    let Some(state) = store.get_module_state(&key)? else {
        return Ok(0);
    };
    let cursor: LiveSyncCursor = serde_json::from_str(&state.state_json)?;
    Ok(cursor.cursor_ms)
}

fn store_live_sync_cursor(
    store: &Store,
    peer_node: &str,
    stream: &str,
    cursor_ms: i64,
) -> Result<(), NodeError> {
    store.put_module_state(&StoredModuleState {
        module: live_sync_cursor_key(peer_node, stream),
        state_json: serde_json::to_string(&LiveSyncCursor { cursor_ms })?,
        updated_at_ms: geth_store::now_ms(),
    })?;
    Ok(())
}

async fn request_peer_control(
    node: &LocalNode,
    peer_node: &str,
    operation: &str,
    build_request: impl FnOnce(PeerCard, String) -> PeerControlRequest,
) -> Result<PeerControlResponse, NodeError> {
    let store = Store::open(&node.paths.metadata_db())?;
    let stored = store
        .get_peer_card(peer_node)?
        .ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?;
    let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
    peer_card.validate_candidate()?;
    let candidate = peer_card
        .endpoints
        .first()
        .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
    ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
    let node_addr = iroh_node_addr_from_candidate(candidate)?;
    let endpoint = node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .clone()
        .ok_or(NodeError::IrohEndpointUnavailable)?;
    let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?;
    let nonce = geth_crypto::blake3_hex(
        format!(
            "{}\0{}\0{}\0{}",
            node.node_id,
            peer_node,
            operation,
            geth_store::now_ms()
        )
        .as_bytes(),
    );
    let request = build_request(self_card, nonce.clone());
    let conn = endpoint
        .endpoint()
        .connect(node_addr, geth_iroh::ALPN_CONTROL)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.write_all(geth_control::encode_peer_request(&request)?.as_bytes())
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.finish()
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let bytes = recv
        .read_to_end(16 * 1024 * 1024)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let text =
        std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let response = geth_control::decode_peer_response(text)?;
    match &response {
        PeerControlResponse::SshCertSynced {
            nonce: response_nonce,
            ..
        }
        | PeerControlResponse::SshRevocationSynced {
            nonce: response_nonce,
            ..
        } if response_nonce == &nonce => Ok(response),
        PeerControlResponse::Error { .. } => Ok(response),
        _ => Err(NodeError::IrohPeer(format!(
            "peer {operation} response did not match request"
        ))),
    }
}

fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
    let raw_endpoint = endpoint.endpoint();
    tokio::spawn(async move {
        while let Some(incoming) = raw_endpoint.accept().await {
            let node = node.clone();
            tokio::spawn(async move {
                if let Err(error) = handle_iroh_control_connection(node, incoming).await {
                    tracing::warn!(%error, "iroh control request failed");
                }
            });
        }
    });
}

fn spawn_background_live_sync(node: LocalNode) {
    tokio::spawn(async move {
        if let Err(error) = run_live_sync_once(&node).await {
            tracing::debug!(%error, "initial live sync tick failed");
        }
        let mut interval = tokio::time::interval(LIVE_SYNC_INTERVAL);
        loop {
            interval.tick().await;
            if let Err(error) = run_live_sync_once(&node).await {
                tracing::debug!(%error, "live sync tick failed");
            }
        }
    });
}

async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
    if node
        .iroh_endpoint
        .lock()
        .map_err(|_| NodeError::RuntimeLockPoisoned)?
        .is_none()
    {
        return Ok(());
    }
    let peers = Store::open(&node.paths.metadata_db())?.list_peer_cards()?;
    for peer in peers {
        if let Err(error) = ssh_cert_sync_from_peer(node, &peer.peer_id).await {
            tracing::debug!(peer = %peer.peer_id, %error, "SSH cert live sync failed");
        }
        if let Err(error) = ssh_revocation_sync_from_peer(node, &peer.peer_id).await {
            tracing::debug!(peer = %peer.peer_id, %error, "SSH revocation live sync failed");
        }
    }
    Ok(())
}

async fn handle_iroh_control_connection(
    node: LocalNode,
    incoming: iroh::endpoint::Incoming,
) -> Result<(), NodeError> {
    let conn = incoming
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let remote_endpoint_id = conn
        .remote_node_id()
        .map(|node_id| node_id.to_string())
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let alpn = conn
        .alpn()
        .map(display_alpn)
        .unwrap_or_else(|| "unknown".to_owned());
    let (mut send, mut recv) = conn
        .accept_bi()
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let bytes = recv
        .read_to_end(64 * 1024)
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let text =
        std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let response = match geth_control::decode_peer_request(text)? {
        PeerControlRequest::Ping { peer_card, nonce } => {
            peer_card.validate_candidate()?;
            ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
            let discovered = DiscoveredPeer::candidate(
                peer_card.clone(),
                UnixMillis(geth_store::now_ms()),
                DiscoverySource::PeerExchange,
            )?;
            let store = Store::open(&node.paths.metadata_db())?;
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: peer_card.node_id.to_string(),
                card_json: serde_json::to_string(&peer_card)?,
                updated_at_ms: discovered.discovered_at.0,
            })?;
            PeerControlResponse::Pong {
                node_id: node.node_id.clone(),
                agent_id: node.agent_id.clone(),
                endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
                remote_endpoint_id,
                alpn,
                nonce,
                note: "peer endpoint authenticated by Iroh and peer-card signature; candidate status does not grant resource capabilities".to_owned(),
            }
        }
        PeerControlRequest::AuthCheck {
            peer_card,
            resource,
            capability,
            nonce,
        } => {
            peer_card.validate_candidate()?;
            ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
            let discovered = DiscoveredPeer::candidate(
                peer_card.clone(),
                UnixMillis(geth_store::now_ms()),
                DiscoverySource::PeerExchange,
            )?;
            let store = Store::open(&node.paths.metadata_db())?;
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: peer_card.node_id.to_string(),
                card_json: serde_json::to_string(&peer_card)?,
                updated_at_ms: discovered.discovered_at.0,
            })?;
            let explanation = geth_auth::explain_auth_ops(
                &load_auth_ops_for_resource(&store, &resource)?,
                PrincipalId::new(peer_card.node_id.to_string()),
                ResourceId::new(resource.clone()),
                Capability::new(capability.clone()),
            );
            PeerControlResponse::AuthChecked {
                node_id: node.node_id.clone(),
                agent_id: node.agent_id.clone(),
                endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
                remote_endpoint_id,
                resource,
                capability,
                allowed: explanation.allowed,
                reason: explanation.reason,
                evaluated_ops: explanation.evaluated_ops,
                nonce,
                note: "protected peer request authenticated endpoint/card binding before resource capability evaluation".to_owned(),
            }
        }
        PeerControlRequest::CasFetch {
            peer_card,
            hash,
            nonce,
        } => {
            peer_card.validate_candidate()?;
            ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
            let discovered = DiscoveredPeer::candidate(
                peer_card.clone(),
                UnixMillis(geth_store::now_ms()),
                DiscoverySource::PeerExchange,
            )?;
            let store = Store::open(&node.paths.metadata_db())?;
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: peer_card.node_id.to_string(),
                card_json: serde_json::to_string(&peer_card)?,
                updated_at_ms: discovered.discovered_at.0,
            })?;
            let resource = "resource:cas:local".to_owned();
            let capability = "cas.fetch".to_owned();
            let explanation = geth_auth::explain_auth_ops(
                &load_auth_ops_for_resource(&store, &resource)?,
                PrincipalId::new(peer_card.node_id.to_string()),
                ResourceId::new(resource),
                Capability::new(capability),
            );
            if explanation.allowed {
                match LocalCas::new(node.paths.cas_dir()).read_bytes(&hash) {
                    Ok(content) => {
                        let size_bytes = content.len() as u64;
                        PeerControlResponse::CasFetched {
                            node_id: node.node_id.clone(),
                            agent_id: node.agent_id.clone(),
                            endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
                            remote_endpoint_id,
                            hash,
                            size_bytes,
                            content_base64: Some(
                                base64::engine::general_purpose::STANDARD.encode(content),
                            ),
                            allowed: true,
                            reason: explanation.reason,
                            evaluated_ops: explanation.evaluated_ops,
                            nonce,
                            note: "CAS fetch authenticated endpoint/card binding and required cas.fetch on resource:cas:local".to_owned(),
                        }
                    }
                    Err(error) => PeerControlResponse::Error {
                        message: format!("CAS blob {hash} is not available: {error}"),
                    },
                }
            } else {
                PeerControlResponse::CasFetched {
                    node_id: node.node_id.clone(),
                    agent_id: node.agent_id.clone(),
                    endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
                    remote_endpoint_id,
                    hash,
                    size_bytes: 0,
                    content_base64: None,
                    allowed: false,
                    reason: explanation.reason,
                    evaluated_ops: explanation.evaluated_ops,
                    nonce,
                    note: "CAS fetch authenticated endpoint/card binding and required cas.fetch on resource:cas:local".to_owned(),
                }
            }
        }
        PeerControlRequest::SshCertSync {
            peer_card,
            since_ms,
            nonce,
        } => {
            peer_card.validate_candidate()?;
            ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
            let discovered = DiscoveredPeer::candidate(
                peer_card.clone(),
                UnixMillis(geth_store::now_ms()),
                DiscoverySource::PeerExchange,
            )?;
            let store = Store::open(&node.paths.metadata_db())?;
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: peer_card.node_id.to_string(),
                card_json: serde_json::to_string(&peer_card)?,
                updated_at_ms: discovered.discovered_at.0,
            })?;
            let resource = "resource:ssh:certs".to_owned();
            let capability = "ssh_cert.sync".to_owned();
            let high_water_ms = geth_store::now_ms();
            let explanation = geth_auth::explain_auth_ops(
                &load_auth_ops_for_resource(&store, &resource)?,
                PrincipalId::new(peer_card.node_id.to_string()),
                ResourceId::new(resource),
                Capability::new(capability),
            );
            let (requests, certificates) = if explanation.allowed {
                (
                    store
                        .list_ssh_cert_requests_since(since_ms)?
                        .into_iter()
                        .map(ssh_cert_request_from_stored)
                        .collect::<Result<Vec<_>, _>>()?,
                    store
                        .list_ssh_certificates_since(since_ms)?
                        .into_iter()
                        .map(ssh_certificate_from_stored)
                        .collect(),
                )
            } else {
                (Vec::new(), Vec::new())
            };
            PeerControlResponse::SshCertSynced {
                node_id: node.node_id.clone(),
                agent_id: node.agent_id.clone(),
                endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
                remote_endpoint_id,
                requests,
                certificates,
                high_water_ms,
                allowed: explanation.allowed,
                reason: explanation.reason,
                evaluated_ops: explanation.evaluated_ops,
                nonce,
                note: "SSH certificate metadata sync authenticated endpoint/card binding and required ssh_cert.sync on resource:ssh:certs".to_owned(),
            }
        }
        PeerControlRequest::SshRevocationSync {
            peer_card,
            since_ms,
            nonce,
        } => {
            peer_card.validate_candidate()?;
            ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
            let discovered = DiscoveredPeer::candidate(
                peer_card.clone(),
                UnixMillis(geth_store::now_ms()),
                DiscoverySource::PeerExchange,
            )?;
            let store = Store::open(&node.paths.metadata_db())?;
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: peer_card.node_id.to_string(),
                card_json: serde_json::to_string(&peer_card)?,
                updated_at_ms: discovered.discovered_at.0,
            })?;
            let resource = "resource:ssh:revocations".to_owned();
            let capability = "ssh_revocation.sync".to_owned();
            let high_water_ms = geth_store::now_ms();
            let explanation = geth_auth::explain_auth_ops(
                &load_auth_ops_for_resource(&store, &resource)?,
                PrincipalId::new(peer_card.node_id.to_string()),
                ResourceId::new(resource),
                Capability::new(capability),
            );
            let revocations = if explanation.allowed {
                store
                    .list_ssh_revocations_since(since_ms)?
                    .into_iter()
                    .map(ssh_revocation_from_stored)
                    .collect::<Result<Vec<_>, _>>()?
            } else {
                Vec::new()
            };
            PeerControlResponse::SshRevocationSynced {
                node_id: node.node_id.clone(),
                agent_id: node.agent_id.clone(),
                endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
                remote_endpoint_id,
                revocations,
                high_water_ms,
                allowed: explanation.allowed,
                reason: explanation.reason,
                evaluated_ops: explanation.evaluated_ops,
                nonce,
                note: "SSH revocation sync authenticated endpoint/card binding and required ssh_revocation.sync on resource:ssh:revocations".to_owned(),
            }
        }
    };
    send.write_all(geth_control::encode_peer_response(&response)?.as_bytes())
        .await
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    send.finish()
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    Ok(())
}

fn iroh_node_addr_from_candidate(
    candidate: &EndpointCandidate,
) -> Result<iroh::NodeAddr, NodeError> {
    let node_id = candidate
        .endpoint_id
        .parse::<iroh::NodeId>()
        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
    let direct_addresses = candidate
        .direct_addresses
        .iter()
        .map(|addr| {
            addr.parse::<std::net::SocketAddr>()
                .map_err(|error| NodeError::IrohPeer(error.to_string()))
        })
        .collect::<Result<Vec<_>, _>>()?;
    let mut node_addr = iroh::NodeAddr::new(node_id).with_direct_addresses(direct_addresses);
    if let Some(relay_url) = &candidate.relay_url {
        node_addr = node_addr.with_relay_url(
            relay_url
                .parse::<iroh::RelayUrl>()
                .map_err(|error| NodeError::IrohPeer(error.to_string()))?,
        );
    }
    Ok(node_addr)
}

fn ensure_peer_card_matches_endpoint(card: &PeerCard, endpoint_id: &str) -> Result<(), NodeError> {
    if card
        .endpoints
        .iter()
        .any(|candidate| candidate.endpoint_id == endpoint_id)
    {
        Ok(())
    } else {
        Err(NodeError::IrohPeer(format!(
            "signed peer card for {} does not bind Iroh endpoint {}",
            card.node_id, endpoint_id
        )))
    }
}

fn display_alpn(alpn: Vec<u8>) -> String {
    String::from_utf8_lossy(&alpn).into_owned()
}

pub fn handle_request(
    node: &LocalNode,
    request: ControlRequest,
) -> Result<ControlResponse, NodeError> {
    let store = Store::open(&node.paths.metadata_db())?;
    match request {
        ControlRequest::Status => Ok(ControlResponse::Status(StatusResponse {
            home: node.paths.home().to_path_buf(),
            socket: node.paths.socket_path(),
            agent_id: node.agent_id.clone(),
            node_id: node.node_id.clone(),
            iroh_enabled: node.iroh_status.enabled,
            endpoint_id: node.iroh_status.endpoint_id.clone(),
            iroh_relay_mode: node.iroh_status.relay_mode.clone(),
            iroh_local_discovery: node.iroh_status.local_discovery,
            iroh: node.iroh_status.note.clone(),
        })),
        ControlRequest::NodeId => Ok(ControlResponse::NodeId(NodeIdResponse {
            agent_id: node.agent_id.clone(),
            node_id: node.node_id.clone(),
            endpoint_id: node.iroh_status.endpoint_id.clone(),
        })),
        ControlRequest::PeerCardExport { out } => {
            let endpoint_id = node
                .iroh_status
                .endpoint_id
                .clone()
                .ok_or(NodeError::IrohEndpointUnavailable)?;
            let key = AgentKey::load(&node.paths.agent_key())?;
            let card = PeerCard::signed(
                NodeId::new(node.node_id.clone()),
                &key,
                vec![EndpointCandidate {
                    endpoint_id,
                    relay_url: None,
                    direct_addresses: Vec::new(),
                    source: DiscoverySource::Manual,
                }],
                UnixMillis(geth_store::now_ms()),
            )?;
            if let Some(path) = &out {
                std::fs::write(path, serde_json::to_string_pretty(&card)?)?;
            }
            Ok(ControlResponse::PeerCardExported {
                card,
                out,
                note: discovery_is_untrusted_note().to_owned(),
            })
        }
        ControlRequest::PeerCardImport { path } => {
            let card_json = std::fs::read_to_string(&path)?;
            let card: PeerCard = serde_json::from_str(&card_json)?;
            card.validate_candidate()?;
            let discovered = DiscoveredPeer::candidate(
                card.clone(),
                UnixMillis(geth_store::now_ms()),
                DiscoverySource::Imported,
            )?;
            store.upsert_peer_card(&StoredPeerCard {
                peer_id: card.node_id.to_string(),
                card_json: serde_json::to_string(&card)?,
                updated_at_ms: discovered.discovered_at.0,
            })?;
            Ok(ControlResponse::PeerCardImported {
                peer: discovered,
                note: discovery_is_untrusted_note().to_owned(),
            })
        }
        ControlRequest::PeerCardList => {
            let peers = store
                .list_peer_cards()?
                .into_iter()
                .map(discovered_peer_from_stored)
                .collect::<Result<Vec<_>, _>>()?;
            Ok(ControlResponse::PeerCardList {
                peers,
                note: discovery_is_untrusted_note().to_owned(),
            })
        }
        ControlRequest::PeerPing { .. } => Err(NodeError::IrohEndpointUnavailable),
        ControlRequest::PeerAuthCheck { .. } => Err(NodeError::IrohEndpointUnavailable),
        ControlRequest::CasFetch { .. } => Err(NodeError::IrohEndpointUnavailable),
        ControlRequest::SshCertSync { .. } => Err(NodeError::IrohEndpointUnavailable),
        ControlRequest::SshRevocationSync { .. } => Err(NodeError::IrohEndpointUnavailable),
        ControlRequest::ResourceList => Ok(ControlResponse::ResourceList {
            resources: store
                .list_resources()?
                .into_iter()
                .map(stored_resource_to_descriptor)
                .collect::<Result<Vec<_>, _>>()?,
        }),
        ControlRequest::ResourceCreate { kind, name } => {
            let kind = kind
                .parse::<ResourceKind>()
                .map_err(|_| NodeError::InvalidResourceKind(kind.clone()))?;
            let id = format!("resource:{}:{}", kind, name);
            let stored = StoredResource {
                resource_id: id,
                kind: kind.to_string(),
                name,
                status: "active".to_owned(),
            };
            store.insert_resource(&stored)?;
            Ok(ControlResponse::ResourceCreated {
                resource: stored_resource_to_descriptor(stored)?,
            })
        }
        ControlRequest::CasAdd { path } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let info = cas.add_path(&path)?;
            store.record_cas_object(
                info.hash.as_str(),
                info.size_bytes,
                &info.path.to_string_lossy(),
            )?;
            Ok(ControlResponse::CasAdded {
                hash: info.hash,
                size_bytes: info.size_bytes,
            })
        }
        ControlRequest::CasGet { hash, out } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let size_bytes = cas.get_to_path(&hash, &out)?;
            Ok(ControlResponse::CasGot {
                hash,
                out,
                size_bytes,
            })
        }
        ControlRequest::CasHash { path } => Ok(ControlResponse::CasHash {
            hash: hash_path(&path)?,
        }),
        ControlRequest::CasHas { hash } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let present = cas.has(&hash)?;
            Ok(ControlResponse::CasHas { hash, present })
        }
        ControlRequest::CasPin { hash } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            if !cas.has(&hash)? {
                return Err(NodeError::Cas(geth_cas::CasError::NotFound(
                    hash.to_string(),
                )));
            }
            store.pin_cas_object(hash.as_str())?;
            Ok(ControlResponse::CasPinned { hash, pinned: true })
        }
        ControlRequest::CasUnpin { hash } => {
            geth_cas::validate_hash(&hash)?;
            store.unpin_cas_object(hash.as_str())?;
            Ok(ControlResponse::CasPinned {
                hash,
                pinned: false,
            })
        }
        ControlRequest::CasCleanup { dry_run } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let mut removed = Vec::new();
            let mut retained_pinned = Vec::new();
            for blob in cas.list()? {
                if store.is_cas_object_pinned(blob.hash.as_str())? {
                    retained_pinned.push(blob.hash);
                    continue;
                }
                if !dry_run && cas.remove(&blob.hash)? {
                    store.delete_cas_object(blob.hash.as_str())?;
                }
                removed.push(blob.hash);
            }
            Ok(ControlResponse::CasCleanup {
                removed,
                retained_pinned,
                dry_run,
            })
        }
        ControlRequest::CasList => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let blobs = cas
                .list()?
                .into_iter()
                .map(|blob| {
                    Ok(CasBlob {
                        pinned: store.is_cas_object_pinned(blob.hash.as_str())?,
                        hash: blob.hash,
                        size_bytes: blob.size_bytes,
                    })
                })
                .collect::<Result<Vec<_>, NodeError>>()?;
            Ok(ControlResponse::CasList { blobs })
        }
        ControlRequest::CasRootAdd { name, path } => {
            geth_cas::validate_file_root_name(&name)?;
            if !path.is_dir() {
                return Err(NodeError::Cas(geth_cas::CasError::TreeRootNotDirectory(
                    path.display().to_string(),
                )));
            }
            let path = std::fs::canonicalize(path)?;
            let resource_id = format!("resource:cas-tree:{name}");
            store.insert_resource(&StoredResource {
                resource_id: resource_id.clone(),
                kind: ResourceKind::Cas.to_string(),
                name: format!("file-root:{name}"),
                status: "active".to_owned(),
            })?;
            let stored = StoredFileRoot {
                root_id: format!("file-root:{name}"),
                resource_id,
                name,
                path: path.display().to_string(),
                latest_tree_hash: None,
                latest_tree_json: None,
                updated_at_ms: geth_store::now_ms(),
            };
            store.upsert_file_root(&stored)?;
            Ok(ControlResponse::CasRootAdded {
                root: file_root_from_stored(&stored),
            })
        }
        ControlRequest::CasRootList => Ok(ControlResponse::CasRootList {
            roots: store
                .list_file_roots()?
                .iter()
                .map(file_root_from_stored)
                .collect(),
        }),
        ControlRequest::CasRootScan { name } => {
            geth_cas::validate_file_root_name(&name)?;
            let mut stored = store
                .get_file_root_by_name(&name)?
                .ok_or_else(|| NodeError::ResourceNotFound(format!("file-root:{name}")))?;
            let previous_tree = stored
                .latest_tree_json
                .as_deref()
                .map(serde_json::from_str)
                .transpose()?;
            let cas = LocalCas::new(node.paths.cas_dir());
            let scanned = cas.add_tree_path(Path::new(&stored.path))?;
            store.record_cas_object(
                scanned.object.hash.as_str(),
                scanned.object.size_bytes,
                &scanned.object.path.to_string_lossy(),
            )?;
            let changes = geth_cas::diff_tree_objects(previous_tree.as_ref(), &scanned.tree);
            stored.latest_tree_hash = Some(scanned.object.hash.to_string());
            stored.latest_tree_json = Some(serde_json::to_string(&scanned.tree)?);
            stored.updated_at_ms = geth_store::now_ms();
            store.upsert_file_root(&stored)?;
            Ok(ControlResponse::CasRootScanned {
                scan: FileRootScan {
                    root: file_root_from_stored(&stored),
                    tree: BlobInfoSummary {
                        hash: scanned.object.hash,
                        size_bytes: scanned.object.size_bytes,
                    },
                    changes,
                    note: geth_cas::file_root_scan_note().to_owned(),
                },
            })
        }
        ControlRequest::CasConflictRecord {
            root,
            path,
            kind,
            detail,
            base_tree,
            local_tree,
            remote_tree,
        } => {
            geth_cas::validate_file_root_name(&root)?;
            if let Some(hash) = &base_tree {
                geth_cas::validate_hash(hash)?;
            }
            if let Some(hash) = &local_tree {
                geth_cas::validate_hash(hash)?;
            }
            if let Some(hash) = &remote_tree {
                geth_cas::validate_hash(hash)?;
            }
            let root_record = store
                .get_file_root_by_name(&root)?
                .ok_or_else(|| NodeError::ResourceNotFound(format!("file-root:{root}")))?;
            let kind = FileConflictKind::parse(&kind)?;
            let created_at = geth_store::now_ms();
            let conflict_id = generated_file_conflict_id(&root, &path, kind.as_str(), created_at);
            let conflict = StoredFileConflict {
                conflict_id,
                root_name: root,
                resource_id: root_record.resource_id,
                path,
                kind: kind.as_str().to_owned(),
                status: FileConflictStatus::Open.as_str().to_owned(),
                base_tree_hash: base_tree.map(|hash| hash.to_string()),
                local_tree_hash: local_tree.map(|hash| hash.to_string()),
                remote_tree_hash: remote_tree.map(|hash| hash.to_string()),
                detail,
                resolution: None,
                resolution_note: None,
                created_at_ms: created_at,
                resolved_at_ms: None,
            };
            store.upsert_file_conflict(&conflict)?;
            Ok(ControlResponse::CasConflictRecorded {
                conflict: file_conflict_from_stored(conflict)?,
            })
        }
        ControlRequest::CasConflictList { root } => {
            if let Some(root) = root.as_deref() {
                geth_cas::validate_file_root_name(root)?;
            }
            let conflicts = store
                .list_file_conflicts(root.as_deref())?
                .into_iter()
                .map(file_conflict_from_stored)
                .collect::<Result<Vec<_>, _>>()?;
            Ok(ControlResponse::CasConflictList { conflicts })
        }
        ControlRequest::CasConflictResolve {
            conflict_id,
            resolution,
            note,
        } => {
            let resolution = FileConflictResolution::parse(&resolution)?;
            let mut conflict = store
                .get_file_conflict(&conflict_id)?
                .ok_or_else(|| NodeError::ResourceNotFound(conflict_id.clone()))?;
            conflict.status = FileConflictStatus::Resolved.as_str().to_owned();
            conflict.resolution = Some(resolution.as_str().to_owned());
            conflict.resolution_note = note;
            conflict.resolved_at_ms = Some(geth_store::now_ms());
            store.upsert_file_conflict(&conflict)?;
            Ok(ControlResponse::CasConflictResolved {
                conflict: file_conflict_from_stored(conflict)?,
            })
        }
        ControlRequest::KeychainInit { admin_key_path } => {
            let mut ops = Vec::new();
            let created_at = UnixMillis(geth_store::now_ms());
            let init = KeychainOp {
                id: generated_keychain_op_id("keychain-init", "local", created_at),
                created_at,
                kind: KeychainOpKind::KeychainInit,
            };
            store_keychain_op(&store, &init)?;
            ops.push(init);

            if let Some(admin_key_path) = admin_key_path {
                let public_key = std::fs::read_to_string(admin_key_path)?;
                let created_at = UnixMillis(geth_store::now_ms());
                let admin_key = KeyId::new(ssh_public_key_fingerprint(&public_key));
                let op = KeychainOp {
                    id: generated_keychain_op_id("admin-key-add", admin_key.as_str(), created_at),
                    created_at,
                    kind: KeychainOpKind::AdminKeyAdd { key: admin_key },
                };
                store_keychain_op(&store, &op)?;
                ops.push(op);
            }

            Ok(ControlResponse::KeychainInitialized { ops })
        }
        ControlRequest::KeychainStatus => {
            let view = geth_keychain::reduce_keychain_ops(&load_keychain_ops(&store)?);
            Ok(ControlResponse::KeychainStatus(KeychainStatusResponse {
                initialized: view.initialized,
                admin_keys: view.admin_keys.len(),
                users: view.users.len(),
                devices: view.devices.len(),
                nodes: view.nodes.len(),
            }))
        }
        ControlRequest::SecretStatus => Ok(ControlResponse::SecretStatus {
            secrets: store
                .list_resource_secrets()?
                .into_iter()
                .map(resource_secret_from_stored)
                .collect(),
        }),
        ControlRequest::SecretCreate { resource } => {
            ensure_resource_exists(&store, &resource)?;
            let secret = create_resource_secret(&store, &resource, 1)?;
            Ok(ControlResponse::SecretCreated { secret })
        }
        ControlRequest::SecretRotate { resource } => {
            ensure_resource_exists(&store, &resource)?;
            let next_epoch = store
                .latest_resource_secret(&resource)?
                .map(|secret| secret.epoch + 1)
                .unwrap_or(1);
            let secret = create_resource_secret(&store, &resource, next_epoch)?;
            Ok(ControlResponse::SecretCreated { secret })
        }
        ControlRequest::SecretBearerCreate {
            resource,
            capabilities,
            expires_at_ms,
        } => {
            ensure_resource_exists(&store, &resource)?;
            let capabilities = capabilities
                .into_iter()
                .map(Capability::new)
                .collect::<Vec<_>>();
            geth_secrets::validate_bearer_capabilities(&capabilities)?;
            let created_at = UnixMillis(geth_store::now_ms());
            let secret = geth_types::SecretId::new(format!(
                "bearer:{}",
                geth_crypto::blake3_hex(
                    format!(
                        "{resource}\0{}\0{}",
                        capabilities
                            .iter()
                            .map(ToString::to_string)
                            .collect::<Vec<_>>()
                            .join(","),
                        created_at.0
                    )
                    .as_bytes()
                )
            ));
            let access = BearerAccess::resource_scoped(
                secret.clone(),
                ResourceId::new(resource.clone()),
                capabilities.clone(),
            );
            let op = AuthOp {
                id: generated_auth_op_id("bearer-create", &resource, secret.as_str(), created_at),
                resource: ResourceId::new(resource),
                created_at,
                kind: AuthOpKind::BearerAccessCreate {
                    secret,
                    capabilities,
                    expires_at: expires_at_ms.map(UnixMillis),
                },
            };
            store_auth_op(&store, &op)?;
            Ok(ControlResponse::SecretBearerCreated {
                access: BearerAccess {
                    expires_at: expires_at_ms.map(UnixMillis),
                    ..access
                },
            })
        }
        ControlRequest::SecretBearerList => Ok(ControlResponse::SecretBearerList {
            access: load_bearer_access(&store)?,
        }),
        ControlRequest::SecretBearerRevoke { resource, secret } => {
            ensure_resource_exists(&store, &resource)?;
            let created_at = UnixMillis(geth_store::now_ms());
            let op = AuthOp {
                id: generated_auth_op_id("bearer-revoke", &resource, &secret, created_at),
                resource: ResourceId::new(resource.clone()),
                created_at,
                kind: AuthOpKind::BearerAccessRevoke {
                    secret: secret.clone().into(),
                },
            };
            store_auth_op(&store, &op)?;
            Ok(ControlResponse::SecretBearerRevoked { resource, secret })
        }
        ControlRequest::AuthExplain {
            subject,
            resource,
            capability,
        } => {
            let ops = load_auth_ops_for_resource(&store, &resource)?;
            let discovered = store.get_peer_card(&subject)?.is_some();
            if ops.is_empty() {
                if discovered {
                    Ok(ControlResponse::AuthExplain(
                        AuthExplanation::discovered_candidate(subject, resource, capability),
                    ))
                } else {
                    Ok(ControlResponse::AuthExplain(AuthExplanation::stub(
                        subject, resource, capability,
                    )))
                }
            } else {
                let mut explanation = geth_auth::explain_auth_ops(
                    &ops,
                    PrincipalId::new(subject.clone()),
                    ResourceId::new(resource.clone()),
                    Capability::new(capability.clone()),
                );
                if discovered && !explanation.allowed {
                    explanation.reason = format!(
                        "subject is a discovered peer candidate only; discovery does not grant trust or authorization; {}",
                        explanation.reason
                    );
                }
                Ok(ControlResponse::AuthExplain(explanation))
            }
        }
        ControlRequest::AuthGrant {
            subject,
            resource,
            capability,
            grant_id,
        } => {
            let created_at = UnixMillis(geth_store::now_ms());
            let grant_id =
                grant_id.unwrap_or_else(|| generated_grant_id(&subject, &resource, &capability));
            let op = AuthOp {
                id: generated_auth_op_id("grant-create", &resource, &grant_id, created_at),
                resource: ResourceId::new(resource),
                created_at,
                kind: AuthOpKind::GrantCreate {
                    grant_id,
                    principal: PrincipalId::new(subject),
                    capabilities: vec![Capability::new(capability)],
                },
            };
            store_auth_op(&store, &op)?;
            Ok(ControlResponse::AuthOpRecorded { op })
        }
        ControlRequest::AuthRevoke { resource, grant_id } => {
            let created_at = UnixMillis(geth_store::now_ms());
            let op = AuthOp {
                id: generated_auth_op_id("grant-revoke", &resource, &grant_id, created_at),
                resource: ResourceId::new(resource),
                created_at,
                kind: AuthOpKind::GrantRevoke { grant_id },
            };
            store_auth_op(&store, &op)?;
            Ok(ControlResponse::AuthOpRecorded { op })
        }
        ControlRequest::SshCertRequest {
            public_key_path,
            cert_kind,
            principals,
            requested_validity,
            renewal_of,
            reason,
        } => {
            if principals.is_empty() {
                return Err(NodeError::MissingSshCertPrincipal);
            }
            let cert_kind = cert_kind
                .parse::<SshCertKind>()
                .map_err(|_| NodeError::InvalidSshCertKind(cert_kind.clone()))?;
            let public_key = std::fs::read_to_string(&public_key_path)?;
            let created_at = UnixMillis(geth_store::now_ms());
            let request = SshCertRequest {
                id: cert_request_id(
                    &NodeId::new(node.node_id.clone()),
                    &public_key,
                    &principals,
                    created_at,
                ),
                requester_node: NodeId::new(node.node_id.clone()),
                public_key_fingerprint: ssh_public_key_fingerprint(&public_key),
                public_key,
                cert_kind,
                principals,
                requested_validity,
                renewal_of: renewal_of.map(SshCertId::new),
                reason,
                status: SshCertRequestStatus::Pending,
                created_at,
            };
            store.insert_ssh_cert_request(&stored_from_ssh_cert_request(&request))?;
            Ok(ControlResponse::SshCertRequested { request })
        }
        ControlRequest::SshCertRequests => Ok(ControlResponse::SshCertRequests {
            requests: store
                .list_ssh_cert_requests()?
                .into_iter()
                .map(ssh_cert_request_from_stored)
                .collect::<Result<Vec<_>, _>>()?,
        }),
        ControlRequest::SshCertApprove {
            request_id,
            ca_key_path,
            valid_for,
            serial,
            out,
        } => {
            let stored = store
                .get_ssh_cert_request(&request_id)?
                .ok_or_else(|| NodeError::SshCertRequestNotFound(request_id.clone()))?;
            let mut request = ssh_cert_request_from_stored(stored)?;
            request.status = SshCertRequestStatus::Approved;
            store.update_ssh_cert_request_status(request.id.as_str(), request.status.as_str())?;
            let public_key_path = out.clone().unwrap_or_else(|| {
                node.paths
                    .home()
                    .join("ssh-cert-requests")
                    .join(format!("{}.pub", request.id))
            });
            if let Some(parent) = public_key_path.parent() {
                std::fs::create_dir_all(parent)?;
            }
            std::fs::write(&public_key_path, &request.public_key)?;
            let valid_for = valid_for
                .or_else(|| request.requested_validity.clone())
                .unwrap_or_else(|| "+52w".to_owned());
            let signing_command = build_ssh_cert_sign_command(
                &request,
                &ca_key_path,
                &public_key_path,
                &valid_for,
                serial,
            )?;
            let approval = SshCertApproval {
                request_id: request.id,
                approved_by_node: NodeId::new(node.node_id.clone()),
                ca_key_path: ca_key_path.display().to_string(),
                key_id: request_id,
                valid_for,
                serial,
                output_path: Some(expected_openssh_cert_path(&public_key_path)),
                signing_command,
                note: "request approved; run the signing command on the CA/YubiKey machine, then import the resulting -cert.pub file".to_owned(),
            };
            Ok(ControlResponse::SshCertApproved { approval })
        }
        ControlRequest::SshCertImport {
            request_id,
            cert_path,
        } => {
            let certificate = std::fs::read_to_string(&cert_path)?;
            let record = SshCertificateRecord {
                id: certificate_id(&certificate),
                request_id: SshCertRequestId::new(request_id.clone()),
                certificate_fingerprint: ssh_public_key_fingerprint(&certificate),
                certificate,
                imported_at: UnixMillis(geth_store::now_ms()),
            };
            store.insert_ssh_certificate(&stored_from_ssh_certificate(&record))?;
            store.update_ssh_cert_request_status(
                &request_id,
                SshCertRequestStatus::Signed.as_str(),
            )?;
            Ok(ControlResponse::SshCertImported {
                certificate: record,
            })
        }
        ControlRequest::SshCertList => Ok(ControlResponse::SshCertList {
            requests: store
                .list_ssh_cert_requests()?
                .into_iter()
                .map(ssh_cert_request_from_stored)
                .collect::<Result<Vec<_>, _>>()?,
            certificates: store
                .list_ssh_certificates()?
                .into_iter()
                .map(ssh_certificate_from_stored)
                .collect(),
        }),
        ControlRequest::SshRevocationAdd {
            kind,
            target,
            reason,
        } => {
            let kind = kind
                .parse::<SshRevocationKind>()
                .map_err(|_| NodeError::InvalidSshRevocationKind(kind.clone()))?;
            let created_at = UnixMillis(geth_store::now_ms());
            let revocation = SshRevocationEntry {
                id: revocation_id(&kind, &target, created_at),
                kind,
                target,
                reason,
                created_at,
                published: true,
            };
            store.insert_ssh_revocation(&stored_from_ssh_revocation(&revocation))?;
            Ok(ControlResponse::SshRevocationAdded { revocation })
        }
        ControlRequest::SshRevocationList => Ok(ControlResponse::SshRevocationList {
            revocations: store
                .list_ssh_revocations()?
                .into_iter()
                .map(ssh_revocation_from_stored)
                .collect::<Result<Vec<_>, _>>()?,
        }),
        ControlRequest::SshRevocationExport {
            out,
            format,
            ca_public,
        } => {
            let revocations = store
                .list_ssh_revocations()?
                .into_iter()
                .map(ssh_revocation_from_stored)
                .collect::<Result<Vec<_>, _>>()?;
            let format = format
                .parse::<SshRevocationExportFormat>()
                .map_err(NodeError::SshIdentity)?;
            if let Some(parent) = out.parent() {
                std::fs::create_dir_all(parent)?;
            }
            let (body, note) = match format {
                SshRevocationExportFormat::Jsonl => {
                    let mut body = String::new();
                    for revocation in &revocations {
                        body.push_str(&serde_json::to_string(revocation)?);
                        body.push('\n');
                    }
                    (
                        body,
                        "JSONL geth revocation metadata; not an OpenSSH KRL binary".to_owned(),
                    )
                }
                SshRevocationExportFormat::OpenSshKrlSpec => (
                    openssh_krl_spec(&revocations)?,
                    "OpenSSH KRL specification; generate a binary KRL with ssh-keygen -k -f <krl> [-s <ca.pub>] <spec>".to_owned(),
                ),
                SshRevocationExportFormat::OpenSshKrl => {
                    write_openssh_krl(&revocations, &out, ca_public.as_deref())?;
                    (
                        String::new(),
                        "OpenSSH binary KRL generated with ssh-keygen; use ssh-keygen -Q -f <krl> <key-or-cert> to query it".to_owned(),
                    )
                }
            };
            if format != SshRevocationExportFormat::OpenSshKrl {
                std::fs::write(&out, body)?;
            }
            Ok(ControlResponse::SshRevocationExported {
                out,
                format: format.to_string(),
                count: revocations.len(),
                note,
            })
        }
        ControlRequest::SshRevocationImport { path, format } => {
            let body = std::fs::read_to_string(&path)?;
            let created_at = UnixMillis(geth_store::now_ms());
            let revocations = match format.as_str() {
                "jsonl" => body
                    .lines()
                    .filter(|line| !line.trim().is_empty())
                    .map(|line| serde_json::from_str(line).map_err(NodeError::from))
                    .collect::<Result<Vec<SshRevocationEntry>, NodeError>>()?,
                "openssh-krl-spec" | "krl-spec" => parse_openssh_krl_spec(&body)?
                    .into_iter()
                    .enumerate()
                    .map(|(index, (kind, target))| {
                        let created_at = UnixMillis(created_at.0 + index as i64);
                        SshRevocationEntry {
                            id: revocation_id(&kind, &target, created_at),
                            kind,
                            target,
                            reason: Some(format!("imported from {}", path.display())),
                            created_at,
                            published: true,
                        }
                    })
                    .collect(),
                "openssh-krl" | "krl" => {
                    return Err(NodeError::SshIdentity(
                        geth_ssh_identity::SshIdentityError::BinaryKrlImportUnsupported,
                    ));
                }
                _ => {
                    return Err(NodeError::SshIdentity(
                        geth_ssh_identity::SshIdentityError::InvalidRevocationExportFormat(
                            format.clone(),
                        ),
                    ));
                }
            };
            for revocation in &revocations {
                store.insert_ssh_revocation(&stored_from_ssh_revocation(revocation))?;
            }
            Ok(ControlResponse::SshRevocationImported {
                count: revocations.len(),
                revocations,
                format,
                note: "imported revocation metadata; binary OpenSSH KRL files cannot be enumerated, import JSONL or the KRL spec source instead".to_owned(),
            })
        }
        ControlRequest::DbAdd { name, path } => {
            geth_db::validate_db_name(&name).map_err(|_| NodeError::InvalidDbName(name.clone()))?;
            if !path.is_file() {
                return Err(NodeError::InvalidDbPath(path.display().to_string()));
            }
            let path = std::fs::canonicalize(path)?;
            let schema_metadata = geth_db::schema_metadata(&path)?;
            let crsqlite_changes = geth_db::crsqlite_change_metadata(&path)?;
            let resource_id = format!("resource:db:{name}");
            let db_id = format!("db:{name}");
            let resource = StoredResource {
                resource_id: resource_id.clone(),
                kind: ResourceKind::Db.to_string(),
                name: name.clone(),
                status: "active".to_owned(),
            };
            store.insert_resource(&resource)?;
            let stored = StoredDbResource {
                db_id,
                resource_id,
                name,
                path: path.display().to_string(),
            };
            store.insert_db_resource(&stored)?;
            Ok(ControlResponse::DbAdded {
                db: db_resource_from_stored_with_schema(&stored, schema_metadata, crsqlite_changes),
            })
        }
        ControlRequest::DbStatus { name } => {
            geth_db::validate_db_name(&name).map_err(|_| NodeError::InvalidDbName(name.clone()))?;
            let stored = store
                .get_db_resource_by_name(&name)?
                .ok_or_else(|| NodeError::DbNotFound(name.clone()))?;
            Ok(ControlResponse::DbStatus {
                db: db_resource_from_stored(&stored)?,
            })
        }
        ControlRequest::DbChanges {
            name,
            after_db_version,
            limit,
        } => {
            geth_db::validate_db_name(&name).map_err(|_| NodeError::InvalidDbName(name.clone()))?;
            let stored = store
                .get_db_resource_by_name(&name)?
                .ok_or_else(|| NodeError::DbNotFound(name.clone()))?;
            let batch = geth_db::extract_crsqlite_changes(
                Path::new(&stored.path),
                after_db_version,
                limit,
            )?;
            Ok(ControlResponse::DbChanges {
                db: db_resource_from_stored(&stored)?,
                batch,
            })
        }
        ControlRequest::KvCreate { name } => {
            geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
            let resource_id = format!("resource:kv:{name}");
            let kv_id = format!("kv:{name}");
            let resource = StoredResource {
                resource_id: resource_id.clone(),
                kind: ResourceKind::Kv.to_string(),
                name: name.clone(),
                status: "active".to_owned(),
            };
            store.insert_resource(&resource)?;
            let stored = StoredKvStore {
                kv_id,
                resource_id,
                name,
            };
            store.insert_kv_store(&stored)?;
            Ok(ControlResponse::KvCreated {
                kv: kv_resource_from_stored(&stored),
            })
        }
        ControlRequest::KvSet {
            name,
            key,
            value,
            subject,
        } => {
            geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
            geth_kv::validate_kv_key(&key).map_err(|_| NodeError::InvalidKvKey(key.clone()))?;
            let kv = store
                .get_kv_store_by_name(&name)?
                .ok_or_else(|| NodeError::KvNotFound(name.clone()))?;
            ensure_kv_write_authorized(&store, node, subject, &kv.resource_id, &key)?;
            let stored = StoredKvEntry {
                kv_id: kv.kv_id,
                key,
                value,
                updated_at_ms: geth_store::now_ms(),
            };
            store.set_kv_entry(&stored)?;
            Ok(ControlResponse::KvSet {
                entry: kv_entry_from_stored(stored),
            })
        }
        ControlRequest::KvGet { name, key } => {
            geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
            geth_kv::validate_kv_key(&key).map_err(|_| NodeError::InvalidKvKey(key.clone()))?;
            let kv = store
                .get_kv_store_by_name(&name)?
                .ok_or_else(|| NodeError::KvNotFound(name.clone()))?;
            let entry = store
                .get_kv_entry(&kv.kv_id, &key)?
                .map(kv_entry_from_stored);
            Ok(ControlResponse::KvGet { entry })
        }
        ControlRequest::DocumentCreate { name } => {
            geth_document::validate_document_name(&name)
                .map_err(|_| NodeError::InvalidDocumentName(name.clone()))?;
            let resource_id = format!("resource:document:{name}");
            let document_id = format!("document:{name}");
            let resource = StoredResource {
                resource_id: resource_id.clone(),
                kind: ResourceKind::Document.to_string(),
                name: name.clone(),
                status: "active".to_owned(),
            };
            store.insert_resource(&resource)?;
            let stored = StoredDocumentResource {
                document_id,
                resource_id,
                name,
                state_json: "{}".to_owned(),
                updated_at_ms: geth_store::now_ms(),
            };
            store.insert_document_resource(&stored)?;
            Ok(ControlResponse::DocumentCreated {
                document: document_resource_from_stored(&stored),
            })
        }
        ControlRequest::DocumentStatus { name } => {
            geth_document::validate_document_name(&name)
                .map_err(|_| NodeError::InvalidDocumentName(name.clone()))?;
            let stored = store
                .get_document_resource_by_name(&name)?
                .ok_or_else(|| NodeError::DocumentNotFound(name.clone()))?;
            Ok(ControlResponse::DocumentStatus {
                document: document_resource_from_stored(&stored),
            })
        }
        ControlRequest::DocumentSet { name, state_json } => {
            geth_document::validate_document_name(&name)
                .map_err(|_| NodeError::InvalidDocumentName(name.clone()))?;
            let mut stored = store
                .get_document_resource_by_name(&name)?
                .ok_or_else(|| NodeError::DocumentNotFound(name.clone()))?;
            stored.state_json = geth_document::normalize_document_state(&state_json)?;
            stored.updated_at_ms = geth_store::now_ms();
            store.insert_document_resource(&stored)?;
            Ok(ControlResponse::DocumentSet {
                state: document_state_from_stored(&stored),
            })
        }
        ControlRequest::DocumentGet { name } => {
            geth_document::validate_document_name(&name)
                .map_err(|_| NodeError::InvalidDocumentName(name.clone()))?;
            let stored = store
                .get_document_resource_by_name(&name)?
                .ok_or_else(|| NodeError::DocumentNotFound(name.clone()))?;
            Ok(ControlResponse::DocumentGet {
                state: document_state_from_stored(&stored),
            })
        }
        ControlRequest::PubsubPub { topic, message } => {
            geth_pubsub::validate_topic(&topic)?;
            geth_pubsub::validate_message(&message)?;
            let message = PubsubMessage {
                topic: topic.clone().into(),
                message,
                published_at: UnixMillis(geth_store::now_ms()),
            };
            let mut runtime = node
                .runtime
                .pubsub
                .lock()
                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
            runtime.messages.push_back(message.clone());
            while runtime.messages.len() > PUBSUB_RING_LIMIT {
                runtime.messages.pop_front();
            }
            Ok(ControlResponse::PubsubPublished { message })
        }
        ControlRequest::PubsubSub { topic } => {
            geth_pubsub::validate_topic(&topic)?;
            let runtime = node
                .runtime
                .pubsub
                .lock()
                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
            let messages = runtime
                .messages
                .iter()
                .filter(|message| message.topic.as_str() == topic)
                .cloned()
                .collect();
            Ok(ControlResponse::PubsubMessages {
                topic,
                messages,
                note: geth_pubsub::pubsub_storage_warning().to_owned(),
            })
        }
        ControlRequest::PipeListen { name } => {
            geth_pipe::validate_pipe_name(&name)?;
            let listener = PipeListener {
                id: format!("pipe:{name}").into(),
                name: name.clone(),
                listened_at: UnixMillis(geth_store::now_ms()),
                note: geth_pipe::local_pipe_runtime_note().to_owned(),
            };
            let mut runtime = node
                .runtime
                .pipes
                .lock()
                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
            runtime.listeners.insert(name, listener.clone());
            Ok(ControlResponse::PipeListening { listener })
        }
        ControlRequest::PipeConnect { target } => {
            geth_pipe::validate_pipe_name(&target)?;
            let mut runtime = node
                .runtime
                .pipes
                .lock()
                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
            let local_listener_found = runtime.listeners.contains_key(&target);
            let connection = PipeConnection {
                target,
                connected_at: UnixMillis(geth_store::now_ms()),
                local_listener_found,
                note: geth_pipe::local_pipe_runtime_note().to_owned(),
            };
            runtime.connections.push_back(connection.clone());
            while runtime.connections.len() > PIPE_CONNECTION_RING_LIMIT {
                runtime.connections.pop_front();
            }
            Ok(ControlResponse::PipeConnected { connection })
        }
        ControlRequest::ModuleStub { module, command } => {
            Ok(ControlResponse::NotImplemented { module, command })
        }
    }
}

fn stored_resource_to_descriptor(stored: StoredResource) -> Result<ResourceDescriptor, NodeError> {
    let kind = stored
        .kind
        .parse::<ResourceKind>()
        .map_err(|_| NodeError::InvalidResourceKind(stored.kind.clone()))?;
    Ok(ResourceDescriptor::local(
        ResourceId::new(stored.resource_id),
        kind,
        ResourceName::new(stored.name),
    ))
}

fn db_resource_from_stored(stored: &StoredDbResource) -> Result<DbResource, NodeError> {
    let path = Path::new(&stored.path);
    let metadata = path.metadata().ok();
    let (schema_metadata, crsqlite_changes) = if metadata
        .as_ref()
        .is_some_and(std::fs::Metadata::is_file)
    {
        (
            geth_db::schema_metadata(path).unwrap_or_else(|error| format!("unavailable:{error}")),
            geth_db::crsqlite_change_metadata(path)
                .unwrap_or_else(|error| geth_db::CrSqliteChangeMetadata::error(error.to_string())),
        )
    } else {
        (
            "unavailable:path-missing".to_owned(),
            geth_db::CrSqliteChangeMetadata::error("path-missing"),
        )
    };
    Ok(db_resource_from_stored_with_schema(
        stored,
        schema_metadata,
        crsqlite_changes,
    ))
}

fn db_resource_from_stored_with_schema(
    stored: &StoredDbResource,
    schema_metadata: String,
    crsqlite_changes: geth_db::CrSqliteChangeMetadata,
) -> DbResource {
    let path = Path::new(&stored.path);
    let metadata = path.metadata().ok();
    DbResource {
        id: stored.db_id.clone().into(),
        resource: stored.resource_id.clone().into(),
        name: stored.name.clone(),
        path: stored.path.clone(),
        path_exists: metadata.as_ref().is_some_and(std::fs::Metadata::is_file),
        size_bytes: metadata.map(|metadata| metadata.len()),
        schema_metadata,
        crsqlite_changes,
        sync_status: "local-only".to_owned(),
    }
}

fn kv_resource_from_stored(stored: &StoredKvStore) -> KvResource {
    KvResource {
        id: stored.kv_id.clone().into(),
        resource: stored.resource_id.clone().into(),
        name: stored.name.clone(),
        sync_status: "local-only".to_owned(),
    }
}

fn kv_entry_from_stored(stored: StoredKvEntry) -> KvEntry {
    KvEntry {
        store: stored.kv_id.into(),
        key: stored.key,
        value: stored.value,
    }
}

fn document_resource_from_stored(stored: &StoredDocumentResource) -> DocumentResource {
    DocumentResource {
        id: stored.document_id.clone().into(),
        resource: stored.resource_id.clone().into(),
        name: stored.name.clone(),
        sync_status: "local-only".to_owned(),
        state_bytes: stored.state_json.len() as u64,
    }
}

fn document_state_from_stored(stored: &StoredDocumentResource) -> DocumentState {
    DocumentState {
        document: document_resource_from_stored(stored),
        state_json: stored.state_json.clone(),
        updated_at: UnixMillis(stored.updated_at_ms),
    }
}

fn file_root_from_stored(stored: &StoredFileRoot) -> FileRoot {
    FileRoot {
        id: stored.root_id.clone(),
        resource: stored.resource_id.clone(),
        name: stored.name.clone(),
        path: stored.path.clone(),
        latest_tree: stored.latest_tree_hash.clone().map(Into::into),
        updated_at_ms: stored.updated_at_ms,
    }
}

fn discovered_peer_from_stored(stored: StoredPeerCard) -> Result<DiscoveredPeer, NodeError> {
    let card: PeerCard = serde_json::from_str(&stored.card_json)?;
    Ok(DiscoveredPeer::candidate(
        card,
        UnixMillis(stored.updated_at_ms),
        DiscoverySource::Imported,
    )?)
}

fn file_conflict_from_stored(stored: StoredFileConflict) -> Result<FileConflict, NodeError> {
    Ok(FileConflict {
        id: stored.conflict_id,
        root: stored.root_name,
        resource: stored.resource_id,
        path: stored.path,
        kind: FileConflictKind::parse(&stored.kind)?,
        status: FileConflictStatus::parse(&stored.status)?,
        base_tree: stored.base_tree_hash.map(Into::into),
        local_tree: stored.local_tree_hash.map(Into::into),
        remote_tree: stored.remote_tree_hash.map(Into::into),
        detail: stored.detail,
        resolution: stored
            .resolution
            .as_deref()
            .map(FileConflictResolution::parse)
            .transpose()?,
        resolution_note: stored.resolution_note,
        created_at_ms: stored.created_at_ms,
        resolved_at_ms: stored.resolved_at_ms,
    })
}

fn ensure_resource_exists(store: &Store, resource_id: &str) -> Result<(), NodeError> {
    if store
        .list_resources()?
        .into_iter()
        .any(|resource| resource.resource_id == resource_id)
    {
        Ok(())
    } else {
        Err(NodeError::ResourceNotFound(resource_id.to_owned()))
    }
}

fn ensure_kv_write_authorized(
    store: &Store,
    node: &LocalNode,
    subject: Option<String>,
    resource_id: &str,
    key: &str,
) -> Result<(), NodeError> {
    let Some(subject) = subject else {
        return Ok(());
    };
    if subject == node.node_id || subject == node.agent_id {
        return Ok(());
    }

    let capability = format!("kv.write_key:{key}");
    let ops = load_auth_ops_for_resource(store, resource_id)?;
    let explanation = geth_auth::explain_auth_ops(
        &ops,
        PrincipalId::new(subject.clone()),
        ResourceId::new(resource_id.to_owned()),
        Capability::new(capability.clone()),
    );
    if explanation.allowed {
        Ok(())
    } else {
        Err(NodeError::Unauthorized(format!(
            "{subject} lacks {capability} on {resource_id}: {}",
            explanation.reason
        )))
    }
}

fn create_resource_secret(
    store: &Store,
    resource_id: &str,
    epoch: u64,
) -> Result<ResourceMasterSecret, NodeError> {
    let created_at = UnixMillis(geth_store::now_ms());
    let secret_id = format!(
        "secret:{}",
        geth_crypto::blake3_hex(format!("{resource_id}\0{epoch}\0{}", created_at.0).as_bytes())
    );
    let stored = StoredResourceSecret {
        secret_id,
        resource_id: resource_id.to_owned(),
        epoch,
        status: "active".to_owned(),
        created_at_ms: created_at.0,
    };
    store.insert_resource_secret(&stored)?;
    Ok(resource_secret_from_stored(stored))
}

fn resource_secret_from_stored(stored: StoredResourceSecret) -> ResourceMasterSecret {
    ResourceMasterSecret {
        id: stored.secret_id.into(),
        resource: stored.resource_id.into(),
        epoch: stored.epoch,
        created_at: UnixMillis(stored.created_at_ms),
    }
}

fn load_bearer_access(store: &Store) -> Result<Vec<BearerAccess>, NodeError> {
    let ops = store
        .list_auth_ops()?
        .into_iter()
        .map(|stored| serde_json::from_str(&stored.op_json).map_err(NodeError::from))
        .collect::<Result<Vec<AuthOp>, NodeError>>()?;
    let view = geth_auth::reduce_auth_ops(&ops);
    Ok(view
        .bearer_access
        .into_values()
        .map(|record| BearerAccess {
            secret: record.secret,
            resource: record.resource,
            capabilities: record.capabilities,
            expires_at: record.expires_at,
            may_delegate: false,
        })
        .collect())
}

fn store_auth_op(store: &Store, op: &AuthOp) -> Result<(), NodeError> {
    store.insert_auth_op(&StoredAuthOp {
        op_id: op.id.to_string(),
        resource_id: op.resource.to_string(),
        op_json: serde_json::to_string(op)?,
        created_at_ms: op.created_at.0,
    })?;
    Ok(())
}

fn load_auth_ops_for_resource(store: &Store, resource: &str) -> Result<Vec<AuthOp>, NodeError> {
    store
        .list_auth_ops_for_resource(resource)?
        .into_iter()
        .map(|stored| serde_json::from_str(&stored.op_json).map_err(NodeError::from))
        .collect()
}

fn store_keychain_op(store: &Store, op: &KeychainOp) -> Result<(), NodeError> {
    store.insert_keychain_op(&StoredKeychainOp {
        op_id: op.id.to_string(),
        op_json: serde_json::to_string(op)?,
        created_at_ms: op.created_at.0,
    })?;
    Ok(())
}

fn load_keychain_ops(store: &Store) -> Result<Vec<KeychainOp>, NodeError> {
    store
        .list_keychain_ops()?
        .into_iter()
        .map(|stored| serde_json::from_str(&stored.op_json).map_err(NodeError::from))
        .collect()
}

fn generated_grant_id(subject: &str, resource: &str, capability: &str) -> String {
    format!(
        "grant:{}",
        geth_crypto::blake3_hex(format!("{subject}\0{resource}\0{capability}").as_bytes())
    )
}

fn generated_file_conflict_id(root: &str, path: &str, kind: &str, created_at_ms: i64) -> String {
    format!(
        "file-conflict:{}",
        geth_crypto::blake3_hex(format!("{created_at_ms}\0{root}\0{path}\0{kind}").as_bytes())
    )
}

fn generated_auth_op_id(
    kind: &str,
    resource: &str,
    stable_id: &str,
    created_at: UnixMillis,
) -> AuthOpId {
    AuthOpId::new(format!(
        "auth-op:{}",
        geth_crypto::blake3_hex(
            format!("{}\0{kind}\0{resource}\0{stable_id}", created_at.0).as_bytes()
        )
    ))
}

fn generated_keychain_op_id(kind: &str, stable_id: &str, created_at: UnixMillis) -> AuthOpId {
    AuthOpId::new(format!(
        "keychain-op:{}",
        geth_crypto::blake3_hex(format!("{}\0{kind}\0{stable_id}", created_at.0).as_bytes())
    ))
}

fn stable_node_id(agent_id: &str) -> String {
    format!("node:{agent_id}")
}

async fn start_daemon_iroh_endpoint(
    node: &mut LocalNode,
) -> Result<Option<GethIrohEndpoint>, NodeError> {
    let node_config = GethConfig::load(&node.paths.config_file())?;
    let relay_mode = node_config.iroh.relay_mode.clone();
    let relay_mode_label = relay_mode.label();
    let local_discovery = node_config.iroh.local_discovery;
    let iroh_relay_mode = config_relay_mode_to_iroh(&relay_mode, &node_config.iroh.relay_maps);
    let mut config = GethIrohConfig::local_with_relay(node.paths.iroh_key(), iroh_relay_mode);
    config.local_discovery = local_discovery;
    match geth_iroh::start_endpoint(&config).await {
        Ok(endpoint) => {
            let status = endpoint.status();
            if let Some(endpoint_id) = &status.endpoint_id {
                let store = Store::open(&node.paths.metadata_db())?;
                store.upsert_node_endpoint(endpoint_id, &node.node_id, &node.agent_id, "iroh")?;
            }
            node.iroh_status = status;
            *node
                .iroh_endpoint
                .lock()
                .map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(endpoint.clone());
            Ok(Some(endpoint))
        }
        Err(error) => {
            node.iroh_status = EndpointStatus {
                enabled: false,
                endpoint_id: None,
                relay_mode: relay_mode_label,
                local_discovery,
                note: format!("Iroh endpoint failed to start: {error}"),
            };
            Ok(None)
        }
    }
}

fn config_relay_mode_to_iroh(
    mode: &RelayMode,
    relay_maps: &std::collections::BTreeMap<String, geth_config::RelayMapConfig>,
) -> GethRelayMode {
    match mode {
        RelayMode::Disabled => GethRelayMode::Disabled,
        RelayMode::Default => GethRelayMode::Default,
        RelayMode::Staging => GethRelayMode::Staging,
        RelayMode::Custom { map } => {
            let relay_map = relay_maps
                .get(map)
                .expect("custom relay map was validated during config load");
            GethRelayMode::Custom {
                name: map.clone(),
                relay_urls: relay_map.urls.clone(),
            }
        }
    }
}

fn expected_openssh_cert_path(public_key_path: &Path) -> String {
    let text = public_key_path.display().to_string();
    if let Some(prefix) = text.strip_suffix(".pub") {
        format!("{prefix}-cert.pub")
    } else {
        format!("{text}-cert.pub")
    }
}

fn stored_from_ssh_cert_request(request: &SshCertRequest) -> StoredSshCertRequest {
    StoredSshCertRequest {
        request_id: request.id.to_string(),
        requester_node: request.requester_node.to_string(),
        public_key: request.public_key.clone(),
        public_key_fingerprint: request.public_key_fingerprint.clone(),
        cert_kind: request.cert_kind.to_string(),
        principals: request.principals.clone(),
        requested_validity: request.requested_validity.clone(),
        renewal_of: request.renewal_of.as_ref().map(ToString::to_string),
        reason: request.reason.clone(),
        status: request.status.to_string(),
        created_at_ms: request.created_at.0,
    }
}

fn ssh_cert_request_from_stored(stored: StoredSshCertRequest) -> Result<SshCertRequest, NodeError> {
    let cert_kind = stored
        .cert_kind
        .parse::<SshCertKind>()
        .map_err(|_| NodeError::InvalidSshCertKind(stored.cert_kind.clone()))?;
    let status = stored
        .status
        .parse::<SshCertRequestStatus>()
        .map_err(|_| NodeError::InvalidSshCertStatus(stored.status.clone()))?;
    Ok(SshCertRequest {
        id: SshCertRequestId::new(stored.request_id),
        requester_node: NodeId::new(stored.requester_node),
        public_key: stored.public_key,
        public_key_fingerprint: stored.public_key_fingerprint,
        cert_kind,
        principals: stored.principals,
        requested_validity: stored.requested_validity,
        renewal_of: stored.renewal_of.map(SshCertId::new),
        reason: stored.reason,
        status,
        created_at: UnixMillis(stored.created_at_ms),
    })
}

fn stored_from_ssh_certificate(certificate: &SshCertificateRecord) -> StoredSshCertificate {
    StoredSshCertificate {
        cert_id: certificate.id.to_string(),
        request_id: certificate.request_id.to_string(),
        certificate: certificate.certificate.clone(),
        certificate_fingerprint: certificate.certificate_fingerprint.clone(),
        imported_at_ms: certificate.imported_at.0,
    }
}

fn ssh_certificate_from_stored(stored: StoredSshCertificate) -> SshCertificateRecord {
    SshCertificateRecord {
        id: SshCertId::new(stored.cert_id),
        request_id: SshCertRequestId::new(stored.request_id),
        certificate: stored.certificate,
        certificate_fingerprint: stored.certificate_fingerprint,
        imported_at: UnixMillis(stored.imported_at_ms),
    }
}

fn stored_from_ssh_revocation(revocation: &SshRevocationEntry) -> StoredSshRevocation {
    StoredSshRevocation {
        revocation_id: revocation.id.to_string(),
        kind: revocation.kind.to_string(),
        target: revocation.target.clone(),
        reason: revocation.reason.clone(),
        created_at_ms: revocation.created_at.0,
        published: revocation.published,
    }
}

fn ssh_revocation_from_stored(
    stored: StoredSshRevocation,
) -> Result<SshRevocationEntry, NodeError> {
    let kind = stored
        .kind
        .parse::<SshRevocationKind>()
        .map_err(|_| NodeError::InvalidSshRevocationKind(stored.kind.clone()))?;
    Ok(SshRevocationEntry {
        id: geth_types::SshRevocationId::new(stored.revocation_id),
        kind,
        target: stored.target,
        reason: stored.reason,
        created_at: UnixMillis(stored.created_at_ms),
        published: stored.published,
    })
}

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

    fn write_offline_iroh_config(paths: &GethPaths) {
        std::fs::write(
            paths.config_file(),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }

    #[test]
    fn lan_discovery_address_selection_uses_iroh_direct_addresses() {
        let key = AgentKey::generate();
        let card = PeerCard::signed(
            "node:test".into(),
            &key,
            vec![EndpointCandidate {
                endpoint_id: "endpoint:test".to_owned(),
                relay_url: None,
                direct_addresses: vec![
                    "127.0.0.1:1111".to_owned(),
                    "127.0.0.2:1111".to_owned(),
                    "127.0.0.3:2222".to_owned(),
                ],
                source: DiscoverySource::Mdns,
            }],
            UnixMillis(1),
        )
        .expect("peer card");

        let (port, addrs) = lan_discovery_addrs(&card).expect("lan addresses");

        assert_eq!(port, 1111);
        assert_eq!(addrs.len(), 2);
    }

    #[tokio::test]
    async fn peer_ping_uses_signed_peer_card_over_iroh() {
        let left_home = tempfile::tempdir().expect("left home");
        let right_home = tempfile::tempdir().expect("right home");
        let left_paths = GethPaths::from_home(left_home.path());
        let right_paths = GethPaths::from_home(right_home.path());
        let mut left = init_node(&left_paths).expect("init left");
        let mut right = init_node(&right_paths).expect("init right");
        write_offline_iroh_config(&left_paths);
        write_offline_iroh_config(&right_paths);

        let Some(left_endpoint) = start_daemon_iroh_endpoint(&mut left)
            .await
            .expect("left iroh")
        else {
            eprintln!("skipping peer ping assertion; left Iroh endpoint unavailable");
            return;
        };
        let Some(right_endpoint) = start_daemon_iroh_endpoint(&mut right)
            .await
            .expect("right iroh")
        else {
            eprintln!("skipping peer ping assertion; right Iroh endpoint unavailable");
            left_endpoint.shutdown().await;
            return;
        };
        spawn_iroh_control_accept_loop(right.clone(), right_endpoint.clone());

        let exported = handle_request_async(&right, ControlRequest::PeerCardExport { out: None })
            .await
            .expect("export right peer card");
        let right_card = match exported {
            ControlResponse::PeerCardExported { card, .. } => card,
            other => panic!("unexpected export response: {other:?}"),
        };
        assert!(!right_card.endpoints[0].direct_addresses.is_empty());
        Store::open(&left_paths.metadata_db())
            .expect("open left store")
            .upsert_peer_card(&StoredPeerCard {
                peer_id: right_card.node_id.to_string(),
                card_json: serde_json::to_string(&right_card).expect("card json"),
                updated_at_ms: geth_store::now_ms(),
            })
            .expect("insert right peer");
        let right_blob = LocalCas::new(right.paths.cas_dir())
            .add_bytes(b"remote cas bytes")
            .expect("right cas add");
        let right_pubkey = right_home.path().join("request.pub");
        std::fs::write(
            &right_pubkey,
            "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGV0aA== node\n",
        )
        .expect("write right public key");
        let requested = handle_request(
            &right,
            ControlRequest::SshCertRequest {
                public_key_path: right_pubkey,
                cert_kind: "user".to_owned(),
                principals: vec!["eric".to_owned()],
                requested_validity: Some("+52w".to_owned()),
                renewal_of: None,
                reason: Some("test sync".to_owned()),
            },
        )
        .expect("right ssh cert request");
        let right_request_id = match requested {
            ControlResponse::SshCertRequested { request } => request.id,
            other => panic!("unexpected SSH cert request response: {other:?}"),
        };
        let added_revocation = handle_request(
            &right,
            ControlRequest::SshRevocationAdd {
                kind: "key-id".to_owned(),
                target: "old-node-key".to_owned(),
                reason: Some("test sync".to_owned()),
            },
        )
        .expect("right ssh revocation");
        let right_revocation_id = match added_revocation {
            ControlResponse::SshRevocationAdded { revocation } => revocation.id,
            other => panic!("unexpected SSH revocation response: {other:?}"),
        };

        let ping = handle_request_async(
            &left,
            ControlRequest::PeerPing {
                node: right_card.node_id.to_string(),
            },
        )
        .await
        .expect("peer ping");

        match ping {
            ControlResponse::PeerPinged {
                peer_node_id,
                peer_agent_id,
                alpn,
                note,
                ..
            } => {
                assert_eq!(peer_node_id, right.node_id);
                assert_eq!(peer_agent_id, right.agent_id);
                assert_eq!(alpn, "/geth/control/1");
                assert!(note.contains("does not grant resource capabilities"));
            }
            other => panic!("unexpected ping response: {other:?}"),
        }

        let denied = handle_request_async(
            &left,
            ControlRequest::PeerAuthCheck {
                node: right_card.node_id.to_string(),
                resource: "resource:cas:local".to_owned(),
                capability: "cas.fetch".to_owned(),
            },
        )
        .await
        .expect("denied peer auth check");
        match denied {
            ControlResponse::PeerAuthChecked {
                allowed,
                reason,
                note,
                ..
            } => {
                assert!(!allowed);
                assert!(reason.contains("no active direct or group grant"));
                assert!(note.contains("authenticated endpoint/card binding"));
            }
            other => panic!("unexpected denied auth response: {other:?}"),
        }

        let denied_fetch = handle_request_async(
            &left,
            ControlRequest::CasFetch {
                node: right_card.node_id.to_string(),
                hash: right_blob.hash.clone(),
            },
        )
        .await
        .expect("denied cas fetch");
        match denied_fetch {
            ControlResponse::CasFetched {
                allowed,
                reason,
                size_bytes,
                ..
            } => {
                assert!(!allowed);
                assert_eq!(size_bytes, 0);
                assert!(reason.contains("no active direct or group grant"));
                assert!(
                    !LocalCas::new(left.paths.cas_dir())
                        .has(&right_blob.hash)
                        .expect("left cas has after denied fetch")
                );
            }
            other => panic!("unexpected denied CAS fetch response: {other:?}"),
        }

        handle_request(
            &right,
            ControlRequest::AuthGrant {
                subject: left.node_id.clone(),
                resource: "resource:cas:local".to_owned(),
                capability: "cas.fetch".to_owned(),
                grant_id: Some("grant:left-cas-fetch".to_owned()),
            },
        )
        .expect("grant left peer");

        let allowed = handle_request_async(
            &left,
            ControlRequest::PeerAuthCheck {
                node: right_card.node_id.to_string(),
                resource: "resource:cas:local".to_owned(),
                capability: "cas.fetch".to_owned(),
            },
        )
        .await
        .expect("allowed peer auth check");
        match allowed {
            ControlResponse::PeerAuthChecked {
                allowed,
                reason,
                evaluated_ops,
                ..
            } => {
                assert!(allowed);
                assert!(reason.contains("direct grant"));
                assert_eq!(evaluated_ops, 1);
            }
            other => panic!("unexpected allowed auth response: {other:?}"),
        }

        let fetched = handle_request_async(
            &left,
            ControlRequest::CasFetch {
                node: right_card.node_id.to_string(),
                hash: right_blob.hash.clone(),
            },
        )
        .await
        .expect("allowed cas fetch");
        match fetched {
            ControlResponse::CasFetched {
                allowed,
                hash,
                size_bytes,
                reason,
                note,
                ..
            } => {
                assert!(allowed);
                assert_eq!(hash, right_blob.hash);
                assert_eq!(size_bytes, right_blob.size_bytes);
                assert!(reason.contains("direct grant"));
                assert!(note.contains("required cas.fetch"));
                assert_eq!(
                    LocalCas::new(left.paths.cas_dir())
                        .read_bytes(&right_blob.hash)
                        .expect("left cas read fetched blob"),
                    b"remote cas bytes"
                );
            }
            other => panic!("unexpected allowed CAS fetch response: {other:?}"),
        }

        let denied_cert_sync = handle_request_async(
            &left,
            ControlRequest::SshCertSync {
                node: right_card.node_id.to_string(),
            },
        )
        .await
        .expect("denied ssh cert sync");
        match denied_cert_sync {
            ControlResponse::SshCertSynced {
                allowed,
                requests_imported,
                certificates_imported,
                reason,
                ..
            } => {
                assert!(!allowed);
                assert_eq!(requests_imported, 0);
                assert_eq!(certificates_imported, 0);
                assert!(reason.contains("no active direct or group grant"));
            }
            other => panic!("unexpected denied SSH cert sync response: {other:?}"),
        }

        handle_request(
            &right,
            ControlRequest::AuthGrant {
                subject: left.node_id.clone(),
                resource: "resource:ssh:certs".to_owned(),
                capability: "ssh_cert.sync".to_owned(),
                grant_id: Some("grant:left-ssh-cert-sync".to_owned()),
            },
        )
        .expect("grant left ssh cert sync");
        handle_request(
            &right,
            ControlRequest::AuthGrant {
                subject: left.node_id.clone(),
                resource: "resource:ssh:revocations".to_owned(),
                capability: "ssh_revocation.sync".to_owned(),
                grant_id: Some("grant:left-ssh-revocation-sync".to_owned()),
            },
        )
        .expect("grant left ssh revocation sync");

        let cert_sync = handle_request_async(
            &left,
            ControlRequest::SshCertSync {
                node: right_card.node_id.to_string(),
            },
        )
        .await
        .expect("allowed ssh cert sync");
        match cert_sync {
            ControlResponse::SshCertSynced {
                allowed,
                requests_imported,
                certificates_imported,
                reason,
                note,
                ..
            } => {
                assert!(allowed);
                assert_eq!(requests_imported, 1);
                assert_eq!(certificates_imported, 0);
                assert!(reason.contains("direct grant"));
                assert!(note.contains("ssh_cert.sync"));
            }
            other => panic!("unexpected allowed SSH cert sync response: {other:?}"),
        }
        let left_requests = Store::open(&left_paths.metadata_db())
            .expect("open left after cert sync")
            .list_ssh_cert_requests()
            .expect("list synced requests");
        assert!(
            left_requests
                .iter()
                .any(|request| request.request_id == right_request_id.as_str())
        );

        let revocation_sync = handle_request_async(
            &left,
            ControlRequest::SshRevocationSync {
                node: right_card.node_id.to_string(),
            },
        )
        .await
        .expect("allowed ssh revocation sync");
        match revocation_sync {
            ControlResponse::SshRevocationSynced {
                allowed,
                revocations_imported,
                reason,
                note,
                ..
            } => {
                assert!(allowed);
                assert_eq!(revocations_imported, 1);
                assert!(reason.contains("direct grant"));
                assert!(note.contains("ssh_revocation.sync"));
            }
            other => panic!("unexpected allowed SSH revocation sync response: {other:?}"),
        }
        let left_revocations = Store::open(&left_paths.metadata_db())
            .expect("open left after revocation sync")
            .list_ssh_revocations()
            .expect("list synced revocations");
        assert!(
            left_revocations
                .iter()
                .any(|revocation| revocation.revocation_id == right_revocation_id.as_str())
        );

        tokio::time::sleep(Duration::from_millis(2)).await;
        let second_pubkey = right_home.path().join("request-2.pub");
        std::fs::write(
            &second_pubkey,
            "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGV0aDI= node2\n",
        )
        .expect("write second public key");
        let second_requested = handle_request(
            &right,
            ControlRequest::SshCertRequest {
                public_key_path: second_pubkey,
                cert_kind: "user".to_owned(),
                principals: vec!["admin".to_owned()],
                requested_validity: Some("+4w".to_owned()),
                renewal_of: None,
                reason: Some("background sync test".to_owned()),
            },
        )
        .expect("right second ssh cert request");
        let second_request_id = match second_requested {
            ControlResponse::SshCertRequested { request } => request.id,
            other => panic!("unexpected second SSH cert request response: {other:?}"),
        };
        let second_revocation = handle_request(
            &right,
            ControlRequest::SshRevocationAdd {
                kind: "key-id".to_owned(),
                target: "newly-revoked-key".to_owned(),
                reason: Some("background sync test".to_owned()),
            },
        )
        .expect("right second ssh revocation");
        let second_revocation_id = match second_revocation {
            ControlResponse::SshRevocationAdded { revocation } => revocation.id,
            other => panic!("unexpected second SSH revocation response: {other:?}"),
        };

        run_live_sync_once(&left)
            .await
            .expect("background live sync tick");
        let left_store = Store::open(&left_paths.metadata_db()).expect("open left after live sync");
        assert!(
            left_store
                .list_ssh_cert_requests()
                .expect("list live-synced requests")
                .iter()
                .any(|request| request.request_id == second_request_id.as_str())
        );
        assert!(
            left_store
                .list_ssh_revocations()
                .expect("list live-synced revocations")
                .iter()
                .any(|revocation| revocation.revocation_id == second_revocation_id.as_str())
        );

        left_endpoint.shutdown().await;
        right_endpoint.shutdown().await;
    }
}