diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs index 254ecbf..4551f95 100644 --- a/crates/maestro-daemon/src/broker.rs +++ b/crates/maestro-daemon/src/broker.rs @@ -19,7 +19,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use maestro_protocol::{ - AttachCandidates, CreateSessionRequest, DaemonOutbound, ReplyTo, RpcResult, Session, + AttachCandidates, CreateSessionRequest, DaemonOutbound, DirListing, ReplyTo, RpcResult, Session, }; use tokio::sync::{mpsc, oneshot}; use uuid::Uuid; @@ -71,6 +71,8 @@ pub struct BrokerHandle { pending_create: Arc>>>>, /// Pending `ForwardKill` replies, keyed by `reply_to` (NAP-327). pending_kill: Arc>>>>, + /// Pending `ForwardListDir` replies, keyed by `reply_to` (NAP-329). + pending_list_dir: 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 @@ -92,6 +94,7 @@ impl BrokerHandle { pending: Arc::new(Mutex::new(HashMap::new())), 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())), connected: Arc::new(AtomicBool::new(false)), }; (handle, out_rx) @@ -232,6 +235,63 @@ impl BrokerHandle { } } + /// Broker a cross-daemon directory listing: ask the registry to relay the + /// `ListDir` to the daemon named by `daemon_id` (which may be a peer) and + /// return its listing. Same fail-fast-while-disconnected + RAII-cleanup + /// discipline as the other broker calls (NAP-329). + pub async fn forward_list_dir( + &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_list_dir.lock().unwrap().insert(reply_to, tx); + let _guard = PendingGuard { + map: &*self.pending_list_dir, + key: reply_to, + }; + if self + .out_tx + .try_send(DaemonOutbound::ForwardListDir { + 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_list_dir`](Self::forward_list_dir); called by + /// the registry-client task on a `ForwardListDirResult` frame. + pub async fn complete_forward_list_dir( + &self, + reply_to: ReplyTo, + result: RpcResult, + ) -> bool { + let entry = self.pending_list_dir.lock().unwrap().remove(&reply_to); + match entry { + Some(tx) => tx.send(result).is_ok(), + None => { + tracing::debug!(?reply_to, "no waiter for ForwardListDirResult"); + 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( @@ -457,6 +517,62 @@ mod tests { } } + // ----- NAP-329: forward_list_dir ----- + + /// A `forward_list_dir` issues a `ForwardListDir` frame and resolves with + /// the listing once the registry's `ForwardListDirResult` is delivered. + #[tokio::test] + async fn forward_list_dir_round_trips_listing() { + let (handle, mut out_rx) = BrokerHandle::new(); + handle.set_connected(true); + + let h = handle.clone(); + let task = + tokio::spawn(async move { h.forward_list_dir("konishi".into(), "/work".into()).await }); + + let reply_to = match out_rx.recv().await.expect("ForwardListDir went out") { + DaemonOutbound::ForwardListDir { + reply_to, + daemon_id, + path, + } => { + assert_eq!(daemon_id, "konishi"); + assert_eq!(path, "/work"); + reply_to + } + other => panic!("expected ForwardListDir, got {other:?}"), + }; + handle + .complete_forward_list_dir( + reply_to, + RpcResult::Ok { + value: DirListing { + path: std::path::PathBuf::from("/work"), + entries: vec!["repo".into()], + truncated: false, + }, + }, + ) + .await; + + let got = task.await.unwrap().expect("list ok"); + assert_eq!(got.entries, vec!["repo".to_string()]); + } + + /// While disconnected, `forward_list_dir` fails fast and leaves no pending + /// entry (NAP-263 / NAP-259). + #[tokio::test] + async fn forward_list_dir_fails_fast_while_disconnected() { + let (handle, _out_rx) = BrokerHandle::new(); // starts disconnected + assert!(matches!( + handle + .forward_list_dir("konishi".into(), "/work".into()) + .await, + Err(BrokerError::NoControlLink) + )); + assert_eq!(handle.pending_list_dir.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/dirlist.rs b/crates/maestro-daemon/src/dirlist.rs new file mode 100644 index 0000000..4c423d4 --- /dev/null +++ b/crates/maestro-daemon/src/dirlist.rs @@ -0,0 +1,188 @@ +//! Directory listing for the spawn-form working-directory autocomplete (NAP-329). +//! +//! A daemon answers `ListDir { path }` by listing the **immediate +//! subdirectories** of one directory on its own host. The web client lists one +//! directory at a time and filters the trailing path segment locally, so this +//! never walks recursively and never returns files. +//! +//! Path expansion is done against the daemon's own `$HOME` so a client can send +//! an abbreviated path (empty, `~`, or relative) and get a sensible result: +//! +//! - empty → home dir +//! - `~` / `~/sub` → home / home + `sub` +//! - relative `a/b` → home + `a/b` +//! - absolute `/a/b` → as-is +//! +//! The pure [`list_dir`] is unit-tested in isolation; the registry-client task +//! just wraps its `Result` in an `RpcResult` for the wire. + +use std::path::{Path, PathBuf}; + +use maestro_protocol::{DirListing, ErrorInfo}; + +/// Maximum number of entries returned in a single listing. A directory with +/// more subdirectories than this is reported `truncated`; the client narrows +/// its query by typing more of the name. Bounds the wire payload regardless of +/// how large a directory is. +pub const MAX_ENTRIES: usize = 1000; + +/// List the immediate subdirectories of `requested`, expanding it against +/// `home_dir` first. Returns directory **names** only (not full paths), sorted +/// case-insensitively, with hidden entries included. +pub fn list_dir(home_dir: &str, requested: &str) -> Result { + let dir = resolve(home_dir, requested); + + let read = std::fs::read_dir(&dir).map_err(|e| io_error(&dir, &e))?; + + let mut entries: Vec = Vec::new(); + for entry in read.flatten() { + if is_dir(&entry) { + entries.push(entry.file_name().to_string_lossy().into_owned()); + } + } + // Case-insensitive sort so `Projects` and `projects` interleave the way a + // user scanning the list expects. + entries.sort_by_key(|s| s.to_lowercase()); + + let truncated = entries.len() > MAX_ENTRIES; + entries.truncate(MAX_ENTRIES); + + Ok(DirListing { + path: dir, + entries, + truncated, + }) +} + +/// Expand an abbreviated client path against the daemon's home dir. +fn resolve(home_dir: &str, requested: &str) -> PathBuf { + let home = PathBuf::from(home_dir); + let trimmed = requested.trim(); + if trimmed.is_empty() { + return home; + } + if trimmed == "~" { + return home; + } + if let Some(rest) = trimmed.strip_prefix("~/") { + return home.join(rest); + } + let p = PathBuf::from(trimmed); + if p.is_absolute() { + p + } else { + home.join(p) + } +} + +/// Whether a `DirEntry` denotes a directory, following symlinks so that a +/// symlinked project dir still completes (shells do the same). +fn is_dir(entry: &std::fs::DirEntry) -> bool { + match entry.file_type() { + Ok(ft) if ft.is_dir() => true, + Ok(ft) if ft.is_symlink() => std::fs::metadata(entry.path()) + .map(|m| m.is_dir()) + .unwrap_or(false), + _ => false, + } +} + +/// Map a `read_dir` failure to a stable [`ErrorInfo`] code the client can react +/// to. `NotADirectory` isn't a stable `ErrorKind` across platforms, so we probe +/// with a `metadata` call to distinguish "file, not dir" from "doesn't exist". +fn io_error(dir: &Path, e: &std::io::Error) -> ErrorInfo { + use std::io::ErrorKind; + let (code, message) = match e.kind() { + ErrorKind::NotFound => ("not_found", "no such directory"), + ErrorKind::PermissionDenied => ("permission_denied", "permission denied"), + _ if dir.is_file() => ("not_a_directory", "path is a file, not a directory"), + _ => ("io_error", "could not read directory"), + }; + ErrorInfo { + code: code.into(), + message: format!("{message}: {}", dir.display()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + + /// Build a temp home with a few subdirs (one hidden), a file, and a symlink + /// to a dir. Returns the temp root path (its `Drop` cleans up). + fn fixture() -> tempfile::TempDir { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path(); + fs::create_dir(home.join("projects")).unwrap(); + fs::create_dir(home.join("Downloads")).unwrap(); + fs::create_dir(home.join(".config")).unwrap(); + fs::write(home.join("notes.txt"), b"x").unwrap(); + #[cfg(unix)] + std::os::unix::fs::symlink(home.join("projects"), home.join("link-to-projects")).unwrap(); + tmp + } + + #[test] + fn lists_dirs_only_sorted_case_insensitively_including_hidden() { + let tmp = fixture(); + let home = tmp.path().to_string_lossy(); + let listing = list_dir(&home, "").unwrap(); + // The file is excluded; dirs (incl. hidden + symlinked dir) are present. + assert!(listing.entries.contains(&".config".to_string())); + assert!(listing.entries.contains(&"projects".to_string())); + assert!(listing.entries.contains(&"Downloads".to_string())); + assert!(!listing.entries.contains(&"notes.txt".to_string())); + #[cfg(unix)] + assert!(listing.entries.contains(&"link-to-projects".to_string())); + // Case-insensitive order: .config, Downloads, link-to-projects, projects + let lower: Vec = listing.entries.iter().map(|s| s.to_lowercase()).collect(); + let mut sorted = lower.clone(); + sorted.sort(); + assert_eq!(lower, sorted); + assert!(!listing.truncated); + assert_eq!(listing.path, tmp.path()); + } + + #[test] + fn empty_and_tilde_resolve_to_home() { + let tmp = fixture(); + let home = tmp.path().to_string_lossy(); + assert_eq!(list_dir(&home, "").unwrap().path, tmp.path()); + assert_eq!(list_dir(&home, "~").unwrap().path, tmp.path()); + } + + #[test] + fn relative_and_tilde_slash_join_onto_home() { + let tmp = fixture(); + let home = tmp.path().to_string_lossy(); + let rel = list_dir(&home, "projects").unwrap(); + assert_eq!(rel.path, tmp.path().join("projects")); + let tilde = list_dir(&home, "~/projects").unwrap(); + assert_eq!(tilde.path, tmp.path().join("projects")); + } + + #[test] + fn absolute_path_is_listed_as_is() { + let tmp = fixture(); + let abs = tmp.path().join("projects"); + // home is irrelevant for an absolute path. + let listing = list_dir("/nonexistent-home", &abs.to_string_lossy()).unwrap(); + assert_eq!(listing.path, abs); + assert!(listing.entries.is_empty()); + } + + #[test] + fn missing_dir_is_not_found() { + let tmp = fixture(); + let err = list_dir(&tmp.path().to_string_lossy(), "does-not-exist").unwrap_err(); + assert_eq!(err.code, "not_found"); + } + + #[test] + fn file_target_is_not_a_directory() { + let tmp = fixture(); + let err = list_dir(&tmp.path().to_string_lossy(), "notes.txt").unwrap_err(); + assert_eq!(err.code, "not_a_directory"); + } +} diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index f93f871..1c4f1f6 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -12,6 +12,7 @@ pub mod catalog_mirror; pub mod cli; pub mod config; pub mod configure; +pub mod dirlist; pub mod jwt; pub mod local_api; pub mod local_events; diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 90fd489..a4dd1c2 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -25,13 +25,15 @@ use std::sync::Arc; -use axum::extract::{Path, State}; +use axum::extract::{Path, Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::{any, delete, get}; use axum::{Json, Router}; -use maestro_protocol::{CreateSessionRequest, DaemonInfo, ErrorInfo, Session, PROTOCOL_VERSION}; -use serde::Serialize; +use maestro_protocol::{ + CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, Session, PROTOCOL_VERSION, +}; +use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::catalog_mirror::CatalogMirror; @@ -73,6 +75,7 @@ pub fn router(state: LocalApiState) -> Router { .route("/api/sessions", get(list_sessions).post(create_session)) .route("/api/sessions/:id", delete(kill_session)) .route("/api/daemons", get(list_daemons)) + .route("/api/daemons/:id/dirs", get(list_dir)) .route("/api/web/info", get(web_info)) .route("/api/events", any(events_ws)) .with_state(state) @@ -97,6 +100,46 @@ async fn list_daemons(State(s): State) -> Json> { Json(out) } +/// Query string for `GET /api/daemons/:id/dirs`. `path` is the directory to +/// list (the parent of what the user is typing); empty/`~`/relative are +/// resolved against the target daemon's home dir. +#[derive(Debug, Deserialize)] +struct DirQuery { + #[serde(default)] + path: String, +} + +/// List one directory's immediate subdirectories on `daemon_id`'s host, for the +/// spawn form's working-directory autocomplete (NAP-329). Lists locally when +/// the target is this daemon; otherwise brokers a `ListDir` to the owning +/// daemon over the registry control link. +async fn list_dir( + State(s): State, + Path(daemon_id): Path, + Query(q): Query, +) -> Response { + if daemon_id == s.manager.daemon_id() { + return match crate::dirlist::list_dir(&s.home_dir, &q.path) { + Ok(listing) => Json(listing).into_response(), + Err(info) => error_response(dir_error_status(&info.code), &info.code, &info.message), + }; + } + match s.broker.forward_list_dir(daemon_id, q.path).await { + Ok(listing) => Json::(listing).into_response(), + Err(e) => broker_error_response(&e), + } +} + +/// HTTP status for a local [`crate::dirlist`] error code. +fn dir_error_status(code: &str) -> StatusCode { + match code { + "not_found" => StatusCode::NOT_FOUND, + "permission_denied" => StatusCode::FORBIDDEN, + "not_a_directory" => StatusCode::BAD_REQUEST, + _ => StatusCode::BAD_GATEWAY, + } +} + async fn web_info(State(s): State) -> Json { Json(WebInfo { daemon_id: s.manager.daemon_id().to_string(), diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index b4ef6a4..66cd187 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -311,6 +311,23 @@ impl RegistryClient { DaemonInbound::ForwardKillResult { reply_to, result } => { self.broker.complete_forward_kill(reply_to, result).await; } + DaemonInbound::ListDir { reply_to, path } => { + // List a directory on this host for a client's spawn-form + // autocomplete (NAP-329). Resolved against this daemon's home. + let result = match crate::dirlist::list_dir(&self.home_dir, &path) { + Ok(listing) => RpcResult::Ok { value: listing }, + Err(error) => RpcResult::Err { error }, + }; + sink.send(Message::Text(serde_json::to_string( + &DaemonOutbound::ListDirResult { reply_to, result }, + )?)) + .await?; + } + DaemonInbound::ForwardListDirResult { reply_to, result } => { + self.broker + .complete_forward_list_dir(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 3edb910..8913e1e 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -43,6 +43,7 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { manager.clone(), hostname, daemon_url, + "/home/test", catalog.clone(), events, broker.clone(), diff --git a/crates/maestro-registry/src/client_api.rs b/crates/maestro-registry/src/client_api.rs index 4ce3749..f0fc141 100644 --- a/crates/maestro-registry/src/client_api.rs +++ b/crates/maestro-registry/src/client_api.rs @@ -7,12 +7,13 @@ use std::sync::Arc; use std::time::Duration; -use axum::extract::{ws::WebSocketUpgrade, Path, State}; +use axum::extract::{ws::WebSocketUpgrade, Path, Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::{any, get}; use axum::{Json, Router}; -use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, Session}; +use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, DirListing, Session}; +use serde::Deserialize; use uuid::Uuid; use crate::events::events_ws; @@ -34,6 +35,7 @@ pub fn routes(state: ClientApiState) -> Router { // connection-orchestration spec §"UI hosting"). Only the JSON/WS API here. Router::new() .route("/api/daemons", get(list_daemons)) + .route("/api/daemons/:id/dirs", get(list_dir)) .route("/api/sessions", get(list_sessions).post(create_session)) .route("/api/sessions/:id", axum::routing::delete(delete_session)) .route("/api/sessions/:id/connect-info", get(session_connect_info)) @@ -56,6 +58,64 @@ async fn list_sessions(State(s): State) -> Json> { Json(s.registry.sessions().await) } +/// Query string for `GET /api/daemons/:id/dirs`. `path` is the directory to +/// list; empty/`~`/relative are resolved against the target daemon's home dir. +#[derive(Debug, Deserialize)] +struct DirQuery { + #[serde(default)] + path: String, +} + +/// List one directory's immediate subdirectories on `daemon_id`'s host, for the +/// web spawn form's working-directory autocomplete (NAP-329). Drives a `ListDir` +/// RPC to the target daemon, same machinery + error mapping as `create_session`. +async fn list_dir( + State(s): State, + Path(daemon_id): Path, + Query(q): Query, +) -> Response { + let outcome = + tokio::time::timeout(s.rpc_timeout, s.registry.list_dir(&daemon_id, q.path)).await; + match outcome { + Err(_) => ( + StatusCode::GATEWAY_TIMEOUT, + Json(error_body("timeout", "daemon did not reply in time")), + ) + .into_response(), + Ok(Err(RegistryError::DaemonNotConnected(_))) => ( + StatusCode::NOT_FOUND, + Json(error_body("unknown_daemon", "daemon is not connected")), + ) + .into_response(), + Ok(Err(RegistryError::Timeout(_))) => ( + StatusCode::GATEWAY_TIMEOUT, + Json(error_body("timeout", "daemon did not reply in time")), + ) + .into_response(), + Ok(Err(e)) => ( + StatusCode::BAD_GATEWAY, + Json(error_body("transport", &e.to_string())), + ) + .into_response(), + Ok(Ok(Err(rpc_err))) => ( + dir_error_status(&rpc_err.0.code), + Json(error_body(&rpc_err.0.code, &rpc_err.0.message)), + ) + .into_response(), + Ok(Ok(Ok(listing))) => Json::(listing).into_response(), + } +} + +/// HTTP status for a daemon-reported directory-listing error code (NAP-329). +fn dir_error_status(code: &str) -> StatusCode { + match code { + "not_found" => StatusCode::NOT_FOUND, + "permission_denied" => StatusCode::FORBIDDEN, + "not_a_directory" => StatusCode::BAD_REQUEST, + _ => StatusCode::BAD_GATEWAY, + } +} + async fn create_session( State(s): State, Json(body): Json, diff --git a/crates/maestro-registry/src/handler.rs b/crates/maestro-registry/src/handler.rs index 742e061..35fb525 100644 --- a/crates/maestro-registry/src/handler.rs +++ b/crates/maestro-registry/src/handler.rs @@ -227,6 +227,20 @@ async fn dispatch_text( .forward_kill(daemon_id, reply_to, &target, id) .await; } + DaemonOutbound::ListDirResult { reply_to, result } => { + registry.complete_list_dir(reply_to, result).await; + } + DaemonOutbound::ForwardListDir { + reply_to, + daemon_id: target, + path, + } => { + // Relay a directory listing to its owning daemon and route the reply + // back to this (home) daemon — control-plane only (NAP-329). + registry + .forward_list_dir(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 8ba3428..652d258 100644 --- a/crates/maestro-registry/src/state.rs +++ b/crates/maestro-registry/src/state.rs @@ -15,7 +15,7 @@ use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use maestro_protocol::{ - AttachCandidates, CreateSessionRequest, DaemonInbound, DaemonInfo, ErrorInfo, + AttachCandidates, CreateSessionRequest, DaemonInbound, DaemonInfo, DirListing, ErrorInfo, PerimeterCandidate, RegistryEvent, ReplyTo, RpcResult, Session, }; use tokio::sync::{broadcast, mpsc, oneshot, Mutex}; @@ -108,6 +108,9 @@ pub struct Registry { /// drop — including when the awaiting HTTP handler is cancelled (NAP-259). pending_create: Arc>>>>, pending_kill: Arc>>>>, + /// Outstanding `ListDir` RPC waiters (NAP-329). Same `PendingGuard` + /// discipline as `pending_create` / `pending_kill`. + pending_list_dir: Arc>>>>, } #[derive(Debug, Default)] @@ -155,6 +158,7 @@ impl Registry { events, 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())), } } @@ -487,6 +491,30 @@ impl Registry { .await; } + /// Relay a home daemon's `ForwardListDir` to the target daemon and push the + /// `ForwardListDirResult` back to the home daemon (NAP-329). Reuses + /// [`list_dir`](Self::list_dir), then routes the reply to `home_daemon_id`. + pub async fn forward_list_dir( + &self, + home_daemon_id: &str, + reply_to: ReplyTo, + target_daemon_id: &str, + path: String, + ) { + let result = match self.list_dir(target_daemon_id, path).await { + Ok(Ok(listing)) => RpcResult::Ok { value: listing }, + Ok(Err(RpcError(error))) => RpcResult::Err { error }, + Err(e) => RpcResult::Err { + error: registry_error_info(&e), + }, + }; + self.reply_to_daemon( + home_daemon_id, + DaemonInbound::ForwardListDirResult { 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). @@ -518,6 +546,19 @@ impl Registry { } } + /// Resolve a ListDir RPC reply. Returns true if a waiter was found (NAP-329). + pub async fn complete_list_dir( + &self, + reply_to: ReplyTo, + result: RpcResult, + ) -> bool { + let entry = self.pending_list_dir.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) // ----------------------------------------------------------------------- @@ -610,6 +651,40 @@ impl Registry { Ok(rpc) } + /// Forward a `ListDir` to the named daemon and wait for the matching + /// `ListDirResult` (NAP-329). No catalog side-effects — listings aren't + /// cached registry-side. Mirrors [`create_session`](Self::create_session). + pub async fn list_dir( + &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_list_dir.lock().unwrap().insert(reply_to, tx); + let _guard = PendingGuard { + map: &*self.pending_list_dir, + key: reply_to, + }; + if outbound + .send(DaemonInbound::ListDir { reply_to, path }) + .await + .is_err() + { + return Err(RegistryError::DaemonGone); + } + await_rpc(rx).await.map(rpc_into_result) + } + // ----------------------------------------------------------------------- // Read-only views (mostly for tests) // ----------------------------------------------------------------------- @@ -720,6 +795,7 @@ mod tests { "host".into(), "0.1.0".into(), format!("ws://{daemon_id}:7070"), + "/home/test".into(), tx, ) .await; @@ -994,6 +1070,7 @@ mod nap259_tests { "host".into(), "0.1.0".into(), "ws://127.0.0.1:1".into(), + "/home/test".into(), tx, ) .await;