| 1 | //! Per-session control socket — the supervised-operation control surface. |
| 2 | //! |
| 3 | //! This module is the codewhale side of "session control/communication API |
| 4 | //! for supervised operation" (#5533). When the |
| 5 | //! `[control_socket]` config table sets `enabled = true`, the interactive |
| 6 | //! TUI binds one unix domain socket per *running* session at |
| 7 | //! |
| 8 | //! ```text |
| 9 | //! <sessions-dir>/<session-id>/control.sock (mode 0600) |
| 10 | //! ``` |
| 11 | //! |
| 12 | //! where `<sessions-dir>` is the same directory the session store uses |
| 13 | //! (`SessionManager::sessions_dir`, typically `~/.codewhale/sessions`) and |
| 14 | //! `<session-id>` is the session the TUI currently owns. The socket lives |
| 15 | //! inside the per-session artifact directory, so `delete_session` and the |
| 16 | //! orphan-reclaim sweep remove it together with the rest of the session's |
| 17 | //! artifacts, and a crashed process leaves at most a stale socket file that |
| 18 | //! the next bind takes over (connect-probe + unlink, a known-good |
| 19 | //! socket-ownership pattern). |
| 20 | //! |
| 21 | //! # Transport |
| 22 | //! |
| 23 | //! Newline-framed JSON-RPC, one request per connection: connect, write one |
| 24 | //! request line, read one response line, close. Requests: |
| 25 | //! |
| 26 | //! ```json |
| 27 | //! {"id":"1","method":"message","params":{"text":"hello"}} |
| 28 | //! {"id":"2","method":"interrupt","params":{}} |
| 29 | //! {"id":"3","method":"relaunch","params":{}} |
| 30 | //! {"id":"4","method":"status","params":{}} |
| 31 | //! ``` |
| 32 | //! |
| 33 | //! Success responses echo the id and carry a `type`-tagged result: |
| 34 | //! |
| 35 | //! ```json |
| 36 | //! {"id":"1","result":{"type":"message_sent","delivery":"dispatched"}} |
| 37 | //! {"id":"2","result":{"type":"interrupted","cancelled":true}} |
| 38 | //! {"id":"3","result":{"type":"relaunching"}} |
| 39 | //! {"id":"4","result":{"type":"status","turn_state":"idle","goal":{"objective":null,"status":"active","paused":false}}} |
| 40 | //! ``` |
| 41 | //! |
| 42 | //! Failures are `{"id":…,"error":{"code":…,"message":…}}` with codes |
| 43 | //! `invalid_request`, `command_error`, `timeout`, and `server_unavailable`. |
| 44 | //! |
| 45 | //! # Verbs |
| 46 | //! |
| 47 | //! - `message` — delivers `text` as a structured user message through the |
| 48 | //! ordinary composer dispatch path (`dispatch_composer_message`): dispatched |
| 49 | //! immediately when the app is idle, queued when a turn is in flight |
| 50 | //! (queued delivery is the default under load, matching the supervisor |
| 51 | //! contract). The response's `delivery` field reports which happened. |
| 52 | //! - `interrupt` — the exact Esc-shaped "cancel the active turn" body |
| 53 | //! (`escape_cancel_request`), shared with the Esc key path so the two |
| 54 | //! cannot drift. `cancelled` reports whether active work was in flight. |
| 55 | //! - `relaunch` — routed through the slash-command path |
| 56 | //! (`crate::commands::execute("/relaunch", app)`): **no relaunch logic |
| 57 | //! lives here**. The `/relaunch` command is built on the |
| 58 | //! `pr/relaunch-command` branch; this verb is the seam that calls the same |
| 59 | //! command the user's `/relaunch` would. Until that command lands, the |
| 60 | //! verb reports the command's own "unknown command" error verbatim, and |
| 61 | //! once it lands the verb inherits its save-and-quit handoff with no |
| 62 | //! changes here. |
| 63 | //! - `status` — answered by the socket thread directly from a snapshot the |
| 64 | //! event loop republishes every iteration: `turn_state` |
| 65 | //! (`idle | in_progress | waiting`) and `goal` |
| 66 | //! (`objective`, `status`, `paused`). |
| 67 | //! |
| 68 | //! # Wiring (insertion points) |
| 69 | //! |
| 70 | //! 1. `run_event_loop` (crates/tui/src/tui/ui/event_loop.rs) constructs a |
| 71 | //! [`SessionControl`] once and, at the top of the frame loop, calls |
| 72 | //! [`SessionControl::reconcile`] (bind/rebind/unbind when the owned |
| 73 | //! session id changes), [`SessionControl::update_status`] (publish the |
| 74 | //! snapshot for `status`), and [`SessionControl::drain`] (execute queued |
| 75 | //! verbs on the UI thread; a `true` return asks the loop to quit, which |
| 76 | //! is how `relaunch` reuses the ordinary `/exit` teardown). |
| 77 | //! 2. The socket runs on background threads; verbs that touch UI state cross |
| 78 | //! to the event loop over an mpsc channel and answer over a response |
| 79 | //! channel with a 5 s timeout (a dispatch-to-app pattern). |
| 80 | //! |
| 81 | //! The feature is off unless `[control_socket] enabled = true`; an unset |
| 82 | //! table changes nothing. Unix-only: on non-unix platforms the config key |
| 83 | //! parses but binding is refused at runtime. |
| 84 | |
| 85 | use std::io; |
| 86 | #[cfg(unix)] |
| 87 | use std::io::{BufRead, BufReader, Read, Write}; |
| 88 | use std::path::{Path, PathBuf}; |
| 89 | #[cfg(unix)] |
| 90 | use std::sync::atomic::{AtomicBool, Ordering}; |
| 91 | use std::sync::{Arc, Mutex, mpsc}; |
| 92 | use std::time::{Duration, Instant}; |
| 93 | |
| 94 | #[cfg(unix)] |
| 95 | use std::fs; |
| 96 | #[cfg(unix)] |
| 97 | use std::os::unix::fs::{FileTypeExt, PermissionsExt}; |
| 98 | #[cfg(unix)] |
| 99 | use std::os::unix::net::{UnixListener, UnixStream}; |
| 100 | #[cfg(unix)] |
| 101 | use std::thread; |
| 102 | |
| 103 | use serde::{Deserialize, Serialize}; |
| 104 | |
| 105 | use crate::tui::app::{App, AppAction, ComposerSubmitAction, QueuedMessage, SubmitDisposition}; |
| 106 | use crate::tui::streaming::StreamDisplayClock; |
| 107 | use crate::tui::ui::{DispatchRecovery, dispatch_composer_message, escape_cancel_request}; |
| 108 | |
| 109 | /// Socket file name inside the per-session artifact directory. |
| 110 | pub(crate) const SOCKET_FILE_NAME: &str = "control.sock"; |
| 111 | |
| 112 | /// Hard cap on one request line (initial-request bound). |
| 113 | #[cfg(unix)] |
| 114 | const MAX_REQUEST_BYTES: usize = 1024 * 1024; |
| 115 | |
| 116 | /// Accept-loop poll interval while the listener is idle. |
| 117 | #[cfg(unix)] |
| 118 | const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100); |
| 119 | |
| 120 | /// A client that connects and never sends is dropped after this long. |
| 121 | #[cfg(unix)] |
| 122 | const REQUEST_READ_TIMEOUT: Duration = Duration::from_secs(5); |
| 123 | |
| 124 | /// Response writes give up after this long rather than blocking forever. |
| 125 | #[cfg(unix)] |
| 126 | const RESPONSE_WRITE_TIMEOUT: Duration = Duration::from_secs(5); |
| 127 | |
| 128 | /// How long a verb may wait for the event loop to handle it |
| 129 | /// (`APP_RESPONSE_TIMEOUT`). |
| 130 | #[cfg(unix)] |
| 131 | const DISPATCH_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); |
| 132 | |
| 133 | /// Minimum pause between bind retries after a refused takeover, so another |
| 134 | /// live process holding the socket cannot turn the per-frame reconcile into |
| 135 | /// a connect-probe and warn-log flood. Shortened under `#[cfg(test)]` so the |
| 136 | /// backoff itself is testable without sleeping for seconds. |
| 137 | #[cfg(not(test))] |
| 138 | const BIND_RETRY_BACKOFF: Duration = Duration::from_secs(5); |
| 139 | #[cfg(test)] |
| 140 | const BIND_RETRY_BACKOFF: Duration = Duration::from_millis(200); |
| 141 | |
| 142 | // ── Protocol ──────────────────────────────────────────────────────────────── |
| 143 | |
| 144 | /// One request line: `{"id": … , "method": … , "params": …}`. |
| 145 | /// Windows builds construct this type only in the portable protocol tests; |
| 146 | /// the plain Windows lib build leaves it unreachable, so the lint allowance |
| 147 | /// below is scoped to exactly that case (unix builds use it via the socket |
| 148 | /// runtime, and CI denies dead code on the MSVC test gate). |
| 149 | #[cfg_attr(not(unix), allow(dead_code))] |
| 150 | #[derive(Debug, Deserialize)] |
| 151 | struct Request { |
| 152 | id: String, |
| 153 | #[serde(flatten)] |
| 154 | method: Method, |
| 155 | } |
| 156 | |
| 157 | #[cfg_attr(not(unix), allow(dead_code))] |
| 158 | #[derive(Debug, Deserialize)] |
| 159 | #[serde(tag = "method", content = "params", rename_all = "snake_case")] |
| 160 | enum Method { |
| 161 | Message(MessageParams), |
| 162 | Interrupt(EmptyParams), |
| 163 | Relaunch(EmptyParams), |
| 164 | Status(EmptyParams), |
| 165 | } |
| 166 | |
| 167 | #[cfg_attr(not(unix), allow(dead_code))] |
| 168 | #[derive(Debug, Deserialize)] |
| 169 | struct MessageParams { |
| 170 | text: String, |
| 171 | } |
| 172 | |
| 173 | #[cfg_attr(not(unix), allow(dead_code))] |
| 174 | #[derive(Debug, Deserialize)] |
| 175 | struct EmptyParams {} |
| 176 | |
| 177 | /// A verb the socket thread hands to the event loop, plus the way back. |
| 178 | #[derive(Debug)] |
| 179 | pub(crate) struct PendingCommand { |
| 180 | pub(crate) id: String, |
| 181 | pub(crate) command: ControlCommand, |
| 182 | pub(crate) respond_to: mpsc::Sender<String>, |
| 183 | } |
| 184 | |
| 185 | #[cfg_attr(not(unix), allow(dead_code))] |
| 186 | #[derive(Debug)] |
| 187 | pub(crate) enum ControlCommand { |
| 188 | Message { text: String }, |
| 189 | Interrupt, |
| 190 | Relaunch, |
| 191 | } |
| 192 | |
| 193 | #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] |
| 194 | #[serde(rename_all = "snake_case")] |
| 195 | pub(crate) enum TurnState { |
| 196 | Idle, |
| 197 | InProgress, |
| 198 | Waiting, |
| 199 | } |
| 200 | |
| 201 | /// Goal state visible to supervisors over the `status` verb. |
| 202 | #[derive(Debug, Clone, PartialEq, Eq, Serialize)] |
| 203 | pub(crate) struct GoalSnapshot { |
| 204 | pub(crate) objective: Option<String>, |
| 205 | pub(crate) status: String, |
| 206 | pub(crate) paused: bool, |
| 207 | } |
| 208 | |
| 209 | /// The `status` answer, republished by the event loop every iteration. |
| 210 | #[derive(Debug, Clone, PartialEq, Eq)] |
| 211 | pub(crate) struct StatusSnapshot { |
| 212 | pub(crate) turn_state: TurnState, |
| 213 | pub(crate) goal: GoalSnapshot, |
| 214 | } |
| 215 | |
| 216 | /// Success envelope (`SuccessResponse` shape). |
| 217 | #[derive(Debug, Serialize)] |
| 218 | struct SuccessResponse { |
| 219 | id: String, |
| 220 | result: ResponseResult, |
| 221 | } |
| 222 | |
| 223 | #[cfg_attr(not(unix), allow(dead_code))] |
| 224 | #[derive(Debug, Serialize)] |
| 225 | #[serde(tag = "type", rename_all = "snake_case")] |
| 226 | enum ResponseResult { |
| 227 | MessageSent { |
| 228 | delivery: &'static str, |
| 229 | }, |
| 230 | Interrupted { |
| 231 | cancelled: bool, |
| 232 | }, |
| 233 | Relaunching, |
| 234 | Status { |
| 235 | turn_state: TurnState, |
| 236 | goal: GoalSnapshot, |
| 237 | }, |
| 238 | } |
| 239 | |
| 240 | /// Error envelope (`ErrorResponse` shape). |
| 241 | #[derive(Debug, Serialize)] |
| 242 | struct ErrorResponse { |
| 243 | id: String, |
| 244 | error: ErrorBody, |
| 245 | } |
| 246 | |
| 247 | #[derive(Debug, Serialize)] |
| 248 | struct ErrorBody { |
| 249 | code: &'static str, |
| 250 | message: String, |
| 251 | } |
| 252 | |
| 253 | fn response_ok(id: String, result: ResponseResult) -> String { |
| 254 | serde_json::to_string(&SuccessResponse { id, result }).unwrap_or_else(|_| { |
| 255 | r#"{"id":"","error":{"code":"internal_error","message":"failed to encode response"}}"# |
| 256 | .to_string() |
| 257 | }) |
| 258 | } |
| 259 | |
| 260 | fn response_error(id: &str, code: &'static str, message: String) -> String { |
| 261 | serde_json::to_string(&ErrorResponse { |
| 262 | id: id.to_string(), |
| 263 | error: ErrorBody { code, message }, |
| 264 | }) |
| 265 | .unwrap_or_else(|_| { |
| 266 | r#"{"id":"","error":{"code":"internal_error","message":"failed to encode response"}}"# |
| 267 | .to_string() |
| 268 | }) |
| 269 | } |
| 270 | |
| 271 | // ── UI-side handle ────────────────────────────────────────────────────────── |
| 272 | |
| 273 | /// The event-loop side of the control surface. Cheap to poll every frame: |
| 274 | /// reconcile/update/drain are all no-ops (or near no-ops) when disabled. |
| 275 | pub(crate) struct SessionControl { |
| 276 | enabled: bool, |
| 277 | sessions_dir: Option<PathBuf>, |
| 278 | bound_session: Option<String>, |
| 279 | socket: Option<ControlSocketHandle>, |
| 280 | commands_tx: Option<mpsc::Sender<PendingCommand>>, |
| 281 | commands_rx: mpsc::Receiver<PendingCommand>, |
| 282 | status: Arc<Mutex<StatusSnapshot>>, |
| 283 | /// When the last bind attempt failed (e.g. another live process owns the |
| 284 | /// socket), retries for *that session* back off so a refused takeover |
| 285 | /// cannot become a per-frame connect-probe and log flood. |
| 286 | last_bind_failure: Option<(String, Instant)>, |
| 287 | } |
| 288 | |
| 289 | impl SessionControl { |
| 290 | pub(crate) fn new(enabled: bool) -> Self { |
| 291 | Self::new_with_sessions_dir(enabled, None) |
| 292 | } |
| 293 | |
| 294 | /// Test seam: `sessions_dir` bypasses `SessionManager::default_location()` |
| 295 | /// so tests never touch the real `~/.codewhale/sessions`. |
| 296 | fn new_with_sessions_dir(enabled: bool, sessions_dir: Option<PathBuf>) -> Self { |
| 297 | let (commands_tx, commands_rx) = mpsc::channel(); |
| 298 | Self { |
| 299 | enabled, |
| 300 | sessions_dir, |
| 301 | bound_session: None, |
| 302 | socket: None, |
| 303 | commands_tx: enabled.then_some(commands_tx), |
| 304 | commands_rx, |
| 305 | status: Arc::new(Mutex::new(StatusSnapshot { |
| 306 | turn_state: TurnState::Idle, |
| 307 | goal: GoalSnapshot { |
| 308 | objective: None, |
| 309 | status: "active".to_string(), |
| 310 | paused: false, |
| 311 | }, |
| 312 | })), |
| 313 | last_bind_failure: None, |
| 314 | } |
| 315 | } |
| 316 | |
| 317 | /// Bind/rebind the socket when the owned session id appears or changes, |
| 318 | /// and unbind when it disappears (session switch or teardown). Runs on |
| 319 | /// the event-loop thread but only spawns a thread on an actual change. |
| 320 | pub(crate) fn reconcile(&mut self, current_session_id: Option<&str>) { |
| 321 | if !self.enabled { |
| 322 | return; |
| 323 | } |
| 324 | let Some(id) = current_session_id |
| 325 | .map(str::trim) |
| 326 | .filter(|id| !id.is_empty()) |
| 327 | else { |
| 328 | // No session yet (fresh session before the first snapshot) or |
| 329 | // the id went away: release whatever we hold. |
| 330 | self.socket = None; |
| 331 | self.bound_session = None; |
| 332 | return; |
| 333 | }; |
| 334 | if self.bound_session.as_deref() == Some(id) { |
| 335 | return; |
| 336 | } |
| 337 | // A refused takeover must not retry every frame: back off so the |
| 338 | // connect probe and its warning log run at most every few seconds. |
| 339 | // Keyed on the session id so switching sessions is never delayed by |
| 340 | // another session's refusal. |
| 341 | if let Some((failed_id, failed_at)) = &self.last_bind_failure |
| 342 | && failed_id == id |
| 343 | && failed_at.elapsed() < BIND_RETRY_BACKOFF |
| 344 | { |
| 345 | return; |
| 346 | } |
| 347 | // Session id changed: drop the old listener first so the socket file |
| 348 | // is unlinked before the new one binds. |
| 349 | self.socket = None; |
| 350 | self.bound_session = None; |
| 351 | |
| 352 | let sessions_dir = match self.sessions_dir.clone() { |
| 353 | Some(dir) => dir, |
| 354 | None => { |
| 355 | let manager = match crate::session_manager::SessionManager::default_location() { |
| 356 | Ok(manager) => manager, |
| 357 | Err(error) => { |
| 358 | tracing::warn!(%error, "control socket: cannot resolve the sessions directory"); |
| 359 | return; |
| 360 | } |
| 361 | }; |
| 362 | let dir = manager.sessions_dir().to_path_buf(); |
| 363 | self.sessions_dir = Some(dir.clone()); |
| 364 | dir |
| 365 | } |
| 366 | }; |
| 367 | let Some(commands_tx) = self.commands_tx.clone() else { |
| 368 | return; |
| 369 | }; |
| 370 | match bind_control_socket(&sessions_dir, id, commands_tx, Arc::clone(&self.status)) { |
| 371 | Ok(handle) => { |
| 372 | tracing::info!( |
| 373 | session = id, |
| 374 | path = %sessions_dir.join(id).join(SOCKET_FILE_NAME).display(), |
| 375 | "control socket listening" |
| 376 | ); |
| 377 | self.bound_session = Some(id.to_string()); |
| 378 | self.socket = Some(handle); |
| 379 | self.last_bind_failure = None; |
| 380 | } |
| 381 | Err(error) => { |
| 382 | tracing::warn!(session = id, %error, "control socket: bind failed; session control unavailable"); |
| 383 | self.last_bind_failure = Some((id.to_string(), Instant::now())); |
| 384 | } |
| 385 | } |
| 386 | } |
| 387 | |
| 388 | /// Republish the `status` snapshot from the current app state. Runs every |
| 389 | /// frame; the mutex write happens only when something actually changed. |
| 390 | pub(crate) fn update_status(&self, app: &App) { |
| 391 | if !self.enabled { |
| 392 | return; |
| 393 | } |
| 394 | let snapshot = snapshot_from_app(app); |
| 395 | let Ok(mut guard) = self.status.try_lock() else { |
| 396 | return; // the socket thread is answering a `status` request; skip a frame |
| 397 | }; |
| 398 | if *guard != snapshot { |
| 399 | *guard = snapshot; |
| 400 | } |
| 401 | } |
| 402 | |
| 403 | /// Execute verbs queued by the socket thread on the UI thread and answer |
| 404 | /// their clients. Returns `true` when a verb requested app quit (the |
| 405 | /// `relaunch` seam) — the caller returns from the event loop and reuses |
| 406 | /// the ordinary `/exit` teardown path. |
| 407 | pub(crate) async fn drain( |
| 408 | &mut self, |
| 409 | app: &mut App, |
| 410 | config: &crate::config::Config, |
| 411 | engine_handle: &crate::core::engine::EngineHandle, |
| 412 | current_streaming_text: &mut String, |
| 413 | stream_display_clock: &mut StreamDisplayClock, |
| 414 | ) -> bool { |
| 415 | if !self.enabled { |
| 416 | return false; |
| 417 | } |
| 418 | let mut quit = false; |
| 419 | while let Ok(pending) = self.commands_rx.try_recv() { |
| 420 | let (do_quit, response) = execute_command( |
| 421 | app, |
| 422 | config, |
| 423 | engine_handle, |
| 424 | current_streaming_text, |
| 425 | stream_display_clock, |
| 426 | pending.id.clone(), |
| 427 | pending.command, |
| 428 | ) |
| 429 | .await; |
| 430 | // The client may have disconnected while we worked; that must |
| 431 | // never fail the loop. |
| 432 | let _ = pending.respond_to.send(response); |
| 433 | quit |= do_quit; |
| 434 | } |
| 435 | quit |
| 436 | } |
| 437 | } |
| 438 | |
| 439 | /// The session's coarse turn state, shared by the control-socket `status` |
| 440 | /// answer and the session-state hook transitions (#6004). `Waiting` covers |
| 441 | /// every wait on the person — an open approval prompt, a presented |
| 442 | /// `request_user_input` question, or a parked goal continuation — not only |
| 443 | /// the continuation wait it used to map. |
| 444 | pub(crate) fn turn_state_from_app(app: &App) -> TurnState { |
| 445 | if app.goal_continuation_waiting |
| 446 | || app.pending_user_input_prompt.is_some() |
| 447 | || app.view_stack.top_kind() == Some(crate::tui::views::ModalKind::Approval) |
| 448 | { |
| 449 | return TurnState::Waiting; |
| 450 | } |
| 451 | if app.is_loading || matches!(app.runtime_turn_status.as_deref(), Some("in_progress")) { |
| 452 | return TurnState::InProgress; |
| 453 | } |
| 454 | TurnState::Idle |
| 455 | } |
| 456 | |
| 457 | fn snapshot_from_app(app: &App) -> StatusSnapshot { |
| 458 | let turn_state = turn_state_from_app(app); |
| 459 | // A paused goal parks its objective in `paused_goal_objective`, so the |
| 460 | // snapshot surfaces the objective that is actually in flight. |
| 461 | let objective = app |
| 462 | .goal |
| 463 | .objective |
| 464 | .clone() |
| 465 | .or_else(|| app.paused_goal_objective.clone()); |
| 466 | StatusSnapshot { |
| 467 | turn_state, |
| 468 | goal: GoalSnapshot { |
| 469 | objective, |
| 470 | status: app.goal.status.as_str().to_string(), |
| 471 | paused: app.paused || app.paused_goal_objective.is_some(), |
| 472 | }, |
| 473 | } |
| 474 | } |
| 475 | |
| 476 | /// Execute one verb on the UI thread. Returns `(quit, response_json)`. |
| 477 | async fn execute_command( |
| 478 | app: &mut App, |
| 479 | config: &crate::config::Config, |
| 480 | engine_handle: &crate::core::engine::EngineHandle, |
| 481 | current_streaming_text: &mut String, |
| 482 | stream_display_clock: &mut StreamDisplayClock, |
| 483 | id: String, |
| 484 | command: ControlCommand, |
| 485 | ) -> (bool, String) { |
| 486 | match command { |
| 487 | ControlCommand::Message { text } => { |
| 488 | if text.trim().is_empty() { |
| 489 | return ( |
| 490 | false, |
| 491 | response_error( |
| 492 | &id, |
| 493 | "invalid_request", |
| 494 | "message text must not be empty".to_string(), |
| 495 | ), |
| 496 | ); |
| 497 | } |
| 498 | // Queued delivery is the default under load: while a turn is in |
| 499 | // flight the message waits like any queued follow-up; an idle |
| 500 | // app dispatches immediately. |
| 501 | let busy = |
| 502 | app.is_loading || matches!(app.runtime_turn_status.as_deref(), Some("in_progress")); |
| 503 | let disposition = if busy { |
| 504 | SubmitDisposition::Queue |
| 505 | } else { |
| 506 | SubmitDisposition::Immediate |
| 507 | }; |
| 508 | let message = QueuedMessage::new(text, None); |
| 509 | // Delivery failures surface through the app's own status/toast |
| 510 | // and recovery paths; the verb still answers with what it asked |
| 511 | // for (dispatched vs queued). |
| 512 | let _ = dispatch_composer_message( |
| 513 | app, |
| 514 | config, |
| 515 | engine_handle, |
| 516 | message, |
| 517 | DispatchRecovery::Immediate, |
| 518 | ComposerSubmitAction::Submit(disposition), |
| 519 | ) |
| 520 | .await; |
| 521 | app.needs_redraw = true; |
| 522 | let delivery = if busy { "queued" } else { "dispatched" }; |
| 523 | ( |
| 524 | false, |
| 525 | response_ok(id, ResponseResult::MessageSent { delivery }), |
| 526 | ) |
| 527 | } |
| 528 | ControlCommand::Interrupt => { |
| 529 | let had_active_work = app.is_loading |
| 530 | || app.is_compacting |
| 531 | || app.manual_compaction_queued |
| 532 | || app.goal_continuation_waiting |
| 533 | || app.paused |
| 534 | || app.paused_goal_objective.is_some() |
| 535 | || matches!(app.runtime_turn_status.as_deref(), Some("in_progress")); |
| 536 | if !had_active_work { |
| 537 | // Nothing Esc-cancel would cancel: quiet no-op, like an Esc |
| 538 | // on an idle app that has nothing else to act on. |
| 539 | return ( |
| 540 | false, |
| 541 | response_ok(id, ResponseResult::Interrupted { cancelled: false }), |
| 542 | ); |
| 543 | } |
| 544 | let _ = escape_cancel_request( |
| 545 | app, |
| 546 | engine_handle, |
| 547 | current_streaming_text, |
| 548 | stream_display_clock, |
| 549 | ); |
| 550 | app.needs_redraw = true; |
| 551 | ( |
| 552 | false, |
| 553 | response_ok(id, ResponseResult::Interrupted { cancelled: true }), |
| 554 | ) |
| 555 | } |
| 556 | ControlCommand::Relaunch => { |
| 557 | // Seam: the exact same command path `/relaunch` uses. When the |
| 558 | // /relaunch command lands (pr/relaunch-command), this returns its |
| 559 | // save-and-quit action and the quit flag below reuses the /exit |
| 560 | // teardown; until then the command's own error is reported. |
| 561 | let result = crate::commands::execute("/relaunch", app); |
| 562 | if result.is_error { |
| 563 | return ( |
| 564 | false, |
| 565 | response_error( |
| 566 | &id, |
| 567 | "command_error", |
| 568 | result |
| 569 | .message |
| 570 | .unwrap_or_else(|| "relaunch failed".to_string()), |
| 571 | ), |
| 572 | ); |
| 573 | } |
| 574 | let quit = matches!(result.action, Some(AppAction::Quit)); |
| 575 | app.needs_redraw = true; |
| 576 | (quit, response_ok(id, ResponseResult::Relaunching)) |
| 577 | } |
| 578 | } |
| 579 | } |
| 580 | |
| 581 | // ── Socket server (unix only) ─────────────────────────────────────────────── |
| 582 | |
| 583 | /// Bound listener + its accept thread. Dropping unbinds: the accept thread |
| 584 | /// stops within one poll interval, the socket file is unlinked if this |
| 585 | /// process still owns it, and in-flight connections finish on their own. |
| 586 | #[cfg(unix)] |
| 587 | pub(crate) struct ControlSocketHandle { |
| 588 | stop: Arc<AtomicBool>, |
| 589 | thread: Option<thread::JoinHandle<()>>, |
| 590 | } |
| 591 | |
| 592 | #[cfg(unix)] |
| 593 | impl Drop for ControlSocketHandle { |
| 594 | fn drop(&mut self) { |
| 595 | self.stop.store(true, Ordering::Release); |
| 596 | if let Some(thread) = self.thread.take() { |
| 597 | // The accept loop polls at CONNECTION_POLL_INTERVAL and never |
| 598 | // blocks on a connection (each connection has its own thread), |
| 599 | // so this join is bounded and cannot deadlock. |
| 600 | let _ = thread.join(); |
| 601 | } |
| 602 | } |
| 603 | } |
| 604 | |
| 605 | #[cfg(unix)] |
| 606 | impl std::fmt::Debug for ControlSocketHandle { |
| 607 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 608 | f.debug_struct("ControlSocketHandle") |
| 609 | .field("stopped", &self.stop.load(Ordering::Relaxed)) |
| 610 | .finish_non_exhaustive() |
| 611 | } |
| 612 | } |
| 613 | |
| 614 | #[cfg(not(unix))] |
| 615 | #[derive(Debug)] |
| 616 | #[allow(dead_code)] // kept so the SessionControl field type is portable |
| 617 | pub(crate) struct ControlSocketHandle; |
| 618 | |
| 619 | /// Bind `<sessions-dir>/<session-id>/control.sock` (0600) and serve it. |
| 620 | /// Refused when another live process already serves that path; a stale file |
| 621 | /// (crash leftover, nothing answering) is taken over. |
| 622 | #[cfg(unix)] |
| 623 | pub(crate) fn bind_control_socket( |
| 624 | sessions_dir: &Path, |
| 625 | session_id: &str, |
| 626 | commands_tx: mpsc::Sender<PendingCommand>, |
| 627 | status: Arc<Mutex<StatusSnapshot>>, |
| 628 | ) -> io::Result<ControlSocketHandle> { |
| 629 | let session_dir = sessions_dir.join(session_id); |
| 630 | fs::create_dir_all(&session_dir)?; |
| 631 | let path = session_dir.join(SOCKET_FILE_NAME); |
| 632 | prepare_socket_path(&path)?; |
| 633 | |
| 634 | let listener = UnixListener::bind(&path)?; |
| 635 | let identity = socket_file_identity(&path); |
| 636 | fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?; |
| 637 | listener.set_nonblocking(true)?; |
| 638 | |
| 639 | let stop = Arc::new(AtomicBool::new(false)); |
| 640 | let thread_stop = Arc::clone(&stop); |
| 641 | let thread = thread::Builder::new() |
| 642 | .name(format!("codewhale-control-{session_id}")) |
| 643 | .spawn(move || serve(listener, path, identity, thread_stop, commands_tx, status))?; |
| 644 | |
| 645 | Ok(ControlSocketHandle { |
| 646 | stop, |
| 647 | thread: Some(thread), |
| 648 | }) |
| 649 | } |
| 650 | |
| 651 | #[cfg(not(unix))] |
| 652 | pub(crate) fn bind_control_socket( |
| 653 | _sessions_dir: &Path, |
| 654 | _session_id: &str, |
| 655 | _commands_tx: mpsc::Sender<PendingCommand>, |
| 656 | _status: Arc<Mutex<StatusSnapshot>>, |
| 657 | ) -> io::Result<ControlSocketHandle> { |
| 658 | Err(io::Error::new( |
| 659 | io::ErrorKind::Unsupported, |
| 660 | "the per-session control socket is unix-only", |
| 661 | )) |
| 662 | } |
| 663 | |
| 664 | /// Take over the socket path, or refuse when a live server already holds it. |
| 665 | #[cfg(unix)] |
| 666 | fn prepare_socket_path(path: &Path) -> io::Result<()> { |
| 667 | match fs::symlink_metadata(path) { |
| 668 | Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), |
| 669 | Err(error) => Err(error), |
| 670 | Ok(metadata) => { |
| 671 | if !metadata.file_type().is_socket() { |
| 672 | // A plain file (or directory) in the way: not ours to keep. |
| 673 | fs::remove_file(path)?; |
| 674 | return Ok(()); |
| 675 | } |
| 676 | match UnixStream::connect(path) { |
| 677 | // Someone answers: a live process owns this session's socket. |
| 678 | // Do not steal it (a "socket busy" refusal). |
| 679 | Ok(_) => Err(io::Error::new( |
| 680 | io::ErrorKind::AddrInUse, |
| 681 | format!("control socket already live at {}", path.display()), |
| 682 | )), |
| 683 | // Stale: the file exists but nothing listens. Take over. |
| 684 | Err(_) => { |
| 685 | fs::remove_file(path)?; |
| 686 | Ok(()) |
| 687 | } |
| 688 | } |
| 689 | } |
| 690 | } |
| 691 | } |
| 692 | |
| 693 | /// (device, inode) so an unlink never removes a file this process did not bind. |
| 694 | #[cfg(unix)] |
| 695 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 696 | struct SocketFileIdentity { |
| 697 | dev: u64, |
| 698 | ino: u64, |
| 699 | } |
| 700 | |
| 701 | #[cfg(unix)] |
| 702 | fn socket_file_identity(path: &Path) -> Option<SocketFileIdentity> { |
| 703 | let metadata = fs::metadata(path).ok()?; |
| 704 | use std::os::unix::fs::MetadataExt; |
| 705 | Some(SocketFileIdentity { |
| 706 | dev: metadata.dev(), |
| 707 | ino: metadata.ino(), |
| 708 | }) |
| 709 | } |
| 710 | |
| 711 | #[cfg(unix)] |
| 712 | fn serve( |
| 713 | listener: UnixListener, |
| 714 | path: PathBuf, |
| 715 | identity: Option<SocketFileIdentity>, |
| 716 | stop: Arc<AtomicBool>, |
| 717 | commands_tx: mpsc::Sender<PendingCommand>, |
| 718 | status: Arc<Mutex<StatusSnapshot>>, |
| 719 | ) { |
| 720 | while !stop.load(Ordering::Acquire) { |
| 721 | match listener.accept() { |
| 722 | Ok((stream, _)) => { |
| 723 | // The listener is nonblocking, and on BSD-family platforms |
| 724 | // (macOS, FreeBSD) an accepted socket *inherits* O_NONBLOCK |
| 725 | // from the listener — Linux does not. The per-connection |
| 726 | // handler expects blocking reads/writes (bounded by request |
| 727 | // caps and timeouts), so make that explicit: without it, a |
| 728 | // read on macOS returns EAGAIN mid-frame on a large request |
| 729 | // and the connection dies with a broken pipe on the client. |
| 730 | let _ = stream.set_nonblocking(false); |
| 731 | // One thread per connection: a silent client |
| 732 | // must not stall other clients or the stop check. |
| 733 | let tx = commands_tx.clone(); |
| 734 | let status = Arc::clone(&status); |
| 735 | let _ = thread::Builder::new() |
| 736 | .name("codewhale-control-conn".to_string()) |
| 737 | .spawn(move || handle_connection(stream, &tx, &status)); |
| 738 | } |
| 739 | Err(error) if error.kind() == io::ErrorKind::WouldBlock => { |
| 740 | thread::sleep(CONNECTION_POLL_INTERVAL); |
| 741 | } |
| 742 | Err(error) if error.kind() == io::ErrorKind::Interrupted => {} |
| 743 | Err(error) => { |
| 744 | // Listener gone (e.g. the session dir was removed out from |
| 745 | // under us) — stop serving; connections fail to connect from |
| 746 | // here on, which is the honest state. |
| 747 | tracing::debug!(%error, "control socket listener closed"); |
| 748 | break; |
| 749 | } |
| 750 | } |
| 751 | } |
| 752 | if let Some(identity) = identity |
| 753 | && socket_file_identity(&path) == Some(identity) |
| 754 | { |
| 755 | let _ = fs::remove_file(&path); |
| 756 | } |
| 757 | } |
| 758 | |
| 759 | /// Serve exactly one request: read one bounded line, answer, close. |
| 760 | #[cfg(unix)] |
| 761 | fn handle_connection( |
| 762 | stream: UnixStream, |
| 763 | commands_tx: &mpsc::Sender<PendingCommand>, |
| 764 | status: &Arc<Mutex<StatusSnapshot>>, |
| 765 | ) { |
| 766 | let _ = stream.set_read_timeout(Some(REQUEST_READ_TIMEOUT)); |
| 767 | let _ = stream.set_write_timeout(Some(RESPONSE_WRITE_TIMEOUT)); |
| 768 | let mut stream = BufReader::new(stream); |
| 769 | |
| 770 | let line = match read_request_line(&mut stream) { |
| 771 | Ok(Some(line)) => line, |
| 772 | Ok(None) => return, // EOF, empty frame, or timeout: close silently |
| 773 | Err(error) if error.kind() == io::ErrorKind::InvalidData => { |
| 774 | // Oversized frame: the reader drained it, so the client can |
| 775 | // finish writing and read this rejection. |
| 776 | let response = response_error("", "invalid_request", error.to_string()); |
| 777 | write_response_line(stream.get_mut(), &response); |
| 778 | return; |
| 779 | } |
| 780 | Err(_) => return, |
| 781 | }; |
| 782 | let trimmed = line.trim(); |
| 783 | if trimmed.is_empty() { |
| 784 | return; |
| 785 | } |
| 786 | |
| 787 | let request: Request = match serde_json::from_str(trimmed) { |
| 788 | Ok(request) => request, |
| 789 | Err(error) => { |
| 790 | let response = |
| 791 | response_error("", "invalid_request", format!("invalid request: {error}")); |
| 792 | write_response_line(stream.get_mut(), &response); |
| 793 | return; |
| 794 | } |
| 795 | }; |
| 796 | |
| 797 | let response = match request.method { |
| 798 | Method::Status(_) => { |
| 799 | let snapshot = status |
| 800 | .lock() |
| 801 | .unwrap_or_else(|poisoned| poisoned.into_inner()) |
| 802 | .clone(); |
| 803 | response_ok( |
| 804 | request.id, |
| 805 | ResponseResult::Status { |
| 806 | turn_state: snapshot.turn_state, |
| 807 | goal: snapshot.goal, |
| 808 | }, |
| 809 | ) |
| 810 | } |
| 811 | Method::Message(params) => dispatch_to_app( |
| 812 | request.id, |
| 813 | ControlCommand::Message { text: params.text }, |
| 814 | commands_tx, |
| 815 | ), |
| 816 | Method::Interrupt(_) => dispatch_to_app(request.id, ControlCommand::Interrupt, commands_tx), |
| 817 | Method::Relaunch(_) => dispatch_to_app(request.id, ControlCommand::Relaunch, commands_tx), |
| 818 | }; |
| 819 | write_response_line(stream.get_mut(), &response); |
| 820 | } |
| 821 | |
| 822 | /// Hand a verb to the event loop and wait (bounded) for its answer. |
| 823 | #[cfg(unix)] |
| 824 | fn dispatch_to_app( |
| 825 | id: String, |
| 826 | command: ControlCommand, |
| 827 | commands_tx: &mpsc::Sender<PendingCommand>, |
| 828 | ) -> String { |
| 829 | let (respond_to, rx) = mpsc::channel(); |
| 830 | if let Err(error) = commands_tx.send(PendingCommand { |
| 831 | id: id.clone(), |
| 832 | command, |
| 833 | respond_to, |
| 834 | }) { |
| 835 | return response_error( |
| 836 | &id, |
| 837 | "server_unavailable", |
| 838 | format!("failed to dispatch request: {error}"), |
| 839 | ); |
| 840 | } |
| 841 | match rx.recv_timeout(DISPATCH_RESPONSE_TIMEOUT) { |
| 842 | Ok(response) => response, |
| 843 | Err(mpsc::RecvTimeoutError::Timeout) => response_error( |
| 844 | &id, |
| 845 | "timeout", |
| 846 | format!( |
| 847 | "timed out waiting for the app to handle the request after {} ms", |
| 848 | DISPATCH_RESPONSE_TIMEOUT.as_millis() |
| 849 | ), |
| 850 | ), |
| 851 | Err(mpsc::RecvTimeoutError::Disconnected) => response_error( |
| 852 | &id, |
| 853 | "server_unavailable", |
| 854 | "request handling failed: app response channel closed".to_string(), |
| 855 | ), |
| 856 | } |
| 857 | } |
| 858 | |
| 859 | /// One newline-terminated line, bounded. `Ok(None)` = EOF before any content. |
| 860 | /// The cap is enforced *while* reading (a hostile peer cannot make us buffer |
| 861 | /// an unbounded line), and on oversize the remainder of the frame is |
| 862 | /// discarded through a fixed-size buffer — memory stays bounded, and a client |
| 863 | /// that wrote the whole request can still receive the rejection. |
| 864 | #[cfg(unix)] |
| 865 | fn read_request_line(stream: &mut BufReader<UnixStream>) -> io::Result<Option<String>> { |
| 866 | let mut capped = stream.by_ref().take(MAX_REQUEST_BYTES as u64 + 1); |
| 867 | let mut line = Vec::new(); |
| 868 | let read = capped.read_until(b'\n', &mut line)?; |
| 869 | if read == 0 { |
| 870 | return Ok(None); // EOF before any content |
| 871 | } |
| 872 | if line.last() != Some(&b'\n') { |
| 873 | // The frame exceeded the cap (or was torn mid-line). Discard the |
| 874 | // remainder so a well-behaved client finishes its write and reads |
| 875 | // the rejection; a torn frame's write lands nowhere and is ignored. |
| 876 | let mut buf = [0u8; 8192]; |
| 877 | loop { |
| 878 | match stream.read(&mut buf) { |
| 879 | Ok(0) | Err(_) => break, |
| 880 | Ok(n) => { |
| 881 | if buf[..n].contains(&b'\n') { |
| 882 | break; |
| 883 | } |
| 884 | } |
| 885 | } |
| 886 | } |
| 887 | return Err(io::Error::new( |
| 888 | io::ErrorKind::InvalidData, |
| 889 | format!("request exceeds {MAX_REQUEST_BYTES} bytes"), |
| 890 | )); |
| 891 | } |
| 892 | Ok(Some(String::from_utf8_lossy(&line).into_owned())) |
| 893 | } |
| 894 | |
| 895 | #[cfg(unix)] |
| 896 | fn write_response_line(stream: &mut UnixStream, value: &str) { |
| 897 | let _ = writeln!(stream, "{value}"); |
| 898 | let _ = stream.flush(); |
| 899 | } |
| 900 | |
| 901 | #[cfg(test)] |
| 902 | mod tests { |
| 903 | use super::*; |
| 904 | |
| 905 | // ── Verb parsing ────────────────────────────────────────────────────── |
| 906 | |
| 907 | #[test] |
| 908 | fn parses_each_verb_request() { |
| 909 | let message: Request = |
| 910 | serde_json::from_str(r#"{"id":"1","method":"message","params":{"text":"hi"}}"#) |
| 911 | .expect("message request"); |
| 912 | assert_eq!(message.id, "1"); |
| 913 | assert!(matches!(message.method, Method::Message(p) if p.text == "hi")); |
| 914 | |
| 915 | for (raw, want) in [ |
| 916 | ( |
| 917 | r#"{"id":"2","method":"interrupt","params":{}}"#, |
| 918 | "interrupt", |
| 919 | ), |
| 920 | (r#"{"id":"3","method":"relaunch","params":{}}"#, "relaunch"), |
| 921 | (r#"{"id":"4","method":"status","params":{}}"#, "status"), |
| 922 | ] { |
| 923 | let request: Request = serde_json::from_str(raw).expect("verb request"); |
| 924 | let got = match request.method { |
| 925 | Method::Message(_) => "message", |
| 926 | Method::Interrupt(_) => "interrupt", |
| 927 | Method::Relaunch(_) => "relaunch", |
| 928 | Method::Status(_) => "status", |
| 929 | }; |
| 930 | assert_eq!(got, want); |
| 931 | } |
| 932 | } |
| 933 | |
| 934 | #[test] |
| 935 | fn rejects_unknown_verb() { |
| 936 | let error = serde_json::from_str::<Request>(r#"{"id":"1","method":"dance","params":{}}"#) |
| 937 | .expect_err("unknown verb must not parse"); |
| 938 | let message = error.to_string(); |
| 939 | assert!(message.contains("unknown variant"), "{message}"); |
| 940 | } |
| 941 | |
| 942 | #[test] |
| 943 | fn rejects_missing_or_wrong_params() { |
| 944 | // message without `text` |
| 945 | let error = serde_json::from_str::<Request>(r#"{"id":"1","method":"message","params":{}}"#) |
| 946 | .expect_err("message without text must not parse"); |
| 947 | assert!(error.to_string().contains("missing field"), "{error}"); |
| 948 | |
| 949 | // missing params entirely |
| 950 | let error = serde_json::from_str::<Request>(r#"{"id":"1","method":"status"}"#) |
| 951 | .expect_err("missing params must not parse"); |
| 952 | assert!(!error.to_string().is_empty()); |
| 953 | |
| 954 | // non-string text |
| 955 | let error = |
| 956 | serde_json::from_str::<Request>(r#"{"id":"1","method":"message","params":{"text":7}}"#) |
| 957 | .expect_err("numeric text must not parse"); |
| 958 | assert!(!error.to_string().is_empty()); |
| 959 | |
| 960 | // non-string id |
| 961 | let error = serde_json::from_str::<Request>(r#"{"id":7,"method":"status","params":{}}"#) |
| 962 | .expect_err("numeric id must not parse"); |
| 963 | assert!(!error.to_string().is_empty()); |
| 964 | } |
| 965 | |
| 966 | #[test] |
| 967 | fn serializes_responses_in_the_response_envelope_shape() { |
| 968 | let sent = response_ok( |
| 969 | "1".into(), |
| 970 | ResponseResult::MessageSent { delivery: "queued" }, |
| 971 | ); |
| 972 | assert_eq!( |
| 973 | sent, |
| 974 | r#"{"id":"1","result":{"type":"message_sent","delivery":"queued"}}"# |
| 975 | ); |
| 976 | |
| 977 | let interrupted = response_ok("2".into(), ResponseResult::Interrupted { cancelled: true }); |
| 978 | assert_eq!( |
| 979 | interrupted, |
| 980 | r#"{"id":"2","result":{"type":"interrupted","cancelled":true}}"# |
| 981 | ); |
| 982 | |
| 983 | let relaunching = response_ok("3".into(), ResponseResult::Relaunching); |
| 984 | assert_eq!(relaunching, r#"{"id":"3","result":{"type":"relaunching"}}"#); |
| 985 | |
| 986 | let status = response_ok( |
| 987 | "4".into(), |
| 988 | ResponseResult::Status { |
| 989 | turn_state: TurnState::Idle, |
| 990 | goal: GoalSnapshot { |
| 991 | objective: Some("ship it".to_string()), |
| 992 | status: "active".to_string(), |
| 993 | paused: false, |
| 994 | }, |
| 995 | }, |
| 996 | ); |
| 997 | assert_eq!( |
| 998 | status, |
| 999 | r#"{"id":"4","result":{"type":"status","turn_state":"idle","goal":{"objective":"ship it","status":"active","paused":false}}}"# |
| 1000 | ); |
| 1001 | |
| 1002 | let error = response_error("9", "invalid_request", "nope".to_string()); |
| 1003 | assert_eq!( |
| 1004 | error, |
| 1005 | r#"{"id":"9","error":{"code":"invalid_request","message":"nope"}}"# |
| 1006 | ); |
| 1007 | } |
| 1008 | |
| 1009 | // ── Socket framing (unix) ───────────────────────────────────────────── |
| 1010 | |
| 1011 | #[cfg(unix)] |
| 1012 | fn test_endpoint() -> ( |
| 1013 | tempfile::TempDir, |
| 1014 | PathBuf, |
| 1015 | mpsc::Receiver<PendingCommand>, |
| 1016 | ControlSocketHandle, |
| 1017 | ) { |
| 1018 | let temp = tempfile::TempDir::new().expect("temp dir"); |
| 1019 | let sessions_dir = temp.path().join("sessions"); |
| 1020 | let (tx, rx) = mpsc::channel(); |
| 1021 | let status = Arc::new(Mutex::new(StatusSnapshot { |
| 1022 | turn_state: TurnState::Idle, |
| 1023 | goal: GoalSnapshot { |
| 1024 | objective: Some("goal".to_string()), |
| 1025 | status: "active".to_string(), |
| 1026 | paused: false, |
| 1027 | }, |
| 1028 | })); |
| 1029 | let handle = bind_control_socket(&sessions_dir, "test-session", tx, status).expect("bind"); |
| 1030 | ( |
| 1031 | temp, |
| 1032 | sessions_dir.join("test-session").join(SOCKET_FILE_NAME), |
| 1033 | rx, |
| 1034 | handle, |
| 1035 | ) |
| 1036 | } |
| 1037 | |
| 1038 | #[cfg(unix)] |
| 1039 | fn request_response(path: &Path, request: &str) -> String { |
| 1040 | let mut stream = UnixStream::connect(path).expect("connect"); |
| 1041 | stream |
| 1042 | .set_read_timeout(Some(Duration::from_secs(5))) |
| 1043 | .expect("read timeout"); |
| 1044 | writeln!(stream, "{request}").expect("write request"); |
| 1045 | let mut response = String::new(); |
| 1046 | BufReader::new(stream) |
| 1047 | .read_line(&mut response) |
| 1048 | .expect("read response"); |
| 1049 | response |
| 1050 | } |
| 1051 | |
| 1052 | #[cfg(unix)] |
| 1053 | #[test] |
| 1054 | fn status_verb_answers_over_the_socket() { |
| 1055 | let (_temp, path, _rx, _handle) = test_endpoint(); |
| 1056 | let response = request_response(&path, r#"{"id":"4","method":"status","params":{}}"#); |
| 1057 | let value: serde_json::Value = serde_json::from_str(&response).expect("response is json"); |
| 1058 | assert_eq!(value["id"], "4"); |
| 1059 | assert_eq!(value["result"]["type"], "status"); |
| 1060 | assert_eq!(value["result"]["turn_state"], "idle"); |
| 1061 | assert_eq!(value["result"]["goal"]["objective"], "goal"); |
| 1062 | assert_eq!(value["result"]["goal"]["paused"], false); |
| 1063 | } |
| 1064 | |
| 1065 | #[cfg(unix)] |
| 1066 | #[test] |
| 1067 | fn message_verb_reaches_the_app_channel_and_answers() { |
| 1068 | let (_temp, path, rx, _handle) = test_endpoint(); |
| 1069 | |
| 1070 | // The test stands in for the event loop on the other end of the |
| 1071 | // channel: it receives the verb and answers like `drain` would. |
| 1072 | let server = std::thread::spawn(move || { |
| 1073 | let pending = rx |
| 1074 | .recv_timeout(Duration::from_secs(5)) |
| 1075 | .expect("verb queued"); |
| 1076 | assert_eq!(pending.id, "1"); |
| 1077 | match pending.command { |
| 1078 | ControlCommand::Message { text } => assert_eq!(text, "hello"), |
| 1079 | other => panic!("expected Message, got {other:?}"), |
| 1080 | } |
| 1081 | pending |
| 1082 | .respond_to |
| 1083 | .send(response_ok( |
| 1084 | "1".into(), |
| 1085 | ResponseResult::MessageSent { |
| 1086 | delivery: "dispatched", |
| 1087 | }, |
| 1088 | )) |
| 1089 | .expect("answer"); |
| 1090 | }); |
| 1091 | |
| 1092 | let response = request_response( |
| 1093 | &path, |
| 1094 | r#"{"id":"1","method":"message","params":{"text":"hello"}}"#, |
| 1095 | ); |
| 1096 | server.join().expect("server thread"); |
| 1097 | let value: serde_json::Value = serde_json::from_str(&response).expect("response is json"); |
| 1098 | assert_eq!(value["id"], "1"); |
| 1099 | assert_eq!(value["result"]["type"], "message_sent"); |
| 1100 | assert_eq!(value["result"]["delivery"], "dispatched"); |
| 1101 | } |
| 1102 | |
| 1103 | #[cfg(unix)] |
| 1104 | #[test] |
| 1105 | fn malformed_json_gets_invalid_request_error() { |
| 1106 | let (_temp, path, _rx, _handle) = test_endpoint(); |
| 1107 | let response = request_response(&path, "{not json"); |
| 1108 | let value: serde_json::Value = serde_json::from_str(&response).expect("response is json"); |
| 1109 | assert_eq!(value["id"], ""); |
| 1110 | assert_eq!(value["error"]["code"], "invalid_request"); |
| 1111 | } |
| 1112 | |
| 1113 | #[cfg(unix)] |
| 1114 | #[test] |
| 1115 | fn unknown_verb_gets_invalid_request_error() { |
| 1116 | let (_temp, path, _rx, _handle) = test_endpoint(); |
| 1117 | let response = request_response(&path, r#"{"id":"7","method":"dance","params":{}}"#); |
| 1118 | let value: serde_json::Value = serde_json::from_str(&response).expect("response is json"); |
| 1119 | assert_eq!(value["id"], ""); |
| 1120 | assert_eq!(value["error"]["code"], "invalid_request"); |
| 1121 | } |
| 1122 | |
| 1123 | #[cfg(unix)] |
| 1124 | #[test] |
| 1125 | fn empty_line_closes_without_a_response() { |
| 1126 | let (_temp, path, _rx, _handle) = test_endpoint(); |
| 1127 | let mut stream = UnixStream::connect(&path).expect("connect"); |
| 1128 | stream |
| 1129 | .set_read_timeout(Some(Duration::from_secs(5))) |
| 1130 | .expect("read timeout"); |
| 1131 | writeln!(stream).expect("write empty line"); |
| 1132 | let mut response = String::new(); |
| 1133 | let read = BufReader::new(stream) |
| 1134 | .read_line(&mut response) |
| 1135 | .expect("read"); |
| 1136 | assert_eq!(read, 0, "empty line must close the connection silently"); |
| 1137 | assert!(response.is_empty()); |
| 1138 | } |
| 1139 | |
| 1140 | #[cfg(unix)] |
| 1141 | #[test] |
| 1142 | fn oversized_request_is_rejected_with_an_error() { |
| 1143 | let (_temp, path, _rx, _handle) = test_endpoint(); |
| 1144 | let mut stream = UnixStream::connect(&path).expect("connect"); |
| 1145 | stream |
| 1146 | .set_read_timeout(Some(Duration::from_secs(5))) |
| 1147 | .expect("read timeout"); |
| 1148 | let blob = "x".repeat(MAX_REQUEST_BYTES + 16); |
| 1149 | let request = format!(r#"{{"id":"1","method":"message","params":{{"text":"{blob}"}}}}"#); |
| 1150 | writeln!(stream, "{request}").expect("write oversized request"); |
| 1151 | let mut response = String::new(); |
| 1152 | BufReader::new(stream) |
| 1153 | .read_line(&mut response) |
| 1154 | .expect("read rejection"); |
| 1155 | let value: serde_json::Value = serde_json::from_str(&response).expect("response is json"); |
| 1156 | assert_eq!(value["error"]["code"], "invalid_request"); |
| 1157 | } |
| 1158 | |
| 1159 | #[cfg(unix)] |
| 1160 | #[test] |
| 1161 | fn bind_refuses_a_live_socket_and_takes_over_a_stale_file() { |
| 1162 | let temp = tempfile::TempDir::new().expect("temp dir"); |
| 1163 | let sessions_dir = temp.path().join("sessions"); |
| 1164 | let socket_path = sessions_dir.join("test-session").join(SOCKET_FILE_NAME); |
| 1165 | let status = Arc::new(Mutex::new(StatusSnapshot { |
| 1166 | turn_state: TurnState::Idle, |
| 1167 | goal: GoalSnapshot { |
| 1168 | objective: None, |
| 1169 | status: "active".to_string(), |
| 1170 | paused: false, |
| 1171 | }, |
| 1172 | })); |
| 1173 | let (tx, _rx) = mpsc::channel(); |
| 1174 | |
| 1175 | // A stale plain file is taken over. |
| 1176 | fs::create_dir_all(socket_path.parent().expect("parent")).expect("mkdir"); |
| 1177 | fs::write(&socket_path, b"stale").expect("write stale file"); |
| 1178 | let handle = bind_control_socket( |
| 1179 | &sessions_dir, |
| 1180 | "test-session", |
| 1181 | tx.clone(), |
| 1182 | Arc::clone(&status), |
| 1183 | ) |
| 1184 | .expect("bind over a stale file"); |
| 1185 | drop(handle); |
| 1186 | |
| 1187 | // A live listener is refused. |
| 1188 | let _ = fs::remove_file(&socket_path); |
| 1189 | let live = UnixListener::bind(&socket_path).expect("bind live listener"); |
| 1190 | let error = bind_control_socket(&sessions_dir, "test-session", tx, status) |
| 1191 | .expect_err("must refuse a live socket"); |
| 1192 | assert_eq!(error.kind(), io::ErrorKind::AddrInUse); |
| 1193 | drop(live); |
| 1194 | let _ = fs::remove_file(&socket_path); |
| 1195 | } |
| 1196 | |
| 1197 | #[cfg(unix)] |
| 1198 | #[test] |
| 1199 | fn drop_unbinds_and_unlinks_the_socket() { |
| 1200 | let temp = tempfile::TempDir::new().expect("temp dir"); |
| 1201 | let sessions_dir = temp.path().join("sessions"); |
| 1202 | let socket_path = sessions_dir.join("test-session").join(SOCKET_FILE_NAME); |
| 1203 | let (tx, _rx) = mpsc::channel(); |
| 1204 | let status = Arc::new(Mutex::new(StatusSnapshot { |
| 1205 | turn_state: TurnState::Idle, |
| 1206 | goal: GoalSnapshot { |
| 1207 | objective: None, |
| 1208 | status: "active".to_string(), |
| 1209 | paused: false, |
| 1210 | }, |
| 1211 | })); |
| 1212 | let handle = bind_control_socket(&sessions_dir, "test-session", tx, status).expect("bind"); |
| 1213 | assert!(socket_path.exists(), "socket file exists while bound"); |
| 1214 | drop(handle); |
| 1215 | // The accept thread unlinks within one poll interval. |
| 1216 | let deadline = std::time::Instant::now() + Duration::from_secs(5); |
| 1217 | while socket_path.exists() && std::time::Instant::now() < deadline { |
| 1218 | std::thread::sleep(Duration::from_millis(20)); |
| 1219 | } |
| 1220 | assert!( |
| 1221 | !socket_path.exists(), |
| 1222 | "socket file must be unlinked after drop" |
| 1223 | ); |
| 1224 | } |
| 1225 | |
| 1226 | #[cfg(unix)] |
| 1227 | #[test] |
| 1228 | fn reconcile_backs_off_after_a_refused_takeover() { |
| 1229 | let temp = tempfile::TempDir::new().expect("temp dir"); |
| 1230 | let sessions_dir = temp.path().join("sessions"); |
| 1231 | let socket_path = sessions_dir.join("sess").join(SOCKET_FILE_NAME); |
| 1232 | fs::create_dir_all(socket_path.parent().expect("parent")).expect("mkdir"); |
| 1233 | |
| 1234 | let mut control = SessionControl::new_with_sessions_dir(true, Some(sessions_dir.clone())); |
| 1235 | |
| 1236 | // A live listener occupies the path: the takeover is refused. |
| 1237 | let live = UnixListener::bind(&socket_path).expect("bind live listener"); |
| 1238 | control.reconcile(Some("sess")); |
| 1239 | assert!( |
| 1240 | control.bound_session.is_none(), |
| 1241 | "refused bind must not claim" |
| 1242 | ); |
| 1243 | |
| 1244 | // The other process goes away, but the backoff still holds. |
| 1245 | drop(live); |
| 1246 | let _ = fs::remove_file(&socket_path); |
| 1247 | control.reconcile(Some("sess")); |
| 1248 | assert!( |
| 1249 | control.bound_session.is_none(), |
| 1250 | "backoff must suppress an immediate rebind" |
| 1251 | ); |
| 1252 | |
| 1253 | // After the backoff window, the same session binds successfully. |
| 1254 | std::thread::sleep(BIND_RETRY_BACKOFF + Duration::from_millis(50)); |
| 1255 | control.reconcile(Some("sess")); |
| 1256 | assert_eq!(control.bound_session.as_deref(), Some("sess")); |
| 1257 | |
| 1258 | // Reconcile with the same id is a no-op; a different id rebinds. |
| 1259 | control.reconcile(Some("sess")); |
| 1260 | assert_eq!(control.bound_session.as_deref(), Some("sess")); |
| 1261 | control.reconcile(Some("other")); |
| 1262 | assert_eq!(control.bound_session.as_deref(), Some("other")); |
| 1263 | drop(control); |
| 1264 | } |
| 1265 | } |
| 1266 |