diff --git a/CHANGELOG.md b/CHANGELOG.md index 438c7221b5..6e6c43cffc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Public roster language is Pod. `/pod` is the customer surface; fleet remains the internal wire, storage, and migration name (#5776). +- Compaction replacement history keeps a bounded last user round (assistant + + tool results) instead of dropping them behind a summary. `/context` names + the compaction path and `/anchor` survival. Failed compact still does not + replace live history (#4394). - Provider catalogs: compatible hosts (Baseten, Groq, Cerebras, SenseNova, Command Code) no longer compile a frozen model roster. Descriptors name the wire, URL, and env; live `GET /v1/models` and a Codewhale-owned catalog @@ -39,6 +43,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Compaction publishes a structured survival contract for session-tree + journal entry types (`crates/tui/src/compaction/SURVIVAL_CONTRACT.md`) and + fails closed when the last user round, tool results, `/anchor` text, or + checkpoint receipt would vanish (#4394). - Internal: `codewhale-config` gains `RouteAuthoritySnapshot`, one immutable authority that owns a compiled provider catalog together with the route resolver projected from it, so a picker, a readiness view, and an execution @@ -241,6 +249,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fresh interactive sessions no longer leave a phantom one-message duplicate + behind. The TUI claimed one session id (Runtime store lock, turn-start crash + checkpoint) while the engine minted a second one; the first `SessionUpdated` + re-keyed the App, the completion commit cleared only the engine id's + checkpoint, and `codewhale --continue` later "recovered" the orphaned + checkpoint as a duplicate session instead of the real one. The engine now + adopts the host-owned id at spawn (`EngineConfig::session_id`) and `/clear` + mints the next id in the App like `/new`. A non-TTY `--continue` no longer + promotes and consumes the crash checkpoint before failing the terminal + check, and the root `codewhale --resume ` / `--session-id ` flags + documented in the operations runbook now parse instead of being swallowed + as a prompt. - Website: `/signin`, `/signup`, and `/auth/callback` are locale-aware public routes instead of localized 404s. Sign-in and create-account send the person to the CWC app; OAuth callbacks hop to `app.codewhale.net` with the query diff --git a/Cargo.lock b/Cargo.lock index a1543db95f..9d84879347 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -769,6 +769,7 @@ dependencies = [ "codewhale-execpolicy", "codewhale-hooks", "codewhale-mcp", + "codewhale-paths", "codewhale-protocol", "codewhale-release", "codewhale-state", @@ -778,6 +779,7 @@ dependencies = [ "serde", "serde_json", "tempfile", + "thiserror 2.0.20", "tokio", "tower", "tower-http 0.7.0", diff --git a/crates/app-server/Cargo.toml b/crates/app-server/Cargo.toml index c437df7373..722cc12a2d 100644 --- a/crates/app-server/Cargo.toml +++ b/crates/app-server/Cargo.toml @@ -21,6 +21,7 @@ codewhale-core = { path = "../core", version = "0.9.11" } codewhale-execpolicy = { path = "../execpolicy", version = "0.9.11" } codewhale-hooks = { path = "../hooks", version = "0.9.11" } codewhale-mcp = { path = "../mcp", version = "0.9.11" } +codewhale-paths = { path = "../paths", version = "0.9.11" } codewhale-protocol = { path = "../protocol", version = "0.9.11" } codewhale-release = { path = "../release", version = "0.9.11" } codewhale-state = { path = "../state", version = "0.9.11" } @@ -28,6 +29,7 @@ codewhale-tools = { path = "../tools", version = "0.9.11" } serde.workspace = true serde_json.workspace = true rustls.workspace = true +thiserror.workspace = true tokio.workspace = true tower-http.workspace = true tracing.workspace = true diff --git a/crates/app-server/src/daemon_socket.rs b/crates/app-server/src/daemon_socket.rs new file mode 100644 index 0000000000..6ce22ad5c2 --- /dev/null +++ b/crates/app-server/src/daemon_socket.rs @@ -0,0 +1,955 @@ +//! Unix-domain-socket daemon transport (Desktop Phase 0, socket half). +//! +//! The desktop shell attaches to a long-lived `codewhale app-server --socket` +//! daemon over a local socket instead of a TCP port: local multi-client, +//! peer-credential auth, nothing to firewall (CORE-PROTOCOL spec §5). The +//! wire is *identical* to the `--stdio` transport — newline-delimited +//! JSON-RPC 2.0 driven by the same [`crate::run_stdio_loop`] — with exactly +//! one addition in front of it: a `daemon/attach` handshake that establishes +//! who this client is and whether it owns the daemon. +//! +//! # Endpoint resolution +//! +//! In precedence order (see [`resolve_socket_path`]): +//! +//! 1. an explicit path (`--socket-path`); +//! 2. `$CODEWHALE_HOME/run/daemon.sock` when `CODEWHALE_HOME` is set — an +//! explicit home is an isolation boundary, so its daemon must not collide +//! with the default one; +//! 3. `$XDG_RUNTIME_DIR/codewhale/daemon.sock`; +//! 4. macOS: `~/Library/Application Support/codewhale/daemon.sock`; +//! 5. `~/.codewhale/run/daemon.sock`. +//! +//! Windows is reserved as the named pipe [`WINDOWS_NAMED_PIPE`]; binding +//! there returns [`DaemonSocketError::UnsupportedPlatform`] rather than +//! silently falling back to TCP. +//! +//! # Ownership +//! +//! Hermes' claim model, server-side: a client attaches with `mode: "claim"` +//! (it spawned the daemon and will manage its lifetime) or `mode: "attach"` +//! (it found a healthy daemon and is a guest). Only the current owner may +//! `shutdown` the daemon; guests get `not_daemon_owner`. When the owner +//! disconnects the slot frees, so a relaunched shell can re-claim the daemon +//! it left running — sessions survive UI restarts because the daemon does. + +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +/// Basename of the daemon socket inside the Codewhale runtime directory. +pub const DAEMON_SOCKET_FILE_NAME: &str = "daemon.sock"; + +/// Reserved Windows endpoint. Not implemented yet; binding on Windows fails +/// with [`DaemonSocketError::UnsupportedPlatform`] naming this pipe. +pub const WINDOWS_NAMED_PIPE: &str = r"\\.\pipe\codewhale-daemon"; + +/// JSON-RPC method a client must send first on a daemon-socket connection. +pub const ATTACH_METHOD: &str = "daemon/attach"; + +/// Longest socket path the kernel accepts (`sun_path` minus the NUL). +pub const MAX_SOCKET_PATH_BYTES: usize = if cfg!(any(target_os = "macos", target_os = "ios")) { + 103 +} else { + 107 +}; + +/// Typed failures of the daemon socket transport. +#[derive(Debug, thiserror::Error)] +pub enum DaemonSocketError { + /// The platform has no daemon socket implementation. Never a silent + /// fallback: the caller must pick another transport explicitly. + #[error( + "the daemon socket transport is not supported on {platform}; the reserved endpoint \ + there is the named pipe {planned_endpoint}, which is not implemented yet" + )] + UnsupportedPlatform { + platform: &'static str, + planned_endpoint: &'static str, + }, + /// No home directory (or runtime directory) to derive a default path from. + #[error( + "cannot resolve the Codewhale runtime directory for the daemon socket: no home directory" + )] + RuntimeDirUnavailable, + /// `CODEWHALE_HOME` is set but not a usable absolute path. + #[error("invalid CODEWHALE_HOME override: {0}")] + InvalidHomeOverride(String), + /// Unix socket paths are limited to roughly one hundred bytes. + #[error("daemon socket path {} is {len} bytes; this platform allows at most {max}", path.display())] + PathTooLong { + path: PathBuf, + len: usize, + max: usize, + }, + /// Something other than a socket already sits at the path. Refused so a + /// misconfigured path can never delete a user's file. + #[error("{} exists and is not a unix socket; refusing to remove it", path.display())] + NotASocket { path: PathBuf }, + /// A daemon answered on the socket: this one must not replace it. + #[error( + "a live listener already answers on {}; refusing to replace it (another codewhale daemon, or something else bound to this path)", + path.display() + )] + AlreadyRunning { path: PathBuf }, + /// The liveness probe neither connected nor was refused within the + /// budget. Refused rather than clobbered; remove the file by hand if the + /// old daemon is truly gone. + #[error("liveness probe of {} timed out; refusing to replace a socket that may be live", path.display())] + ProbeTimedOut { path: PathBuf }, + /// Filesystem or socket I/O failed. + #[error("{context} ({})", path.display())] + Io { + context: &'static str, + path: PathBuf, + #[source] + source: std::io::Error, + }, + /// The app-server state (config, state store, runtime) failed to build. + #[error("failed to build daemon state")] + State(#[source] anyhow::Error), +} + +/// How to start the daemon socket transport. +#[derive(Debug, Clone, Default)] +pub struct DaemonSocketOptions { + /// Explicit socket path; `None` resolves the platform default. + pub socket_path: Option, + /// Explicit config file, like `app-server --config`. + pub config_path: Option, +} + +/// Inputs to [`resolve_socket_path`], separated from the environment so the +/// precedence rules are a pure, testable function. +#[derive(Debug, Clone, Default)] +pub struct SocketPathInputs { + /// `--socket-path`. + pub explicit: Option, + /// A valid explicit `CODEWHALE_HOME`. + pub codewhale_home_override: Option, + /// `$XDG_RUNTIME_DIR`, when set and non-empty. + pub xdg_runtime_dir: Option, + /// The user's home directory. + pub user_home: Option, + /// Whether the macOS Application Support layout applies. + pub macos: bool, +} + +impl SocketPathInputs { + /// Capture the live environment. + pub fn from_environment(explicit: Option) -> Result { + let codewhale_home_override = codewhale_paths::codewhale_home_override() + .map_err(|err| DaemonSocketError::InvalidHomeOverride(err.to_string()))?; + let xdg_runtime_dir = std::env::var_os("XDG_RUNTIME_DIR") + .filter(|value| !value.is_empty()) + .map(PathBuf::from); + Ok(Self { + explicit, + codewhale_home_override, + xdg_runtime_dir, + user_home: codewhale_paths::user_home(), + macos: cfg!(target_os = "macos"), + }) + } +} + +/// Apply the precedence rules documented at the module level and enforce the +/// kernel's path-length limit. +pub fn resolve_socket_path(inputs: &SocketPathInputs) -> Result { + let path = if let Some(explicit) = inputs.explicit.clone() { + explicit + } else if let Some(home) = inputs.codewhale_home_override.clone() { + home.join("run").join(DAEMON_SOCKET_FILE_NAME) + } else if let Some(runtime_dir) = inputs.xdg_runtime_dir.clone() { + runtime_dir.join("codewhale").join(DAEMON_SOCKET_FILE_NAME) + } else { + let user_home = inputs + .user_home + .clone() + .ok_or(DaemonSocketError::RuntimeDirUnavailable)?; + if inputs.macos { + user_home + .join("Library") + .join("Application Support") + .join("codewhale") + .join(DAEMON_SOCKET_FILE_NAME) + } else { + user_home + .join(codewhale_paths::CODEWHALE_APP_DIR) + .join("run") + .join(DAEMON_SOCKET_FILE_NAME) + } + }; + let len = path.as_os_str().len(); + if len > MAX_SOCKET_PATH_BYTES { + return Err(DaemonSocketError::PathTooLong { + path, + len, + max: MAX_SOCKET_PATH_BYTES, + }); + } + Ok(path) +} + +/// The socket path this host would use with no explicit override. +#[cfg(unix)] +pub fn default_socket_path() -> Result { + resolve_socket_path(&SocketPathInputs::from_environment(None)?) +} + +/// The socket path this host would use with no explicit override. +#[cfg(not(unix))] +pub fn default_socket_path() -> Result { + Err(unsupported_platform()) +} + +/// Who is on the other end of a daemon-socket connection. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ClientIdentity { + /// Product name of the client, e.g. `codewhale-desktop`. + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pid: Option, +} + +/// Ownership intent carried by `daemon/attach`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AttachMode { + /// A guest: use the daemon, never stop it. + #[default] + Attach, + /// The daemon's owner: may `shutdown`. Fails if a live owner exists. + Claim, +} + +/// Role granted by a successful attach. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AttachRole { + Owner, + Attached, +} + +/// `daemon/attach` params. +#[derive(Debug, Clone, Deserialize)] +pub struct AttachParams { + pub client: ClientIdentity, + #[serde(default)] + pub mode: AttachMode, + /// Bundle-skew guard: when set, the daemon refuses the attach unless its + /// own version string matches exactly. + #[serde(default)] + pub expect_daemon_version: Option, +} + +/// The typed refusal every non-unix entry point returns. Unused in the unix +/// library build by construction; the tests pin its wording on every host. +#[cfg_attr(unix, allow(dead_code))] +fn unsupported_platform() -> DaemonSocketError { + DaemonSocketError::UnsupportedPlatform { + platform: std::env::consts::OS, + planned_endpoint: WINDOWS_NAMED_PIPE, + } +} + +#[cfg(unix)] +mod platform { + use std::collections::HashMap; + use std::os::unix::fs::{DirBuilderExt, FileTypeExt, MetadataExt, PermissionsExt}; + use std::path::{Path, PathBuf}; + use std::sync::{Arc, Mutex}; + use std::time::{Duration, Instant}; + + use anyhow::Result; + use serde_json::{Value, json}; + use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, BufReader, Lines}; + use tokio::net::{UnixListener, UnixStream}; + use tokio::sync::watch; + use tokio::task::JoinSet; + + use super::{ + ATTACH_METHOD, AttachMode, AttachParams, AttachRole, ClientIdentity, DaemonSocketError, + DaemonSocketOptions, SocketPathInputs, resolve_socket_path, + }; + use crate::{ + AppState, AppTransport, JsonRpcError, ParsedStdioLine, ShutdownAuthority, StdioLoopExit, + StdioLoopPolicy, build_state_with_transport, dispatch_stdio_request_with_writer, + jsonrpc_error, jsonrpc_result, legacy_deepseek_compat, params_or_object, parse_params, + parse_stdio_line, run_stdio_loop, write_stdio_line, + }; + + /// How long the stale-socket probe waits for a connect to resolve. + const PROBE_TIMEOUT: Duration = Duration::from_secs(1); + + /// Facts about this daemon, reported in every attach reply. + #[derive(Debug)] + struct DaemonInfo { + pid: u32, + version: &'static str, + /// Owner uid of the socket file, i.e. the daemon's effective uid. + uid: u32, + socket_path: PathBuf, + started_at: Instant, + } + + #[derive(Debug, Default)] + struct ConnectionRegistry { + next_id: u64, + connections: HashMap, + owner: Option, + } + + impl ConnectionRegistry { + fn register(&mut self, client: ClientIdentity) -> u64 { + self.next_id += 1; + let id = self.next_id; + self.connections.insert(id, client); + id + } + + /// Take the owner slot, or report who holds it. + fn claim(&mut self, id: u64) -> Result<(), ClientIdentity> { + if let Some(owner_id) = self.owner + && owner_id != id + && let Some(owner) = self.connections.get(&owner_id) + { + return Err(owner.clone()); + } + self.owner = Some(id); + Ok(()) + } + + fn owner(&self) -> Option { + self.owner.and_then(|id| self.connections.get(&id)).cloned() + } + + fn remove(&mut self, id: u64) { + self.connections.remove(&id); + if self.owner == Some(id) { + self.owner = None; + } + } + } + + /// Releases the registry slot (and the owner claim) on drop, whichever + /// way the connection ends. + struct ConnectionGuard { + registry: Arc>, + id: u64, + role: AttachRole, + } + + impl Drop for ConnectionGuard { + fn drop(&mut self) { + if let Ok(mut registry) = self.registry.lock() { + registry.remove(self.id); + } + } + } + + /// Removes the socket file when the server stops, however it stops. + struct SocketFileGuard(PathBuf); + + impl Drop for SocketFileGuard { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } + } + + /// Asks a running [`DaemonSocket::serve`] to stop. + #[derive(Debug, Clone)] + pub struct DaemonShutdownHandle(Arc>); + + impl DaemonShutdownHandle { + /// Idempotent; safe to call from any task or signal handler. + pub fn trigger(&self) { + self.0.send_replace(true); + } + } + + #[derive(Clone)] + struct ConnectionContext { + state: AppState, + registry: Arc>, + info: Arc, + shutdown: DaemonShutdownHandle, + } + + /// A bound, not yet serving, daemon socket. + pub struct DaemonSocket { + listener: UnixListener, + path: PathBuf, + state: AppState, + shutdown: Arc>, + } + + impl DaemonSocket { + /// Where clients connect. + #[must_use] + pub fn local_path(&self) -> &Path { + &self.path + } + + /// A handle that stops [`Self::serve`] from outside (signals, tests). + #[must_use] + pub fn shutdown_handle(&self) -> DaemonShutdownHandle { + DaemonShutdownHandle(Arc::clone(&self.shutdown)) + } + + /// Accept clients until the owner sends `shutdown` or the handle is + /// triggered. Removes the socket file on the way out. + pub async fn serve(self) -> Result<(), DaemonSocketError> { + let Self { + listener, + path, + state, + shutdown, + } = self; + let _socket_file = SocketFileGuard(path.clone()); + let uid = std::fs::metadata(&path) + .map_err(|source| DaemonSocketError::Io { + context: "failed to stat the daemon socket", + path: path.clone(), + source, + })? + .uid(); + let context = ConnectionContext { + state, + registry: Arc::new(Mutex::new(ConnectionRegistry::default())), + info: Arc::new(DaemonInfo { + pid: std::process::id(), + version: env!("CARGO_PKG_VERSION"), + uid, + socket_path: path.clone(), + started_at: Instant::now(), + }), + shutdown: DaemonShutdownHandle(Arc::clone(&shutdown)), + }; + let mut shutdown_rx = shutdown.subscribe(); + let mut connections = JoinSet::new(); + + loop { + if *shutdown_rx.borrow() { + break; + } + tokio::select! { + accepted = listener.accept() => match accepted { + Ok((stream, _)) => { + connections.spawn(handle_connection(context.clone(), stream)); + } + Err(err) => { + tracing::warn!(error = %err, "daemon socket accept failed"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + }, + changed = shutdown_rx.changed() => { + if changed.is_err() || *shutdown_rx.borrow() { + break; + } + } + } + } + + // The owner's `shutdown` reply was flushed before its loop + // returned, so aborting what is left loses nothing a client + // still needs. + connections.shutdown().await; + Ok(()) + } + } + + /// Resolve the path, clear a stale socket, bind with `0600`, and build + /// the shared app state. Does not accept anything until + /// [`DaemonSocket::serve`]. + pub async fn bind_daemon_socket( + options: DaemonSocketOptions, + ) -> Result { + let path = resolve_socket_path(&SocketPathInputs::from_environment(options.socket_path)?)?; + ensure_private_parent_dir(&path)?; + clear_stale_socket(&path).await?; + + let listener = UnixListener::bind(&path).map_err(|source| DaemonSocketError::Io { + context: "failed to bind the daemon socket", + path: path.clone(), + source, + })?; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).map_err( + |source| DaemonSocketError::Io { + context: "failed to restrict daemon socket permissions to 0600", + path: path.clone(), + source, + }, + )?; + + let state = build_state_with_transport(options.config_path, None, AppTransport::Socket) + .map_err(DaemonSocketError::State)?; + let (shutdown, _) = watch::channel(false); + Ok(DaemonSocket { + listener, + path, + state, + shutdown: Arc::new(shutdown), + }) + } + + /// Create the socket's directory as `0700` when it does not exist. An + /// existing directory is left as the operator made it. + fn ensure_private_parent_dir(path: &Path) -> Result<(), DaemonSocketError> { + let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + else { + return Ok(()); + }; + if parent.is_dir() { + return Ok(()); + } + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(parent) + .map_err(|source| DaemonSocketError::Io { + context: "failed to create the daemon runtime directory", + path: parent.to_path_buf(), + source, + }) + } + + /// Remove a socket file nobody answers on; refuse to touch anything else. + async fn clear_stale_socket(path: &Path) -> Result<(), DaemonSocketError> { + let metadata = match std::fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(source) => { + return Err(DaemonSocketError::Io { + context: "failed to inspect the daemon socket path", + path: path.to_path_buf(), + source, + }); + } + }; + if !metadata.file_type().is_socket() { + return Err(DaemonSocketError::NotASocket { + path: path.to_path_buf(), + }); + } + match tokio::time::timeout(PROBE_TIMEOUT, UnixStream::connect(path)).await { + Ok(Ok(_live)) => Err(DaemonSocketError::AlreadyRunning { + path: path.to_path_buf(), + }), + Ok(Err(_refused)) => { + std::fs::remove_file(path).map_err(|source| DaemonSocketError::Io { + context: "failed to remove a stale daemon socket", + path: path.to_path_buf(), + source, + }) + } + Err(_elapsed) => Err(DaemonSocketError::ProbeTimedOut { + path: path.to_path_buf(), + }), + } + } + + async fn handle_connection(context: ConnectionContext, stream: UnixStream) { + match stream.peer_cred() { + Ok(cred) if cred.uid() == context.info.uid => {} + Ok(cred) => { + tracing::warn!( + peer_uid = cred.uid(), + daemon_uid = context.info.uid, + "rejected daemon socket peer: uid mismatch" + ); + return; + } + Err(err) => { + tracing::warn!(error = %err, "rejected daemon socket peer: no peer credentials"); + return; + } + } + + let (rx, mut writer) = stream.into_split(); + let mut lines = BufReader::new(rx).lines(); + let guard = match handshake(&context, &mut lines, &mut writer).await { + Ok(Some(guard)) => guard, + Ok(None) => return, + Err(err) => { + tracing::debug!(error = %err, "daemon socket handshake aborted"); + return; + } + }; + + let policy = StdioLoopPolicy { + transport: AppTransport::Socket, + shutdown: match guard.role { + AttachRole::Owner => ShutdownAuthority::Granted, + AttachRole::Attached => ShutdownAuthority::Denied, + }, + }; + // The guard moves into the loop so the claim is released when the + // socket closes, not when a long-running turn finally returns. + let exit = run_stdio_loop(&context.state, lines, writer, policy, Some(guard)).await; + match exit { + Ok(StdioLoopExit::Shutdown) => context.shutdown.trigger(), + Ok(StdioLoopExit::InputClosed) => {} + Err(err) => tracing::debug!(error = %err, "daemon socket connection ended with error"), + } + } + + /// Serve `healthz` and wait for `daemon/attach`; everything else is + /// refused with `attach_required` until the client attaches. + async fn handshake( + context: &ConnectionContext, + lines: &mut Lines, + writer: &mut W, + ) -> Result> + where + R: AsyncBufRead + Unpin, + W: AsyncWrite + Unpin, + { + loop { + let Some(line) = lines.next_line().await? else { + return Ok(None); + }; + let request = match parse_stdio_line(&line) { + ParsedStdioLine::Blank => continue, + ParsedStdioLine::Rejected(response) => { + write_stdio_line(writer, &response).await?; + continue; + } + ParsedStdioLine::Request(request) => request, + }; + let id = request.id.clone(); + match request.method.as_str() { + "healthz" | "app/healthz" => { + let response = match dispatch_stdio_request_with_writer( + &context.state, + writer, + &request.method, + request.params, + AppTransport::Socket, + ) + .await + { + Ok(dispatch) => jsonrpc_result(id, dispatch.result), + Err(err) => jsonrpc_error(id, err), + }; + write_stdio_line(writer, &response).await?; + } + ATTACH_METHOD => match attach(context, request.params) { + Ok((result, guard)) => { + write_stdio_line(writer, &jsonrpc_result(id, result)).await?; + return Ok(Some(guard)); + } + Err(err) => write_stdio_line(writer, &jsonrpc_error(id, err)).await?, + }, + other => { + write_stdio_line( + writer, + &jsonrpc_error(id, JsonRpcError::attach_required(other)), + ) + .await?; + } + } + } + } + + fn attach( + context: &ConnectionContext, + params: Value, + ) -> Result<(Value, ConnectionGuard), JsonRpcError> { + let params: AttachParams = parse_params(params_or_object(params))?; + if params.client.name.trim().is_empty() { + return Err(JsonRpcError::invalid_params( + "client.name must not be empty", + )); + } + if let Some(expected) = params.expect_daemon_version.as_deref() + && expected != context.info.version + { + return Err(JsonRpcError::daemon_version_skew( + expected, + context.info.version, + )); + } + + let mut registry = context + .registry + .lock() + .map_err(|_| JsonRpcError::internal("daemon connection registry poisoned"))?; + let id = registry.register(params.client.clone()); + let role = match params.mode { + AttachMode::Attach => AttachRole::Attached, + AttachMode::Claim => match registry.claim(id) { + Ok(()) => AttachRole::Owner, + Err(owner) => { + registry.remove(id); + let owner = serde_json::to_value(owner) + .map_err(|err| JsonRpcError::internal(err.to_string()))?; + return Err(JsonRpcError::daemon_already_claimed(&owner)); + } + }, + }; + let owner = registry.owner(); + let connections = registry.connections.len(); + drop(registry); + + let info = &context.info; + let result = json!({ + "attached": true, + "connection_id": id, + "role": role, + "transport": AppTransport::Socket.label(), + "daemon": { + "service": legacy_deepseek_compat::SERVICE_NAME, + "pid": info.pid, + "version": info.version, + "socket_path": info.socket_path.display().to_string(), + "uptime_ms": u64::try_from(info.started_at.elapsed().as_millis()).unwrap_or(u64::MAX), + }, + "owner": owner, + "connections": connections, + }); + Ok(( + result, + ConnectionGuard { + registry: Arc::clone(&context.registry), + id, + role, + }, + )) + } + + #[cfg(test)] + mod tests { + use super::*; + + fn client(name: &str) -> ClientIdentity { + ClientIdentity { + name: name.to_string(), + version: None, + pid: None, + } + } + + #[test] + fn registry_claim_is_exclusive_until_the_owner_leaves() { + let mut registry = ConnectionRegistry::default(); + let first = registry.register(client("desktop-a")); + let second = registry.register(client("desktop-b")); + + assert!(registry.claim(first).is_ok()); + assert_eq!(registry.claim(second), Err(client("desktop-a"))); + assert!( + registry.claim(first).is_ok(), + "re-claim by the owner is idempotent" + ); + + registry.remove(first); + assert_eq!(registry.owner(), None); + assert!(registry.claim(second).is_ok()); + assert_eq!(registry.owner(), Some(client("desktop-b"))); + } + + #[test] + fn removing_a_guest_keeps_the_owner() { + let mut registry = ConnectionRegistry::default(); + let owner = registry.register(client("owner")); + let guest = registry.register(client("guest")); + registry.claim(owner).expect("claim"); + registry.remove(guest); + assert_eq!(registry.owner(), Some(client("owner"))); + assert_eq!(registry.connections.len(), 1); + } + } +} + +#[cfg(not(unix))] +mod platform { + use std::path::Path; + + use super::{DaemonSocketError, DaemonSocketOptions, unsupported_platform}; + + /// Placeholder until the Windows named pipe lands; cannot be constructed. + pub struct DaemonSocket { + never: std::convert::Infallible, + } + + /// Placeholder handle for the unsupported platform. + #[derive(Debug, Clone)] + pub struct DaemonShutdownHandle(()); + + impl DaemonShutdownHandle { + pub fn trigger(&self) {} + } + + impl DaemonSocket { + #[must_use] + pub fn local_path(&self) -> &Path { + match self.never {} + } + + #[must_use] + pub fn shutdown_handle(&self) -> DaemonShutdownHandle { + match self.never {} + } + + pub async fn serve(self) -> Result<(), DaemonSocketError> { + match self.never {} + } + } + + /// Always [`DaemonSocketError::UnsupportedPlatform`] here. + pub async fn bind_daemon_socket( + _options: DaemonSocketOptions, + ) -> Result { + Err(unsupported_platform()) + } +} + +pub use platform::{DaemonShutdownHandle, DaemonSocket, bind_daemon_socket}; + +/// `codewhale app-server --socket`: bind, announce, serve until the owner's +/// `shutdown` or a termination signal. +pub async fn run_daemon_socket(options: DaemonSocketOptions) -> anyhow::Result<()> { + let daemon = bind_daemon_socket(options).await?; + let path: &Path = daemon.local_path(); + tracing::info!(path = %path.display(), "codewhale daemon listening on unix socket"); + eprintln!("codewhale daemon: listening on {}", path.display()); + + let handle = daemon.shutdown_handle(); + tokio::spawn(async move { + crate::shutdown_signal().await; + handle.trigger(); + }); + daemon.serve().await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn inputs() -> SocketPathInputs { + SocketPathInputs { + explicit: None, + codewhale_home_override: None, + xdg_runtime_dir: None, + user_home: Some(PathBuf::from("/home/whale")), + macos: false, + } + } + + #[test] + fn explicit_path_wins() { + let resolved = resolve_socket_path(&SocketPathInputs { + explicit: Some(PathBuf::from("/tmp/x.sock")), + codewhale_home_override: Some(PathBuf::from("/iso")), + xdg_runtime_dir: Some(PathBuf::from("/run/user/1000")), + ..inputs() + }) + .expect("resolve"); + assert_eq!(resolved, PathBuf::from("/tmp/x.sock")); + } + + #[test] + fn explicit_codewhale_home_isolates_the_daemon() { + let resolved = resolve_socket_path(&SocketPathInputs { + codewhale_home_override: Some(PathBuf::from("/iso/home")), + xdg_runtime_dir: Some(PathBuf::from("/run/user/1000")), + ..inputs() + }) + .expect("resolve"); + assert_eq!(resolved, PathBuf::from("/iso/home/run/daemon.sock")); + } + + #[test] + fn xdg_runtime_dir_beats_home_layouts() { + let resolved = resolve_socket_path(&SocketPathInputs { + xdg_runtime_dir: Some(PathBuf::from("/run/user/1000")), + macos: true, + ..inputs() + }) + .expect("resolve"); + assert_eq!( + resolved, + PathBuf::from("/run/user/1000/codewhale/daemon.sock") + ); + } + + #[test] + fn macos_defaults_to_application_support() { + let resolved = resolve_socket_path(&SocketPathInputs { + macos: true, + user_home: Some(PathBuf::from("/Users/whale")), + ..inputs() + }) + .expect("resolve"); + assert_eq!( + resolved, + PathBuf::from("/Users/whale/Library/Application Support/codewhale/daemon.sock") + ); + } + + #[test] + fn linux_defaults_to_dot_codewhale_run() { + let resolved = resolve_socket_path(&inputs()).expect("resolve"); + assert_eq!( + resolved, + PathBuf::from("/home/whale/.codewhale/run/daemon.sock") + ); + } + + #[test] + fn no_home_is_a_typed_error() { + let err = resolve_socket_path(&SocketPathInputs { + user_home: None, + ..inputs() + }) + .expect_err("must fail"); + assert!( + matches!(err, DaemonSocketError::RuntimeDirUnavailable), + "{err}" + ); + } + + #[test] + fn over_long_paths_are_refused_before_bind() { + let long = PathBuf::from(format!( + "/{}/daemon.sock", + "d".repeat(MAX_SOCKET_PATH_BYTES) + )); + let err = resolve_socket_path(&SocketPathInputs { + explicit: Some(long.clone()), + ..inputs() + }) + .expect_err("must fail"); + match err { + DaemonSocketError::PathTooLong { path, len, max } => { + assert_eq!(path, long); + assert!(len > max); + assert_eq!(max, MAX_SOCKET_PATH_BYTES); + } + other => panic!("unexpected error: {other}"), + } + } + + #[test] + fn unsupported_platform_error_names_the_named_pipe() { + let err = unsupported_platform(); + let text = err.to_string(); + assert!(text.contains(WINDOWS_NAMED_PIPE), "{text}"); + assert!(text.contains("not implemented"), "{text}"); + } + + #[test] + fn attach_mode_defaults_to_guest() { + let params: AttachParams = + serde_json::from_value(serde_json::json!({ "client": { "name": "x" } })) + .expect("parse"); + assert_eq!(params.mode, AttachMode::Attach); + assert_eq!(params.expect_daemon_version, None); + } +} diff --git a/crates/app-server/src/lib.rs b/crates/app-server/src/lib.rs index 00fa7885b7..373ab3c4c5 100644 --- a/crates/app-server/src/lib.rs +++ b/crates/app-server/src/lib.rs @@ -32,6 +32,7 @@ use tower_http::cors::CorsLayer; use uuid::Uuid; mod chat_completions; +pub mod daemon_socket; /// Legacy DeepSeek-era naming kept for external compatibility. /// @@ -160,6 +161,22 @@ struct JsonRpcRequest { const RUNTIME_UNAVAILABLE_CODE: i64 = -32005; /// Server error: the named thread does not exist. const THREAD_NOT_FOUND_CODE: i64 = -32004; +/// Server error: a daemon-socket client tried to act before `daemon/attach`. +/// Only the unix listener raises it; gated so the Windows build (where the +/// listener is a typed-unsupported stub) does not fail `warnings = "deny"` +/// on dead code. +#[cfg(unix)] +const ATTACH_REQUIRED_CODE: i64 = -32010; +/// Server error: a `daemon/attach` claim lost to a live owner. +#[cfg(unix)] +const DAEMON_ALREADY_CLAIMED_CODE: i64 = -32011; +/// Server error: only the owning client may `shutdown` the daemon. +const NOT_DAEMON_OWNER_CODE: i64 = -32012; +/// Server error: the client refused the daemon's version at attach time. +#[cfg(unix)] +const DAEMON_VERSION_SKEW_CODE: i64 = -32013; +/// Server error: `daemon/attach` sent twice on one connection. +const ALREADY_ATTACHED_CODE: i64 = -32014; #[derive(Debug)] struct JsonRpcError { @@ -217,6 +234,58 @@ struct TurnTranscript { enum AppTransport { Http, Stdio, + /// Unix-domain-socket daemon transport (`daemon_socket`). Speaks the + /// stdio JSON-RPC protocol verbatim after a `daemon/attach` handshake. + Socket, +} + +impl AppTransport { + /// Wire label reported by `healthz` / `capabilities`. + fn label(self) -> &'static str { + match self { + Self::Http => "http", + Self::Stdio => "stdio", + Self::Socket => "unix-socket", + } + } +} + +/// Whether the peer driving a JSON-RPC loop may stop the whole server. +/// +/// The process-owned stdio loop always may (its peer *is* the supervisor). +/// On the daemon socket only the client that claimed the daemon may; every +/// other attached client is refused with `not_daemon_owner` — the brief's +/// "never terminate a daemon the app did not spawn", enforced server-side. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ShutdownAuthority { + Granted, + Denied, +} + +/// Per-connection policy for [`run_stdio_loop`]. +#[derive(Debug, Clone, Copy)] +struct StdioLoopPolicy { + transport: AppTransport, + shutdown: ShutdownAuthority, +} + +impl StdioLoopPolicy { + /// The loop owned by the process's own stdin/stdout. + const fn process_stdio() -> Self { + Self { + transport: AppTransport::Stdio, + shutdown: ShutdownAuthority::Granted, + } + } +} + +/// Why [`run_stdio_loop`] returned. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StdioLoopExit { + /// The peer closed its write side; nothing asked the server to stop. + InputClosed, + /// The peer sent an honoured `shutdown`. + Shutdown, } #[derive(Debug, Deserialize)] @@ -311,19 +380,36 @@ pub async fn run_stdio(config_path: Option) -> Result<()> { let state = build_state_with_transport(config_path, None, AppTransport::Stdio)?; let reader = BufReader::new(tokio::io::stdin()).lines(); let writer = tokio::io::BufWriter::new(tokio::io::stdout()); - run_stdio_loop(&state, reader, writer).await + run_stdio_loop( + &state, + reader, + writer, + StdioLoopPolicy::process_stdio(), + None::<()>, + ) + .await + .map(|_exit| ()) } /// The stdio JSON-RPC loop, generic over its transport so it can be driven by /// a duplex pipe in tests rather than the process's real stdin/stdout. -async fn run_stdio_loop( +async fn run_stdio_loop( state: &AppState, mut reader: tokio::io::Lines, mut writer: W, -) -> Result<()> + policy: StdioLoopPolicy, + // Dropped the moment input closes, not when the in-flight turn ends. The + // socket transport passes its owner claim here: a `thread/message` can run + // for minutes, and an owner who disconnects mid-turn must not keep the + // daemon claimed for the rest of it, or a relaunched client is locked out + // with `daemon_already_claimed` and cannot even shut the daemon down. + // Process stdio has no claim and passes `None`. + mut input_claim: Option, +) -> Result where R: AsyncBufRead + Unpin, W: AsyncWrite + Unpin, + C: Send, { // Work that arrived while a turn was streaming. The turn owns the writer // for its whole duration, so these wait for it rather than interleaving @@ -340,10 +426,10 @@ where Some(PendingStdioWork::Request(request)) => request, None => { if !stdin_open { - break; + return Ok(StdioLoopExit::InputClosed); } let Some(line) = reader.next_line().await? else { - break; + return Ok(StdioLoopExit::InputClosed); }; match parse_stdio_line(&line) { ParsedStdioLine::Blank => continue, @@ -357,6 +443,14 @@ where }; let id = request.id.clone(); + if request.method == "shutdown" && policy.shutdown == ShutdownAuthority::Denied { + write_stdio_line( + &mut writer, + &jsonrpc_error(id, JsonRpcError::not_daemon_owner()), + ) + .await?; + continue; + } let dispatched = if request.method == "thread/message" { // A turn can run for minutes. Keep reading stdin while it streams // so an interrupt (or a shutdown) can actually reach it — with a @@ -366,6 +460,7 @@ where &mut writer, &request.method, request.params, + policy.transport, ); tokio::pin!(dispatch); loop { @@ -373,24 +468,35 @@ where outcome = &mut dispatch => break outcome, line = reader.next_line(), if stdin_open => { match line? { - None => stdin_open = false, + None => { + stdin_open = false; + // Release the claim here, not after `dispatch` + // resolves. + drop(input_claim.take()); + } Some(line) => { - handle_line_during_turn(state, &line, &mut pending).await; + handle_line_during_turn(state, &line, &mut pending, policy).await; } } } } } } else { - dispatch_stdio_request_with_writer(state, &mut writer, &request.method, request.params) - .await + dispatch_stdio_request_with_writer( + state, + &mut writer, + &request.method, + request.params, + policy.transport, + ) + .await }; match dispatched { Ok(dispatch) => { write_stdio_line(&mut writer, &jsonrpc_result(id, dispatch.result)).await?; if dispatch.should_exit { - break; + return Ok(StdioLoopExit::Shutdown); } } Err(err) => { @@ -398,8 +504,6 @@ where } } } - - Ok(()) } /// Work deferred until a streaming turn releases the writer. @@ -454,6 +558,7 @@ async fn handle_line_during_turn( state: &AppState, line: &str, pending: &mut VecDeque, + policy: StdioLoopPolicy, ) { let request = match parse_stdio_line(line) { ParsedStdioLine::Blank => return, @@ -481,6 +586,14 @@ async fn handle_line_during_turn( }; pending.push_back(PendingStdioWork::Response(response)); } + "shutdown" if policy.shutdown == ShutdownAuthority::Denied => { + // A non-owner may not even interrupt the live turns: that is the + // first half of what shutdown does. + pending.push_back(PendingStdioWork::Response(jsonrpc_error( + request.id, + JsonRpcError::not_daemon_owner(), + ))); + } "shutdown" => { let live: Vec = state.in_flight_turns.lock().await.keys().cloned().collect(); for thread_id in live { @@ -920,6 +1033,70 @@ impl JsonRpcError { data: None, } } + + /// Server error (-32000..-32099): the daemon-socket connection has not + /// completed `daemon/attach`, so nothing but `healthz` is allowed yet. + #[cfg(unix)] + fn attach_required(method: &str) -> Self { + Self { + code: ATTACH_REQUIRED_CODE, + message: format!("send daemon/attach before `{method}`"), + data: Some(json!({ + "error": "attach_required", + "method": method, + "attach_method": daemon_socket::ATTACH_METHOD, + })), + } + } + + /// Server error (-32000..-32099): a `claim` attach lost to a live owner. + #[cfg(unix)] + fn daemon_already_claimed(owner: &Value) -> Self { + Self { + code: DAEMON_ALREADY_CLAIMED_CODE, + message: "daemon already claimed by another client; attach with mode=attach" + .to_string(), + data: Some(json!({ + "error": "daemon_already_claimed", + "owner": owner, + })), + } + } + + /// Server error (-32000..-32099): only the owner may stop the daemon. + fn not_daemon_owner() -> Self { + Self { + code: NOT_DAEMON_OWNER_CODE, + message: "only the client that claimed this daemon may shut it down".to_string(), + data: Some(json!({ "error": "not_daemon_owner" })), + } + } + + /// Server error (-32000..-32099): the client's expected daemon version + /// does not match the running binary (bundle skew). + #[cfg(unix)] + fn daemon_version_skew(expected: &str, actual: &str) -> Self { + Self { + code: DAEMON_VERSION_SKEW_CODE, + message: format!( + "daemon version {actual} does not match the client's expected {expected}" + ), + data: Some(json!({ + "error": "daemon_version_skew", + "expected": expected, + "actual": actual, + })), + } + } + + /// Server error (-32000..-32099): `daemon/attach` after attaching. + fn already_attached() -> Self { + Self { + code: ALREADY_ATTACHED_CODE, + message: "this connection is already attached".to_string(), + data: Some(json!({ "error": "already_attached" })), + } + } } async fn handle_thread_request( @@ -1692,14 +1869,15 @@ async fn dispatch_stdio_request( params: Value, ) -> std::result::Result { let mut sink = tokio::io::sink(); - dispatch_stdio_request_with_writer(state, &mut sink, method, params).await + dispatch_stdio_request_with_writer(state, &mut sink, method, params, AppTransport::Stdio).await } async fn dispatch_stdio_app_request( state: &AppState, request: AppRequest, + transport: AppTransport, ) -> std::result::Result { - let response = Box::pin(process_app_request(state, request, AppTransport::Stdio)).await; + let response = Box::pin(process_app_request(state, request, transport)).await; Ok(StdioDispatchResult { result: serde_json::to_value(response) .map_err(|err| JsonRpcError::internal(err.to_string()))?, @@ -1712,55 +1890,64 @@ async fn dispatch_stdio_request_with_writer( writer: &mut W, method: &str, params: Value, + transport: AppTransport, ) -> std::result::Result { let outcome = match method { "healthz" | "app/healthz" => StdioDispatchResult { result: json!({ "status": "ok", "service": legacy_deepseek_compat::SERVICE_NAME, - "transport": "stdio" - }), - should_exit: false, - }, - "capabilities" => StdioDispatchResult { - result: json!({ - "transport": "stdio", - "families": ["thread/*", "app/*", "prompt/*"], - "methods": [ - "healthz", - "thread/capabilities", - "thread/request", - "thread/create", - "thread/start", - "thread/resume", - "thread/fork", - "thread/list", - "thread/read", - "thread/set_name", - "thread/goal/set", - "thread/goal/get", - "thread/goal/clear", - "thread/archive", - "thread/unarchive", - "thread/message", - "thread/interrupt", - "app/capabilities", - "app/request", - "app/config/get", - "app/config/set", - "app/config/unset", - "app/config/list", - "app/config/reload", - "app/models", - "app/thread_loaded_list", - "prompt/capabilities", - "prompt/request", - "prompt/run", - "shutdown" - ] + "transport": transport.label() }), should_exit: false, }, + "capabilities" => { + let mut methods = vec![ + "healthz", + "thread/capabilities", + "thread/request", + "thread/create", + "thread/start", + "thread/resume", + "thread/fork", + "thread/list", + "thread/read", + "thread/set_name", + "thread/goal/set", + "thread/goal/get", + "thread/goal/clear", + "thread/archive", + "thread/unarchive", + "thread/message", + "thread/interrupt", + "app/capabilities", + "app/request", + "app/config/get", + "app/config/set", + "app/config/unset", + "app/config/list", + "app/config/reload", + "app/models", + "app/thread_loaded_list", + "prompt/capabilities", + "prompt/request", + "prompt/run", + "shutdown", + ]; + if transport == AppTransport::Socket { + // The daemon handshake exists only on the socket transport; + // stdio/HTTP clients never see it, so the stdio pin is unchanged. + methods.insert(1, daemon_socket::ATTACH_METHOD); + } + StdioDispatchResult { + result: json!({ + "transport": transport.label(), + "families": ["thread/*", "app/*", "prompt/*"], + "methods": methods, + }), + should_exit: false, + } + } "thread/capabilities" => StdioDispatchResult { result: json!({ "methods": [ @@ -1965,14 +2152,17 @@ async fn dispatch_stdio_request_with_writer( should_exit: false, } } - "app/capabilities" => dispatch_stdio_app_request(state, AppRequest::Capabilities).await?, + "app/capabilities" => { + dispatch_stdio_app_request(state, AppRequest::Capabilities, transport).await? + } "app/request" => { let request: AppRequest = parse_params(params)?; - dispatch_stdio_app_request(state, request).await? + dispatch_stdio_app_request(state, request, transport).await? } "app/config/get" => { let parsed: ConfigGetParams = parse_params(params_or_object(params))?; - dispatch_stdio_app_request(state, AppRequest::ConfigGet { key: parsed.key }).await? + dispatch_stdio_app_request(state, AppRequest::ConfigGet { key: parsed.key }, transport) + .await? } "app/config/set" => { let parsed: ConfigSetParams = parse_params(params_or_object(params))?; @@ -1982,18 +2172,28 @@ async fn dispatch_stdio_request_with_writer( key: parsed.key, value: parsed.value, }, + transport, ) .await? } "app/config/unset" => { let parsed: ConfigGetParams = parse_params(params_or_object(params))?; - dispatch_stdio_app_request(state, AppRequest::ConfigUnset { key: parsed.key }).await? + dispatch_stdio_app_request( + state, + AppRequest::ConfigUnset { key: parsed.key }, + transport, + ) + .await? + } + "app/config/list" => { + dispatch_stdio_app_request(state, AppRequest::ConfigList, transport).await? } - "app/config/list" => dispatch_stdio_app_request(state, AppRequest::ConfigList).await?, - "app/config/reload" => dispatch_stdio_app_request(state, AppRequest::ConfigReload).await?, - "app/models" => dispatch_stdio_app_request(state, AppRequest::Models).await?, + "app/config/reload" => { + dispatch_stdio_app_request(state, AppRequest::ConfigReload, transport).await? + } + "app/models" => dispatch_stdio_app_request(state, AppRequest::Models, transport).await?, "app/thread_loaded_list" | "app/thread-loaded-list" => { - dispatch_stdio_app_request(state, AppRequest::ThreadLoadedList).await? + dispatch_stdio_app_request(state, AppRequest::ThreadLoadedList, transport).await? } "prompt/capabilities" => StdioDispatchResult { result: json!({ @@ -2038,6 +2238,9 @@ async fn dispatch_stdio_request_with_writer( should_exit: true, } } + daemon_socket::ATTACH_METHOD if transport == AppTransport::Socket => { + return Err(JsonRpcError::already_attached()); + } _ => return Err(JsonRpcError::method_not_found(method)), }; Ok(outcome) @@ -3099,7 +3302,14 @@ mod tests { let loop_state = state.clone(); let loop_handle = tokio::spawn(async move { let (rx, tx) = tokio::io::split(server_side); - run_stdio_loop(&loop_state, BufReader::new(rx).lines(), tx).await + run_stdio_loop( + &loop_state, + BufReader::new(rx).lines(), + tx, + StdioLoopPolicy::process_stdio(), + None::<()>, + ) + .await }); // Start the runaway turn. @@ -3443,6 +3653,7 @@ mod tests { &mut writer, "prompt/request", json!({ "prompt": "what is 2+2" }), + AppTransport::Stdio, ) .await .expect("prompt/request dispatch"); @@ -3677,6 +3888,36 @@ mod tests { ); } + /// The socket transport advertises the `daemon/attach` handshake right + /// after `healthz`; the stdio pin above must stay untouched by it. + #[tokio::test] + async fn socket_transport_advertises_daemon_attach() { + let (state, _tmp) = capability_test_state(); + let mut sink = tokio::io::sink(); + let caps = dispatch_stdio_request_with_writer( + &state, + &mut sink, + "capabilities", + json!({}), + AppTransport::Socket, + ) + .await + .expect("capabilities dispatch"); + assert_eq!(caps.result["transport"], json!("unix-socket")); + let methods: Vec = caps.result["methods"] + .as_array() + .expect("methods array") + .iter() + .map(|m| m.as_str().expect("method string").to_string()) + .collect(); + let mut expected: Vec = EXPECTED_CAPABILITY_METHODS + .iter() + .map(|m| m.to_string()) + .collect(); + expected.insert(1, daemon_socket::ATTACH_METHOD.to_string()); + assert_eq!(methods, expected); + } + #[tokio::test] async fn every_advertised_capability_is_dispatchable() { let (state, _tmp) = capability_test_state(); diff --git a/crates/app-server/tests/daemon_socket.rs b/crates/app-server/tests/daemon_socket.rs new file mode 100644 index 0000000000..8bb076117a --- /dev/null +++ b/crates/app-server/tests/daemon_socket.rs @@ -0,0 +1,401 @@ +//! Desktop Phase 0 acceptance for the daemon socket: spawn the daemon, connect +//! over the unix socket, complete the attach/claim handshake, round-trip +//! requests through the same JSON-RPC dispatcher the stdio transport uses, +//! and shut down cleanly (socket file removed, listener gone). + +#![cfg(unix)] + +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use codewhale_app_server::daemon_socket::{ + DaemonSocketError, DaemonSocketOptions, bind_daemon_socket, +}; +use serde_json::{Value, json}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixStream; +use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::task::JoinHandle; + +static NONCE: AtomicU64 = AtomicU64::new(0); + +/// A short, unique socket path: unix socket paths are capped near 100 bytes, +/// so `std::env::temp_dir()` (deep under `/var/folders` on macOS) is too long. +/// `/tmp` is the same choice the hooks crate's socket test makes. +fn short_socket_root(label: &str) -> PathBuf { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock") + .as_millis() + % 1_000_000; + let nonce = NONCE.fetch_add(1, Ordering::Relaxed); + let pid = std::process::id(); + let root = PathBuf::from("/tmp").join(format!("cw-ds-{label}-{pid}-{nonce}-{millis}")); + assert!( + root.as_os_str().len() < 60, + "socket root too long for a unix socket test: {}", + root.display() + ); + root +} + +struct Harness { + root: PathBuf, + socket_path: PathBuf, + _config_dir: tempfile::TempDir, +} + +impl Harness { + fn new(label: &str) -> Self { + let root = short_socket_root(label); + let config_dir = tempfile::tempdir().expect("tempdir"); + std::fs::write(config_dir.path().join("config.toml"), "").expect("config"); + Self { + socket_path: root.join("run").join("daemon.sock"), + root, + _config_dir: config_dir, + } + } + + fn options(&self) -> DaemonSocketOptions { + DaemonSocketOptions { + socket_path: Some(self.socket_path.clone()), + config_path: Some(self._config_dir.path().join("config.toml")), + } + } + + /// Bind and serve on a background task; returns the serve join handle. + async fn spawn_daemon(&self) -> JoinHandle> { + let daemon = bind_daemon_socket(self.options()) + .await + .expect("bind daemon socket"); + assert_eq!(daemon.local_path(), self.socket_path.as_path()); + tokio::spawn(daemon.serve()) + } +} + +impl Drop for Harness { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.root); + } +} + +struct Client { + reader: BufReader, + writer: OwnedWriteHalf, +} + +impl Client { + async fn connect(path: &Path) -> Self { + let stream = tokio::time::timeout(Duration::from_secs(5), UnixStream::connect(path)) + .await + .expect("connect timeout") + .expect("connect"); + let (rx, writer) = stream.into_split(); + Self { + reader: BufReader::new(rx), + writer, + } + } + + async fn call(&mut self, id: u64, method: &str, params: Value) -> Value { + let line = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + })) + .expect("encode"); + self.writer + .write_all(format!("{line}\n").as_bytes()) + .await + .expect("write"); + let mut response = String::new(); + let read = tokio::time::timeout( + Duration::from_secs(10), + self.reader.read_line(&mut response), + ) + .await + .expect("response timeout") + .expect("read"); + assert!( + read > 0, + "daemon closed the connection before answering `{method}`" + ); + let value: Value = serde_json::from_str(&response).expect("json response"); + assert_eq!(value["id"], json!(id), "response id mismatch: {value}"); + value + } + + async fn attach(&mut self, id: u64, name: &str, mode: &str) -> Value { + self.call( + id, + "daemon/attach", + json!({ "client": { "name": name, "version": "0.0.0-test", "pid": std::process::id() }, "mode": mode }), + ) + .await + } + + /// Read until EOF; proves the daemon closed the socket. + async fn wait_for_close(mut self) { + let mut sink = String::new(); + let read = tokio::time::timeout(Duration::from_secs(10), self.reader.read_line(&mut sink)) + .await + .expect("close timeout") + .expect("read"); + assert_eq!(read, 0, "expected EOF, got: {sink}"); + } +} + +async fn wait_for_socket_removed(path: &Path) { + tokio::time::timeout(Duration::from_secs(10), async { + while path.exists() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("socket file must be removed on shutdown"); +} + +#[tokio::test] +async fn owner_attaches_round_trips_and_shuts_down_cleanly() { + let harness = Harness::new("owner"); + let server = harness.spawn_daemon().await; + + let socket_mode = std::fs::metadata(&harness.socket_path) + .expect("socket metadata") + .permissions() + .mode() + & 0o777; + assert_eq!(socket_mode, 0o600, "socket must be private to the user"); + let dir_mode = std::fs::metadata(harness.socket_path.parent().expect("parent")) + .expect("dir metadata") + .permissions() + .mode() + & 0o777; + assert_eq!(dir_mode, 0o700, "runtime dir must be private to the user"); + + let mut client = Client::connect(&harness.socket_path).await; + + // Anything but healthz before attaching is refused with a typed error: + // a read-only probe, a thread/* read, and a prompt run alike. + for (id, method, params) in [ + (1, "capabilities", json!({})), + (10, "thread/list", json!({})), + (11, "prompt/run", json!({ "prompt": "hi" })), + ] { + let early = client.call(id, method, params).await; + assert_eq!(early["error"]["code"], json!(-32010), "{method}: {early}"); + assert_eq!(early["error"]["data"]["error"], json!("attach_required")); + assert_eq!(early["error"]["data"]["method"], json!(method)); + } + + // healthz is allowed pre-attach so a shell can probe liveness first. + let health = client.call(2, "healthz", json!({})).await; + assert_eq!(health["result"]["status"], json!("ok"), "{health}"); + assert_eq!(health["result"]["transport"], json!("unix-socket")); + + let attached = client.attach(3, "codewhale-desktop", "claim").await; + assert_eq!(attached["result"]["attached"], json!(true), "{attached}"); + assert_eq!(attached["result"]["role"], json!("owner")); + assert_eq!(attached["result"]["transport"], json!("unix-socket")); + assert_eq!( + attached["result"]["daemon"]["pid"], + json!(std::process::id()) + ); + assert_eq!( + attached["result"]["daemon"]["version"], + json!(env!("CARGO_PKG_VERSION")) + ); + assert_eq!( + attached["result"]["owner"]["name"], + json!("codewhale-desktop") + ); + assert_eq!(attached["result"]["connections"], json!(1)); + + // Post-attach, the socket transport advertises its own handshake next to + // the stdio method set. + let advertised = client.call(8, "capabilities", json!({})).await; + let methods = advertised["result"]["methods"] + .as_array() + .expect("methods array"); + assert_eq!(methods[0], json!("healthz"), "{advertised}"); + assert_eq!(methods[1], json!("daemon/attach"), "{advertised}"); + assert!(methods.contains(&json!("shutdown"))); + + // Round-trip JSON-RPC requests through the shared dispatcher: `app/*` + // methods in, their JSON results out — byte-for-byte the shapes the + // stdio transport emits. (No protocol-crate Op/EventMsg envelope is on + // this wire; the framing is the stdio transport's newline-delimited + // JSON-RPC.) + let caps = client.call(4, "app/capabilities", json!({})).await; + assert_eq!(caps["result"]["ok"], json!(true), "{caps}"); + assert!(caps["result"]["data"]["routes"].is_array()); + let config = client + .call(5, "app/config/get", json!({ "key": "model" })) + .await; + assert_eq!(config["result"]["ok"], json!(true), "{config}"); + assert_eq!(config["result"]["data"]["key"], json!("model")); + + // A second attach on an attached connection is a typed refusal, not + // method_not_found. + let again = client.attach(6, "codewhale-desktop", "attach").await; + assert_eq!(again["error"]["code"], json!(-32014), "{again}"); + + let stopped = client.call(7, "shutdown", json!({})).await; + assert_eq!(stopped["result"]["status"], json!("stopped"), "{stopped}"); + + let outcome = tokio::time::timeout(Duration::from_secs(10), server) + .await + .expect("daemon must exit after the owner's shutdown") + .expect("join"); + outcome.expect("serve result"); + wait_for_socket_removed(&harness.socket_path).await; + client.wait_for_close().await; +} + +#[tokio::test] +async fn guests_share_the_daemon_but_cannot_stop_it() { + let harness = Harness::new("guest"); + let server = harness.spawn_daemon().await; + + let mut owner = Client::connect(&harness.socket_path).await; + let claimed = owner.attach(1, "desktop-window-1", "claim").await; + assert_eq!(claimed["result"]["role"], json!("owner"), "{claimed}"); + + let mut guest = Client::connect(&harness.socket_path).await; + let lost = guest.attach(1, "desktop-window-2", "claim").await; + assert_eq!(lost["error"]["code"], json!(-32011), "{lost}"); + assert_eq!( + lost["error"]["data"]["owner"]["name"], + json!("desktop-window-1") + ); + + let attached = guest.attach(2, "desktop-window-2", "attach").await; + assert_eq!(attached["result"]["role"], json!("attached"), "{attached}"); + assert_eq!( + attached["result"]["owner"]["name"], + json!("desktop-window-1") + ); + assert_eq!(attached["result"]["connections"], json!(2)); + + let health = guest.call(3, "healthz", json!({})).await; + assert_eq!(health["result"]["status"], json!("ok")); + + let refused = guest.call(4, "shutdown", json!({})).await; + assert_eq!(refused["error"]["code"], json!(-32012), "{refused}"); + assert_eq!(refused["error"]["data"]["error"], json!("not_daemon_owner")); + assert!( + !server.is_finished(), + "a guest's shutdown must not stop the daemon" + ); + assert!(harness.socket_path.exists()); + + // Once the owner leaves, the slot frees and a relaunched shell can claim. + drop(owner); + let mut relaunched = Client::connect(&harness.socket_path).await; + let reclaimed = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let response = relaunched.attach(1, "desktop-relaunch", "claim").await; + if response.get("result").is_some() { + return response; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("owner slot must free when the owner disconnects"); + assert_eq!(reclaimed["result"]["role"], json!("owner"), "{reclaimed}"); + + // The guest is still attached and served while the new owner is in. + let health = guest.call(5, "healthz", json!({})).await; + assert_eq!(health["result"]["status"], json!("ok")); + + let stopped = relaunched.call(2, "shutdown", json!({})).await; + assert_eq!(stopped["result"]["status"], json!("stopped")); + tokio::time::timeout(Duration::from_secs(10), server) + .await + .expect("daemon exits") + .expect("join") + .expect("serve result"); + wait_for_socket_removed(&harness.socket_path).await; + // The owner's shutdown closes every other connection, not just its own. + guest.wait_for_close().await; + relaunched.wait_for_close().await; +} + +#[tokio::test] +async fn version_skew_is_refused_at_attach() { + let harness = Harness::new("skew"); + let server = harness.spawn_daemon().await; + let mut client = Client::connect(&harness.socket_path).await; + let refused = client + .call( + 1, + "daemon/attach", + json!({ "client": { "name": "old-desktop" }, "expect_daemon_version": "0.0.1-other" }), + ) + .await; + assert_eq!(refused["error"]["code"], json!(-32013), "{refused}"); + assert_eq!( + refused["error"]["data"]["actual"], + json!(env!("CARGO_PKG_VERSION")) + ); + + let daemon = bind_daemon_socket(harness.options()).await; + // Meanwhile the original daemon is live, so a second bind must refuse. + match daemon { + Err(DaemonSocketError::AlreadyRunning { path }) => { + assert_eq!(path, harness.socket_path); + } + Err(other) => panic!("unexpected error: {other}"), + Ok(_) => panic!("second daemon must not replace a live socket"), + } + server.abort(); + let _ = server.await; +} + +#[tokio::test] +async fn stale_socket_is_cleaned_up_and_foreign_files_are_refused() { + let harness = Harness::new("stale"); + std::fs::create_dir_all(harness.socket_path.parent().expect("parent")).expect("mkdir"); + + // A socket file whose listener is gone: bind must reclaim it. + { + let dead = tokio::net::UnixListener::bind(&harness.socket_path).expect("bind dead"); + drop(dead); + } + assert!( + harness.socket_path.exists(), + "dropping a listener leaves the file" + ); + let daemon = bind_daemon_socket(harness.options()) + .await + .expect("stale socket must be reclaimed"); + let handle = daemon.shutdown_handle(); + let server = tokio::spawn(daemon.serve()); + let mut client = Client::connect(&harness.socket_path).await; + let health = client.call(1, "healthz", json!({})).await; + assert_eq!(health["result"]["status"], json!("ok")); + handle.trigger(); + tokio::time::timeout(Duration::from_secs(10), server) + .await + .expect("daemon exits on handle") + .expect("join") + .expect("serve result"); + wait_for_socket_removed(&harness.socket_path).await; + + // A regular file at the path is never deleted. + std::fs::write(&harness.socket_path, b"not a socket").expect("write file"); + match bind_daemon_socket(harness.options()).await { + Err(DaemonSocketError::NotASocket { path }) => assert_eq!(path, harness.socket_path), + Err(other) => panic!("unexpected error: {other}"), + Ok(_) => panic!("must refuse to replace a non-socket"), + } + assert_eq!( + std::fs::read(&harness.socket_path).expect("file intact"), + b"not a socket" + ); +} diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index 57e5614aab..d2aad43c69 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -17,6 +17,7 @@ use anyhow::{Context, Result, anyhow, bail}; use clap::{Args, CommandFactory, FromArgMatches, Parser, Subcommand, ValueEnum}; use clap_complete::{Shell, generate}; use codewhale_agent::ModelRegistry; +use codewhale_app_server::daemon_socket::{DaemonSocketOptions, run_daemon_socket}; use codewhale_app_server::{ AppServerOptions, run as run_app_server, run_stdio as run_app_server_stdio, }; @@ -1679,6 +1680,18 @@ struct AppServerArgs { /// Used by local SDKs and JSON-RPC integrations. #[arg(long, default_value_t = false)] stdio: bool, + /// Run as the desktop daemon: the same JSON-RPC control transport as + /// `--stdio`, served on a user-private unix domain socket under the + /// Codewhale runtime directory. Clients must `daemon/attach` first. + /// Not yet supported on Windows (fails with a typed error). + #[arg(long, default_value_t = false, conflicts_with_all = ["stdio", "http", "mobile"])] + socket: bool, + /// Socket path override for --socket. Defaults to + /// `$CODEWHALE_HOME/run/daemon.sock`, else `$XDG_RUNTIME_DIR/codewhale/daemon.sock`, + /// else `~/Library/Application Support/codewhale/daemon.sock` (macOS) or + /// `~/.codewhale/run/daemon.sock`. + #[arg(long = "socket-path", requires = "socket")] + socket_path: Option, /// Show a QR code for the mobile URL in the terminal (requires --mobile). #[arg(long, requires = "mobile")] qr: bool, @@ -4613,6 +4626,14 @@ fn run_app_server_command( finish_cli_telemetry(session, &outcome); return outcome; } + if args.socket { + let outcome = runtime.block_on(run_daemon_socket(DaemonSocketOptions { + socket_path: args.socket_path, + config_path: args.config, + })); + finish_cli_telemetry(session, &outcome); + return outcome; + } // Legacy in-process app-server HTTP transport (`/healthz`, `/thread`, `/app`, // `/prompt`, `/tool`, `/jobs`). Kept for backward compatibility; defaults to // 127.0.0.1:8787 to avoid colliding with the runtime API default of :7878. @@ -6096,16 +6117,54 @@ verbosity = "project-imported" })) )); + assert!(matches!( + parse_ok(&["deepseek", "app-server", "--socket"]).command, + Some(Commands::AppServer(AppServerArgs { + socket: true, + socket_path: None, + http: false, + mobile: false, + stdio: false, + .. + })) + )); + for argv in [ ["deepseek", "app-server", "--http", "--mobile"].as_slice(), ["deepseek", "app-server", "--http", "--stdio"].as_slice(), ["deepseek", "app-server", "--mobile", "--stdio"].as_slice(), + ["deepseek", "app-server", "--socket", "--stdio"].as_slice(), + ["deepseek", "app-server", "--socket", "--http"].as_slice(), + ["deepseek", "app-server", "--socket", "--mobile"].as_slice(), ] { let err = Cli::try_parse_from(argv).expect_err("conflicting transports must fail"); assert_eq!(err.kind(), ErrorKind::ArgumentConflict, "argv={argv:?}"); } } + #[test] + fn app_server_socket_path_requires_socket() { + let err = Cli::try_parse_from(["deepseek", "app-server", "--socket-path", "/tmp/d.sock"]) + .expect_err("--socket-path without --socket must fail"); + assert_eq!(err.kind(), ErrorKind::MissingRequiredArgument); + match parse_ok(&[ + "deepseek", + "app-server", + "--socket", + "--socket-path", + "/tmp/d.sock", + ]) + .command + { + Some(Commands::AppServer(AppServerArgs { + socket: true, + socket_path: Some(path), + .. + })) => assert_eq!(path, PathBuf::from("/tmp/d.sock")), + other => panic!("unexpected parse: {other:?}"), + } + } + #[test] fn app_server_qr_requires_mobile() { let err = Cli::try_parse_from(["deepseek", "app-server", "--qr"]) @@ -6127,6 +6186,8 @@ verbosity = "project-imported" http: true, mobile: false, stdio: false, + socket: false, + socket_path: None, qr: false, host: Some("127.0.0.1".to_string()), port: Some(9000), @@ -6165,6 +6226,8 @@ verbosity = "project-imported" http: false, mobile: true, stdio: false, + socket: false, + socket_path: None, qr: true, host: None, port: None, diff --git a/crates/tui/CHANGELOG.md b/crates/tui/CHANGELOG.md index 050f9d56b3..14ef9c6a10 100644 --- a/crates/tui/CHANGELOG.md +++ b/crates/tui/CHANGELOG.md @@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Public roster language is Pod. `/pod` is the customer surface; fleet remains the internal wire, storage, and migration name (#5776). +- Compaction replacement history keeps a bounded last user round (assistant + + tool results) instead of dropping them behind a summary. `/context` names + the compaction path and `/anchor` survival. Failed compact still does not + replace live history (#4394). - Provider catalogs: compatible hosts (Baseten, Groq, Cerebras, SenseNova, Command Code) no longer compile a frozen model roster. Descriptors name the wire, URL, and env; live `GET /v1/models` and a Codewhale-owned catalog @@ -39,6 +43,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Compaction publishes a structured survival contract for session-tree + journal entry types (`crates/tui/src/compaction/SURVIVAL_CONTRACT.md`) and + fails closed when the last user round, tool results, `/anchor` text, or + checkpoint receipt would vanish (#4394). - Internal: `codewhale-config` gains `RouteAuthoritySnapshot`, one immutable authority that owns a compiled provider catalog together with the route resolver projected from it, so a picker, a readiness view, and an execution @@ -241,6 +249,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fresh interactive sessions no longer leave a phantom one-message duplicate + behind. The TUI claimed one session id (Runtime store lock, turn-start crash + checkpoint) while the engine minted a second one; the first `SessionUpdated` + re-keyed the App, the completion commit cleared only the engine id's + checkpoint, and `codewhale --continue` later "recovered" the orphaned + checkpoint as a duplicate session instead of the real one. The engine now + adopts the host-owned id at spawn (`EngineConfig::session_id`) and `/clear` + mints the next id in the App like `/new`. A non-TTY `--continue` no longer + promotes and consumes the crash checkpoint before failing the terminal + check, and the root `codewhale --resume ` / `--session-id ` flags + documented in the operations runbook now parse instead of being swallowed + as a prompt. - Website: `/signin`, `/signup`, and `/auth/callback` are locale-aware public routes instead of localized 404s. Sign-in and create-account send the person to the CWC app; OAuth callbacks hop to `app.codewhale.net` with the query diff --git a/docs/RUNTIME_API.md b/docs/RUNTIME_API.md index 3ee105af74..f49e3fb9a8 100644 --- a/docs/RUNTIME_API.md +++ b/docs/RUNTIME_API.md @@ -28,6 +28,7 @@ local supervisor / SDK / automation harness ├─ codewhale app-server --http → HTTP/SSE runtime API (/v1/*) [canonical] ├─ codewhale app-server --mobile → runtime API + mobile control page ├─ codewhale app-server --stdio → JSON-RPC control transport over stdio + ├─ codewhale app-server --socket → same JSON-RPC over a unix domain socket (desktop daemon) ├─ codewhale doctor --json → machine-readable health & capability ├─ codewhale serve --acp → ACP stdio agent for editors such as Zed ├─ codewhale serve --mcp → MCP stdio server @@ -50,6 +51,7 @@ CLI/API surfaces are not implemented yet. | `codewhale app-server --http` | HTTP/SSE on `127.0.0.1:7878` | Full `/v1/*` runtime API (canonical) | | `codewhale app-server --mobile` | HTTP/SSE on `0.0.0.0:7878` + `/mobile` | Runtime API + phone control page | | `codewhale app-server --stdio` | JSON-RPC 2.0 over stdio | Local SDK / control probe (no listener) | +| `codewhale app-server --socket [--socket-path P]` | JSON-RPC 2.0 over a `0600` unix domain socket | Desktop daemon: multi-client, peer-uid checked, `daemon/attach` claim handshake (macOS/Linux; Windows named pipe reserved, not implemented) | | `codewhale app-server` | HTTP on `127.0.0.1:8787` | Legacy in-process app-server (`/healthz`, `/thread`, `/app`, `/prompt`, `/tool`, `/jobs`); `/prompt` and `/thread` messages execute real turns via the runtime bridge | | `codewhale serve --http` / `--mobile` | same server as `app-server --http`/`--mobile` | Compatibility aliases | @@ -118,6 +120,60 @@ printf '%s\n' \ set is pinned by a drift test in `crates/app-server/src/lib.rs`, so SDK and local integration clients can rely on it not changing silently. +### Daemon socket: `codewhale app-server --socket` + +The desktop shell (DESKTOP-APP-BRIEF §2) attaches to a long-lived daemon over +a unix domain socket. The wire is the `--stdio` transport verbatim — the same +newline-delimited JSON-RPC 2.0 methods, dispatched by the same code — with one +handshake in front of it. + +**Endpoint.** `--socket-path` if given; else `$CODEWHALE_HOME/run/daemon.sock` +when `CODEWHALE_HOME` is set (an explicit home is an isolation boundary); else +`$XDG_RUNTIME_DIR/codewhale/daemon.sock`; else +`~/Library/Application Support/codewhale/daemon.sock` on macOS or +`~/.codewhale/run/daemon.sock` elsewhere. The directory is created `0700`, the +socket is `0600`, and every accepted peer must present the daemon's own uid. +On start, a socket file nobody answers on is removed; a live one makes the new +daemon exit with `a live listener already answers on ; refusing to +replace it`; a non-socket file at the path is never touched. On Windows `--socket` fails with a typed +`UnsupportedPlatform` error naming the reserved pipe `\\.\pipe\codewhale-daemon` +— there is no silent TCP fallback. The daemon prints +`codewhale daemon: listening on ` to stderr once it is accepting. + +**Handshake.** The first request on a connection must be `daemon/attach` +(`healthz` is also allowed beforehand, so a shell can probe liveness). Every +other method is refused with `-32010 attach_required` until then. + +```json +{"jsonrpc":"2.0","id":1,"method":"daemon/attach","params":{ + "client":{"name":"codewhale-desktop","version":"1.2.3","pid":4242}, + "mode":"claim", + "expect_daemon_version":"0.9.11"}} +``` + +`mode` is `"claim"` (this client spawned the daemon and manages its lifetime) +or `"attach"` (default: a guest that found a healthy daemon). A claim while +another connection owns the daemon fails with `-32011 daemon_already_claimed` +(`data.owner` names the holder) and the client should retry with `attach`. +`expect_daemon_version`, when present, must equal the daemon's crate version +or the attach fails with `-32013 daemon_version_skew` (the bundle-skew guard). +The reply reports the granted `role` (`owner` / `attached`), the daemon's +`pid`, `version`, `socket_path`, and `uptime_ms`, the current `owner`, and the +live `connections` count. A second `daemon/attach` on an attached connection +is `-32014 already_attached`. + +**Capabilities.** On this transport `capabilities.methods` is the pinned stdio +set plus `daemon/attach` (second entry, after `healthz`); `transport` reads +`unix-socket`. `shutdown` is advertised to every connection because the method +exists, but only the owner may call it (below). + +**Ownership.** Only the owner may `shutdown`; a guest's `shutdown` is refused +with `-32012 not_daemon_owner` and does not interrupt anyone's turn. When the +owner disconnects the slot frees, so a relaunched shell re-claims the daemon it +left running. The owner's `shutdown` stops the listener, closes every +connection, and removes the socket file. Journal replay from a client's +last-seen `seq` is not part of this transport yet. + ### Interrupting a turn `thread/message` streams until the turn reaches a terminal state, which can