feat(maestro): daemon broker T2 + pair-JWT verifier + allow_inbound (NAP-247)

* New PairAttachVerifier (crates/maestro-daemon/src/jwt.rs): checks
  signature, aud == self, exp > now, and sub ∈ allow_inbound. Default
  deny — empty allow_inbound rejects every cross-daemon attach.
* /sessions/:id/stream tries the legacy verifier first, falls back to
  the pair verifier on mismatch. Compat with existing direct-attach.
* New BrokerHandle (crates/maestro-daemon/src/broker.rs): shared between
  the registry-client task and HTTP handlers; calls ResolveAttach over
  the control link and correlates the reply via a pending map.
* RegistryClient now owns a BrokerHandle, drains its outbound mpsc
  alongside session events / heartbeats, and routes
  ResolveAttachResult back into the pending map. First heartbeat fires
  immediately so the registry has the daemon's perimeter info before
  the first ResolveAttach.
* /api/attach/🆔 self-owned → local PTY; peer-owned (per catalog
  mirror) → ResolveAttach + T2 dial + bidirectional frame proxy.
  Returns structured errors for missing local perimeter, peer on a
  different perimeter, etc. T1/T3 stay stubbed.

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

crates/maestro-daemon/Cargo.toml

@@ -32,6 +32,7 @@ tokio-tungstenite = { workspace = true, features = ["rustls-tls-webpki-roots"] }
 toml.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
+urlencoding = "2"
 uuid.workspace = true
 
 [dev-dependencies]

crates/maestro-daemon/src/attach.rs

@@ -18,13 +18,18 @@ use serde::Deserialize;
 use tokio::sync::broadcast::error::RecvError;
 use uuid::Uuid;
 
