Finish moving codex exec to app-server (#15424)

This PR completes the conversion of non-interactive `codex exec` to use
app server rather than directly using core events and methods.

### Summary
- move `codex-exec` off exec-owned `AuthManager` and `ThreadManager`
state
- route exec bootstrap, resume, and auth refresh through existing
app-server paths
- replace legacy `codex/event/*` decoding in exec with typed app-server
notification handling
- update human and JSONL exec output adapters to translate existing
app-server notifications only
- clean up "app server client" layer by eliminating support for legacy
notifications; this is no longer needed
- remove exposure of `authManager` and `threadManager` from "app server
client" layer

### Testing
- `exec` has pretty extensive unit and integration tests already, and
these all pass
- In addition, I asked Codex to put together a comprehensive manual set
of tests to cover all of the `codex exec` functionality (including
command-line options), and it successfully generated and ran these tests
This commit is contained in:
Eric Traut authored and GitHub committed 2026-03-24 08:51:32 -06:00
1 parent 1db6cb9789
commit 45f68843b8
22 files changed
+3438 -4238

No files matched your search

-5
View File
@@ -1489,7 +1489,6 @@ dependencies = [
"codex-app-server-protocol", "codex-app-server-protocol",
"codex-arg0", "codex-arg0",
"codex-core", "codex-core",
"codex-features",
"codex-feedback", "codex-feedback",
"codex-protocol", "codex-protocol",
"futures", "futures",
@@ -1999,9 +1998,7 @@ dependencies = [
"codex-utils-absolute-path", "codex-utils-absolute-path",
"codex-utils-cargo-bin", "codex-utils-cargo-bin",
"codex-utils-cli", "codex-utils-cli",
"codex-utils-elapsed",
"codex-utils-oss", "codex-utils-oss",
"codex-utils-sandbox-summary",
"core_test_support", "core_test_support",
"libc", "libc",
"opentelemetry", "opentelemetry",
@@ -2009,10 +2006,8 @@ dependencies = [
"owo-colors", "owo-colors",
"predicates", "predicates",
"pretty_assertions", "pretty_assertions",
"rmcp",
"serde", "serde",
"serde_json", "serde_json",
"shlex",
"supports-color 3.0.2", "supports-color 3.0.2",
"tempfile", "tempfile",
"tokio", "tokio",
-1
View File
@@ -16,7 +16,6 @@ codex-app-server = { workspace = true }
codex-app-server-protocol = { workspace = true } codex-app-server-protocol = { workspace = true }
codex-arg0 = { workspace = true } codex-arg0 = { workspace = true }
codex-core = { workspace = true } codex-core = { workspace = true }
codex-features = { workspace = true }
codex-feedback = { workspace = true } codex-feedback = { workspace = true }
codex-protocol = { workspace = true } codex-protocol = { workspace = true }
futures = { workspace = true } futures = { workspace = true }
+60 -127
View File
@@ -35,19 +35,14 @@ use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::RequestId; use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result as JsonRpcResult; use codex_app_server_protocol::Result as JsonRpcResult;
use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequest;
use codex_arg0::Arg0DispatchPaths; use codex_arg0::Arg0DispatchPaths;
use codex_core::AuthManager;
use codex_core::ThreadManager;
use codex_core::config::Config; use codex_core::config::Config;
use codex_core::config_loader::CloudRequirementsLoader; use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides; use codex_core::config_loader::LoaderOverrides;
use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_features::Feature;
use codex_feedback::CodexFeedback; use codex_feedback::CodexFeedback;
use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SessionSource;
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
@@ -73,7 +68,6 @@ pub type RequestResult = std::result::Result<JsonRpcResult, JSONRPCErrorError>;
pub enum AppServerEvent { pub enum AppServerEvent {
Lagged { skipped: usize }, Lagged { skipped: usize },
ServerNotification(ServerNotification), ServerNotification(ServerNotification),
LegacyNotification(JSONRPCNotification),
ServerRequest(ServerRequest), ServerRequest(ServerRequest),
Disconnected { message: String }, Disconnected { message: String },
} }
@@ -85,9 +79,6 @@ impl From<InProcessServerEvent> for AppServerEvent {
InProcessServerEvent::ServerNotification(notification) => { InProcessServerEvent::ServerNotification(notification) => {
Self::ServerNotification(notification) Self::ServerNotification(notification)
} }
InProcessServerEvent::LegacyNotification(notification) => {
Self::LegacyNotification(notification)
}
InProcessServerEvent::ServerRequest(request) => Self::ServerRequest(request), InProcessServerEvent::ServerRequest(request) => Self::ServerRequest(request),
} }
} }
@@ -97,19 +88,12 @@ fn event_requires_delivery(event: &InProcessServerEvent) -> bool {
// These terminal events drive surface shutdown/completion state. Dropping // These terminal events drive surface shutdown/completion state. Dropping
// them under backpressure can leave exec/TUI waiting forever even though // them under backpressure can leave exec/TUI waiting forever even though
// the underlying turn has already ended. // the underlying turn has already ended.
match event { matches!(
event,
InProcessServerEvent::ServerNotification( InProcessServerEvent::ServerNotification(
codex_app_server_protocol::ServerNotification::TurnCompleted(_), codex_app_server_protocol::ServerNotification::TurnCompleted(_),
) => true, )
InProcessServerEvent::LegacyNotification(notification) => matches!( )
notification
.method
.strip_prefix("codex/event/")
.unwrap_or(&notification.method),
"task_complete" | "turn_aborted" | "shutdown_complete"
),
_ => false,
}
} }
/// Layered error for [`InProcessAppServerClient::request_typed`]. /// Layered error for [`InProcessAppServerClient::request_typed`].
@@ -159,16 +143,6 @@ impl Error for TypedRequestError {
} }
} }
#[derive(Clone)]
struct SharedCoreManagers {
// Temporary bootstrap escape hatch for embedders that still need direct
// core handles during the in-process app-server migration. Once TUI/exec
// stop depending on direct manager access, remove this wrapper and keep
// manager ownership entirely inside the app-server runtime.
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
}
#[derive(Clone)] #[derive(Clone)]
pub struct InProcessClientStartArgs { pub struct InProcessClientStartArgs {
/// Resolved argv0 dispatch paths used by command execution internals. /// Resolved argv0 dispatch paths used by command execution internals.
@@ -202,30 +176,6 @@ pub struct InProcessClientStartArgs {
} }
impl InProcessClientStartArgs { impl InProcessClientStartArgs {
fn shared_core_managers(&self) -> SharedCoreManagers {
let auth_manager = AuthManager::shared(
self.config.codex_home.clone(),
self.enable_codex_api_key_env,
self.config.cli_auth_credentials_store_mode,
);
let thread_manager = Arc::new(ThreadManager::new(
self.config.as_ref(),
auth_manager.clone(),
self.session_source.clone(),
CollaborationModesConfig {
default_mode_request_user_input: self
.config
.features
.enabled(Feature::DefaultModeRequestUserInput),
},
));
SharedCoreManagers {
auth_manager,
thread_manager,
}
}
/// Builds initialize params from caller-provided metadata. /// Builds initialize params from caller-provided metadata.
pub fn initialize_params(&self) -> InitializeParams { pub fn initialize_params(&self) -> InitializeParams {
let capabilities = InitializeCapabilities { let capabilities = InitializeCapabilities {
@@ -247,7 +197,7 @@ impl InProcessClientStartArgs {
} }
} }
fn into_runtime_start_args(self, shared_core: &SharedCoreManagers) -> InProcessStartArgs { fn into_runtime_start_args(self) -> InProcessStartArgs {
let initialize = self.initialize_params(); let initialize = self.initialize_params();
InProcessStartArgs { InProcessStartArgs {
arg0_paths: self.arg0_paths, arg0_paths: self.arg0_paths,
@@ -255,8 +205,6 @@ impl InProcessClientStartArgs {
cli_overrides: self.cli_overrides, cli_overrides: self.cli_overrides,
loader_overrides: self.loader_overrides, loader_overrides: self.loader_overrides,
cloud_requirements: self.cloud_requirements, cloud_requirements: self.cloud_requirements,
auth_manager: Some(shared_core.auth_manager.clone()),
thread_manager: Some(shared_core.thread_manager.clone()),
feedback: self.feedback, feedback: self.feedback,
config_warnings: self.config_warnings, config_warnings: self.config_warnings,
session_source: self.session_source, session_source: self.session_source,
@@ -310,8 +258,6 @@ pub struct InProcessAppServerClient {
command_tx: mpsc::Sender<ClientCommand>, command_tx: mpsc::Sender<ClientCommand>,
event_rx: mpsc::Receiver<InProcessServerEvent>, event_rx: mpsc::Receiver<InProcessServerEvent>,
worker_handle: tokio::task::JoinHandle<()>, worker_handle: tokio::task::JoinHandle<()>,
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
} }
#[derive(Clone)] #[derive(Clone)]
@@ -338,9 +284,8 @@ impl InProcessAppServerClient {
/// with overload error instead of being silently dropped. /// with overload error instead of being silently dropped.
pub async fn start(args: InProcessClientStartArgs) -> IoResult<Self> { pub async fn start(args: InProcessClientStartArgs) -> IoResult<Self> {
let channel_capacity = args.channel_capacity.max(1); let channel_capacity = args.channel_capacity.max(1);
let shared_core = args.shared_core_managers();
let mut handle = let mut handle =
codex_app_server::in_process::start(args.into_runtime_start_args(&shared_core)).await?; codex_app_server::in_process::start(args.into_runtime_start_args()).await?;
let request_sender = handle.sender(); let request_sender = handle.sender();
let (command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(channel_capacity); let (command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(channel_capacity);
let (event_tx, event_rx) = mpsc::channel::<InProcessServerEvent>(channel_capacity); let (event_tx, event_rx) = mpsc::channel::<InProcessServerEvent>(channel_capacity);
@@ -401,6 +346,25 @@ impl InProcessAppServerClient {
let Some(event) = event else { let Some(event) = event else {
break; break;
}; };
if let InProcessServerEvent::ServerRequest(
ServerRequest::ChatgptAuthTokensRefresh { request_id, .. }
) = &event
{
let send_result = request_sender.fail_server_request(
request_id.clone(),
JSONRPCErrorError {
code: -32000,
message: "chatgpt auth token refresh is not supported for in-process app-server clients".to_string(),
data: None,
},
);
if let Err(err) = send_result {
warn!(
"failed to reject unsupported chatgpt auth token refresh request: {err}"
);
}
continue;
}
if skipped_events > 0 { if skipped_events > 0 {
if event_requires_delivery(&event) { if event_requires_delivery(&event) {
@@ -491,21 +455,9 @@ impl InProcessAppServerClient {
command_tx, command_tx,
event_rx, event_rx,
worker_handle, worker_handle,
auth_manager: shared_core.auth_manager,
thread_manager: shared_core.thread_manager,
}) })
} }
/// Temporary bootstrap escape hatch for embedders migrating toward RPC-only usage.
pub fn auth_manager(&self) -> Arc<AuthManager> {
self.auth_manager.clone()
}
/// Temporary bootstrap escape hatch for embedders migrating toward RPC-only usage.
pub fn thread_manager(&self) -> Arc<ThreadManager> {
self.thread_manager.clone()
}
pub fn request_handle(&self) -> InProcessAppServerRequestHandle { pub fn request_handle(&self) -> InProcessAppServerRequestHandle {
InProcessAppServerRequestHandle { InProcessAppServerRequestHandle {
command_tx: self.command_tx.clone(), command_tx: self.command_tx.clone(),
@@ -664,8 +616,6 @@ impl InProcessAppServerClient {
command_tx, command_tx,
event_rx, event_rx,
worker_handle, worker_handle,
auth_manager: _,
thread_manager: _,
} = self; } = self;
let mut worker_handle = worker_handle; let mut worker_handle = worker_handle;
// Drop the caller-facing receiver before asking the worker to shut // Drop the caller-facing receiver before asking the worker to shut
@@ -857,8 +807,6 @@ mod tests {
use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ToolRequestUserInputParams; use codex_app_server_protocol::ToolRequestUserInputParams;
use codex_app_server_protocol::ToolRequestUserInputQuestion; use codex_app_server_protocol::ToolRequestUserInputQuestion;
use codex_core::AuthManager;
use codex_core::ThreadManager;
use codex_core::config::ConfigBuilder; use codex_core::config::ConfigBuilder;
use futures::SinkExt; use futures::SinkExt;
use futures::StreamExt; use futures::StreamExt;
@@ -1052,7 +1000,7 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn shared_thread_manager_tracks_threads_started_via_app_server() { async fn threads_started_via_app_server_are_visible_through_typed_requests() {
let client = start_test_client(SessionSource::Cli).await; let client = start_test_client(SessionSource::Cli).await;
let response: ThreadStartResponse = client let response: ThreadStartResponse = client
@@ -1065,17 +1013,19 @@ mod tests {
}) })
.await .await
.expect("thread/start should succeed"); .expect("thread/start should succeed");
let created_thread_id = codex_protocol::ThreadId::from_string(&response.thread.id) let read = client
.expect("thread id should parse"); .request_typed::<codex_app_server_protocol::ThreadReadResponse>(
timeout( ClientRequest::ThreadRead {
Duration::from_secs(2), request_id: RequestId::Integer(4),
client.thread_manager().get_thread(created_thread_id), params: codex_app_server_protocol::ThreadReadParams {
) thread_id: response.thread.id.clone(),
.await include_turns: false,
.expect("timed out waiting for retained thread manager to observe started thread") },
.expect("started thread should be visible through the shared thread manager"); },
let thread_ids = client.thread_manager().list_thread_ids().await; )
assert!(thread_ids.contains(&created_thread_id)); .await
.expect("thread/read should return the newly started thread");
assert_eq!(read.thread.id, response.thread.id);
client.shutdown().await.expect("shutdown should complete"); client.shutdown().await.expect("shutdown should complete");
} }
@@ -1472,22 +1422,6 @@ mod tests {
let (command_tx, _command_rx) = mpsc::channel(1); let (command_tx, _command_rx) = mpsc::channel(1);
let (event_tx, event_rx) = mpsc::channel(1); let (event_tx, event_rx) = mpsc::channel(1);
let worker_handle = tokio::spawn(async {}); let worker_handle = tokio::spawn(async {});
let config = build_test_config().await;
let auth_manager = AuthManager::shared(
config.codex_home.clone(),
false,
config.cli_auth_credentials_store_mode,
);
let thread_manager = Arc::new(ThreadManager::new(
&config,
auth_manager.clone(),
SessionSource::Exec,
CollaborationModesConfig {
default_mode_request_user_input: config
.features
.enabled(Feature::DefaultModeRequestUserInput),
},
));
event_tx event_tx
.send(InProcessServerEvent::Lagged { skipped: 3 }) .send(InProcessServerEvent::Lagged { skipped: 3 })
.await .await
@@ -1498,8 +1432,6 @@ mod tests {
command_tx, command_tx,
event_rx, event_rx,
worker_handle, worker_handle,
auth_manager,
thread_manager,
}; };
let event = timeout(Duration::from_secs(2), client.next_event()) let event = timeout(Duration::from_secs(2), client.next_event())
@@ -1530,37 +1462,38 @@ mod tests {
) )
) )
)); ));
assert!(event_requires_delivery(
&InProcessServerEvent::LegacyNotification(
codex_app_server_protocol::JSONRPCNotification {
method: "codex/event/turn_aborted".to_string(),
params: None,
}
)
));
assert!(!event_requires_delivery(&InProcessServerEvent::Lagged { assert!(!event_requires_delivery(&InProcessServerEvent::Lagged {
skipped: 1 skipped: 1
})); }));
} }
#[tokio::test] #[tokio::test]
async fn accessors_expose_retained_shared_managers() { async fn runtime_start_args_leave_manager_bootstrap_to_app_server() {
let client = start_test_client(SessionSource::Cli).await; let config = Arc::new(build_test_config().await);
assert!( let runtime_args = InProcessClientStartArgs {
Arc::ptr_eq(&client.auth_manager(), &client.auth_manager()), arg0_paths: Arg0DispatchPaths::default(),
"auth_manager accessor should clone the retained shared manager" config: config.clone(),
); cli_overrides: Vec::new(),
assert!( loader_overrides: LoaderOverrides::default(),
Arc::ptr_eq(&client.thread_manager(), &client.thread_manager()), cloud_requirements: CloudRequirementsLoader::default(),
"thread_manager accessor should clone the retained shared manager" feedback: CodexFeedback::new(),
); config_warnings: Vec::new(),
session_source: SessionSource::Exec,
enable_codex_api_key_env: false,
client_name: "codex-app-server-client-test".to_string(),
client_version: "0.0.0-test".to_string(),
experimental_api: true,
opt_out_notification_methods: Vec::new(),
channel_capacity: DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
}
.into_runtime_start_args();
client.shutdown().await.expect("shutdown should complete"); assert_eq!(runtime_args.config, config);
} }
#[tokio::test] #[tokio::test]
async fn shutdown_completes_promptly_with_retained_shared_managers() { async fn shutdown_completes_promptly_without_retained_managers() {
let client = start_test_client(SessionSource::Cli).await; let client = start_test_client(SessionSource::Cli).await;
timeout(Duration::from_secs(1), client.shutdown()) timeout(Duration::from_secs(1), client.shutdown())
+20 -24
View File
@@ -272,18 +272,19 @@ impl RemoteAppServerClient {
} }
} }
Ok(JSONRPCMessage::Notification(notification)) => { Ok(JSONRPCMessage::Notification(notification)) => {
let event = app_server_event_from_notification(notification); if let Some(event) =
if let Err(err) = deliver_event( app_server_event_from_notification(notification)
&event_tx, && let Err(err) = deliver_event(
&mut skipped_events, &event_tx,
event, &mut skipped_events,
&mut stream, event,
) &mut stream,
.await )
{ .await
warn!(%err, "failed to deliver remote app-server event"); {
break; warn!(%err, "failed to deliver remote app-server event");
} break;
}
} }
Ok(JSONRPCMessage::Request(request)) => { Ok(JSONRPCMessage::Request(request)) => {
let request_id = request.id.clone(); let request_id = request.id.clone();
@@ -673,7 +674,9 @@ async fn initialize_remote_connection(
))); )));
} }
JSONRPCMessage::Notification(notification) => { JSONRPCMessage::Notification(notification) => {
pending_events.push(app_server_event_from_notification(notification)); if let Some(event) = app_server_event_from_notification(notification) {
pending_events.push(event);
}
} }
JSONRPCMessage::Request(request) => { JSONRPCMessage::Request(request) => {
let request_id = request.id.clone(); let request_id = request.id.clone();
@@ -756,10 +759,10 @@ async fn initialize_remote_connection(
Ok(pending_events) Ok(pending_events)
} }
fn app_server_event_from_notification(notification: JSONRPCNotification) -> AppServerEvent { fn app_server_event_from_notification(notification: JSONRPCNotification) -> Option<AppServerEvent> {
match ServerNotification::try_from(notification.clone()) { match ServerNotification::try_from(notification) {
Ok(notification) => AppServerEvent::ServerNotification(notification), Ok(notification) => Some(AppServerEvent::ServerNotification(notification)),
Err(_) => AppServerEvent::LegacyNotification(notification), Err(_) => None,
} }
} }
@@ -852,13 +855,6 @@ async fn reject_if_server_request_dropped(
fn event_requires_delivery(event: &AppServerEvent) -> bool { fn event_requires_delivery(event: &AppServerEvent) -> bool {
match event { match event {
AppServerEvent::ServerNotification(ServerNotification::TurnCompleted(_)) => true, AppServerEvent::ServerNotification(ServerNotification::TurnCompleted(_)) => true,
AppServerEvent::LegacyNotification(notification) => matches!(
notification
.method
.strip_prefix("codex/event/")
.unwrap_or(&notification.method),
"task_complete" | "turn_aborted" | "shutdown_complete"
),
AppServerEvent::Disconnected { .. } => true, AppServerEvent::Disconnected { .. } => true,
AppServerEvent::Lagged { .. } AppServerEvent::Lagged { .. }
| AppServerEvent::ServerNotification(_) | AppServerEvent::ServerNotification(_)
@@ -12,7 +12,6 @@ use crate::models::supported_models;
use crate::outgoing_message::ConnectionId; use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::ConnectionRequestId; use crate::outgoing_message::ConnectionRequestId;
use crate::outgoing_message::OutgoingMessageSender; use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotification;
use crate::outgoing_message::RequestContext; use crate::outgoing_message::RequestContext;
use crate::outgoing_message::ThreadScopedOutgoingMessageSender; use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
use crate::thread_status::ThreadWatchManager; use crate::thread_status::ThreadWatchManager;
@@ -6831,43 +6830,9 @@ impl CodexMessageProcessor {
} }
}; };
// For now, we send a notification for every event, // Track the event before emitting any typed
// Legacy `codex/event/*` notifications are still // translations so thread-local state such as raw event
// produced here because the in-process app-server lane // opt-in stays synchronized with the conversation.
// (`codex exec` and other in-process consumers) still
// depends on them. External transports now drop
// `OutgoingMessage::Notification` in `transport.rs`,
// so stdio/websocket clients only observe the typed
// `ServerNotification` translations emitted below.
//
// TODO: remove this raw legacy-notification emission
// entirely once the remaining in-process consumers are
// migrated off `codex/event/*`.
let event_formatted = match &event.msg {
EventMsg::TurnStarted(_) => "task_started",
EventMsg::TurnComplete(_) => "task_complete",
_ => &event.msg.to_string(),
};
let request_event_name = format!("codex/event/{event_formatted}");
tracing::trace!(
conversation_id = %conversation_id,
"app-server event: {request_event_name}"
);
let mut params = match serde_json::to_value(event.clone()) {
Ok(serde_json::Value::Object(map)) => map,
Ok(_) => {
error!("event did not serialize to an object");
continue;
}
Err(err) => {
error!("failed to serialize event: {err}");
continue;
}
};
params.insert(
"conversationId".to_string(),
conversation_id.to_string().into(),
);
let raw_events_enabled = { let raw_events_enabled = {
let mut thread_state = thread_state.lock().await; let mut thread_state = thread_state.lock().await;
thread_state.track_current_turn_event(&event.msg); thread_state.track_current_turn_event(&event.msg);
@@ -6880,18 +6845,6 @@ impl CodexMessageProcessor {
continue; continue;
} }
if !subscribed_connection_ids.is_empty() {
outgoing_for_task
.send_notification_to_connections(
&subscribed_connection_ids,
OutgoingNotification {
method: request_event_name,
params: Some(params.into()),
},
)
.await;
}
let thread_outgoing = ThreadScopedOutgoingMessageSender::new( let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing_for_task.clone(), outgoing_for_task.clone(),
subscribed_connection_ids, subscribed_connection_ids,
+1 -81
View File
@@ -68,14 +68,11 @@ use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::RequestId; use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result; use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequest;
use codex_arg0::Arg0DispatchPaths; use codex_arg0::Arg0DispatchPaths;
use codex_core::AuthManager;
use codex_core::ThreadManager;
use codex_core::config::Config; use codex_core::config::Config;
use codex_core::config_loader::CloudRequirementsLoader; use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides; use codex_core::config_loader::LoaderOverrides;
@@ -98,16 +95,6 @@ fn server_notification_requires_delivery(notification: &ServerNotification) -> b
matches!(notification, ServerNotification::TurnCompleted(_)) matches!(notification, ServerNotification::TurnCompleted(_))
} }
fn legacy_notification_requires_delivery(notification: &JSONRPCNotification) -> bool {
matches!(
notification
.method
.strip_prefix("codex/event/")
.unwrap_or(&notification.method),
"task_complete" | "turn_aborted" | "shutdown_complete"
)
}
/// Input needed to start an in-process app-server runtime. /// Input needed to start an in-process app-server runtime.
/// ///
/// These fields mirror the pieces of ambient process state that stdio and /// These fields mirror the pieces of ambient process state that stdio and
@@ -124,10 +111,6 @@ pub struct InProcessStartArgs {
pub loader_overrides: LoaderOverrides, pub loader_overrides: LoaderOverrides,
/// Preloaded cloud requirements provider. /// Preloaded cloud requirements provider.
pub cloud_requirements: CloudRequirementsLoader, pub cloud_requirements: CloudRequirementsLoader,
/// Optional prebuilt auth manager reused by an embedding caller.
pub auth_manager: Option<Arc<AuthManager>>,
/// Optional prebuilt thread manager reused by an embedding caller.
pub thread_manager: Option<Arc<ThreadManager>>,
/// Feedback sink used by app-server/core telemetry and logs. /// Feedback sink used by app-server/core telemetry and logs.
pub feedback: CodexFeedback, pub feedback: CodexFeedback,
/// Startup warnings emitted after initialize succeeds. /// Startup warnings emitted after initialize succeeds.
@@ -144,11 +127,6 @@ pub struct InProcessStartArgs {
/// Event emitted from the app-server to the in-process client. /// Event emitted from the app-server to the in-process client.
/// ///
/// The stream carries three event families because CLI surfaces are mid-migration
/// from the legacy `codex_protocol::Event` model to the typed app-server
/// notification model. Once all surfaces consume only [`ServerNotification`],
/// [`LegacyNotification`](Self::LegacyNotification) can be removed.
///
/// [`Lagged`](Self::Lagged) is a transport health marker, not an application /// [`Lagged`](Self::Lagged) is a transport health marker, not an application
/// event — it signals that the consumer fell behind and some events were dropped. /// event — it signals that the consumer fell behind and some events were dropped.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
@@ -157,8 +135,6 @@ pub enum InProcessServerEvent {
ServerRequest(ServerRequest), ServerRequest(ServerRequest),
/// App-server notification directed to the embedded client. /// App-server notification directed to the embedded client.
ServerNotification(ServerNotification), ServerNotification(ServerNotification),
/// Legacy JSON-RPC notification from core event bridge.
LegacyNotification(JSONRPCNotification),
/// Indicates one or more events were dropped due to backpressure. /// Indicates one or more events were dropped due to backpressure.
Lagged { skipped: usize }, Lagged { skipped: usize },
} }
@@ -390,7 +366,6 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
Arc::clone(&outbound_initialized), Arc::clone(&outbound_initialized),
Arc::clone(&outbound_experimental_api_enabled), Arc::clone(&outbound_experimental_api_enabled),
Arc::clone(&outbound_opted_out_notification_methods), Arc::clone(&outbound_opted_out_notification_methods),
/*allow_legacy_notifications*/ true,
/*disconnect_sender*/ None, /*disconnect_sender*/ None,
), ),
); );
@@ -410,8 +385,6 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
cli_overrides: args.cli_overrides, cli_overrides: args.cli_overrides,
loader_overrides: args.loader_overrides, loader_overrides: args.loader_overrides,
cloud_requirements: args.cloud_requirements, cloud_requirements: args.cloud_requirements,
auth_manager: args.auth_manager,
thread_manager: args.thread_manager,
feedback: args.feedback, feedback: args.feedback,
log_db: None, log_db: None,
config_warnings: args.config_warnings, config_warnings: args.config_warnings,
@@ -655,32 +628,6 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
} }
} }
} }
OutgoingMessage::Notification(notification) => {
let notification = JSONRPCNotification {
method: notification.method,
params: notification.params,
};
if legacy_notification_requires_delivery(&notification) {
if event_tx
.send(InProcessServerEvent::LegacyNotification(notification))
.await
.is_err()
{
break;
}
} else if let Err(send_error) =
event_tx.try_send(InProcessServerEvent::LegacyNotification(notification))
{
match send_error {
mpsc::error::TrySendError::Full(_) => {
warn!("dropping in-process legacy notification (queue full)");
}
mpsc::error::TrySendError::Closed(_) => {
break;
}
}
}
}
} }
} }
} }
@@ -759,8 +706,6 @@ mod tests {
cli_overrides: Vec::new(), cli_overrides: Vec::new(),
loader_overrides: LoaderOverrides::default(), loader_overrides: LoaderOverrides::default(),
cloud_requirements: CloudRequirementsLoader::default(), cloud_requirements: CloudRequirementsLoader::default(),
auth_manager: None,
thread_manager: None,
feedback: CodexFeedback::new(), feedback: CodexFeedback::new(),
config_warnings: Vec::new(), config_warnings: Vec::new(),
session_source, session_source,
@@ -858,7 +803,7 @@ mod tests {
} }
#[test] #[test]
fn guaranteed_delivery_helpers_cover_terminal_notifications() { fn guaranteed_delivery_helpers_cover_terminal_server_notifications() {
assert!(server_notification_requires_delivery( assert!(server_notification_requires_delivery(
&ServerNotification::TurnCompleted(TurnCompletedNotification { &ServerNotification::TurnCompleted(TurnCompletedNotification {
thread_id: "thread-1".to_string(), thread_id: "thread-1".to_string(),
@@ -870,30 +815,5 @@ mod tests {
}, },
}) })
)); ));
assert!(legacy_notification_requires_delivery(
&JSONRPCNotification {
method: "codex/event/task_complete".to_string(),
params: None,
}
));
assert!(legacy_notification_requires_delivery(
&JSONRPCNotification {
method: "codex/event/turn_aborted".to_string(),
params: None,
}
));
assert!(legacy_notification_requires_delivery(
&JSONRPCNotification {
method: "codex/event/shutdown_complete".to_string(),
params: None,
}
));
assert!(!legacy_notification_requires_delivery(
&JSONRPCNotification {
method: "codex/event/item_started".to_string(),
params: None,
}
));
} }
} }
-8
View File
@@ -104,8 +104,6 @@ enum OutboundControlEvent {
Opened { Opened {
connection_id: ConnectionId, connection_id: ConnectionId,
writer: mpsc::Sender<crate::outgoing_message::OutgoingMessage>, writer: mpsc::Sender<crate::outgoing_message::OutgoingMessage>,
// Allow codex/event/* notifications to be emitted.
allow_legacy_notifications: bool,
disconnect_sender: Option<CancellationToken>, disconnect_sender: Option<CancellationToken>,
initialized: Arc<AtomicBool>, initialized: Arc<AtomicBool>,
experimental_api_enabled: Arc<AtomicBool>, experimental_api_enabled: Arc<AtomicBool>,
@@ -562,7 +560,6 @@ pub async fn run_main_with_transport(
OutboundControlEvent::Opened { OutboundControlEvent::Opened {
connection_id, connection_id,
writer, writer,
allow_legacy_notifications,
disconnect_sender, disconnect_sender,
initialized, initialized,
experimental_api_enabled, experimental_api_enabled,
@@ -575,7 +572,6 @@ pub async fn run_main_with_transport(
initialized, initialized,
experimental_api_enabled, experimental_api_enabled,
opted_out_notification_methods, opted_out_notification_methods,
allow_legacy_notifications,
disconnect_sender, disconnect_sender,
), ),
); );
@@ -618,8 +614,6 @@ pub async fn run_main_with_transport(
cli_overrides, cli_overrides,
loader_overrides, loader_overrides,
cloud_requirements: cloud_requirements.clone(), cloud_requirements: cloud_requirements.clone(),
auth_manager: None,
thread_manager: None,
feedback: feedback.clone(), feedback: feedback.clone(),
log_db, log_db,
config_warnings, config_warnings,
@@ -675,7 +669,6 @@ pub async fn run_main_with_transport(
TransportEvent::ConnectionOpened { TransportEvent::ConnectionOpened {
connection_id, connection_id,
writer, writer,
allow_legacy_notifications,
disconnect_sender, disconnect_sender,
} => { } => {
let outbound_initialized = Arc::new(AtomicBool::new(false)); let outbound_initialized = Arc::new(AtomicBool::new(false));
@@ -687,7 +680,6 @@ pub async fn run_main_with_transport(
.send(OutboundControlEvent::Opened { .send(OutboundControlEvent::Opened {
connection_id, connection_id,
writer, writer,
allow_legacy_notifications,
disconnect_sender, disconnect_sender,
initialized: Arc::clone(&outbound_initialized), initialized: Arc::clone(&outbound_initialized),
experimental_api_enabled: Arc::clone( experimental_api_enabled: Arc::clone(
+15 -26
View File
@@ -172,8 +172,6 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) cli_overrides: Vec<(String, TomlValue)>, pub(crate) cli_overrides: Vec<(String, TomlValue)>,
pub(crate) loader_overrides: LoaderOverrides, pub(crate) loader_overrides: LoaderOverrides,
pub(crate) cloud_requirements: CloudRequirementsLoader, pub(crate) cloud_requirements: CloudRequirementsLoader,
pub(crate) auth_manager: Option<Arc<AuthManager>>,
pub(crate) thread_manager: Option<Arc<ThreadManager>>,
pub(crate) feedback: CodexFeedback, pub(crate) feedback: CodexFeedback,
pub(crate) log_db: Option<LogDbLayer>, pub(crate) log_db: Option<LogDbLayer>,
pub(crate) config_warnings: Vec<ConfigWarningNotification>, pub(crate) config_warnings: Vec<ConfigWarningNotification>,
@@ -192,36 +190,27 @@ impl MessageProcessor {
cli_overrides, cli_overrides,
loader_overrides, loader_overrides,
cloud_requirements, cloud_requirements,
auth_manager,
thread_manager,
feedback, feedback,
log_db, log_db,
config_warnings, config_warnings,
session_source, session_source,
enable_codex_api_key_env, enable_codex_api_key_env,
} = args; } = args;
let (auth_manager, thread_manager) = match (auth_manager, thread_manager) { let auth_manager = AuthManager::shared(
(Some(auth_manager), Some(thread_manager)) => (auth_manager, thread_manager), config.codex_home.clone(),
(None, None) => { enable_codex_api_key_env,
let auth_manager = AuthManager::shared( config.cli_auth_credentials_store_mode,
config.codex_home.clone(), );
enable_codex_api_key_env, let thread_manager = Arc::new(ThreadManager::new(
config.cli_auth_credentials_store_mode, config.as_ref(),
); auth_manager.clone(),
let thread_manager = Arc::new(ThreadManager::new( session_source,
config.as_ref(), CollaborationModesConfig {
auth_manager.clone(), default_mode_request_user_input: config
session_source, .features
CollaborationModesConfig { .enabled(Feature::DefaultModeRequestUserInput),
default_mode_request_user_input: config },
.features ));
.enabled(Feature::DefaultModeRequestUserInput),
},
));
(auth_manager, thread_manager)
}
_ => panic!("MessageProcessorArgs must provide both auth_manager and thread_manager"),
};
auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone()); auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone());
auth_manager.set_external_auth_refresher(Arc::new(ExternalAuthRefreshBridge { auth_manager.set_external_auth_refresher(Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(), outgoing: outgoing.clone(),
@@ -239,8 +239,6 @@ fn build_test_processor(
cli_overrides: Vec::new(), cli_overrides: Vec::new(),
loader_overrides: LoaderOverrides::default(), loader_overrides: LoaderOverrides::default(),
cloud_requirements: CloudRequirementsLoader::default(), cloud_requirements: CloudRequirementsLoader::default(),
auth_manager: None,
thread_manager: None,
feedback: CodexFeedback::new(), feedback: CodexFeedback::new(),
log_db: None, log_db: None,
config_warnings: Vec::new(), config_warnings: Vec::new(),
@@ -527,38 +527,6 @@ impl OutgoingMessageSender {
} }
} }
pub(crate) async fn send_notification_to_connections(
&self,
connection_ids: &[ConnectionId],
notification: OutgoingNotification,
) {
let outgoing_message = OutgoingMessage::Notification(notification);
if connection_ids.is_empty() {
if let Err(err) = self
.sender
.send(OutgoingEnvelope::Broadcast {
message: outgoing_message,
})
.await
{
warn!("failed to send notification to client: {err:?}");
}
return;
}
for connection_id in connection_ids {
if let Err(err) = self
.sender
.send(OutgoingEnvelope::ToConnection {
connection_id: *connection_id,
message: outgoing_message.clone(),
})
.await
{
warn!("failed to send notification to client: {err:?}");
}
}
}
pub(crate) async fn send_error( pub(crate) async fn send_error(
&self, &self,
request_id: ConnectionRequestId, request_id: ConnectionRequestId,
@@ -616,7 +584,6 @@ impl OutgoingMessageSender {
#[serde(untagged)] #[serde(untagged)]
pub(crate) enum OutgoingMessage { pub(crate) enum OutgoingMessage {
Request(ServerRequest), Request(ServerRequest),
Notification(OutgoingNotification),
/// AppServerNotification is specific to the case where this is run as an /// AppServerNotification is specific to the case where this is run as an
/// "app server" as opposed to an MCP server. /// "app server" as opposed to an MCP server.
AppServerNotification(ServerNotification), AppServerNotification(ServerNotification),
@@ -624,13 +591,6 @@ pub(crate) enum OutgoingMessage {
Error(OutgoingError), Error(OutgoingError),
} }
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingNotification {
pub method: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub params: Option<serde_json::Value>,
}
#[derive(Debug, Clone, PartialEq, Serialize)] #[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingResponse { pub(crate) struct OutgoingResponse {
pub id: RequestId, pub id: RequestId,
+84 -161
View File
@@ -188,7 +188,6 @@ pub(crate) enum TransportEvent {
ConnectionOpened { ConnectionOpened {
connection_id: ConnectionId, connection_id: ConnectionId,
writer: mpsc::Sender<OutgoingMessage>, writer: mpsc::Sender<OutgoingMessage>,
allow_legacy_notifications: bool,
disconnect_sender: Option<CancellationToken>, disconnect_sender: Option<CancellationToken>,
}, },
ConnectionClosed { ConnectionClosed {
@@ -226,7 +225,6 @@ pub(crate) struct OutboundConnectionState {
pub(crate) initialized: Arc<AtomicBool>, pub(crate) initialized: Arc<AtomicBool>,
pub(crate) experimental_api_enabled: Arc<AtomicBool>, pub(crate) experimental_api_enabled: Arc<AtomicBool>,
pub(crate) opted_out_notification_methods: Arc<RwLock<HashSet<String>>>, pub(crate) opted_out_notification_methods: Arc<RwLock<HashSet<String>>>,
pub(crate) allow_legacy_notifications: bool,
pub(crate) writer: mpsc::Sender<OutgoingMessage>, pub(crate) writer: mpsc::Sender<OutgoingMessage>,
disconnect_sender: Option<CancellationToken>, disconnect_sender: Option<CancellationToken>,
} }
@@ -237,14 +235,12 @@ impl OutboundConnectionState {
initialized: Arc<AtomicBool>, initialized: Arc<AtomicBool>,
experimental_api_enabled: Arc<AtomicBool>, experimental_api_enabled: Arc<AtomicBool>,
opted_out_notification_methods: Arc<RwLock<HashSet<String>>>, opted_out_notification_methods: Arc<RwLock<HashSet<String>>>,
allow_legacy_notifications: bool,
disconnect_sender: Option<CancellationToken>, disconnect_sender: Option<CancellationToken>,
) -> Self { ) -> Self {
Self { Self {
initialized, initialized,
experimental_api_enabled, experimental_api_enabled,
opted_out_notification_methods, opted_out_notification_methods,
allow_legacy_notifications,
writer, writer,
disconnect_sender, disconnect_sender,
} }
@@ -272,7 +268,6 @@ pub(crate) async fn start_stdio_connection(
.send(TransportEvent::ConnectionOpened { .send(TransportEvent::ConnectionOpened {
connection_id, connection_id,
writer: writer_tx, writer: writer_tx,
allow_legacy_notifications: false,
disconnect_sender: None, disconnect_sender: None,
}) })
.await .await
@@ -376,7 +371,6 @@ async fn run_websocket_connection(
.send(TransportEvent::ConnectionOpened { .send(TransportEvent::ConnectionOpened {
connection_id, connection_id,
writer: writer_tx, writer: writer_tx,
allow_legacy_notifications: false,
disconnect_sender: Some(disconnect_token.clone()), disconnect_sender: Some(disconnect_token.clone()),
}) })
.await .await
@@ -584,16 +578,6 @@ fn should_skip_notification_for_connection(
connection_state: &OutboundConnectionState, connection_state: &OutboundConnectionState,
message: &OutgoingMessage, message: &OutgoingMessage,
) -> bool { ) -> bool {
if !connection_state.allow_legacy_notifications
&& matches!(message, OutgoingMessage::Notification(_))
{
// Raw legacy `codex/event/*` notifications are still emitted upstream
// for in-process compatibility, but they are no longer part of the
// external app-server contract. Keep dropping them here until the
// producer path can be deleted entirely.
return true;
}
let Ok(opted_out_notification_methods) = connection_state.opted_out_notification_methods.read() let Ok(opted_out_notification_methods) = connection_state.opted_out_notification_methods.read()
else { else {
warn!("failed to read outbound opted-out notifications"); warn!("failed to read outbound opted-out notifications");
@@ -604,9 +588,6 @@ fn should_skip_notification_for_connection(
let method = notification.to_string(); let method = notification.to_string();
opted_out_notification_methods.contains(method.as_str()) opted_out_notification_methods.contains(method.as_str())
} }
OutgoingMessage::Notification(notification) => {
opted_out_notification_methods.contains(notification.method.as_str())
}
_ => false, _ => false,
} }
} }
@@ -719,6 +700,8 @@ mod tests {
use super::*; use super::*;
use crate::error_code::OVERLOADED_ERROR_CODE; use crate::error_code::OVERLOADED_ERROR_CODE;
use codex_app_server_protocol::CommandExecutionRequestApprovalSkillMetadata; use codex_app_server_protocol::CommandExecutionRequestApprovalSkillMetadata;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::ServerNotification;
use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use serde_json::json; use serde_json::json;
@@ -921,11 +904,13 @@ mod tests {
.expect("transport queue should accept first message"); .expect("transport queue should accept first message");
writer_tx writer_tx
.send(OutgoingMessage::Notification( .send(OutgoingMessage::AppServerNotification(
crate::outgoing_message::OutgoingNotification { ServerNotification::ConfigWarning(ConfigWarningNotification {
method: "queued".to_string(), summary: "queued".to_string(),
params: None, details: None,
}, path: None,
range: None,
}),
)) ))
.await .await
.expect("writer queue should accept first message"); .expect("writer queue should accept first message");
@@ -950,7 +935,16 @@ mod tests {
.await .await
.expect("writer queue should still contain original message"); .expect("writer queue should still contain original message");
let queued_json = serde_json::to_value(queued_outgoing).expect("serialize queued message"); let queued_json = serde_json::to_value(queued_outgoing).expect("serialize queued message");
assert_eq!(queued_json, json!({ "method": "queued" })); assert_eq!(
queued_json,
json!({
"method": "configWarning",
"params": {
"summary": "queued",
"details": null,
},
})
);
} }
#[tokio::test] #[tokio::test]
@@ -958,9 +952,8 @@ mod tests {
let connection_id = ConnectionId(7); let connection_id = ConnectionId(7);
let (writer_tx, mut writer_rx) = mpsc::channel(1); let (writer_tx, mut writer_rx) = mpsc::channel(1);
let initialized = Arc::new(AtomicBool::new(true)); let initialized = Arc::new(AtomicBool::new(true));
let opted_out_notification_methods = Arc::new(RwLock::new(HashSet::from([ let opted_out_notification_methods =
"codex/event/task_started".to_string(), Arc::new(RwLock::new(HashSet::from(["configWarning".to_string()])));
])));
let mut connections = HashMap::new(); let mut connections = HashMap::new();
connections.insert( connections.insert(
@@ -970,7 +963,6 @@ mod tests {
initialized, initialized,
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
opted_out_notification_methods, opted_out_notification_methods,
false,
None, None,
), ),
); );
@@ -979,12 +971,14 @@ mod tests {
&mut connections, &mut connections,
OutgoingEnvelope::ToConnection { OutgoingEnvelope::ToConnection {
connection_id, connection_id,
message: OutgoingMessage::Notification( message: OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
crate::outgoing_message::OutgoingNotification { ConfigWarningNotification {
method: "codex/event/task_started".to_string(), summary: "task_started".to_string(),
params: None, details: None,
path: None,
range: None,
}, },
), )),
}, },
) )
.await; .await;
@@ -995,89 +989,6 @@ mod tests {
); );
} }
#[tokio::test]
async fn to_connection_legacy_notifications_are_dropped_for_external_clients() {
let connection_id = ConnectionId(10);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
let mut connections = HashMap::new();
connections.insert(
connection_id,
OutboundConnectionState::new(
writer_tx,
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())),
false,
None,
),
);
route_outgoing_envelope(
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::Notification(
crate::outgoing_message::OutgoingNotification {
method: "codex/event/task_started".to_string(),
params: None,
},
),
},
)
.await;
assert!(
writer_rx.try_recv().is_err(),
"legacy notifications should not reach external clients"
);
}
#[tokio::test]
async fn to_connection_legacy_notifications_are_preserved_for_in_process_clients() {
let connection_id = ConnectionId(11);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
let mut connections = HashMap::new();
connections.insert(
connection_id,
OutboundConnectionState::new(
writer_tx,
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())),
true,
None,
),
);
route_outgoing_envelope(
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::Notification(
crate::outgoing_message::OutgoingNotification {
method: "codex/event/task_started".to_string(),
params: None,
},
),
},
)
.await;
let message = writer_rx
.recv()
.await
.expect("legacy notification should reach in-process clients");
assert!(matches!(
message,
OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification {
method,
params: None,
}) if method == "codex/event/task_started"
));
}
#[tokio::test] #[tokio::test]
async fn command_execution_request_approval_strips_experimental_fields_without_capability() { async fn command_execution_request_approval_strips_experimental_fields_without_capability() {
let connection_id = ConnectionId(8); let connection_id = ConnectionId(8);
@@ -1091,7 +1002,6 @@ mod tests {
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(false)), Arc::new(AtomicBool::new(false)),
Arc::new(RwLock::new(HashSet::new())), Arc::new(RwLock::new(HashSet::new())),
false,
None, None,
), ),
); );
@@ -1158,7 +1068,6 @@ mod tests {
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())), Arc::new(RwLock::new(HashSet::new())),
false,
None, None,
), ),
); );
@@ -1246,7 +1155,6 @@ mod tests {
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())), Arc::new(RwLock::new(HashSet::new())),
false,
Some(fast_disconnect_token.clone()), Some(fast_disconnect_token.clone()),
), ),
); );
@@ -1257,25 +1165,30 @@ mod tests {
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())), Arc::new(RwLock::new(HashSet::new())),
false,
Some(slow_disconnect_token.clone()), Some(slow_disconnect_token.clone()),
), ),
); );
let queued_message = let queued_message = OutgoingMessage::AppServerNotification(
OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { ServerNotification::ConfigWarning(ConfigWarningNotification {
method: "codex/event/already-buffered".to_string(), summary: "already-buffered".to_string(),
params: None, details: None,
}); path: None,
range: None,
}),
);
slow_writer_tx slow_writer_tx
.try_send(queued_message) .try_send(queued_message)
.expect("channel should have room"); .expect("channel should have room");
let broadcast_message = let broadcast_message = OutgoingMessage::AppServerNotification(
OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { ServerNotification::ConfigWarning(ConfigWarningNotification {
method: "codex/event/test".to_string(), summary: "test".to_string(),
params: None, details: None,
}); path: None,
range: None,
}),
);
timeout( timeout(
Duration::from_millis(100), Duration::from_millis(100),
route_outgoing_envelope( route_outgoing_envelope(
@@ -1286,24 +1199,28 @@ mod tests {
), ),
) )
.await .await
.expect("broadcast should return even when legacy notifications are dropped"); .expect("broadcast should return even when one connection is slow");
assert!(connections.contains_key(&slow_connection_id)); assert!(!connections.contains_key(&slow_connection_id));
assert!(!slow_disconnect_token.is_cancelled()); assert!(slow_disconnect_token.is_cancelled());
assert!(!fast_disconnect_token.is_cancelled()); assert!(!fast_disconnect_token.is_cancelled());
assert!( let fast_message = fast_writer_rx
fast_writer_rx.try_recv().is_err(), .try_recv()
"broadcast legacy notification should be dropped for fast connections" .expect("fast connection should receive the broadcast notification");
); assert!(matches!(
fast_message,
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification { summary, .. }
)) if summary == "test"
));
let slow_message = slow_writer_rx let slow_message = slow_writer_rx
.try_recv() .try_recv()
.expect("slow connection should retain its original buffered message"); .expect("slow connection should retain its original buffered message");
assert!(matches!( assert!(matches!(
slow_message, slow_message,
OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
method, ConfigWarningNotification { summary, .. }
params: None, )) if summary == "already-buffered"
}) if method == "codex/event/already-buffered"
)); ));
} }
@@ -1312,11 +1229,13 @@ mod tests {
let connection_id = ConnectionId(3); let connection_id = ConnectionId(3);
let (writer_tx, mut writer_rx) = mpsc::channel(1); let (writer_tx, mut writer_rx) = mpsc::channel(1);
writer_tx writer_tx
.send(OutgoingMessage::Notification( .send(OutgoingMessage::AppServerNotification(
crate::outgoing_message::OutgoingNotification { ServerNotification::ConfigWarning(ConfigWarningNotification {
method: "queued".to_string(), summary: "queued".to_string(),
params: None, details: None,
}, path: None,
range: None,
}),
)) ))
.await .await
.expect("channel should accept the first queued message"); .expect("channel should accept the first queued message");
@@ -1329,7 +1248,6 @@ mod tests {
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())), Arc::new(RwLock::new(HashSet::new())),
false,
None, None,
), ),
); );
@@ -1339,11 +1257,13 @@ mod tests {
&mut connections, &mut connections,
OutgoingEnvelope::ToConnection { OutgoingEnvelope::ToConnection {
connection_id, connection_id,
message: OutgoingMessage::Notification( message: OutgoingMessage::AppServerNotification(
crate::outgoing_message::OutgoingNotification { ServerNotification::ConfigWarning(ConfigWarningNotification {
method: "second".to_string(), summary: "second".to_string(),
params: None, details: None,
}, path: None,
range: None,
}),
), ),
}, },
) )
@@ -1356,20 +1276,23 @@ mod tests {
.expect("first queued message should exist"); .expect("first queued message should exist");
timeout(Duration::from_millis(100), route_task) timeout(Duration::from_millis(100), route_task)
.await .await
.expect("routing should finish immediately when legacy notifications are dropped") .expect("routing should finish after the first queued message is drained")
.expect("routing task should succeed"); .expect("routing task should succeed");
assert!(matches!( assert!(matches!(
first, first,
OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
method, ConfigWarningNotification { summary, .. }
params: None, )) if summary == "queued"
}) if method == "queued"
)); ));
let second = writer_rx
.try_recv()
.expect("second notification should be delivered once the queue has room");
assert!(matches!( assert!(matches!(
writer_rx.try_recv(), second,
Err(tokio::sync::mpsc::error::TryRecvError::Empty) OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
| Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) ConfigWarningNotification { summary, .. }
)) if summary == "second"
)); ));
} }
} }
+5 -4
View File
@@ -3,6 +3,7 @@ name = "codex-exec"
version.workspace = true version.workspace = true
edition.workspace = true edition.workspace = true
license.workspace = true license.workspace = true
autotests = false
[[bin]] [[bin]]
name = "codex-exec" name = "codex-exec"
@@ -12,6 +13,10 @@ path = "src/main.rs"
name = "codex_exec" name = "codex_exec"
path = "src/lib.rs" path = "src/lib.rs"
[[test]]
name = "all"
path = "tests/all.rs"
[lints] [lints]
workspace = true workspace = true
@@ -28,13 +33,10 @@ codex-otel = { workspace = true }
codex-protocol = { workspace = true } codex-protocol = { workspace = true }
codex-utils-absolute-path = { workspace = true } codex-utils-absolute-path = { workspace = true }
codex-utils-cli = { workspace = true } codex-utils-cli = { workspace = true }
codex-utils-elapsed = { workspace = true }
codex-utils-oss = { workspace = true } codex-utils-oss = { workspace = true }
codex-utils-sandbox-summary = { workspace = true }
owo-colors = { workspace = true } owo-colors = { workspace = true }
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true } serde_json = { workspace = true }
shlex = { workspace = true }
supports-color = { workspace = true } supports-color = { workspace = true }
tokio = { workspace = true, features = [ tokio = { workspace = true, features = [
"io-std", "io-std",
@@ -63,7 +65,6 @@ opentelemetry = { workspace = true }
opentelemetry_sdk = { workspace = true } opentelemetry_sdk = { workspace = true }
predicates = { workspace = true } predicates = { workspace = true }
pretty_assertions = { workspace = true } pretty_assertions = { workspace = true }
rmcp = { workspace = true }
tempfile = { workspace = true } tempfile = { workspace = true }
tracing-opentelemetry = { workspace = true } tracing-opentelemetry = { workspace = true }
uuid = { workspace = true } uuid = { workspace = true }
-4
View File
@@ -86,10 +86,6 @@ pub struct Cli {
#[arg(long = "color", value_enum, default_value_t = Color::Auto)] #[arg(long = "color", value_enum, default_value_t = Color::Auto)]
pub color: Color, pub color: Color,
/// Force cursor-based progress updates in exec mode.
#[arg(long = "progress-cursor", default_value_t = false)]
pub progress_cursor: bool,
/// Print events to stdout as JSONL. /// Print events to stdout as JSONL.
#[arg( #[arg(
long = "json", long = "json",
+8 -5
View File
@@ -1,13 +1,13 @@
use std::path::Path; use std::path::Path;
use codex_app_server_protocol::ServerNotification;
use codex_core::config::Config; use codex_core::config::Config;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionConfiguredEvent;
pub(crate) enum CodexStatus { #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CodexStatus {
Running, Running,
InitiateShutdown, InitiateShutdown,
Shutdown,
} }
pub(crate) trait EventProcessor { pub(crate) trait EventProcessor {
@@ -19,8 +19,11 @@ pub(crate) trait EventProcessor {
session_configured: &SessionConfiguredEvent, session_configured: &SessionConfiguredEvent,
); );
/// Handle a single event emitted by the agent. /// Handle a single typed app-server notification emitted by the agent.
fn process_event(&mut self, event: Event) -> CodexStatus; fn process_server_notification(&mut self, notification: ServerNotification) -> CodexStatus;
/// Handle a local exec warning that is not represented as an app-server notification.
fn process_warning(&mut self, message: String) -> CodexStatus;
fn print_final_output(&mut self) {} fn print_final_output(&mut self) {}
} }
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+434 -390
View File
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
@@ -178,10 +178,11 @@ trust_level = "trusted"
anyhow::bail!("timed out waiting for codex resume to exit"); anyhow::bail!("timed out waiting for codex resume to exit");
} }
}; };
let output_text = String::from_utf8_lossy(&output);
let interrupted_startup = exit_code == 1 && output_text.trim() == "^C";
anyhow::ensure!( anyhow::ensure!(
exit_code == 0 || exit_code == 130, exit_code == 0 || exit_code == 130 || interrupted_startup,
"unexpected exit code from codex resume: {exit_code}; output: {}", "unexpected exit code from codex resume: {exit_code}; output: {output_text}",
String::from_utf8_lossy(&output)
); );
let config_contents = std::fs::read_to_string(codex_home.path().join("config.toml"))?; let config_contents = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
@@ -122,9 +122,6 @@ impl App {
self.handle_server_notification_event(app_server_client, notification) self.handle_server_notification_event(app_server_client, notification)
.await; .await;
} }
AppServerEvent::LegacyNotification(_) => {
tracing::debug!("ignoring legacy app-server notification in tui_app_server");
}
AppServerEvent::ServerRequest(request) => { AppServerEvent::ServerRequest(request) => {
self.handle_server_request_event(app_server_client, request) self.handle_server_request_event(app_server_client, request)
.await; .await;
@@ -504,7 +504,6 @@ pub(crate) async fn run_onboarding_app(
return Err(color_eyre::eyre::eyre!(message)); return Err(color_eyre::eyre::eyre!(message));
} }
AppServerEvent::Lagged { .. } AppServerEvent::Lagged { .. }
| AppServerEvent::LegacyNotification(_)
| AppServerEvent::ServerRequest(_) => {} | AppServerEvent::ServerRequest(_) => {}
} }
} }
@@ -178,10 +178,11 @@ trust_level = "trusted"
anyhow::bail!("timed out waiting for codex resume to exit"); anyhow::bail!("timed out waiting for codex resume to exit");
} }
}; };
let output_text = String::from_utf8_lossy(&output);
let interrupted_startup = exit_code == 1 && output_text.trim() == "^C";
anyhow::ensure!( anyhow::ensure!(
exit_code == 0 || exit_code == 130, exit_code == 0 || exit_code == 130 || interrupted_startup,
"unexpected exit code from codex resume: {exit_code}; output: {}", "unexpected exit code from codex resume: {exit_code}; output: {output_text}",
String::from_utf8_lossy(&output)
); );
let config_contents = std::fs::read_to_string(codex_home.path().join("config.toml"))?; let config_contents = std::fs::read_to_string(codex_home.path().join("config.toml"))?;