fix: make remote app-server proxy acquisition idempotent

This commit is contained in:
Edward Frazer
2026-05-18 18:57:34 -07:00
parent 84d941d07f
commit 5b41d03376
7 changed files with 447 additions and 2 deletions
+4
View File
@@ -6,16 +6,20 @@ pub use outgoing_message::OutgoingError;
pub use outgoing_message::OutgoingMessage;
pub use outgoing_message::OutgoingResponse;
pub use outgoing_message::QueuedOutgoingMessage;
pub use transport::AppServerListenerStartup;
pub use transport::AppServerTransport;
pub use transport::AppServerTransportParseError;
pub use transport::CHANNEL_CAPACITY;
pub use transport::ConnectionOrigin;
pub use transport::EnsuredAppServerListener;
pub use transport::RemoteControlHandle;
pub use transport::RemoteControlStartConfig;
pub use transport::RemoteControlUnavailable;
pub use transport::TransportEvent;
pub use transport::app_server_control_socket_path;
pub use transport::app_server_startup_lock_path;
pub use transport::auth;
pub use transport::ensure_control_socket_listener;
pub use transport::start_control_socket_acceptor;
pub use transport::start_remote_control;
pub use transport::start_stdio_connection;
@@ -35,6 +35,9 @@ pub use remote_control::RemoteControlStartConfig;
pub use remote_control::RemoteControlUnavailable;
pub use remote_control::start_remote_control;
pub use stdio::start_stdio_connection;
pub use unix_socket::AppServerListenerStartup;
pub use unix_socket::EnsuredAppServerListener;
pub use unix_socket::ensure_control_socket_listener;
pub use unix_socket::start_control_socket_acceptor;
pub use websocket::start_websocket_acceptor;
@@ -42,6 +45,7 @@ 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(
@@ -51,6 +55,14 @@ pub fn app_server_control_socket_path(codex_home: &Path) -> std::io::Result<Abso
)
}
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,
@@ -1,3 +1,5 @@
use std::fs::OpenOptions;
use std::future::Future;
use std::io::ErrorKind;
use std::io::Result as IoResult;
use std::path::Path;
@@ -11,6 +13,7 @@ use futures::StreamExt;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::Duration;
use tokio::time::sleep;
use tokio_tungstenite::accept_async;
use tokio_util::sync::CancellationToken;
use tracing::error;
@@ -19,6 +22,18 @@ use tracing::warn;
#[cfg(unix)]
const CONTROL_SOCKET_MODE: u32 = 0o600;
const CONTROL_SOCKET_READY_POLL_INTERVAL: Duration = Duration::from_millis(50);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum EnsuredAppServerListener {
ReusedExisting,
StartedNew,
}
pub enum AppServerListenerStartup {
Untracked,
Child(std::process::Child),
}
pub async fn start_control_socket_acceptor(
socket_path: AbsolutePathBuf,
@@ -42,6 +57,33 @@ pub async fn start_control_socket_acceptor(
)))
}
pub async fn ensure_control_socket_listener<F, Fut>(
socket_path: AbsolutePathBuf,
startup_lock_path: AbsolutePathBuf,
start_listener: F,
) -> IoResult<EnsuredAppServerListener>
where
F: FnOnce() -> Fut,
Fut: Future<Output = IoResult<AppServerListenerStartup>>,
{
if probe_existing_control_socket(socket_path.as_path()).await? {
return Ok(EnsuredAppServerListener::ReusedExisting);
}
if let Some(parent) = startup_lock_path.as_path().parent() {
codex_uds::prepare_private_socket_directory(parent).await?;
}
let _startup_lock = acquire_app_server_startup_lock(startup_lock_path).await?;
if probe_existing_control_socket(socket_path.as_path()).await? {
return Ok(EnsuredAppServerListener::ReusedExisting);
}
let startup = start_listener().await?;
wait_for_control_socket_listener(socket_path.as_path(), startup).await?;
Ok(EnsuredAppServerListener::StartedNew)
}
async fn run_control_socket_acceptor(
mut listener: UnixListener,
transport_event_tx: mpsc::Sender<TransportEvent>,
@@ -130,6 +172,67 @@ async fn prepare_control_socket_path(socket_path: &Path) -> IoResult<()> {
tokio::fs::remove_file(socket_path).await
}
async fn probe_existing_control_socket(socket_path: &Path) -> IoResult<bool> {
match UnixStream::connect(socket_path).await {
Ok(_stream) => Ok(true),
Err(err)
if matches!(
err.kind(),
ErrorKind::ConnectionRefused | ErrorKind::NotFound
) =>
{
Ok(false)
}
Err(err) => {
if !socket_path.exists() {
return Ok(false);
}
Err(err)
}
}
}
async fn wait_for_control_socket_listener(
socket_path: &Path,
mut startup: AppServerListenerStartup,
) -> IoResult<()> {
loop {
if probe_existing_control_socket(socket_path).await? {
return Ok(());
}
if let AppServerListenerStartup::Child(child) = &mut startup
&& let Some(status) = child.try_wait()?
{
return Err(std::io::Error::other(format!(
"detached app-server listener exited before opening {} with status {status}",
socket_path.display()
)));
}
sleep(CONTROL_SOCKET_READY_POLL_INTERVAL).await;
}
}
struct AppServerStartupLock {
_file: std::fs::File,
}
async fn acquire_app_server_startup_lock(
startup_lock_path: AbsolutePathBuf,
) -> IoResult<AppServerStartupLock> {
tokio::task::spawn_blocking(move || {
let file = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(startup_lock_path.as_path())?;
file.lock()?;
Ok(AppServerStartupLock { _file: file })
})
.await
.map_err(|err| std::io::Error::other(format!("startup lock task failed: {err}")))?
}
#[cfg(unix)]
async fn set_control_socket_permissions(socket_path: &Path) -> IoResult<()> {
use std::os::unix::fs::PermissionsExt;
@@ -1,7 +1,10 @@
use super::AppServerListenerStartup;
use super::AppServerTransport;
use super::CHANNEL_CAPACITY;
use super::EnsuredAppServerListener;
use super::TransportEvent;
use super::app_server_control_socket_path;
use super::ensure_control_socket_listener;
use super::start_control_socket_acceptor;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
@@ -13,6 +16,10 @@ use futures::StreamExt;
use pretty_assertions::assert_eq;
use std::io::Result as IoResult;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use tokio::sync::Notify;
use tokio::sync::mpsc;
use tokio::time::Duration;
use tokio::time::timeout;
@@ -140,6 +147,160 @@ async fn control_socket_acceptor_upgrades_and_forwards_websocket_text_messages_a
assert_socket_path_removed(socket_path.as_path());
}
#[tokio::test]
async fn ensure_control_socket_listener_reuses_a_live_listener() {
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let socket_path = test_socket_path(temp_dir.path());
let lock_path = test_startup_lock_path(temp_dir.path());
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let accept_handle = start_control_socket_acceptor(
socket_path.clone(),
transport_event_tx,
shutdown_token.clone(),
)
.await
.expect("control socket acceptor should start");
let start_attempts = Arc::new(AtomicUsize::new(0));
let listener = ensure_control_socket_listener(socket_path, lock_path, {
let start_attempts = Arc::clone(&start_attempts);
move || async move {
start_attempts.fetch_add(1, Ordering::SeqCst);
Ok(AppServerListenerStartup::Untracked)
}
})
.await
.expect("existing listener should be reused");
assert_eq!(listener, EnsuredAppServerListener::ReusedExisting);
assert_eq!(start_attempts.load(Ordering::SeqCst), 0);
shutdown_token.cancel();
accept_handle.await.expect("acceptor should join");
}
#[tokio::test]
async fn ensure_control_socket_listener_serializes_concurrent_creators() {
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let socket_path = test_socket_path(temp_dir.path());
let lock_path = test_startup_lock_path(temp_dir.path());
let start_attempts = Arc::new(AtomicUsize::new(0));
let ensure_one = ensure_control_socket_listener(socket_path.clone(), lock_path.clone(), {
let socket_path = socket_path.clone();
let start_attempts = Arc::clone(&start_attempts);
move || start_test_listener(socket_path, start_attempts)
});
let ensure_two = ensure_control_socket_listener(socket_path.clone(), lock_path.clone(), {
let socket_path = socket_path.clone();
let start_attempts = Arc::clone(&start_attempts);
move || start_test_listener(socket_path, start_attempts)
});
let ensure_three = ensure_control_socket_listener(socket_path.clone(), lock_path, {
let start_attempts = Arc::clone(&start_attempts);
move || start_test_listener(socket_path, start_attempts)
});
let (one, two, three) = tokio::join!(ensure_one, ensure_two, ensure_three);
let listeners = [
one.expect("first ensure should succeed"),
two.expect("second ensure should succeed"),
three.expect("third ensure should succeed"),
];
assert_eq!(start_attempts.load(Ordering::SeqCst), 1);
assert_eq!(
listeners
.iter()
.filter(|listener| **listener == EnsuredAppServerListener::StartedNew)
.count(),
1
);
}
#[tokio::test]
async fn ensure_control_socket_listener_waits_for_slow_startup_without_spawning_again() {
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let socket_path = test_socket_path(temp_dir.path());
let lock_path = test_startup_lock_path(temp_dir.path());
let start_attempts = Arc::new(AtomicUsize::new(0));
let slow_start_entered = Arc::new(Notify::new());
let allow_slow_start_to_finish = Arc::new(Notify::new());
let ensure_one = tokio::spawn(ensure_control_socket_listener(
socket_path.clone(),
lock_path.clone(),
{
let socket_path = socket_path.clone();
let start_attempts = Arc::clone(&start_attempts);
let slow_start_entered = Arc::clone(&slow_start_entered);
let allow_slow_start_to_finish = Arc::clone(&allow_slow_start_to_finish);
move || async move {
slow_start_entered.notify_one();
allow_slow_start_to_finish.notified().await;
start_test_listener(socket_path, start_attempts).await
}
},
));
slow_start_entered.notified().await;
let ensure_two = ensure_control_socket_listener(socket_path.clone(), lock_path.clone(), {
let socket_path = socket_path.clone();
let start_attempts = Arc::clone(&start_attempts);
move || start_test_listener(socket_path, start_attempts)
});
let ensure_three = ensure_control_socket_listener(socket_path, lock_path, {
move || async move {
Err(std::io::Error::other(
"third ensure must not start another listener",
))
}
});
allow_slow_start_to_finish.notify_one();
let (one, two, three) = tokio::join!(ensure_one, ensure_two, ensure_three);
let listeners = [
one.expect("first ensure task should join")
.expect("first ensure should succeed"),
two.expect("second ensure should succeed"),
three.expect("third ensure should succeed"),
];
assert_eq!(start_attempts.load(Ordering::SeqCst), 1);
assert_eq!(
listeners
.iter()
.filter(|listener| **listener == EnsuredAppServerListener::StartedNew)
.count(),
1
);
}
#[cfg(unix)]
#[tokio::test]
async fn ensure_control_socket_listener_reports_child_exit_before_bind() {
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let socket_path = test_socket_path(temp_dir.path());
let lock_path = test_startup_lock_path(temp_dir.path());
let err = ensure_control_socket_listener(socket_path, lock_path, move || async move {
let child = std::process::Command::new("sh")
.arg("-c")
.arg("exit 17")
.spawn()?;
Ok(AppServerListenerStartup::Child(child))
})
.await
.expect_err("child exit before socket bind should surface as an error");
assert!(
err.to_string()
.contains("detached app-server listener exited before opening")
);
}
#[cfg(unix)]
#[tokio::test]
async fn control_socket_file_is_private_after_bind() {
@@ -185,6 +346,28 @@ fn test_socket_path(temp_dir: &Path) -> AbsolutePathBuf {
.expect("socket path should resolve")
}
fn test_startup_lock_path(temp_dir: &Path) -> AbsolutePathBuf {
AbsolutePathBuf::from_absolute_path(
temp_dir
.join("app-server-control")
.join("app-server-startup.lock"),
)
.expect("startup lock path should resolve")
}
async fn start_test_listener(
socket_path: AbsolutePathBuf,
start_attempts: Arc<AtomicUsize>,
) -> IoResult<AppServerListenerStartup> {
start_attempts.fetch_add(1, Ordering::SeqCst);
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let _accept_handle =
start_control_socket_acceptor(socket_path, transport_event_tx, CancellationToken::new())
.await?;
Ok(AppServerListenerStartup::Untracked)
}
async fn connect_to_socket(socket_path: &Path) -> IoResult<UnixStream> {
UnixStream::connect(socket_path).await
}
+3
View File
@@ -100,11 +100,14 @@ mod transport;
pub use crate::error_code::INPUT_TOO_LARGE_ERROR_CODE;
pub use crate::error_code::INVALID_PARAMS_ERROR_CODE;
pub use crate::transport::AppServerListenerStartup;
pub use crate::transport::AppServerTransport;
pub use crate::transport::app_server_control_socket_path;
pub use crate::transport::app_server_startup_lock_path;
pub use crate::transport::auth::AppServerWebsocketAuthArgs;
pub use crate::transport::auth::AppServerWebsocketAuthSettings;
pub use crate::transport::auth::WebsocketAuthCliMode;
pub use crate::transport::ensure_control_socket_listener;
const LOG_FORMAT_ENV_VAR: &str = "LOG_FORMAT";
const OTEL_SERVICE_NAME: &str = "codex-app-server";
+3
View File
@@ -12,6 +12,7 @@ use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::warn;
pub use codex_app_server_transport::AppServerListenerStartup;
pub use codex_app_server_transport::AppServerTransport;
pub(crate) use codex_app_server_transport::CHANNEL_CAPACITY;
pub(crate) use codex_app_server_transport::ConnectionId;
@@ -23,7 +24,9 @@ pub(crate) use codex_app_server_transport::RemoteControlStartConfig;
pub(crate) use codex_app_server_transport::RemoteControlUnavailable;
pub(crate) use codex_app_server_transport::TransportEvent;
pub use codex_app_server_transport::app_server_control_socket_path;
pub use codex_app_server_transport::app_server_startup_lock_path;
pub use codex_app_server_transport::auth;
pub use codex_app_server_transport::ensure_control_socket_listener;
pub(crate) use codex_app_server_transport::start_control_socket_acceptor;
pub(crate) use codex_app_server_transport::start_remote_control;
pub(crate) use codex_app_server_transport::start_stdio_connection;
+139 -2
View File
@@ -42,6 +42,7 @@ use codex_utils_cli::resume_hint;
use owo_colors::OwoColorize;
use std::io::IsTerminal;
use std::path::PathBuf;
use std::process::Stdio;
use supports_color::Stream;
#[cfg(any(target_os = "macos", target_os = "windows"))]
@@ -558,6 +559,10 @@ struct AppServerProxyCommand {
/// Path to the app-server Unix domain socket to connect to.
#[arg(long = "sock", value_name = "SOCKET_PATH", value_parser = parse_socket_path)]
socket_path: Option<AbsolutePathBuf>,
/// Start the default app-server Unix socket listener if it is not already running.
#[arg(long = "ensure-listener")]
ensure_listener: bool,
}
#[derive(Debug, Args)]
@@ -612,6 +617,42 @@ fn parse_socket_path(raw: &str) -> Result<AbsolutePathBuf, String> {
.map_err(|err| format!("failed to resolve socket path `{raw}`: {err}"))
}
async fn start_detached_app_server_listener(
socket_path: AbsolutePathBuf,
log_path: PathBuf,
cli_config_overrides: CliConfigOverrides,
analytics_default_enabled: bool,
) -> std::io::Result<codex_app_server::AppServerListenerStartup> {
let log_file = std::fs::File::create(&log_path)?;
let log_file_stderr = log_file.try_clone()?;
let current_exe = std::env::current_exe()?;
#[cfg(unix)]
let mut command = {
let mut command = std::process::Command::new("nohup");
command.arg(current_exe);
command
};
#[cfg(not(unix))]
let mut command = std::process::Command::new(current_exe);
for raw_override in cli_config_overrides.raw_overrides {
command.arg("-c").arg(raw_override);
}
command.arg("app-server");
if analytics_default_enabled {
command.arg("--analytics-default-enabled");
}
let child = command
.arg("--listen")
.arg(format!("unix://{}", socket_path.display()))
.stdin(Stdio::null())
.stdout(Stdio::from(log_file))
.stderr(Stdio::from(log_file_stderr))
.spawn()?;
Ok(codex_app_server::AppServerListenerStartup::Child(child))
}
fn format_exit_messages(exit_info: AppExitInfo, color_enabled: bool) -> Vec<String> {
let AppExitInfo {
token_usage,
@@ -1025,6 +1066,10 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
}
},
Some(AppServerSubcommand::Proxy(proxy_cli)) => {
reject_config_overrides_for_ensure_listener(
&proxy_cli,
&root_config_overrides,
)?;
let socket_path = match proxy_cli.socket_path {
Some(socket_path) => socket_path,
None => {
@@ -1032,6 +1077,42 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
codex_app_server::app_server_control_socket_path(&codex_home)?
}
};
if proxy_cli.ensure_listener {
let codex_home = find_codex_home()?;
let startup_lock_path =
codex_app_server::app_server_startup_lock_path(&codex_home)?;
let startup_log_path = startup_lock_path
.as_path()
.parent()
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!(
"app-server startup lock path has no parent: {}",
startup_lock_path.display()
),
)
})?
.join("app-server.log");
codex_app_server::ensure_control_socket_listener(
socket_path.clone(),
startup_lock_path,
{
let socket_path = socket_path.clone();
let startup_log_path = startup_log_path.clone();
let cli_config_overrides = root_config_overrides.clone();
move || {
start_detached_app_server_listener(
socket_path,
startup_log_path,
cli_config_overrides,
analytics_default_enabled,
)
}
},
)
.await?;
}
codex_stdio_to_uds::run(socket_path.as_path()).await?;
}
Some(AppServerSubcommand::GenerateTs(gen_cli)) => {
@@ -1966,6 +2047,18 @@ async fn print_app_server_remote_control_output(
Ok(())
}
fn reject_config_overrides_for_ensure_listener(
proxy_cli: &AppServerProxyCommand,
root_config_overrides: &CliConfigOverrides,
) -> anyhow::Result<()> {
if proxy_cli.ensure_listener && !root_config_overrides.raw_overrides.is_empty() {
anyhow::bail!(
"`codex app-server proxy --ensure-listener` cannot use root `-c`/`--config` overrides because it may reuse an existing app-server listener; start a dedicated listener or pass per-thread settings through the app-server protocol instead"
);
}
Ok(())
}
fn read_remote_auth_token_from_env_var_with<F>(
env_var_name: &str,
get_var: F,
@@ -3135,7 +3228,21 @@ mod tests {
assert!(matches!(
app_server.subcommand,
Some(AppServerSubcommand::Proxy(AppServerProxyCommand {
socket_path: None
socket_path: None,
ensure_listener: false,
}))
));
}
#[test]
fn app_server_proxy_ensure_listener_parses() {
let app_server =
app_server_from_args(["codex", "app-server", "proxy", "--ensure-listener"].as_ref());
assert!(matches!(
app_server.subcommand,
Some(AppServerSubcommand::Proxy(AppServerProxyCommand {
socket_path: None,
ensure_listener: true,
}))
));
}
@@ -3222,7 +3329,10 @@ mod tests {
#[test]
fn reject_remote_auth_token_env_for_app_server_proxy() {
let subcommand = AppServerSubcommand::Proxy(AppServerProxyCommand { socket_path: None });
let subcommand = AppServerSubcommand::Proxy(AppServerProxyCommand {
socket_path: None,
ensure_listener: false,
});
let err = reject_remote_mode_for_app_server_subcommand(
/*remote*/ None,
Some("CODEX_REMOTE_AUTH_TOKEN"),
@@ -3246,6 +3356,33 @@ mod tests {
assert!(err.to_string().contains("app-server daemon version"));
}
#[test]
fn app_server_proxy_ensure_listener_rejects_root_config_overrides() {
let proxy_cli = AppServerProxyCommand {
socket_path: None,
ensure_listener: true,
};
let overrides = CliConfigOverrides {
raw_overrides: vec!["model=o3".to_string()],
};
let err = reject_config_overrides_for_ensure_listener(&proxy_cli, &overrides)
.expect_err("ensure-listener should reject process-scoped root overrides");
assert!(err.to_string().contains("--ensure-listener"));
}
#[test]
fn app_server_proxy_attach_only_accepts_root_config_overrides() {
let proxy_cli = AppServerProxyCommand {
socket_path: None,
ensure_listener: false,
};
let overrides = CliConfigOverrides {
raw_overrides: vec!["model=o3".to_string()],
};
reject_config_overrides_for_ensure_listener(&proxy_cli, &overrides)
.expect("attach-only proxy should preserve existing override behavior");
}
#[test]
fn app_server_capability_token_flags_parse() {
let app_server = app_server_from_args(