diff --git a/crates/maestro-cli/src/attach.rs b/crates/maestro-cli/src/attach.rs index 4b924bf..8276827 100644 --- a/crates/maestro-cli/src/attach.rs +++ b/crates/maestro-cli/src/attach.rs @@ -19,6 +19,12 @@ use tokio_tungstenite::tungstenite::Message; pub enum AttachError { #[error("ws: {0}")] WebSocket(#[from] tokio_tungstenite::tungstenite::Error), + #[error("daemon returned {status} ({code}): {message}")] + Rejected { + status: String, + code: String, + message: String, + }, #[error("io: {0}")] Io(#[from] io::Error), #[error("termios: {0}")] @@ -32,6 +38,21 @@ pub enum AttachError { } impl AttachError { + fn from_websocket(error: tokio_tungstenite::tungstenite::Error) -> Self { + if let tokio_tungstenite::tungstenite::Error::Http(response) = &error { + if let Some(body) = response.body().as_deref() { + if let Ok(error_body) = serde_json::from_slice::(body) { + return Self::Rejected { + status: response.status().to_string(), + code: error_body.error.code, + message: error_body.error.message, + }; + } + } + } + Self::WebSocket(error) + } + /// True when the failure was an inability to reach the WS endpoint at the /// transport level (connection refused), as opposed to an HTTP-level /// rejection from a daemon that did answer. @@ -111,7 +132,9 @@ fn current_winsize() -> Option<(u16, u16)> { /// byte that exits cleanly without killing the session. #[allow(clippy::result_large_err)] pub async fn run_attach_url(url: &str, detach_byte: u8) -> Result<(), AttachError> { - let (ws, _resp) = tokio_tungstenite::connect_async(url).await?; + let (ws, _resp) = tokio_tungstenite::connect_async(url) + .await + .map_err(AttachError::from_websocket)?; let (mut sink, mut stream) = ws.split(); let raw = RawTty::enable()?; @@ -233,4 +256,25 @@ mod tests { Err(AttachError::InvalidDetachKey(_)) )); } + + #[test] + fn attach_surfaces_structured_handshake_errors() { + let body = serde_json::to_vec(&maestro_protocol::ApiError::new( + "not_terminal", + "session is not a terminal session", + )) + .unwrap(); + let response = tokio_tungstenite::tungstenite::http::Response::builder() + .status(409) + .body(Some(body)) + .unwrap(); + + let error = + AttachError::from_websocket(tokio_tungstenite::tungstenite::Error::Http(response)); + + assert_eq!( + error.to_string(), + "daemon returned 409 Conflict (not_terminal): session is not a terminal session" + ); + } } diff --git a/crates/maestro-cli/src/sessions.rs b/crates/maestro-cli/src/sessions.rs index 21296a5..f98d9eb 100644 --- a/crates/maestro-cli/src/sessions.rs +++ b/crates/maestro-cli/src/sessions.rs @@ -6,7 +6,7 @@ use std::io::Write; use maestro_protocol::display::abbreviate_dir; use maestro_protocol::{ ActivityState, ClientError, CreateSessionRequest, DaemonClient, DaemonInfo, SendInputRequest, - Session, SessionStatus, WorktreeSpec, + Session, SessionStatus, SessionView, WorktreeSpec, }; use serde_json::Value; @@ -179,8 +179,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon } writeln!( out, - " {:<36} {:<8} {:<9} {:<28} {:<40} {}", - "ID", "STATUS", "ACTIVITY", "DIR", "COMMAND", "BLOCKED" + " {:<36} {:<8} {:<10} {:<9} {:<28} {:<40} {}", + "ID", "STATUS", "VIEW", "ACTIVITY", "DIR", "COMMAND", "BLOCKED" ) .ok(); for s in sorted { @@ -190,6 +190,10 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon SessionStatus::Running => "running", SessionStatus::Dead => "dead", }; + let view = match s.view { + SessionView::Terminal => "terminal", + SessionView::Structured => "structured", + }; let activity = match s.activity { ActivityState::Active => "active", ActivityState::Idle => "idle", @@ -207,8 +211,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon let blocked = blocked_label(s); writeln!( out, - " {:<36} {:<8} {:<9} {:<28} {:<40} {blocked}", - s.id, status, activity, dir, command + " {:<36} {:<8} {:<10} {:<9} {:<28} {:<40} {blocked}", + s.id, status, view, activity, dir, command ) .ok(); } @@ -230,7 +234,7 @@ mod tests { use std::path::PathBuf; use chrono::{TimeZone, Utc}; - use maestro_protocol::{ActivityState, Session, SessionStatus}; + use maestro_protocol::{ActivityState, Session, SessionStatus, SessionView}; use uuid::Uuid; use super::DaemonInfo; @@ -286,6 +290,22 @@ mod tests { assert!(s.contains("idle"), "{s}"); } + #[test] + fn table_exposes_terminal_and_structured_views() { + let terminal = session("/tmp", "bash", 1); + let mut structured = session("/tmp", "claude", 2); + structured.view = SessionView::Structured; + + let mut buf = Vec::new(); + super::print_session_table(&mut buf, &[terminal, structured], &info("")); + let out = String::from_utf8(buf).unwrap(); + let lines: Vec<&str> = out.lines().collect(); + let header = lines.iter().position(|line| line.contains("VIEW")).unwrap(); + + assert!(lines[header + 1].contains("terminal"), "{out}"); + assert!(lines[header + 2].contains("structured"), "{out}"); + } + #[test] fn live_cwd_overrides_spawn_dir_in_table() { let mut s = session("/home/naps/maestro", "bash", 1); diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 81f10fe..e5ceee4 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -26,7 +26,7 @@ use maestro_protocol::{ use serde::Deserialize; use uuid::Uuid; -use crate::session::{QueueAcpPrompt, SessionManager}; +use crate::session::{QueueAcpPrompt, SessionError, SessionManager}; use crate::sidecar::EventCursor; #[derive(Debug, Clone)] @@ -179,6 +179,7 @@ fn session_error_status(code: &str) -> StatusCode { // The session outlived its pane: nothing to type into, and nothing // the daemon can do about it. "not_running" => StatusCode::CONFLICT, + "not_terminal" => StatusCode::CONFLICT, "shutting_down" => StatusCode::SERVICE_UNAVAILABLE, _ => StatusCode::BAD_GATEWAY, } @@ -534,12 +535,23 @@ async fn attach_ws( State(s): State, ws: axum::extract::WebSocketUpgrade, ) -> Result { - let Some((replay, rx)) = s.manager.subscribe_pty_with_replay(id).await else { - return Err(error_response( - StatusCode::NOT_FOUND, - "unknown_session", - "no such session on this daemon", - )); + let (replay, rx) = match s.manager.subscribe_pty_with_replay(id).await { + Ok(subscription) => subscription, + Err(SessionError::NotFound(_)) => { + return Err(error_response( + StatusCode::NOT_FOUND, + "unknown_session", + "no such session on this daemon", + )); + } + Err(error) => { + let info = error.to_error_info(); + return Err(error_response( + session_error_status(&info.code), + &info.code, + &info.message, + )); + } }; let manager = s.manager.clone(); Ok(ws.on_upgrade(move |socket| async move { diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index caeeca8..06cd2c7 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -169,6 +169,8 @@ pub enum SessionError { /// a stray keystroke on an attach socket, which is dropped. #[error("session is no longer running: {0}")] NotRunning(Uuid), + #[error("session is not a terminal session")] + NotTerminal(Uuid), #[error("daemon is shutting down")] ShuttingDown, /// The kill was refused (#79): the session's worktree holds unsaved work. @@ -205,6 +207,7 @@ impl SessionError { SessionError::NotFound(_) => "not_found", SessionError::InvalidEnv(_) => "invalid_env", SessionError::NotRunning(_) => "not_running", + SessionError::NotTerminal(_) => "not_terminal", SessionError::ShuttingDown => "shutting_down", SessionError::KillGuarded(_) => "kill_guarded", }; @@ -879,11 +882,17 @@ impl SessionManager { } } - /// Subscribe to a session's PTY broadcast. `None` if the session is - /// unknown to this daemon. - pub async fn subscribe_pty(&self, id: Uuid) -> Option>> { + /// Subscribe to a terminal session's PTY broadcast. + pub async fn subscribe_pty( + &self, + id: Uuid, + ) -> Result>, SessionError> { let sessions = self.inner.sessions.lock().await; - sessions.get(&id).map(|r| r.bridge.subscribe()) + let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; + if record.session.view != SessionView::Terminal { + return Err(SessionError::NotTerminal(id)); + } + Ok(record.bridge.subscribe()) } /// Subscribe to a session's PTY broadcast and return the replay buffer @@ -897,11 +906,14 @@ impl SessionManager { pub async fn subscribe_pty_with_replay( &self, id: Uuid, - ) -> Option<(Vec, broadcast::Receiver>)> { + ) -> Result<(Vec, broadcast::Receiver>), SessionError> { let tmux = self.inner.tmux.clone(); let (result, updated) = { let mut sessions = self.inner.sessions.lock().await; - let rec = sessions.get_mut(&id)?; + let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?; + if rec.session.view != SessionView::Terminal { + return Err(SessionError::NotTerminal(id)); + } rec.session.last_focused_at = Some(Utc::now()); let updated = rec.session.clone(); // For a live session, tmux's own screen beats the byte tail: the @@ -932,7 +944,7 @@ impl SessionManager { .inner .events .send(CatalogEvent::SessionUpdated { session: updated }); - Some(result) + Ok(result) } pub async fn has(&self, id: Uuid) -> bool { @@ -945,6 +957,15 @@ impl SessionManager { sessions.get(&id).map(|r| r.session.status) } + async fn terminal_status(&self, id: Uuid) -> Result { + let sessions = self.inner.sessions.lock().await; + let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; + if record.session.view != SessionView::Terminal { + return Err(SessionError::NotTerminal(id)); + } + Ok(record.session.status) + } + /// Whether the natural-exit settle step (#82) has finished for this /// session: `Some(true)` once applied or refused, `Some(false)` while /// still pending, `None` if the session is unknown. `Dead` is observable @@ -1626,10 +1647,9 @@ impl SessionManager { /// its pane is gone, so this is a no-op rather than a tmux failure the /// client would have to interpret. pub async fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), SessionError> { - match self.status(id).await { - None => return Err(SessionError::NotFound(id)), - Some(SessionStatus::Dead) => return Ok(()), - Some(SessionStatus::Running) => {} + match self.terminal_status(id).await? { + SessionStatus::Dead => return Ok(()), + SessionStatus::Running => {} } let tmux = self.inner.tmux.clone(); let bytes = bytes.to_vec(); @@ -1645,10 +1665,9 @@ impl SessionManager { /// asked for text to land somewhere, and a silent success would let a /// script believe an agent had been told something it never heard. pub async fn send_input(&self, id: Uuid, text: &str, submit: bool) -> Result<(), SessionError> { - match self.status(id).await { - None => return Err(SessionError::NotFound(id)), - Some(SessionStatus::Dead) => return Err(SessionError::NotRunning(id)), - Some(SessionStatus::Running) => {} + match self.terminal_status(id).await? { + SessionStatus::Dead => return Err(SessionError::NotRunning(id)), + SessionStatus::Running => {} } let tmux = self.inner.tmux.clone(); let bytes = text.as_bytes().to_vec(); @@ -1668,10 +1687,9 @@ impl SessionManager { /// Resize a session's window. No-op for a dead session, as with /// [`Self::send_keys`]. pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> Result<(), SessionError> { - match self.status(id).await { - None => return Err(SessionError::NotFound(id)), - Some(SessionStatus::Dead) => return Ok(()), - Some(SessionStatus::Running) => {} + match self.terminal_status(id).await? { + SessionStatus::Dead => return Ok(()), + SessionStatus::Running => {} } let tmux = self.inner.tmux.clone(); tmux_blocking(move || tmux.resize_window(id, cols, rows)) @@ -1703,6 +1721,7 @@ impl SessionManager { cols: u16, rows: u16, ) -> Result<(), SessionError> { + self.terminal_status(id).await?; let target = self .inner .viewports @@ -2807,13 +2826,15 @@ mod tests { async fn manager_with_structured_stream() -> ( tempfile::TempDir, SessionManager, + Arc, uuid::Uuid, Arc>, uuid::Uuid, ) { let state = tempfile::tempdir().unwrap(); + let panes = Arc::new(FakePanes::new()); let manager = SessionManager::new( - Arc::new(FakePanes::new()), + panes.clone(), SidecarStore::new(state.path().join("sessions")), WorktreeManager::new(None), state.path().join("fifos"), @@ -2848,12 +2869,12 @@ mod tests { record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD); record.acp_events = Some(events.clone()); } - (state, manager, structured.id, events, terminal.id) + (state, manager, panes, structured.id, events, terminal.id) } #[tokio::test] async fn blocked_structured_activity_read_does_not_block_the_catalog() { - let (state, manager, structured_id, events, terminal_id) = + let (state, manager, _panes, structured_id, events, terminal_id) = manager_with_structured_stream().await; let held_stream = events.lock().await; let activity_pass = run_structured_activity_pass(&manager.inner); @@ -2898,9 +2919,55 @@ mod tests { assert_eq!(structured.activity, ActivityState::Idle); } + #[tokio::test] + async fn structured_sessions_reject_every_pty_operation_before_panes() { + let (_state, manager, panes, structured_id, _events, terminal_id) = + manager_with_structured_stream().await; + let before = panes.calls(); + + assert!(matches!( + manager.subscribe_pty(structured_id).await, + Err(super::SessionError::NotTerminal(id)) if id == structured_id + )); + assert!(matches!( + manager.subscribe_pty_with_replay(structured_id).await, + Err(super::SessionError::NotTerminal(id)) if id == structured_id + )); + assert!(matches!( + manager.send_keys(structured_id, b"ignored").await, + Err(super::SessionError::NotTerminal(id)) if id == structured_id + )); + assert!(matches!( + manager.send_input(structured_id, "ignored", false).await, + Err(super::SessionError::NotTerminal(id)) if id == structured_id + )); + assert!(matches!( + manager.resize(structured_id, 80, 24).await, + Err(super::SessionError::NotTerminal(id)) if id == structured_id + )); + assert!(matches!( + manager + .set_viewport(structured_id, manager.attach_viewport(), 80, 24) + .await, + Err(super::SessionError::NotTerminal(id)) if id == structured_id + )); + + let after = panes.calls(); + assert_eq!(after, before, "structured PTY operations touched panes"); + + manager + .send_input(terminal_id, "works", false) + .await + .expect("terminal input remains available"); + manager + .resize(terminal_id, 100, 30) + .await + .expect("terminal resize remains available"); + } + #[tokio::test] async fn delayed_activity_from_a_replaced_stream_is_discarded() { - let (_state, manager, structured_id, old_events, _terminal_id) = + let (_state, manager, _panes, structured_id, old_events, _terminal_id) = manager_with_structured_stream().await; let replacement = Arc::new(tokio::sync::Mutex::new(stream( manager diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 7203cff..d2aa08f 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -1311,6 +1311,39 @@ async fn attach_ws_streams_pty_bytes() { assert!(saw_hello, "PTY did not echo through the attach WS"); } +#[tokio::test] +async fn pty_routes_reject_structured_sessions_as_not_terminal() { + let h = spawn("naps-laptop").await; + let fake = FakeAcp::new(); + let id = structured_session(&h, &fake).await; + + let response = reqwest::Client::new() + .post(url(h.addr, &format!("/api/sessions/{id}/input"))) + .json(&maestro_protocol::SendInputRequest { + text: "ignored".into(), + submit: false, + }) + .send() + .await + .unwrap(); + assert_eq!(response.status(), reqwest::StatusCode::CONFLICT); + let error: ApiError = response.json().await.unwrap(); + assert_eq!(error.error.code, "not_terminal"); + assert_eq!(error.error.message, "session is not a terminal session"); + + let error = connect_async(ws_url(h.addr, &format!("/api/attach/{id}"))) + .await + .expect_err("structured session must not upgrade to a PTY socket"); + let tokio_tungstenite::tungstenite::Error::Http(response) = error else { + panic!("expected an HTTP handshake rejection, got {error}"); + }; + assert_eq!(response.status(), reqwest::StatusCode::CONFLICT); + let body = response.body().as_deref().expect("ApiError response body"); + let error: ApiError = serde_json::from_slice(body).unwrap(); + assert_eq!(error.error.code, "not_terminal"); + assert_eq!(error.error.message, "session is not a terminal session"); +} + #[tokio::test] async fn attach_ws_for_unknown_session_returns_404() { let h = spawn("naps-laptop").await;