feat(maestro): daemon brokers cross-daemon create/kill instead of 501/404 (NAP-327)

The home daemon's `POST /api/sessions` (peer `daemon_id`) and
`DELETE /api/sessions/:id` (peer-owned) now broker the op to the owning daemon
over the registry control link rather than returning 501/404. This fixes
cross-daemon spawn/kill for *all* local clients — notably maestro-web, which
talks only to the local daemon and had no registry fallback.

- broker: `BrokerHandle::{forward_create,forward_kill}` + completion handlers,
  mirroring `resolve_attach` (fail-fast while disconnected, RAII pending
  cleanup, RESOLVE_TIMEOUT).
- registry_client: route `Forward{Create,Kill}Result` back to the broker.
- local_api: create forwards on peer `daemon_id`; kill looks up the owner in
  the catalog mirror and forwards (only a truly-unknown session is 404).
  `broker_error_response` maps BrokerError → HTTP (503/504/502/404/...).
- cli: thread the broker into `LocalApiState`.

Tests: broker round-trip / fail-fast / registry-err propagation; updated the
local_api integration test (peer create now 503s without a registry link
instead of 501, proving it brokers).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/maestro-daemon/src/broker.rs

@@ -11,7 +11,9 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
-use maestro_protocol::{AttachCandidates, DaemonOutbound, ReplyTo, RpcResult};
+use maestro_protocol::{
+    AttachCandidates, CreateSessionRequest, DaemonOutbound, ReplyTo, RpcResult, Session,
+};
 use tokio::sync::{mpsc, oneshot};
 use uuid::Uuid;
 
