From cc8412ca596fc9627cf3886e0f9b44d3e46df29c Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Sat, 20 Jun 2026 21:09:18 +0000 Subject: [PATCH] feat(maestro): spawn protocol types + relay + daemon HTTP for recents/gitroot/presets (#26) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose the daemon-side spawn data (#24 recents, #25 gitroot, #22 presets) to the web, mirroring the NAP-329 ListDir broker plumbing end to end so it works across federated hosts. Wire (maestro-protocol): `RecentDir`/`RecentDirs`/`GitRoot` payloads; `DaemonInbound::RecentDirs`/`GitRoot` requests + `ForwardRecentDirs`/ `ForwardGitRoot` relay pairs + their `*Result` replies on both enums. All forward-compatible via the existing `RpcResult` + `#[serde(other)] Unknown`. Relay (broker + registry + registry_client): home daemon → registry → owning daemon → back, with the same fail-fast-while-disconnected + RAII pending-entry cleanup discipline as ListDir. The owning daemon answers RecentDirs from its spawn history and GitRoot by resolving the path against its home dir (shared `dirlist::resolve`) then walking up via `gitroot::find_root`. Daemon HTTP (maestro-daemon): - GET /api/daemons/:id/recent-dirs → RecentDirs - GET /api/daemons/:id/gitroot?path= → GitRoot - Presets ride GET /api/web/info (the documented DaemonInfo sibling) so the form needs no extra round-trip; resolved once at startup from [spawn].presets (a bad preset list now refuses startup). Tests: protocol round-trip + forward-compat for the new variants; broker forward round-trips + fail-fast; HTTP handler tests (self recents/gitroot, presets advertisement, peer 503 without a registry link). Closes #26 Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/maestro-daemon/src/broker.rs | 222 ++++++++++++++++++- crates/maestro-daemon/src/cli.rs | 1 + crates/maestro-daemon/src/config.rs | 11 + crates/maestro-daemon/src/dirlist.rs | 6 +- crates/maestro-daemon/src/local_api.rs | 55 ++++- crates/maestro-daemon/src/registry_client.rs | 38 ++++ crates/maestro-daemon/src/spawn_history.rs | 17 ++ crates/maestro-daemon/tests/local_api.rs | 73 ++++++ crates/maestro-protocol/src/lib.rs | 187 ++++++++++++++++ crates/maestro-registry/src/handler.rs | 27 +++ crates/maestro-registry/src/state.rs | 144 +++++++++++- 11 files changed, 776 insertions(+), 5 deletions(-) diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs index 4551f95..1ae6d73 100644 --- a/crates/maestro-daemon/src/broker.rs +++ b/crates/maestro-daemon/src/broker.rs @@ -19,7 +19,8 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use maestro_protocol::{ - AttachCandidates, CreateSessionRequest, DaemonOutbound, DirListing, ReplyTo, RpcResult, Session, + AttachCandidates, CreateSessionRequest, DaemonOutbound, DirListing, GitRoot, RecentDirs, + ReplyTo, RpcResult, Session, }; use tokio::sync::{mpsc, oneshot}; use uuid::Uuid; @@ -73,6 +74,10 @@ pub struct BrokerHandle { pending_kill: Arc>>>>, /// Pending `ForwardListDir` replies, keyed by `reply_to` (NAP-329). pending_list_dir: Arc>>>>, + /// Pending `ForwardRecentDirs` replies, keyed by `reply_to` (#26). + pending_recent_dirs: Arc>>>>, + /// Pending `ForwardGitRoot` replies, keyed by `reply_to` (#26). + pending_git_root: 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 @@ -95,6 +100,8 @@ impl BrokerHandle { pending_create: Arc::new(Mutex::new(HashMap::new())), pending_kill: Arc::new(Mutex::new(HashMap::new())), pending_list_dir: Arc::new(Mutex::new(HashMap::new())), + pending_recent_dirs: Arc::new(Mutex::new(HashMap::new())), + pending_git_root: Arc::new(Mutex::new(HashMap::new())), connected: Arc::new(AtomicBool::new(false)), }; (handle, out_rx) @@ -292,6 +299,117 @@ impl BrokerHandle { } } + /// Broker a cross-daemon recent-dirs query: ask the registry to relay a + /// `RecentDirs` request to the daemon named by `daemon_id` (which may be a + /// peer) and return its recents. Same discipline as the other broker calls + /// (#26). + pub async fn forward_recent_dirs(&self, daemon_id: String) -> Result { + if !self.connected.load(Ordering::Acquire) { + return Err(BrokerError::NoControlLink); + } + let reply_to = Uuid::new_v4(); + let (tx, rx) = oneshot::channel(); + self.pending_recent_dirs + .lock() + .unwrap() + .insert(reply_to, tx); + let _guard = PendingGuard { + map: &*self.pending_recent_dirs, + key: reply_to, + }; + if self + .out_tx + .try_send(DaemonOutbound::ForwardRecentDirs { + reply_to, + daemon_id, + }) + .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)), + } + } + + /// Resolve a waiting [`forward_recent_dirs`](Self::forward_recent_dirs); + /// called by the registry-client task on a `ForwardRecentDirsResult` frame. + pub async fn complete_forward_recent_dirs( + &self, + reply_to: ReplyTo, + result: RpcResult, + ) -> bool { + let entry = self.pending_recent_dirs.lock().unwrap().remove(&reply_to); + match entry { + Some(tx) => tx.send(result).is_ok(), + None => { + tracing::debug!(?reply_to, "no waiter for ForwardRecentDirsResult"); + false + } + } + } + + /// Broker a cross-daemon gitroot probe: ask the registry to relay a + /// `GitRoot` request for `path` to the daemon named by `daemon_id` (which + /// may be a peer) and return its result (#26). + pub async fn forward_git_root( + &self, + daemon_id: String, + path: String, + ) -> Result { + if !self.connected.load(Ordering::Acquire) { + return Err(BrokerError::NoControlLink); + } + let reply_to = Uuid::new_v4(); + let (tx, rx) = oneshot::channel(); + self.pending_git_root.lock().unwrap().insert(reply_to, tx); + let _guard = PendingGuard { + map: &*self.pending_git_root, + key: reply_to, + }; + if self + .out_tx + .try_send(DaemonOutbound::ForwardGitRoot { + reply_to, + daemon_id, + path, + }) + .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)), + } + } + + /// Resolve a waiting [`forward_git_root`](Self::forward_git_root); called by + /// the registry-client task on a `ForwardGitRootResult` frame. + pub async fn complete_forward_git_root( + &self, + reply_to: ReplyTo, + result: RpcResult, + ) -> bool { + let entry = self.pending_git_root.lock().unwrap().remove(&reply_to); + match entry { + Some(tx) => tx.send(result).is_ok(), + None => { + tracing::debug!(?reply_to, "no waiter for ForwardGitRootResult"); + false + } + } + } + /// Resolve a waiting [`forward_create`](Self::forward_create); called by the /// registry-client task on a `ForwardCreateResult` frame. pub async fn complete_forward_create( @@ -573,6 +691,108 @@ mod tests { assert_eq!(handle.pending_list_dir.lock().unwrap().len(), 0); } + // ----- #26: forward_recent_dirs / forward_git_root ----- + + /// `forward_recent_dirs` issues a `ForwardRecentDirs` frame and resolves + /// with the recents once the registry's `ForwardRecentDirsResult` lands. + #[tokio::test] + async fn forward_recent_dirs_round_trips() { + use maestro_protocol::{RecentDir, RecentDirs}; + let (handle, mut out_rx) = BrokerHandle::new(); + handle.set_connected(true); + + let h = handle.clone(); + let task = tokio::spawn(async move { h.forward_recent_dirs("konishi".into()).await }); + + let reply_to = match out_rx.recv().await.expect("ForwardRecentDirs went out") { + DaemonOutbound::ForwardRecentDirs { + reply_to, + daemon_id, + } => { + assert_eq!(daemon_id, "konishi"); + reply_to + } + other => panic!("expected ForwardRecentDirs, got {other:?}"), + }; + handle + .complete_forward_recent_dirs( + reply_to, + RpcResult::Ok { + value: RecentDirs { + dirs: vec![RecentDir { + cwd: std::path::PathBuf::from("/work"), + last_command: "claude".into(), + last_used: chrono::Utc::now(), + }], + truncated: false, + }, + }, + ) + .await; + + let got = task.await.unwrap().expect("recents ok"); + assert_eq!(got.dirs.len(), 1); + assert_eq!(got.dirs[0].cwd, std::path::PathBuf::from("/work")); + } + + /// `forward_git_root` issues a `ForwardGitRoot` frame carrying the path and + /// resolves with the probe result. + #[tokio::test] + async fn forward_git_root_round_trips() { + use maestro_protocol::GitRoot; + let (handle, mut out_rx) = BrokerHandle::new(); + handle.set_connected(true); + + let h = handle.clone(); + let task = tokio::spawn(async move { + h.forward_git_root("konishi".into(), "/work/sub".into()) + .await + }); + + let reply_to = match out_rx.recv().await.expect("ForwardGitRoot went out") { + DaemonOutbound::ForwardGitRoot { + reply_to, + daemon_id, + path, + } => { + assert_eq!(daemon_id, "konishi"); + assert_eq!(path, "/work/sub"); + reply_to + } + other => panic!("expected ForwardGitRoot, got {other:?}"), + }; + handle + .complete_forward_git_root( + reply_to, + RpcResult::Ok { + value: GitRoot { + root: Some(std::path::PathBuf::from("/work")), + }, + }, + ) + .await; + + let got = task.await.unwrap().expect("gitroot ok"); + assert_eq!(got.root, Some(std::path::PathBuf::from("/work"))); + } + + /// While disconnected, both new broker calls fail fast and leave no pending + /// entry (NAP-263 / NAP-259). + #[tokio::test] + async fn forward_recents_and_gitroot_fail_fast_while_disconnected() { + let (handle, _out_rx) = BrokerHandle::new(); // starts disconnected + assert!(matches!( + handle.forward_recent_dirs("konishi".into()).await, + Err(BrokerError::NoControlLink) + )); + assert!(matches!( + handle.forward_git_root("konishi".into(), "/x".into()).await, + Err(BrokerError::NoControlLink) + )); + assert_eq!(handle.pending_recent_dirs.lock().unwrap().len(), 0); + assert_eq!(handle.pending_git_root.lock().unwrap().len(), 0); + } + /// 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 52adb32..eb62dfb 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.hostname.clone(), config.daemon_url.clone(), config.home_dir.clone(), + config.spawn_presets.clone(), catalog.clone(), events, client.broker(), diff --git a/crates/maestro-daemon/src/config.rs b/crates/maestro-daemon/src/config.rs index 6dcfece..e753e63 100644 --- a/crates/maestro-daemon/src/config.rs +++ b/crates/maestro-daemon/src/config.rs @@ -63,6 +63,8 @@ pub enum ConfigError { env: &'static str, path: PathBuf, }, + #[error("invalid [spawn].presets: {0}")] + SpawnPresets(#[from] maestro_config::SpawnConfigError), } /// Resolved, validated daemon configuration ready to drive the binary. @@ -101,6 +103,10 @@ pub struct DaemonConfig { /// form's "recent dirs" picker (#24). `0` disables history. Default /// [`DEFAULT_SPAWN_HISTORY_DAYS`]. pub spawn_history_days: u32, + /// Resolved spawn-form command presets (#22/#26): `[spawn].presets` (or the + /// built-ins) with the implicit `custom` tile appended. Advertised to the + /// web in `GET /api/web/info`. + pub spawn_presets: Vec, } impl DaemonConfig { @@ -195,6 +201,10 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result Result Result PathBuf { +/// +/// Shared with the gitroot probe (#26) so a `~`/relative/empty path the client +/// sends is resolved against the daemon's `$HOME` the same way before walking +/// up for the enclosing repo. +pub fn resolve(home_dir: &str, requested: &str) -> PathBuf { let home = PathBuf::from(home_dir); let trimmed = requested.trim(); if trimmed.is_empty() { diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index a4dd1c2..221528d 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -30,8 +30,10 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::{any, delete, get}; use axum::{Json, Router}; +use maestro_config::SpawnPreset; use maestro_protocol::{ - CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, Session, PROTOCOL_VERSION, + CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, GitRoot, RecentDirs, Session, + PROTOCOL_VERSION, }; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -42,13 +44,18 @@ use crate::session::SessionManager; /// Daemon identity returned by `GET /api/web/info`. Browsers render this in the /// header ("you're connected to konishi"); the field set mirrors `DaemonInfo` -/// plus the running daemon's protocol version. +/// plus the running daemon's protocol version and its spawn presets (#26). #[derive(Debug, Clone, Serialize)] pub struct WebInfo { pub daemon_id: String, pub hostname: String, pub version: String, pub daemon_url: String, + /// Spawn-form command presets resolved from `[spawn].presets` (#22), with + /// the implicit `custom` tile appended. The web renders these as one-click + /// tiles; it rides this existing payload so opening the form needs no extra + /// round-trip (#26). + pub presets: Vec, } #[derive(Debug, Clone)] @@ -59,6 +66,8 @@ pub struct LocalApiState { /// Home directory of the OS user this daemon runs as, advertised so clients /// can prefill a spawn form's working directory. Empty if unresolved. pub home_dir: Arc, + /// Resolved spawn presets (#22/#26) advertised in `GET /api/web/info`. + pub presets: Arc>, /// Peer-only mirror of the global catalog. Empty until the daemon's /// registry client connects and receives the snapshot. pub catalog: CatalogMirror, @@ -76,6 +85,8 @@ pub fn router(state: LocalApiState) -> Router { .route("/api/sessions/:id", delete(kill_session)) .route("/api/daemons", get(list_daemons)) .route("/api/daemons/:id/dirs", get(list_dir)) + .route("/api/daemons/:id/recent-dirs", get(recent_dirs)) + .route("/api/daemons/:id/gitroot", get(git_root)) .route("/api/web/info", get(web_info)) .route("/api/events", any(events_ws)) .with_state(state) @@ -140,12 +151,49 @@ fn dir_error_status(code: &str) -> StatusCode { } } +/// Recent working dirs on `daemon_id`'s host for the spawn form's recents +/// picker (#26). Reads locally when the target is this daemon; otherwise brokers +/// a `RecentDirs` query to the owning daemon over the registry control link. +async fn recent_dirs(State(s): State, Path(daemon_id): Path) -> Response { + if daemon_id == s.manager.daemon_id() { + let value = crate::spawn_history::to_wire(s.manager.recent_dirs()); + return Json(value).into_response(); + } + match s.broker.forward_recent_dirs(daemon_id).await { + Ok(recents) => Json::(recents).into_response(), + Err(e) => broker_error_response(&e), + } +} + +/// Probe the enclosing git repo of `path` on `daemon_id`'s host, to infer the +/// worktree `base_repo` for the spawn form (#26). Local when the target is this +/// daemon; otherwise brokered to the owning daemon. `path` is resolved against +/// the target daemon's home dir (empty/`~`/relative), like [`list_dir`]. +async fn git_root( + State(s): State, + Path(daemon_id): Path, + Query(q): Query, +) -> Response { + if daemon_id == s.manager.daemon_id() { + let dir = crate::dirlist::resolve(&s.home_dir, &q.path); + let value = GitRoot { + root: crate::gitroot::find_root(&dir), + }; + return Json(value).into_response(); + } + match s.broker.forward_git_root(daemon_id, q.path).await { + Ok(root) => Json::(root).into_response(), + Err(e) => broker_error_response(&e), + } +} + async fn web_info(State(s): State) -> Json { Json(WebInfo { daemon_id: s.manager.daemon_id().to_string(), hostname: (*s.hostname).clone(), version: PROTOCOL_VERSION.into(), daemon_url: (*s.daemon_url).clone(), + presets: (*s.presets).clone(), }) } @@ -249,11 +297,13 @@ fn error_response(status: StatusCode, code: &str, message: &str) -> Response { /// Convenience constructor — pulls the bits `LocalApiState` needs from /// `DaemonConfig` so callers don't have to assemble it by hand. +#[allow(clippy::too_many_arguments)] pub fn state_for( manager: SessionManager, hostname: impl Into, daemon_url: impl Into, home_dir: impl Into, + presets: Vec, catalog: CatalogMirror, events: LocalEventBus, broker: BrokerHandle, @@ -263,6 +313,7 @@ pub fn state_for( hostname: Arc::new(hostname.into()), daemon_url: Arc::new(daemon_url.into()), home_dir: Arc::new(home_dir.into()), + presets: Arc::new(presets), catalog, events, broker, diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index f3eb98f..f485ae4 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -379,6 +379,44 @@ impl RegistryClient { .complete_forward_list_dir(reply_to, result) .await; } + DaemonInbound::RecentDirs { reply_to } => { + // Answer a peer's recent-dirs query from this host's spawn + // history (#26). Daemon-wide, so no path to resolve. + let value = crate::spawn_history::to_wire(self.manager.recent_dirs()); + sink.send(Message::Text(serde_json::to_string( + &DaemonOutbound::RecentDirsResult { + reply_to, + result: RpcResult::Ok { value }, + }, + )?)) + .await?; + } + DaemonInbound::ForwardRecentDirsResult { reply_to, result } => { + self.broker + .complete_forward_recent_dirs(reply_to, result) + .await; + } + DaemonInbound::GitRoot { reply_to, path } => { + // Probe the enclosing repo of `path` on this host (#26), + // resolving the abbreviated path against this daemon's home the + // same way `ListDir` does. + let dir = crate::dirlist::resolve(&self.home_dir, &path); + let value = maestro_protocol::GitRoot { + root: crate::gitroot::find_root(&dir), + }; + sink.send(Message::Text(serde_json::to_string( + &DaemonOutbound::GitRootResult { + reply_to, + result: RpcResult::Ok { value }, + }, + )?)) + .await?; + } + DaemonInbound::ForwardGitRootResult { reply_to, result } => { + self.broker + .complete_forward_git_root(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/src/spawn_history.rs b/crates/maestro-daemon/src/spawn_history.rs index 5711939..668bca4 100644 --- a/crates/maestro-daemon/src/spawn_history.rs +++ b/crates/maestro-daemon/src/spawn_history.rs @@ -61,6 +61,23 @@ pub struct RecentDir { pub last_used: DateTime, } +/// Map daemon-local recents to the wire payload (#26). `recent_dirs` already +/// caps and dedups, so `truncated` is always false here — the field exists for +/// wire symmetry with [`maestro_protocol::DirListing`]. +pub fn to_wire(dirs: Vec) -> maestro_protocol::RecentDirs { + maestro_protocol::RecentDirs { + dirs: dirs + .into_iter() + .map(|d| maestro_protocol::RecentDir { + cwd: d.cwd, + last_command: d.last_command, + last_used: d.last_used, + }) + .collect(), + truncated: false, + } +} + /// Append-only JSONL spawn log rooted at a file path, with a recency cap. /// Cheap to clone — just the path + caps. #[derive(Debug, Clone)] diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 8913e1e..30e2eff 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -39,11 +39,15 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { // resolve_attach / forward_create / forward_kill fails fast with // NoControlLink (the handle starts disconnected). let (broker, _broker_rx) = BrokerHandle::new(); + // Built-in presets (claude/opencode/zsh + custom) so web_info advertises a + // known set (#26). + let presets = maestro_config::SpawnSection::default().resolve().unwrap(); let state = local_api::state_for( manager.clone(), hostname, daemon_url, "/home/test", + presets, catalog.clone(), events, broker.clone(), @@ -239,6 +243,75 @@ async fn post_to_peer_daemon_id_brokers_and_503s_without_registry_link() { assert_eq!(body["error"]["code"], "no_registry_link"); } +#[tokio::test] +async fn web_info_advertises_spawn_presets() { + // #26: the connected daemon's presets ride the web_info payload so the + // spawn form needs no extra round-trip. Built-ins + the implicit custom. + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/web/info")) + .await + .unwrap() + .json() + .await + .unwrap(); + let ids: Vec<&str> = resp["presets"] + .as_array() + .unwrap() + .iter() + .map(|p| p["id"].as_str().unwrap()) + .collect(); + assert_eq!(ids, ["claude", "opencode", "zsh", "custom"]); +} + +#[tokio::test] +async fn recent_dirs_self_returns_empty_without_history() { + // #26: the self-path reads this daemon's spawn history. The test fixture + // wires no history, so the list is empty but the endpoint is well-formed. + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/daemons/laptop/recent-dirs")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(resp["dirs"].as_array().unwrap().len(), 0); + assert_eq!(resp["truncated"], false); +} + +#[tokio::test] +async fn gitroot_self_probes_path() { + // #26: the self-path resolves the query against the daemon home dir and + // walks up for a `.git`. A `/tmp`-rooted temp dir is not in a repo → null. + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let tmp = tempfile::tempdir().unwrap(); + let resp: serde_json::Value = reqwest::get(url( + h.addr, + &format!( + "/api/daemons/laptop/gitroot?path={}", + urlencoding::encode(&tmp.path().to_string_lossy()) + ), + )) + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(resp.get("root").is_none() || resp["root"].is_null()); +} + +#[tokio::test] +async fn recent_dirs_for_peer_503s_without_registry_link() { + // #26: a peer target brokers over the (absent) registry link, failing fast + // with a retryable 503 — same shape as the cross-daemon create path. + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let resp = reqwest::get(url(h.addr, "/api/daemons/konishi/recent-dirs")) + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["error"]["code"], "no_registry_link"); +} + #[tokio::test] async fn delete_unknown_session_returns_404() { let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index a8aadc6..e58a4e4 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -85,6 +85,36 @@ pub struct DirListing { pub truncated: bool, } +/// One deduplicated recent working directory for the spawn form's "recent +/// dirs" picker (#24/#26). Newest use of a `cwd`, with the command last run +/// there. Mirrors the daemon-local `spawn_history::RecentDir`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecentDir { + pub cwd: PathBuf, + pub last_command: String, + pub last_used: DateTime, +} + +/// Result of a `RecentDirs` query: a daemon's recently-used working dirs, +/// newest-first and deduped by `cwd` (#26). `truncated` mirrors [`DirListing`]: +/// true when the daemon capped the list. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecentDirs { + pub dirs: Vec, + #[serde(default)] + pub truncated: bool, +} + +/// Result of a `GitRoot` probe: the enclosing git repository toplevel of a +/// path on a daemon's host, or `None` when the path isn't inside a repo (#26). +/// Drives the spawn form's worktree section — a present `root` is the inferred +/// `base_repo`; `None` disables the worktree toggle. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct GitRoot { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub root: Option, +} + /// Structured failure used everywhere we don't want to leak a string-only error. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ErrorInfo { @@ -192,6 +222,34 @@ pub enum DaemonOutbound { daemon_id: String, path: String, }, + /// Owning daemon's reply to a [`DaemonInbound::RecentDirs`] — its recent + /// working dirs. (#26) + RecentDirsResult { + reply_to: ReplyTo, + result: RpcResult, + }, + /// Home daemon asks the registry to relay a recent-dirs query to the daemon + /// named by `daemon_id` (which may be a peer). Recents are daemon-wide so + /// there's no path argument. Reply: + /// [`DaemonInbound::ForwardRecentDirsResult`]. (#26) + ForwardRecentDirs { + reply_to: ReplyTo, + daemon_id: String, + }, + /// Owning daemon's reply to a [`DaemonInbound::GitRoot`] — the enclosing + /// repo toplevel of the requested path on its host. (#26) + GitRootResult { + reply_to: ReplyTo, + result: RpcResult, + }, + /// Home daemon asks the registry to relay a gitroot probe to the daemon + /// named by `daemon_id` (which may be a peer). Reply: + /// [`DaemonInbound::ForwardGitRootResult`]. (#26) + ForwardGitRoot { + reply_to: ReplyTo, + daemon_id: String, + path: String, + }, /// Target daemon answers a [`DaemonInbound::RelaySetup`]: whether it will /// accept relay traffic for the requested session. Registry only allocates /// relay capacity when `accepted == true`. @@ -251,6 +309,31 @@ pub enum DaemonInbound { reply_to: ReplyTo, result: RpcResult, }, + /// Registry asks a daemon for its recent working dirs (#26). Daemon-wide, + /// so no arguments beyond the correlation token; the daemon replies with + /// [`DaemonOutbound::RecentDirsResult`]. + RecentDirs { + reply_to: ReplyTo, + }, + /// Registry's reply to [`DaemonOutbound::ForwardRecentDirs`] — the owning + /// daemon's recents, relayed back to the home daemon. (#26) + ForwardRecentDirsResult { + reply_to: ReplyTo, + result: RpcResult, + }, + /// Registry asks a daemon to probe the enclosing git repo of `path` on its + /// host (#26). The daemon expands `~`/relative/empty against its own home + /// dir, then replies with [`DaemonOutbound::GitRootResult`]. + GitRoot { + reply_to: ReplyTo, + path: String, + }, + /// Registry's reply to [`DaemonOutbound::ForwardGitRoot`] — the owning + /// daemon's gitroot probe, relayed back to the home daemon. (#26) + ForwardGitRootResult { + reply_to: ReplyTo, + result: RpcResult, + }, /// Registry signals a target daemon that a home daemon wants to attach via /// the registry-mediated relay (T3). Target decides whether to accept the /// relay and replies with [`DaemonOutbound::RelayAck`]. @@ -1002,6 +1085,110 @@ mod tests { assert_eq!(res, round_trip(&res)); } + #[test] + fn daemon_recent_dirs_round_trips_and_tag() { + let req = DaemonInbound::RecentDirs { + reply_to: Uuid::from_u128(71), + }; + let v = serde_json::to_value(&req).unwrap(); + assert_eq!(v["type"], "RecentDirs"); + assert_eq!(req, round_trip(&req)); + + let res = DaemonOutbound::RecentDirsResult { + reply_to: Uuid::from_u128(71), + result: RpcResult::Ok { + value: RecentDirs { + dirs: vec![RecentDir { + cwd: PathBuf::from("/home/naps62/git/maestro"), + last_command: "claude".into(), + last_used: DateTime::parse_from_rfc3339("2026-06-20T10:00:00Z") + .unwrap() + .with_timezone(&Utc), + }], + truncated: false, + }, + }, + }; + let v = serde_json::to_value(&res).unwrap(); + assert_eq!(v["type"], "RecentDirsResult"); + assert_eq!(res, round_trip(&res)); + + let fwd = DaemonOutbound::ForwardRecentDirs { + reply_to: Uuid::from_u128(72), + daemon_id: "konishi".into(), + }; + let v = serde_json::to_value(&fwd).unwrap(); + assert_eq!(v["type"], "ForwardRecentDirs"); + assert_eq!(v["daemon_id"], "konishi"); + assert_eq!(fwd, round_trip(&fwd)); + + let fwd_res = DaemonInbound::ForwardRecentDirsResult { + reply_to: Uuid::from_u128(72), + result: RpcResult::Ok { + value: RecentDirs { + dirs: vec![], + truncated: false, + }, + }, + }; + let v = serde_json::to_value(&fwd_res).unwrap(); + assert_eq!(v["type"], "ForwardRecentDirsResult"); + assert_eq!(fwd_res, round_trip(&fwd_res)); + } + + #[test] + fn recent_dirs_truncated_defaults_when_absent() { + // Older daemons omit `truncated`; it must default to false. + let json = serde_json::json!({ "dirs": [] }); + let parsed: RecentDirs = serde_json::from_value(json).unwrap(); + assert!(!parsed.truncated); + } + + #[test] + fn daemon_git_root_round_trips_and_tag() { + let req = DaemonInbound::GitRoot { + reply_to: Uuid::from_u128(81), + path: "~/git/maestro/crates".into(), + }; + let v = serde_json::to_value(&req).unwrap(); + assert_eq!(v["type"], "GitRoot"); + assert_eq!(v["path"], "~/git/maestro/crates"); + assert_eq!(req, round_trip(&req)); + + let res = DaemonOutbound::GitRootResult { + reply_to: Uuid::from_u128(81), + result: RpcResult::Ok { + value: GitRoot { + root: Some(PathBuf::from("/home/naps62/git/maestro")), + }, + }, + }; + let v = serde_json::to_value(&res).unwrap(); + assert_eq!(v["type"], "GitRootResult"); + assert_eq!(res, round_trip(&res)); + + let fwd = DaemonOutbound::ForwardGitRoot { + reply_to: Uuid::from_u128(82), + daemon_id: "konishi".into(), + path: "/work".into(), + }; + let v = serde_json::to_value(&fwd).unwrap(); + assert_eq!(v["type"], "ForwardGitRoot"); + assert_eq!(fwd, round_trip(&fwd)); + + // A `None` root (path not inside a repo) drops the field from the wire. + let none = DaemonInbound::ForwardGitRootResult { + reply_to: Uuid::from_u128(82), + result: RpcResult::Ok { + value: GitRoot { root: None }, + }, + }; + let v = serde_json::to_value(&none).unwrap(); + assert_eq!(v["type"], "ForwardGitRootResult"); + assert_eq!(v["result"]["value"], serde_json::json!({})); + assert_eq!(none, round_trip(&none)); + } + #[test] fn daemon_outbound_relay_ack_round_trip_and_tag() { let msg = DaemonOutbound::RelayAck { diff --git a/crates/maestro-registry/src/handler.rs b/crates/maestro-registry/src/handler.rs index 35fb525..580b0c5 100644 --- a/crates/maestro-registry/src/handler.rs +++ b/crates/maestro-registry/src/handler.rs @@ -241,6 +241,33 @@ async fn dispatch_text( .forward_list_dir(daemon_id, reply_to, &target, path) .await; } + DaemonOutbound::RecentDirsResult { reply_to, result } => { + registry.complete_recent_dirs(reply_to, result).await; + } + DaemonOutbound::ForwardRecentDirs { + reply_to, + daemon_id: target, + } => { + // Relay a recent-dirs query to its owning daemon and route the reply + // back to this (home) daemon — control-plane only (#26). + registry + .forward_recent_dirs(daemon_id, reply_to, &target) + .await; + } + DaemonOutbound::GitRootResult { reply_to, result } => { + registry.complete_git_root(reply_to, result).await; + } + DaemonOutbound::ForwardGitRoot { + reply_to, + daemon_id: target, + path, + } => { + // Relay a gitroot probe to its owning daemon and route the reply + // back to this (home) daemon — control-plane only (#26). + registry + .forward_git_root(daemon_id, reply_to, &target, path) + .await; + } DaemonOutbound::RelayAck { reply_to, .. } => { // T3 (relay) is gated on NAP-238. Land the variant so newer daemons // don't crash the connection; the registry doesn't act on it yet. diff --git a/crates/maestro-registry/src/state.rs b/crates/maestro-registry/src/state.rs index 55983c4..447e357 100644 --- a/crates/maestro-registry/src/state.rs +++ b/crates/maestro-registry/src/state.rs @@ -16,7 +16,7 @@ use std::time::Duration; use maestro_protocol::{ AttachCandidates, CreateSessionRequest, DaemonInbound, DaemonInfo, DirListing, ErrorInfo, - PerimeterCandidate, RegistryEvent, ReplyTo, RpcResult, Session, + GitRoot, PerimeterCandidate, RecentDirs, RegistryEvent, ReplyTo, RpcResult, Session, }; use tokio::sync::{broadcast, mpsc, oneshot, Mutex}; use uuid::Uuid; @@ -113,6 +113,10 @@ pub struct Registry { /// Outstanding `ListDir` RPC waiters (NAP-329). Same `PendingGuard` /// discipline as `pending_create` / `pending_kill`. pending_list_dir: Arc>>>>, + /// Outstanding `RecentDirs` RPC waiters (#26). + pending_recent_dirs: Arc>>>>, + /// Outstanding `GitRoot` RPC waiters (#26). + pending_git_root: Arc>>>>, } #[derive(Debug, Default)] @@ -161,6 +165,8 @@ impl Registry { pending_create: Arc::new(StdMutex::new(HashMap::new())), pending_kill: Arc::new(StdMutex::new(HashMap::new())), pending_list_dir: Arc::new(StdMutex::new(HashMap::new())), + pending_recent_dirs: Arc::new(StdMutex::new(HashMap::new())), + pending_git_root: Arc::new(StdMutex::new(HashMap::new())), } } @@ -517,6 +523,51 @@ impl Registry { .await; } + /// Relay a home daemon's `ForwardRecentDirs` to the target daemon and push + /// the `ForwardRecentDirsResult` back to the home daemon (#26). + pub async fn forward_recent_dirs( + &self, + home_daemon_id: &str, + reply_to: ReplyTo, + target_daemon_id: &str, + ) { + let result = match self.recent_dirs(target_daemon_id).await { + Ok(Ok(recents)) => RpcResult::Ok { value: recents }, + Ok(Err(RpcError(error))) => RpcResult::Err { error }, + Err(e) => RpcResult::Err { + error: registry_error_info(&e), + }, + }; + self.reply_to_daemon( + home_daemon_id, + DaemonInbound::ForwardRecentDirsResult { reply_to, result }, + ) + .await; + } + + /// Relay a home daemon's `ForwardGitRoot` to the target daemon and push the + /// `ForwardGitRootResult` back to the home daemon (#26). + pub async fn forward_git_root( + &self, + home_daemon_id: &str, + reply_to: ReplyTo, + target_daemon_id: &str, + path: String, + ) { + let result = match self.git_root(target_daemon_id, path).await { + Ok(Ok(root)) => RpcResult::Ok { value: root }, + Ok(Err(RpcError(error))) => RpcResult::Err { error }, + Err(e) => RpcResult::Err { + error: registry_error_info(&e), + }, + }; + self.reply_to_daemon( + home_daemon_id, + DaemonInbound::ForwardGitRootResult { reply_to, result }, + ) + .await; + } + /// Send a single frame to a connected daemon's outbound channel, dropping /// it silently if the daemon has gone away mid-flight (the home daemon's /// own RPC timeout will surface the failure). @@ -561,6 +612,28 @@ impl Registry { } } + /// Resolve a RecentDirs RPC reply. Returns true if a waiter was found (#26). + pub async fn complete_recent_dirs( + &self, + reply_to: ReplyTo, + result: RpcResult, + ) -> bool { + let entry = self.pending_recent_dirs.lock().unwrap().remove(&reply_to); + match entry { + Some(tx) => tx.send(result).is_ok(), + None => false, + } + } + + /// Resolve a GitRoot RPC reply. Returns true if a waiter was found (#26). + pub async fn complete_git_root(&self, reply_to: ReplyTo, result: RpcResult) -> bool { + let entry = self.pending_git_root.lock().unwrap().remove(&reply_to); + match entry { + Some(tx) => tx.send(result).is_ok(), + None => false, + } + } + // ----------------------------------------------------------------------- // Outbound R→D requests (internal API for NAP-215) // ----------------------------------------------------------------------- @@ -688,6 +761,75 @@ impl Registry { await_rpc(rx).await.map(rpc_into_result) } + /// Forward a `RecentDirs` query to the named daemon and wait for its + /// `RecentDirsResult` (#26). No catalog side-effects. Mirrors + /// [`list_dir`](Self::list_dir). + pub async fn recent_dirs( + &self, + daemon_id: &str, + ) -> Result, RegistryError> { + let reply_to = Uuid::new_v4(); + let (tx, rx) = oneshot::channel(); + let outbound = { + let state = self.inner.lock().await; + state + .daemons + .get(daemon_id) + .ok_or_else(|| RegistryError::DaemonNotConnected(daemon_id.to_string()))? + .outbound + .clone() + }; + self.pending_recent_dirs + .lock() + .unwrap() + .insert(reply_to, tx); + let _guard = PendingGuard { + map: &*self.pending_recent_dirs, + key: reply_to, + }; + if outbound + .send(DaemonInbound::RecentDirs { reply_to }) + .await + .is_err() + { + return Err(RegistryError::DaemonGone); + } + await_rpc(rx).await.map(rpc_into_result) + } + + /// Forward a `GitRoot` probe to the named daemon and wait for its + /// `GitRootResult` (#26). Mirrors [`list_dir`](Self::list_dir). + pub async fn git_root( + &self, + daemon_id: &str, + path: String, + ) -> Result, RegistryError> { + let reply_to = Uuid::new_v4(); + let (tx, rx) = oneshot::channel(); + let outbound = { + let state = self.inner.lock().await; + state + .daemons + .get(daemon_id) + .ok_or_else(|| RegistryError::DaemonNotConnected(daemon_id.to_string()))? + .outbound + .clone() + }; + self.pending_git_root.lock().unwrap().insert(reply_to, tx); + let _guard = PendingGuard { + map: &*self.pending_git_root, + key: reply_to, + }; + if outbound + .send(DaemonInbound::GitRoot { reply_to, path }) + .await + .is_err() + { + return Err(RegistryError::DaemonGone); + } + await_rpc(rx).await.map(rpc_into_result) + } + // ----------------------------------------------------------------------- // Read-only views (mostly for tests) // -----------------------------------------------------------------------