agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

app-server: notify clients of remote-control status changes (#19919)

Browse Source 
## Why

Remote-control app-server enrollments have both an internal server id
and the environment id exposed to remote-control clients. App-server
clients need one current status snapshot that says whether remote
control is usable and which environment id, if any, is exposed.

A temporary websocket disconnect is not itself an identity change.
Account changes, stale enrollment invalidation, successful
re-enrollment, and missing ChatGPT auth are meaningful status changes.
Disabled remote control remains `disabled` regardless of auth or SQLite
state. SQLite startup failure disablement and enrollment persistence
failures are handled in #20068; this PR reports the resulting effective
status to clients.

## What changed

- Adds v2 `remoteControl/status/changed` carrying `state` and
`environmentId`.
- Adds `RemoteControlConnectionState` values: `disabled`, `connecting`,
`connected`, and `errored`.
- Exposes remote-control status updates through `RemoteControlHandle`
using a Tokio watch channel.
- Always sends the current remote-control status snapshot to newly
initialized app-server clients.
- Broadcasts status changes to initialized app-server clients when state
or environment id changes.
- Treats missing ChatGPT auth as an `errored` status while leaving it
retryable because auth can change at runtime.
- Clears `environmentId` when enrollment is cleared for account changes,
auth loss, stale backend invalidation, or disabled remote control.
- Updates app-server protocol schema fixtures, generated TypeScript,
app-server README, remote-control tests, and TUI exhaustive notification
matches.

## Stack

- Builds on #20068.

## Verification

- `just write-app-server-schema`
- `cargo test -p codex-app-server-protocol`
- `cargo test -p codex-app-server transport::remote_control --lib`
- `cargo check -p codex-tui`
- `just fix -p codex-app-server-protocol`
- `just fix -p codex-app-server`
- `just fix -p codex-tui`
This commit is contained in:
Ruslan Nigmatullin
2026-04-28 16:52:14 -07:00
committed by GitHub
Unverified
parent 5e6cbbadf7
commit c6465c1ec2
17 changed files with 743 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+47
View File
@@ -2603,6 +2603,33 @@
],
"type": "object"
},
"RemoteControlConnectionStatus": {
"enum": [
"disabled",
"connecting",
"connected",
"errored"
],
"type": "string"
},
"RemoteControlStatusChangedNotification": {
"description": "Current remote-control connection status and environment id exposed to clients.",
"properties": {
"environmentId": {
"type": [
"string",
"null"
]
},
"status": {
"$ref": "#/definitions/RemoteControlConnectionStatus"
}
},
"required": [
"status"
],
"type": "object"
},
"RequestId": {
"anyOf": [
{
@@ -5342,6 +5369,26 @@
"title": "App/list/updatedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"remoteControl/status/changed"
],
"title": "RemoteControl/status/changedNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/RemoteControlStatusChangedNotification"
}
},
"required": [
"method",
"params"
],
"title": "RemoteControl/status/changedNotification",
"type": "object"
},
{
"properties": {
"method": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+49
View File
@@ -4348,6 +4348,26 @@
"title": "App/list/updatedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"remoteControl/status/changed"
],
"title": "RemoteControl/status/changedNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/RemoteControlStatusChangedNotification"
}
},
"required": [
"method",
"params"
],
"title": "RemoteControl/status/changedNotification",
"type": "object"
},
{
"properties": {
"method": {
@@ -12499,6 +12519,35 @@
],
"type": "string"
},
"RemoteControlConnectionStatus": {
"enum": [
"disabled",
"connecting",
"connected",
"errored"
],
"type": "string"
},
"RemoteControlStatusChangedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "Current remote-control connection status and environment id exposed to clients.",
"properties": {
"environmentId": {
"type": [
"string",
"null"
]
},
"status": {
"$ref": "#/definitions/v2/RemoteControlConnectionStatus"
}
},
"required": [
"status"
],
"title": "RemoteControlStatusChangedNotification",
"type": "object"
},
"RequestId": {
"anyOf": [
{
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+49
View File
@@ -9173,6 +9173,35 @@
],
"type": "string"
},
"RemoteControlConnectionStatus": {
"enum": [
"disabled",
"connecting",
"connected",
"errored"
],
"type": "string"
},
"RemoteControlStatusChangedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "Current remote-control connection status and environment id exposed to clients.",
"properties": {
"environmentId": {
"type": [
"string",
"null"
]
},
"status": {
"$ref": "#/definitions/RemoteControlConnectionStatus"
}
},
"required": [
"status"
],
"title": "RemoteControlStatusChangedNotification",
"type": "object"
},
"RequestId": {
"anyOf": [
{
@@ -10909,6 +10938,26 @@
"title": "App/list/updatedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"remoteControl/status/changed"
],
"title": "RemoteControl/status/changedNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/RemoteControlStatusChangedNotification"
}
},
"required": [
"method",
"params"
],
"title": "RemoteControl/status/changedNotification",
"type": "object"
},
{
"properties": {
"method": {
codex-rs/app-server-protocol/schema/json/v2/RemoteControlStatusChangedNotification.json
Generated
+31
View File
@@ -0,0 +1,31 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"RemoteControlConnectionStatus": {
"enum": [
"disabled",
"connecting",
"connected",
"errored"
],
"type": "string"
}
},
"description": "Current remote-control connection status and environment id exposed to clients.",
"properties": {
"environmentId": {
"type": [
"string",
"null"
]
},
"status": {
"$ref": "#/definitions/RemoteControlConnectionStatus"
}
},
"required": [
"status"
],
"title": "RemoteControlStatusChangedNotification",
"type": "object"
}
codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts
Generated
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlConnectionStatus.ts
Generated
+5
View File
@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type RemoteControlConnectionStatus = "disabled" | "connecting" | "connected" | "errored";
codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlStatusChangedNotification.ts
Generated
+9
View File
@@ -0,0 +1,9 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { RemoteControlConnectionStatus } from "./RemoteControlConnectionStatus";
/**
* Current remote-control connection status and environment id exposed to clients.
*/
export type RemoteControlStatusChangedNotification = { status: RemoteControlConnectionStatus, environmentId: string | null, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
+2
View File
@@ -282,6 +282,8 @@ export type { ReasoningSummaryTextDeltaNotification } from "./ReasoningSummaryTe
export type { ReasoningTextDeltaNotification } from "./ReasoningTextDeltaNotification";
export type { RemoteControlClientConnectionAudience } from "./RemoteControlClientConnectionAudience";
export type { RemoteControlClientEnrollmentAudience } from "./RemoteControlClientEnrollmentAudience";
export type { RemoteControlConnectionStatus } from "./RemoteControlConnectionStatus";
export type { RemoteControlStatusChangedNotification } from "./RemoteControlStatusChangedNotification";
export type { RequestPermissionProfile } from "./RequestPermissionProfile";
export type { ResidencyRequirement } from "./ResidencyRequirement";
export type { ReviewDelivery } from "./ReviewDelivery";
codex-rs/app-server-protocol/src/protocol/common.rs
+1
View File
@@ -1258,6 +1258,7 @@ server_notification_definitions! {
AccountUpdated => "account/updated" (v2::AccountUpdatedNotification),
AccountRateLimitsUpdated => "account/rateLimits/updated" (v2::AccountRateLimitsUpdatedNotification),
AppListUpdated => "app/list/updated" (v2::AppListUpdatedNotification),
RemoteControlStatusChanged => "remoteControl/status/changed" (v2::RemoteControlStatusChangedNotification),
ExternalAgentConfigImportCompleted => "externalAgentConfig/import/completed" (v2::ExternalAgentConfigImportCompletedNotification),
FsChanged => "fs/changed" (v2::FsChangedNotification),
ReasoningSummaryTextDelta => "item/reasoning/summaryTextDelta" (v2::ReasoningSummaryTextDeltaNotification),
codex-rs/app-server-protocol/src/protocol/v2.rs
+19
View File
@@ -2878,6 +2878,25 @@ pub struct DeviceKeyPublicResponse {
pub protection_class: DeviceKeyProtectionClass,
}
/// Current remote-control connection status and environment id exposed to clients.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct RemoteControlStatusChangedNotification {
pub status: RemoteControlConnectionStatus,
pub environment_id: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase", export_to = "v2/")]
pub enum RemoteControlConnectionStatus {
Disabled,
Connecting,
Connected,
Errored,
}
/// Audience for a remote-control client connection device-key proof.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
codex-rs/app-server/README.md
+1
View File
@@ -205,6 +205,7 @@ Example with notification opt-out:
- `device/key/create` — create or load a controller-local device signing key for an account/client binding. This local-key API is available only over local transports such as stdio and in-process; remote transports reject it. Hardware-backed providers are the target protection class; an OS-protected non-extractable fallback is allowed only with `protectionPolicy: "allow_os_protected_nonextractable"` and returns the reported `protectionClass`.
- `device/key/public` — return a device key's SPKI DER public key as base64 plus its `algorithm` and `protectionClass`.
- `device/key/sign` — sign one of the accepted structured payload variants with a controller-local device key. The only accepted payload today is `remoteControlClientConnection`, which binds a server-issued `/client` websocket challenge to the enrolled controller device without signing the bearer token itself; this is intentionally not an arbitrary-byte signing API.
- `remoteControl/status/changed` — notification emitted when the remote-control status or client-visible environment id changes. `status` is one of `disabled`, `connecting`, `connected`, or `errored`; `environmentId` is a string when the app-server has a current enrollment and `null` when that enrollment is cleared, invalidated, or remote control is disabled. Newly initialized app-server clients always receive the current status snapshot.
- `skills/config/write` — write user-level skill config by name or absolute path.
- `plugin/install` — install a plugin from a discovered marketplace entry, rejecting marketplace entries marked unavailable for install, install MCPs if any, and return the effective plugin auth policy plus any apps that still need auth (**under development; do not call from production clients yet**).
- `plugin/uninstall` — uninstall a local plugin by `pluginId` in `<plugin>@<marketplace>` form by removing its cached files and clearing its user-level config entry, or uninstall a remote ChatGPT plugin by backend `pluginId` by forwarding the uninstall to the ChatGPT plugin backend and removing any downloaded remote-plugin cache (**under development; do not call from production clients yet**).
codex-rs/app-server/src/lib.rs
+32 -1
View File
@@ -40,6 +40,8 @@ use codex_analytics::AppServerRpcTransport;
use codex_app_server_protocol::ConfigLayerSource;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::RemoteControlStatusChangedNotification;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::TextPosition as AppTextPosition;
use codex_app_server_protocol::TextRange as AppTextRange;
use codex_config::ConfigLoadError;
@@ -724,6 +726,7 @@ pub async fn run_main_with_transport_options(
let processor_handle = tokio::spawn({
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let initialize_notification_sender = outgoing_message_sender.clone();
let outbound_control_tx = outbound_control_tx;
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
@@ -739,12 +742,14 @@ pub async fn run_main_with_transport_options(
session_source,
auth_manager,
rpc_transport: analytics_rpc_transport(&transport),
remote_control_handle: Some(remote_control_handle),
remote_control_handle: Some(remote_control_handle.clone()),
plugin_startup_tasks: runtime_options.plugin_startup_tasks,
}));
let mut thread_created_rx = processor.thread_created_receiver();
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
let mut connections = HashMap::<ConnectionId, ConnectionState>::new();
let mut remote_control_status_rx = remote_control_handle.status_receiver();
let mut remote_control_status = remote_control_status_rx.borrow().clone();
let transport_shutdown_token = transport_shutdown_token.clone();
async move {
let mut listen_for_threads = true;
@@ -884,6 +889,14 @@ pub async fn run_main_with_transport_options(
connection_id,
)
.await;
initialize_notification_sender
.send_server_notification_to_connections(
&[connection_id],
ServerNotification::RemoteControlStatusChanged(
remote_control_status.clone(),
),
)
.await;
processor.connection_initialized(connection_id).await;
connection_state
.outbound_initialized
@@ -915,6 +928,24 @@ pub async fn run_main_with_transport_options(
}
}
}
changed = remote_control_status_rx.changed() => {
if changed.is_err() {
continue;
}
let status = remote_control_status_rx.borrow().clone();
if remote_control_status == status {
continue;
}
remote_control_status = status.clone();
initialize_notification_sender
.send_server_notification(ServerNotification::RemoteControlStatusChanged(
RemoteControlStatusChangedNotification {
status: status.status,
environment_id: status.environment_id,
},
))
.await;
}
created = thread_created_rx.recv(), if listen_for_threads => {
match created {
Ok(thread_id) => {
codex-rs/app-server/src/transport/remote_control/mod.rs
+26 -1
View File
@@ -3,6 +3,8 @@ mod enroll;
mod protocol;
mod websocket;
use crate::transport::remote_control::websocket::RemoteControlChannels;
use crate::transport::remote_control::websocket::RemoteControlStatusPublisher;
use crate::transport::remote_control::websocket::RemoteControlWebsocket;
pub use self::protocol::ClientId;
@@ -12,6 +14,8 @@ use self::protocol::normalize_remote_control_url;
use super::CHANNEL_CAPACITY;
use super::TransportEvent;
use super::next_connection_id;
use codex_app_server_protocol::RemoteControlConnectionStatus;
use codex_app_server_protocol::RemoteControlStatusChangedNotification;
use codex_login::AuthManager;
use codex_state::StateRuntime;
use std::io;
@@ -33,6 +37,7 @@ pub(super) struct QueuedServerEnvelope {
#[derive(Clone)]
pub(crate) struct RemoteControlHandle {
enabled_tx: Arc<watch::Sender<bool>>,
status_tx: Arc<watch::Sender<RemoteControlStatusChangedNotification>>,
state_db_available: bool,
}
@@ -49,6 +54,12 @@ impl RemoteControlHandle {
changed
});
}
pub(crate) fn status_receiver(
&self,
) -> watch::Receiver<RemoteControlStatusChangedNotification> {
self.status_tx.subscribe()
}
}
pub(crate) async fn start_remote_control(
@@ -73,13 +84,26 @@ pub(crate) async fn start_remote_control(
};
let (enabled_tx, enabled_rx) = watch::channel(initial_enabled);
let initial_status = RemoteControlStatusChangedNotification {
status: if initial_enabled {
RemoteControlConnectionStatus::Connecting
} else {
RemoteControlConnectionStatus::Disabled
},
environment_id: None,
};
let (status_tx, _status_rx) = watch::channel(initial_status);
let status_publisher = RemoteControlStatusPublisher::new(status_tx.clone());
let join_handle = tokio::spawn(async move {
RemoteControlWebsocket::new(
remote_control_url,
remote_control_target,
state_db,
auth_manager,
transport_event_tx,
RemoteControlChannels {
transport_event_tx,
status_publisher,
},
shutdown_token,
enabled_rx,
)
@@ -91,6 +115,7 @@ pub(crate) async fn start_remote_control(
join_handle,
RemoteControlHandle {
enabled_tx: Arc::new(enabled_tx),
status_tx: Arc::new(status_tx),
state_db_available,
},
))
codex-rs/app-server/src/transport/remote_control/tests.rs
+142 -6
View File
@@ -18,6 +18,8 @@ use base64::Engine;
use codex_app_server_protocol::AuthMode;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::RemoteControlConnectionStatus;
use codex_app_server_protocol::RemoteControlStatusChangedNotification;
use codex_app_server_protocol::ServerNotification;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::test_support::auth_manager_from_auth;
@@ -45,6 +47,7 @@ use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::sync::watch;
use tokio::time::Duration;
use tokio::time::timeout;
use tokio_tungstenite::WebSocketStream;
@@ -115,6 +118,50 @@ fn remote_control_url_for_listener(listener: &TcpListener) -> String {
format!("http://{addr}/backend-api/")
}
async fn expect_remote_control_status(
status_rx: &mut watch::Receiver<RemoteControlStatusChangedNotification>,
expected_status: Option<RemoteControlConnectionStatus>,
expected_environment_id: Option<&str>,
) {
timeout(Duration::from_secs(5), status_rx.changed())
.await
.expect("remote control status event should arrive in time")
.expect("remote control status watch should remain open");
let status = status_rx.borrow();
if let Some(expected_status) = expected_status {
assert_eq!(status.status, expected_status);
}
assert_eq!(status.environment_id.as_deref(), expected_environment_id);
}
async fn expect_remote_control_status_snapshot(
status_rx: &mut watch::Receiver<RemoteControlStatusChangedNotification>,
expected_status: RemoteControlStatusChangedNotification,
) {
if *status_rx.borrow() == expected_status {
return;
}
let expected_status_for_wait = expected_status.clone();
let result = timeout(Duration::from_secs(5), async {
loop {
status_rx
.changed()
.await
.expect("remote control status watch should remain open");
if *status_rx.borrow() == expected_status_for_wait {
return;
}
}
})
.await;
assert!(
result.is_ok(),
"remote control status snapshot should arrive in time; expected {expected_status:?}, latest {:?}",
status_rx.borrow().clone()
);
}
#[tokio::test]
async fn remote_control_transport_manages_virtual_clients_and_routes_messages() {
let listener = TcpListener::bind("127.0.0.1:0")
@@ -125,7 +172,7 @@ 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_task, _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(),
@@ -136,6 +183,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages()
)
.await
.expect("remote control should start");
let mut status_rx = remote_handle.status_receiver();
let enroll_request = accept_http_request(&listener).await;
assert_eq!(
enroll_request.request_line,
@@ -147,6 +195,12 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages()
)
.await;
let mut websocket = accept_remote_control_connection(&listener).await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_test"),
)
.await;
let client_id = ClientId("client-1".to_string());
send_client_event(
@@ -394,7 +448,7 @@ 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_task, _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(),
@@ -405,6 +459,7 @@ async fn remote_control_transport_reconnects_after_disconnect() {
)
.await
.expect("remote control should start");
let mut status_rx = remote_handle.status_receiver();
let enroll_request = accept_http_request(&listener).await;
assert_eq!(
@@ -424,6 +479,12 @@ async fn remote_control_transport_reconnects_after_disconnect() {
drop(first_websocket);
let mut second_websocket = accept_remote_control_connection(&listener).await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_test"),
)
.await;
send_client_event(
&mut second_websocket,
ClientEnvelope {
@@ -526,7 +587,7 @@ async fn remote_control_start_allows_missing_auth_when_enabled() {
}
#[tokio::test]
async fn remote_control_start_disables_remote_control_without_state_db() {
async fn remote_control_start_reports_missing_state_db_as_disabled_when_enabled() {
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("listener should bind");
@@ -545,6 +606,14 @@ async fn remote_control_start_disables_remote_control_without_state_db() {
)
.await
.expect("remote control should start disabled without sqlite state db");
let mut status_rx = remote_handle.status_receiver();
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Disabled,
environment_id: None,
}
);
timeout(Duration::from_millis(100), listener.accept())
.await
@@ -554,6 +623,9 @@ async fn remote_control_start_disables_remote_control_without_state_db() {
timeout(Duration::from_millis(100), listener.accept())
.await
.expect_err("remote control should remain disabled without sqlite state db");
timeout(Duration::from_millis(20), status_rx.changed())
.await
.expect_err("status should remain disabled without sqlite state db");
shutdown_token.cancel();
timeout(Duration::from_secs(1), remote_task)
@@ -583,6 +655,7 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() {
)
.await
.expect("remote control should start");
let mut status_rx = remote_handle.status_receiver();
let enroll_request = accept_http_request(&listener).await;
assert_eq!(
@@ -595,8 +668,24 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() {
)
.await;
let mut first_websocket = accept_remote_control_connection(&listener).await;
expect_remote_control_status_snapshot(
&mut status_rx,
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connected,
environment_id: Some("env_test".to_string()),
},
)
.await;
remote_handle.set_enabled(/*enabled*/ false);
expect_remote_control_status_snapshot(
&mut status_rx,
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Disabled,
environment_id: None,
},
)
.await;
timeout(Duration::from_secs(1), first_websocket.next())
.await
.expect("disabling remote control should close the websocket");
@@ -605,7 +694,21 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() {
.expect_err("disabled remote control should not reconnect");
remote_handle.set_enabled(/*enabled*/ true);
expect_remote_control_status_snapshot(
&mut status_rx,
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connecting,
environment_id: Some("env_test".to_string()),
},
)
.await;
let mut second_websocket = accept_remote_control_connection(&listener).await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_test"),
)
.await;
second_websocket
.close(None)
.await
@@ -625,7 +728,7 @@ 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_task, _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(),
@@ -636,6 +739,7 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
)
.await
.expect("remote control should start");
let mut status_rx = remote_handle.status_receiver();
let enroll_request = accept_http_request(&listener).await;
respond_with_json(
@@ -644,6 +748,12 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
)
.await;
let mut first_websocket = accept_remote_control_connection(&listener).await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_test"),
)
.await;
let client_id = ClientId("client-1".to_string());
let initialize_message = JSONRPCMessage::Request(codex_app_server_protocol::JSONRPCRequest {
@@ -793,7 +903,7 @@ 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_task, _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(),
@@ -804,6 +914,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
)
.await
.expect("remote control should start");
let mut status_rx = remote_handle.status_receiver();
let enroll_request = accept_http_request(&listener).await;
assert_eq!(
@@ -836,6 +947,12 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
let (handshake_request, mut websocket) =
accept_remote_control_backend_connection(&listener).await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_test"),
)
.await;
assert_eq!(
handshake_request.path,
"/backend-api/wham/remote/control/server"
@@ -1220,7 +1337,7 @@ 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_task, _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),
@@ -1231,6 +1348,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404()
)
.await
.expect("remote control should start");
let mut status_rx = remote_handle.status_receiver();
let websocket_request = accept_http_request(&listener).await;
assert_eq!(
@@ -1241,7 +1359,19 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404()
websocket_request.headers.get("x-codex-server-id"),
Some(&stale_enrollment.server_id)
);
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_stale"),
)
.await;
respond_with_status(websocket_request.stream, "404 Not Found", "").await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
/*expected_environment_id*/ None,
)
.await;
let enroll_request = accept_http_request(&listener).await;
assert_eq!(
@@ -1258,6 +1388,12 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404()
.await;
let (handshake_request, _websocket) = accept_remote_control_backend_connection(&listener).await;
expect_remote_control_status(
&mut status_rx,
/*expected_status*/ None,
Some("env_refreshed"),
)
.await;
assert_eq!(
handshake_request.headers.get("x-codex-server-id"),
Some(&refreshed_enrollment.server_id)
codex-rs/app-server/src/transport/remote_control/websocket.rs
+326 -26
View File
@@ -17,6 +17,8 @@ use super::protocol::ServerEnvelope;
use super::protocol::StreamId;
use axum::http::HeaderValue;
use base64::Engine;
use codex_app_server_protocol::RemoteControlConnectionStatus;
use codex_app_server_protocol::RemoteControlStatusChangedNotification;
use codex_core::util::backoff;
use codex_login::AuthManager;
use codex_login::UnauthorizedRecovery;
@@ -117,6 +119,7 @@ pub(crate) struct RemoteControlWebsocket {
remote_control_target: Option<RemoteControlTarget>,
state_db: Option<Arc<StateRuntime>>,
auth_manager: Arc<AuthManager>,
status_publisher: RemoteControlStatusPublisher,
shutdown_token: CancellationToken,
reconnect_attempt: u64,
enrollment: Option<RemoteControlEnrollment>,
@@ -134,20 +137,82 @@ enum ConnectOutcome {
Shutdown,
}
pub(super) struct RemoteControlChannels {
pub(super) transport_event_tx: mpsc::Sender<TransportEvent>,
pub(super) status_publisher: RemoteControlStatusPublisher,
}
#[derive(Clone)]
pub(super) struct RemoteControlStatusPublisher {
tx: watch::Sender<RemoteControlStatusChangedNotification>,
}
impl RemoteControlStatusPublisher {
pub(super) fn new(tx: watch::Sender<RemoteControlStatusChangedNotification>) -> Self {
Self { tx }
}
fn publish_status(&self, connection_status: RemoteControlConnectionStatus) {
self.tx.send_if_modified(|status| {
let next_status = RemoteControlStatusChangedNotification {
status: connection_status,
environment_id: if connection_status == RemoteControlConnectionStatus::Disabled {
None
} else {
status.environment_id.clone()
},
};
if *status == next_status {
return false;
}
*status = next_status;
true
});
}
fn publish_environment_id(&self, environment_id: Option<String>) {
self.tx.send_if_modified(|status| {
if status.status == RemoteControlConnectionStatus::Disabled {
return false;
}
let next_status = RemoteControlStatusChangedNotification {
status: status.status,
environment_id,
};
if *status == next_status {
return false;
}
*status = next_status;
true
});
}
}
#[derive(Clone, Copy)]
pub(super) struct RemoteControlConnectOptions<'a> {
subscribe_cursor: Option<&'a str>,
app_server_client_name: Option<&'a str>,
}
impl RemoteControlWebsocket {
pub(crate) fn new(
remote_control_url: String,
remote_control_target: Option<RemoteControlTarget>,
state_db: Option<Arc<StateRuntime>>,
auth_manager: Arc<AuthManager>,
transport_event_tx: mpsc::Sender<TransportEvent>,
channels: RemoteControlChannels,
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);
let client_tracker =
ClientTracker::new(server_event_tx, transport_event_tx, &shutdown_token);
let client_tracker = ClientTracker::new(
server_event_tx,
channels.transport_event_tx,
&shutdown_token,
);
let (outbound_buffer, used_rx) = BoundedOutboundBuffer::new();
let auth_recovery = auth_manager.unauthorized_recovery();
@@ -156,6 +221,7 @@ impl RemoteControlWebsocket {
remote_control_target,
state_db,
auth_manager,
status_publisher: channels.status_publisher,
shutdown_token,
reconnect_attempt: 0,
enrollment: None,
@@ -202,7 +268,11 @@ impl RemoteControlWebsocket {
.await
{
ConnectOutcome::Connected(websocket_connection) => *websocket_connection,
ConnectOutcome::Disabled => continue,
ConnectOutcome::Disabled => {
self.status_publisher
.publish_status(RemoteControlConnectionStatus::Disabled);
continue;
}
ConnectOutcome::Shutdown => break,
};
@@ -243,6 +313,8 @@ impl RemoteControlWebsocket {
shutdown_token: &CancellationToken,
app_server_client_name: Option<&str>,
) -> ConnectOutcome {
self.status_publisher
.publish_status(RemoteControlConnectionStatus::Connecting);
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) {
@@ -251,6 +323,8 @@ impl RemoteControlWebsocket {
remote_control_target
}
Err(err) => {
self.status_publisher
.publish_status(RemoteControlConnectionStatus::Errored);
warn!("remote control is enabled but the URL is invalid: {err}");
tokio::select! {
_ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown,
@@ -267,6 +341,10 @@ impl RemoteControlWebsocket {
loop {
let subscribe_cursor = self.state.lock().await.subscribe_cursor.clone();
let connect_options = RemoteControlConnectOptions {
subscribe_cursor: subscribe_cursor.as_deref(),
app_server_client_name,
};
let connect_result = tokio::select! {
_ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown,
changed = self.enabled_rx.wait_for(|enabled| !*enabled) => {
@@ -281,15 +359,20 @@ impl RemoteControlWebsocket {
&self.auth_manager,
&mut self.auth_recovery,
&mut self.enrollment,
subscribe_cursor.as_deref(),
app_server_client_name,
connect_options,
&self.status_publisher,
) => connect_result,
};
match connect_result {
Ok((websocket_connection, response)) => {
if !*self.enabled_rx.borrow() {
return ConnectOutcome::Disabled;
}
self.reconnect_attempt = 0;
self.auth_recovery = self.auth_manager.unauthorized_recovery();
self.status_publisher
.publish_status(RemoteControlConnectionStatus::Connected);
info!(
"connected to app-server remote control websocket: {}, {}",
remote_control_target.websocket_url,
@@ -298,9 +381,14 @@ impl RemoteControlWebsocket {
return ConnectOutcome::Connected(Box::new(websocket_connection));
}
Err(err) => {
if !*self.enabled_rx.borrow() {
return ConnectOutcome::Disabled;
}
let reconnect_delay = if err.kind() == ErrorKind::WouldBlock {
REMOTE_CONTROL_ACCOUNT_ID_RETRY_INTERVAL
} else {
self.status_publisher
.publish_status(RemoteControlConnectionStatus::Errored);
warn!(
"failed to connect to app-server remote control websocket: {}, err: {}",
remote_control_target.websocket_url, err
@@ -351,9 +439,15 @@ impl RemoteControlWebsocket {
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(),
}
changed = enabled_rx.wait_for(|enabled| !*enabled) => {
if changed.is_ok() {
self.status_publisher
.publish_status(RemoteControlConnectionStatus::Disabled);
}
}
_ = join_set.join_next() => {}
};
shutdown_token.cancel();
join_set.join_all().await;
}
@@ -745,8 +839,8 @@ pub(super) async fn connect_remote_control_websocket(
auth_manager: &Arc<AuthManager>,
auth_recovery: &mut UnauthorizedRecovery,
enrollment: &mut Option<RemoteControlEnrollment>,
subscribe_cursor: Option<&str>,
app_server_client_name: Option<&str>,
connect_options: RemoteControlConnectOptions<'_>,
status_publisher: &RemoteControlStatusPublisher,
) -> io::Result<(
WebSocketStream<MaybeTlsStream<TcpStream>>,
tungstenite::http::Response<()>,
@@ -761,7 +855,16 @@ pub(super) async fn connect_remote_control_websocket(
));
};
let auth = load_remote_control_auth(auth_manager).await?;
let auth = match load_remote_control_auth(auth_manager).await {
Ok(auth) => auth,
Err(err) => {
if err.kind() == ErrorKind::PermissionDenied {
*enrollment = None;
status_publisher.publish_environment_id(/*environment_id*/ None);
}
return Err(err);
}
};
let enrollment_account_id = enrollment.as_ref().map(|enrollment| &enrollment.account_id);
if enrollment_account_id.is_some_and(|account_id| account_id != &auth.account_id) {
info!(
@@ -773,16 +876,25 @@ pub(super) async fn connect_remote_control_websocket(
auth.account_id
);
*enrollment = None;
status_publisher.publish_environment_id(/*environment_id*/ None);
}
if let Some(enrollment) = enrollment.as_ref() {
status_publisher.publish_environment_id(Some(enrollment.environment_id.clone()));
}
if enrollment.is_none() {
*enrollment = load_persisted_remote_control_enrollment(
let loaded_enrollment = load_persisted_remote_control_enrollment(
Some(state_db),
remote_control_target,
&auth.account_id,
app_server_client_name,
connect_options.app_server_client_name,
)
.await?;
if let Some(loaded_enrollment) = loaded_enrollment.as_ref() {
status_publisher.publish_environment_id(Some(loaded_enrollment.environment_id.clone()));
}
*enrollment = loaded_enrollment;
}
if enrollment.is_none() {
@@ -807,7 +919,7 @@ pub(super) async fn connect_remote_control_websocket(
Some(state_db),
remote_control_target,
&auth.account_id,
app_server_client_name,
connect_options.app_server_client_name,
Some(&new_enrollment),
)
.await
@@ -823,6 +935,7 @@ pub(super) async fn connect_remote_control_websocket(
new_enrollment.server_id,
new_enrollment.environment_id
);
status_publisher.publish_environment_id(Some(new_enrollment.environment_id.clone()));
*enrollment = Some(new_enrollment);
}
@@ -833,7 +946,7 @@ pub(super) async fn connect_remote_control_websocket(
&remote_control_target.websocket_url,
enrollment_ref,
&auth,
subscribe_cursor,
connect_options.subscribe_cursor,
)?;
match connect_async(request).await {
@@ -852,7 +965,7 @@ pub(super) async fn connect_remote_control_websocket(
Some(state_db),
remote_control_target,
&auth.account_id,
app_server_client_name,
connect_options.app_server_client_name,
/*enrollment*/ None,
)
.await
@@ -862,6 +975,7 @@ pub(super) async fn connect_remote_control_websocket(
);
}
*enrollment = None;
status_publisher.publish_environment_id(/*environment_id*/ None);
}
tungstenite::Error::Http(response)
if matches!(response.status().as_u16(), 401 | 403) =>
@@ -968,6 +1082,17 @@ mod tests {
#[cfg(not(windows))]
const TEST_HTTP_ACCEPT_TIMEOUT: Duration = Duration::from_secs(5);
fn remote_control_status_channel() -> (
RemoteControlStatusPublisher,
watch::Receiver<RemoteControlStatusChangedNotification>,
) {
let (status_tx, status_rx) = watch::channel(RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connecting,
environment_id: None,
});
(RemoteControlStatusPublisher::new(status_tx), status_rx)
}
async fn remote_control_state_runtime(codex_home: &TempDir) -> Arc<StateRuntime> {
StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string())
.await
@@ -1059,6 +1184,7 @@ mod tests {
server_id: "srv_e_test".to_string(),
server_name: "test-server".to_string(),
});
let (status_publisher, status_rx) = remote_control_status_channel();
let err = match connect_remote_control_websocket(
&remote_control_target,
@@ -1066,8 +1192,11 @@ mod tests {
&auth_manager,
&mut auth_recovery,
&mut enrollment,
/*subscribe_cursor*/ None,
/*app_server_client_name*/ None,
RemoteControlConnectOptions {
subscribe_cursor: None,
app_server_client_name: None,
},
&status_publisher,
)
.await
{
@@ -1077,6 +1206,13 @@ mod tests {
server_task.await.expect("server task should succeed");
assert_eq!(err.to_string(), expected_error);
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connecting,
environment_id: Some("env_test".to_string()),
}
);
}
#[tokio::test]
@@ -1109,6 +1245,7 @@ mod tests {
server_id: "srv_e_test".to_string(),
server_name: "test-server".to_string(),
});
let (status_publisher, status_rx) = remote_control_status_channel();
save_auth(
codex_home.path(),
&remote_control_auth_dot_json("fresh-token"),
@@ -1131,13 +1268,23 @@ mod tests {
&auth_manager,
&mut auth_recovery,
&mut enrollment,
/*subscribe_cursor*/ None,
/*app_server_client_name*/ None,
RemoteControlConnectOptions {
subscribe_cursor: None,
app_server_client_name: None,
},
&status_publisher,
)
.await
.expect_err("unauthorized response should fail the websocket connect");
server_task.await.expect("server task should succeed");
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connecting,
environment_id: Some("env_test".to_string()),
}
);
assert_eq!(
err.to_string(),
"remote control websocket auth failed with HTTP 401 Unauthorized; retrying after auth recovery"
@@ -1187,6 +1334,7 @@ mod tests {
.await;
let mut auth_recovery = auth_manager.unauthorized_recovery();
let mut enrollment = None;
let (status_publisher, status_rx) = remote_control_status_channel();
save_auth(
codex_home.path(),
&remote_control_auth_dot_json("fresh-token"),
@@ -1200,13 +1348,21 @@ mod tests {
&auth_manager,
&mut auth_recovery,
&mut enrollment,
/*subscribe_cursor*/ None,
/*app_server_client_name*/ None,
RemoteControlConnectOptions {
subscribe_cursor: None,
app_server_client_name: None,
},
&status_publisher,
)
.await
.expect_err("unauthorized enrollment should fail the websocket connect");
server_task.await.expect("server task should succeed");
assert!(
!status_rx
.has_changed()
.expect("remote control status watch should remain open")
);
assert_eq!(
err.to_string(),
format!(
@@ -1236,6 +1392,7 @@ mod tests {
server_id: "srv_e_test".to_string(),
server_name: "test-server".to_string(),
});
let (status_publisher, _status_rx) = remote_control_status_channel();
let err = connect_remote_control_websocket(
&remote_control_target,
@@ -1243,8 +1400,11 @@ mod tests {
&auth_manager,
&mut auth_recovery,
&mut enrollment,
/*subscribe_cursor*/ None,
/*app_server_client_name*/ None,
RemoteControlConnectOptions {
subscribe_cursor: None,
app_server_client_name: None,
},
&status_publisher,
)
.await
.expect_err("missing sqlite state db should fail remote control");
@@ -1254,6 +1414,63 @@ mod tests {
assert_eq!(enrollment, None);
}
#[tokio::test]
async fn connect_remote_control_websocket_requires_chatgpt_auth() {
let remote_control_target = normalize_remote_control_url("http://127.0.0.1:9/backend-api/")
.expect("target should parse");
let codex_home = TempDir::new().expect("temp dir should create");
let state_db = remote_control_state_runtime(&codex_home).await;
let auth_manager = AuthManager::shared(
codex_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await;
let mut auth_recovery = auth_manager.unauthorized_recovery();
let mut enrollment = Some(RemoteControlEnrollment {
account_id: "account_id".to_string(),
environment_id: "env_test".to_string(),
server_id: "srv_e_test".to_string(),
server_name: "test-server".to_string(),
});
let (status_publisher, mut status_rx) = remote_control_status_channel();
status_publisher.publish_environment_id(Some("env_test".to_string()));
status_rx
.changed()
.await
.expect("remote control status watch should remain open");
let err = connect_remote_control_websocket(
&remote_control_target,
Some(state_db.as_ref()),
&auth_manager,
&mut auth_recovery,
&mut enrollment,
RemoteControlConnectOptions {
subscribe_cursor: None,
app_server_client_name: None,
},
&status_publisher,
)
.await
.expect_err("missing auth should fail remote control");
assert_eq!(err.kind(), ErrorKind::PermissionDenied);
assert_eq!(
err.to_string(),
"remote control requires ChatGPT authentication"
);
assert_eq!(enrollment, None);
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connecting,
environment_id: None,
}
);
}
#[tokio::test]
async fn run_remote_control_websocket_loop_shutdown_cancels_reconnect_backoff() {
let listener = TcpListener::bind("127.0.0.1:0")
@@ -1266,6 +1483,7 @@ mod tests {
normalize_remote_control_url(&remote_control_url).expect("target should parse");
let (transport_event_tx, transport_event_rx) = mpsc::channel(1);
drop(transport_event_rx);
let (status_publisher, _status_rx) = remote_control_status_channel();
let shutdown_token = CancellationToken::new();
let (_enabled_tx, enabled_rx) = watch::channel(true);
let websocket_task = tokio::spawn({
@@ -1276,7 +1494,10 @@ mod tests {
Some(remote_control_target),
/*state_db*/ None,
remote_control_auth_manager(),
transport_event_tx,
RemoteControlChannels {
transport_event_tx,
status_publisher,
},
shutdown_token,
enabled_rx,
)
@@ -1294,6 +1515,85 @@ mod tests {
.expect("websocket task should join");
}
#[tokio::test]
async fn publish_status_if_changed_sends_only_status_changes() {
let (status_publisher, mut status_rx) = remote_control_status_channel();
status_publisher.publish_environment_id(/*environment_id*/ None);
assert!(
timeout(Duration::from_millis(20), status_rx.changed())
.await
.is_err()
);
status_publisher.publish_environment_id(Some("env_first".to_string()));
status_rx
.changed()
.await
.expect("remote control status watch should remain open");
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connecting,
environment_id: Some("env_first".to_string()),
}
);
status_publisher.publish_environment_id(Some("env_first".to_string()));
assert!(
timeout(Duration::from_millis(20), status_rx.changed())
.await
.is_err()
);
status_publisher.publish_status(RemoteControlConnectionStatus::Connected);
status_rx
.changed()
.await
.expect("remote control status watch should remain open");
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connected,
environment_id: Some("env_first".to_string()),
}
);
status_publisher.publish_environment_id(/*environment_id*/ None);
status_rx
.changed()
.await
.expect("remote control status watch should remain open");
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Connected,
environment_id: None,
}
);
status_publisher.publish_environment_id(Some("env_disabled".to_string()));
status_publisher.publish_status(RemoteControlConnectionStatus::Disabled);
status_rx
.changed()
.await
.expect("remote control status watch should remain open");
assert_eq!(
status_rx.borrow().clone(),
RemoteControlStatusChangedNotification {
status: RemoteControlConnectionStatus::Disabled,
environment_id: None,
}
);
status_publisher.publish_environment_id(Some("env_disabled".to_string()));
assert!(
timeout(Duration::from_millis(20), status_rx.changed())
.await
.is_err()
);
}
#[tokio::test]
async fn run_server_writer_inner_sends_periodic_ping_frames() {
let (client_stream, mut server_stream) = connected_websocket_pair().await;
codex-rs/tui/src/app/app_server_adapter.rs
+1
View File
@@ -441,6 +441,7 @@ fn server_notification_thread_target(
| ServerNotification::AccountUpdated(_)
| ServerNotification::AccountRateLimitsUpdated(_)
| ServerNotification::AppListUpdated(_)
| ServerNotification::RemoteControlStatusChanged(_)
| ServerNotification::ExternalAgentConfigImportCompleted(_)
| ServerNotification::DeprecationNotice(_)
| ServerNotification::ConfigWarning(_)
codex-rs/tui/src/chatwidget.rs
+1
View File
@@ -7261,6 +7261,7 @@ impl ChatWidget {
| ServerNotification::McpToolCallProgress(_)
| ServerNotification::McpServerOauthLoginCompleted(_)
| ServerNotification::AppListUpdated(_)
| ServerNotification::RemoteControlStatusChanged(_)
| ServerNotification::ExternalAgentConfigImportCompleted(_)
| ServerNotification::FsChanged(_)
| ServerNotification::FuzzyFileSearchSessionUpdated(_)