openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into rhan/oob-thread-metadata

Browse Source
This commit is contained in:
rhan-oai
2026-03-26 21:33:51 -07:00
committed by GitHub
parent 960bd3e028 21a03f1671
commit b1ee6e9cec
330 changed files with 12468 additions and 9409 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
-5
View File
@@ -26,7 +26,6 @@ use codex_app_server_protocol::CommandExecutionApprovalDecision;
use codex_app_server_protocol::CommandExecutionOutputDeltaNotification;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
use codex_app_server_protocol::CommandExecutionRequestApprovalSkillMetadata;
use codex_app_server_protocol::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::ContextCompactedNotification;
@@ -608,7 +607,6 @@ pub(crate) async fn apply_bespoke_event_handling(
proposed_execpolicy_amendment,
proposed_network_policy_amendments,
additional_permissions,
skill_metadata,
parsed_cmd,
..
} = ev;
@@ -680,8 +678,6 @@ pub(crate) async fn apply_bespoke_event_handling(
});
let additional_permissions =
additional_permissions.map(V2AdditionalPermissionProfile::from);
let skill_metadata =
skill_metadata.map(CommandExecutionRequestApprovalSkillMetadata::from);
let params = CommandExecutionRequestApprovalParams {
thread_id: conversation_id.to_string(),
@@ -694,7 +690,6 @@ pub(crate) async fn apply_bespoke_event_handling(
cwd,
command_actions,
additional_permissions,
skill_metadata,
proposed_execpolicy_amendment: proposed_execpolicy_amendment_v2,
proposed_network_policy_amendments: proposed_network_policy_amendments_v2,
available_decisions: Some(available_decisions),
codex-rs/app-server/src/codex_message_processor.rs
+8 -2
View File
@@ -545,6 +545,7 @@ impl CodexMessageProcessor {
data: None,
})?;
apply_runtime_feature_enablement(&mut config, &self.current_runtime_feature_enablement());
config.codex_self_exe = self.arg0_paths.codex_self_exe.clone();
config.codex_linux_sandbox_exe = self.arg0_paths.codex_linux_sandbox_exe.clone();
config.main_execve_wrapper_exe = self.arg0_paths.main_execve_wrapper_exe.clone();
Ok(config)
@@ -1411,7 +1412,12 @@ impl CodexMessageProcessor {
requires_openai_auth: Some(false),
}
} else {
match self.auth_manager.auth().await {
let auth = if do_refresh {
self.auth_manager.auth_cached()
} else {
self.auth_manager.auth().await
};
match auth {
Some(auth) => {
let permanent_refresh_failure =
self.auth_manager.refresh_failure_for_auth(&auth).is_some();
@@ -7947,7 +7953,7 @@ fn validate_dynamic_tools(tools: &[ApiDynamicToolSpec]) -> Result<(), String> {
return Err(format!("duplicate dynamic tool name: {name}"));
}
if let Err(err) = codex_core::parse_tool_input_schema(&tool.input_schema) {
if let Err(err) = codex_tools::parse_tool_input_schema(&tool.input_schema) {
return Err(format!(
"dynamic tool input schema is not supported for {name}: {err}"
));
codex-rs/app-server/src/config_api.rs
+1 -1
View File
@@ -130,7 +130,7 @@ impl ConfigApi {
.unwrap_or_default()
}
async fn load_latest_config(
pub(crate) async fn load_latest_config(
&self,
fallback_cwd: Option<PathBuf>,
) -> Result<Config, JSONRPCErrorError> {
codex-rs/app-server/src/in_process.rs
+2
View File
@@ -77,6 +77,7 @@ use codex_arg0::Arg0DispatchPaths;
use codex_core::config::Config;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_exec_server::EnvironmentManager;
use codex_feedback::CodexFeedback;
use codex_protocol::protocol::SessionSource;
use tokio::sync::mpsc;
@@ -383,6 +384,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
outgoing: Arc::clone(&processor_outgoing),
arg0_paths: args.arg0_paths,
config: args.config,
environment_manager: Arc::new(EnvironmentManager::from_env()),
cli_overrides: args.cli_overrides,
loader_overrides: args.loader_overrides,
cloud_requirements: args.cloud_requirements,
codex-rs/app-server/src/lib.rs
+31 -49
View File
@@ -40,6 +40,7 @@ use codex_core::ExecPolicyError;
use codex_core::check_execpolicy_for_warnings;
use codex_core::config_loader::ConfigLoadError;
use codex_core::config_loader::TextRange as CoreTextRange;
use codex_exec_server::EnvironmentManager;
use codex_feedback::CodexFeedback;
use codex_protocol::protocol::SessionSource;
use codex_state::log_db;
@@ -355,44 +356,13 @@ pub async fn run_main_with_transport(
session_source: SessionSource,
auth: AppServerWebsocketAuthSettings,
) -> IoResult<()> {
let environment_manager = Arc::new(EnvironmentManager::from_env());
let (transport_event_tx, mut transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingEnvelope>(CHANNEL_CAPACITY);
let (outbound_control_tx, mut outbound_control_rx) =
mpsc::channel::<OutboundControlEvent>(CHANNEL_CAPACITY);
enum TransportRuntime {
Stdio,
WebSocket {
accept_handle: JoinHandle<()>,
shutdown_token: CancellationToken,
},
}
let mut stdio_handles = Vec::<JoinHandle<()>>::new();
let transport_runtime = match transport {
AppServerTransport::Stdio => {
start_stdio_connection(transport_event_tx.clone(), &mut stdio_handles).await?;
TransportRuntime::Stdio
}
AppServerTransport::WebSocket { bind_address } => {
let shutdown_token = CancellationToken::new();
let accept_handle = start_websocket_acceptor(
bind_address,
transport_event_tx.clone(),
shutdown_token.clone(),
policy_from_settings(&auth)?,
)
.await?;
TransportRuntime::WebSocket {
accept_handle,
shutdown_token,
}
}
};
let single_client_mode = matches!(&transport_runtime, TransportRuntime::Stdio);
let shutdown_when_no_connections = single_client_mode;
let graceful_signal_restart_enabled = !single_client_mode;
// Parse CLI overrides once and derive the base Config eagerly so later
// components do not need to work with raw TOML values.
let cli_kv_overrides = cli_config_overrides.parse_overrides().map_err(|e| {
@@ -556,6 +526,30 @@ pub async fn run_main_with_transport(
}
}
let transport_shutdown_token = CancellationToken::new();
let mut transport_accept_handles = Vec::<JoinHandle<()>>::new();
let single_client_mode = matches!(&transport, AppServerTransport::Stdio);
let shutdown_when_no_connections = single_client_mode;
let graceful_signal_restart_enabled = !single_client_mode;
match transport {
AppServerTransport::Stdio => {
start_stdio_connection(transport_event_tx.clone(), &mut transport_accept_handles)
.await?;
}
AppServerTransport::WebSocket { bind_address } => {
let accept_handle = start_websocket_acceptor(
bind_address,
transport_event_tx.clone(),
transport_shutdown_token.clone(),
policy_from_settings(&auth)?,
)
.await?;
transport_accept_handles.push(accept_handle);
}
}
let outbound_handle = tokio::spawn(async move {
let mut outbound_connections = HashMap::<ConnectionId, OutboundConnectionState>::new();
loop {
@@ -620,6 +614,7 @@ pub async fn run_main_with_transport(
outgoing: outgoing_message_sender,
arg0_paths,
config: Arc::new(config),
environment_manager,
cli_overrides,
loader_overrides,
cloud_requirements: cloud_requirements.clone(),
@@ -632,10 +627,7 @@ pub async fn run_main_with_transport(
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 websocket_accept_shutdown = match &transport_runtime {
TransportRuntime::WebSocket { shutdown_token, .. } => Some(shutdown_token.clone()),
TransportRuntime::Stdio => None,
};
let transport_shutdown_token = transport_shutdown_token.clone();
async move {
let mut listen_for_threads = true;
let mut shutdown_state = ShutdownState::default();
@@ -648,9 +640,7 @@ pub async fn run_main_with_transport(
shutdown_state.update(running_turn_count, connections.len()),
ShutdownAction::Finish
) {
if let Some(shutdown_token) = &websocket_accept_shutdown {
shutdown_token.cancel();
}
transport_shutdown_token.cancel();
let _ = outbound_control_tx
.send(OutboundControlEvent::DisconnectAll)
.await;
@@ -844,16 +834,8 @@ pub async fn run_main_with_transport(
let _ = processor_handle.await;
let _ = outbound_handle.await;
if let TransportRuntime::WebSocket {
accept_handle,
shutdown_token,
} = transport_runtime
{
shutdown_token.cancel();
let _ = accept_handle.await;
}
for handle in stdio_handles {
transport_shutdown_token.cancel();
for handle in transport_accept_handles {
let _ = handle.await;
}
codex-rs/app-server/src/message_processor.rs
+87 -7
View File
@@ -20,6 +20,7 @@ use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::RequestContext;
use crate::transport::AppServerTransport;
use async_trait::async_trait;
use codex_app_server_protocol::AppListUpdatedNotification;
use codex_app_server_protocol::ChatgptAuthTokensRefreshParams;
use codex_app_server_protocol::ChatgptAuthTokensRefreshReason;
use codex_app_server_protocol::ChatgptAuthTokensRefreshResponse;
@@ -53,6 +54,7 @@ use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::experimental_required_message;
use codex_arg0::Arg0DispatchPaths;
use codex_chatgpt::connectors;
use codex_core::AnalyticsEventsClient;
use codex_core::AuthManager;
use codex_core::ThreadManager;
@@ -66,6 +68,7 @@ use codex_core::default_client::get_codex_user_agent;
use codex_core::default_client::set_default_client_residency_requirement;
use codex_core::default_client::set_default_originator;
use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_exec_server::EnvironmentManager;
use codex_features::Feature;
use codex_feedback::CodexFeedback;
use codex_login::auth::ExternalAuthRefreshContext;
@@ -177,6 +180,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) arg0_paths: Arg0DispatchPaths,
pub(crate) config: Arc<Config>,
pub(crate) environment_manager: Arc<EnvironmentManager>,
pub(crate) cli_overrides: Vec<(String, TomlValue)>,
pub(crate) loader_overrides: LoaderOverrides,
pub(crate) cloud_requirements: CloudRequirementsLoader,
@@ -195,6 +199,7 @@ impl MessageProcessor {
outgoing,
arg0_paths,
config,
environment_manager,
cli_overrides,
loader_overrides,
cloud_requirements,
@@ -218,6 +223,7 @@ impl MessageProcessor {
.features
.enabled(Feature::DefaultModeRequestUserInput),
},
environment_manager,
));
auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone());
auth_manager.set_external_auth_refresher(Arc::new(ExternalAuthRefreshBridge {
@@ -882,13 +888,87 @@ impl MessageProcessor {
request_id: ConnectionRequestId,
params: ExperimentalFeatureEnablementSetParams,
) {
self.handle_config_mutation_result(
request_id,
self.config_api
.set_experimental_feature_enablement(params)
.await,
)
.await;
let should_refresh_apps_list = params.enablement.get("apps").copied() == Some(true);
match self
.config_api
.set_experimental_feature_enablement(params)
.await
{
Ok(response) => {
self.codex_message_processor.clear_plugin_related_caches();
self.codex_message_processor
.maybe_start_plugin_startup_tasks_for_latest_config()
.await;
self.outgoing.send_response(request_id, response).await;
if should_refresh_apps_list {
self.refresh_apps_list_after_experimental_feature_enablement_set()
.await;
}
}
Err(error) => self.outgoing.send_error(request_id, error).await,
}
}
async fn refresh_apps_list_after_experimental_feature_enablement_set(&self) {
let config = match self
.config_api
.load_latest_config(/*fallback_cwd*/ None)
.await
{
Ok(config) => config,
Err(error) => {
tracing::warn!(
"failed to load config for apps list refresh after experimental feature enablement: {}",
error.message
);
return;
}
};
if !config.features.apps_enabled(Some(&self.auth_manager)).await {
return;
}
let outgoing = Arc::clone(&self.outgoing);
tokio::spawn(async move {
let (all_connectors_result, accessible_connectors_result) = tokio::join!(
connectors::list_all_connectors_with_options(&config, /*force_refetch*/ true),
connectors::list_accessible_connectors_from_mcp_tools_with_options(
&config, /*force_refetch*/ true,
),
);
let all_connectors = match all_connectors_result {
Ok(connectors) => connectors,
Err(err) => {
tracing::warn!(
"failed to force-refresh directory apps after experimental feature enablement: {err:#}"
);
return;
}
};
let accessible_connectors = match accessible_connectors_result {
Ok(connectors) => connectors,
Err(err) => {
tracing::warn!(
"failed to force-refresh accessible apps after experimental feature enablement: {err:#}"
);
return;
}
};
let data = connectors::with_app_enabled_state(
connectors::merge_connectors_with_accessible(
all_connectors,
accessible_connectors,
/*all_connectors_loaded*/ true,
),
&config,
);
outgoing
.send_server_notification(ServerNotification::AppListUpdated(
AppListUpdatedNotification { data },
))
.await;
});
}
async fn handle_config_mutation_result<T: serde::Serialize>(
codex-rs/app-server/src/message_processor/tracing_tests.rs
+2
View File
@@ -24,6 +24,7 @@ use codex_core::config::Config;
use codex_core::config::ConfigBuilder;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_exec_server::EnvironmentManager;
use codex_feedback::CodexFeedback;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::W3cTraceContext;
@@ -236,6 +237,7 @@ fn build_test_processor(
outgoing,
arg0_paths: Arg0DispatchPaths::default(),
config,
environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
cli_overrides: Vec::new(),
loader_overrides: LoaderOverrides::default(),
cloud_requirements: CloudRequirementsLoader::default(),
codex-rs/app-server/src/transport.rs → codex-rs/app-server/src/transport/mod.rs
+7 -390
View File
@@ -1,8 +1,5 @@
pub(crate) mod auth;
use self::auth::WebsocketAuthPolicy;
use self::auth::authorize_upgrade;
use self::auth::should_warn_about_unauthenticated_non_loopback_listener;
use crate::error_code::OVERLOADED_ERROR_CODE;
use crate::message_processor::ConnectionSessionState;
use crate::outgoing_message::ConnectionId;
@@ -10,53 +7,20 @@ use crate::outgoing_message::OutgoingEnvelope;
use crate::outgoing_message::OutgoingError;
use crate::outgoing_message::OutgoingMessage;
use crate::outgoing_message::QueuedOutgoingMessage;
use axum::Router;
use axum::body::Body;
use axum::extract::ConnectInfo;
use axum::extract::State;
use axum::extract::ws::Message as WebSocketMessage;
use axum::extract::ws::WebSocket;
use axum::extract::ws::WebSocketUpgrade;
use axum::http::HeaderMap;
use axum::http::Request;
use axum::http::StatusCode;
use axum::http::header::ORIGIN;
use axum::middleware;
use axum::middleware::Next;
use axum::response::IntoResponse;
use axum::response::Response;
use axum::routing::any;
use axum::routing::get;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::ServerRequest;
use futures::SinkExt;
use futures::StreamExt;
use owo_colors::OwoColorize;
use owo_colors::Stream;
use owo_colors::Style;
use std::collections::HashMap;
use std::collections::HashSet;
use std::io::ErrorKind;
use std::io::Result as IoResult;
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::io::{self};
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::warn;
/// Size of the bounded channels used to communicate between tasks. The value
@@ -64,85 +28,11 @@ use tracing::warn;
/// plenty for an interactive CLI.
pub(crate) const CHANNEL_CAPACITY: usize = 128;
fn colorize(text: &str, style: Style) -> String {
text.if_supports_color(Stream::Stderr, |value| value.style(style))
.to_string()
}
mod stdio;
mod websocket;
#[allow(clippy::print_stderr)]
fn print_websocket_startup_banner(addr: SocketAddr) {
let title = colorize("codex app-server (WebSockets)", Style::new().bold().cyan());
let listening_label = colorize("listening on:", Style::new().dimmed());
let listen_url = colorize(&format!("ws://{addr}"), Style::new().green());
let ready_label = colorize("readyz:", Style::new().dimmed());
let ready_url = colorize(&format!("http://{addr}/readyz"), Style::new().green());
let health_label = colorize("healthz:", Style::new().dimmed());
let health_url = colorize(&format!("http://{addr}/healthz"), Style::new().green());
let note_label = colorize("note:", Style::new().dimmed());
eprintln!("{title}");
eprintln!(" {listening_label} {listen_url}");
eprintln!(" {ready_label} {ready_url}");
eprintln!(" {health_label} {health_url}");
if addr.ip().is_loopback() {
eprintln!(
" {note_label} binds localhost only (use SSH port-forwarding for remote access)"
);
} else {
eprintln!(
" {note_label} websocket auth is opt-in in this build; configure `--ws-auth ...` before real remote use"
);
}
}
#[derive(Clone)]
struct WebSocketListenerState {
transport_event_tx: mpsc::Sender<TransportEvent>,
connection_counter: Arc<AtomicU64>,
auth_policy: Arc<WebsocketAuthPolicy>,
}
async fn health_check_handler() -> StatusCode {
StatusCode::OK
}
async fn reject_requests_with_origin_header(
request: Request<Body>,
next: Next,
) -> Result<Response, StatusCode> {
if request.headers().contains_key(ORIGIN) {
warn!(
method = %request.method(),
uri = %request.uri(),
"rejecting websocket listener request with Origin header"
);
Err(StatusCode::FORBIDDEN)
} else {
Ok(next.run(request).await)
}
}
async fn websocket_upgrade_handler(
websocket: WebSocketUpgrade,
ConnectInfo(peer_addr): ConnectInfo<SocketAddr>,
State(state): State<WebSocketListenerState>,
headers: HeaderMap,
) -> impl IntoResponse {
if let Err(err) = authorize_upgrade(&headers, state.auth_policy.as_ref()) {
warn!(
%peer_addr,
message = err.message(),
"rejecting websocket client during upgrade"
);
return (err.status_code(), err.message()).into_response();
}
let connection_id = ConnectionId(state.connection_counter.fetch_add(1, Ordering::Relaxed));
info!(%peer_addr, "websocket client connected");
websocket
.on_upgrade(move |stream| async move {
run_websocket_connection(connection_id, stream, state.transport_event_tx).await;
})
.into_response()
}
pub(crate) use stdio::start_stdio_connection;
pub(crate) use websocket::start_websocket_acceptor;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum AppServerTransport {
@@ -276,262 +166,6 @@ impl OutboundConnectionState {
}
}
pub(crate) async fn start_stdio_connection(
transport_event_tx: mpsc::Sender<TransportEvent>,
stdio_handles: &mut Vec<JoinHandle<()>>,
) -> IoResult<()> {
let connection_id = ConnectionId(0);
let (writer_tx, mut writer_rx) = mpsc::channel::<QueuedOutgoingMessage>(CHANNEL_CAPACITY);
let writer_tx_for_reader = writer_tx.clone();
transport_event_tx
.send(TransportEvent::ConnectionOpened {
connection_id,
writer: writer_tx,
disconnect_sender: None,
})
.await
.map_err(|_| std::io::Error::new(ErrorKind::BrokenPipe, "processor unavailable"))?;
let transport_event_tx_for_reader = transport_event_tx.clone();
stdio_handles.push(tokio::spawn(async move {
let stdin = io::stdin();
let reader = BufReader::new(stdin);
let mut lines = reader.lines();
loop {
match lines.next_line().await {
Ok(Some(line)) => {
if !forward_incoming_message(
&transport_event_tx_for_reader,
&writer_tx_for_reader,
connection_id,
&line,
)
.await
{
break;
}
}
Ok(None) => break,
Err(err) => {
error!("Failed reading stdin: {err}");
break;
}
}
}
let _ = transport_event_tx_for_reader
.send(TransportEvent::ConnectionClosed { connection_id })
.await;
debug!("stdin reader finished (EOF)");
}));
stdio_handles.push(tokio::spawn(async move {
let mut stdout = io::stdout();
while let Some(queued_message) = writer_rx.recv().await {
let Some(mut json) = serialize_outgoing_message(queued_message.message) else {
continue;
};
json.push('\n');
if let Err(err) = stdout.write_all(json.as_bytes()).await {
error!("Failed to write to stdout: {err}");
break;
}
if let Some(write_complete_tx) = queued_message.write_complete_tx {
let _ = write_complete_tx.send(());
}
}
info!("stdout writer exited (channel closed)");
}));
Ok(())
}
pub(crate) async fn start_websocket_acceptor(
bind_address: SocketAddr,
transport_event_tx: mpsc::Sender<TransportEvent>,
shutdown_token: CancellationToken,
auth_policy: WebsocketAuthPolicy,
) -> IoResult<JoinHandle<()>> {
if should_warn_about_unauthenticated_non_loopback_listener(bind_address, &auth_policy) {
warn!(
%bind_address,
"starting non-loopback websocket listener without auth; websocket auth is opt-in for now and will become the default in a future release"
);
}
let listener = TcpListener::bind(bind_address).await?;
let local_addr = listener.local_addr()?;
print_websocket_startup_banner(local_addr);
info!("app-server websocket listening on ws://{local_addr}");
let router = Router::new()
.route("/readyz", get(health_check_handler))
.route("/healthz", get(health_check_handler))
.fallback(any(websocket_upgrade_handler))
.layer(middleware::from_fn(reject_requests_with_origin_header))
.with_state(WebSocketListenerState {
transport_event_tx,
connection_counter: Arc::new(AtomicU64::new(1)),
auth_policy: Arc::new(auth_policy),
});
let server = axum::serve(
listener,
router.into_make_service_with_connect_info::<SocketAddr>(),
)
.with_graceful_shutdown(async move {
shutdown_token.cancelled().await;
});
Ok(tokio::spawn(async move {
if let Err(err) = server.await {
error!("websocket acceptor failed: {err}");
}
info!("websocket acceptor shutting down");
}))
}
async fn run_websocket_connection(
connection_id: ConnectionId,
websocket_stream: WebSocket,
transport_event_tx: mpsc::Sender<TransportEvent>,
) {
let (writer_tx, writer_rx) = mpsc::channel::<QueuedOutgoingMessage>(CHANNEL_CAPACITY);
let writer_tx_for_reader = writer_tx.clone();
let disconnect_token = CancellationToken::new();
if transport_event_tx
.send(TransportEvent::ConnectionOpened {
connection_id,
writer: writer_tx,
disconnect_sender: Some(disconnect_token.clone()),
})
.await
.is_err()
{
return;
}
let (websocket_writer, websocket_reader) = websocket_stream.split();
let (writer_control_tx, writer_control_rx) =
mpsc::channel::<WebSocketMessage>(CHANNEL_CAPACITY);
let mut outbound_task = tokio::spawn(run_websocket_outbound_loop(
websocket_writer,
writer_rx,
writer_control_rx,
disconnect_token.clone(),
));
let mut inbound_task = tokio::spawn(run_websocket_inbound_loop(
websocket_reader,
transport_event_tx.clone(),
writer_tx_for_reader,
writer_control_tx,
connection_id,
disconnect_token.clone(),
));
tokio::select! {
_ = &mut outbound_task => {
disconnect_token.cancel();
inbound_task.abort();
}
_ = &mut inbound_task => {
disconnect_token.cancel();
outbound_task.abort();
}
}
let _ = transport_event_tx
.send(TransportEvent::ConnectionClosed { connection_id })
.await;
}
async fn run_websocket_outbound_loop(
mut websocket_writer: futures::stream::SplitSink<WebSocket, WebSocketMessage>,
mut writer_rx: mpsc::Receiver<QueuedOutgoingMessage>,
mut writer_control_rx: mpsc::Receiver<WebSocketMessage>,
disconnect_token: CancellationToken,
) {
loop {
tokio::select! {
_ = disconnect_token.cancelled() => {
break;
}
message = writer_control_rx.recv() => {
let Some(message) = message else {
break;
};
if websocket_writer.send(message).await.is_err() {
break;
}
}
queued_message = writer_rx.recv() => {
let Some(queued_message) = queued_message else {
break;
};
let Some(json) = serialize_outgoing_message(queued_message.message) else {
continue;
};
if websocket_writer.send(WebSocketMessage::Text(json.into())).await.is_err() {
break;
}
if let Some(write_complete_tx) = queued_message.write_complete_tx {
let _ = write_complete_tx.send(());
}
}
}
}
}
async fn run_websocket_inbound_loop(
mut websocket_reader: futures::stream::SplitStream<WebSocket>,
transport_event_tx: mpsc::Sender<TransportEvent>,
writer_tx_for_reader: mpsc::Sender<QueuedOutgoingMessage>,
writer_control_tx: mpsc::Sender<WebSocketMessage>,
connection_id: ConnectionId,
disconnect_token: CancellationToken,
) {
loop {
tokio::select! {
_ = disconnect_token.cancelled() => {
break;
}
incoming_message = websocket_reader.next() => {
match incoming_message {
Some(Ok(WebSocketMessage::Text(text))) => {
if !forward_incoming_message(
&transport_event_tx,
&writer_tx_for_reader,
connection_id,
text.as_ref(),
)
.await
{
break;
}
}
Some(Ok(WebSocketMessage::Ping(payload))) => {
match writer_control_tx.try_send(WebSocketMessage::Pong(payload)) {
Ok(()) => {}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => break,
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
warn!("websocket control queue full while replying to ping; closing connection");
break;
}
}
}
Some(Ok(WebSocketMessage::Pong(_))) => {}
Some(Ok(WebSocketMessage::Close(_))) | None => break,
Some(Ok(WebSocketMessage::Binary(_))) => {
warn!("dropping unsupported binary websocket message");
}
Some(Err(err)) => {
warn!("websocket receive error: {err}");
break;
}
}
}
}
}
}
async fn forward_incoming_message(
transport_event_tx: &mpsc::Sender<TransportEvent>,
writer: &mpsc::Sender<QueuedOutgoingMessage>,
@@ -745,7 +379,6 @@ pub(crate) async fn route_outgoing_envelope(
mod tests {
use super::*;
use crate::error_code::OVERLOADED_ERROR_CODE;
use codex_app_server_protocol::CommandExecutionRequestApprovalSkillMetadata;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::ServerNotification;
use codex_utils_absolute_path::AbsolutePathBuf;
@@ -1127,7 +760,7 @@ mod tests {
}
#[tokio::test]
async fn command_execution_request_approval_strips_experimental_fields_without_capability() {
async fn command_execution_request_approval_strips_additional_permissions_without_capability() {
let connection_id = ConnectionId(8);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
@@ -1168,12 +801,8 @@ mod tests {
write: None,
},
),
macos: None,
},
),
skill_metadata: Some(CommandExecutionRequestApprovalSkillMetadata {
path_to_skills_md: PathBuf::from("/tmp/SKILLS.md"),
}),
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
@@ -1190,11 +819,10 @@ mod tests {
.expect("request should be delivered to the connection");
let json = serde_json::to_value(message.message).expect("request should serialize");
assert_eq!(json["params"].get("additionalPermissions"), None);
assert_eq!(json["params"].get("skillMetadata"), None);
}
#[tokio::test]
async fn command_execution_request_approval_keeps_experimental_fields_with_capability() {
async fn command_execution_request_approval_keeps_additional_permissions_with_capability() {
let connection_id = ConnectionId(9);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
@@ -1235,12 +863,8 @@ mod tests {
write: None,
},
),
macos: None,
},
),
skill_metadata: Some(CommandExecutionRequestApprovalSkillMetadata {
path_to_skills_md: PathBuf::from("/tmp/SKILLS.md"),
}),
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
@@ -1263,15 +887,8 @@ mod tests {
"network": null,
"fileSystem": {
"read": [allowed_path],
"write": null,
"write": null,
},
"macos": null,
})
);
assert_eq!(
json["params"]["skillMetadata"],
json!({
"pathToSkillsMd": "/tmp/SKILLS.md",
})
);
}
codex-rs/app-server/src/transport/stdio.rs
+88
View File
@@ -0,0 +1,88 @@
use super::CHANNEL_CAPACITY;
use super::TransportEvent;
use super::forward_incoming_message;
use super::serialize_outgoing_message;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::QueuedOutgoingMessage;
use std::io::ErrorKind;
use std::io::Result as IoResult;
use tokio::io;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::debug;
use tracing::error;
use tracing::info;
pub(crate) async fn start_stdio_connection(
transport_event_tx: mpsc::Sender<TransportEvent>,
stdio_handles: &mut Vec<JoinHandle<()>>,
) -> IoResult<()> {
let connection_id = ConnectionId(0);
let (writer_tx, mut writer_rx) = mpsc::channel::<QueuedOutgoingMessage>(CHANNEL_CAPACITY);
let writer_tx_for_reader = writer_tx.clone();
transport_event_tx
.send(TransportEvent::ConnectionOpened {
connection_id,
writer: writer_tx,
disconnect_sender: None,
})
.await
.map_err(|_| std::io::Error::new(ErrorKind::BrokenPipe, "processor unavailable"))?;
let transport_event_tx_for_reader = transport_event_tx.clone();
stdio_handles.push(tokio::spawn(async move {
let stdin = io::stdin();
let reader = BufReader::new(stdin);
let mut lines = reader.lines();
loop {
match lines.next_line().await {
Ok(Some(line)) => {
if !forward_incoming_message(
&transport_event_tx_for_reader,
&writer_tx_for_reader,
connection_id,
&line,
)
.await
{
break;
}
}
Ok(None) => break,
Err(err) => {
error!("Failed reading stdin: {err}");
break;
}
}
}
let _ = transport_event_tx_for_reader
.send(TransportEvent::ConnectionClosed { connection_id })
.await;
debug!("stdin reader finished (EOF)");
}));
stdio_handles.push(tokio::spawn(async move {
let mut stdout = io::stdout();
while let Some(queued_message) = writer_rx.recv().await {
let Some(mut json) = serialize_outgoing_message(queued_message.message) else {
continue;
};
json.push('\n');
if let Err(err) = stdout.write_all(json.as_bytes()).await {
error!("Failed to write to stdout: {err}");
break;
}
if let Some(write_complete_tx) = queued_message.write_complete_tx {
let _ = write_complete_tx.send(());
}
}
info!("stdout writer exited (channel closed)");
}));
Ok(())
}
codex-rs/app-server/src/transport/websocket.rs
+308
View File
@@ -0,0 +1,308 @@
use super::CHANNEL_CAPACITY;
use super::TransportEvent;
use super::auth::WebsocketAuthPolicy;
use super::auth::authorize_upgrade;
use super::auth::should_warn_about_unauthenticated_non_loopback_listener;
use super::forward_incoming_message;
use super::serialize_outgoing_message;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::QueuedOutgoingMessage;
use axum::Router;
use axum::body::Body;
use axum::extract::ConnectInfo;
use axum::extract::State;
use axum::extract::ws::Message as WebSocketMessage;
use axum::extract::ws::WebSocket;
use axum::extract::ws::WebSocketUpgrade;
use axum::http::HeaderMap;
use axum::http::Request;
use axum::http::StatusCode;
use axum::http::header::ORIGIN;
use axum::middleware;
use axum::middleware::Next;
use axum::response::IntoResponse;
use axum::response::Response;
use axum::routing::any;
use axum::routing::get;
use futures::SinkExt;
use futures::StreamExt;
use owo_colors::OwoColorize;
use owo_colors::Stream;
use owo_colors::Style;
use std::io::Result as IoResult;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::error;
use tracing::info;
use tracing::warn;
fn colorize(text: &str, style: Style) -> String {
text.if_supports_color(Stream::Stderr, |value| value.style(style))
.to_string()
}
#[allow(clippy::print_stderr)]
fn print_websocket_startup_banner(addr: SocketAddr) {
let title = colorize("codex app-server (WebSockets)", Style::new().bold().cyan());
let listening_label = colorize("listening on:", Style::new().dimmed());
let listen_url = colorize(&format!("ws://{addr}"), Style::new().green());
let ready_label = colorize("readyz:", Style::new().dimmed());
let ready_url = colorize(&format!("http://{addr}/readyz"), Style::new().green());
let health_label = colorize("healthz:", Style::new().dimmed());
let health_url = colorize(&format!("http://{addr}/healthz"), Style::new().green());
let note_label = colorize("note:", Style::new().dimmed());
eprintln!("{title}");
eprintln!(" {listening_label} {listen_url}");
eprintln!(" {ready_label} {ready_url}");
eprintln!(" {health_label} {health_url}");
if addr.ip().is_loopback() {
eprintln!(
" {note_label} binds localhost only (use SSH port-forwarding for remote access)"
);
} else {
eprintln!(
" {note_label} websocket auth is opt-in in this build; configure `--ws-auth ...` before real remote use"
);
}
}
#[derive(Clone)]
struct WebSocketListenerState {
transport_event_tx: mpsc::Sender<TransportEvent>,
connection_counter: Arc<AtomicU64>,
auth_policy: Arc<WebsocketAuthPolicy>,
}
async fn health_check_handler() -> StatusCode {
StatusCode::OK
}
async fn reject_requests_with_origin_header(
request: Request<Body>,
next: Next,
) -> Result<Response, StatusCode> {
if request.headers().contains_key(ORIGIN) {
warn!(
method = %request.method(),
uri = %request.uri(),
"rejecting websocket listener request with Origin header"
);
Err(StatusCode::FORBIDDEN)
} else {
Ok(next.run(request).await)
}
}
async fn websocket_upgrade_handler(
websocket: WebSocketUpgrade,
ConnectInfo(peer_addr): ConnectInfo<SocketAddr>,
State(state): State<WebSocketListenerState>,
headers: HeaderMap,
) -> impl IntoResponse {
if let Err(err) = authorize_upgrade(&headers, state.auth_policy.as_ref()) {
warn!(
%peer_addr,
message = err.message(),
"rejecting websocket client during upgrade"
);
return (err.status_code(), err.message()).into_response();
}
let connection_id = ConnectionId(state.connection_counter.fetch_add(1, Ordering::Relaxed));
info!(%peer_addr, "websocket client connected");
websocket
.on_upgrade(move |stream| async move {
run_websocket_connection(connection_id, stream, state.transport_event_tx).await;
})
.into_response()
}
pub(crate) async fn start_websocket_acceptor(
bind_address: SocketAddr,
transport_event_tx: mpsc::Sender<TransportEvent>,
shutdown_token: CancellationToken,
auth_policy: WebsocketAuthPolicy,
) -> IoResult<JoinHandle<()>> {
if should_warn_about_unauthenticated_non_loopback_listener(bind_address, &auth_policy) {
warn!(
%bind_address,
"starting non-loopback websocket listener without auth; websocket auth is opt-in for now and will become the default in a future release"
);
}
let listener = TcpListener::bind(bind_address).await?;
let local_addr = listener.local_addr()?;
print_websocket_startup_banner(local_addr);
info!("app-server websocket listening on ws://{local_addr}");
let router = Router::new()
.route("/readyz", get(health_check_handler))
.route("/healthz", get(health_check_handler))
.fallback(any(websocket_upgrade_handler))
.layer(middleware::from_fn(reject_requests_with_origin_header))
.with_state(WebSocketListenerState {
transport_event_tx,
connection_counter: Arc::new(AtomicU64::new(1)),
auth_policy: Arc::new(auth_policy),
});
let server = axum::serve(
listener,
router.into_make_service_with_connect_info::<SocketAddr>(),
)
.with_graceful_shutdown(async move {
shutdown_token.cancelled().await;
});
Ok(tokio::spawn(async move {
if let Err(err) = server.await {
error!("websocket acceptor failed: {err}");
}
info!("websocket acceptor shutting down");
}))
}
async fn run_websocket_connection(
connection_id: ConnectionId,
websocket_stream: WebSocket,
transport_event_tx: mpsc::Sender<TransportEvent>,
) {
let (writer_tx, writer_rx) = mpsc::channel::<QueuedOutgoingMessage>(CHANNEL_CAPACITY);
let writer_tx_for_reader = writer_tx.clone();
let disconnect_token = CancellationToken::new();
if transport_event_tx
.send(TransportEvent::ConnectionOpened {
connection_id,
writer: writer_tx,
disconnect_sender: Some(disconnect_token.clone()),
})
.await
.is_err()
{
return;
}
let (websocket_writer, websocket_reader) = websocket_stream.split();
let (writer_control_tx, writer_control_rx) =
mpsc::channel::<WebSocketMessage>(CHANNEL_CAPACITY);
let mut outbound_task = tokio::spawn(run_websocket_outbound_loop(
websocket_writer,
writer_rx,
writer_control_rx,
disconnect_token.clone(),
));
let mut inbound_task = tokio::spawn(run_websocket_inbound_loop(
websocket_reader,
transport_event_tx.clone(),
writer_tx_for_reader,
writer_control_tx,
connection_id,
disconnect_token.clone(),
));
tokio::select! {
_ = &mut outbound_task => {
disconnect_token.cancel();
inbound_task.abort();
}
_ = &mut inbound_task => {
disconnect_token.cancel();
outbound_task.abort();
}
}
let _ = transport_event_tx
.send(TransportEvent::ConnectionClosed { connection_id })
.await;
}
async fn run_websocket_outbound_loop(
mut websocket_writer: futures::stream::SplitSink<WebSocket, WebSocketMessage>,
mut writer_rx: mpsc::Receiver<QueuedOutgoingMessage>,
mut writer_control_rx: mpsc::Receiver<WebSocketMessage>,
disconnect_token: CancellationToken,
) {
loop {
tokio::select! {
_ = disconnect_token.cancelled() => {
break;
}
message = writer_control_rx.recv() => {
let Some(message) = message else {
break;
};
if websocket_writer.send(message).await.is_err() {
break;
}
}
queued_message = writer_rx.recv() => {
let Some(queued_message) = queued_message else {
break;
};
let Some(json) = serialize_outgoing_message(queued_message.message) else {
continue;
};
if websocket_writer.send(WebSocketMessage::Text(json.into())).await.is_err() {
break;
}
if let Some(write_complete_tx) = queued_message.write_complete_tx {
let _ = write_complete_tx.send(());
}
}
}
}
}
async fn run_websocket_inbound_loop(
mut websocket_reader: futures::stream::SplitStream<WebSocket>,
transport_event_tx: mpsc::Sender<TransportEvent>,
writer_tx_for_reader: mpsc::Sender<QueuedOutgoingMessage>,
writer_control_tx: mpsc::Sender<WebSocketMessage>,
connection_id: ConnectionId,
disconnect_token: CancellationToken,
) {
loop {
tokio::select! {
_ = disconnect_token.cancelled() => {
break;
}
incoming_message = websocket_reader.next() => {
match incoming_message {
Some(Ok(WebSocketMessage::Text(text))) => {
if !forward_incoming_message(
&transport_event_tx,
&writer_tx_for_reader,
connection_id,
text.as_ref(),
)
.await
{
break;
}
}
Some(Ok(WebSocketMessage::Ping(payload))) => {
match writer_control_tx.try_send(WebSocketMessage::Pong(payload)) {
Ok(()) => {}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => break,
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
warn!("websocket control queue full while replying to ping; closing connection");
break;
}
}
}
Some(Ok(WebSocketMessage::Pong(_))) => {}
Some(Ok(WebSocketMessage::Close(_))) | None => break,
Some(Ok(WebSocketMessage::Binary(_))) => {
warn!("dropping unsupported binary websocket message");
}
Some(Err(err)) => {
warn!("websocket receive error: {err}");
break;
}
}
}
}
}
}