feat(maestro): drop CLI registry fallback for cross-daemon create/kill + e2e (NAP-327)

The home daemon now brokers cross-daemon create/kill itself, so the CLI only
falls back to the legacy registry HTTP path when the home daemon is
*unreachable* — not on 501/404. Removes the now-dead `is_remote_op_unsupported`
/ `is_not_found` / `has_status` predicates.

Adds an e2e (real tmux, two daemons + registry): POST to alpha with
daemon_id=beta brokers the create to beta and returns beta's session; a kill
via alpha reaches beta. Both verified against the registry catalog.

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

crates/maestro-cli/src/dispatch.rs

@@ -15,10 +15,10 @@
 //!
 //! If the home daemon is **unreachable** (connection refused / not running),
 //! every command falls back to the legacy registry path (registry HTTP +
-//! direct WS) with a one-line stderr warning, so migration is painless. A few
-//! operations the daemon can't yet satisfy locally (cross-daemon spawn → 501,
-//! cross-daemon kill → 404) also fall back to the registry, which forwards
-//! them the old way. The fallback is removed in a future ticket.
+//! direct WS) with a one-line stderr warning, so migration is painless. As of
+//! NAP-327 the daemon brokers cross-daemon create *and* kill over the registry
+//! control link, so those no longer need a fallback — only an unreachable home
+//! daemon does.
 
 use std::path::{Path, PathBuf};
 
@@ -157,11 +157,12 @@ async fn spawn(
     // `spawn --attach` has no detach-key flag; use the attach default (Ctrl-q).
     let detach_key = 0x11;
 
-    // Cross-daemon create isn't brokered by the local daemon yet (501), so we
-    // fall back to the registry on that as well as on unreachable.
+    // The home daemon now brokers cross-daemon create over the registry
+    // control link (NAP-327), so the only reason to fall back to the legacy
+    // registry HTTP path is the home daemon being unreachable.
     let session = match run_spawn(&daemon, args.clone()).await {
         Ok(s) => s,
-        Err(e) if e.is_unreachable() || e.is_remote_op_unsupported() => {
+        Err(e) if e.is_unreachable() => {
             warn_fallback(&home_url, &e);
             let registry = registry_client(registry_url_override.clone(), &cfg, config_path)?;
             run_spawn(&registry, args).await?
@@ -216,11 +217,12 @@ async fn kill(
     let home_url = cfg.resolve_home_daemon_url(home_daemon_url_override);
     let daemon = daemon_client(&home_url, config_path);
 
-    // The local daemon only kills sessions it owns (peer sessions → 404); the
-    // registry forwards cross-daemon kills, so fall back on 404 too.
+    // The home daemon brokers cross-daemon kills too (NAP-327) and 404s only on
+    // a session no daemon knows about, so we no longer fall back on 404 — only
+    // when the home daemon itself is unreachable.
     match run_kill(&daemon, &args).await {
         Ok(()) => {}
-        Err(e) if e.is_unreachable() || e.is_not_found() => {
+        Err(e) if e.is_unreachable() => {
             warn_fallback(&home_url, &e);
             let registry = registry_client(registry_url_override, &cfg, config_path)?;
             run_kill(&registry, &args).await?;

crates/maestro-cli/src/http.rs

@@ -61,25 +61,6 @@ impl HttpError {
         matches!(self, HttpError::Reqwest(e) if e.is_connect())
     }
 
-    /// True when a *reachable* home daemon replied with the given HTTP status.
-    pub fn has_status(&self, want: StatusCode) -> bool {
-        matches!(self, HttpError::ServerError { status, .. } if *status == want)
-    }
-
-    /// The home daemon's local `POST /api/sessions` only creates sessions it
-    /// owns; cross-daemon creates return `501 remote_create_unsupported`
-    /// (broker create-forwarding is future work). `maestro spawn --daemon
-    /// <peer>` falls back to the registry on this so it keeps working.
-    pub fn is_remote_op_unsupported(&self) -> bool {
-        self.has_status(StatusCode::NOT_IMPLEMENTED)
-    }
-
-    /// The home daemon's local `DELETE /api/sessions/:id` only kills sessions
-    /// it owns; a peer-owned session yields `404 unknown_session`. `maestro
-    /// kill` falls back to the registry on this so cross-daemon kill works.
-    pub fn is_not_found(&self) -> bool {
-        self.has_status(StatusCode::NOT_FOUND)
-    }
 }
 
 /// HTTP client bound to a single registry. Persists the auth cookie back to

crates/maestro-e2e/tests/e2e.rs

@@ -162,6 +162,75 @@ async fn daemon_local_api_mirrors_peer_catalog() {
     client.kill_session(b.id).await.expect("kill b");
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn home_daemon_brokers_cross_daemon_create_and_kill() {
+    // NAP-327: POST /api/sessions to alpha (home) with daemon_id=beta must
+    // broker the create to beta over the registry control link and return
+    // beta's session — and a kill via alpha must reach beta too. This is the
+    // path maestro-web uses (it only talks to its local daemon).
+    let (_reg, daemons, registry_client) = bring_up(&["alpha", "beta"]).await;
+    let cwd = TempDir::new().unwrap();
+
+    // A client pointed at alpha's *local* API (not the registry).
+    let alpha_base = format!("http://{}", daemons[0].bind_addr);
+    let alpha = TestClient::new(&alpha_base, &alpha_base);
+
+    // Create a beta-owned session by asking alpha.
+    let s = alpha
+        .spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
+        .await
+        .expect("alpha brokers create to beta");
+    assert_eq!(s.daemon_id, "beta", "session should be owned by beta");
+
+    // The registry catalog (and thus every client) sees it on beta.
+    wait_for_async("beta session visible in catalog", Duration::from_secs(5), || {
+        let client = registry_client.clone();
+        let want = s.id;
+        async move {
+            client
+                .list_sessions()
+                .await
+                .map(|ss| ss.iter().any(|x| x.id == want && x.daemon_id == "beta"))
+                .unwrap_or(false)
+        }
+    })
+    .await;
+
+    // alpha's mirror must reflect beta's session before it can broker the kill.
+    wait_for_async("alpha mirrors beta's session", Duration::from_secs(5), || {
+        let base = alpha_base.clone();
+        let want = s.id;
+        async move {
+            let resp = match reqwest::get(format!("{base}/api/sessions")).await {
+                Ok(r) => r,
+                Err(_) => return false,
+            };
+            let list: Vec<maestro_protocol::Session> = match resp.json().await {
+                Ok(v) => v,
+                Err(_) => return false,
+            };
+            list.iter().any(|x| x.id == want)
+        }
+    })
+    .await;
+
+    // Kill the beta session via alpha — brokered to beta.
+    alpha.kill_session(s.id).await.expect("alpha brokers kill to beta");
+
+    wait_for_async("beta session gone from catalog", Duration::from_secs(5), || {
+        let client = registry_client.clone();
+        let gone = s.id;
+        async move {
+            client
+                .list_sessions()
+                .await
+                .map(|ss| !ss.iter().any(|x| x.id == gone))
+                .unwrap_or(false)
+        }
+    })
+    .await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn multi_attach_sees_same_output() {
     let (_reg, _daemons, client) = bring_up(&["solo"]).await;