test(maestro): e2e for daemon catalog mirror (NAP-246)

Spins up registry + alpha + beta, creates a session on beta, verifies
alpha's `/api/sessions` and `/api/daemons` reflect the peer state within
5 s of the create round-tripping.

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

crates/maestro-e2e/tests/e2e.rs

@@ -88,6 +88,80 @@ async fn lists_sessions_from_both_daemons() {
     client.kill_session(b.id).await.expect("kill b");
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn daemon_local_api_mirrors_peer_catalog() {
+    // NAP-246: a daemon's GET /api/daemons + /api/sessions should reflect
+    // peer state (not just self), because the registry fans out catalog
+    // events to every connected daemon.
+    let (_reg, daemons, client) = bring_up(&["alpha", "beta"]).await;
+    let cwd = TempDir::new().unwrap();
+
+    // Create a session on beta via the registry path.
+    let b = client
+        .spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
+        .await
+        .expect("spawn beta");
+
+    // Now query alpha's local HTTP API and confirm it sees beta + beta's
+    // session. The mirror is eventually consistent, so poll.
+    let alpha_addr = daemons[0].bind_addr;
+    let alpha_base = format!("http://{alpha_addr}");
+
+    wait_for_async(
+        "alpha mirrors beta in /api/daemons",
+        Duration::from_secs(5),
+        || {
+            let base = alpha_base.clone();
+            async move {
+                let resp = match reqwest::get(format!("{base}/api/daemons")).await {
+                    Ok(r) => r,
+                    Err(_) => return false,
+                };
+                let list: Vec<maestro_protocol::DaemonInfo> = match resp.json().await {
+                    Ok(v) => v,
+                    Err(_) => return false,
+                };
+                list.iter().any(|d| d.daemon_id == "beta")
+            }
+        },
+    )
+    .await;
+
+    wait_for_async(
+        "alpha mirrors beta's session in /api/sessions",
+        Duration::from_secs(5),
+        || {
+            let base = alpha_base.clone();
+            let want = b.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(|s| s.id == want && s.daemon_id == "beta")
+            }
+        },
+    )
+    .await;
+
+    // Both self and peer come back.
+    let list: Vec<maestro_protocol::DaemonInfo> = reqwest::get(format!("{alpha_base}/api/daemons"))
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let ids: Vec<&str> = list.iter().map(|d| d.daemon_id.as_str()).collect();
+    assert!(ids.contains(&"alpha"));
+    assert!(ids.contains(&"beta"));
+
+    client.kill_session(b.id).await.expect("kill b");
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn multi_attach_sees_same_output() {
     let (_reg, _daemons, client) = bring_up(&["solo"]).await;