From daeacc750feda95461da752b3ea1bd2c47340e91 Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Mon, 15 Jun 2026 16:21:11 +0000 Subject: [PATCH] feat(maestro): registry relays Forward{Create,Kill} RPC for cross-daemon ops (NAP-327) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the home-daemon → registry control-link RPC that brokers a cross-daemon session create/kill without putting any session bytes on the registry: the owning daemon still does the tmux spawn, the registry just relays request and reply. - protocol: `DaemonOutbound::{ForwardCreate,ForwardKill}` + `DaemonInbound::{ForwardCreateResult,ForwardKillResult}` (+ round-trip tests). - registry: `Registry::{forward_create,forward_kill}` reuse the existing `create_session`/`kill_session` R→D machinery and route the reply back to the home daemon; transport faults map to stable `ErrorInfo` codes. Wired into the `WS /daemon` dispatch. Next: daemon-side broker handle + local_api wiring. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/maestro-protocol/src/lib.rs | 89 +++++++++++ crates/maestro-registry/src/handler.rs | 14 ++ crates/maestro-registry/src/state.rs | 195 +++++++++++++++++++++++++ 3 files changed, 298 insertions(+) diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index e937a6b..106e135 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -136,6 +136,23 @@ pub enum DaemonOutbound { 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, + }, /// Target daemon answers a [`DaemonInbound::RelaySetup`]: whether it will /// accept relay traffic for the requested session. Registry only allocates /// relay capacity when `accepted == true`. @@ -170,6 +187,18 @@ pub enum DaemonInbound { reply_to: ReplyTo, result: RpcResult, }, + /// 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, + }, + /// Registry's reply to [`DaemonOutbound::ForwardKill`]. (NAP-327) + ForwardKillResult { + reply_to: ReplyTo, + result: RpcResult<()>, + }, /// 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`]. @@ -731,6 +760,66 @@ mod tests { 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 daemon_outbound_relay_ack_round_trip_and_tag() { let msg = DaemonOutbound::RelayAck { diff --git a/crates/maestro-registry/src/handler.rs b/crates/maestro-registry/src/handler.rs index 796e64d..ff1cf42 100644 --- a/crates/maestro-registry/src/handler.rs +++ b/crates/maestro-registry/src/handler.rs @@ -211,6 +211,20 @@ async fn dispatch_text( .resolve_attach(daemon_id, reply_to, session_id, shared_secret) .await; } + DaemonOutbound::ForwardCreate { reply_to, request } => { + // Relay the create to its owning daemon and route the reply back to + // this (home) daemon — control-plane only (NAP-327). + registry.forward_create(daemon_id, reply_to, request).await; + } + DaemonOutbound::ForwardKill { + reply_to, + daemon_id: target, + id, + } => { + registry + .forward_kill(daemon_id, reply_to, &target, id) + .await; + } DaemonOutbound::RelayAck { reply_to, .. } => { // T3 (relay) is gated on NAP-238. Land the variant so newer daemons // don't crash the connection; the registry doesn't act on it yet. diff --git a/crates/maestro-registry/src/state.rs b/crates/maestro-registry/src/state.rs index 829a275..3116e15 100644 --- a/crates/maestro-registry/src/state.rs +++ b/crates/maestro-registry/src/state.rs @@ -429,6 +429,70 @@ impl Registry { } } + /// Relay a home daemon's `ForwardCreate` to the owning daemon and push the + /// result back to the home daemon as `ForwardCreateResult` (NAP-327). + /// + /// Control-plane only: the owning daemon does the actual tmux spawn; the + /// registry just relays request and reply, so no session bytes traverse it. + /// Reuses [`create_session`](Self::create_session) (the same R→D machinery + /// the legacy HTTP path uses), then routes the reply to `home_daemon_id`. + pub async fn forward_create( + &self, + home_daemon_id: &str, + reply_to: ReplyTo, + request: CreateSessionRequest, + ) { + let owner = request.daemon_id.clone(); + let result = match self.create_session(&owner, request).await { + Ok(Ok(session)) => RpcResult::Ok { value: session }, + Ok(Err(RpcError(error))) => RpcResult::Err { error }, + Err(e) => RpcResult::Err { + error: registry_error_info(&e), + }, + }; + self.reply_to_daemon( + home_daemon_id, + DaemonInbound::ForwardCreateResult { reply_to, result }, + ) + .await; + } + + /// Relay a home daemon's `ForwardKill` to the owning daemon and push the + /// `ForwardKillResult` back to the home daemon (NAP-327). + pub async fn forward_kill( + &self, + home_daemon_id: &str, + reply_to: ReplyTo, + target_daemon_id: &str, + id: Uuid, + ) { + let result = match self.kill_session(target_daemon_id, id).await { + Ok(Ok(())) => RpcResult::Ok { value: () }, + Ok(Err(RpcError(error))) => RpcResult::Err { error }, + Err(e) => RpcResult::Err { + error: registry_error_info(&e), + }, + }; + self.reply_to_daemon( + home_daemon_id, + DaemonInbound::ForwardKillResult { reply_to, result }, + ) + .await; + } + + /// Send a single frame to a connected daemon's outbound channel, dropping + /// it silently if the daemon has gone away mid-flight (the home daemon's + /// own RPC timeout will surface the failure). + async fn reply_to_daemon(&self, daemon_id: &str, frame: DaemonInbound) { + let outbound = { + let state = self.inner.lock().await; + state.daemons.get(daemon_id).map(|e| e.outbound.clone()) + }; + if let Some(outbound) = outbound { + let _ = outbound.send(frame).await; + } + } + /// Resolve a CreateSession RPC reply. Returns true if a waiter was found. pub async fn complete_create(&self, reply_to: ReplyTo, result: RpcResult) -> bool { let entry = self.pending_create.lock().unwrap().remove(&reply_to); @@ -598,6 +662,22 @@ async fn await_rpc(rx: oneshot::Receiver>) -> Result ErrorInfo { + let code = match e { + RegistryError::DaemonNotConnected(_) => "daemon_not_connected", + RegistryError::DaemonGone => "daemon_gone", + RegistryError::ReplyChannelClosed => "reply_closed", + RegistryError::Timeout(_) => "rpc_timeout", + }; + ErrorInfo { + code: code.into(), + message: e.to_string(), + } +} + fn rpc_into_result(rpc: RpcResult) -> Result { match rpc { RpcResult::Ok { value } => Ok(value), @@ -750,6 +830,121 @@ mod tests { } } + // ----- NAP-327: forwarded create / kill ----- + + /// A `ForwardCreate` from the home daemon is relayed to the owning daemon as + /// a `CreateSession`, and the owner's success reply comes back to the home + /// daemon as a `ForwardCreateResult` carrying the new session. + #[tokio::test] + async fn forward_create_relays_to_owner_and_replies_to_home() { + let reg = Registry::new(); + let (_th, mut home_rx) = register(®, "laptop").await; + let (_tt, mut target_rx) = register(®, "konishi").await; + + let reply_to = Uuid::new_v4(); + let request = CreateSessionRequest { + daemon_id: "konishi".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + + let r = reg.clone(); + let task = tokio::spawn(async move { r.forward_create("laptop", reply_to, request).await }); + + // The owner receives a CreateSession; learn its reply_to and answer. + let sid = Uuid::new_v4(); + let owner_reply_to = match target_rx.recv().await.expect("owner got CreateSession") { + DaemonInbound::CreateSession { reply_to, request } => { + assert_eq!(request.daemon_id, "konishi"); + reply_to + } + other => panic!("expected CreateSession, got {other:?}"), + }; + reg.complete_create( + owner_reply_to, + RpcResult::Ok { + value: sample_session(sid, "konishi"), + }, + ) + .await; + task.await.unwrap(); + + // The home daemon gets the relayed result. + match home_rx.recv().await.expect("home got ForwardCreateResult") { + DaemonInbound::ForwardCreateResult { reply_to: got, result } => { + assert_eq!(got, reply_to); + match result { + RpcResult::Ok { value } => { + assert_eq!(value.id, sid); + assert_eq!(value.daemon_id, "konishi"); + } + RpcResult::Err { error } => panic!("unexpected err {error:?}"), + } + } + other => panic!("expected ForwardCreateResult, got {other:?}"), + } + } + + /// Forwarding to an owner that isn't connected surfaces a structured + /// `daemon_not_connected` error back to the home daemon (NAP-327). + #[tokio::test] + async fn forward_create_owner_offline_replies_error_to_home() { + let reg = Registry::new(); + let (_th, mut home_rx) = register(®, "laptop").await; + + let request = CreateSessionRequest { + daemon_id: "ghost".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + reg.forward_create("laptop", Uuid::new_v4(), request).await; + + match home_rx.recv().await.expect("home got ForwardCreateResult") { + DaemonInbound::ForwardCreateResult { result, .. } => match result { + RpcResult::Err { error } => assert_eq!(error.code, "daemon_not_connected"), + RpcResult::Ok { .. } => panic!("create to offline owner must fail"), + }, + other => panic!("expected ForwardCreateResult, got {other:?}"), + } + } + + /// A `ForwardKill` is relayed to the owner and the success reply returns to + /// the home daemon (NAP-327). + #[tokio::test] + async fn forward_kill_relays_to_owner_and_replies_to_home() { + let reg = Registry::new(); + let (_th, mut home_rx) = register(®, "laptop").await; + let (_tt, mut target_rx) = register(®, "konishi").await; + let id = Uuid::new_v4(); + + let r = reg.clone(); + let task = tokio::spawn(async move { r.forward_kill("laptop", Uuid::new_v4(), "konishi", id).await }); + + let owner_reply_to = match target_rx.recv().await.expect("owner got KillSession") { + DaemonInbound::KillSession { reply_to, id: got } => { + assert_eq!(got, id); + reply_to + } + other => panic!("expected KillSession, got {other:?}"), + }; + reg.complete_kill(owner_reply_to, RpcResult::Ok { value: () }) + .await; + task.await.unwrap(); + + match home_rx.recv().await.expect("home got ForwardKillResult") { + DaemonInbound::ForwardKillResult { result, .. } => { + assert!(matches!(result, RpcResult::Ok { .. })); + } + other => panic!("expected ForwardKillResult, got {other:?}"), + } + } + /// Resolving an attach for a session no daemon owns yields /// `unknown_session`, not a JWT for some default daemon (NAP-270). #[tokio::test]