feat(maestro): spawn protocol types + relay + daemon HTTP for recents/gitroot/presets (#26)

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) <noreply@anthropic.com>
This commit is contained in:
2026-06-20 21:09:18 +00:00
co-authored by Claude Opus 4.8
parent ac11997f58
commit cc8412ca59
11 changed files with 776 additions and 5 deletions
+221 -1
View File
@@ -19,7 +19,8 @@ use std::sync::{Arc, Mutex};
use std::time::Duration; use std::time::Duration;
use maestro_protocol::{ 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 tokio::sync::{mpsc, oneshot};
use uuid::Uuid; use uuid::Uuid;
@@ -73,6 +74,10 @@ pub struct BrokerHandle {
pending_kill: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<()>>>>>, pending_kill: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<()>>>>>,
/// Pending `ForwardListDir` replies, keyed by `reply_to` (NAP-329). /// Pending `ForwardListDir` replies, keyed by `reply_to` (NAP-329).
pending_list_dir: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<DirListing>>>>>, pending_list_dir: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<DirListing>>>>>,
/// Pending `ForwardRecentDirs` replies, keyed by `reply_to` (#26).
pending_recent_dirs: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<RecentDirs>>>>>,
/// Pending `ForwardGitRoot` replies, keyed by `reply_to` (#26).
pending_git_root: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<GitRoot>>>>>,
/// Whether the registry control link is currently up. Flipped by the /// Whether the registry control link is currently up. Flipped by the
/// registry-client task via [`BrokerHandle::set_connected`]. When down, /// registry-client task via [`BrokerHandle::set_connected`]. When down,
/// `resolve_attach` fails fast instead of queueing onto an undrained /// `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_create: Arc::new(Mutex::new(HashMap::new())),
pending_kill: Arc::new(Mutex::new(HashMap::new())), pending_kill: Arc::new(Mutex::new(HashMap::new())),
pending_list_dir: 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)), connected: Arc::new(AtomicBool::new(false)),
}; };
(handle, out_rx) (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<RecentDirs, BrokerError> {
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<RecentDirs>,
) -> 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<GitRoot, BrokerError> {
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<GitRoot>,
) -> 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 /// Resolve a waiting [`forward_create`](Self::forward_create); called by the
/// registry-client task on a `ForwardCreateResult` frame. /// registry-client task on a `ForwardCreateResult` frame.
pub async fn complete_forward_create( pub async fn complete_forward_create(
@@ -573,6 +691,108 @@ mod tests {
assert_eq!(handle.pending_list_dir.lock().unwrap().len(), 0); 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 /// NAP-270: if the registry never replies, `resolve_attach` resolves to
/// `BrokerError::Timeout` after `RESOLVE_TIMEOUT` (driven here with a paused /// `BrokerError::Timeout` after `RESOLVE_TIMEOUT` (driven here with a paused
/// clock so the test is instant). /// clock so the test is instant).
+1
View File
@@ -140,6 +140,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
config.hostname.clone(), config.hostname.clone(),
config.daemon_url.clone(), config.daemon_url.clone(),
config.home_dir.clone(), config.home_dir.clone(),
config.spawn_presets.clone(),
catalog.clone(), catalog.clone(),
events, events,
client.broker(), client.broker(),
+11
View File
@@ -63,6 +63,8 @@ pub enum ConfigError {
env: &'static str, env: &'static str,
path: PathBuf, path: PathBuf,
}, },
#[error("invalid [spawn].presets: {0}")]
SpawnPresets(#[from] maestro_config::SpawnConfigError),
} }
/// Resolved, validated daemon configuration ready to drive the binary. /// 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 /// form's "recent dirs" picker (#24). `0` disables history. Default
/// [`DEFAULT_SPAWN_HISTORY_DAYS`]. /// [`DEFAULT_SPAWN_HISTORY_DAYS`].
pub spawn_history_days: u32, 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<maestro_config::SpawnPreset>,
} }
impl DaemonConfig { impl DaemonConfig {
@@ -195,6 +201,10 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
let shared = file.shared; let shared = file.shared;
let daemon = file.daemon; let daemon = file.daemon;
// Resolve spawn presets up front so a bad `[spawn].presets` (empty/dup id,
// empty command) refuses startup with a clear error, like other config
// validation (#22/#26).
let spawn_presets = file.spawn.resolve()?;
let hostname = daemon.hostname.unwrap_or_else(system_hostname); let hostname = daemon.hostname.unwrap_or_else(system_hostname);
let home_dir = dirs::home_dir() let home_dir = dirs::home_dir()
@@ -269,6 +279,7 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
pty_replay_bytes, pty_replay_bytes,
tmux_tmpdir, tmux_tmpdir,
spawn_history_days, spawn_history_days,
spawn_presets,
}) })
} }
+5 -1
View File
@@ -61,7 +61,11 @@ pub fn list_dir(home_dir: &str, requested: &str) -> Result<DirListing, ErrorInfo
} }
/// Expand an abbreviated client path against the daemon's home dir. /// Expand an abbreviated client path against the daemon's home dir.
fn resolve(home_dir: &str, requested: &str) -> 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 home = PathBuf::from(home_dir);
let trimmed = requested.trim(); let trimmed = requested.trim();
if trimmed.is_empty() { if trimmed.is_empty() {
+53 -2
View File
@@ -30,8 +30,10 @@ use axum::http::StatusCode;
use axum::response::{IntoResponse, Response}; use axum::response::{IntoResponse, Response};
use axum::routing::{any, delete, get}; use axum::routing::{any, delete, get};
use axum::{Json, Router}; use axum::{Json, Router};
use maestro_config::SpawnPreset;
use maestro_protocol::{ use maestro_protocol::{
CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, Session, PROTOCOL_VERSION, CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, GitRoot, RecentDirs, Session,
PROTOCOL_VERSION,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use uuid::Uuid; use uuid::Uuid;
@@ -42,13 +44,18 @@ use crate::session::SessionManager;
/// Daemon identity returned by `GET /api/web/info`. Browsers render this in the /// Daemon identity returned by `GET /api/web/info`. Browsers render this in the
/// header ("you're connected to konishi"); the field set mirrors `DaemonInfo` /// 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)] #[derive(Debug, Clone, Serialize)]
pub struct WebInfo { pub struct WebInfo {
pub daemon_id: String, pub daemon_id: String,
pub hostname: String, pub hostname: String,
pub version: String, pub version: String,
pub daemon_url: 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<SpawnPreset>,
} }
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
@@ -59,6 +66,8 @@ pub struct LocalApiState {
/// Home directory of the OS user this daemon runs as, advertised so clients /// Home directory of the OS user this daemon runs as, advertised so clients
/// can prefill a spawn form's working directory. Empty if unresolved. /// can prefill a spawn form's working directory. Empty if unresolved.
pub home_dir: Arc<String>, pub home_dir: Arc<String>,
/// Resolved spawn presets (#22/#26) advertised in `GET /api/web/info`.
pub presets: Arc<Vec<SpawnPreset>>,
/// Peer-only mirror of the global catalog. Empty until the daemon's /// Peer-only mirror of the global catalog. Empty until the daemon's
/// registry client connects and receives the snapshot. /// registry client connects and receives the snapshot.
pub catalog: CatalogMirror, pub catalog: CatalogMirror,
@@ -76,6 +85,8 @@ pub fn router(state: LocalApiState) -> Router {
.route("/api/sessions/:id", delete(kill_session)) .route("/api/sessions/:id", delete(kill_session))
.route("/api/daemons", get(list_daemons)) .route("/api/daemons", get(list_daemons))
.route("/api/daemons/:id/dirs", get(list_dir)) .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/web/info", get(web_info))
.route("/api/events", any(events_ws)) .route("/api/events", any(events_ws))
.with_state(state) .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<LocalApiState>, Path(daemon_id): Path<String>) -> 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::<RecentDirs>(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<LocalApiState>,
Path(daemon_id): Path<String>,
Query(q): Query<DirQuery>,
) -> 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::<GitRoot>(root).into_response(),
Err(e) => broker_error_response(&e),
}
}
async fn web_info(State(s): State<LocalApiState>) -> Json<WebInfo> { async fn web_info(State(s): State<LocalApiState>) -> Json<WebInfo> {
Json(WebInfo { Json(WebInfo {
daemon_id: s.manager.daemon_id().to_string(), daemon_id: s.manager.daemon_id().to_string(),
hostname: (*s.hostname).clone(), hostname: (*s.hostname).clone(),
version: PROTOCOL_VERSION.into(), version: PROTOCOL_VERSION.into(),
daemon_url: (*s.daemon_url).clone(), 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 /// Convenience constructor — pulls the bits `LocalApiState` needs from
/// `DaemonConfig` so callers don't have to assemble it by hand. /// `DaemonConfig` so callers don't have to assemble it by hand.
#[allow(clippy::too_many_arguments)]
pub fn state_for( pub fn state_for(
manager: SessionManager, manager: SessionManager,
hostname: impl Into<String>, hostname: impl Into<String>,
daemon_url: impl Into<String>, daemon_url: impl Into<String>,
home_dir: impl Into<String>, home_dir: impl Into<String>,
presets: Vec<SpawnPreset>,
catalog: CatalogMirror, catalog: CatalogMirror,
events: LocalEventBus, events: LocalEventBus,
broker: BrokerHandle, broker: BrokerHandle,
@@ -263,6 +313,7 @@ pub fn state_for(
hostname: Arc::new(hostname.into()), hostname: Arc::new(hostname.into()),
daemon_url: Arc::new(daemon_url.into()), daemon_url: Arc::new(daemon_url.into()),
home_dir: Arc::new(home_dir.into()), home_dir: Arc::new(home_dir.into()),
presets: Arc::new(presets),
catalog, catalog,
events, events,
broker, broker,
@@ -379,6 +379,44 @@ impl RegistryClient {
.complete_forward_list_dir(reply_to, result) .complete_forward_list_dir(reply_to, result)
.await; .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 { .. } => { DaemonInbound::RelaySetup { .. } => {
// T3 (relay) is gated on NAP-238. Lands here only so newer // T3 (relay) is gated on NAP-238. Lands here only so newer
// registries sending it don't crash this daemon on parse. // registries sending it don't crash this daemon on parse.
@@ -61,6 +61,23 @@ pub struct RecentDir {
pub last_used: DateTime<Utc>, pub last_used: DateTime<Utc>,
} }
/// 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<RecentDir>) -> 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. /// Append-only JSONL spawn log rooted at a file path, with a recency cap.
/// Cheap to clone — just the path + caps. /// Cheap to clone — just the path + caps.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
+73
View File
@@ -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 // resolve_attach / forward_create / forward_kill fails fast with
// NoControlLink (the handle starts disconnected). // NoControlLink (the handle starts disconnected).
let (broker, _broker_rx) = BrokerHandle::new(); 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( let state = local_api::state_for(
manager.clone(), manager.clone(),
hostname, hostname,
daemon_url, daemon_url,
"/home/test", "/home/test",
presets,
catalog.clone(), catalog.clone(),
events, events,
broker.clone(), 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"); 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] #[tokio::test]
async fn delete_unknown_session_returns_404() { async fn delete_unknown_session_returns_404() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
+187
View File
@@ -85,6 +85,36 @@ pub struct DirListing {
pub truncated: bool, 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<Utc>,
}
/// 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<RecentDir>,
#[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<PathBuf>,
}
/// Structured failure used everywhere we don't want to leak a string-only error. /// Structured failure used everywhere we don't want to leak a string-only error.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ErrorInfo { pub struct ErrorInfo {
@@ -192,6 +222,34 @@ pub enum DaemonOutbound {
daemon_id: String, daemon_id: String,
path: String, path: String,
}, },
/// Owning daemon's reply to a [`DaemonInbound::RecentDirs`] — its recent
/// working dirs. (#26)
RecentDirsResult {
reply_to: ReplyTo,
result: RpcResult<RecentDirs>,
},
/// 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<GitRoot>,
},
/// 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 /// Target daemon answers a [`DaemonInbound::RelaySetup`]: whether it will
/// accept relay traffic for the requested session. Registry only allocates /// accept relay traffic for the requested session. Registry only allocates
/// relay capacity when `accepted == true`. /// relay capacity when `accepted == true`.
@@ -251,6 +309,31 @@ pub enum DaemonInbound {
reply_to: ReplyTo, reply_to: ReplyTo,
result: RpcResult<DirListing>, result: RpcResult<DirListing>,
}, },
/// 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<RecentDirs>,
},
/// 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<GitRoot>,
},
/// Registry signals a target daemon that a home daemon wants to attach via /// Registry signals a target daemon that a home daemon wants to attach via
/// the registry-mediated relay (T3). Target decides whether to accept the /// the registry-mediated relay (T3). Target decides whether to accept the
/// relay and replies with [`DaemonOutbound::RelayAck`]. /// relay and replies with [`DaemonOutbound::RelayAck`].
@@ -1002,6 +1085,110 @@ mod tests {
assert_eq!(res, round_trip(&res)); 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] #[test]
fn daemon_outbound_relay_ack_round_trip_and_tag() { fn daemon_outbound_relay_ack_round_trip_and_tag() {
let msg = DaemonOutbound::RelayAck { let msg = DaemonOutbound::RelayAck {
+27
View File
@@ -241,6 +241,33 @@ async fn dispatch_text(
.forward_list_dir(daemon_id, reply_to, &target, path) .forward_list_dir(daemon_id, reply_to, &target, path)
.await; .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, .. } => { DaemonOutbound::RelayAck { reply_to, .. } => {
// T3 (relay) is gated on NAP-238. Land the variant so newer daemons // 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. // don't crash the connection; the registry doesn't act on it yet.
+143 -1
View File
@@ -16,7 +16,7 @@ use std::time::Duration;
use maestro_protocol::{ use maestro_protocol::{
AttachCandidates, CreateSessionRequest, DaemonInbound, DaemonInfo, DirListing, ErrorInfo, 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 tokio::sync::{broadcast, mpsc, oneshot, Mutex};
use uuid::Uuid; use uuid::Uuid;
@@ -113,6 +113,10 @@ pub struct Registry {
/// Outstanding `ListDir` RPC waiters (NAP-329). Same `PendingGuard` /// Outstanding `ListDir` RPC waiters (NAP-329). Same `PendingGuard`
/// discipline as `pending_create` / `pending_kill`. /// discipline as `pending_create` / `pending_kill`.
pending_list_dir: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<DirListing>>>>>, pending_list_dir: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<DirListing>>>>>,
/// Outstanding `RecentDirs` RPC waiters (#26).
pending_recent_dirs: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<RecentDirs>>>>>,
/// Outstanding `GitRoot` RPC waiters (#26).
pending_git_root: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<GitRoot>>>>>,
} }
#[derive(Debug, Default)] #[derive(Debug, Default)]
@@ -161,6 +165,8 @@ impl Registry {
pending_create: Arc::new(StdMutex::new(HashMap::new())), pending_create: Arc::new(StdMutex::new(HashMap::new())),
pending_kill: Arc::new(StdMutex::new(HashMap::new())), pending_kill: Arc::new(StdMutex::new(HashMap::new())),
pending_list_dir: 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; .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 /// 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 /// it silently if the daemon has gone away mid-flight (the home daemon's
/// own RPC timeout will surface the failure). /// 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<RecentDirs>,
) -> 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<GitRoot>) -> 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) // Outbound R→D requests (internal API for NAP-215)
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
@@ -688,6 +761,75 @@ impl Registry {
await_rpc(rx).await.map(rpc_into_result) 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<Result<RecentDirs, RpcError>, 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<Result<GitRoot, RpcError>, 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) // Read-only views (mostly for tests)
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------