maestro/crates/maestro-protocol/src/lib.rs

//! Wire types shared between the Maestro daemon, registry, and clients.
//!
//! See `docs/superpowers/specs/2026-05-24-maestro-design.md` § "Wire protocol"
//! and § "Session model" for the source-of-truth schema.

#![deny(missing_debug_implementations)]

use std::collections::HashMap;
use std::path::PathBuf;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;

/// A correlation token used to pair request/response messages over a single WS.
pub type ReplyTo = Uuid;

/// The protocol version this crate implements. Daemons advertise it in `Hello`.
pub const PROTOCOL_VERSION: &str = env!("CARGO_PKG_VERSION");

/// A live or recently-dead tmux session managed by a daemon.
///
/// The daemon is the source of truth; the registry caches snapshots.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Session {
    pub id: Uuid,
    pub daemon_id: String,
    pub cwd: PathBuf,
    pub command: String,
    #[serde(default)]
    pub env: HashMap<String, String>,
    #[serde(default)]
    pub metadata: HashMap<String, Value>,
    pub status: SessionStatus,
    pub created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SessionStatus {
    Running,
    Dead,
}

/// Optional worktree request attached to a `CreateSessionRequest`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorktreeSpec {
    pub base_repo: PathBuf,
    pub branch: String,
    #[serde(default)]
    pub cleanup_on_kill: bool,
}

/// HTTP `POST /api/sessions` body, also forwarded daemon-ward inside
/// `DaemonInbound::CreateSession`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CreateSessionRequest {
    pub daemon_id: String,
    pub cwd: PathBuf,
    pub command: String,
    #[serde(default)]
    pub env: HashMap<String, String>,
    #[serde(default)]
    pub metadata: HashMap<String, Value>,
    #[serde(default)]
    pub worktree: Option<WorktreeSpec>,
}

/// Structured failure used everywhere we don't want to leak a string-only error.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ErrorInfo {
    pub code: String,
    pub message: String,
}

/// Generic request-reply result body. Used by `CreateSessionResult`,
/// `KillSessionResult`, and any future correlated reply.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum RpcResult<T> {
    Ok { value: T },
    Err { error: ErrorInfo },
}

// ---------------------------------------------------------------------------
// Daemon ↔ Registry messages
// ---------------------------------------------------------------------------

/// Messages a daemon sends to the registry over its outbound WS.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum DaemonOutbound {
    Hello {
        daemon_id: String,
        hostname: String,
        version: String,
        token: String,
        /// Reachable URL of this daemon's per-session WS endpoint, as the
        /// operator wants web clients to dial it. The registry hands this back
        /// in `GET /api/sessions/:id/connect-info` so the client can connect
        /// directly. Example: `ws://10.0.0.1:8081`.
        daemon_url: String,
    },
    SessionsSnapshot {
        sessions: Vec<Session>,
    },
    SessionAdded {
        session: Session,
    },
    SessionRemoved {
        id: Uuid,
    },
    SessionUpdated {
        session: Session,
    },
    Heartbeat,
    CreateSessionResult {
        reply_to: ReplyTo,
        result: RpcResult<Session>,
    },
    KillSessionResult {
        reply_to: ReplyTo,
        result: RpcResult<()>,
    },
    /// Home daemon asks the registry to resolve a session it doesn't own:
    /// "where is session X and how do I reach it?" The registry replies with
    /// [`DaemonInbound::ResolveAttachResult`] containing [`AttachCandidates`].
    ResolveAttach {
        reply_to: ReplyTo,
        session_id: Uuid,
    },
    /// Target daemon answers a [`DaemonInbound::RelaySetup`]: whether it will
    /// accept relay traffic for the requested session. Registry only allocates
    /// relay capacity when `accepted == true`.
    RelayAck {
        reply_to: ReplyTo,
        accepted: bool,
    },
}

/// Messages the registry sends back to a daemon over the same WS.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum DaemonInbound {
    Ack,
    Error(ErrorInfo),
    CreateSession {
        reply_to: ReplyTo,
        request: CreateSessionRequest,
    },
    KillSession {
        reply_to: ReplyTo,
        id: Uuid,
    },
    /// Registry's response to [`DaemonOutbound::ResolveAttach`]. `Err` covers
    /// "session not found", "no daemon owns this session", etc.
    ResolveAttachResult {
        reply_to: ReplyTo,
        result: RpcResult<AttachCandidates>,
    },
    /// Registry signals a target daemon that a home daemon wants to attach via
    /// the registry-mediated relay (T3). Target evaluates `allow_inbound` and
    /// replies with [`DaemonOutbound::RelayAck`].
    RelaySetup {
        reply_to: ReplyTo,
        home_daemon: String,
        session_id: Uuid,
    },
}