@@ -57,6 +59,11 @@ pub enum BrokerError {
 pub struct BrokerHandle {
     out_tx: mpsc::Sender<DaemonOutbound>,
     pending: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<AttachCandidates>>>>>,
+    /// Pending `ForwardCreate` replies, keyed by `reply_to` (NAP-327). Same
+    /// fail-fast + RAII-cleanup discipline as `pending`.
+    pending_create: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<Session>>>>>,
+    /// Pending `ForwardKill` replies, keyed by `reply_to` (NAP-327).
+    pending_kill: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<()>>>>>,
     /// Whether the registry control link is currently up. Flipped by the
     /// registry-client task via [`BrokerHandle::set_connected`]. When down,
     /// `resolve_attach` fails fast instead of queueing onto an undrained
@@ -76,6 +83,8 @@ impl BrokerHandle {
         let handle = Self {
             out_tx,
             pending: Arc::new(Mutex::new(HashMap::new())),
+            pending_create: Arc::new(Mutex::new(HashMap::new())),
+            pending_kill: Arc::new(Mutex::new(HashMap::new())),
             connected: Arc::new(AtomicBool::new(false)),
         };
         (handle, out_rx)
@@ -146,6 +155,101 @@ impl BrokerHandle {
             }
         }
     }
+
+    /// Broker a cross-daemon create: ask the registry to relay the request to
+    /// the owning daemon (`request.daemon_id`) and return the created session.
+    /// Same fail-fast-while-disconnected + RAII-cleanup discipline as
+    /// [`resolve_attach`](Self::resolve_attach) (NAP-263 / NAP-259, NAP-327).
+    pub async fn forward_create(
+        &self,
+        request: CreateSessionRequest,
+    ) -> Result<Session, BrokerError> {
+        if !self.connected.load(Ordering::Acquire) {
+            return Err(BrokerError::NoControlLink);
+        }
+        let reply_to = Uuid::new_v4();
+        let (tx, rx) = oneshot::channel();
+        self.pending_create.lock().unwrap().insert(reply_to, tx);
+        let _guard = PendingGuard {
+            map: &*self.pending_create,
+            key: reply_to,
+        };
+        if self
+            .out_tx
+            .try_send(DaemonOutbound::ForwardCreate { reply_to, request })
+            .is_err()
+        {
+            return Err(BrokerError::NoControlLink);
+        }
+        match tokio::time::timeout(RESOLVE_TIMEOUT, rx).await {
+            Ok(Ok(RpcResult::Ok { value })) => Ok(value),
+            Ok(Ok(RpcResult::Err { error })) => Err(BrokerError::Registry(error.code, error.message)),
+            Ok(Err(_)) => Err(BrokerError::ReplyClosed),
+            Err(_) => Err(BrokerError::Timeout(RESOLVE_TIMEOUT)),
+        }
+    }
+
+    /// Broker a cross-daemon kill: ask the registry to relay the kill to the
+    /// daemon that owns `id` (NAP-327).
+    pub async fn forward_kill(&self, daemon_id: String, id: Uuid) -> Result<(), BrokerError> {
+        if !self.connected.load(Ordering::Acquire) {
+            return Err(BrokerError::NoControlLink);
+        }
+        let reply_to = Uuid::new_v4();
+        let (tx, rx) = oneshot::channel();
+        self.pending_kill.lock().unwrap().insert(reply_to, tx);
+        let _guard = PendingGuard {
+            map: &*self.pending_kill,
+            key: reply_to,
+        };
+        if self
+            .out_tx
+            .try_send(DaemonOutbound::ForwardKill {
+                reply_to,
+                daemon_id,
+                id,
+            })
+            .is_err()
+        {
+            return Err(BrokerError::NoControlLink);
+        }
+        match tokio::time::timeout(RESOLVE_TIMEOUT, rx).await {
+            Ok(Ok(RpcResult::Ok { .. })) => Ok(()),
+            Ok(Ok(RpcResult::Err { error })) => Err(BrokerError::Registry(error.code, error.message)),
+            Ok(Err(_)) => Err(BrokerError::ReplyClosed),
+            Err(_) => Err(BrokerError::Timeout(RESOLVE_TIMEOUT)),
+        }
+    }
+
+    /// Resolve a waiting [`forward_create`](Self::forward_create); called by the
+    /// registry-client task on a `ForwardCreateResult` frame.
+    pub async fn complete_forward_create(
+        &self,
+        reply_to: ReplyTo,
+        result: RpcResult<Session>,
+    ) -> bool {
+        let entry = self.pending_create.lock().unwrap().remove(&reply_to);
+        match entry {
+            Some(tx) => tx.send(result).is_ok(),
+            None => {
+                tracing::debug!(?reply_to, "no waiter for ForwardCreateResult");
+                false
+            }
+        }
+    }
+
+    /// Resolve a waiting [`forward_kill`](Self::forward_kill); called by the
+    /// registry-client task on a `ForwardKillResult` frame.
+    pub async fn complete_forward_kill(&self, reply_to: ReplyTo, result: RpcResult<()>) -> bool {
+        let entry = self.pending_kill.lock().unwrap().remove(&reply_to);
+        match entry {
+            Some(tx) => tx.send(result).is_ok(),
+            None => {
+                tracing::debug!(?reply_to, "no waiter for ForwardKillResult");
+                false
+            }
+        }
+    }
 }
 
 #[cfg(test)]
@@ -239,6 +343,103 @@ mod tests {
         }
     }
 
