This repository has been archived on 2026-10-08. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
maestro/crates/maestro-protocol/src/lib.rs
T
naps62-yoloandClaude Opus 4.8 75d162dea3
ci / rust (push) Successful in 6m45s
refactor(maestro): share cookie name + extraction across CLI and registry (NAP-273 F)
Promote `COOKIE_NAME` and a robust `extract_cookie` (iterates every `;`-pair,
so it works on both `Cookie:` and `Set-Cookie:` headers regardless of position)
into `maestro-protocol`. The registry re-exports them; the CLI drops its weaker
`extract_cookie_value` (which only inspected the first pair) and its duplicate
`COOKIE_NAME`. Both ends of the wire now agree on one definition.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-18 14:35:30 +00:00

1279 lines
45 KiB
Rust

//! 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>,
}
/// Result of a `ListDir` query: the immediate subdirectories of one directory
/// on a daemon's host. Used to drive the web spawn form's working-directory
/// autocomplete (NAP-329).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DirListing {
/// The resolved absolute directory that was listed, after the daemon
/// expanded `~`/relative/empty inputs against its own home dir. The client
/// uses it to know what an abbreviated query resolved to.
pub path: PathBuf,
/// Immediate subdirectory names (not full paths), sorted case-insensitively.
/// Directories only — the spawn form's targets are always dirs. Hidden
/// directories (dotfiles) are excluded.
pub entries: Vec<String>,
/// True when `entries` was truncated at the daemon's cap; the client can
/// surface "narrow your search" rather than implying the dir is small.
#[serde(default)]
pub truncated: bool,
}
/// 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,
/// Absolute home directory of the OS user this daemon runs as, used by
/// clients to prefill a spawn form's working directory. Empty when the
/// daemon could not resolve `$HOME`. (NAP-328)
#[serde(default)]
home_dir: String,
},
SessionsSnapshot {
sessions: Vec<Session>,
},
SessionAdded {
session: Session,
},
SessionRemoved {
id: Uuid,
},
SessionUpdated {
session: Session,
},
/// Liveness ticker. The optional `perimeter` advertises the daemon's
/// reachability for the T2 transport (NAP-242); the typed object encodes the
/// "both id and addr, or neither" invariant once (NAP-272). Old daemons that
/// don't set it still produce the bare `{"type":"Heartbeat"}` shape.
Heartbeat {
#[serde(default, skip_serializing_if = "Option::is_none")]
perimeter: Option<PerimeterCandidate>,
},
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,
},
/// Home daemon asks the registry to relay a session create to the daemon
/// that should own it (`request.daemon_id`). Control-plane only: the
/// registry forwards the request to the owner over its control link and
/// relays the owner's reply — no session bytes ever pass through the
/// registry, so the data-plane stays daemon-to-daemon. Reply:
/// [`DaemonInbound::ForwardCreateResult`]. (NAP-327)
ForwardCreate {
reply_to: ReplyTo,
request: CreateSessionRequest,
},
/// Home daemon asks the registry to relay a kill to the daemon that owns
/// `id`. Reply: [`DaemonInbound::ForwardKillResult`]. (NAP-327)
ForwardKill {
reply_to: ReplyTo,
daemon_id: String,
id: Uuid,
},
/// Owning daemon's reply to a [`DaemonInbound::ListDir`] — the requested
/// directory listing on its host. (NAP-329)
ListDirResult {
reply_to: ReplyTo,
result: RpcResult<DirListing>,
},
/// Home daemon asks the registry to relay a directory listing to the daemon
/// named by `daemon_id` (which may be a peer). Control-plane only, like
/// [`ForwardCreate`](Self::ForwardCreate). Reply:
/// [`DaemonInbound::ForwardListDirResult`]. (NAP-329)
ForwardListDir {
reply_to: ReplyTo,
daemon_id: String,
path: String,
},
/// 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,
},
/// Forward-compat catch-all: a frame whose `type` this build doesn't
/// recognize (e.g. a newer daemon emitting a not-yet-merged variant).
/// Readers log and skip it instead of tearing the connection down (NAP-269).
#[serde(other)]
Unknown,
}
/// 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's reply to [`DaemonOutbound::ForwardCreate`] — the owning
/// daemon's create result, relayed back to the home daemon. `Err` covers
/// "owner offline", the owner's own spawn failure, etc. (NAP-327)
ForwardCreateResult {
reply_to: ReplyTo,
result: RpcResult<Session>,
},
/// Registry's reply to [`DaemonOutbound::ForwardKill`]. (NAP-327)
ForwardKillResult {
reply_to: ReplyTo,
result: RpcResult<()>,
},
/// Registry asks a daemon to list `path` on its host. The daemon expands
/// `~`/relative/empty against its own home dir, then replies with
/// [`DaemonOutbound::ListDirResult`]. (NAP-329)
ListDir {
reply_to: ReplyTo,
path: String,
},
/// Registry's reply to [`DaemonOutbound::ForwardListDir`] — the owning
/// daemon's listing, relayed back to the home daemon. (NAP-329)
ForwardListDirResult {
reply_to: ReplyTo,
result: RpcResult<DirListing>,
},
/// 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,
},
/// Catalog delta forwarded from the registry so each connected daemon can
/// maintain a local mirror of the global catalog (peer daemons + their
/// sessions). The registry filters out events originating at the
/// recipient daemon — a daemon only sees *peer* deltas. The first event
/// a freshly-connected daemon receives is a `RegistryEvent::Snapshot`.
CatalogEvent {
event: RegistryEvent,
},
/// Forward-compat catch-all: a frame whose `type` this build doesn't
/// recognize (e.g. a newer registry emitting a not-yet-merged variant).
/// Readers log and skip it instead of tearing the connection down (NAP-269).
#[serde(other)]
Unknown,
}
// ---------------------------------------------------------------------------
// 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,
/// Absolute home directory of the OS user the daemon runs as. Clients use
/// it to prefill the spawn form's working directory. Empty when unknown
/// (older daemon, or `$HOME` unresolved). (NAP-328)
#[serde(default)]
pub home_dir: 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,
},
}
impl RegistryEvent {
/// Filter this event for a daemon recipient that should only see *peer*
/// state. Returns `None` if the entire event concerns the recipient,
/// otherwise a possibly-pruned event with self-references removed.
///
/// Used by the registry's daemon fan-out (NAP-246) so each daemon's
/// local catalog mirror reflects every other daemon, not itself.
pub fn for_peer(self, self_daemon_id: &str) -> Option<RegistryEvent> {
match self {
RegistryEvent::Snapshot { daemons, sessions } => {
let daemons = daemons
.into_iter()
.filter(|d| d.daemon_id != self_daemon_id)
.collect::<Vec<_>>();
let sessions = sessions
.into_iter()
.filter(|s| s.daemon_id != self_daemon_id)
.collect::<Vec<_>>();
Some(RegistryEvent::Snapshot { daemons, sessions })
}
RegistryEvent::DaemonAdded { ref daemon } if daemon.daemon_id == self_daemon_id => None,
RegistryEvent::DaemonRemoved { ref daemon_id } if daemon_id == self_daemon_id => None,
RegistryEvent::SessionAdded { ref session } if session.daemon_id == self_daemon_id => {
None
}
RegistryEvent::SessionRemoved { ref daemon_id, .. } if daemon_id == self_daemon_id => {
None
}
RegistryEvent::SessionUpdated { ref session }
if session.daemon_id == self_daemon_id =>
{
None
}
// Everything else (Lagged, peer adds/removes/updates) is forwarded
// as-is.
other => Some(other),
}
}
}
// ---------------------------------------------------------------------------
// 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.
///
/// Superseded by [`PairAttachClaims`] for the broker-mediated attach flow
/// (NAP-242). Retained as a compat path while clients migrate.
#[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>,
}
/// `iss` claim baked into every registry-issued JWT (pair-attach credentials
/// and auth cookies alike). Verifiers bind to it so a token minted for one
/// purpose can't be replayed as another even though both share the signing
/// secret (NAP-266).
pub const REGISTRY_ISSUER: &str = "maestro-registry";
/// Name of the auth cookie the registry sets and the CLI persists. Defined
/// once here so both ends of the wire agree (NAP-273).
pub const COOKIE_NAME: &str = "maestro_session";
/// Find a named cookie's value inside a `Cookie:` / `Set-Cookie:` header value.
///
/// Iterates every `;`-separated `name=value` pair, so it works whether the
/// target pair is first or buried among other cookies or `Set-Cookie`
/// attributes (`HttpOnly`, `Path=/`, …). Returns the first match, or `None`.
/// Shared by the registry's inbound-cookie check and the CLI's `Set-Cookie`
/// capture (NAP-273).
pub fn extract_cookie<'a>(header_value: &'a str, name: &str) -> Option<&'a str> {
for part in header_value.split(';') {
let part = part.trim();
if let Some((k, v)) = part.split_once('=') {
if k.trim() == name {
return Some(v.trim());
}
}
}
None
}
/// Pair-scoped attach credential. Issued by the registry inside
/// [`AttachCandidates::jwt`]; verified by the target daemon for direct
/// transports (T1/T2) and gated by `allow_inbound` after JWT verification.
///
/// "Pair-scoped" means: `sub` (home daemon) may initiate one attach to
/// `aud` (target daemon) for session `sid`, valid until `exp`. A stolen
/// token still only works between the two daemons it was issued for.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PairAttachClaims {
/// Issuer — registry id.
pub iss: String,
/// Subject — home daemon-id (the daemon initiating attach).
pub sub: String,
/// Audience — target daemon-id (the daemon owning the session).
pub aud: String,
/// Session id this credential is scoped to.
pub sid: Uuid,
/// Expiry (unix seconds). Short-lived; v1 default 60s.
pub exp: i64,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn extract_cookie_finds_named_value_anywhere() {
// Cookie: header — pair may be buried among others.
let h = "foo=1; maestro_session=abc.def.ghi; bar=2";
assert_eq!(extract_cookie(h, COOKIE_NAME), Some("abc.def.ghi"));
assert_eq!(extract_cookie(h, "foo"), Some("1"));
assert_eq!(extract_cookie(h, "missing"), None);
// Set-Cookie: header — cookie first, attributes (no `=` or other keys)
// after it must be skipped, not matched.
let sc = "maestro_session=xyz; HttpOnly; SameSite=Strict; Path=/; Max-Age=2592000";
assert_eq!(extract_cookie(sc, COOKIE_NAME), Some("xyz"));
assert_eq!(extract_cookie("other=1; HttpOnly", COOKIE_NAME), None);
}
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(),
home_dir: String::new(),
};
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 { perimeter: None },
DaemonOutbound::Heartbeat {
perimeter: Some(PerimeterCandidate {
perimeter_id: "wg-naps62".into(),
private_addr: "10.0.0.7:7070".into(),
}),
},
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_deserializes_to_unknown() {
// A newer peer can introduce a frame `type` this build has never heard
// of. It must decode to the forward-compat `Unknown` variant rather
// than erroring, so the reader skips it instead of dropping the WS
// (NAP-269).
let bad = serde_json::json!({ "type": "Bogus", "extra": 1 });
assert_eq!(
serde_json::from_value::<DaemonOutbound>(bad).unwrap(),
DaemonOutbound::Unknown
);
let bad_inbound = serde_json::json!({ "type": "AlsoBogus" });
assert_eq!(
serde_json::from_value::<DaemonInbound>(bad_inbound).unwrap(),
DaemonInbound::Unknown
);
}
#[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(),
home_dir: "/home/naps62".into(),
};
assert_eq!(d, round_trip(&d));
}
#[test]
fn daemon_info_home_dir_defaults_when_absent() {
// Older daemons / pre-NAP-328 payloads omit `home_dir`; it must
// deserialize to an empty string rather than failing.
let json = r#"{
"daemon_id": "laptop",
"hostname": "naps-laptop",
"version": "0.1.0",
"daemon_url": "ws://10.0.0.1:8081"
}"#;
let d: DaemonInfo = serde_json::from_str(json).unwrap();
assert_eq!(d.home_dir, "");
}
#[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(),
home_dir: String::new(),
};
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 heartbeat_legacy_payload_parses_with_none_fields() {
// Daemons predating NAP-242 send the bare `{"type":"Heartbeat"}` shape.
// It must still deserialize, with both new fields defaulting to None.
let legacy = serde_json::json!({ "type": "Heartbeat" });
let parsed: DaemonOutbound = serde_json::from_value(legacy).unwrap();
match parsed {
DaemonOutbound::Heartbeat { perimeter } => {
assert!(perimeter.is_none());
}
other => panic!("expected Heartbeat, got {other:?}"),
}
}
#[test]
fn heartbeat_empty_serializes_to_legacy_shape() {
// A new daemon that doesn't set a perimeter must produce wire bytes
// identical to an old daemon — registries doing equality checks or
// logging payloads see no surprise.
let hb = DaemonOutbound::Heartbeat { perimeter: None };
let v = serde_json::to_value(&hb).unwrap();
assert_eq!(v, serde_json::json!({ "type": "Heartbeat" }));
}
#[test]
fn heartbeat_with_fields_round_trips() {
let hb = DaemonOutbound::Heartbeat {
perimeter: Some(PerimeterCandidate {
perimeter_id: "wg-naps62".into(),
private_addr: "10.0.0.7:7070".into(),
}),
};
let v = serde_json::to_value(&hb).unwrap();
assert_eq!(v["type"], "Heartbeat");
assert_eq!(v["perimeter"]["perimeter_id"], "wg-naps62");
assert_eq!(v["perimeter"]["private_addr"], "10.0.0.7:7070");
assert_eq!(hb, round_trip(&hb));
}
#[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_forward_create_round_trip_and_tag() {
let msg = DaemonOutbound::ForwardCreate {
reply_to: Uuid::from_u128(51),
request: CreateSessionRequest {
daemon_id: "konishi".into(),
cwd: PathBuf::from("/tmp"),
command: "bash".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
},
};
let v = serde_json::to_value(&msg).unwrap();
assert_eq!(v["type"], "ForwardCreate");
assert_eq!(v["request"]["daemon_id"], "konishi");
assert_eq!(msg, round_trip(&msg));
}
#[test]
fn daemon_outbound_forward_kill_round_trip_and_tag() {
let msg = DaemonOutbound::ForwardKill {
reply_to: Uuid::from_u128(52),
daemon_id: "konishi".into(),
id: Uuid::from_u128(99),
};
let v = serde_json::to_value(&msg).unwrap();
assert_eq!(v["type"], "ForwardKill");
assert_eq!(v["daemon_id"], "konishi");
assert_eq!(msg, round_trip(&msg));
}
#[test]
fn daemon_inbound_forward_results_round_trip() {
let create_ok = DaemonInbound::ForwardCreateResult {
reply_to: Uuid::from_u128(51),
result: RpcResult::Ok {
value: sample_session(),
},
};
let create_err = DaemonInbound::ForwardCreateResult {
reply_to: Uuid::from_u128(51),
result: RpcResult::Err {
error: ErrorInfo {
code: "daemon_not_connected".into(),
message: "owner offline".into(),
},
},
};
let kill_ok = DaemonInbound::ForwardKillResult {
reply_to: Uuid::from_u128(52),
result: RpcResult::Ok { value: () },
};
let v = serde_json::to_value(&create_ok).unwrap();
assert_eq!(v["type"], "ForwardCreateResult");
for m in [create_ok, create_err, kill_ok] {
assert_eq!(m, round_trip(&m));
}
}
#[test]
fn dir_listing_round_trips_and_truncated_defaults() {
let listing = DirListing {
path: PathBuf::from("/home/naps62"),
entries: vec!["projects".into(), ".config".into()],
truncated: true,
};
assert_eq!(listing, round_trip(&listing));
// Older daemons omit `truncated`; it must default to false.
let json = serde_json::json!({
"path": "/home/naps62",
"entries": ["projects"]
});
let parsed: DirListing = serde_json::from_value(json).unwrap();
assert!(!parsed.truncated);
}
#[test]
fn daemon_list_dir_round_trips_and_tag() {
let req = DaemonInbound::ListDir {
reply_to: Uuid::from_u128(61),
path: "~/projects".into(),
};
let v = serde_json::to_value(&req).unwrap();
assert_eq!(v["type"], "ListDir");
assert_eq!(v["path"], "~/projects");
assert_eq!(req, round_trip(&req));
let res = DaemonOutbound::ListDirResult {
reply_to: Uuid::from_u128(61),
result: RpcResult::Ok {
value: DirListing {
path: PathBuf::from("/home/naps62/projects"),
entries: vec!["maestro".into()],
truncated: false,
},
},
};
let v = serde_json::to_value(&res).unwrap();
assert_eq!(v["type"], "ListDirResult");
assert_eq!(res, round_trip(&res));
}
#[test]
fn daemon_forward_list_dir_round_trips_and_tag() {
let fwd = DaemonOutbound::ForwardListDir {
reply_to: Uuid::from_u128(62),
daemon_id: "konishi".into(),
path: "/work".into(),
};
let v = serde_json::to_value(&fwd).unwrap();
assert_eq!(v["type"], "ForwardListDir");
assert_eq!(v["daemon_id"], "konishi");
assert_eq!(fwd, round_trip(&fwd));
let res = DaemonInbound::ForwardListDirResult {
reply_to: Uuid::from_u128(62),
result: RpcResult::Err {
error: ErrorInfo {
code: "permission_denied".into(),
message: "EACCES".into(),
},
},
};
let v = serde_json::to_value(&res).unwrap();
assert_eq!(v["type"], "ForwardListDirResult");
assert_eq!(res, round_trip(&res));
}
#[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 registry_event_for_peer_filters_self() {
let mine = sample_session(); // daemon_id = "laptop"
let mut peer_session = sample_session();
peer_session.daemon_id = "konishi".into();
peer_session.id = Uuid::from_u128(99);
let snapshot = RegistryEvent::Snapshot {
daemons: vec![
DaemonInfo {
daemon_id: "laptop".into(),
hostname: "naps-laptop".into(),
version: "0.1.0".into(),
daemon_url: "ws://10.0.0.1:7070".into(),
home_dir: String::new(),
},
DaemonInfo {
daemon_id: "konishi".into(),
hostname: "konishi".into(),
version: "0.1.0".into(),
daemon_url: "ws://10.0.0.2:7070".into(),
home_dir: String::new(),
},
],
sessions: vec![mine.clone(), peer_session.clone()],
};
let filtered = snapshot.clone().for_peer("laptop").unwrap();
match filtered {
RegistryEvent::Snapshot { daemons, sessions } => {
assert_eq!(daemons.len(), 1);
assert_eq!(daemons[0].daemon_id, "konishi");
assert_eq!(sessions.len(), 1);
assert_eq!(sessions[0].daemon_id, "konishi");
}
other => panic!("unexpected {other:?}"),
}
}
#[test]
fn registry_event_for_peer_drops_self_deltas() {
let mine = sample_session(); // daemon_id "laptop"
assert!(RegistryEvent::SessionAdded {
session: mine.clone()
}
.for_peer("laptop")
.is_none());
assert!(RegistryEvent::SessionUpdated {
session: mine.clone()
}
.for_peer("laptop")
.is_none());
assert!(RegistryEvent::SessionRemoved {
daemon_id: "laptop".into(),
id: mine.id,
}
.for_peer("laptop")
.is_none());
assert!(RegistryEvent::DaemonAdded {
daemon: DaemonInfo {
daemon_id: "laptop".into(),
hostname: "x".into(),
version: "y".into(),
daemon_url: "z".into(),
home_dir: String::new(),
}
}
.for_peer("laptop")
.is_none());
assert!(RegistryEvent::DaemonRemoved {
daemon_id: "laptop".into()
}
.for_peer("laptop")
.is_none());
}
#[test]
fn registry_event_for_peer_forwards_peer_deltas_and_lagged() {
let mut peer = sample_session();
peer.daemon_id = "konishi".into();
assert!(matches!(
RegistryEvent::SessionAdded {
session: peer.clone()
}
.for_peer("laptop"),
Some(RegistryEvent::SessionAdded { .. })
));
assert!(matches!(
RegistryEvent::Lagged { skipped: 3 }.for_peer("laptop"),
Some(RegistryEvent::Lagged { skipped: 3 })
));
}
#[test]
fn daemon_inbound_catalog_event_round_trip_and_tag() {
let snapshot = DaemonInbound::CatalogEvent {
event: RegistryEvent::Snapshot {
daemons: vec![DaemonInfo {
daemon_id: "peer".into(),
hostname: "peer-host".into(),
version: "0.1.0".into(),
daemon_url: "ws://peer:7070".into(),
home_dir: String::new(),
}],
sessions: vec![sample_session()],
},
};
let v = serde_json::to_value(&snapshot).unwrap();
assert_eq!(v["type"], "CatalogEvent");
assert_eq!(snapshot, round_trip(&snapshot));
let delta = DaemonInbound::CatalogEvent {
event: RegistryEvent::SessionAdded {
session: sample_session(),
},
};
assert_eq!(delta, round_trip(&delta));
}
#[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());
}
#[test]
fn pair_attach_claims_round_trip() {
let c = PairAttachClaims {
iss: "registry.naps62".into(),
sub: "laptop".into(),
aud: "konishi".into(),
sid: Uuid::from_u128(7),
exp: 1_700_000_060,
};
assert_eq!(c, round_trip(&c));
// Standard JWT short claim names — keep them on the wire as-is so the
// existing `jsonwebtoken` validators don't need a custom name map.
let v = serde_json::to_value(&c).unwrap();
assert_eq!(v["iss"], "registry.naps62");
assert_eq!(v["sub"], "laptop");
assert_eq!(v["aud"], "konishi");
assert_eq!(v["exp"], 1_700_000_060_i64);
}
#[test]
fn pair_attach_claims_reject_missing_audience() {
// aud is mandatory — without it a token isn't pair-scoped at all and
// the target daemon couldn't verify itself as the intended recipient.
let bad = serde_json::json!({
"iss": "registry.naps62",
"sub": "laptop",
"sid": Uuid::from_u128(7),
"exp": 1_700_000_060_i64,
});
assert!(serde_json::from_value::<PairAttachClaims>(bad).is_err());
}
}