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

use geth_auth::{AuthExplanation, AuthOp, AuthOpSignature};
use geth_cas::{FileConflict, FileRoot, FileRootScan};
use geth_db::{CrSqliteChangeBatch, DbResource};
use geth_discovery::{DiscoveredPeer, PeerCard};
use geth_document::{DocumentResource, DocumentState};
use geth_keychain::{KeychainOp, KeychainOpSignature, NodeEnrollmentRequest, NodeRecord};
use geth_kv::{KvEntry, KvResource, KvSyncEntry};
use geth_overlay::{OverlayJoinPlan, OverlayNetworkStatus, OverlayPlan};
use geth_pipe::{PipeConnection, PipeListener, PipeMessage};
use geth_pubsub::PubsubMessage;
use geth_resource::ResourceDescriptor;
use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSecret};
use geth_ssh_identity::{
    SshCertApproval, SshCertRequest, SshCertificateRecord, SshDistributionLogEntry,
    SshRevocationEntry,
};
use geth_ssh_proxy::SshProxyConnection;
use geth_types::BlobHash;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum ControlRequest {
    Status,
    NodeId,
    PeerCardExport {
        out: Option<PathBuf>,
    },
    PeerCardImport {
        path: PathBuf,
    },
    PeerCardList,
    PeerPing {
        node: String,
    },
    PeerAuthCheck {
        node: String,
        resource: String,
        capability: String,
    },
    ResourceList,
    ResourceCreate {
        kind: String,
        name: String,
    },
    OverlayStatus,
    OverlayPlan {
        name: String,
        cidr: Option<String>,
    },
    OverlayJoin {
        name: String,
        secret: String,
        cidr: Option<String>,
    },
    OverlayLeave {
        name: String,
    },
    CasAdd {
        path: PathBuf,
    },
    CasAddPrivate {
        resource: String,
        path: PathBuf,
    },
    CasGet {
        hash: BlobHash,
        out: PathBuf,
    },
    CasGetPrivate {
        resource: String,
        hash: BlobHash,
        out: PathBuf,
    },
    CasFetch {
        node: String,
        hash: BlobHash,
        bearer_secret: Option<String>,
    },
    CasHash {
        path: PathBuf,
    },
    CasHas {
        hash: BlobHash,
    },
    CasPin {
        hash: BlobHash,
    },
    CasUnpin {
        hash: BlobHash,
    },
    CasCleanup {
        dry_run: bool,
    },
    CasProviders {
        hash: BlobHash,
    },
    CasList,
    CasRootAdd {
        name: String,
        path: PathBuf,
    },
    CasRootList,
    CasRootScan {
        name: String,
    },
    CasRootSync {
        node: String,
        name: String,
        bearer_secret: Option<String>,
    },
    CasRootApply {
        source: String,
        target: PathBuf,
        dry_run: bool,
    },
    CasConflictRecord {
        root: String,
        path: String,
        kind: String,
        detail: String,
        base_tree: Option<BlobHash>,
        local_tree: Option<BlobHash>,
        remote_tree: Option<BlobHash>,
    },
    CasConflictList {
        root: Option<String>,
    },
    CasConflictResolve {
        conflict_id: String,
        resolution: String,
        note: Option<String>,
    },
    NodeList,
    NodeRename {
        node: String,
        name: String,
        signing_key_path: Option<PathBuf>,
    },
    NodeRevoke {
        node: String,
        signing_key_path: Option<PathBuf>,
    },
    NodeGrant {
        node: String,
        resource: String,
        capability: String,
        grant_id: Option<String>,
        signing_key_path: Option<PathBuf>,
        admin_key_path: Option<PathBuf>,
    },
    NodeRevokeGrant {
        resource: String,
        grant_id: String,
        signing_key_path: Option<PathBuf>,
        admin_key_path: Option<PathBuf>,
    },
    NodeEndpointAdd {
        node: String,
        endpoint: String,
        signing_key_path: Option<PathBuf>,
    },
    NodeEndpointRevoke {
        node: String,
        endpoint: String,
        signing_key_path: Option<PathBuf>,
    },
    NodeEnrollRequest {
        node_name: String,
        capabilities: Vec<String>,
        reason: Option<String>,
        out: Option<PathBuf>,
    },
    NodeEnrollSubmit {
        owner_node: String,
        request_id: Option<String>,
        path: Option<PathBuf>,
    },
    NodeEnrollImport {
        path: PathBuf,
    },
    NodeEnrollList {
        status: Option<String>,
    },
    NodeEnrollApprove {
        request_id: String,
        signing_key_path: PathBuf,
        admin_key_path: Option<PathBuf>,
        node_name: Option<String>,
        capabilities: Vec<String>,
    },
    NodeEnrollSync {
        owner_node: String,
    },
    KeychainInit {
        admin_key_path: Option<PathBuf>,
        signing_key_path: Option<PathBuf>,
    },
    KeychainStatus,
    KeychainSync {
        node: String,
    },
    AuthSync {
        node: String,
    },
    SyncStatus,
    SyncNow {
        node: Option<String>,
    },
    SecretStatus,
    SecretCreate {
        resource: String,
    },
    SecretRotate {
        resource: String,
    },
    SecretBearerCreate {
        resource: String,
        capabilities: Vec<String>,
        expires_at_ms: Option<i64>,
    },
    SecretBearerList,
    SecretBearerChallenge {
        resource: String,
        capabilities: Vec<String>,
    },
    SecretBearerProve {
        secret: String,
        resource: String,
        capabilities: Vec<String>,
        nonce: String,
    },
    SecretBearerVerify {
        secret: String,
        resource: String,
        capabilities: Vec<String>,
        nonce: String,
        response: String,
    },
    SecretBearerRevoke {
        resource: String,
        secret: String,
    },
    AuthExplain {
        subject: String,
        resource: String,
        capability: String,
    },
    AuthGrant {
        subject: String,
        resource: String,
        capability: String,
        grant_id: Option<String>,
        signing_key_path: Option<PathBuf>,
        admin_key_path: Option<PathBuf>,
    },
    AuthRevoke {
        resource: String,
        grant_id: String,
        signing_key_path: Option<PathBuf>,
        admin_key_path: Option<PathBuf>,
    },
    SshCertRequest {
        public_key_path: PathBuf,
        cert_kind: String,
        principals: Vec<String>,
        requested_validity: Option<String>,
        renewal_of: Option<String>,
        reason: Option<String>,
        subject: Option<String>,
    },
    SshCertRequests {
        subject: Option<String>,
    },
    SshCertApprove {
        request_id: String,
        ca_key_path: PathBuf,
        valid_for: Option<String>,
        serial: Option<u64>,
        out: Option<PathBuf>,
        sign: bool,
        subject: Option<String>,
    },
    SshCertImport {
        request_id: String,
        cert_path: PathBuf,
        subject: Option<String>,
    },
    SshCertList {
        subject: Option<String>,
    },
    SshCertSync {
        node: String,
        bearer_secret: Option<String>,
    },
    SshRevocationAdd {
        kind: String,
        target: String,
        reason: Option<String>,
        subject: Option<String>,
    },
    SshRevocationList {
        subject: Option<String>,
    },
    SshRevocationExport {
        out: PathBuf,
        format: String,
        ca_public: Option<PathBuf>,
        subject: Option<String>,
    },
    SshRevocationImport {
        path: PathBuf,
        format: String,
        subject: Option<String>,
    },
    SshRevocationSync {
        node: String,
        bearer_secret: Option<String>,
    },
    SshProxyConnect {
        node: String,
        bearer_secret: Option<String>,
    },
    SshProxyStream {
        node: String,
        bearer_secret: Option<String>,
    },
    SshAdminShell {
        node: String,
        command: String,
        bearer_secret: Option<String>,
    },
    DbAdd {
        name: String,
        path: PathBuf,
    },
    DbStatus {
        name: String,
    },
    DbChanges {
        name: String,
        after_db_version: Option<i64>,
        limit: u32,
    },
    DbSync {
        node: String,
        name: String,
        limit: u32,
        bearer_secret: Option<String>,
    },
    KvCreate {
        name: String,
    },
    KvSet {
        name: String,
        key: String,
        value: String,
        subject: Option<String>,
    },
    KvGet {
        name: String,
        key: String,
    },
    KvSync {
        node: String,
        name: String,
        bearer_secret: Option<String>,
    },
    DocumentCreate {
        name: String,
    },
    DocumentStatus {
        name: String,
    },
    DocumentSet {
        name: String,
        state_json: String,
    },
    DocumentGet {
        name: String,
    },
    DocumentSync {
        node: String,
        name: String,
        bearer_secret: Option<String>,
    },
    PubsubPub {
        topic: String,
        message: String,
        node: Option<String>,
        bearer_secret: Option<String>,
    },
    PubsubSub {
        topic: String,
        node: Option<String>,
        bearer_secret: Option<String>,
    },
    PipeListen {
        name: String,
        node: Option<String>,
        bearer_secret: Option<String>,
    },
    PipeConnect {
        target: String,
        node: Option<String>,
        bearer_secret: Option<String>,
    },
    PipeTcpForward {
        node: String,
        listen_addr: String,
        target_addr: String,
        bearer_secret: Option<String>,
    },
    PipeTcpStream {
        node: String,
        target_addr: String,
        bearer_secret: Option<String>,
    },
    PipeUnixForward {
        node: String,
        listen_path: PathBuf,
        target_path: PathBuf,
        bearer_secret: Option<String>,
    },
    PipeUnixStream {
        node: String,
        target_path: PathBuf,
        bearer_secret: Option<String>,
    },
    PipeSend {
        target: String,
        data_base64: String,
        node: Option<String>,
        bearer_secret: Option<String>,
    },
    PipeRecv {
        name: String,
        peek: bool,
    },
    ModuleStub {
        module: String,
        command: String,
    },
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum ControlResponse {
    Status(StatusResponse),
    NodeId(NodeIdResponse),
    PeerCardExported {
        card: PeerCard,
        out: Option<PathBuf>,
        note: String,
    },
    PeerCardImported {
        peer: DiscoveredPeer,
        note: String,
    },
    PeerCardList {
        peers: Vec<DiscoveredPeer>,
        note: String,
    },
    PeerPinged {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        alpn: String,
        note: String,
    },
    PeerAuthChecked {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        resource: String,
        capability: String,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        note: String,
    },
    ResourceList {
        resources: Vec<ResourceDescriptor>,
    },
    ResourceCreated {
        resource: ResourceDescriptor,
    },
    OverlayStatus {
        networks: Vec<OverlayNetworkStatus>,
        note: String,
    },
    OverlayPlanned {
        plan: OverlayPlan,
    },
    OverlayJoined {
        join: OverlayJoinPlan,
    },
    OverlayLeft {
        name: String,
        stopped: bool,
        note: String,
    },
    CasAdded {
        hash: BlobHash,
        size_bytes: u64,
    },
    CasPrivateAdded {
        resource: String,
        epoch: u64,
        plaintext_hash: BlobHash,
        encrypted_hash: BlobHash,
        size_bytes: u64,
        note: String,
    },
    CasGot {
        hash: BlobHash,
        out: PathBuf,
        size_bytes: u64,
    },
    CasPrivateGot {
        resource: String,
        hash: BlobHash,
        plaintext_hash: BlobHash,
        out: PathBuf,
        size_bytes: u64,
        note: String,
    },
    CasFetched {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        hash: BlobHash,
        size_bytes: u64,
        allowed: bool,
        reason: String,
        note: String,
    },
    CasHash {
        hash: BlobHash,
    },
    CasHas {
        hash: BlobHash,
        present: bool,
    },
    CasPinned {
        hash: BlobHash,
        pinned: bool,
    },
    CasCleanup {
        removed: Vec<BlobHash>,
        retained_pinned: Vec<BlobHash>,
        dry_run: bool,
    },
    CasList {
        blobs: Vec<CasBlob>,
    },
    CasProviders {
        hash: BlobHash,
        providers: Vec<CasProvider>,
    },
    CasRootAdded {
        root: FileRoot,
    },
    CasRootList {
        roots: Vec<FileRoot>,
    },
    CasRootScanned {
        scan: FileRootScan,
    },
    CasRootSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        name: String,
        root: Option<FileRoot>,
        tree_bytes_imported: bool,
        sync_conflicts: Vec<FileConflict>,
        allowed: bool,
        reason: String,
        note: String,
    },
    CasRootApplied {
        source: String,
        target: PathBuf,
        files_written: usize,
        dirs_created: usize,
        conflicts: Vec<FileConflict>,
        dry_run: bool,
        note: String,
    },
    CasConflictRecorded {
        conflict: FileConflict,
    },
    CasConflictList {
        conflicts: Vec<FileConflict>,
    },
    CasConflictResolved {
        conflict: FileConflict,
    },
    KeychainStatus(KeychainStatusResponse),
    KeychainInitialized {
        ops: Vec<KeychainOp>,
        signatures: Vec<KeychainOpSignature>,
    },
    KeychainSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        ops_imported: usize,
        signatures_imported: usize,
        invalid_ops_rejected: usize,
        high_water_ms: i64,
        note: String,
    },
    AuthSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        ops_imported: usize,
        signatures_imported: usize,
        invalid_ops_rejected: usize,
        high_water_ms: i64,
        note: String,
    },
    SyncStatus {
        peers: Vec<SyncPeerStatus>,
        note: String,
    },
    SyncRan {
        peers: Vec<SyncPeerRun>,
        note: String,
    },
    SecretStatus {
        secrets: Vec<ResourceMasterSecret>,
    },
    SecretCreated {
        secret: ResourceMasterSecret,
    },
    SecretBearerCreated {
        access: BearerAccess,
    },
    SecretBearerList {
        access: Vec<BearerAccess>,
    },
    SecretBearerChallenge {
        challenge: BearerChallenge,
    },
    SecretBearerProof {
        proof: BearerProof,
    },
    SecretBearerVerified {
        secret: String,
        resource: String,
        capabilities: Vec<String>,
        verified: bool,
        reason: String,
    },
    SecretBearerRevoked {
        resource: String,
        secret: String,
    },
    AuthExplain(AuthExplanation),
    AuthOpRecorded {
        op: AuthOp,
        signatures: Vec<AuthOpSignature>,
    },
    NodeList {
        nodes: Vec<NodeRecord>,
        note: String,
    },
    NodeKeychainUpdated {
        ops: Vec<KeychainOp>,
        signatures: Vec<KeychainOpSignature>,
        note: String,
    },
    NodeGrantUpdated {
        op: AuthOp,
        signatures: Vec<AuthOpSignature>,
        note: String,
    },
    NodeEnrollmentRequested {
        request: NodeEnrollmentRequest,
        out: Option<PathBuf>,
        note: String,
    },
    NodeEnrollmentSubmitted {
        request_id: String,
        owner_node_id: String,
        accepted: bool,
        note: String,
    },
    NodeEnrollmentImported {
        request: NodeEnrollmentRequest,
        note: String,
    },
    NodeEnrollmentList {
        requests: Vec<NodeEnrollmentRequest>,
        note: String,
    },
    NodeEnrollmentApproved {
        request: NodeEnrollmentRequest,
        keychain_ops: Vec<KeychainOp>,
        keychain_signatures: Vec<KeychainOpSignature>,
        auth_ops: Vec<AuthOp>,
        auth_signatures: Vec<AuthOpSignature>,
        note: String,
    },
    NodeEnrollmentSynced {
        owner_node: String,
        keychain_ops_imported: usize,
        keychain_signatures_imported: usize,
        auth_ops_imported: usize,
        auth_signatures_imported: usize,
        invalid_ops_rejected: usize,
        note: String,
    },
    SshCertRequested {
        request: SshCertRequest,
    },
    SshCertRequests {
        requests: Vec<SshCertRequest>,
    },
    SshCertApproved {
        approval: SshCertApproval,
    },
    SshCertImported {
        certificate: SshCertificateRecord,
    },
    SshCertList {
        requests: Vec<SshCertRequest>,
        certificates: Vec<SshCertificateRecord>,
    },
    SshCertSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        requests_imported: usize,
        certificates_imported: usize,
        allowed: bool,
        reason: String,
        note: String,
    },
    SshRevocationAdded {
        revocation: SshRevocationEntry,
    },
    SshRevocationList {
        revocations: Vec<SshRevocationEntry>,
    },
    SshRevocationExported {
        out: PathBuf,
        format: String,
        count: usize,
        note: String,
    },
    SshRevocationImported {
        revocations: Vec<SshRevocationEntry>,
        format: String,
        count: usize,
        note: String,
    },
    SshRevocationSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        revocations_imported: usize,
        allowed: bool,
        reason: String,
        note: String,
    },
    DbAdded {
        db: DbResource,
    },
    DbStatus {
        db: DbResource,
    },
    DbChanges {
        db: DbResource,
        batch: CrSqliteChangeBatch,
    },
    DbSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        name: String,
        changes_received: usize,
        changes_applied: usize,
        max_db_version: Option<i64>,
        schema_match: bool,
        allowed: bool,
        reason: String,
        note: String,
    },
    KvCreated {
        kv: KvResource,
    },
    KvSet {
        entry: KvEntry,
    },
    KvGet {
        entry: Option<KvEntry>,
    },
    KvSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        name: String,
        entries_imported: usize,
        allowed: bool,
        reason: String,
        note: String,
    },
    DocumentCreated {
        document: DocumentResource,
    },
    DocumentStatus {
        document: DocumentResource,
    },
    DocumentSet {
        state: DocumentState,
    },
    DocumentGet {
        state: DocumentState,
    },
    DocumentSynced {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        name: String,
        updated: bool,
        allowed: bool,
        reason: String,
        note: String,
    },
    PubsubPublished {
        message: PubsubMessage,
    },
    PubsubRemotePublished {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        message: PubsubMessage,
        allowed: bool,
        reason: String,
        note: String,
    },
    PubsubMessages {
        topic: String,
        messages: Vec<PubsubMessage>,
        note: String,
    },
    PubsubRemoteMessages {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        topic: String,
        messages: Vec<PubsubMessage>,
        allowed: bool,
        reason: String,
        note: String,
    },
    PipeListening {
        listener: PipeListener,
    },
    PipeRemoteListening {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        listener: Option<PipeListener>,
        allowed: bool,
        reason: String,
        note: String,
    },
    PipeConnected {
        connection: PipeConnection,
    },
    PipeRemoteConnected {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        connection: PipeConnection,
        allowed: bool,
        reason: String,
        note: String,
    },
    PipeSent {
        message: Option<PipeMessage>,
        listener_found: bool,
        note: String,
    },
    PipeRemoteSent {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        message: Option<PipeMessage>,
        listener_found: bool,
        allowed: bool,
        reason: String,
        note: String,
    },
    PipeMessages {
        name: String,
        messages: Vec<PipeMessage>,
        drained: bool,
        note: String,
    },
    SshProxyConnected {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        connection: Option<SshProxyConnection>,
        allowed: bool,
        reason: String,
        note: String,
    },
    SshAdminShellOutput {
        peer_node_id: String,
        peer_agent_id: String,
        endpoint_id: String,
        command: String,
        output: String,
        allowed: bool,
        reason: String,
        note: String,
    },
    NotImplemented {
        module: String,
        command: String,
    },
    Error {
        message: String,
    },
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct StatusResponse {
    pub home: PathBuf,
    pub socket: PathBuf,
    pub agent_id: String,
    pub node_id: String,
    pub iroh_enabled: bool,
    pub endpoint_id: Option<String>,
    pub iroh_relay_mode: String,
    pub iroh_local_discovery: bool,
    pub iroh: String,
    #[serde(default)]
    pub native_backends: Vec<NativeBackendStatus>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NativeBackendStatus {
    pub module: String,
    pub current_backend: String,
    pub target_crate: String,
    pub target_version: String,
    pub status: String,
    pub blocker: String,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NodeIdResponse {
    pub agent_id: String,
    pub node_id: String,
    pub endpoint_id: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct KeychainStatusResponse {
    pub initialized: bool,
    pub admin_keys: usize,
    pub signatures: usize,
    pub verified_signatures: usize,
    pub failed_signatures: usize,
    pub users: usize,
    pub devices: usize,
    pub nodes: usize,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CasBlob {
    pub hash: BlobHash,
    pub size_bytes: u64,
    pub pinned: bool,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CasProvider {
    pub peer_node_id: String,
    pub endpoint_id: String,
    pub last_seen_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum PeerControlRequest {
    Ping {
        peer_card: PeerCard,
        nonce: String,
    },
    AuthCheck {
        peer_card: PeerCard,
        resource: String,
        capability: String,
        nonce: String,
    },
    SyncStatus {
        peer_card: PeerCard,
        nonce: String,
    },
    KeychainSync {
        peer_card: PeerCard,
        since_ms: i64,
        nonce: String,
    },
    AuthSync {
        peer_card: PeerCard,
        since_ms: i64,
        nonce: String,
    },
    NodeEnrollmentSubmit {
        peer_card: PeerCard,
        request: NodeEnrollmentRequest,
        nonce: String,
    },
    CasFetch {
        peer_card: PeerCard,
        hash: BlobHash,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    CasRootSync {
        peer_card: PeerCard,
        name: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    SshCertSync {
        peer_card: PeerCard,
        since_ms: i64,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    SshRevocationSync {
        peer_card: PeerCard,
        since_ms: i64,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    KvSync {
        peer_card: PeerCard,
        name: String,
        since_ms: i64,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    PubsubPublish {
        peer_card: PeerCard,
        topic: String,
        message: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    PubsubSubscribe {
        peer_card: PeerCard,
        topic: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    PipeConnect {
        peer_card: PeerCard,
        target: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    PipeListen {
        peer_card: PeerCard,
        name: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    SshProxyConnect {
        peer_card: PeerCard,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    SshAdminShell {
        peer_card: PeerCard,
        command: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    DocumentSync {
        peer_card: PeerCard,
        name: String,
        since_ms: i64,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    DbSync {
        peer_card: PeerCard,
        name: String,
        after_db_version: Option<i64>,
        limit: u32,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum PeerControlResponse {
    Pong {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        alpn: String,
        nonce: String,
        note: String,
    },
    AuthChecked {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        resource: String,
        capability: String,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    SyncStatus {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        watermarks: Vec<SyncWatermark>,
        nonce: String,
        note: String,
    },
    KeychainSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        ops: Vec<KeychainOp>,
        signatures: Vec<KeychainOpSignature>,
        high_water_ms: i64,
        nonce: String,
        note: String,
    },
    AuthSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        ops: Vec<AuthOp>,
        signatures: Vec<AuthOpSignature>,
        high_water_ms: i64,
        nonce: String,
        note: String,
    },
    NodeEnrollmentSubmitted {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        request_id: String,
        accepted: bool,
        nonce: String,
        note: String,
    },
    CasFetched {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        hash: BlobHash,
        size_bytes: u64,
        content_base64: Option<String>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    CasRootSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        name: String,
        root: Option<FileRoot>,
        tree_content_base64: Option<String>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    SshCertSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        requests: Vec<SshCertRequest>,
        certificates: Vec<SshCertificateRecord>,
        log_entries: Vec<SshDistributionLogEntry>,
        high_water_ms: i64,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    SshRevocationSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        revocations: Vec<SshRevocationEntry>,
        log_entries: Vec<SshDistributionLogEntry>,
        high_water_ms: i64,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    KvSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        name: String,
        entries: Vec<KvSyncEntry>,
        high_water_ms: i64,
        docs_ticket: Option<String>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    PubsubPublished {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        message: Option<PubsubMessage>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    PubsubSubscribed {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        topic: String,
        messages: Vec<PubsubMessage>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    PipeConnected {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        connection: Option<PipeConnection>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    PipeListening {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        listener: Option<PipeListener>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    SshProxyConnected {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        connection: Option<SshProxyConnection>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    SshAdminShellOutput {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        command: String,
        output: String,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    DocumentSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        name: String,
        state: Option<DocumentState>,
        high_water_ms: i64,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    DbSynced {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        name: String,
        batch: Option<CrSqliteChangeBatch>,
        high_water_db_version: Option<i64>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    Error {
        message: String,
    },
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum PipeWireRequest {
    Send {
        peer_card: PeerCard,
        target: String,
        data_base64: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    TcpConnect {
        peer_card: PeerCard,
        target_addr: String,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
    UnixConnect {
        peer_card: PeerCard,
        target_path: PathBuf,
        nonce: String,
        bearer_proof: Option<BearerProof>,
    },
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum PipeWireResponse {
    Sent {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        message: Box<Option<PipeMessage>>,
        listener_found: bool,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    Connected {
        node_id: String,
        agent_id: String,
        endpoint_id: String,
        remote_endpoint_id: String,
        connection: Option<PipeConnection>,
        allowed: bool,
        reason: String,
        evaluated_ops: usize,
        nonce: String,
        note: String,
    },
    Error {
        message: String,
    },
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncWatermark {
    pub stream: String,
    pub high_water: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncPeerStatus {
    pub peer_node_id: String,
    pub streams: Vec<SyncStreamStatus>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncStreamStatus {
    pub stream: String,
    pub cursor_ms: i64,
    pub state: String,
    pub stale: bool,
    pub stale_after_ms: i64,
    pub last_attempt_ms: Option<i64>,
    pub last_success_ms: Option<i64>,
    pub last_error: Option<String>,
    pub last_imported: usize,
    pub last_rejected: usize,
    pub next_action: String,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncPeerRun {
    pub peer_node_id: String,
    pub streams: Vec<SyncStreamRun>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncStreamRun {
    pub stream: String,
    pub attempted: bool,
    pub success: bool,
    pub imported: usize,
    pub rejected: usize,
    pub cursor_ms: i64,
    pub error: Option<String>,
}

#[derive(Debug, thiserror::Error)]
pub enum ControlError {
    #[error("json error: {0}")]
    Json(#[from] serde_json::Error),
}

pub fn encode_request(request: &ControlRequest) -> Result<String, ControlError> {
    let mut line = serde_json::to_string(request)?;
    line.push('\n');
    Ok(line)
}

pub fn decode_request(line: &str) -> Result<ControlRequest, ControlError> {
    serde_json::from_str(line).map_err(ControlError::from)
}

pub fn encode_response(response: &ControlResponse) -> Result<String, ControlError> {
    let mut line = serde_json::to_string(response)?;
    line.push('\n');
    Ok(line)
}

pub fn decode_response(line: &str) -> Result<ControlResponse, ControlError> {
    serde_json::from_str(line).map_err(ControlError::from)
}

pub fn encode_peer_request(request: &PeerControlRequest) -> Result<String, ControlError> {
    let mut line = serde_json::to_string(request)?;
    line.push('\n');
    Ok(line)
}

pub fn decode_peer_request(line: &str) -> Result<PeerControlRequest, ControlError> {
    serde_json::from_str(line).map_err(ControlError::from)
}

pub fn encode_peer_response(response: &PeerControlResponse) -> Result<String, ControlError> {
    let mut line = serde_json::to_string(response)?;
    line.push('\n');
    Ok(line)
}

pub fn decode_peer_response(line: &str) -> Result<PeerControlResponse, ControlError> {
    serde_json::from_str(line).map_err(ControlError::from)
}

pub fn encode_pipe_wire_request(request: &PipeWireRequest) -> Result<String, ControlError> {
    let mut line = serde_json::to_string(request)?;
    line.push('\n');
    Ok(line)
}

pub fn decode_pipe_wire_request(line: &str) -> Result<PipeWireRequest, ControlError> {
    serde_json::from_str(line).map_err(ControlError::from)
}

pub fn encode_pipe_wire_response(response: &PipeWireResponse) -> Result<String, ControlError> {
    let mut line = serde_json::to_string(response)?;
    line.push('\n');
    Ok(line)
}

pub fn decode_pipe_wire_response(line: &str) -> Result<PipeWireResponse, ControlError> {
    serde_json::from_str(line).map_err(ControlError::from)
}

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

    #[test]
    fn control_request_response_serialization_roundtrip() {
        let request = ControlRequest::KeychainInit {
            admin_key_path: Some(PathBuf::from("admin.pub")),
            signing_key_path: Some(PathBuf::from("admin")),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::NodeRename {
            node: "laptop".to_owned(),
            name: "work-laptop".to_owned(),
            signing_key_path: Some(PathBuf::from("admin")),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::NodeEnrollRequest {
            node_name: "workstation".to_owned(),
            capabilities: vec!["resource:ssh-proxy:local=ssh_proxy.connect".to_owned()],
            reason: Some("new machine".to_owned()),
            out: Some(PathBuf::from("enrollment.json")),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::NodeEnrollSync {
            owner_node: "owner-laptop".to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::KeychainSync {
            node: "work-laptop".to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::CasHas {
            hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::OverlayPlan {
            name: "home-lan".to_owned(),
            cidr: Some("172.22.0.0/24".to_owned()),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::CasHas {
            hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
            present: true,
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::OverlayPlanned {
            plan: geth_overlay::plan_overlay("home-lan", None, geth_overlay::OVERLAY_ALPN)
                .expect("overlay plan"),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::NodeList {
            nodes: vec![NodeRecord {
                id: geth_types::NodeId::new("node:local"),
                device: geth_types::DeviceId::new("device:local"),
                name: "work-laptop".to_owned(),
                endpoints: Vec::new(),
            }],
            note: "reduced keychain view".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::KeychainSynced {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            ops_imported: 2,
            signatures_imported: 2,
            invalid_ops_rejected: 1,
            high_water_ms: 42,
            note: "trusted admin signatures only".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::NodeEnrollmentSynced {
            owner_node: "owner-laptop".to_owned(),
            keychain_ops_imported: 3,
            keychain_signatures_imported: 3,
            auth_ops_imported: 1,
            auth_signatures_imported: 1,
            invalid_ops_rejected: 0,
            note: "synced".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::AuthSynced {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            ops_imported: 1,
            signatures_imported: 1,
            invalid_ops_rejected: 0,
            high_water_ms: 42,
            note: "trusted admin signatures only".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::SyncStatus {
            peers: vec![SyncPeerStatus {
                peer_node_id: "node:peer".to_owned(),
                streams: vec![SyncStreamStatus {
                    stream: "keychain".to_owned(),
                    cursor_ms: 42,
                    state: "ok".to_owned(),
                    stale: false,
                    stale_after_ms: 120_000,
                    last_attempt_ms: Some(43),
                    last_success_ms: Some(43),
                    last_error: None,
                    last_imported: 2,
                    last_rejected: 0,
                    next_action: "none".to_owned(),
                }],
            }],
            note: "local health".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::SyncRan {
            peers: vec![SyncPeerRun {
                peer_node_id: "node:peer".to_owned(),
                streams: vec![SyncStreamRun {
                    stream: "auth".to_owned(),
                    attempted: true,
                    success: true,
                    imported: 1,
                    rejected: 0,
                    cursor_ms: 44,
                    error: None,
                }],
            }],
            note: "ran".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::SecretBearerVerify {
            secret: "bearer:test".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.fetch".to_owned()],
            nonce: "nonce".to_owned(),
            response: "response".to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::SecretBearerVerified {
            secret: "bearer:test".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.fetch".to_owned()],
            verified: true,
            reason: "ok".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::CasFetch {
            node: "node:peer".to_owned(),
            hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
            bearer_secret: Some("bearer:test".to_owned()),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::CasProviders {
            hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::CasProviders {
            hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
            providers: vec![CasProvider {
                peer_node_id: "node:peer".to_owned(),
                endpoint_id: "endpoint:peer".to_owned(),
                last_seen_ms: 42,
            }],
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::SshRevocationExport {
            out: PathBuf::from("revocations.krl-spec"),
            format: "openssh-krl-spec".to_owned(),
            ca_public: None,
            subject: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::SshRevocationImport {
            path: PathBuf::from("revocations.jsonl"),
            format: "jsonl".to_owned(),
            subject: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::SshCertSync {
            node: "node:ca".to_owned(),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::SshRevocationSync {
            node: "node:ca".to_owned(),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::SshRevocationExported {
            out: PathBuf::from("revocations.krl-spec"),
            format: "openssh-krl-spec".to_owned(),
            count: 2,
            note: "OpenSSH KRL specification".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::SshCertSynced {
            peer_node_id: "node:ca".to_owned(),
            peer_agent_id: "agent:ca".to_owned(),
            endpoint_id: "endpoint:ca".to_owned(),
            requests_imported: 1,
            certificates_imported: 0,
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "cert sync".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::SshRevocationSynced {
            peer_node_id: "node:ca".to_owned(),
            peer_agent_id: "agent:ca".to_owned(),
            endpoint_id: "endpoint:ca".to_owned(),
            revocations_imported: 2,
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "revocation sync".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::KvSync {
            node: "node:peer".to_owned(),
            name: "prefs".to_owned(),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::KvSynced {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            name: "prefs".to_owned(),
            entries_imported: 1,
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "kv sync".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::DocumentSet {
            name: "notes".to_owned(),
            state_json: r#"{"title":"notes"}"#.to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PipeListen {
            name: "inbox".to_owned(),
            node: Some("node:peer".to_owned()),
            bearer_secret: Some("bearer:test".to_owned()),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PipeConnect {
            target: "inbox".to_owned(),
            node: Some("node:peer".to_owned()),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PipeSend {
            target: "inbox".to_owned(),
            data_base64: "aGVsbG8=".to_owned(),
            node: Some("node:peer".to_owned()),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PipeRecv {
            name: "inbox".to_owned(),
            peek: true,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::PipeMessages {
            name: "inbox".to_owned(),
            messages: vec![PipeMessage {
                pipe: "inbox".to_owned(),
                data_base64: "aGVsbG8=".to_owned(),
                received_at: geth_types::UnixMillis(1),
                source_node: Some("node:peer".to_owned()),
                note: "pipe".to_owned(),
            }],
            drained: false,
            note: "pipe messages".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::PubsubPub {
            topic: "presence/test".to_owned(),
            message: "online".to_owned(),
            node: Some("node:peer".to_owned()),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PubsubSub {
            topic: "presence/test".to_owned(),
            node: Some("node:peer".to_owned()),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::PubsubRemotePublished {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            message: PubsubMessage {
                topic: "presence/test".into(),
                message: "online".to_owned(),
                published_at: geth_types::UnixMillis(1),
            },
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "lossy".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::PubsubRemoteMessages {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            topic: "presence/test".to_owned(),
            messages: vec![PubsubMessage {
                topic: "presence/test".into(),
                message: "online".to_owned(),
                published_at: geth_types::UnixMillis(1),
            }],
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "lossy".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::PipeRemoteConnected {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            connection: PipeConnection {
                target: "inbox".to_owned(),
                connected_at: geth_types::UnixMillis(1),
                local_listener_found: true,
                note: "remote".to_owned(),
            },
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "pipe connect".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::PipeRemoteListening {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            listener: Some(PipeListener {
                id: "pipe:inbox".into(),
                name: "inbox".to_owned(),
                listened_at: geth_types::UnixMillis(1),
                note: "remote".to_owned(),
            }),
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "pipe listen".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::SshProxyConnect {
            node: "node:peer".to_owned(),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::SshProxyStream {
            node: "node:peer".to_owned(),
            bearer_secret: Some("bearer:test".to_owned()),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::SshAdminShell {
            node: "node:peer".to_owned(),
            command: "status".to_owned(),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::SshProxyConnected {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            connection: Some(SshProxyConnection {
                target_node: "node:peer".into(),
                connected_at: geth_types::UnixMillis(1),
                local_sshd_target: Some("127.0.0.1:22".to_owned()),
                admin_shell_available: false,
                note: "proxy".to_owned(),
            }),
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "ssh proxy".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let response = ControlResponse::SshAdminShellOutput {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            command: "status".to_owned(),
            output: "node_id=node:peer".to_owned(),
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "restricted admin shell".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::DbChanges {
            name: "notes".to_owned(),
            after_db_version: Some(7),
            limit: 10,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::DbSync {
            node: "node:peer".to_owned(),
            name: "notes".to_owned(),
            limit: 50,
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::DbSynced {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            name: "notes".to_owned(),
            changes_received: 1,
            changes_applied: 1,
            max_db_version: Some(7),
            schema_match: true,
            allowed: true,
            reason: "direct grant".to_owned(),
            note: "exchange only".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::CasRootScan {
            name: "notes".to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::CasRootSync {
            node: "node:peer".to_owned(),
            name: "notes".to_owned(),
            bearer_secret: None,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::CasRootSynced {
            peer_node_id: "node:peer".to_owned(),
            peer_agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            name: "notes".to_owned(),
            root: None,
            tree_bytes_imported: false,
            sync_conflicts: Vec::new(),
            allowed: false,
            reason: "no grant".to_owned(),
            note: "file-root sync".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::CasRootApply {
            source: "remote-node-peer-notes".to_owned(),
            target: PathBuf::from("/tmp/notes"),
            dry_run: true,
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = ControlResponse::CasRootApplied {
            source: "remote-node-peer-notes".to_owned(),
            target: PathBuf::from("/tmp/notes"),
            files_written: 1,
            dirs_created: 1,
            conflicts: Vec::new(),
            dry_run: true,
            note: "safe apply".to_owned(),
        };
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
            response
        );

        let request = ControlRequest::CasConflictResolve {
            conflict_id: "file-conflict:notes:1".to_owned(),
            resolution: "keep-local".to_owned(),
            note: Some("local file is authoritative".to_owned()),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PeerCardExport { out: None };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PeerPing {
            node: "node:peer".to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = ControlRequest::PeerAuthCheck {
            node: "node:peer".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = PeerControlResponse::Pong {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            alpn: "/geth/control/1".to_owned(),
            nonce: "nonce".to_owned(),
            note: "candidate only".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::AuthChecked {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
            allowed: false,
            reason: "no grant".to_owned(),
            evaluated_ops: 0,
            nonce: "nonce".to_owned(),
            note: "protected".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::SyncStatus {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            watermarks: vec![SyncWatermark {
                stream: "kv:prefs".to_owned(),
                high_water: 42,
            }],
            nonce: "nonce".to_owned(),
            note: "status".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::CasFetched {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
            size_bytes: 5,
            content_base64: Some("aGVsbG8".to_owned()),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "cas fetch".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::CasRootSynced {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            name: "notes".to_owned(),
            root: None,
            tree_content_base64: None,
            allowed: false,
            reason: "no grant".to_owned(),
            evaluated_ops: 0,
            nonce: "nonce".to_owned(),
            note: "file-root sync".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let request = PeerControlRequest::DbSync {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            name: "notes".to_owned(),
            after_db_version: Some(7),
            limit: 10,
            nonce: "nonce".to_owned(),
            bearer_proof: None,
        };
        assert_eq!(
            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = PeerControlRequest::SyncStatus {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            nonce: "nonce".to_owned(),
        };
        assert_eq!(
            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = PeerControlResponse::SshCertSynced {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            requests: Vec::new(),
            certificates: Vec::new(),
            log_entries: Vec::new(),
            high_water_ms: 42,
            allowed: false,
            reason: "no grant".to_owned(),
            evaluated_ops: 0,
            nonce: "nonce".to_owned(),
            note: "cert sync".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::SshRevocationSynced {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            revocations: Vec::new(),
            log_entries: Vec::new(),
            high_water_ms: 42,
            allowed: false,
            reason: "no grant".to_owned(),
            evaluated_ops: 0,
            nonce: "nonce".to_owned(),
            note: "revocation sync".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::KvSynced {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            name: "prefs".to_owned(),
            entries: vec![KvSyncEntry {
                key: "apps/foo/theme".to_owned(),
                value: "dark".to_owned(),
                updated_at_ms: 42,
            }],
            high_water_ms: 43,
            docs_ticket: Some("doc-ticket".to_owned()),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "kv sync".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::PubsubPublished {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            message: Some(PubsubMessage {
                topic: "presence/test".into(),
                message: "online".to_owned(),
                published_at: geth_types::UnixMillis(1),
            }),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "lossy".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let request = PeerControlRequest::PubsubSubscribe {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            topic: "presence/test".to_owned(),
            nonce: "nonce".to_owned(),
            bearer_proof: None,
        };
        assert_eq!(
            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
            request
        );

        let response = PeerControlResponse::PubsubSubscribed {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            topic: "presence/test".to_owned(),
            messages: vec![PubsubMessage {
                topic: "presence/test".into(),
                message: "online".to_owned(),
                published_at: geth_types::UnixMillis(1),
            }],
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "lossy".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::PipeConnected {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            connection: Some(PipeConnection {
                target: "inbox".to_owned(),
                connected_at: geth_types::UnixMillis(1),
                local_listener_found: true,
                note: "remote".to_owned(),
            }),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "pipe connect".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let request = PeerControlRequest::SshProxyConnect {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            nonce: "nonce".to_owned(),
            bearer_proof: Some(BearerProof {
                secret: "bearer:test".into(),
                resource: "resource:ssh-proxy:local".into(),
                capabilities: vec!["ssh_proxy.connect".into()],
                nonce: "nonce".to_owned(),
                response: "response".to_owned(),
            }),
        };
        assert_eq!(
            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = PeerControlRequest::SshAdminShell {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            command: "status".to_owned(),
            nonce: "nonce".to_owned(),
            bearer_proof: Some(BearerProof {
                secret: "bearer:test".into(),
                resource: "resource:ssh-proxy:local".into(),
                capabilities: vec!["ssh_proxy.admin_shell".into()],
                nonce: "nonce".to_owned(),
                response: "response".to_owned(),
            }),
        };
        assert_eq!(
            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = PeerControlRequest::PipeListen {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            name: "inbox".to_owned(),
            nonce: "nonce".to_owned(),
            bearer_proof: None,
        };
        assert_eq!(
            decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"),
            request
        );

        let request = PipeWireRequest::Send {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            target: "inbox".to_owned(),
            data_base64: "aGVsbG8=".to_owned(),
            nonce: "nonce".to_owned(),
            bearer_proof: None,
        };
        assert_eq!(
            decode_pipe_wire_request(&encode_pipe_wire_request(&request).expect("encode"))
                .expect("decode"),
            request
        );

        let response = PipeWireResponse::Sent {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            message: Box::new(Some(PipeMessage {
                pipe: "inbox".to_owned(),
                data_base64: "aGVsbG8=".to_owned(),
                received_at: geth_types::UnixMillis(1),
                source_node: Some("node:caller".to_owned()),
                note: "pipe".to_owned(),
            })),
            listener_found: true,
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "pipe wire".to_owned(),
        };
        assert_eq!(
            decode_pipe_wire_response(&encode_pipe_wire_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let request = PipeWireRequest::TcpConnect {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            target_addr: "127.0.0.1:22".to_owned(),
            nonce: "nonce".to_owned(),
            bearer_proof: Some(BearerProof {
                secret: "bearer:test".into(),
                resource: "resource:pipe-tcp:127.0.0.1:22".into(),
                capabilities: vec!["pipe.forward".into()],
                nonce: "nonce".to_owned(),
                response: "response".to_owned(),
            }),
        };
        assert_eq!(
            decode_pipe_wire_request(&encode_pipe_wire_request(&request).expect("encode"))
                .expect("decode"),
            request
        );

        let response = PipeWireResponse::Connected {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            connection: Some(PipeConnection {
                target: "127.0.0.1:22".to_owned(),
                connected_at: geth_types::UnixMillis(1),
                local_listener_found: true,
                note: "tcp".to_owned(),
            }),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "pipe tcp".to_owned(),
        };
        assert_eq!(
            decode_pipe_wire_response(&encode_pipe_wire_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let request = PipeWireRequest::UnixConnect {
            peer_card: PeerCard {
                node_id: "node:caller".into(),
                agent_id: "agent:caller".into(),
                endpoints: Vec::new(),
                issued_at: geth_types::UnixMillis(1),
                signature: geth_discovery::SignatureMetadata {
                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
                    signer: "agent:caller".to_owned(),
                    public_key: "key".to_owned(),
                    signature: "sig".to_owned(),
                },
            },
            target_path: PathBuf::from("/tmp/geth-test.sock"),
            nonce: "nonce".to_owned(),
            bearer_proof: None,
        };
        assert_eq!(
            decode_pipe_wire_request(&encode_pipe_wire_request(&request).expect("encode"))
                .expect("decode"),
            request
        );

        let response = PeerControlResponse::SshProxyConnected {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            connection: Some(SshProxyConnection {
                target_node: "node:peer".into(),
                connected_at: geth_types::UnixMillis(1),
                local_sshd_target: Some("127.0.0.1:22".to_owned()),
                admin_shell_available: false,
                note: "proxy".to_owned(),
            }),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "ssh proxy".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::SshAdminShellOutput {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            command: "status".to_owned(),
            output: "node_id=node:peer".to_owned(),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "restricted admin shell".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );

        let response = PeerControlResponse::DbSynced {
            node_id: "node:peer".to_owned(),
            agent_id: "agent:peer".to_owned(),
            endpoint_id: "endpoint:peer".to_owned(),
            remote_endpoint_id: "endpoint:caller".to_owned(),
            name: "notes".to_owned(),
            batch: Some(CrSqliteChangeBatch {
                schema_metadata: "tables=1 schema_hash=abc".to_owned(),
                max_db_version: Some(7),
                changes: Vec::new(),
            }),
            high_water_db_version: Some(7),
            allowed: true,
            reason: "direct grant".to_owned(),
            evaluated_ops: 1,
            nonce: "nonce".to_owned(),
            note: "db sync".to_owned(),
        };
        assert_eq!(
            decode_peer_response(&encode_peer_response(&response).expect("encode"))
                .expect("decode"),
            response
        );
    }
}