+    // ----- NAP-327: forward_create / forward_kill -----
+
+    fn sample_request() -> CreateSessionRequest {
+        CreateSessionRequest {
+            daemon_id: "konishi".into(),
+            cwd: std::path::PathBuf::from("/tmp"),
+            command: "bash".into(),
+            env: Default::default(),
+            metadata: Default::default(),
+            worktree: None,
+        }
+    }
+
+    /// A `forward_create` issues a `ForwardCreate` frame and resolves with the
+    /// session once the registry's `ForwardCreateResult` is delivered.
+    #[tokio::test]
+    async fn forward_create_round_trips_session() {
+        let (handle, mut out_rx) = BrokerHandle::new();
+        handle.set_connected(true);
+
+        let h = handle.clone();
+        let task = tokio::spawn(async move { h.forward_create(sample_request()).await });
+
+        let reply_to = match out_rx.recv().await.expect("ForwardCreate went out") {
+            DaemonOutbound::ForwardCreate { reply_to, request } => {
+                assert_eq!(request.daemon_id, "konishi");
+                reply_to
+            }
+            other => panic!("expected ForwardCreate, got {other:?}"),
+        };
+        let session = Session {
+            id: Uuid::new_v4(),
+            daemon_id: "konishi".into(),
+            cwd: std::path::PathBuf::from("/tmp"),
+            command: "bash".into(),
+            env: Default::default(),
+            metadata: Default::default(),
+            status: maestro_protocol::SessionStatus::Running,
+            created_at: chrono::Utc::now(),
+        };
+        handle
+            .complete_forward_create(reply_to, RpcResult::Ok { value: session.clone() })
+            .await;
+
+        let got = task.await.unwrap().expect("create ok");
+        assert_eq!(got.id, session.id);
+    }
+
+    /// While disconnected, `forward_create` / `forward_kill` fail fast with
+    /// `NoControlLink` and leave no pending entry (NAP-263 / NAP-259).
+    #[tokio::test]
+    async fn forward_ops_fail_fast_while_disconnected() {
+        let (handle, _out_rx) = BrokerHandle::new(); // starts disconnected
+        assert!(matches!(
+            handle.forward_create(sample_request()).await,
+            Err(BrokerError::NoControlLink)
+        ));
+        assert!(matches!(
+            handle.forward_kill("konishi".into(), Uuid::new_v4()).await,
+            Err(BrokerError::NoControlLink)
+        ));
+        assert_eq!(handle.pending_create.lock().unwrap().len(), 0);
+        assert_eq!(handle.pending_kill.lock().unwrap().len(), 0);
+    }
+
+    /// A registry-reported error in the reply surfaces as `BrokerError::Registry`
+    /// carrying the code, so the HTTP handler can map it to a status.
+    #[tokio::test]
+    async fn forward_kill_propagates_registry_err() {
+        let (handle, mut out_rx) = BrokerHandle::new();
+        handle.set_connected(true);
+
+        let h = handle.clone();
+        let task = tokio::spawn(async move { h.forward_kill("konishi".into(), Uuid::new_v4()).await });
+
+        let reply_to = match out_rx.recv().await.expect("ForwardKill went out") {
+            DaemonOutbound::ForwardKill { reply_to, .. } => reply_to,
+            other => panic!("expected ForwardKill, got {other:?}"),
+        };
+        handle
+            .complete_forward_kill(
+                reply_to,
+                RpcResult::Err {
+                    error: maestro_protocol::ErrorInfo {
+                        code: "daemon_not_connected".into(),
+                        message: "owner offline".into(),
+                    },
+                },
+            )
+            .await;
+
+        match task.await.unwrap() {
+            Err(BrokerError::Registry(code, _)) => assert_eq!(code, "daemon_not_connected"),
+            other => panic!("expected Registry error, got {other:?}"),
+        }
+    }
+
     /// NAP-270: if the registry never replies, `resolve_attach` resolves to
     /// `BrokerError::Timeout` after `RESOLVE_TIMEOUT` (driven here with a paused
     /// clock so the test is instant).

crates/maestro-daemon/src/cli.rs

@@ -140,6 +140,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
         config.daemon_url.clone(),
         catalog.clone(),
         events,
