pub mod auth;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingError;
use crate::outgoing_message::OutgoingMessage;
use crate::outgoing_message::QueuedOutgoingMessage;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_core::config::find_codex_home;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::net::SocketAddr;
use std::path::Path;
use std::str::FromStr;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::error;
use tracing::warn;
pub const CHANNEL_CAPACITY: usize = 128;
mod remote_control;
mod stdio;
mod unix_socket;
#[cfg(test)]
mod unix_socket_tests;
mod websocket;
pub use remote_control::REMOTE_CONTROL_DISABLED_ENV_VAR;
pub use remote_control::RemoteControlDisabledByRequirements;
pub use remote_control::RemoteControlEnableError;
pub use remote_control::RemoteControlHandle;
pub use remote_control::RemoteControlPolicy;
pub use remote_control::RemoteControlStartConfig;
pub use remote_control::RemoteControlStartupMode;
pub use remote_control::RemoteControlUnavailable;
pub use remote_control::start_remote_control;
pub use remote_control::take_remote_control_disabled_env;
pub use stdio::start_stdio_connection;
pub use unix_socket::AppServerStartupLock;
pub use unix_socket::acquire_app_server_startup_lock;
pub use unix_socket::prepare_control_socket_path;
pub use unix_socket::start_control_socket_acceptor;
pub use websocket::start_websocket_acceptor;
const OVERLOADED_ERROR_CODE: i64 = -32001;
const APP_SERVER_CONTROL_SOCKET_DIR_NAME: &str = "app-server-control";
const APP_SERVER_CONTROL_SOCKET_FILE_NAME: &str = "app-server-control.sock";
const APP_SERVER_STARTUP_LOCK_FILE_NAME: &str = "app-server-startup.lock";
pub fn app_server_control_socket_path(codex_home: &Path) -> std::io::Result<AbsolutePathBuf> {
AbsolutePathBuf::from_absolute_path(
codex_home
.join(APP_SERVER_CONTROL_SOCKET_DIR_NAME)
.join(APP_SERVER_CONTROL_SOCKET_FILE_NAME),
)
}
pub fn app_server_startup_lock_path(codex_home: &Path) -> std::io::Result<AbsolutePathBuf> {
AbsolutePathBuf::from_absolute_path(
codex_home
.join(APP_SERVER_CONTROL_SOCKET_DIR_NAME)
.join(APP_SERVER_STARTUP_LOCK_FILE_NAME),
)
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum AppServerTransport {
Stdio,
UnixSocket { socket_path: AbsolutePathBuf },
WebSocket { bind_address: SocketAddr },
Off,
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum AppServerTransportParseError {
UnsupportedListenUrl(String),
InvalidUnixSocketPath { listen_url: String, message: String },
InvalidWebSocketListenUrl(String),
}
impl std::fmt::Display for AppServerTransportParseError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AppServerTransportParseError::UnsupportedListenUrl(listen_url) => write!(
f,
"unsupported --listen URL `{listen_url}`; expected `stdio://`, `unix://`, `unix://PATH`, `ws://IP:PORT`, or `off`"
),
AppServerTransportParseError::InvalidUnixSocketPath {
listen_url,
message,
} => write!(
f,
"invalid unix socket --listen URL `{listen_url}`; failed to resolve socket path: {message}"
),
AppServerTransportParseError::InvalidWebSocketListenUrl(listen_url) => write!(
f,
"invalid websocket --listen URL `{listen_url}`; expected `ws://IP:PORT`"
),
}
}
}
impl std::error::Error for AppServerTransportParseError {}
impl AppServerTransport {
pub const DEFAULT_LISTEN_URL: &'static str = "stdio://";
pub fn from_listen_url(listen_url: &str) -> Result<Self, AppServerTransportParseError> {
if listen_url == Self::DEFAULT_LISTEN_URL {
return Ok(Self::Stdio);
}
if let Some(raw_socket_path) = listen_url.strip_prefix("unix://") {
let socket_path = if raw_socket_path.is_empty() {
let codex_home = find_codex_home().map_err(|err| {
AppServerTransportParseError::InvalidUnixSocketPath {
listen_url: listen_url.to_string(),
message: format!("failed to resolve CODEX_HOME: {err}"),
}
})?;
app_server_control_socket_path(&codex_home).map_err(|err| {
AppServerTransportParseError::InvalidUnixSocketPath {
listen_url: listen_url.to_string(),
message: err.to_string(),
}
})?
} else {
AbsolutePathBuf::relative_to_current_dir(raw_socket_path).map_err(|err| {
AppServerTransportParseError::InvalidUnixSocketPath {
listen_url: listen_url.to_string(),
message: err.to_string(),
}
})?
};
return Ok(Self::UnixSocket { socket_path });
}
if listen_url == "off" {
return Ok(Self::Off);
}
if let Some(socket_addr) = listen_url.strip_prefix("ws://") {
let bind_address = socket_addr.parse::<SocketAddr>().map_err(|_| {
AppServerTransportParseError::InvalidWebSocketListenUrl(listen_url.to_string())
})?;
return Ok(Self::WebSocket { bind_address });
}
Err(AppServerTransportParseError::UnsupportedListenUrl(
listen_url.to_string(),
))
}
}
impl FromStr for AppServerTransport {
type Err = AppServerTransportParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::from_listen_url(s)
}
}
#[derive(Debug)]
pub enum TransportEvent {
ConnectionOpened {
connection_id: ConnectionId,
origin: ConnectionOrigin,
writer: mpsc::Sender<QueuedOutgoingMessage>,
disconnect_sender: Option<CancellationToken>,
},
ConnectionClosed {
connection_id: ConnectionId,
},
IncomingMessage {
connection_id: ConnectionId,
message: JSONRPCMessage,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionOrigin {
Stdio,
InProcess,
WebSocket,
RemoteControl,
}
static CONNECTION_ID_COUNTER: AtomicU64 = AtomicU64::new(0);
fn next_connection_id() -> ConnectionId {
ConnectionId(CONNECTION_ID_COUNTER.fetch_add(1, Ordering::Relaxed))
}
async fn forward_incoming_message(
transport_event_tx: &mpsc::Sender<TransportEvent>,
writer: &mpsc::Sender<QueuedOutgoingMessage>,
connection_id: ConnectionId,
payload: &str,
) -> bool {
match serde_json::from_str::<JSONRPCMessage>(payload) {
Ok(message) => {
enqueue_incoming_message(transport_event_tx, writer, connection_id, message).await
}
Err(err) => {
error!("Failed to deserialize JSONRPCMessage: {err}");
true
}
}
}
async fn enqueue_incoming_message(
transport_event_tx: &mpsc::Sender<TransportEvent>,
writer: &mpsc::Sender<QueuedOutgoingMessage>,
connection_id: ConnectionId,
message: JSONRPCMessage,
) -> bool {
let event = TransportEvent::IncomingMessage {
connection_id,
message,
};
match transport_event_tx.try_send(event) {
Ok(()) => true,
Err(mpsc::error::TrySendError::Closed(_)) => false,
Err(mpsc::error::TrySendError::Full(TransportEvent::IncomingMessage {
connection_id,
message: JSONRPCMessage::Request(request),
})) => {
let overload_error = OutgoingMessage::Error(OutgoingError {
id: request.id,
error: JSONRPCErrorError {
code: OVERLOADED_ERROR_CODE,
message: "Server overloaded; retry later.".to_string(),
data: None,
},
});
match writer.try_send(QueuedOutgoingMessage::new(overload_error)) {
Ok(()) => true,
Err(mpsc::error::TrySendError::Closed(_)) => false,
Err(mpsc::error::TrySendError::Full(_overload_error)) => {
warn!(
"dropping overload response for connection {:?}: outbound queue is full",
connection_id
);
true
}
}
}
Err(mpsc::error::TrySendError::Full(event)) => transport_event_tx.send(event).await.is_ok(),
}
}
fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option<String> {
let value = match serde_json::to_value(outgoing_message) {
Ok(value) => value,
Err(err) => {
error!("Failed to convert OutgoingMessage to JSON value: {err}");
return None;
}
};
match serde_json::to_string(&value) {
Ok(json) => Some(json),
Err(err) => {
error!("Failed to serialize JSONRPCMessage: {err}");
None
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::time::Duration;
use tokio::time::timeout;
#[test]
fn listen_off_parses_as_off_transport() {
assert_eq!(
AppServerTransport::from_listen_url("off"),
Ok(AppServerTransport::Off)
);
}
#[tokio::test]
async fn enqueue_incoming_request_returns_overload_error_when_queue_is_full() {
let connection_id = ConnectionId(42);
let (transport_event_tx, mut transport_event_rx) = mpsc::channel(1);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
let first_message = JSONRPCMessage::Notification(JSONRPCNotification {
method: "initialized".to_string(),
params: None,
});
transport_event_tx
.send(TransportEvent::IncomingMessage {
connection_id,
message: first_message.clone(),
})
.await
.expect("queue should accept first message");
let request = JSONRPCMessage::Request(JSONRPCRequest {
id: RequestId::Integer(7),
method: "config/read".to_string(),
params: Some(json!({ "includeLayers": false })),
trace: None,
});
assert!(
enqueue_incoming_message(&transport_event_tx, &writer_tx, connection_id, request).await
);
let queued_event = transport_event_rx
.recv()
.await
.expect("first event should stay queued");
match queued_event {
TransportEvent::IncomingMessage {
connection_id: queued_connection_id,
message,
} => {
assert_eq!(queued_connection_id, connection_id);
assert_eq!(message, first_message);
}
_ => panic!("expected queued incoming message"),
}
let overload = writer_rx
.recv()
.await
.expect("request should receive overload error");
let overload_json =
serde_json::to_value(overload.message).expect("serialize overload error");
assert_eq!(
overload_json,
json!({
"id": 7,
"error": {
"code": OVERLOADED_ERROR_CODE,
"message": "Server overloaded; retry later."
}
})
);
}
#[tokio::test]
async fn enqueue_incoming_response_waits_instead_of_dropping_when_queue_is_full() {
let connection_id = ConnectionId(42);
let (transport_event_tx, mut transport_event_rx) = mpsc::channel(1);
let (writer_tx, _writer_rx) = mpsc::channel(1);
let first_message = JSONRPCMessage::Notification(JSONRPCNotification {
method: "initialized".to_string(),
params: None,
});
transport_event_tx
.send(TransportEvent::IncomingMessage {
connection_id,
message: first_message.clone(),
})
.await
.expect("queue should accept first message");
let response = JSONRPCMessage::Response(JSONRPCResponse {
id: RequestId::Integer(7),
result: json!({"ok": true}),
});
let transport_event_tx_for_enqueue = transport_event_tx.clone();
let writer_tx_for_enqueue = writer_tx.clone();
let enqueue_handle = tokio::spawn(async move {
enqueue_incoming_message(
&transport_event_tx_for_enqueue,
&writer_tx_for_enqueue,
connection_id,
response,
)
.await
});
let queued_event = transport_event_rx
.recv()
.await
.expect("first event should be dequeued");
match queued_event {
TransportEvent::IncomingMessage {
connection_id: queued_connection_id,
message,
} => {
assert_eq!(queued_connection_id, connection_id);
assert_eq!(message, first_message);
}
_ => panic!("expected queued incoming message"),
}
let enqueue_result = enqueue_handle.await.expect("enqueue task should not panic");
assert!(enqueue_result);
let forwarded_event = transport_event_rx
.recv()
.await
.expect("response should be forwarded instead of dropped");
match forwarded_event {
TransportEvent::IncomingMessage {
connection_id: queued_connection_id,
message: JSONRPCMessage::Response(JSONRPCResponse { id, result }),
} => {
assert_eq!(queued_connection_id, connection_id);
assert_eq!(id, RequestId::Integer(7));
assert_eq!(result, json!({"ok": true}));
}
_ => panic!("expected forwarded response message"),
}
}
#[tokio::test]
async fn enqueue_incoming_request_does_not_block_when_writer_queue_is_full() {
let connection_id = ConnectionId(42);
let (transport_event_tx, _transport_event_rx) = mpsc::channel(1);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
transport_event_tx
.send(TransportEvent::IncomingMessage {
connection_id,
message: JSONRPCMessage::Notification(JSONRPCNotification {
method: "initialized".to_string(),
params: None,
}),
})
.await
.expect("transport queue should accept first message");
writer_tx
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "queued".to_string(),
details: None,
path: None,
range: None,
}),
emitted_at_ms: Some(1_234),
}),
))
.await
.expect("writer queue should accept first message");
let request = JSONRPCMessage::Request(JSONRPCRequest {
id: RequestId::Integer(7),
method: "config/read".to_string(),
params: Some(json!({ "includeLayers": false })),
trace: None,
});
let enqueue_result = timeout(
Duration::from_millis(100),
enqueue_incoming_message(&transport_event_tx, &writer_tx, connection_id, request),
)
.await
.expect("enqueue should not block while writer queue is full");
assert!(enqueue_result);
let queued_outgoing = writer_rx
.recv()
.await
.expect("writer queue should still contain original message");
let queued_json =
serde_json::to_value(queued_outgoing.message).expect("serialize queued message");
assert_eq!(
queued_json,
json!({
"method": "configWarning",
"params": {
"summary": "queued",
"details": null,
},
"emittedAtMs": 1_234,
})
);
}
}