-use crate::jwt::JwtVerifier;
+use crate::jwt::{JwtVerifier, PairAttachVerifier};
 use crate::session::SessionManager;
 
 #[derive(Debug, Clone)]
 pub struct AttachState {
     pub manager: SessionManager,
     pub verifier: Arc<JwtVerifier>,
+    /// Pair-scoped JWT verifier (NAP-247). When the legacy verifier fails
+    /// (e.g. the token was minted as a pair-attach JWT for the broker path),
+    /// we fall back to this one before rejecting the client outright. Local
+    /// attaches always go through the legacy verifier.
+    pub pair_verifier: Arc<PairAttachVerifier>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -32,12 +37,17 @@ struct AttachParams {
     token: String,
 }
 
-pub fn router(manager: SessionManager, verifier: JwtVerifier) -> Router {
+pub fn router(
+    manager: SessionManager,
+    verifier: JwtVerifier,
+    pair_verifier: PairAttachVerifier,
+) -> Router {
     Router::new()
         .route("/sessions/:id/stream", any(attach_ws))
         .with_state(AttachState {
             manager,
             verifier: Arc::new(verifier),
+            pair_verifier: Arc::new(pair_verifier),
         })
 }
 
@@ -47,18 +57,38 @@ async fn attach_ws(
     State(state): State<AttachState>,
     ws: WebSocketUpgrade,
 ) -> Result<Response, (StatusCode, String)> {
-    let claims = state
-        .verifier
-        .verify(&params.token)
-        .map_err(|e| (StatusCode::UNAUTHORIZED, e.to_string()))?;
-    if claims.session_id != id {
-        return Err((
-            StatusCode::FORBIDDEN,
-            format!(
-                "token is for session {} but path requested {}",
-                claims.session_id, id
-            ),
-        ));
+    // Try the legacy `JwtClaims` shape first (browser/CLI direct attach). On
+    // any failure, try the pair-scoped shape used by the home-daemon broker
+    // (NAP-247). One — and only one — must accept.
+    let legacy_err = match state.verifier.verify(&params.token) {
+        Ok(claims) => {
+            if claims.session_id != id {
+                return Err((
+                    StatusCode::FORBIDDEN,
+                    format!(
+                        "token is for session {} but path requested {}",
+                        claims.session_id, id
+                    ),
+                ));
+            }
+            None
+        }
+        Err(e) => Some(e),
+    };
+    if let Some(legacy_err) = legacy_err {
+        match state.pair_verifier.verify(&params.token, id) {
+            Ok(sub) => {
+                tracing::debug!(home_daemon = %sub, %id, "pair-scoped attach accepted");
+            }
+            Err(pair_err) => {
+                // Both rejected — return the legacy error since most clients
+                // still mint that shape today.
+                return Err((
+                    StatusCode::UNAUTHORIZED,
+                    format!("legacy: {legacy_err}; pair: {pair_err}"),
+                ));
+            }
+        }
     }
     if !state.manager.has(id).await {
         return Err((StatusCode::NOT_FOUND, format!("no session {id}")));

crates/maestro-daemon/src/broker.rs

@@ -0,0 +1,103 @@
+//! Home-daemon broker handle (NAP-247).
+//!
+//! The HTTP `WS /api/attach/:id` handler uses this to ask the registry to
+//! resolve a peer-owned session into an [`AttachCandidates`] (a pair-scoped
+//! JWT + the target daemon's T2 perimeter info). The actual control-link
+//! send and the reply correlation live here so the handler stays thin.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use maestro_protocol::{AttachCandidates, DaemonOutbound, ReplyTo, RpcResult};
+use tokio::sync::{mpsc, oneshot, Mutex};
+use uuid::Uuid;
+
+/// How long to wait for a `ResolveAttachResult` after the request goes out.
+/// Matches the registry's RPC timeout — a healthy registry replies in ms.
+pub const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[derive(Debug, thiserror::Error)]
+pub enum BrokerError {
+    #[error("registry control link unavailable")]
+    NoControlLink,
+    #[error("registry reply channel closed before response")]
+    ReplyClosed,
+    #[error("registry did not reply within {0:?}")]
+    Timeout(Duration),
+    #[error("registry rejected resolve: {0}: {1}")]
+    Registry(String, String),
+}
+
+/// Shared between the registry-client task (producer of `ResolveAttachResult`)
+/// and the broker HTTP handler (consumer). Clone freely — all fields are
+/// behind `Arc`s.
+#[derive(Debug, Clone)]
+pub struct BrokerHandle {
+    out_tx: mpsc::Sender<DaemonOutbound>,
+    pending: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<AttachCandidates>>>>>,
+}
+
+impl BrokerHandle {
+    /// `(handle, receiver)` — the registry-client task owns the receiver and
+    /// drains it into the control-link WS sink. The handle is shared with
+    /// HTTP handlers and is what they use to issue requests.
+    pub fn new() -> (Self, mpsc::Receiver<DaemonOutbound>) {
+        let (out_tx, out_rx) = mpsc::channel(64);
+        let handle = Self {
+            out_tx,
+            pending: Arc::new(Mutex::new(HashMap::new())),
+        };
+        (handle, out_rx)
+    }
+
+    /// Resolve a peer session: ask the registry for an [`AttachCandidates`].
+    pub async fn resolve_attach(&self, session_id: Uuid) -> Result<AttachCandidates, BrokerError> {
+        let reply_to = Uuid::new_v4();
+        let (tx, rx) = oneshot::channel();
+        self.pending.lock().await.insert(reply_to, tx);
+
+        if self
+            .out_tx
+            .send(DaemonOutbound::ResolveAttach {
+                reply_to,
+                session_id,
+            })
+            .await
+            .is_err()
+        {
+            self.pending.lock().await.remove(&reply_to);
+            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(_) => {
+                self.pending.lock().await.remove(&reply_to);
+                Err(BrokerError::Timeout(RESOLVE_TIMEOUT))
+            }
+        }
+    }
+
+    /// Called by the registry-client task when a `ResolveAttachResult` frame
+    /// comes back from the registry. Resolves the waiting `resolve_attach`
+    /// future if one's still alive.
+    pub async fn complete_resolve(
+        &self,
+        reply_to: ReplyTo,
+        result: RpcResult<AttachCandidates>,
+    ) -> bool {
+        let mut pending = self.pending.lock().await;
+        match pending.remove(&reply_to) {
+            Some(tx) => tx.send(result).is_ok(),
+            None => {
+                tracing::debug!(?reply_to, "no waiter for ResolveAttachResult");
+                false
+            }
+        }
+    }
+}

crates/maestro-daemon/src/jwt.rs

@@ -2,8 +2,11 @@
 //! validated locally by each daemon using the shared secret — no callback to
 //! the registry on the attach hot path. Spec §Auth.
 
+use std::collections::HashSet;
+
 use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
-use maestro_protocol::JwtClaims;
+use maestro_protocol::{JwtClaims, PairAttachClaims};
+use uuid::Uuid;
 
 #[derive(Debug, thiserror::Error)]
 pub enum JwtError {
@@ -11,6 +14,10 @@ pub enum JwtError {
     Decode(#[from] jsonwebtoken::errors::Error),
     #[error("daemon_id mismatch (token wants `{wanted}`, daemon is `{ours}`)")]
     WrongDaemon { wanted: String, ours: String },
+    #[error("session mismatch (token wants `{wanted}`, path is `{path}`)")]
+    WrongSession { wanted: Uuid, path: Uuid },
+    #[error("home daemon `{sub}` not in this daemon's allow_inbound")]
+    NotAllowed { sub: String },
 }
 
 #[derive(Clone)]
@@ -59,6 +66,67 @@ pub fn issue(secret: &str, claims: &JwtClaims) -> Result<String, jsonwebtoken::e
     encode(&header, claims, &key)
 }
 
+/// Verifier for pair-scoped attach JWTs (NAP-247). Checks signature, `exp`,
+/// `aud == self.daemon_id`, and `sub in allow_inbound`. The `allow_inbound`
+/// list is read at daemon startup; hot reload is NAP-237.
+#[derive(Clone)]
+pub struct PairAttachVerifier {
+    decoding: DecodingKey,
+    daemon_id: String,
+    allow_inbound: HashSet<String>,
+}
+
+impl std::fmt::Debug for PairAttachVerifier {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("PairAttachVerifier")
+            .field("daemon_id", &self.daemon_id)
+            .field("allow_inbound", &self.allow_inbound)
+            .field("decoding", &"<redacted>")
+            .finish()
+    }
+}
+
+impl PairAttachVerifier {
+    pub fn new(
+        secret: &str,
+        daemon_id: impl Into<String>,
+        allow_inbound: impl IntoIterator<Item = String>,
+    ) -> Self {
+        Self {
+            decoding: DecodingKey::from_secret(secret.as_bytes()),
+            daemon_id: daemon_id.into(),
+            allow_inbound: allow_inbound.into_iter().collect(),
+        }
+    }
+
+    /// Validate the token for the path-bound `session_id`. On success returns
+    /// the home daemon's id (`sub`) for logging/tracing.
+    pub fn verify(&self, token: &str, session_id: Uuid) -> Result<String, JwtError> {
+        let mut validation = Validation::new(Algorithm::HS256);
+        // Require explicit audience match — the jsonwebtoken crate validates
+        // `aud` against this set on decode.
+        validation.set_audience(&[self.daemon_id.as_str()]);
+        let data = decode::<PairAttachClaims>(token, &self.decoding, &validation)?;
+        let claims = data.claims;
+        if claims.aud != self.daemon_id {
+            return Err(JwtError::WrongDaemon {
+                wanted: claims.aud,
+                ours: self.daemon_id.clone(),
+            });
+        }
+        if claims.sid != session_id {
+            return Err(JwtError::WrongSession {
+                wanted: claims.sid,
+                path: session_id,
+            });
+        }
+        if !self.allow_inbound.contains(&claims.sub) {
+            return Err(JwtError::NotAllowed { sub: claims.sub });
+        }
+        Ok(claims.sub)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -110,4 +178,87 @@ mod tests {
         let token = issue("wrong", &claims_for("laptop", 300)).unwrap();
         assert!(v.verify(&token).is_err());
     }
+
+    // ----- PairAttachVerifier -----
+
+    fn pair_claims(home: &str, target: &str, sid: Uuid, valid_seconds: i64) -> PairAttachClaims {
+        PairAttachClaims {
+            iss: "registry".into(),
+            sub: home.into(),
+            aud: target.into(),
+            sid,
+            exp: (Utc::now() + Duration::seconds(valid_seconds)).timestamp(),
+        }
+    }
+
+    fn issue_pair(secret: &str, claims: &PairAttachClaims) -> String {
+        let header = Header::new(Algorithm::HS256);
+        let key = EncodingKey::from_secret(secret.as_bytes());
+        encode(&header, claims, &key).unwrap()
+    }
+
+    #[test]
+    fn pair_verifier_accepts_allow_listed_home() {
+        let secret = "s";
+        let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
+        let sid = Uuid::new_v4();
+        let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
+        let sub = v.verify(&token, sid).unwrap();
+        assert_eq!(sub, "laptop");
+    }
+
+    #[test]
+    fn pair_verifier_rejects_unlisted_home() {
+        let secret = "s";
+        let v = PairAttachVerifier::new(secret, "konishi", ["other".into()]);
+        let sid = Uuid::new_v4();
+        let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
+        let err = v.verify(&token, sid).unwrap_err();
+        assert!(matches!(err, JwtError::NotAllowed { .. }), "{err:?}");
+    }
+
+    #[test]
+    fn pair_verifier_rejects_empty_allow_inbound() {
+        // Default deny: no peer listed → every cross-daemon attach fails.
+        let secret = "s";
+        let v = PairAttachVerifier::new(secret, "konishi", std::iter::empty::<String>());
+        let sid = Uuid::new_v4();
+        let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
+        assert!(matches!(
+            v.verify(&token, sid).unwrap_err(),
+            JwtError::NotAllowed { .. }
+        ));
+    }
+
+    #[test]
+    fn pair_verifier_rejects_wrong_audience() {
+        let secret = "s";
+        let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
+        let sid = Uuid::new_v4();
+        // Token addressed to a different target — must not pass our checks.
+        let token = issue_pair(secret, &pair_claims("laptop", "elsewhere", sid, 60));
+        assert!(v.verify(&token, sid).is_err());
+    }
+
+    #[test]
+    fn pair_verifier_rejects_session_id_mismatch() {
+        let secret = "s";
+        let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
+        let sid = Uuid::new_v4();
+        let path = Uuid::new_v4();
+        let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
+        assert!(matches!(
+            v.verify(&token, path).unwrap_err(),
+            JwtError::WrongSession { .. }
+        ));
+    }
+
+    #[test]
+    fn pair_verifier_rejects_expired() {
+        let secret = "s";
+        let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
+        let sid = Uuid::new_v4();
+        let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, -120));
+        assert!(v.verify(&token, sid).is_err());
+    }
 }

crates/maestro-daemon/src/lib.rs

@@ -7,6 +7,7 @@
 #![deny(missing_debug_implementations)]
 
 pub mod attach;
+pub mod broker;
 pub mod catalog_mirror;
 pub mod config;
 pub mod configure;
@@ -19,9 +20,10 @@ pub mod sidecar;
 pub mod tmux;
 pub mod worktree;
 
+pub use broker::{BrokerError, BrokerHandle};
 pub use catalog_mirror::CatalogMirror;
 pub use config::{default_config_path, default_state_dir, load_config, DaemonConfig, EnvOverrides};
 pub use configure::{run as run_configure, ConfigureError, ConfigureInputs};
-pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier};
+pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier, PairAttachVerifier};
 pub use registry_client::{ClientError, RegistryClient, HEARTBEAT_INTERVAL};
 pub use session::{SessionError, SessionEvent, SessionManager};

crates/maestro-daemon/src/local_api.rs

@@ -179,33 +179,185 @@ pub fn state_for(
 // Broker WS — `WS /api/attach/:session_id`
 // ---------------------------------------------------------------------------
 //
-// v1 only handles the self case: if this daemon owns the session, delegate to
-// the existing PTY subscribe path. For peer-owned sessions we reply with a
-// structured 501 — NAP-247 (T2 broker) implements the cross-daemon path.
+// Home-daemon broker (NAP-243 + NAP-247):
+//
+//   1. Session owned by self           → delegate to local PTY subscribe.
+//   2. Session owned by a peer daemon  → ResolveAttach via the registry, dial
+//                                        the peer over T2, proxy frames.
+//   3. Unknown session                 → 404.
 
-pub fn broker_router(manager: SessionManager) -> Router {
+use crate::broker::{BrokerError, BrokerHandle};
+
+#[derive(Debug, Clone)]
+pub struct BrokerState {
+    pub manager: SessionManager,
+    pub catalog: CatalogMirror,
+    pub broker: BrokerHandle,
+    pub self_daemon_id: Arc<String>,
+    pub self_perimeter_id: Option<Arc<String>>,
+}
+
+pub fn broker_router(state: BrokerState) -> Router {
     Router::new()
         .route("/api/attach/:id", any(broker_attach_ws))
-        .with_state(manager)
+        .with_state(state)
 }
 
 async fn broker_attach_ws(
     Path(id): Path<Uuid>,
-    State(manager): State<SessionManager>,
+    State(state): State<BrokerState>,
     ws: axum::extract::WebSocketUpgrade,
 ) -> Result<Response, Response> {
-    if !manager.has(id).await {
-        // Peer-owned sessions land here once NAP-246 (peer catalog mirror)
-        // ships. Until then we can't even tell them apart from "doesn't exist"
-        // — both produce a 404. NAP-247 will distinguish the two.
+    if state.manager.has(id).await {
+        // Local short-circuit.
+        let manager = state.manager.clone();
+        return Ok(ws.on_upgrade(move |socket| async move {
+            let _ = crate::attach::serve_local(socket, manager, id).await;
+        }));
+    }
+
+    // Look up which peer owns the session via the catalog mirror.
+    let peer = state
+        .catalog
+        .sessions()
+        .await
+        .into_iter()
+        .find(|s| s.id == id)
+        .map(|s| s.daemon_id);
+    let Some(peer_daemon_id) = peer else {
         return Err(error_response(
             StatusCode::NOT_FOUND,
             "unknown_session",
-            "no such session on this daemon (peer broker is NAP-247)",
+            "no such session on this daemon or any known peer",
         ));
-    }
-    let upgrade_manager = manager.clone();
+    };
+
+    // ResolveAttach via the control link.
+    let candidates = match state.broker.resolve_attach(id).await {
+        Ok(c) => c,
+        Err(e) => {
+            tracing::warn!(%id, %peer_daemon_id, error = %e, "ResolveAttach failed");
+            let (code, msg) = broker_error_code(&e);
+            return Err(error_response(StatusCode::BAD_GATEWAY, code, &msg));
+        }
+    };
+
+    // Pick a transport. T1 is canceled (NAP-235); T3 is gated (NAP-238).
+    // For T2 we need both daemons to declare the same perimeter id, and the
+    // candidate must include a private address.
+    let perimeter = candidates.perimeter.clone();
+    let our_perimeter = state.self_perimeter_id.as_deref().map(|s| s.as_str());
+    let dial_url = match (perimeter, our_perimeter) {
+        (Some(p), Some(ours)) if p.perimeter_id == ours => {
+            format!(
+                "ws://{}/sessions/{}/stream?token={}",
+                p.private_addr,
+                id,
+                urlencoding::encode(&candidates.jwt),
+            )
+        }
+        (Some(_), Some(_)) => {
+            return Err(error_response(
+                StatusCode::BAD_GATEWAY,
+                "perimeter_mismatch",
+                "peer is on a different perimeter; T2 unavailable, T1/T3 not implemented",
+            ));
+        }
+        (Some(_), None) => {
+            return Err(error_response(
+                StatusCode::BAD_GATEWAY,
+                "no_local_perimeter",
+                "this daemon has no perimeter_id set; can't use T2",
+            ));
+        }
+        (None, _) => {
+            return Err(error_response(
+                StatusCode::BAD_GATEWAY,
+                "no_perimeter_candidate",
+                "peer hasn't advertised a perimeter; T2 unavailable",
+            ));
+        }
+    };
+
     Ok(ws.on_upgrade(move |socket| async move {
-        let _ = crate::attach::serve_local(socket, upgrade_manager, id).await;
+        let _ = pump_broker_ws(socket, dial_url).await;
     }))
 }
+
+/// Bidirectional pump between the client WS and the peer daemon's WS.
+async fn pump_broker_ws(
+    client: axum::extract::ws::WebSocket,
+    upstream_url: String,
+) -> Result<(), axum::Error> {
+    use axum::extract::ws::Message as AxMessage;
+    use futures_util::{SinkExt, StreamExt};
+    use tokio_tungstenite::tungstenite::Message as WsMessage;
+
+    let (upstream, _) = match tokio_tungstenite::connect_async(&upstream_url).await {
+        Ok(pair) => pair,
+        Err(e) => {
+            tracing::warn!(url = %upstream_url, error = %e, "T2 dial failed");
+            return Ok(());
+        }
+    };
+
+    let (mut client_sink, mut client_stream) = client.split();
+    let (mut up_sink, mut up_stream) = upstream.split();
+
+    let client_to_up = async {
+        while let Some(Ok(msg)) = client_stream.next().await {
+            let frame = match msg {
+                AxMessage::Text(t) => WsMessage::Text(t),
+                AxMessage::Binary(b) => WsMessage::Binary(b),
+                AxMessage::Ping(p) => WsMessage::Ping(p),
+                AxMessage::Pong(p) => WsMessage::Pong(p),
+                AxMessage::Close(_) => break,
+            };
+            if up_sink.send(frame).await.is_err() {
+                break;
+            }
+        }
+    };
+    let up_to_client = async {
+        while let Some(Ok(msg)) = up_stream.next().await {
+            let frame = match msg {
+                WsMessage::Text(t) => AxMessage::Text(t),
+                WsMessage::Binary(b) => AxMessage::Binary(b),
+                WsMessage::Ping(p) => AxMessage::Ping(p),
+                WsMessage::Pong(p) => AxMessage::Pong(p),
+                WsMessage::Close(_) | WsMessage::Frame(_) => break,
+            };
+            if client_sink.send(frame).await.is_err() {
+                break;
+            }
+        }
+    };
+
+    tokio::select! {
+        _ = client_to_up => {}
+        _ = up_to_client => {}
+    }
+    Ok(())
+}
+
+fn broker_error_code(e: &BrokerError) -> (&'static str, String) {
+    match e {
+        BrokerError::NoControlLink => (
+            "no_registry_link",
+            "this daemon is not connected to the registry".into(),
+        ),
+        BrokerError::ReplyClosed => ("reply_closed", e.to_string()),
+        BrokerError::Timeout(_) => ("resolve_timeout", e.to_string()),
+        BrokerError::Registry(code, msg) => (
+            // Code points are stable across the protocol so the client can
+            // distinguish "unknown_session" from "self_target" etc.
+            match code.as_str() {
+                "unknown_session" => "unknown_session",
+                "self_target" => "self_target",
+                "jwt_failed" => "jwt_failed",
+                _ => "registry_error",
+            },
+            msg.clone(),
+        ),
+    }
+}

crates/maestro-daemon/src/main.rs

@@ -15,7 +15,7 @@ use maestro_daemon::tmux::Tmux;
 use maestro_daemon::worktree::WorktreeManager;
 use maestro_daemon::{
     attach, default_config_path, default_state_dir, load_config, local_api, run_configure,
-    ConfigureInputs, EnvOverrides, JwtVerifier, RegistryClient, SessionManager,
+    ConfigureInputs, EnvOverrides, JwtVerifier, PairAttachVerifier, RegistryClient, SessionManager,
 };
 use tokio::net::TcpListener;
 use tracing_subscriber::EnvFilter;
@@ -133,6 +133,11 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
     manager.start_liveness_watcher(DEFAULT_LIVENESS_INTERVAL);
 
     let verifier = JwtVerifier::new(&config.token, &config.daemon_id);
+    let pair_verifier = PairAttachVerifier::new(
+        &config.token,
+        &config.daemon_id,
+        config.allow_inbound.iter().cloned(),
+    );
 
     // Build the registry client first so we can hand its catalog mirror to
     // the local-API state — both halves share the same Arc<Mutex<…>>.
@@ -151,11 +156,18 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
         manager.clone(),
         config.hostname.clone(),
         config.daemon_url.clone(),
-        catalog,
+        catalog.clone(),
     );
-    let router = attach::router(manager.clone(), verifier)
+    let broker_state = local_api::BrokerState {
+        manager: manager.clone(),
+        catalog,
+        broker: client.broker(),
+        self_daemon_id: std::sync::Arc::new(config.daemon_id.clone()),
+        self_perimeter_id: config.perimeter_id.clone().map(std::sync::Arc::new),
+    };
+    let router = attach::router(manager.clone(), verifier, pair_verifier)
         .merge(local_api::router(local_state))
-        .merge(local_api::broker_router(manager.clone()));
+        .merge(local_api::broker_router(broker_state));
 
     let listener = TcpListener::bind(config.bind_addr).await?;
     let bound = listener.local_addr()?;

crates/maestro-daemon/src/registry_client.rs

@@ -17,6 +17,9 @@ use maestro_protocol::{DaemonInbound, DaemonOutbound, RpcResult, PROTOCOL_VERSIO
 use tokio::sync::broadcast::error::RecvError;
 use tokio_tungstenite::tungstenite::Message;
 
+use tokio::sync::mpsc;
+
+use crate::broker::BrokerHandle;
 use crate::catalog_mirror::CatalogMirror;
 use crate::session::{SessionEvent, SessionManager};
 
@@ -37,7 +40,7 @@ pub enum ClientError {
     Handshake(String),
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct RegistryClient {
     registry_url: String,
     daemon_id: String,
@@ -54,6 +57,13 @@ pub struct RegistryClient {
     /// populated from `DaemonInbound::CatalogEvent` frames the registry
     /// fans out.
     catalog: CatalogMirror,
+    /// Shared with HTTP handlers (NAP-247): when one calls `resolve_attach`,
+    /// we route the request through this client's control link and route the
+    /// reply back via the handle's pending map.
+    broker: BrokerHandle,
+    /// Outbound channel from `BrokerHandle::resolve_attach` (and future broker
+    /// requests). Drained inside `connect_once` and forwarded into the WS.
+    broker_rx: mpsc::Receiver<DaemonOutbound>,
 }
 
 impl RegistryClient {
@@ -65,6 +75,7 @@ impl RegistryClient {
         token: impl Into<String>,
         manager: SessionManager,
     ) -> Self {
+        let (broker, broker_rx) = BrokerHandle::new();
         Self {
             registry_url: registry_url.into(),
             daemon_id: daemon_id.into(),
@@ -75,9 +86,17 @@ impl RegistryClient {
             private_addr: None,
             manager,
             catalog: CatalogMirror::new(),
+            broker,
+            broker_rx,
         }
     }
 
+    /// Cheap clone of the broker handle for HTTP handlers (NAP-247). Sends
+    /// here are pumped through this client's control link.
+    pub fn broker(&self) -> BrokerHandle {
+        self.broker.clone()
+    }
+
     /// Cheap handle to the daemon's peer-catalog mirror. Clone freely;
     /// internal state is `Arc<Mutex<…>>`.
     pub fn catalog(&self) -> CatalogMirror {
@@ -98,7 +117,7 @@ impl RegistryClient {
 
     /// Run forever, reconnecting with exponential backoff between attempts.
     /// Returns only if cancelled by the caller dropping the task.
-    pub async fn run(self) {
+    pub async fn run(mut self) {
         let mut backoff = BACKOFF_MIN;
         loop {
             match self.connect_once().await {
@@ -115,7 +134,7 @@ impl RegistryClient {
         }
     }
 
-    async fn connect_once(&self) -> Result<(), ClientError> {
+    async fn connect_once(&mut self) -> Result<(), ClientError> {
         let (ws, _) = tokio_tungstenite::connect_async(&self.registry_url).await?;
         let (mut sink, mut stream) = ws.split();
 
@@ -157,9 +176,11 @@ impl RegistryClient {
             .await?;
 
         let mut events = self.manager.events();
+        // The first interval tick fires immediately, so the registry gets a
+        // perimeter advertisement (NAP-247) right after the handshake instead
+        // of waiting a full HEARTBEAT_INTERVAL for `update_perimeter` to land.
         let mut hb = tokio::time::interval(HEARTBEAT_INTERVAL);
         hb.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-        let _ = hb.tick().await; // consume the immediate first tick
 
         loop {
             tokio::select! {
@@ -203,6 +224,11 @@ impl RegistryClient {
                         },
                     )?)).await?;
                 }
+                Some(broker_msg) = self.broker_rx.recv() => {
+                    // Outbound broker requests (NAP-247) — e.g. ResolveAttach.
+                    sink.send(Message::Text(serde_json::to_string(&broker_msg)?))
+                        .await?;
+                }
             }
         }
         Ok(())
@@ -242,10 +268,13 @@ impl RegistryClient {
                 )?))
                 .await?;
             }
-            DaemonInbound::ResolveAttachResult { .. } | DaemonInbound::RelaySetup { .. } => {
-                // Wired by NAP-243 (broker mode). Lands here only so registries
-                // emitting these messages don't crash older daemons on parse.
-                tracing::debug!("broker message received but unhandled (NAP-243 pending)");
+            DaemonInbound::ResolveAttachResult { reply_to, result } => {
+                self.broker.complete_resolve(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.
+                tracing::debug!("RelaySetup received but T3 not implemented (NAP-238 pending)");
             }
             DaemonInbound::CatalogEvent { event } => {
                 // NAP-246: peer-filtered catalog delta from the registry.

crates/maestro-daemon/tests/attach.rs

@@ -12,7 +12,7 @@ use std::time::Duration;
 use chrono::{Duration as ChronoDuration, Utc};
 use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
-use maestro_daemon::{attach, issue_jwt, JwtVerifier, SessionManager};
+use maestro_daemon::{attach, issue_jwt, JwtVerifier, PairAttachVerifier, SessionManager};
 use maestro_protocol::{CreateSessionRequest, JwtClaims};
 use tempfile::TempDir;
 use tokio::net::TcpListener;
@@ -36,7 +36,8 @@ async fn spawn_daemon_ws(daemon_id: &str) -> Harness {
     let fx = DaemonFixture::new(daemon_id);
     let manager = fx.manager.clone();
     let verifier = JwtVerifier::new(TOKEN, daemon_id);
-    let router = attach::router(manager.clone(), verifier);
+    let pair_verifier = PairAttachVerifier::new(TOKEN, daemon_id, std::iter::empty::<String>());
+    let router = attach::router(manager.clone(), verifier, pair_verifier);
     let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
         .await
         .unwrap();

crates/maestro-daemon/tests/local_api.rs

@@ -11,7 +11,7 @@ use std::time::Duration;
 
 use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
-use maestro_daemon::{local_api, CatalogMirror};
+use maestro_daemon::{local_api, BrokerHandle, CatalogMirror};
 use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session};
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
@@ -34,7 +34,18 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
     let manager = fx.manager.clone();
     let catalog = CatalogMirror::new();
     let state = local_api::state_for(manager.clone(), hostname, daemon_url, catalog.clone());
-    let router = local_api::router(state).merge(local_api::broker_router(manager.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(),
+        broker,
+        self_daemon_id: std::sync::Arc::new(daemon_id.into()),
+        self_perimeter_id: None,
+    };
+    let router = local_api::router(state).merge(local_api::broker_router(broker_state));
     let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
         .await
         .unwrap();