// ---------------------------------------------------------------------------
// Client ↔ Registry API types
// ---------------------------------------------------------------------------

/// Public, read-only view of a connected daemon. Returned by `GET /api/daemons`
/// and included in `RegistryEvent::Snapshot` / `RegistryEvent::DaemonAdded`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaemonInfo {
    pub daemon_id: String,
    pub hostname: String,
    pub version: String,
    pub daemon_url: String,
}

/// Response body for `GET /api/sessions/:id/connect-info`.
///
/// Superseded by the broker-mediated `ResolveAttach` / [`AttachCandidates`]
/// flow added for the connection-orchestration epic (NAP-236). Retained as a
/// compat path while clients migrate; new code should prefer the broker model.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ConnectInfo {
    pub daemon_url: String,
    pub jwt: String,
}

// ---------------------------------------------------------------------------
// Broker-mediated attach (NAP-242)
//
// Wire types for the multi-transport attach flow defined in
// `docs/superpowers/specs/2026-05-26-connection-orchestration.md`. The home
// daemon asks the registry to `ResolveAttach`; the registry replies with
// `AttachCandidates` describing the transports the home daemon may try (T1
// direct WSS, T2 direct WS over shared perimeter, T3 registry-mediated relay)
// plus a pair-scoped JWT. For T3, the registry signals the target daemon with
// `RelaySetup` and only allocates relay capacity after a `RelayAck`.
// ---------------------------------------------------------------------------

/// T2 transport hint: target daemon's reachable address on a perimeter the
/// home daemon may share. The home daemon dials `ws://<private_addr>/...` only
/// if its own `perimeter_id` matches.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PerimeterCandidate {
    pub perimeter_id: String,
    pub private_addr: String,
}

/// T3 transport marker. The concrete shape (relay endpoint, ticket, etc.)
/// depends on the relay backend selected by NAP-238 (Iroh vs libp2p
/// circuit-v2), so this stays a `#[non_exhaustive]` placeholder for now.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct RelayCandidate {}

/// Registry's reply to `ResolveAttach`. Each `Option` describes one viable
/// transport; the home daemon tries them in T1 → T2 → T3 order. `jwt` is the
/// pair-scoped credential ([`PairAttachClaims`]) target daemons validate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttachCandidates {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub direct_wss: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub perimeter: Option<PerimeterCandidate>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub relay: Option<RelayCandidate>,
    pub jwt: String,
}

/// Messages broadcast by the registry to web clients over `WS /api/events`.
///
/// A subscriber receives a single `Snapshot` immediately on connect, then a
/// stream of deltas as daemons connect/disconnect and their session sets
/// change. Skipped events (broadcast channel lag) surface as `Lagged`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum RegistryEvent {
    Snapshot {
        daemons: Vec<DaemonInfo>,
        sessions: Vec<Session>,
    },
    DaemonAdded {
        daemon: DaemonInfo,
    },
    DaemonRemoved {
        daemon_id: String,
    },
    SessionAdded {
        session: Session,
    },
    SessionRemoved {
        daemon_id: String,
        id: Uuid,
    },
    SessionUpdated {
        session: Session,
    },
    /// Server-issued notification that the broadcast queue dropped frames.
    /// Clients should re-fetch a full snapshot.
    Lagged {
        skipped: u64,
    },
}

// ---------------------------------------------------------------------------
// Client ↔ Daemon control messages (text frames on the attach WS)
// ---------------------------------------------------------------------------

/// Text-frame control messages on a `WS /sessions/:id/stream` connection.
/// PTY bytes travel in binary frames and are not modeled here.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum ClientControl {
    Resize { cols: u16, rows: u16 },
    Ping,
    Pong,
    Error(ErrorInfo),
}

// ---------------------------------------------------------------------------
// JWT claims used on the per-session attach WS
// ---------------------------------------------------------------------------

