mirror of
https://github.com/openai/codex.git
synced 2026-10-19 08:16:48 +00:00
Persist remote-control RPC preferences in managed daemon settings (#52304)
## Why Durable remote-control RPC changes need to update the managed daemon's launch settings so future launches use the saved preference. ## What changed - Save non-ephemeral `remoteControl/enable` and `remoteControl/disable` requests to `remoteControlEnabled` in `app-server-daemon/settings.json` for managed daemons. - Preserve other settings and write updates atomically. Skip persistence when `daemon.lock` is held so concurrent lifecycle operations take precedence without blocking the RPC. - Leave daemon settings untouched for ephemeral requests and unmanaged servers. ## Testing Add integration coverage for durable enable and disable, ephemeral requests, and lock contention. Extend the unmanaged-server test to verify that no daemon settings file is created. GitOrigin-RevId: 2baf261ed62f7c7ad87c9779ffb91c59e923f0f8
This commit is contained in:
@@ -973,11 +973,17 @@ pub async fn run_main_with_transport_options(
|
||||
});
|
||||
|
||||
let recovery_file = daemon_recovery_file_path(&config.codex_home);
|
||||
let daemon_settings_file = managed_daemon.then(|| {
|
||||
config
|
||||
.codex_home
|
||||
.as_path()
|
||||
.join("app-server-daemon/settings.json")
|
||||
});
|
||||
let processor_handle = tokio::spawn({
|
||||
let auth_manager = Arc::clone(&auth_manager);
|
||||
let initialize_notification_sender = outgoing_message_sender.clone();
|
||||
let outbound_control_tx = outbound_control_tx;
|
||||
let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs {
|
||||
let mut processor = MessageProcessor::new(MessageProcessorArgs {
|
||||
outgoing: outgoing_message_sender,
|
||||
analytics_events_client,
|
||||
arg0_paths,
|
||||
@@ -1002,7 +1008,9 @@ pub async fn run_main_with_transport_options(
|
||||
PluginStartupTasks::Start
|
||||
)
|
||||
.then_some(plugin_startup_config),
|
||||
}));
|
||||
});
|
||||
processor.remote_control_processor.daemon_settings_file = daemon_settings_file;
|
||||
let processor = Arc::new(processor);
|
||||
let mut thread_created_rx = processor.thread_created_receiver();
|
||||
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
|
||||
let mut active_admissions_rx = processor.turn_admission.subscribe_active();
|
||||
|
||||
@@ -163,7 +163,7 @@ pub(crate) struct MessageProcessor {
|
||||
mcp_processor: McpRequestProcessor,
|
||||
plugin_processor: PluginRequestProcessor,
|
||||
project_processor: ProjectRequestProcessor,
|
||||
remote_control_processor: RemoteControlRequestProcessor,
|
||||
pub(super) remote_control_processor: RemoteControlRequestProcessor,
|
||||
search_processor: SearchRequestProcessor,
|
||||
thread_goal_processor: ThreadGoalRequestProcessor,
|
||||
thread_queue_processor: ThreadQueueRequestProcessor,
|
||||
|
||||
@@ -15,17 +15,26 @@ use codex_app_server_protocol::RemoteControlPairingStartResponse;
|
||||
use codex_app_server_protocol::RemoteControlPairingStatusParams;
|
||||
use codex_app_server_protocol::RemoteControlPairingStatusResponse;
|
||||
use codex_app_server_protocol::RemoteControlStatusReadResponse;
|
||||
use codex_core::path_utils::write_atomically;
|
||||
use serde_json::Map;
|
||||
use serde_json::Value;
|
||||
use std::fs;
|
||||
use std::fs::OpenOptions;
|
||||
use std::fs::TryLockError;
|
||||
use std::io;
|
||||
use std::path::PathBuf;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct RemoteControlRequestProcessor {
|
||||
remote_control_handle: Option<RemoteControlHandle>,
|
||||
pub(crate) daemon_settings_file: Option<PathBuf>,
|
||||
}
|
||||
|
||||
impl RemoteControlRequestProcessor {
|
||||
pub(crate) fn new(remote_control_handle: Option<RemoteControlHandle>) -> Self {
|
||||
Self {
|
||||
remote_control_handle,
|
||||
daemon_settings_file: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,6 +52,11 @@ impl RemoteControlRequestProcessor {
|
||||
.await
|
||||
.map_err(map_update_error)?
|
||||
};
|
||||
if !ephemeral {
|
||||
self.persist_daemon_preference(/*enabled*/ true)
|
||||
.await
|
||||
.map_err(map_update_error)?;
|
||||
}
|
||||
Ok(RemoteControlEnableResponse::from(status))
|
||||
}
|
||||
|
||||
@@ -60,9 +74,47 @@ impl RemoteControlRequestProcessor {
|
||||
.await
|
||||
.map_err(map_update_error)?
|
||||
};
|
||||
if !ephemeral {
|
||||
self.persist_daemon_preference(/*enabled*/ false)
|
||||
.await
|
||||
.map_err(map_update_error)?;
|
||||
}
|
||||
Ok(RemoteControlDisableResponse::from(status))
|
||||
}
|
||||
|
||||
async fn persist_daemon_preference(&self, enabled: bool) -> io::Result<()> {
|
||||
let Some(path) = self.daemon_settings_file.clone() else {
|
||||
return Ok(());
|
||||
};
|
||||
tokio::task::spawn_blocking(move || {
|
||||
if let Some(parent) = path.parent() {
|
||||
fs::create_dir_all(parent)?;
|
||||
}
|
||||
let lock = OpenOptions::new()
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.read(true)
|
||||
.write(true)
|
||||
.open(path.with_file_name("daemon.lock"))?;
|
||||
// The daemon may be waiting for this RPC during shutdown. Never wait
|
||||
// for it: a concurrent lifecycle operation wins over RPC persistence.
|
||||
match lock.try_lock() {
|
||||
Ok(()) => {}
|
||||
Err(TryLockError::WouldBlock) => return Ok(()),
|
||||
Err(TryLockError::Error(err)) => return Err(err),
|
||||
}
|
||||
let mut settings: Map<String, Value> = match fs::read(&path) {
|
||||
Ok(contents) => serde_json::from_slice(&contents)?,
|
||||
Err(err) if err.kind() == io::ErrorKind::NotFound => Map::new(),
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
settings.insert("remoteControlEnabled".to_string(), Value::Bool(enabled));
|
||||
write_atomically(&path, &serde_json::to_string_pretty(&settings)?)
|
||||
})
|
||||
.await
|
||||
.map_err(io::Error::other)?
|
||||
}
|
||||
|
||||
pub(crate) fn status_read(&self) -> Result<RemoteControlStatusReadResponse, JSONRPCErrorError> {
|
||||
let status = self.handle()?.status();
|
||||
Ok(RemoteControlStatusReadResponse {
|
||||
|
||||
@@ -801,19 +801,25 @@ async fn connect_default_daemon_client(socket_path: &Path) -> Result<WebSocketSt
|
||||
connect_daemon_client(socket_path, InitializeCapabilities::default()).await
|
||||
}
|
||||
|
||||
async fn connect_daemon_client(
|
||||
pub(super) async fn connect_daemon_client(
|
||||
socket_path: &Path,
|
||||
capabilities: InitializeCapabilities,
|
||||
) -> Result<WebSocketStream<UnixStream>> {
|
||||
connect_initialized(socket_path, capabilities, "daemon_recovery_test", "0.1.0").await
|
||||
}
|
||||
|
||||
fn spawn_server(home: &Path, socket_path: &Path) -> Result<Child> {
|
||||
pub(super) fn spawn_server(home: &Path, socket_path: &Path) -> Result<Child> {
|
||||
let binary = codex_utils_cargo_bin::cargo_bin("codex-app-server")?;
|
||||
Ok(Command::new(binary)
|
||||
.args(["--listen", &format!("unix://{}", socket_path.display())])
|
||||
.arg(DISABLE_PLUGIN_STARTUP_TASKS_ARG)
|
||||
.env("CODEX_HOME", home)
|
||||
// Match TestAppServer's isolation from the host's managed configuration.
|
||||
.env(
|
||||
"CODEX_APP_SERVER_MANAGED_CONFIG_PATH",
|
||||
home.join("managed_config.toml"),
|
||||
)
|
||||
.current_dir(home)
|
||||
.arg("--managed-daemon")
|
||||
.env(
|
||||
codex_app_server_transport::DAEMON_SHUTDOWN_SOCKET_ENV,
|
||||
@@ -888,7 +894,7 @@ async fn connect_initialized(
|
||||
Ok(websocket)
|
||||
}
|
||||
|
||||
async fn request(
|
||||
pub(super) async fn request(
|
||||
websocket: &mut WebSocketStream<UnixStream>,
|
||||
id: i64,
|
||||
method: &str,
|
||||
|
||||
@@ -19,6 +19,7 @@ use codex_app_server::PluginStartupTasks;
|
||||
use codex_app_server::RemoteControlStartupMode;
|
||||
use codex_app_server::run_main_with_transport_options;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::InitializeCapabilities;
|
||||
use codex_app_server_protocol::JSONRPCError;
|
||||
use codex_app_server_protocol::JSONRPCResponse;
|
||||
use codex_app_server_protocol::RemoteControlClient;
|
||||
@@ -713,6 +714,84 @@ async fn rpc_updates_durable_preference_but_ephemeral_does_not() -> Result<()> {
|
||||
Some(false)
|
||||
);
|
||||
|
||||
assert!(
|
||||
!codex_home
|
||||
.path()
|
||||
.join("app-server-daemon/settings.json")
|
||||
.exists()
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn managed_rpc_persists_unless_daemon_lock_is_busy() -> Result<()> {
|
||||
use super::daemon_update_recovery::connect_daemon_client;
|
||||
use super::daemon_update_recovery::request;
|
||||
use super::daemon_update_recovery::spawn_server;
|
||||
use serde_json::json;
|
||||
|
||||
let home = TempDir::new()?;
|
||||
let mut backend = BlockingRemoteControlBackend::start(home.path()).await?;
|
||||
let socket = home.path().join("control/server.sock");
|
||||
let settings_file = home.path().join("app-server-daemon/settings.json");
|
||||
let _server = spawn_server(home.path(), &socket)?;
|
||||
let mut client = connect_daemon_client(
|
||||
&socket,
|
||||
InitializeCapabilities {
|
||||
experimental_api: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
let enable = request(
|
||||
&mut client,
|
||||
/*id*/ 2,
|
||||
"remoteControl/enable",
|
||||
json!(null),
|
||||
);
|
||||
let enroll = async {
|
||||
timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??;
|
||||
backend.complete_enrollment()
|
||||
};
|
||||
tokio::try_join!(enable, enroll)?;
|
||||
let mut settings = tokio::fs::read(&settings_file).await?;
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<serde_json::Value>(&settings)?,
|
||||
json!({"remoteControlEnabled": true})
|
||||
);
|
||||
for (id, method, ephemeral) in [
|
||||
(3, "remoteControl/disable", true),
|
||||
(4, "remoteControl/disable", false),
|
||||
(5, "remoteControl/enable", true),
|
||||
] {
|
||||
request(&mut client, id, method, json!({"ephemeral": ephemeral})).await?;
|
||||
let updated = tokio::fs::read(&settings_file).await?;
|
||||
if ephemeral {
|
||||
assert_eq!(updated, settings);
|
||||
} else {
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<serde_json::Value>(&updated)?,
|
||||
json!({"remoteControlEnabled": false})
|
||||
);
|
||||
}
|
||||
settings = updated;
|
||||
}
|
||||
// A lifecycle operation holding daemon.lock must not block the RPC or
|
||||
// allow it to overwrite other settings being changed by the daemon.
|
||||
let lock = std::fs::OpenOptions::new()
|
||||
.read(true)
|
||||
.write(true)
|
||||
.open(settings_file.with_file_name("daemon.lock"))?;
|
||||
lock.lock()?;
|
||||
request(
|
||||
&mut client,
|
||||
/*id*/ 6,
|
||||
"remoteControl/enable",
|
||||
json!(null),
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(tokio::fs::read(&settings_file).await?, settings);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user