From 1e4b65c73c407b010db168d428e3dc1bb4c998b1 Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Mon, 15 Jun 2026 16:24:48 +0000 Subject: [PATCH] feat(maestro): daemon brokers cross-daemon create/kill instead of 501/404 (NAP-327) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- crates/maestro-daemon/src/broker.rs | 203 ++++++++++++++++++- crates/maestro-daemon/src/cli.rs | 1 + crates/maestro-daemon/src/local_api.rs | 78 +++++-- crates/maestro-daemon/src/registry_client.rs | 6 + crates/maestro-daemon/tests/local_api.rs | 22 +- 5 files changed, 281 insertions(+), 29 deletions(-) diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs index 073669e..7b6d7e4 100644 --- a/crates/maestro-daemon/src/broker.rs +++ b/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, pending: Arc>>>>, + /// Pending `ForwardCreate` replies, keyed by `reply_to` (NAP-327). Same + /// fail-fast + RAII-cleanup discipline as `pending`. + pending_create: Arc>>>>, + /// Pending `ForwardKill` replies, keyed by `reply_to` (NAP-327). + pending_kill: Arc>>>>, /// 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 { + 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, + ) -> 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). diff --git a/crates/maestro-daemon/src/cli.rs b/crates/maestro-daemon/src/cli.rs index 6cefeb4..08719c2 100644 --- a/crates/maestro-daemon/src/cli.rs +++ b/crates/maestro-daemon/src/cli.rs @@ -140,6 +140,7 @@ async fn run(config_override: Option) -> Result<(), Box> { config.daemon_url.clone(), catalog.clone(), events, + client.broker(), ); let broker_state = local_api::BrokerState { manager: manager.clone(), diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 9a1a985..a5ad0da 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/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, ) -> 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, Path(id): Path) -> 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, Path(id): Path) -> 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, 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(), diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index 6d36ac7..88fa2a6 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/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. diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 23d5447..7637986 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/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]