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