/// Issued by the registry, validated locally by the daemon. Short-lived
/// (~5 min). The daemon checks `daemon_id` matches itself and `session_id`
/// is one of its live sessions.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JwtClaims {
    pub session_id: Uuid,
    pub daemon_id: String,
    /// Expiry (unix seconds).
    pub exp: i64,
    /// Issued at (unix seconds). Optional, mostly for diagnostics.
    #[serde(default)]
    pub iat: Option<i64>,
}

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

    fn round_trip<T>(value: &T) -> T
    where
        T: Serialize + for<'de> Deserialize<'de>,
    {
        let json = serde_json::to_string(value).expect("serialize");
        serde_json::from_str(&json).expect("deserialize")
    }

    fn sample_session() -> Session {
        Session {
            id: Uuid::from_u128(0x0123_4567_89ab_cdef_0123_4567_89ab_cdef),
            daemon_id: "laptop".into(),
            cwd: PathBuf::from("/home/naps62/projects/maestro"),
            command: "claude".into(),
            env: HashMap::from([("FOO".into(), "bar".into())]),
            metadata: HashMap::from([
                ("name".into(), Value::String("maestro-impl".into())),
                ("label".into(), Value::String("personal/maestro".into())),
            ]),
            status: SessionStatus::Running,
            created_at: DateTime::parse_from_rfc3339("2026-05-24T12:00:00Z")
                .unwrap()
                .with_timezone(&Utc),
        }
    }

    #[test]
    fn session_round_trips() {
        let s = sample_session();
        assert_eq!(s, round_trip(&s));
    }

    #[test]
    fn session_status_serializes_pascal_case() {
        assert_eq!(
            serde_json::to_value(SessionStatus::Running).unwrap(),
            serde_json::json!("Running")
        );
        assert_eq!(
            serde_json::to_value(SessionStatus::Dead).unwrap(),
            serde_json::json!("Dead")
        );
    }

    #[test]
    fn create_session_request_with_worktree_round_trips() {
        let req = CreateSessionRequest {
            daemon_id: "dev-vm".into(),
            cwd: PathBuf::from("/work"),
            command: "bash".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: PathBuf::from("/home/naps62/projects/maestro"),
                branch: "feat/foo".into(),
                cleanup_on_kill: true,
            }),
        };
        assert_eq!(req, round_trip(&req));
    }

    #[test]
    fn create_session_request_omits_optional_worktree() {
        let json = serde_json::json!({
            "daemon_id": "dev-vm",
            "cwd": "/work",
            "command": "bash"
        });
        let req: CreateSessionRequest = serde_json::from_value(json).unwrap();
        assert!(req.worktree.is_none());
        assert!(req.env.is_empty());
        assert!(req.metadata.is_empty());
    }

    #[test]
    fn rpc_result_ok_and_err_round_trip() {
        let ok: RpcResult<Session> = RpcResult::Ok {
            value: sample_session(),
        };
        let err: RpcResult<Session> = RpcResult::Err {
            error: ErrorInfo {
                code: "tmux_failed".into(),
                message: "session name in use".into(),
            },
        };
        assert_eq!(ok, round_trip(&ok));
        assert_eq!(err, round_trip(&err));
    }

    #[test]
    fn daemon_outbound_hello_round_trip_and_tag() {
        let hello = DaemonOutbound::Hello {
            daemon_id: "laptop".into(),
            hostname: "naps-laptop".into(),
            version: PROTOCOL_VERSION.into(),
            token: "shhh".into(),
            daemon_url: "ws://10.0.0.1:8081".into(),
        };
        let v = serde_json::to_value(&hello).unwrap();
        assert_eq!(v["type"], "Hello");
        assert_eq!(v["daemon_id"], "laptop");
        assert_eq!(hello, round_trip(&hello));
    }

    #[test]
    fn daemon_outbound_all_variants_round_trip() {
        let session = sample_session();
        let reply_to = Uuid::from_u128(1);
        let variants = vec![
            DaemonOutbound::SessionsSnapshot {
                sessions: vec![session.clone()],
            },
            DaemonOutbound::SessionAdded {
                session: session.clone(),
            },
            DaemonOutbound::SessionRemoved { id: session.id },
            DaemonOutbound::SessionUpdated {
                session: session.clone(),
            },
            DaemonOutbound::Heartbeat,
            DaemonOutbound::CreateSessionResult {
                reply_to,
                result: RpcResult::Ok {
                    value: session.clone(),
                },
            },
            DaemonOutbound::KillSessionResult {
                reply_to,
                result: RpcResult::Ok { value: () },
            },
        ];
        for v in variants {
            assert_eq!(v, round_trip(&v), "variant did not round-trip");
        }
    }

    #[test]
    fn daemon_inbound_all_variants_round_trip() {
        let reply_to = Uuid::from_u128(7);
        let create = DaemonInbound::CreateSession {
            reply_to,
            request: CreateSessionRequest {
                daemon_id: "laptop".into(),
                cwd: PathBuf::from("/tmp"),
                command: "bash".into(),
                env: HashMap::new(),
                metadata: HashMap::new(),
                worktree: None,
            },
        };
        let kill = DaemonInbound::KillSession {
            reply_to,
            id: Uuid::from_u128(42),
        };
        let ack = DaemonInbound::Ack;
        let err = DaemonInbound::Error(ErrorInfo {
            code: "unknown_session".into(),
            message: "no such id".into(),
        });
        for v in [create, kill, ack, err] {
            assert_eq!(v, round_trip(&v));
        }
    }

    #[test]
    fn daemon_inbound_ack_tag() {
        let v = serde_json::to_value(DaemonInbound::Ack).unwrap();
        assert_eq!(v, serde_json::json!({ "type": "Ack" }));
    }

    #[test]
    fn client_control_round_trips_and_tags() {
        let resize = ClientControl::Resize {
            cols: 120,
            rows: 40,
        };
        let v = serde_json::to_value(&resize).unwrap();
        assert_eq!(v["type"], "Resize");
        assert_eq!(v["cols"], 120);
        assert_eq!(v["rows"], 40);

        for c in [
            ClientControl::Ping,
            ClientControl::Pong,
            ClientControl::Error(ErrorInfo {
                code: "bad_frame".into(),
                message: "decode failed".into(),
            }),
            resize,
        ] {
            assert_eq!(c, round_trip(&c));
        }
    }

    #[test]
    fn unknown_message_type_is_rejected() {
        let bad = serde_json::json!({ "type": "Bogus" });
        let err = serde_json::from_value::<DaemonOutbound>(bad).unwrap_err();
        assert!(err.to_string().contains("Bogus"), "{}", err);
    }

    #[test]
    fn daemon_info_round_trips() {
        let d = DaemonInfo {
            daemon_id: "laptop".into(),
            hostname: "naps-laptop".into(),
            version: "0.1.0".into(),
            daemon_url: "ws://10.0.0.1:8081".into(),
        };
        assert_eq!(d, round_trip(&d));
    }

    #[test]
    fn connect_info_round_trips() {
        let c = ConnectInfo {
            daemon_url: "ws://10.0.0.1:8081".into(),
            jwt: "eyJ...".into(),
        };
        assert_eq!(c, round_trip(&c));
    }

    #[test]
    fn registry_event_variants_round_trip_and_tag() {
        let daemon = DaemonInfo {
            daemon_id: "laptop".into(),
            hostname: "naps-laptop".into(),
            version: "0.1.0".into(),
            daemon_url: "ws://10.0.0.1:8081".into(),
        };
        let session = sample_session();
        let variants = vec![
            RegistryEvent::Snapshot {
                daemons: vec![daemon.clone()],
                sessions: vec![session.clone()],
            },
            RegistryEvent::DaemonAdded {
                daemon: daemon.clone(),
            },
            RegistryEvent::DaemonRemoved {
                daemon_id: "laptop".into(),
            },
            RegistryEvent::SessionAdded {
                session: session.clone(),
            },
            RegistryEvent::SessionRemoved {
                daemon_id: "laptop".into(),
                id: session.id,
            },
            RegistryEvent::SessionUpdated { session },
            RegistryEvent::Lagged { skipped: 12 },
        ];
        for v in variants {
            assert_eq!(v, round_trip(&v));
        }

        let json = serde_json::to_value(RegistryEvent::DaemonRemoved {
            daemon_id: "x".into(),
        })
        .unwrap();
        assert_eq!(json["type"], "DaemonRemoved");
    }

    #[test]
    fn daemon_outbound_resolve_attach_round_trip_and_tag() {
        let msg = DaemonOutbound::ResolveAttach {
            reply_to: Uuid::from_u128(11),
            session_id: Uuid::from_u128(22),
        };
        let v = serde_json::to_value(&msg).unwrap();
        assert_eq!(v["type"], "ResolveAttach");
        assert_eq!(msg, round_trip(&msg));
    }

    #[test]
    fn daemon_outbound_relay_ack_round_trip_and_tag() {
        let msg = DaemonOutbound::RelayAck {
            reply_to: Uuid::from_u128(33),
            accepted: true,
        };
        let v = serde_json::to_value(&msg).unwrap();
        assert_eq!(v["type"], "RelayAck");
        assert_eq!(v["accepted"], true);
        assert_eq!(msg, round_trip(&msg));

        let denied = DaemonOutbound::RelayAck {
            reply_to: Uuid::from_u128(34),
            accepted: false,
        };
        assert_eq!(denied, round_trip(&denied));
    }

    #[test]
    fn daemon_inbound_resolve_attach_result_round_trip() {
        let ok = DaemonInbound::ResolveAttachResult {
            reply_to: Uuid::from_u128(11),
            result: RpcResult::Ok {
                value: AttachCandidates {
                    direct_wss: None,
                    perimeter: Some(PerimeterCandidate {
                        perimeter_id: "wg-naps62".into(),
                        private_addr: "10.0.0.7:7070".into(),
                    }),
                    relay: None,
                    jwt: "eyJ...".into(),
                },
            },
        };
        let err = DaemonInbound::ResolveAttachResult {
            reply_to: Uuid::from_u128(12),
            result: RpcResult::Err {
                error: ErrorInfo {
                    code: "unknown_session".into(),
                    message: "no daemon owns session".into(),
                },
            },
        };
        for v in [ok, err] {
            assert_eq!(v, round_trip(&v));
        }
    }

    #[test]
    fn daemon_inbound_relay_setup_round_trip_and_tag() {
        let msg = DaemonInbound::RelaySetup {
            reply_to: Uuid::from_u128(33),
            home_daemon: "laptop".into(),
            session_id: Uuid::from_u128(44),
        };
        let v = serde_json::to_value(&msg).unwrap();
        assert_eq!(v["type"], "RelaySetup");
        assert_eq!(v["home_daemon"], "laptop");
        assert_eq!(msg, round_trip(&msg));
    }

    #[test]
    fn perimeter_candidate_round_trips() {
        let p = PerimeterCandidate {
            perimeter_id: "wg-naps62".into(),
            private_addr: "10.0.0.7:7070".into(),
        };
        assert_eq!(p, round_trip(&p));
    }

    #[test]
    fn relay_candidate_round_trips_as_empty_object() {
        let r = RelayCandidate::default();
        assert_eq!(r, round_trip(&r));
        assert_eq!(serde_json::to_value(&r).unwrap(), serde_json::json!({}));
    }

    #[test]
    fn attach_candidates_full_and_minimal_round_trip() {
        let full = AttachCandidates {
            direct_wss: Some("wss://konishi.naps62.maestro.example/".into()),
            perimeter: Some(PerimeterCandidate {
                perimeter_id: "wg-naps62".into(),
                private_addr: "10.0.0.7:7070".into(),
            }),
            relay: Some(RelayCandidate::default()),
            jwt: "eyJ...".into(),
        };
        assert_eq!(full, round_trip(&full));

        let minimal = AttachCandidates {
            direct_wss: None,
            perimeter: None,
            relay: None,
            jwt: "eyJ...".into(),
        };
        assert_eq!(minimal, round_trip(&minimal));

        // Optional transport fields are dropped from the wire when absent —
        // keeps minimal payloads tight and avoids accidentally implying a
        // transport is available with a `null` value.
        let v = serde_json::to_value(&minimal).unwrap();
        assert_eq!(v, serde_json::json!({ "jwt": "eyJ..." }));
    }

    #[test]
    fn jwt_claims_round_trip_and_iat_optional() {
        let c = JwtClaims {
            session_id: Uuid::from_u128(7),
            daemon_id: "laptop".into(),
            exp: 1_700_000_000,
            iat: Some(1_699_999_700),
        };
        assert_eq!(c, round_trip(&c));

        let no_iat = serde_json::json!({
            "session_id": Uuid::from_u128(8),
            "daemon_id": "laptop",
            "exp": 1_700_000_000_i64,
        });
        let parsed: JwtClaims = serde_json::from_value(no_iat).unwrap();
        assert!(parsed.iat.is_none());
    }
}