openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-14 21:40:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into pakrym/core-cwd-absolutepathbuf

Browse Source 
# Conflicts:
#	codex-rs/core/src/tools/runtimes/apply_patch.rs
#	codex-rs/core/src/tools/runtimes/apply_patch_tests.rs
#	codex-rs/core/tests/common/lib.rs
#	codex-rs/tui/src/test_support.rs
This commit is contained in:
pakrym-oai
2026-04-07 14:34:21 -07:00
parent 69eb799fe0 e9702411ab
commit 9c04f54e52
71 changed files with 1449 additions and 621 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/README.md
+2 -2
View File
@@ -169,7 +169,7 @@ Example with notification opt-out:
- `fs/readDirectory` — list direct child entries for an absolute directory path; each entry contains `fileName`, `isDirectory`, and `isFile`, and `fileName` is just the child name, not a path.
- `fs/remove` — remove an absolute file or directory tree; `recursive` and `force` default to `true`.
- `fs/copy` — copy between absolute paths; directory copies require `recursive: true`.
- `fs/watch` — subscribe this connection to filesystem change notifications for an absolute file or directory path; returns a `watchId` and canonicalized `path`.
- `fs/watch` — subscribe this connection to filesystem change notifications for an absolute file or directory path and caller-provided `watchId`; returns the canonicalized `path`.
- `fs/unwatch` — stop sending notifications for a prior `fs/watch`; returns `{}`.
- `fs/changed` — notification emitted when watched paths change, including the `watchId` and `changedPaths`.
- `model/list` — list available models (set `includeHidden: true` to include entries with `hidden: true`), with reasoning effort options, optional legacy `upgrade` model ids, optional `upgradeInfo` metadata (`model`, `upgradeCopy`, `modelLink`, `migrationMarkdown`), and optional `availabilityNux` metadata.
@@ -816,10 +816,10 @@ All filesystem paths in this section must be absolute.
```json
{ "method": "fs/watch", "id": 44, "params": {
"watchId": "0195ec6b-1d6f-7c2e-8c7a-56f2c4a8b9d1",
"path": "/Users/me/project/.git/HEAD"
} }
{ "id": 44, "result": {
"watchId": "0195ec6b-1d6f-7c2e-8c7a-56f2c4a8b9d1",
"path": "/Users/me/project/.git/HEAD"
} }
{ "method": "fs/changed", "params": {
codex-rs/app-server/src/bespoke_event_handling.rs
+7 -1
View File
@@ -2843,6 +2843,7 @@ async fn construct_mcp_tool_call_end_notification(
Some(McpToolCallResult {
content: value.content.clone(),
structured_content: value.structured_content.clone(),
meta: value.meta.clone(),
}),
None,
),
@@ -4233,7 +4234,9 @@ mod tests {
content: content.clone(),
is_error: Some(false),
structured_content: None,
meta: None,
meta: Some(serde_json::json!({
"ui/resourceUri": "ui://widget/list-resources.html"
})),
};
let end_event = McpToolCallEndEvent {
@@ -4268,6 +4271,9 @@ mod tests {
result: Some(McpToolCallResult {
content,
structured_content: None,
meta: Some(serde_json::json!({
"ui/resourceUri": "ui://widget/list-resources.html"
})),
}),
error: None,
duration_ms: Some(0),
codex-rs/app-server/src/fs_watch.rs
+81 -33
View File
@@ -1,3 +1,4 @@
use crate::fs_api::invalid_request;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingMessageSender;
use codex_app_server_protocol::FsChangedNotification;
@@ -16,6 +17,7 @@ use codex_core::file_watcher::WatchRegistration;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::hash_map::Entry;
use std::hash::Hash;
use std::path::PathBuf;
use std::sync::Arc;
@@ -26,7 +28,6 @@ use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::time::Instant;
use tracing::warn;
use uuid::Uuid;
const FS_CHANGED_NOTIFICATION_DEBOUNCE: Duration = Duration::from_millis(200);
@@ -120,7 +121,11 @@ impl FsWatchManager {
connection_id: ConnectionId,
params: FsWatchParams,
) -> Result<FsWatchResponse, JSONRPCErrorError> {
let watch_id = Uuid::now_v7().to_string();
let watch_id = params.watch_id;
let watch_key = WatchKey {
connection_id,
watch_id: watch_id.clone(),
};
let outgoing = self.outgoing.clone();
let (subscriber, rx) = self.file_watcher.add_subscriber();
let watch_root = params.path.to_path_buf().clone();
@@ -130,17 +135,20 @@ impl FsWatchManager {
}]);
let (terminate_tx, terminate_rx) = oneshot::channel();
self.state.lock().await.entries.insert(
WatchKey {
connection_id,
watch_id: watch_id.clone(),
},
WatchEntry {
terminate_tx,
_subscriber: subscriber,
_registration: registration,
},
);
match self.state.lock().await.entries.entry(watch_key) {
Entry::Occupied(_) => {
return Err(invalid_request(format!(
"watchId already exists: {watch_id}"
)));
}
Entry::Vacant(entry) => {
entry.insert(WatchEntry {
terminate_tx,
_subscriber: subscriber,
_registration: registration,
});
}
}
let task_watch_id = watch_id.clone();
tokio::spawn(async move {
@@ -175,10 +183,7 @@ impl FsWatchManager {
}
});
Ok(FsWatchResponse {
watch_id,
path: params.path,
})
Ok(FsWatchResponse { path: params.path })
}
pub(crate) async fn unwatch(
@@ -216,7 +221,6 @@ mod tests {
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use uuid::Version;
fn absolute_path(path: PathBuf) -> AbsolutePathBuf {
assert!(
@@ -237,28 +241,33 @@ mod tests {
}
#[tokio::test]
async fn watch_returns_a_v7_id_and_tracks_the_owner_scoped_entry() {
async fn watch_uses_client_id_and_tracks_the_owner_scoped_entry() {
let temp_dir = TempDir::new().expect("temp dir");
let head_path = temp_dir.path().join("HEAD");
std::fs::write(&head_path, "ref: refs/heads/main\n").expect("write HEAD");
let manager = manager_with_noop_watcher();
let path = absolute_path(head_path);
let watch_id = "watch-head".to_string();
let response = manager
.watch(ConnectionId(1), FsWatchParams { path: path.clone() })
.watch(
ConnectionId(1),
FsWatchParams {
watch_id: watch_id.clone(),
path: path.clone(),
},
)
.await
.expect("watch should succeed");
assert_eq!(response.path, path);
let watch_id = Uuid::parse_str(&response.watch_id).expect("watch id should be a UUID");
assert_eq!(watch_id.get_version(), Some(Version::SortRand));
let state = manager.state.lock().await;
assert_eq!(
state.entries.keys().cloned().collect::<HashSet<_>>(),
HashSet::from([WatchKey {
connection_id: ConnectionId(1),
watch_id: response.watch_id,
watch_id,
}])
);
}
@@ -270,10 +279,11 @@ mod tests {
std::fs::write(&head_path, "ref: refs/heads/main\n").expect("write HEAD");
let manager = manager_with_noop_watcher();
let response = manager
manager
.watch(
ConnectionId(1),
FsWatchParams {
watch_id: "watch-head".to_string(),
path: absolute_path(head_path),
},
)
@@ -281,14 +291,14 @@ mod tests {
.expect("watch should succeed");
let watch_key = WatchKey {
connection_id: ConnectionId(1),
watch_id: response.watch_id.clone(),
watch_id: "watch-head".to_string(),
};
manager
.unwatch(
ConnectionId(2),
FsUnwatchParams {
watch_id: response.watch_id.clone(),
watch_id: "watch-head".to_string(),
},
)
.await
@@ -299,7 +309,7 @@ mod tests {
.unwatch(
ConnectionId(1),
FsUnwatchParams {
watch_id: response.watch_id,
watch_id: "watch-head".to_string(),
},
)
.await
@@ -307,6 +317,41 @@ mod tests {
assert!(!manager.state.lock().await.entries.contains_key(&watch_key));
}
#[tokio::test]
async fn watch_rejects_duplicate_id_for_the_same_connection() {
let temp_dir = TempDir::new().expect("temp dir");
let head_path = temp_dir.path().join("HEAD");
let fetch_head_path = temp_dir.path().join("FETCH_HEAD");
std::fs::write(&head_path, "ref: refs/heads/main\n").expect("write HEAD");
std::fs::write(&fetch_head_path, "old-fetch\n").expect("write FETCH_HEAD");
let manager = manager_with_noop_watcher();
manager
.watch(
ConnectionId(1),
FsWatchParams {
watch_id: "watch-head".to_string(),
path: absolute_path(head_path),
},
)
.await
.expect("first watch should succeed");
let error = manager
.watch(
ConnectionId(1),
FsWatchParams {
watch_id: "watch-head".to_string(),
path: absolute_path(fetch_head_path),
},
)
.await
.expect_err("duplicate watch should fail");
assert_eq!(error.message, "watchId already exists: watch-head");
assert_eq!(manager.state.lock().await.entries.len(), 1);
}
#[tokio::test]
async fn connection_closed_removes_only_that_connections_watches() {
let temp_dir = TempDir::new().expect("temp dir");
@@ -318,28 +363,31 @@ mod tests {
std::fs::write(&packed_refs_path, "refs\n").expect("write packed-refs");
let manager = manager_with_noop_watcher();
let response_1 = manager
let response = manager
.watch(
ConnectionId(1),
FsWatchParams {
path: absolute_path(head_path),
watch_id: "watch-head".to_string(),
path: absolute_path(head_path.clone()),
},
)
.await
.expect("first watch should succeed");
let response_2 = manager
manager
.watch(
ConnectionId(1),
FsWatchParams {
watch_id: "watch-fetch-head".to_string(),
path: absolute_path(fetch_head_path),
},
)
.await
.expect("second watch should succeed");
let response_3 = manager
manager
.watch(
ConnectionId(2),
FsWatchParams {
watch_id: "watch-packed-refs".to_string(),
path: absolute_path(packed_refs_path),
},
)
@@ -359,9 +407,9 @@ mod tests {
.collect::<HashSet<_>>(),
HashSet::from([WatchKey {
connection_id: ConnectionId(2),
watch_id: response_3.watch_id,
watch_id: "watch-packed-refs".to_string(),
}])
);
assert_ne!(response_1.watch_id, response_2.watch_id);
assert_eq!(response.path, absolute_path(head_path));
}
}
codex-rs/app-server/src/in_process.rs
+1
View File
@@ -398,6 +398,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
session_source: args.session_source,
auth_manager,
rpc_transport: AppServerRpcTransport::InProcess,
remote_control_handle: None,
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut session = ConnectionSessionState::default();
codex-rs/app-server/src/lib.rs
+15 -13
View File
@@ -564,25 +564,26 @@ pub async fn run_main_with_transport(
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false);
if config.features.enabled(Feature::RemoteControl) {
let accept_handle = start_remote_control(
config.chatgpt_base_url.clone(),
state_db.clone(),
auth_manager.clone(),
transport_event_tx.clone(),
transport_shutdown_token.clone(),
app_server_client_name_rx,
)
.await?;
transport_accept_handles.push(accept_handle);
}
if transport_accept_handles.is_empty() {
let remote_control_enabled = config.features.enabled(Feature::RemoteControl);
if transport_accept_handles.is_empty() && !remote_control_enabled {
return Err(std::io::Error::new(
ErrorKind::InvalidInput,
"no transport configured; use --listen or enable remote control",
));
}
let (remote_control_accept_handle, remote_control_handle) = start_remote_control(
config.chatgpt_base_url.clone(),
state_db.clone(),
auth_manager.clone(),
transport_event_tx.clone(),
transport_shutdown_token.clone(),
app_server_client_name_rx,
remote_control_enabled,
)
.await?;
transport_accept_handles.push(remote_control_accept_handle);
let outbound_handle = tokio::spawn(async move {
let mut outbound_connections = HashMap::<ConnectionId, OutboundConnectionState>::new();
loop {
@@ -659,6 +660,7 @@ pub async fn run_main_with_transport(
session_source,
auth_manager,
rpc_transport: analytics_rpc_transport(transport),
remote_control_handle: Some(remote_control_handle),
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
codex-rs/app-server/src/message_processor.rs
+29 -1
View File
@@ -19,6 +19,7 @@ use crate::outgoing_message::ConnectionRequestId;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::RequestContext;
use crate::transport::AppServerTransport;
use crate::transport::RemoteControlHandle;
use async_trait::async_trait;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppServerRpcTransport;
@@ -170,6 +171,7 @@ pub(crate) struct MessageProcessor {
config: Arc<Config>,
config_warnings: Arc<Vec<ConfigWarningNotification>>,
rpc_transport: AppServerRpcTransport,
remote_control_handle: Option<RemoteControlHandle>,
}
#[derive(Clone, Debug, Default)]
@@ -195,6 +197,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) session_source: SessionSource,
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) rpc_transport: AppServerRpcTransport,
pub(crate) remote_control_handle: Option<RemoteControlHandle>,
}
impl MessageProcessor {
@@ -215,6 +218,7 @@ impl MessageProcessor {
session_source,
auth_manager,
rpc_transport,
remote_control_handle,
} = args;
auth_manager.set_external_auth(Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(),
@@ -285,6 +289,7 @@ impl MessageProcessor {
config,
config_warnings: Arc::new(config_warnings),
rpc_transport,
remote_control_handle,
}
}
@@ -969,13 +974,36 @@ impl MessageProcessor {
) {
match result {
Ok(response) => {
self.codex_message_processor.handle_config_mutation();
self.handle_config_mutation().await;
self.outgoing.send_response(request_id, response).await;
}
Err(error) => self.outgoing.send_error(request_id, error).await,
}
}
async fn handle_config_mutation(&self) {
self.codex_message_processor.handle_config_mutation();
let Some(remote_control_handle) = &self.remote_control_handle else {
return;
};
match self
.config_api
.load_latest_config(/*fallback_cwd*/ None)
.await
{
Ok(config) => {
remote_control_handle.set_enabled(config.features.enabled(Feature::RemoteControl));
}
Err(error) => {
tracing::warn!(
"failed to load config for remote control enablement refresh after config mutation: {}",
error.message
);
}
}
}
async fn handle_config_requirements_read(&self, request_id: ConnectionRequestId) {
match self.config_api.config_requirements_read().await {
Ok(response) => self.outgoing.send_response(request_id, response).await,
codex-rs/app-server/src/message_processor/tracing_tests.rs
+1
View File
@@ -251,6 +251,7 @@ fn build_test_processor(
session_source: SessionSource::VSCode,
auth_manager,
rpc_transport: AppServerRpcTransport::Stdio,
remote_control_handle: None,
});
(processor, outgoing_rx)
}
codex-rs/app-server/src/transport/mod.rs
+1
View File
@@ -33,6 +33,7 @@ mod remote_control;
mod stdio;
mod websocket;
pub(crate) use remote_control::RemoteControlHandle;
pub(crate) use remote_control::start_remote_control;
pub(crate) use stdio::start_stdio_connection;
pub(crate) use websocket::start_websocket_acceptor;
codex-rs/app-server/src/transport/remote_control/mod.rs
+38 -5
View File
@@ -19,6 +19,7 @@ use std::io;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::sync::watch;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
@@ -29,6 +30,21 @@ pub(super) struct QueuedServerEnvelope {
pub(super) write_complete_tx: Option<oneshot::Sender<()>>,
}
#[derive(Clone)]
pub(crate) struct RemoteControlHandle {
enabled_tx: Arc<watch::Sender<bool>>,
}
impl RemoteControlHandle {
pub(crate) fn set_enabled(&self, enabled: bool) {
self.enabled_tx.send_if_modified(|state| {
let changed = *state != enabled;
*state = enabled;
changed
});
}
}
pub(crate) async fn start_remote_control(
remote_control_url: String,
state_db: Option<Arc<StateRuntime>>,
@@ -36,21 +52,38 @@ pub(crate) async fn start_remote_control(
transport_event_tx: mpsc::Sender<TransportEvent>,
shutdown_token: CancellationToken,
app_server_client_name_rx: Option<oneshot::Receiver<String>>,
) -> io::Result<JoinHandle<()>> {
let remote_control_target = normalize_remote_control_url(&remote_control_url)?;
validate_remote_control_auth(&auth_manager).await?;
initial_enabled: bool,
) -> io::Result<(JoinHandle<()>, RemoteControlHandle)> {
let remote_control_target = if initial_enabled {
Some(normalize_remote_control_url(&remote_control_url)?)
} else {
None
};
if initial_enabled {
validate_remote_control_auth(&auth_manager).await?;
}
Ok(tokio::spawn(async move {
let (enabled_tx, enabled_rx) = watch::channel(initial_enabled);
let join_handle = tokio::spawn(async move {
RemoteControlWebsocket::new(
remote_control_url,
remote_control_target,
state_db,
auth_manager,
transport_event_tx,
shutdown_token,
enabled_rx,
)
.run(app_server_client_name_rx)
.await;
}))
});
Ok((
join_handle,
RemoteControlHandle {
enabled_tx: Arc::new(enabled_tx),
},
))
}
pub(crate) async fn validate_remote_control_auth(
codex-rs/app-server/src/transport/remote_control/tests.rs
+101 -16
View File
@@ -125,13 +125,14 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages()
let (transport_event_tx, mut transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(remote_control_state_runtime(&codex_home).await),
remote_control_auth_manager(),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -376,7 +377,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages()
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
@@ -389,13 +390,14 @@ async fn remote_control_transport_reconnects_after_disconnect() {
let (transport_event_tx, mut transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(remote_control_state_runtime(&codex_home).await),
remote_control_auth_manager(),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -452,7 +454,84 @@ async fn remote_control_transport_reconnects_after_disconnect() {
}
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
async fn remote_control_start_allows_remote_control_invalid_url_when_disabled() {
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let (remote_task, _remote_handle) = start_remote_control(
"https://internal.example.com/backend-api/".to_string(),
/*state_db*/ None,
remote_control_auth_manager(),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ false,
)
.await
.expect("disabled remote control should not validate the URL at startup");
shutdown_token.cancel();
timeout(Duration::from_secs(1), remote_task)
.await
.expect("remote control task should stop")
.expect("remote control task should join");
}
#[tokio::test]
async fn remote_control_handle_set_enabled_stops_and_restarts_connections() {
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("listener should bind");
let remote_control_url = remote_control_url_for_listener(&listener);
let codex_home = TempDir::new().expect("temp dir should create");
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let (remote_task, remote_handle) = start_remote_control(
remote_control_url,
Some(remote_control_state_runtime(&codex_home).await),
remote_control_auth_manager(),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
let enroll_request = accept_http_request(&listener).await;
assert_eq!(
enroll_request.request_line,
"POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
);
respond_with_json(
enroll_request.stream,
json!({ "server_id": "srv_e_test", "environment_id": "env_test" }),
)
.await;
let mut first_websocket = accept_remote_control_connection(&listener).await;
remote_handle.set_enabled(/*enabled*/ false);
timeout(Duration::from_secs(1), first_websocket.next())
.await
.expect("disabling remote control should close the websocket");
timeout(Duration::from_millis(100), listener.accept())
.await
.expect_err("disabled remote control should not reconnect");
remote_handle.set_enabled(/*enabled*/ true);
let mut second_websocket = accept_remote_control_connection(&listener).await;
second_websocket
.close(None)
.await
.expect("second websocket should close");
shutdown_token.cancel();
let _ = remote_task.await;
}
#[tokio::test]
@@ -465,13 +544,14 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
let (transport_event_tx, mut transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(remote_control_state_runtime(&codex_home).await),
remote_control_auth_manager(),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -617,7 +697,7 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
@@ -631,13 +711,14 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let expected_server_name = gethostname().to_string_lossy().trim().to_string();
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(remote_control_state_runtime(&codex_home).await),
remote_control_auth_manager(),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -815,7 +896,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
@@ -847,13 +928,14 @@ async fn remote_control_http_mode_reuses_persisted_enrollment_before_reenrolling
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(state_db.clone()),
remote_control_auth_manager_with_home(&codex_home),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -879,7 +961,7 @@ async fn remote_control_http_mode_reuses_persisted_enrollment_before_reenrolling
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
@@ -913,13 +995,14 @@ async fn remote_control_stdio_mode_waits_for_client_name_before_connecting() {
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let (app_server_client_name_tx, app_server_client_name_rx) = oneshot::channel::<String>();
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(state_db.clone()),
remote_control_auth_manager_with_home(&codex_home),
transport_event_tx,
shutdown_token.clone(),
Some(app_server_client_name_rx),
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -936,7 +1019,7 @@ async fn remote_control_stdio_mode_waits_for_client_name_before_connecting() {
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
@@ -969,13 +1052,14 @@ async fn remote_control_waits_for_account_id_before_enrolling() {
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(state_db.clone()),
auth_manager,
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start before account id is available");
@@ -1012,7 +1096,7 @@ async fn remote_control_waits_for_account_id_before_enrolling() {
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[tokio::test]
@@ -1051,13 +1135,14 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404()
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
let remote_handle = start_remote_control(
let (remote_task, _remote_handle) = start_remote_control(
remote_control_url,
Some(state_db.clone()),
remote_control_auth_manager_with_home(&codex_home),
transport_event_tx,
shutdown_token.clone(),
/*app_server_client_name_rx*/ None,
/*initial_enabled*/ true,
)
.await
.expect("remote control should start");
@@ -1104,7 +1189,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404()
);
shutdown_token.cancel();
let _ = remote_handle.await;
let _ = remote_task.await;
}
#[derive(Debug)]
codex-rs/app-server/src/transport/remote_control/websocket.rs
+99 -34
View File
@@ -111,7 +111,8 @@ struct WebsocketState {
}
pub(crate) struct RemoteControlWebsocket {
remote_control_target: RemoteControlTarget,
remote_control_url: String,
remote_control_target: Option<RemoteControlTarget>,
state_db: Option<Arc<StateRuntime>>,
auth_manager: Arc<AuthManager>,
shutdown_token: CancellationToken,
@@ -122,15 +123,24 @@ pub(crate) struct RemoteControlWebsocket {
state: Arc<Mutex<WebsocketState>>,
server_event_rx: Arc<Mutex<mpsc::Receiver<super::QueuedServerEnvelope>>>,
used_rx: watch::Receiver<usize>,
enabled_rx: watch::Receiver<bool>,
}
enum ConnectOutcome {
Connected(Box<WebSocketStream<MaybeTlsStream<TcpStream>>>),
Disabled,
Shutdown,
}
impl RemoteControlWebsocket {
pub(crate) fn new(
remote_control_target: RemoteControlTarget,
remote_control_url: String,
remote_control_target: Option<RemoteControlTarget>,
state_db: Option<Arc<StateRuntime>>,
auth_manager: Arc<AuthManager>,
transport_event_tx: mpsc::Sender<TransportEvent>,
shutdown_token: CancellationToken,
enabled_rx: watch::Receiver<bool>,
) -> Self {
let shutdown_token = shutdown_token.child_token();
let (server_event_tx, server_event_rx) = mpsc::channel(super::CHANNEL_CAPACITY);
@@ -140,6 +150,7 @@ impl RemoteControlWebsocket {
let auth_recovery = auth_manager.unauthorized_recovery();
Self {
remote_control_url,
remote_control_target,
state_db,
auth_manager,
@@ -155,6 +166,7 @@ impl RemoteControlWebsocket {
})),
server_event_rx: Arc::new(Mutex::new(server_event_rx)),
used_rx,
enabled_rx,
}
}
@@ -174,13 +186,18 @@ impl RemoteControlWebsocket {
};
loop {
if !self.wait_until_enabled().await {
break;
}
let shutdown_token = self.shutdown_token.child_token();
let websocket_connection = match self
.connect(&shutdown_token, app_server_client_name.as_deref())
.await
{
Some(websocket_connection) => websocket_connection,
None => break,
ConnectOutcome::Connected(websocket_connection) => *websocket_connection,
ConnectOutcome::Disabled => continue,
ConnectOutcome::Shutdown => break,
};
self.run_connection(websocket_connection, shutdown_token)
@@ -208,50 +225,93 @@ impl RemoteControlWebsocket {
}
}
async fn wait_until_enabled(&mut self) -> bool {
tokio::select! {
_ = self.shutdown_token.cancelled() => false,
enabled = self.enabled_rx.wait_for(|enabled| *enabled) => enabled.is_ok(),
}
}
async fn connect(
&mut self,
shutdown_token: &CancellationToken,
app_server_client_name: Option<&str>,
) -> Option<WebSocketStream<MaybeTlsStream<TcpStream>>> {
) -> ConnectOutcome {
let remote_control_target = match self.remote_control_target.as_ref() {
Some(remote_control_target) => remote_control_target.clone(),
None => match super::protocol::normalize_remote_control_url(&self.remote_control_url) {
Ok(remote_control_target) => {
self.remote_control_target = Some(remote_control_target.clone());
remote_control_target
}
Err(err) => {
warn!("remote control is enabled but the URL is invalid: {err}");
tokio::select! {
_ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown,
changed = self.enabled_rx.wait_for(|enabled| !*enabled) => {
if changed.is_err() {
return ConnectOutcome::Shutdown;
}
return ConnectOutcome::Disabled;
}
}
}
},
};
loop {
let subscribe_cursor = self.state.lock().await.subscribe_cursor.clone();
tokio::select! {
_ = shutdown_token.cancelled() => return None,
let connect_result = tokio::select! {
_ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown,
changed = self.enabled_rx.wait_for(|enabled| !*enabled) => {
if changed.is_err() {
return ConnectOutcome::Shutdown;
}
return ConnectOutcome::Disabled;
}
connect_result = connect_remote_control_websocket(
&self.remote_control_target,
&remote_control_target,
self.state_db.as_deref(),
&self.auth_manager,
&mut self.auth_recovery,
&mut self.enrollment,
subscribe_cursor.as_deref(),
app_server_client_name,
) => {
match connect_result {
Ok((websocket_connection, response)) => {
self.reconnect_attempt = 0;
self.auth_recovery = self.auth_manager.unauthorized_recovery();
info!(
"connected to app-server remote control websocket: {}, {}",
self.remote_control_target.websocket_url,
format_headers(response.headers())
);
return Some(websocket_connection);
}
Err(err) => {
let reconnect_delay = if err.kind() == ErrorKind::WouldBlock {
info!("{err}");
REMOTE_CONTROL_ACCOUNT_ID_RETRY_INTERVAL
} else {
warn!("{err}");
let reconnect_delay = backoff(self.reconnect_attempt);
self.reconnect_attempt += 1;
reconnect_delay
};
tokio::select! {
_ = shutdown_token.cancelled() => return None,
_ = tokio::time::sleep(reconnect_delay) => {}
) => connect_result,
};
match connect_result {
Ok((websocket_connection, response)) => {
self.reconnect_attempt = 0;
self.auth_recovery = self.auth_manager.unauthorized_recovery();
info!(
"connected to app-server remote control websocket: {}, {}",
remote_control_target.websocket_url,
format_headers(response.headers())
);
return ConnectOutcome::Connected(Box::new(websocket_connection));
}
Err(err) => {
let reconnect_delay = if err.kind() == ErrorKind::WouldBlock {
REMOTE_CONTROL_ACCOUNT_ID_RETRY_INTERVAL
} else {
warn!(
"failed to connect to app-server remote control websocket: {}, err: {}",
remote_control_target.websocket_url, err
);
let reconnect_delay = backoff(self.reconnect_attempt);
self.reconnect_attempt += 1;
reconnect_delay
};
tokio::select! {
_ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown,
changed = self.enabled_rx.wait_for(|enabled| !*enabled) => {
if changed.is_err() {
return ConnectOutcome::Shutdown;
}
return ConnectOutcome::Disabled;
}
_ = tokio::time::sleep(reconnect_delay) => {}
}
}
}
@@ -282,8 +342,10 @@ impl RemoteControlWebsocket {
shutdown_token.clone(),
));
let mut enabled_rx = self.enabled_rx.clone();
tokio::select! {
_ = shutdown_token.cancelled() => {}
_ = enabled_rx.wait_for(|enabled| !*enabled) => shutdown_token.cancel(),
_ = join_set.join_next() => shutdown_token.cancel(),
}
@@ -1124,15 +1186,18 @@ mod tests {
let (transport_event_tx, transport_event_rx) = mpsc::channel(1);
drop(transport_event_rx);
let shutdown_token = CancellationToken::new();
let (_enabled_tx, enabled_rx) = watch::channel(true);
let websocket_task = tokio::spawn({
let shutdown_token = shutdown_token.clone();
async move {
RemoteControlWebsocket::new(
remote_control_target,
remote_control_url,
Some(remote_control_target),
/*state_db*/ None,
remote_control_auth_manager(),
transport_event_tx,
shutdown_token,
enabled_rx,
)
.run(/*app_server_client_name_rx*/ None)
.await
codex-rs/app-server/tests/suite/v2/fs.rs
+14 -11
View File
@@ -21,8 +21,6 @@ use std::path::PathBuf;
use tempfile::TempDir;
use tokio::time::Duration;
use tokio::time::timeout;
use uuid::Uuid;
use uuid::Version;
#[cfg(unix)]
use std::os::unix::fs::symlink;
@@ -628,8 +626,10 @@ async fn fs_watch_directory_reports_changed_child_paths_and_unwatch_stops_notifi
std::fs::write(&fetch_head, "old\n")?;
let mut mcp = initialized_mcp(&codex_home).await?;
let watch_id = "watch-git-dir".to_string();
let watch_request_id = mcp
.send_fs_watch_request(codex_app_server_protocol::FsWatchParams {
watch_id: watch_id.clone(),
path: absolute_path(git_dir.clone()),
})
.await?;
@@ -641,8 +641,6 @@ async fn fs_watch_directory_reports_changed_child_paths_and_unwatch_stops_notifi
.await??,
)?;
assert_eq!(watch_response.path, absolute_path(git_dir.clone()));
let watch_id = Uuid::parse_str(&watch_response.watch_id)?;
assert_eq!(watch_id.get_version(), Some(Version::SortRand));
std::fs::write(&fetch_head, "updated\n")?;
@@ -650,7 +648,7 @@ async fn fs_watch_directory_reports_changed_child_paths_and_unwatch_stops_notifi
// Keep validating notification shape when the backend does emit, but do not
// fail the whole suite if no OS event arrives.
if let Some(changed) = maybe_fs_changed_notification(&mut mcp).await? {
assert_eq!(changed.watch_id, watch_response.watch_id.clone());
assert_eq!(changed.watch_id, watch_id.clone());
assert_eq!(
changed.changed_paths,
vec![absolute_path(fetch_head.clone())]
@@ -665,9 +663,7 @@ async fn fs_watch_directory_reports_changed_child_paths_and_unwatch_stops_notifi
{}
let unwatch_request_id = mcp
.send_fs_unwatch_request(FsUnwatchParams {
watch_id: watch_response.watch_id,
})
.send_fs_unwatch_request(FsUnwatchParams { watch_id })
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
@@ -698,8 +694,10 @@ async fn fs_watch_file_reports_atomic_replace_events() -> Result<()> {
std::fs::write(&head_path, "ref: refs/heads/main\n")?;
let mut mcp = initialized_mcp(&codex_home).await?;
let watch_id = "watch-head".to_string();
let watch_request_id = mcp
.send_fs_watch_request(codex_app_server_protocol::FsWatchParams {
watch_id: watch_id.clone(),
path: absolute_path(head_path.clone()),
})
.await?;
@@ -718,7 +716,7 @@ async fn fs_watch_file_reports_atomic_replace_events() -> Result<()> {
assert_eq!(
changed,
FsChangedNotification {
watch_id: watch_response.watch_id,
watch_id,
changed_paths: vec![absolute_path(head_path.clone())],
}
);
@@ -735,8 +733,10 @@ async fn fs_watch_allows_missing_file_targets() -> Result<()> {
std::fs::create_dir_all(&git_dir)?;
let mut mcp = initialized_mcp(&codex_home).await?;
let watch_id = "watch-fetch-head".to_string();
let watch_request_id = mcp
.send_fs_watch_request(codex_app_server_protocol::FsWatchParams {
watch_id: watch_id.clone(),
path: absolute_path(fetch_head.clone()),
})
.await?;
@@ -755,7 +755,7 @@ async fn fs_watch_allows_missing_file_targets() -> Result<()> {
assert_eq!(
changed,
FsChangedNotification {
watch_id: watch_response.watch_id,
watch_id,
changed_paths: vec![absolute_path(fetch_head.clone())],
}
);
@@ -770,7 +770,10 @@ async fn fs_watch_rejects_relative_paths() -> Result<()> {
let mut mcp = initialized_mcp(&codex_home).await?;
let watch_id = mcp
.send_raw_request("fs/watch", Some(json!({ "path": "relative-path" })))
.send_raw_request(
"fs/watch",
Some(json!({ "watchId": "watch-relative", "path": "relative-path" })),
)
.await?;
expect_error_message(
&mut mcp,