feat(maestro): registry relays Forward{Create,Kill} RPC for cross-daemon ops (NAP-327)

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) <noreply@anthropic.com>

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<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 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 {

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.

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<Session>) -> bool {
         let entry = self.pending_create.lock().unwrap().remove(&reply_to);
@@ -598,6 +662,22 @@ async fn await_rpc<T>(rx: oneshot::Receiver<RpcResult<T>>) -> Result<RpcResult<T
     }
 }
 
+/// Map a transport-level [`RegistryError`] to a wire [`ErrorInfo`] with a
+/// stable `code` so the home daemon can translate it to an HTTP status
+/// (NAP-327). Daemon-reported faults keep their own codes via [`RpcError`].
+fn registry_error_info(e: &RegistryError) -> 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<T>(rpc: RpcResult<T>) -> Result<T, RpcError> {
     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(&reg, "laptop").await;
+        let (_tt, mut target_rx) = register(&reg, "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(&reg, "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(&reg, "laptop").await;
+        let (_tt, mut target_rx) = register(&reg, "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]