+        client.broker(),
     );
     let broker_state = local_api::BrokerState {
         manager: manager.clone(),

crates/maestro-daemon/src/local_api.rs

@@ -58,6 +58,9 @@ pub struct LocalApiState {
     /// Daemon-local fan-out of own + peer catalog deltas (NAP-248). The
     /// `/api/events` WS subscribes here; everything else just reads REST.
     pub events: LocalEventBus,
+    /// Control link to the registry, used to broker cross-daemon create/kill
+    /// (NAP-327). When the link is down, brokered ops fail fast.
+    pub broker: BrokerHandle,
 }
 
 pub fn router(state: LocalApiState) -> Router {
@@ -102,18 +105,12 @@ async fn create_session(
     Json(body): Json<CreateSessionRequest>,
 ) -> Response {
     if body.daemon_id != s.manager.daemon_id() {
-        // Forwarding to peer daemons is NAP-247 (broker mode). Until then,
-        // tell the caller exactly why we refused — the registry's API still
-        // accepts cross-daemon creates the old way, so this isn't a regression.
-        return error_response(
-            StatusCode::NOT_IMPLEMENTED,
-            "remote_create_unsupported",
-            &format!(
-                "this daemon ({}) can only create sessions for itself; \
-                 cross-daemon broker is NAP-247",
-                s.manager.daemon_id()
-            ),
-        );
+        // Cross-daemon create: broker it to the owning daemon over the registry
+        // control link (NAP-327). Control-plane only — the owner does the spawn.
+        return match s.broker.forward_create(body).await {
+            Ok(session) => (StatusCode::CREATED, Json(session)).into_response(),
+            Err(e) => broker_error_response(&e),
+        };
     }
     match s.manager.create(body).await {
         Ok(session) => (StatusCode::CREATED, Json(session)).into_response(),
@@ -130,13 +127,27 @@ async fn create_session(
 
 async fn kill_session(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> Response {
     if !s.manager.has(id).await {
-        // The session might exist on a peer daemon, but v1 doesn't broker —
-        // give a clean 404 rather than lying about success.
-        return error_response(
-            StatusCode::NOT_FOUND,
-            "unknown_session",
-            "no such session on this daemon",
-        );
+        // Not ours — if a peer owns it, broker the kill to that daemon over the
+        // registry control link (NAP-327). Only a session no daemon knows about
+        // is a real 404.
+        let owner = s
+            .catalog
+            .sessions()
+            .await
+            .into_iter()
+            .find(|sess| sess.id == id)
+            .map(|sess| sess.daemon_id);
+        return match owner {
+            Some(daemon_id) => match s.broker.forward_kill(daemon_id, id).await {
+                Ok(()) => StatusCode::NO_CONTENT.into_response(),
+                Err(e) => broker_error_response(&e),
+            },
+            None => error_response(
+                StatusCode::NOT_FOUND,
+                "unknown_session",
+                "no such session on this daemon or any known peer",
+            ),
+        };
     }
     match s.manager.kill(id).await {
         Ok(()) => StatusCode::NO_CONTENT.into_response(),
@@ -151,6 +162,29 @@ async fn kill_session(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> R
     }
 }
 
+/// Map a [`BrokerError`] from a forwarded create/kill (NAP-327) to a structured
+/// HTTP error. Mirrors `broker_error_code`'s code points but also picks the
+/// status the CLI/web client should react to.
+fn broker_error_response(e: &BrokerError) -> Response {
+    let (code, message) = broker_error_code(e);
+    let status = match e {
+        // Registry link down or back-pressured: the home daemon can't broker
+        // right now — a retryable 503.
+        BrokerError::NoControlLink => StatusCode::SERVICE_UNAVAILABLE,
+        BrokerError::Timeout(_) => StatusCode::GATEWAY_TIMEOUT,
+        BrokerError::ReplyClosed => StatusCode::BAD_GATEWAY,
+        // Registry- or owner-reported faults. Distinguish "nothing owns this"
+        // (404) from spawn/transport failures (502).
+        BrokerError::Registry(c, _) => match c.as_str() {
+            "unknown_session" => StatusCode::NOT_FOUND,
+            "daemon_not_connected" | "daemon_gone" => StatusCode::BAD_GATEWAY,
+            "rpc_timeout" => StatusCode::GATEWAY_TIMEOUT,
+            _ => StatusCode::BAD_GATEWAY,
+        },
+    };
+    error_response(status, code, &message)
+}
+
 fn error_response(status: StatusCode, code: &str, message: &str) -> Response {
     (
         status,
@@ -172,6 +206,7 @@ pub fn state_for(
     daemon_url: impl Into<String>,
     catalog: CatalogMirror,
     events: LocalEventBus,
+    broker: BrokerHandle,
 ) -> LocalApiState {
     LocalApiState {
         manager,
@@ -179,6 +214,7 @@ pub fn state_for(
         daemon_url: Arc::new(daemon_url.into()),
         catalog,
         events,
+        broker,
     }
 }
 
@@ -446,6 +482,10 @@ fn broker_error_code(e: &BrokerError) -> (&'static str, String) {
                 "unknown_session" => "unknown_session",
                 "self_target" => "self_target",
                 "jwt_failed" => "jwt_failed",
+                // Forwarded create/kill transport faults (NAP-327).
+                "daemon_not_connected" => "daemon_not_connected",
+                "daemon_gone" => "daemon_gone",
+                "rpc_timeout" => "rpc_timeout",
                 _ => "registry_error",
             },
             msg.clone(),

crates/maestro-daemon/src/registry_client.rs

@@ -297,6 +297,12 @@ impl RegistryClient {
             DaemonInbound::ResolveAttachResult { reply_to, result } => {
                 self.broker.complete_resolve(reply_to, result).await;
             }
+            DaemonInbound::ForwardCreateResult { reply_to, result } => {
+                self.broker.complete_forward_create(reply_to, result).await;
+            }
+            DaemonInbound::ForwardKillResult { reply_to, result } => {
+                self.broker.complete_forward_kill(reply_to, result).await;
+            }
             DaemonInbound::RelaySetup { .. } => {
                 // T3 (relay) is gated on NAP-238. Lands here only so newer
                 // registries sending it don't crash this daemon on parse.

crates/maestro-daemon/tests/local_api.rs

@@ -34,17 +34,19 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
     let manager = fx.manager.clone();
     let catalog = CatalogMirror::new();
     let events = LocalEventBus::new();
+    // NAP-247/NAP-327: tests don't exercise the cross-host broker path; hand the
+    // routers an isolated BrokerHandle whose receiver is dropped, so any
+    // resolve_attach / forward_create / forward_kill fails fast with
+    // NoControlLink (the handle starts disconnected).
+    let (broker, _broker_rx) = BrokerHandle::new();
     let state = local_api::state_for(
         manager.clone(),
         hostname,
         daemon_url,
         catalog.clone(),
         events,
+        broker.clone(),
     );
-    // NAP-247: tests don't exercise the cross-host broker path; hand the
-    // router an isolated BrokerHandle whose receiver is dropped, so any
-    // resolve_attach attempt fails fast with NoControlLink.
-    let (broker, _broker_rx) = BrokerHandle::new();
     let broker_state = local_api::BrokerState {
         manager: manager.clone(),
         catalog: catalog.clone(),
@@ -210,9 +212,11 @@ async fn list_sessions_starts_empty_and_reflects_creates() {
 }
 
 #[tokio::test]
-async fn post_to_peer_daemon_id_returns_501_not_proxy() {
-    // Cross-daemon broker is NAP-247; until it lands a daemon must refuse
-    // (rather than silently misroute) a create targeting a peer.
+async fn post_to_peer_daemon_id_brokers_and_503s_without_registry_link() {
+    // Cross-daemon create now brokers to the owner over the registry control
+    // link (NAP-327). This harness has no registry link (the broker starts
+    // disconnected), so the daemon fails fast with a retryable 503 instead of
+    // the old 501 — proving it attempts to broker rather than refusing.
     let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
     let req = CreateSessionRequest {
         daemon_id: "konishi".into(),
@@ -228,9 +232,9 @@ async fn post_to_peer_daemon_id_returns_501_not_proxy() {
         .send()
         .await
         .unwrap();
-    assert_eq!(resp.status(), reqwest::StatusCode::NOT_IMPLEMENTED);
+    assert_eq!(resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
     let body: serde_json::Value = resp.json().await.unwrap();
-    assert_eq!(body["error"]["code"], "remote_create_unsupported");
+    assert_eq!(body["error"]["code"], "no_registry_link");
 }
 
 #[tokio::test]