Add exec-server process RPC implementation

Restore the process RPC client/server implementation on top of the initialize-only base.\n\nCo-authored-by: Codex <noreply@openai.com>

codex-rs/Cargo.lock

@@ -1596,23 +1596,6 @@ dependencies = [
  "tokio-util",
 ]
 
-[[package]]
-name = "codex-auth"
-version = "0.0.0"
-dependencies = [
- "base64 0.22.1",
- "codex-api",
- "codex-app-server-protocol",
- "http 1.4.0",
- "maplit",
- "pretty_assertions",
- "schemars 0.8.22",
- "serde",
- "serde_json",
- "thiserror 2.0.18",
- "toml 0.9.11+spec-1.1.0",
-]
-
 [[package]]
 name = "codex-backend-client"
 version = "0.0.0"
@@ -1857,11 +1840,9 @@ dependencies = [
  "codex-arg0",
  "codex-artifacts",
  "codex-async-utils",
- "codex-auth",
  "codex-client",
  "codex-config",
  "codex-connectors",
- "codex-core-auth",
  "codex-environment",
  "codex-execpolicy",
  "codex-file-search",
@@ -1954,28 +1935,6 @@ dependencies = [
  "zstd",
 ]
 
-[[package]]
-name = "codex-core-auth"
-version = "0.0.0"
-dependencies = [
- "anyhow",
- "base64 0.22.1",
- "chrono",
- "codex-app-server-protocol",
- "codex-auth",
- "codex-keyring-store",
- "keyring",
- "once_cell",
- "pretty_assertions",
- "schemars 0.8.22",
- "serde",
- "serde_json",
- "sha2",
- "tempfile",
- "tokio",
- "tracing",
-]
-
 [[package]]
 name = "codex-debug-client"
 version = "0.0.0"
@@ -2044,6 +2003,26 @@ dependencies = [
  "wiremock",
 ]
 
+[[package]]
+name = "codex-exec-server"
+version = "0.0.0"
+dependencies = [
+ "anyhow",
+ "base64 0.22.1",
+ "clap",
+ "codex-app-server-protocol",
+ "codex-utils-cargo-bin",
+ "codex-utils-pty",
+ "futures",
+ "pretty_assertions",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-tungstenite",
+ "tracing",
+]
+
 [[package]]
 name = "codex-execpolicy"
 version = "0.0.0"

codex-rs/Cargo.toml

@@ -18,16 +18,15 @@ members = [
     "cli",
     "connectors",
     "config",
-    "codex-auth",
     "shell-command",
     "shell-escalation",
     "skills",
     "core",
-    "core/auth",
     "environment",
     "hooks",
     "secrets",
     "exec",
+    "exec-server",
     "execpolicy",
     "execpolicy-legacy",
     "keyring-store",
@@ -89,7 +88,6 @@ license = "Apache-2.0"
 app_test_support = { path = "app-server/tests/common" }
 codex-ansi-escape = { path = "ansi-escape" }
 codex-api = { path = "codex-api" }
-codex-auth = { path = "codex-auth" }
 codex-artifacts = { path = "artifacts" }
 codex-package-manager = { path = "package-manager" }
 codex-app-server = { path = "app-server" }
@@ -107,7 +105,6 @@ codex-cloud-requirements = { path = "cloud-requirements" }
 codex-connectors = { path = "connectors" }
 codex-config = { path = "config" }
 codex-core = { path = "core" }
-codex-core-auth = { path = "core/auth" }
 codex-environment = { path = "environment" }
 codex-exec = { path = "exec" }
 codex-execpolicy = { path = "execpolicy" }

codex-rs/app-server/tests/common/auth_fixtures.rs

@@ -7,11 +7,11 @@ use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use chrono::DateTime;
 use chrono::Utc;
 use codex_app_server_protocol::AuthMode;
-use codex_core::TokenData;
 use codex_core::auth::AuthCredentialsStoreMode;
 use codex_core::auth::AuthDotJson;
 use codex_core::auth::save_auth;
-use codex_core::parse_chatgpt_jwt_claims;
+use codex_core::token_data::TokenData;
+use codex_core::token_data::parse_chatgpt_jwt_claims;
 use serde_json::json;
 
 /// Builder for writing a fake ChatGPT auth.json in tests.

codex-rs/chatgpt/src/chatgpt_token.rs

@@ -1,10 +1,10 @@
 use codex_core::AuthManager;
-use codex_core::TokenData;
 use std::path::Path;
 use std::sync::LazyLock;
 use std::sync::RwLock;
 
 use codex_core::auth::AuthCredentialsStoreMode;
+use codex_core::token_data::TokenData;
 
 static CHATGPT_TOKEN: LazyLock<RwLock<Option<TokenData>>> = LazyLock::new(|| RwLock::new(None));
 

codex-rs/chatgpt/src/connectors.rs

@@ -1,6 +1,6 @@
 use codex_core::AuthManager;
-use codex_core::TokenData;
 use codex_core::config::Config;
+use codex_core::token_data::TokenData;
 use std::collections::HashSet;
 use std::time::Duration;
 

codex-rs/codex-auth/BUILD.bazel

@@ -1,6 +0,0 @@
-load("//:defs.bzl", "codex_rust_crate")
-
-codex_rust_crate(
-    name = "codex-auth",
-    crate_name = "codex_auth",
-)

codex-rs/codex-auth/Cargo.toml

@@ -1,23 +0,0 @@
-[package]
-name = "codex-auth"
-version.workspace = true
-edition.workspace = true
-license.workspace = true
-
-[lints]
-workspace = true
-
-[dependencies]
-base64 = { workspace = true }
-codex-api = { workspace = true }
-codex-app-server-protocol = { workspace = true }
-http = { workspace = true }
-schemars = { workspace = true }
-serde = { workspace = true, features = ["derive"] }
-serde_json = { workspace = true }
-thiserror = { workspace = true }
-
-[dev-dependencies]
-maplit = { workspace = true }
-pretty_assertions = { workspace = true }
-toml = { workspace = true }

codex-rs/codex-auth/src/error.rs

@@ -1,15 +0,0 @@
-#[derive(Debug)]
-pub struct EnvVarError {
-    pub var: String,
-    pub instructions: Option<String>,
-}
-
-impl std::fmt::Display for EnvVarError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "Missing environment variable: `{}`.", self.var)?;
-        if let Some(instructions) = &self.instructions {
-            write!(f, " {instructions}")?;
-        }
-        Ok(())
-    }
-}

codex-rs/codex-auth/src/lib.rs

@@ -1,19 +0,0 @@
-pub mod error;
-pub mod provider;
-pub mod token_data;
-
-#[cfg(test)]
-mod model_provider_info_tests;
-#[cfg(test)]
-mod token_data_tests;
-
-pub use error::EnvVarError;
-pub use provider::DEFAULT_LMSTUDIO_PORT;
-pub use provider::DEFAULT_OLLAMA_PORT;
-pub use provider::LMSTUDIO_OSS_PROVIDER_ID;
-pub use provider::ModelProviderInfo;
-pub use provider::OLLAMA_OSS_PROVIDER_ID;
-pub use provider::OPENAI_PROVIDER_ID;
-pub use provider::WireApi;
-pub use provider::built_in_model_providers;
-pub use provider::create_oss_provider_with_base_url;

codex-rs/codex-auth/src/provider.rs

@@ -1,291 +0,0 @@
-use crate::error::EnvVarError;
-use codex_api::Provider as ApiProvider;
-use codex_api::provider::RetryConfig as ApiRetryConfig;
-use codex_app_server_protocol::AuthMode as ApiAuthMode;
-use http::HeaderMap;
-use http::header::HeaderName;
-use http::header::HeaderValue;
-use schemars::JsonSchema;
-use serde::Deserialize;
-use serde::Serialize;
-use std::collections::HashMap;
-use std::fmt;
-use std::time::Duration;
-
-const DEFAULT_STREAM_IDLE_TIMEOUT_MS: u64 = 300_000;
-const DEFAULT_STREAM_MAX_RETRIES: u64 = 5;
-const DEFAULT_REQUEST_MAX_RETRIES: u64 = 4;
-pub const DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS: u64 = 15_000;
-const MAX_STREAM_MAX_RETRIES: u64 = 100;
-const MAX_REQUEST_MAX_RETRIES: u64 = 100;
-
-const OPENAI_PROVIDER_NAME: &str = "OpenAI";
-pub const OPENAI_PROVIDER_ID: &str = "openai";
-pub const CHAT_WIRE_API_REMOVED_ERROR: &str = "`wire_api = \"chat\"` is no longer supported.\nHow to fix: set `wire_api = \"responses\"` in your provider config.\nMore info: https://github.com/openai/codex/discussions/7782";
-pub const LEGACY_OLLAMA_CHAT_PROVIDER_ID: &str = "ollama-chat";
-pub const OLLAMA_CHAT_PROVIDER_REMOVED_ERROR: &str = "`ollama-chat` is no longer supported.\nHow to fix: replace `ollama-chat` with `ollama` in `model_provider`, `oss_provider`, or `--local-provider`.\nMore info: https://github.com/openai/codex/discussions/7782";
-
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, JsonSchema)]
-#[serde(rename_all = "lowercase")]
-#[schemars(rename = "WireApi")]
-pub enum WireApi {
-    #[default]
-    Responses,
-}
-
-impl fmt::Display for WireApi {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let value = match self {
-            Self::Responses => "responses",
-        };
-        f.write_str(value)
-    }
-}
-
-impl<'de> Deserialize<'de> for WireApi {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        let value = String::deserialize(deserializer)?;
-        match value.as_str() {
-            "responses" => Ok(Self::Responses),
-            "chat" => Err(serde::de::Error::custom(CHAT_WIRE_API_REMOVED_ERROR)),
-            _ => Err(serde::de::Error::unknown_variant(&value, &["responses"])),
-        }
-    }
-}
-
-#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema)]
-#[schemars(deny_unknown_fields)]
-#[schemars(rename = "ModelProviderInfo")]
-pub struct ModelProviderInfo {
-    pub name: String,
-    pub base_url: Option<String>,
-    pub env_key: Option<String>,
-    pub env_key_instructions: Option<String>,
-    pub experimental_bearer_token: Option<String>,
-    #[serde(default)]
-    pub wire_api: WireApi,
-    pub query_params: Option<HashMap<String, String>>,
-    pub http_headers: Option<HashMap<String, String>>,
-    pub env_http_headers: Option<HashMap<String, String>>,
-    pub request_max_retries: Option<u64>,
-    pub stream_max_retries: Option<u64>,
-    pub stream_idle_timeout_ms: Option<u64>,
-    #[schemars(
-        description = "Maximum time (in milliseconds) to wait for a websocket connection attempt before treating it as failed."
-    )]
-    pub websocket_connect_timeout_ms: Option<u64>,
-    #[serde(default)]
-    pub requires_openai_auth: bool,
-    #[serde(default)]
-    pub supports_websockets: bool,
-}
-
-impl ModelProviderInfo {
-    fn build_header_map(&self) -> HeaderMap {
-        let capacity = self.http_headers.as_ref().map_or(0, HashMap::len)
-            + self.env_http_headers.as_ref().map_or(0, HashMap::len);
-        let mut headers = HeaderMap::with_capacity(capacity);
-        if let Some(extra) = &self.http_headers {
-            for (k, v) in extra {
-                if let (Ok(name), Ok(value)) = (HeaderName::try_from(k), HeaderValue::try_from(v)) {
-                    headers.insert(name, value);
-                }
-            }
-        }
-
-        if let Some(env_headers) = &self.env_http_headers {
-            for (header, env_var) in env_headers {
-                if let Ok(val) = std::env::var(env_var)
-                    && !val.trim().is_empty()
-                    && let (Ok(name), Ok(value)) =
-                        (HeaderName::try_from(header), HeaderValue::try_from(val))
-                {
-                    headers.insert(name, value);
-                }
-            }
-        }
-
-        headers
-    }
-
-    pub fn to_api_provider(
-        &self,
-        auth_mode: Option<ApiAuthMode>,
-    ) -> Result<ApiProvider, EnvVarError> {
-        let default_base_url = if matches!(
-            auth_mode,
-            Some(ApiAuthMode::Chatgpt | ApiAuthMode::ChatgptAuthTokens)
-        ) {
-            "https://chatgpt.com/backend-api/codex"
-        } else {
-            "https://api.openai.com/v1"
-        };
-        let base_url = self
-            .base_url
-            .clone()
-            .unwrap_or_else(|| default_base_url.to_string());
-
-        let retry = ApiRetryConfig {
-            max_attempts: self.request_max_retries(),
-            base_delay: Duration::from_millis(200),
-            retry_429: false,
-            retry_5xx: true,
-            retry_transport: true,
-        };
-
-        Ok(ApiProvider {
-            name: self.name.clone(),
-            base_url,
-            query_params: self.query_params.clone(),
-            headers: self.build_header_map(),
-            retry,
-            stream_idle_timeout: self.stream_idle_timeout(),
-        })
-    }
-
-    pub fn api_key(&self) -> Result<Option<String>, EnvVarError> {
-        match &self.env_key {
-            Some(env_key) => {
-                let api_key = std::env::var(env_key)
-                    .ok()
-                    .filter(|v| !v.trim().is_empty())
-                    .ok_or_else(|| EnvVarError {
-                        var: env_key.clone(),
-                        instructions: self.env_key_instructions.clone(),
-                    })?;
-                Ok(Some(api_key))
-            }
-            None => Ok(None),
-        }
-    }
-
-    pub fn request_max_retries(&self) -> u64 {
-        self.request_max_retries
-            .unwrap_or(DEFAULT_REQUEST_MAX_RETRIES)
-            .min(MAX_REQUEST_MAX_RETRIES)
-    }
-
-    pub fn stream_max_retries(&self) -> u64 {
-        self.stream_max_retries
-            .unwrap_or(DEFAULT_STREAM_MAX_RETRIES)
-            .min(MAX_STREAM_MAX_RETRIES)
-    }
-
-    pub fn stream_idle_timeout(&self) -> Duration {
-        self.stream_idle_timeout_ms
-            .map(Duration::from_millis)
-            .unwrap_or(Duration::from_millis(DEFAULT_STREAM_IDLE_TIMEOUT_MS))
-    }
-
-    pub fn websocket_connect_timeout(&self) -> Duration {
-        self.websocket_connect_timeout_ms
-            .map(Duration::from_millis)
-            .unwrap_or(Duration::from_millis(DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS))
-    }
-
-    pub fn create_openai_provider(base_url: Option<String>) -> ModelProviderInfo {
-        ModelProviderInfo {
-            name: OPENAI_PROVIDER_NAME.into(),
-            base_url,
-            env_key: None,
-            env_key_instructions: None,
-            experimental_bearer_token: None,
-            wire_api: WireApi::Responses,
-            query_params: None,
-            http_headers: Some(
-                [("version".to_string(), env!("CARGO_PKG_VERSION").to_string())]
-                    .into_iter()
-                    .collect(),
-            ),
-            env_http_headers: Some(
-                [
-                    (
-                        "OpenAI-Organization".to_string(),
-                        "OPENAI_ORGANIZATION".to_string(),
-                    ),
-                    ("OpenAI-Project".to_string(), "OPENAI_PROJECT".to_string()),
-                ]
-                .into_iter()
-                .collect(),
-            ),
-            request_max_retries: None,
-            stream_max_retries: None,
-            stream_idle_timeout_ms: None,
-            websocket_connect_timeout_ms: None,
-            requires_openai_auth: true,
-            supports_websockets: true,
-        }
-    }
-
-    pub fn is_openai(&self) -> bool {
-        self.name == OPENAI_PROVIDER_NAME
-    }
-}
-
-pub const DEFAULT_LMSTUDIO_PORT: u16 = 1234;
-pub const DEFAULT_OLLAMA_PORT: u16 = 11434;
-
-pub const LMSTUDIO_OSS_PROVIDER_ID: &str = "lmstudio";
-pub const OLLAMA_OSS_PROVIDER_ID: &str = "ollama";
-
-pub fn built_in_model_providers(
-    openai_base_url: Option<String>,
-) -> HashMap<String, ModelProviderInfo> {
-    use ModelProviderInfo as P;
-    let openai_provider = P::create_openai_provider(openai_base_url);
-
-    [
-        (OPENAI_PROVIDER_ID, openai_provider),
-        (
-            OLLAMA_OSS_PROVIDER_ID,
-            create_oss_provider(DEFAULT_OLLAMA_PORT, WireApi::Responses),
-        ),
-        (
-            LMSTUDIO_OSS_PROVIDER_ID,
-            create_oss_provider(DEFAULT_LMSTUDIO_PORT, WireApi::Responses),
-        ),
-    ]
-    .into_iter()
-    .map(|(k, v)| (k.to_string(), v))
-    .collect()
-}
-
-pub fn create_oss_provider(default_provider_port: u16, wire_api: WireApi) -> ModelProviderInfo {
-    let default_codex_oss_base_url = format!(
-        "http://localhost:{codex_oss_port}/v1",
-        codex_oss_port = std::env::var("CODEX_OSS_PORT")
-            .ok()
-            .filter(|value| !value.trim().is_empty())
-            .and_then(|value| value.parse::<u16>().ok())
-            .unwrap_or(default_provider_port)
-    );
-
-    let codex_oss_base_url = std::env::var("CODEX_OSS_BASE_URL")
-        .ok()
-        .filter(|v| !v.trim().is_empty())
-        .unwrap_or(default_codex_oss_base_url);
-    create_oss_provider_with_base_url(&codex_oss_base_url, wire_api)
-}
-
-pub fn create_oss_provider_with_base_url(base_url: &str, wire_api: WireApi) -> ModelProviderInfo {
-    ModelProviderInfo {
-        name: "gpt-oss".into(),
-        base_url: Some(base_url.into()),
-        env_key: None,
-        env_key_instructions: None,
-        experimental_bearer_token: None,
-        wire_api,
-        query_params: None,
-        http_headers: None,
-        env_http_headers: None,
-        request_max_retries: None,
-        stream_max_retries: None,
-        stream_idle_timeout_ms: None,
-        websocket_connect_timeout_ms: None,
-        requires_openai_auth: false,
-        supports_websockets: false,
-    }
-}

codex-rs/core/Cargo.toml

@@ -28,14 +28,12 @@ chardetng = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 clap = { workspace = true, features = ["derive"] }
 codex-api = { workspace = true }
-codex-auth = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-apply-patch = { workspace = true }
 codex-async-utils = { workspace = true }
 codex-client = { workspace = true }
 codex-connectors = { workspace = true }
 codex-config = { workspace = true }
-codex-core-auth = { workspace = true }
 codex-environment = { workspace = true }
 codex-shell-command = { workspace = true }
 codex-skills = { workspace = true }

codex-rs/core/auth/Cargo.toml

@@ -1,29 +0,0 @@
-[package]
-name = "codex-core-auth"
-version.workspace = true
-edition.workspace = true
-license.workspace = true
-
-[lints]
-workspace = true
-
-[dependencies]
-chrono = { workspace = true, features = ["serde"] }
-codex-app-server-protocol = { workspace = true }
-codex-auth = { workspace = true }
-codex-keyring-store = { workspace = true }
-once_cell = { workspace = true }
-schemars = { workspace = true }
-serde = { workspace = true, features = ["derive"] }
-serde_json = { workspace = true }
-sha2 = { workspace = true }
-tracing = { workspace = true }
-
-[dev-dependencies]
-anyhow = { workspace = true }
-base64 = { workspace = true }
-keyring = { workspace = true }
-pretty_assertions = { workspace = true }
-serde = { workspace = true, features = ["derive"] }
-tempfile = { workspace = true }
-tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

codex-rs/core/auth/src/lib.rs

@@ -1,14 +0,0 @@
-//! Auth storage backend for Codex CLI credentials.
-//!
-//! This crate provides the storage layer for auth.json (file, keyring, auto, ephemeral)
-//! and the AuthDotJson / AuthCredentialsStoreMode types. The higher-level auth logic
-//! (CodexAuth, AuthManager, token refresh) lives in codex-core.
-
-pub mod storage;
-
-pub use storage::AuthCredentialsStoreMode;
-pub use storage::AuthDotJson;
-pub use storage::AuthStorageBackend;
-pub use storage::FileAuthStorage;
-pub use storage::create_auth_storage;
-pub use storage::get_auth_file;

codex-rs/core/src/api_bridge.rs

@@ -10,13 +10,13 @@ use http::HeaderMap;
 use serde::Deserialize;
 use serde_json::Value;
 
-use crate::PlanType;
 use crate::auth::CodexAuth;
 use crate::error::CodexErr;
 use crate::error::RetryLimitReachedError;
 use crate::error::UnexpectedResponseError;
 use crate::error::UsageLimitReachedError;
 use crate::model_provider_info::ModelProviderInfo;
+use crate::token_data::PlanType;
 
 pub(crate) fn map_api_error(err: ApiError) -> CodexErr {
     match err {

codex-rs/core/src/auth.rs

@@ -1,3 +1,5 @@
+mod storage;
+
 use async_trait::async_trait;
 use chrono::Utc;
 use reqwest::StatusCode;
@@ -17,19 +19,17 @@ use codex_app_server_protocol::AuthMode as ApiAuthMode;
 use codex_otel::TelemetryAuthMode;
 use codex_protocol::config_types::ForcedLoginMethod;
 
-use codex_core_auth::AuthStorageBackend;
-use codex_core_auth::create_auth_storage;
-
-pub use codex_core_auth::AuthCredentialsStoreMode;
-pub use codex_core_auth::AuthDotJson;
-
-use crate::KnownPlan as InternalKnownPlan;
-use crate::PlanType as InternalPlanType;
-use crate::TokenData;
+pub use crate::auth::storage::AuthCredentialsStoreMode;
+pub use crate::auth::storage::AuthDotJson;
+use crate::auth::storage::AuthStorageBackend;
+use crate::auth::storage::create_auth_storage;
 use crate::config::Config;
 use crate::error::RefreshTokenFailedError;
 use crate::error::RefreshTokenFailedReason;
-use crate::parse_chatgpt_jwt_claims;
+use crate::token_data::KnownPlan as InternalKnownPlan;
+use crate::token_data::PlanType as InternalPlanType;
+use crate::token_data::TokenData;
+use crate::token_data::parse_chatgpt_jwt_claims;
 use crate::util::try_parse_error_message;
 use codex_client::CodexHttpClient;
 use codex_protocol::account::PlanType as AccountPlanType;
@@ -752,6 +752,67 @@ fn refresh_token_endpoint() -> String {
         .unwrap_or_else(|_| REFRESH_TOKEN_URL.to_string())
 }
 
+impl AuthDotJson {
+    fn from_external_tokens(external: &ExternalAuthTokens) -> std::io::Result<Self> {
+        let mut token_info =
+            parse_chatgpt_jwt_claims(&external.access_token).map_err(std::io::Error::other)?;
+        token_info.chatgpt_account_id = Some(external.chatgpt_account_id.clone());
+        token_info.chatgpt_plan_type = external
+            .chatgpt_plan_type
+            .as_deref()
+            .map(InternalPlanType::from_raw_value)
+            .or(token_info.chatgpt_plan_type)
+            .or(Some(InternalPlanType::Unknown("unknown".to_string())));
+        let tokens = TokenData {
+            id_token: token_info,
+            access_token: external.access_token.clone(),
+            refresh_token: String::new(),
+            account_id: Some(external.chatgpt_account_id.clone()),
+        };
+
+        Ok(Self {
+            auth_mode: Some(ApiAuthMode::ChatgptAuthTokens),
+            openai_api_key: None,
+            tokens: Some(tokens),
+            last_refresh: Some(Utc::now()),
+        })
+    }
+
+    fn from_external_access_token(
+        access_token: &str,
+        chatgpt_account_id: &str,
+        chatgpt_plan_type: Option<&str>,
+    ) -> std::io::Result<Self> {
+        let external = ExternalAuthTokens {
+            access_token: access_token.to_string(),
+            chatgpt_account_id: chatgpt_account_id.to_string(),
+            chatgpt_plan_type: chatgpt_plan_type.map(str::to_string),
+        };
+        Self::from_external_tokens(&external)
+    }
+
+    fn resolved_mode(&self) -> ApiAuthMode {
+        if let Some(mode) = self.auth_mode {
+            return mode;
+        }
+        if self.openai_api_key.is_some() {
+            return ApiAuthMode::ApiKey;
+        }
+        ApiAuthMode::Chatgpt
+    }
+
+    fn storage_mode(
+        &self,
+        auth_credentials_store_mode: AuthCredentialsStoreMode,
+    ) -> AuthCredentialsStoreMode {
+        if self.resolved_mode() == ApiAuthMode::ChatgptAuthTokens {
+            AuthCredentialsStoreMode::Ephemeral
+        } else {
+            auth_credentials_store_mode
+        }
+    }
+}
+
 /// Internal cached auth state.
 #[derive(Clone)]
 struct CachedAuth {
@@ -1351,12 +1412,8 @@ impl AuthManager {
                 ),
             )));
         }
-        let auth_dot_json = AuthDotJson::from_external_access_token(
-            &refreshed.access_token,
-            &refreshed.chatgpt_account_id,
-            refreshed.chatgpt_plan_type.as_deref(),
-        )
-        .map_err(RefreshTokenError::Transient)?;
+        let auth_dot_json =
+            AuthDotJson::from_external_tokens(&refreshed).map_err(RefreshTokenError::Transient)?;
         save_auth(
             &self.codex_home,
             &auth_dot_json,

codex-rs/core/src/auth/storage.rs

@@ -19,10 +19,8 @@ use std::sync::Arc;
 use std::sync::Mutex;
 use tracing::warn;
 
+use crate::token_data::TokenData;
 use codex_app_server_protocol::AuthMode;
-use codex_auth::token_data::PlanType;
-use codex_auth::token_data::TokenData;
-use codex_auth::token_data::parse_chatgpt_jwt_claims;
 use codex_keyring_store::DefaultKeyringStore;
 use codex_keyring_store::KeyringStore;
 use once_cell::sync::Lazy;
@@ -58,57 +56,7 @@ pub struct AuthDotJson {
     pub last_refresh: Option<DateTime<Utc>>,
 }
 
-impl AuthDotJson {
-    pub fn from_external_access_token(
-        access_token: &str,
-        chatgpt_account_id: &str,
-        chatgpt_plan_type: Option<&str>,
-    ) -> std::io::Result<Self> {
-        let mut token_info =
-            parse_chatgpt_jwt_claims(access_token).map_err(std::io::Error::other)?;
-        token_info.chatgpt_account_id = Some(chatgpt_account_id.to_string());
-        token_info.chatgpt_plan_type = chatgpt_plan_type
-            .map(PlanType::from_raw_value)
-            .or(token_info.chatgpt_plan_type)
-            .or(Some(PlanType::Unknown("unknown".to_string())));
-        let tokens = TokenData {
-            id_token: token_info,
-            access_token: access_token.to_string(),
-            refresh_token: String::new(),
-            account_id: Some(chatgpt_account_id.to_string()),
-        };
-
-        Ok(Self {
-            auth_mode: Some(AuthMode::ChatgptAuthTokens),
-            openai_api_key: None,
-            tokens: Some(tokens),
-            last_refresh: Some(Utc::now()),
-        })
-    }
-
-    pub fn resolved_mode(&self) -> AuthMode {
-        if let Some(mode) = self.auth_mode {
-            return mode;
-        }
-        if self.openai_api_key.is_some() {
-            return AuthMode::ApiKey;
-        }
-        AuthMode::Chatgpt
-    }
-
-    pub fn storage_mode(
-        &self,
-        auth_credentials_store_mode: AuthCredentialsStoreMode,
-    ) -> AuthCredentialsStoreMode {
-        if self.resolved_mode() == AuthMode::ChatgptAuthTokens {
-            AuthCredentialsStoreMode::Ephemeral
-        } else {
-            auth_credentials_store_mode
-        }
-    }
-}
-
-pub fn get_auth_file(codex_home: &Path) -> PathBuf {
+pub(super) fn get_auth_file(codex_home: &Path) -> PathBuf {
     codex_home.join("auth.json")
 }
 
@@ -121,25 +69,25 @@ pub(super) fn delete_file_if_exists(codex_home: &Path) -> std::io::Result<bool>
     }
 }
 
-pub trait AuthStorageBackend: Debug + Send + Sync {
+pub(super) trait AuthStorageBackend: Debug + Send + Sync {
     fn load(&self) -> std::io::Result<Option<AuthDotJson>>;
     fn save(&self, auth: &AuthDotJson) -> std::io::Result<()>;
     fn delete(&self) -> std::io::Result<bool>;
 }
 
 #[derive(Clone, Debug)]
-pub struct FileAuthStorage {
+pub(super) struct FileAuthStorage {
     codex_home: PathBuf,
 }
 
 impl FileAuthStorage {
-    pub fn new(codex_home: PathBuf) -> Self {
+    pub(super) fn new(codex_home: PathBuf) -> Self {
         Self { codex_home }
     }
 
     /// Attempt to read and parse the `auth.json` file in the given `CODEX_HOME` directory.
     /// Returns the full AuthDotJson structure.
-    pub fn try_read_auth_json(&self, auth_file: &Path) -> std::io::Result<AuthDotJson> {
+    pub(super) fn try_read_auth_json(&self, auth_file: &Path) -> std::io::Result<AuthDotJson> {
         let mut file = File::open(auth_file)?;
         let mut contents = String::new();
         file.read_to_string(&mut contents)?;
@@ -187,7 +135,7 @@ impl AuthStorageBackend for FileAuthStorage {
 const KEYRING_SERVICE: &str = "Codex Auth";
 
 // turns codex_home path into a stable, short key string
-pub(crate) fn compute_store_key(codex_home: &Path) -> std::io::Result<String> {
+fn compute_store_key(codex_home: &Path) -> std::io::Result<String> {
     let canonical = codex_home
         .canonicalize()
         .unwrap_or_else(|_| codex_home.to_path_buf());
@@ -201,13 +149,13 @@ pub(crate) fn compute_store_key(codex_home: &Path) -> std::io::Result<String> {
 }
 
 #[derive(Clone, Debug)]
-pub(crate) struct KeyringAuthStorage {
-    pub(crate) codex_home: PathBuf,
-    pub(crate) keyring_store: Arc<dyn KeyringStore>,
+struct KeyringAuthStorage {
+    codex_home: PathBuf,
+    keyring_store: Arc<dyn KeyringStore>,
 }
 
 impl KeyringAuthStorage {
-    pub(crate) fn new(codex_home: PathBuf, keyring_store: Arc<dyn KeyringStore>) -> Self {
+    fn new(codex_home: PathBuf, keyring_store: Arc<dyn KeyringStore>) -> Self {
         Self {
             codex_home,
             keyring_store,
@@ -275,13 +223,13 @@ impl AuthStorageBackend for KeyringAuthStorage {
 }
 
 #[derive(Clone, Debug)]
-pub(crate) struct AutoAuthStorage {
-    pub(crate) keyring_storage: Arc<KeyringAuthStorage>,
-    pub(crate) file_storage: Arc<FileAuthStorage>,
+struct AutoAuthStorage {
+    keyring_storage: Arc<KeyringAuthStorage>,
+    file_storage: Arc<FileAuthStorage>,
 }
 
 impl AutoAuthStorage {
-    pub(crate) fn new(codex_home: PathBuf, keyring_store: Arc<dyn KeyringStore>) -> Self {
+    fn new(codex_home: PathBuf, keyring_store: Arc<dyn KeyringStore>) -> Self {
         Self {
             keyring_storage: Arc::new(KeyringAuthStorage::new(codex_home.clone(), keyring_store)),
             file_storage: Arc::new(FileAuthStorage::new(codex_home)),
@@ -360,7 +308,7 @@ impl AuthStorageBackend for EphemeralAuthStorage {
     }
 }
 
-pub fn create_auth_storage(
+pub(super) fn create_auth_storage(
     codex_home: PathBuf,
     mode: AuthCredentialsStoreMode,
 ) -> Arc<dyn AuthStorageBackend> {

codex-rs/core/src/auth/storage_tests.rs

@@ -1,16 +1,14 @@
 use super::*;
+use crate::token_data::IdTokenInfo;
 use anyhow::Context;
 use base64::Engine;
-use codex_auth::token_data::IdTokenInfo;
-use codex_auth::token_data::TokenData;
-use codex_auth::token_data::parse_chatgpt_jwt_claims;
-use codex_keyring_store::tests::MockKeyringStore;
-use keyring::Error as KeyringError;
 use pretty_assertions::assert_eq;
-use serde::Serialize;
 use serde_json::json;
 use tempfile::tempdir;
 
+use codex_keyring_store::tests::MockKeyringStore;
+use keyring::Error as KeyringError;
+
 #[tokio::test]
 async fn file_storage_load_returns_auth_dot_json() -> anyhow::Result<()> {
     let codex_home = tempdir()?;
@@ -169,7 +167,7 @@ fn id_token_with_prefix(prefix: &str) -> IdTokenInfo {
     let signature_b64 = encode(b"sig");
     let fake_jwt = format!("{header_b64}.{payload_b64}.{signature_b64}");
 
-    parse_chatgpt_jwt_claims(&fake_jwt).expect("fake JWT should parse")
+    crate::token_data::parse_chatgpt_jwt_claims(&fake_jwt).expect("fake JWT should parse")
 }
 
 fn auth_with_prefix(prefix: &str) -> AuthDotJson {

codex-rs/core/src/auth_tests.rs

@@ -1,12 +1,11 @@
 use super::*;
-use crate::IdTokenInfo;
-use crate::KnownPlan as InternalKnownPlan;
-use crate::PlanType as InternalPlanType;
-use crate::TokenData;
+use crate::auth::storage::FileAuthStorage;
+use crate::auth::storage::get_auth_file;
 use crate::config::Config;
 use crate::config::ConfigBuilder;
-use codex_core_auth::FileAuthStorage;
-use codex_core_auth::get_auth_file;
+use crate::token_data::IdTokenInfo;
+use crate::token_data::KnownPlan as InternalKnownPlan;
+use crate::token_data::PlanType as InternalPlanType;
 use codex_protocol::account::PlanType as AccountPlanType;
 
 use base64::Engine;

codex-rs/core/src/client.rs

@@ -528,7 +528,7 @@ impl ModelClient {
         let api_provider = self
             .state
             .provider
-            .to_api_provider(auth.as_ref().map(CodexAuth::api_auth_mode))?;
+            .to_api_provider(auth.as_ref().map(CodexAuth::auth_mode))?;
         let api_auth = auth_provider_from_auth(auth.clone(), &self.state.provider)?;
         Ok(CurrentClientSetup {
             auth,

codex-rs/core/src/connectors.rs

@@ -25,7 +25,6 @@ use tracing::warn;
 use crate::AuthManager;
 use crate::CodexAuth;
 use crate::SandboxState;
-use crate::TokenData;
 use crate::config::Config;
 use crate::config::types::AppToolApproval;
 use crate::config::types::AppsConfigToml;
@@ -45,6 +44,7 @@ use crate::mcp_connection_manager::codex_apps_tools_cache_key;
 use crate::plugins::AppConnectorId;
 use crate::plugins::PluginsManager;
 use crate::plugins::list_tool_suggest_discoverable_plugins;
+use crate::token_data::TokenData;
 use crate::tools::discoverable::DiscoverablePluginInfo;
 use crate::tools::discoverable::DiscoverableTool;
 

codex-rs/core/src/error.rs

@@ -1,7 +1,7 @@
-use crate::KnownPlan;
-use crate::PlanType;
 use crate::exec::ExecToolCallOutput;
 use crate::network_policy_decision::NetworkPolicyDecisionPayload;
+use crate::token_data::KnownPlan;
+use crate::token_data::PlanType;
 use crate::truncate::TruncationPolicy;
 use crate::truncate::truncate_text;
 use chrono::DateTime;
@@ -9,7 +9,6 @@ use chrono::Datelike;
 use chrono::Local;
 use chrono::Utc;
 use codex_async_utils::CancelErr;
-pub use codex_auth::EnvVarError;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::CodexErrorInfo;
 use codex_protocol::protocol::ErrorEvent;
@@ -192,12 +191,6 @@ impl From<CancelErr> for CodexErr {
     }
 }
 
-impl From<EnvVarError> for CodexErr {
-    fn from(error: EnvVarError) -> Self {
-        Self::EnvVar(error)
-    }
-}
-
 impl CodexErr {
     pub fn is_retryable(&self) -> bool {
         match self {
@@ -558,6 +551,26 @@ fn now_for_retry() -> DateTime<Utc> {
     Utc::now()
 }
 
+#[derive(Debug)]
+pub struct EnvVarError {
+    /// Name of the environment variable that is missing.
+    pub var: String,
+
+    /// Optional instructions to help the user get a valid value for the
+    /// variable and set it.
+    pub instructions: Option<String>,
+}
+
+impl std::fmt::Display for EnvVarError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Missing environment variable: `{}`.", self.var)?;
+        if let Some(instructions) = &self.instructions {
+            write!(f, " {instructions}")?;
+        }
+        Ok(())
+    }
+}
+
 impl CodexErr {
     /// Minimal shim so that existing `e.downcast_ref::<CodexErr>()` checks continue to compile
     /// after replacing `anyhow::Error` in the return signature. This mirrors the behavior of

codex-rs/core/src/lib.rs

@@ -63,7 +63,7 @@ mod mcp_tool_call;
 mod memories;
 pub mod mention_syntax;
 mod mentions;
-pub mod message_history;
+mod message_history;
 mod model_provider_info;
 pub mod path_utils;
 pub mod personality_migration;
@@ -76,16 +76,11 @@ mod shell_detect;
 mod stream_events_utils;
 pub mod test_support;
 mod text_encoding;
+pub mod token_data;
 mod truncate;
 mod unified_exec;
 pub mod windows_sandbox;
 pub use client::X_RESPONSESAPI_INCLUDE_TIMING_METRICS_HEADER;
-pub use codex_auth::token_data::IdTokenInfo;
-pub use codex_auth::token_data::IdTokenInfoError;
-pub use codex_auth::token_data::KnownPlan;
-pub use codex_auth::token_data::PlanType;
-pub use codex_auth::token_data::TokenData;
-pub use codex_auth::token_data::parse_chatgpt_jwt_claims;
 pub use model_provider_info::DEFAULT_LMSTUDIO_PORT;
 pub use model_provider_info::DEFAULT_OLLAMA_PORT;
 pub use model_provider_info::LMSTUDIO_OSS_PROVIDER_ID;

codex-rs/core/src/message_history.rs

@@ -66,22 +66,14 @@ fn history_filepath(config: &Config) -> PathBuf {
     path
 }
 
-/// Append a `text` entry associated with `conversation_id` to the history file.
-///
-/// Uses advisory file locking (`File::try_lock`) with a retry loop to ensure
-/// concurrent writes from multiple TUI processes do not interleave. The lock
-/// acquisition and write are performed inside `spawn_blocking` so the caller's
-/// async runtime is not blocked.
-///
-/// The entry is silently skipped when `config.history.persistence` is
-/// [`HistoryPersistence::None`].
-///
-/// # Errors
-///
-/// Returns an I/O error if the history file cannot be opened/created, the
-/// system clock is before the Unix epoch, or the exclusive lock cannot be
-/// acquired after [`MAX_RETRIES`] attempts.
-pub async fn append_entry(text: &str, conversation_id: &ThreadId, config: &Config) -> Result<()> {
+/// Append a `text` entry associated with `conversation_id` to the history file. Uses
+/// advisory file locking to ensure that concurrent writes do not interleave,
+/// which entails a small amount of blocking I/O internally.
+pub(crate) async fn append_entry(
+    text: &str,
+    conversation_id: &ThreadId,
+    config: &Config,
+) -> Result<()> {
     match config.history.persistence {
         HistoryPersistence::SaveAll => {
             // Save everything: proceed.
@@ -251,29 +243,22 @@ fn trim_target_bytes(max_bytes: u64, newest_entry_len: u64) -> u64 {
     soft_cap_bytes.max(newest_entry_len)
 }
 
-/// Asynchronously fetch the history file's *identifier* and current entry count.
-///
-/// The identifier is the file's inode on Unix or creation time on Windows.
-/// The entry count is derived by counting newline bytes in the file. Returns
-/// `(0, 0)` when the file does not exist or its metadata cannot be read. If
-/// metadata succeeds but the file cannot be opened or scanned, returns
-/// `(log_id, 0)` so callers can still detect that a history file exists.
-pub async fn history_metadata(config: &Config) -> (u64, usize) {
+/// Asynchronously fetch the history file's *identifier* (inode on Unix) and
+/// the current number of entries by counting newline characters.
+pub(crate) async fn history_metadata(config: &Config) -> (u64, usize) {
     let path = history_filepath(config);
     history_metadata_for_file(&path).await
 }
 
-/// Look up a single history entry by file identity and zero-based offset.
+/// Given a `log_id` (on Unix this is the file's inode number,
+/// on Windows this is the file's creation time) and a zero-based
+/// `offset`, return the corresponding `HistoryEntry` if the identifier matches
+/// the current history file **and** the requested offset exists. Any I/O or
+/// parsing errors are logged and result in `None`.
 ///
-/// Returns `Some(entry)` when the current history file's identifier (inode on
-/// Unix, creation time on Windows) matches `log_id` **and** a valid JSON
-/// record exists at `offset`. Returns `None` on any mismatch, I/O error, or
-/// parse failure, all of which are logged at `warn` level.
-///
-/// This function is synchronous because it acquires a shared advisory file lock
-/// via `File::try_lock_shared`. Callers on an async runtime should wrap it in
-/// `spawn_blocking`.
-pub fn lookup(log_id: u64, offset: usize, config: &Config) -> Option<HistoryEntry> {
+/// Note this function is not async because it uses a sync advisory file
+/// locking API.
+pub(crate) fn lookup(log_id: u64, offset: usize, config: &Config) -> Option<HistoryEntry> {
     let path = history_filepath(config);
     lookup_history_entry(&path, log_id, offset)
 }

codex-rs/core/src/model_provider_info.rs

@@ -5,10 +5,10 @@
 //!   2. User-defined entries inside `~/.codex/config.toml` under the `model_providers`
 //!      key. These override or extend the defaults at runtime.
 
+use crate::auth::AuthMode;
 use crate::error::EnvVarError;
 use codex_api::Provider as ApiProvider;
 use codex_api::provider::RetryConfig as ApiRetryConfig;
-use codex_app_server_protocol::AuthMode as ApiAuthMode;
 use http::HeaderMap;
 use http::header::HeaderName;
 use http::header::HeaderValue;
@@ -159,12 +159,9 @@ impl ModelProviderInfo {
 
     pub(crate) fn to_api_provider(
         &self,
-        auth_mode: Option<ApiAuthMode>,
+        auth_mode: Option<AuthMode>,
     ) -> crate::error::Result<ApiProvider> {
-        let default_base_url = if matches!(
-            auth_mode,
-            Some(ApiAuthMode::Chatgpt | ApiAuthMode::ChatgptAuthTokens)
-        ) {
+        let default_base_url = if matches!(auth_mode, Some(AuthMode::Chatgpt)) {
             "https://chatgpt.com/backend-api/codex"
         } else {
             "https://api.openai.com/v1"
@@ -267,6 +264,7 @@ impl ModelProviderInfo {
                 .into_iter()
                 .collect(),
             ),
+            // Use global defaults for retry/timeout unless overridden in config.toml.
             request_max_retries: None,
             stream_max_retries: None,
             stream_idle_timeout_ms: None,
@@ -294,6 +292,10 @@ pub fn built_in_model_providers(
     use ModelProviderInfo as P;
     let openai_provider = P::create_openai_provider(openai_base_url);
 
+    // We do not want to be in the business of adjucating which third-party
+    // providers are bundled with Codex CLI, so we only include the OpenAI and
+    // open source ("oss") providers by default. Users are encouraged to add to
+    // `model_providers` in config.toml to add their own providers.
     [
         (OPENAI_PROVIDER_ID, openai_provider),
         (
@@ -311,6 +313,8 @@ pub fn built_in_model_providers(
 }
 
 pub fn create_oss_provider(default_provider_port: u16, wire_api: WireApi) -> ModelProviderInfo {
+    // These CODEX_OSS_ environment variables are experimental: we may
+    // switch to reading values from config.toml instead.
     let default_codex_oss_base_url = format!(
         "http://localhost:{codex_oss_port}/v1",
         codex_oss_port = std::env::var("CODEX_OSS_PORT")
@@ -346,3 +350,7 @@ pub fn create_oss_provider_with_base_url(base_url: &str, wire_api: WireApi) -> M
         supports_websockets: false,
     }
 }
+
+#[cfg(test)]
+#[path = "model_provider_info_tests.rs"]
+mod tests;

codex-rs/core/src/model_provider_info_tests.rs

@@ -1,4 +1,4 @@
-use super::provider::*;
+use super::*;
 use pretty_assertions::assert_eq;
 
 #[test]

codex-rs/core/src/models_manager/manager.rs

@@ -433,9 +433,7 @@ impl ModelsManager {
             codex_otel::start_global_timer("codex.remote_models.fetch_update.duration_ms", &[]);
         let auth = self.auth_manager.auth().await;
         let auth_mode = auth.as_ref().map(CodexAuth::auth_mode);
-        let api_provider = self
-            .provider
-            .to_api_provider(auth.as_ref().map(CodexAuth::api_auth_mode))?;
+        let api_provider = self.provider.to_api_provider(auth_mode)?;
         let api_auth = auth_provider_from_auth(auth.clone(), &self.provider)?;
         let auth_env = collect_auth_env_telemetry(
             &self.provider,

codex-rs/core/src/realtime_conversation.rs

@@ -22,7 +22,6 @@ use codex_api::RealtimeSessionMode;
 use codex_api::RealtimeWebsocketClient;
 use codex_api::endpoint::realtime_websocket::RealtimeWebsocketEvents;
 use codex_api::endpoint::realtime_websocket::RealtimeWebsocketWriter;
-use codex_app_server_protocol::AuthMode as ApiAuthMode;
 use codex_protocol::protocol::CodexErrorInfo;
 use codex_protocol::protocol::ConversationAudioParams;
 use codex_protocol::protocol::ConversationStartParams;
@@ -455,7 +454,7 @@ async fn prepare_realtime_start(
     let provider = sess.provider().await;
     let auth = sess.services.auth_manager.auth().await;
     let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?;
-    let mut api_provider = provider.to_api_provider(Some(ApiAuthMode::ApiKey))?;
+    let mut api_provider = provider.to_api_provider(Some(crate::auth::AuthMode::ApiKey))?;
     let config = sess.get_config().await;
     if let Some(realtime_ws_base_url) = &config.experimental_realtime_ws_base_url {
         api_provider.base_url = realtime_ws_base_url.clone();

codex-rs/core/src/token_data.rs

@@ -5,21 +5,32 @@ use thiserror::Error;
 
 #[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Default)]
 pub struct TokenData {
+    /// Flat info parsed from the JWT in auth.json.
     #[serde(
         deserialize_with = "deserialize_id_token",
         serialize_with = "serialize_id_token"
     )]
     pub id_token: IdTokenInfo,
+
+    /// This is a JWT.
     pub access_token: String,
+
     pub refresh_token: String,
+
     pub account_id: Option<String>,
 }
 
+/// Flat subset of useful claims in id_token from auth.json.
 #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
 pub struct IdTokenInfo {
     pub email: Option<String>,
-    pub chatgpt_plan_type: Option<PlanType>,
+    /// The ChatGPT subscription plan type
+    /// (e.g., "free", "plus", "pro", "business", "enterprise", "edu").
+    /// (Note: values may vary by backend.)
+    pub(crate) chatgpt_plan_type: Option<PlanType>,
+    /// ChatGPT user identifier associated with the token, if present.
     pub chatgpt_user_id: Option<String>,
+    /// Organization/workspace identifier associated with the token, if present.
     pub chatgpt_account_id: Option<String>,
     pub raw_jwt: String,
 }
@@ -44,13 +55,13 @@ impl IdTokenInfo {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(untagged)]
-pub enum PlanType {
+pub(crate) enum PlanType {
     Known(KnownPlan),
     Unknown(String),
 }
 
 impl PlanType {
-    pub fn from_raw_value(raw: &str) -> Self {
+    pub(crate) fn from_raw_value(raw: &str) -> Self {
         match raw.to_ascii_lowercase().as_str() {
             "free" => Self::Known(KnownPlan::Free),
             "go" => Self::Known(KnownPlan::Go),
@@ -67,7 +78,7 @@ impl PlanType {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
-pub enum KnownPlan {
+pub(crate) enum KnownPlan {
     Free,
     Go,
     Plus,
@@ -117,6 +128,7 @@ pub enum IdTokenInfoError {
 }
 
 pub fn parse_chatgpt_jwt_claims(jwt: &str) -> Result<IdTokenInfo, IdTokenInfoError> {
+    // JWT format: header.payload.signature
     let mut parts = jwt.split('.');
     let (_header_b64, payload_b64, _sig_b64) = match (parts.next(), parts.next(), parts.next()) {
         (Some(h), Some(p), Some(s)) if !h.is_empty() && !p.is_empty() && !s.is_empty() => (h, p, s),
@@ -161,3 +173,7 @@ where
 {
     serializer.serialize_str(&id_token.raw_jwt)
 }
+
+#[cfg(test)]
+#[path = "token_data_tests.rs"]
+mod tests;

codex-rs/core/src/token_data_tests.rs

@@ -1,5 +1,4 @@
-use super::token_data::*;
-use base64::Engine;
+use super::*;
 use pretty_assertions::assert_eq;
 use serde::Serialize;
 

codex-rs/core/tests/suite/auth_refresh.rs

@@ -5,8 +5,6 @@ use chrono::Duration;
 use chrono::Utc;
 use codex_app_server_protocol::AuthMode;
 use codex_core::AuthManager;
-use codex_core::IdTokenInfo;
-use codex_core::TokenData;
 use codex_core::auth::AuthCredentialsStoreMode;
 use codex_core::auth::AuthDotJson;
 use codex_core::auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
@@ -14,6 +12,8 @@ use codex_core::auth::RefreshTokenError;
 use codex_core::auth::load_auth_dot_json;
 use codex_core::auth::save_auth;
 use codex_core::error::RefreshTokenFailedReason;
+use codex_core::token_data::IdTokenInfo;
+use codex_core::token_data::TokenData;
 use core_test_support::skip_if_no_network;
 use pretty_assertions::assert_eq;
 use serde::Serialize;

codex-rs/exec-server/BUILD.bazel

@@ -0,0 +1,7 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "exec-server",
+    crate_name = "codex_exec_server",
+    test_tags = ["no-sandbox"],
+)

codex-rs/exec-server/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "codex-exec-server"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[[bin]]
+name = "codex-exec-server"
+path = "src/bin/codex-exec-server.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+base64 = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
+codex-app-server-protocol = { workspace = true }
+codex-utils-pty = { workspace = true }
+futures = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = [
+    "io-std",
+    "io-util",
+    "macros",
+    "net",
+    "process",
+    "rt-multi-thread",
+    "sync",
+    "time",
+] }
+tokio-tungstenite = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+anyhow = { workspace = true }
+codex-utils-cargo-bin = { workspace = true }
+pretty_assertions = { workspace = true }

codex-rs/exec-server/DESIGN.md

@@ -0,0 +1,242 @@
+# exec-server design notes
+
+This document sketches a likely direction for integrating `codex-exec-server`
+with unified exec without baking the full tool-call policy stack into the
+server.
+
+The goal is:
+
+- keep exec-server generic and reusable
+- keep approval, sandbox, and retry policy in `core`
+- preserve the unified-exec event flow the model already depends on
+- support retained output caps so polling and snapshot-style APIs do not grow
+  memory without bound
+
+## Unified exec today
+
+Today the flow for LLM-visible interactive execution is:
+
+1. The model sees the `exec_command` and `write_stdin` tools.
+2. `UnifiedExecHandler` parses the tool arguments and allocates a process id.
+3. `UnifiedExecProcessManager::exec_command(...)` calls
+   `open_session_with_sandbox(...)`.
+4. `ToolOrchestrator` drives approval, sandbox selection, managed network
+   approval, and sandbox-denial retry behavior.
+5. `UnifiedExecRuntime` builds a `CommandSpec`, asks the current
+   `SandboxAttempt` to transform it into an `ExecRequest`, and passes that
+   resolved request back to the process manager.
+6. `open_session_with_exec_env(...)` spawns the process from that resolved
+   `ExecRequest`.
+7. Unified exec emits an `ExecCommandBegin` event.
+8. Unified exec starts a background output watcher that emits
+   `ExecCommandOutputDelta` events.
+9. The initial tool call collects output until the requested yield deadline and
+   returns an `ExecCommandToolOutput` snapshot to the model.
+10. If the process is still running, unified exec stores it and later emits
+    `ExecCommandEnd` when the exit watcher fires.
+11. A later `write_stdin` tool call writes to the stored process, emits a
+    `TerminalInteraction` event, collects another bounded snapshot, and returns
+    that tool response to the model.
+
+Important observation: the 250ms / 10s yield-window behavior is not really a
+process-server concern. It is a client-side convenience layer for the LLM tool
+API. The server should focus on raw process lifecycle and streaming events.
+
+## Proposed boundary
+
+The clean split is:
+
+- exec-server server: process lifecycle, output streaming, retained output caps
+- exec-server client: `wait`, `communicate`, yield-window helpers, session
+  bookkeeping
+- unified exec in `core`: tool parsing, event emission, approvals, sandboxing,
+  managed networking, retry semantics
+
+If exec-server is used by unified exec later, the boundary should sit between
+step 5 and step 6 above: after policy has produced a resolved spawn request, but
+before the actual PTY or pipe spawn.
+
+## Suggested process API
+
+Start simple and explicit:
+
+- `process/start`
+- `process/write`
+- `process/closeStdin`
+- `process/resize`
+- `process/terminate`
+- `process/wait`
+- `process/snapshot`
+
+Server notifications:
+
+- `process/output`
+- `process/exited`
+- optionally `process/started`
+- optionally `process/failed`
+
+Suggested request shapes:
+
+```rust
+enum ProcessStartRequest {
+    Direct(DirectExecSpec),
+    Prepared(PreparedExecSpec),
+}
+
+struct DirectExecSpec {
+    process_id: String,
+    argv: Vec<String>,
+    cwd: PathBuf,
+    env: HashMap<String, String>,
+    arg0: Option<String>,
+    io: ProcessIo,
+}
+
+struct PreparedExecSpec {
+    process_id: String,
+    request: PreparedExecRequest,
+    io: ProcessIo,
+}
+
+enum ProcessIo {
+    Pty { rows: u16, cols: u16 },
+    Pipe { stdin: StdinMode },
+}
+
+enum StdinMode {
+    Open,
+    Closed,
+}
+
+enum TerminateMode {
+    Graceful { timeout_ms: u64 },
+    Force,
+}
+```
+
+Notes:
+
+- `processId` remains a protocol handle, not an OS pid.
+- `wait` is a good generic API because many callers want process completion
+  without manually wiring notifications.
+- `communicate` is also a reasonable API, but it should probably start as a
+  client helper built on top of `write + closeStdin + wait + snapshot`.
+- If an RPC form of `communicate` is added later, it should be a convenience
+  wrapper rather than the primitive execution model.
+
+## Output capping
+
+Even with event streaming, the server should retain a bounded amount of output
+per process so callers can poll, wait, or reconnect without unbounded memory
+growth.
+
+Suggested behavior:
+
+- stream every output chunk live via `process/output`
+- retain capped output per process in memory
+- keep stdout and stderr separately for pipe-backed processes
+- for PTY-backed processes, treat retained output as a single terminal stream
+- expose truncation metadata on snapshots
+
+Suggested snapshot response:
+
+```rust
+struct ProcessSnapshot {
+    stdout: Vec<u8>,
+    stderr: Vec<u8>,
+    terminal: Vec<u8>,
+    truncated: bool,
+    exit_code: Option<i32>,
+    running: bool,
+}
+```
+
+Implementation-wise, the current `HeadTailBuffer` pattern used by unified exec
+is a good fit. The cap should be server config, not request config, so memory
+use stays predictable.
+
+## Sandboxing and networking
+
+### How unified exec does it today
+
+Unified exec does not hand raw command args directly to the PTY layer for tool
+calls. Instead, it:
+
+1. computes approval requirements
+2. chooses a sandbox attempt
+3. applies managed-network policy if needed
+4. transforms `CommandSpec` into `ExecRequest`
+5. spawns from that resolved `ExecRequest`
+
+That split is already valuable and should be preserved.
+
+### Recommended exec-server design
+
+Do not put approval policy into exec-server.
+
+Instead, support two execution modes:
+
+- `Direct`: raw command, intended for orchestrator-side or already-trusted use
+- `Prepared`: already-resolved spawn request, intended for tool-call execution
+
+For tool calls from the LLM side:
+
+1. `core` runs the existing approval + sandbox + managed-network flow
+2. `core` produces a resolved `ExecRequest`
+3. the exec-server client sends `PreparedExecSpec`
+4. exec-server spawns exactly that request and streams process events
+
+For orchestrator-side execution:
+
+1. caller sends `DirectExecSpec`
+2. exec-server spawns directly without running approval or sandbox policy
+
+This gives one generic process API while keeping the policy-sensitive logic in
+the place that already owns it.
+
+### Why not make exec-server own sandbox selection?
+
+That would force exec-server to understand:
+
+- approval policy
+- exec policy / prefix rules
+- managed-network approval flow
+- sandbox retry semantics
+- guardian routing
+- feature-flag-driven sandbox selection
+- platform-specific sandbox helper configuration
+
+That is too opinionated for a reusable process service.
+
+## Optional future server config
+
+If exec-server grows beyond the current prototype, a config object like this
+would be enough:
+
+```rust
+struct ExecServerConfig {
+    shutdown_grace_period_ms: u64,
+    max_processes_per_connection: usize,
+    retained_output_bytes_per_process: usize,
+    allow_direct_exec: bool,
+    allow_prepared_exec: bool,
+}
+```
+
+That keeps policy surface small:
+
+- lifecycle limits live in the server
+- trust and sandbox policy stay with the caller
+
+## Mapping back to LLM-visible events
+
+If unified exec is later backed by exec-server, the `core` client wrapper should
+keep owning the translation into the existing event model:
+
+- `process/start` success -> `ExecCommandBegin`
+- `process/output` -> `ExecCommandOutputDelta`
+- local `process/write` call -> `TerminalInteraction`
+- `process/exited` plus retained transcript -> `ExecCommandEnd`
+
+That preserves the current LLM-facing contract while making the process backend
+swappable.

codex-rs/exec-server/README.md

@@ -0,0 +1,392 @@
+# codex-exec-server
+
+`codex-exec-server` is a small standalone JSON-RPC server for spawning and
+controlling subprocesses through `codex-utils-pty`.
+
+It currently provides:
+
+- a standalone binary: `codex-exec-server`
+- a transport-agnostic server runtime with stdio and websocket entrypoints
+- a Rust client: `ExecServerClient`
+- a direct in-process client mode: `ExecServerClient::connect_in_process`
+- a separate local launch helper: `spawn_local_exec_server`
+- a small protocol module with shared request/response types
+
+This crate is intentionally narrow. It is not wired into the main Codex CLI or
+unified-exec in this PR; it is only the standalone transport layer.
+
+The internal shape is intentionally closer to `app-server` than the first cut:
+
+- transport adapters are separate from the per-connection request processor
+- JSON-RPC route matching is separate from the stateful exec handler
+- the client only speaks the protocol; it does not spawn a server subprocess
+- the client can also bypass the JSON-RPC transport/routing layer in local
+  in-process mode and call the typed handler directly
+- local child-process launch is handled by a separate helper/factory layer
+
+That split is meant to leave reusable seams if exec-server and app-server later
+share transport or JSON-RPC connection utilities. It also keeps the core
+handler testable without the RPC server implementation itself.
+
+Design notes for a likely future integration with unified exec, including
+rough call flow, buffering, and sandboxing boundaries, live in
+[DESIGN.md](./DESIGN.md).
+
+## Transport
+
+The server speaks the same JSON-RPC message shapes over multiple transports.
+
+The standalone binary supports:
+
+- `stdio://` (default)
+- `ws://IP:PORT`
+
+Wire framing:
+
+- stdio: one newline-delimited JSON-RPC message per line on stdin/stdout
+- websocket: one JSON-RPC message per websocket text frame
+
+Like the app-server transport, messages on the wire omit the `"jsonrpc":"2.0"`
+field and use the shared `codex-app-server-protocol` envelope types.
+
+The current protocol version is:
+
+```text
+exec-server.v0
+```
+
+## Lifecycle
+
+Each connection follows this sequence:
+
+1. Send `initialize`.
+2. Wait for the `initialize` response.
+3. Send `initialized`.
+4. Start and manage processes with `process/start`, `process/read`,
+   `process/write`, and `process/terminate`.
+5. Read streaming notifications from `process/output` and
+   `process/exited`.
+
+If the client sends exec methods before completing the `initialize` /
+`initialized` handshake, the server rejects them.
+
+If a connection closes, the server terminates any remaining managed processes
+for that connection.
+
+TODO: add authentication to the `initialize` setup before this is used across a
+trust boundary.
+
+## API
+
+### `initialize`
+
+Initial handshake request.
+
+Request params:
+
+```json
+{
+  "clientName": "my-client"
+}
+```
+
+Response:
+
+```json
+{
+  "protocolVersion": "exec-server.v0"
+}
+```
+
+### `initialized`
+
+Handshake acknowledgement notification sent by the client after a successful
+`initialize` response. Exec methods are rejected until this arrives.
+
+Params are currently ignored. Sending any other client notification method is a
+protocol error.
+
+### `process/start`
+
+Starts a new managed process.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1",
+  "argv": ["bash", "-lc", "printf 'hello\\n'"],
+  "cwd": "/absolute/working/directory",
+  "env": {
+    "PATH": "/usr/bin:/bin"
+  },
+  "tty": true,
+  "arg0": null
+}
+```
+
+Field definitions:
+
+- `argv`: command vector. It must be non-empty.
+- `cwd`: absolute working directory used for the child process.
+- `env`: environment variables passed to the child process.
+- `tty`: when `true`, spawn a PTY-backed interactive process; when `false`,
+  spawn a pipe-backed process with closed stdin.
+- `arg0`: optional argv0 override forwarded to `codex-utils-pty`.
+
+Response:
+
+```json
+{
+  "processId": "proc-1"
+}
+```
+
+Behavior notes:
+
+- `processId` is chosen by the client and must be unique for the connection.
+- PTY-backed processes accept later writes through `process/write`.
+- Pipe-backed processes are launched with stdin closed and reject writes.
+- Output is streamed asynchronously via `process/output`.
+- Exit is reported asynchronously via `process/exited`.
+
+### `process/write`
+
+Writes raw bytes to a running PTY-backed process stdin.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1",
+  "chunk": "aGVsbG8K"
+}
+```
+
+`chunk` is base64-encoded raw bytes. In the example above it is `hello\n`.
+
+Response:
+
+```json
+{
+  "accepted": true
+}
+```
+
+Behavior notes:
+
+- Writes to an unknown `processId` are rejected.
+- Writes to a non-PTY process are rejected because stdin is already closed.
+
+### `process/read`
+
+Reads retained output from a managed process by sequence number.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1",
+  "afterSeq": 0,
+  "maxBytes": 65536,
+  "waitMs": 250
+}
+```
+
+Response:
+
+```json
+{
+  "chunks": [
+    {
+      "seq": 1,
+      "stream": "pty",
+      "chunk": "aGVsbG8K"
+    }
+  ],
+  "nextSeq": 2,
+  "exited": false,
+  "exitCode": null
+}
+```
+
+Behavior notes:
+
+- Output is retained in bounded server memory so callers can poll without
+  relying only on notifications.
+- `afterSeq` is exclusive: `0` reads from the beginning of the retained buffer.
+- `waitMs` waits briefly for new output or exit if nothing is currently
+  available.
+- Once retained output exceeds the per-process cap, oldest chunks are dropped.
+
+### `process/terminate`
+
+Terminates a running managed process.
+
+Request params:
+
+```json
+{
+  "processId": "proc-1"
+}
+```
+
+Response:
+
+```json
+{
+  "running": true
+}
+```
+
+If the process is already unknown or already removed, the server responds with:
+
+```json
+{
+  "running": false
+}
+```
+
+## Notifications
+
+### `process/output`
+
+Streaming output chunk from a running process.
+
+Params:
+
+```json
+{
+  "processId": "proc-1",
+  "stream": "stdout",
+  "chunk": "aGVsbG8K"
+}
+```
+
+Fields:
+
+- `processId`: process identifier
+- `stream`: `"stdout"`, `"stderr"`, or `"pty"` for PTY-backed processes
+- `chunk`: base64-encoded output bytes
+
+### `process/exited`
+
+Final process exit notification.
+
+Params:
+
+```json
+{
+  "processId": "proc-1",
+  "exitCode": 0
+}
+```
+
+## Errors
+
+The server returns JSON-RPC errors with these codes:
+
+- `-32600`: invalid request
+- `-32602`: invalid params
+- `-32603`: internal error
+
+Typical error cases:
+
+- unknown method
+- malformed params
+- empty `argv`
+- duplicate `processId`
+- writes to unknown processes
+- writes to non-PTY processes
+
+## Rust surface
+
+The crate exports:
+
+- `ExecServerClient`
+- `ExecServerClientConnectOptions`
+- `RemoteExecServerConnectArgs`
+- `ExecServerLaunchCommand`
+- `ExecServerEvent`
+- `SpawnedExecServer`
+- `ExecServerError`
+- `ExecServerTransport`
+- `spawn_local_exec_server(...)`
+- protocol structs such as `ExecParams`, `ExecResponse`,
+  `WriteParams`, `TerminateParams`, `ExecOutputDeltaNotification`, and
+  `ExecExitedNotification`
+- `run_main()` and `run_main_with_transport(...)`
+
+### Binary
+
+Run over stdio:
+
+```text
+codex-exec-server
+```
+
+Run as a websocket server:
+
+```text
+codex-exec-server --listen ws://127.0.0.1:8080
+```
+
+### Client
+
+Connect the client to an existing server transport:
+
+- `ExecServerClient::connect_stdio(...)`
+- `ExecServerClient::connect_websocket(...)`
+- `ExecServerClient::connect_in_process(...)` for a local no-transport mode
+  backed directly by the typed handler
+
+Timeout behavior:
+
+- stdio and websocket clients both enforce an initialize-handshake timeout
+- websocket clients also enforce a connect timeout before the handshake begins
+
+Events:
+
+- `ExecServerClient::event_receiver()` yields `ExecServerEvent`
+- output events include both `stream` (`stdout`, `stderr`, or `pty`) and raw
+  bytes
+- process lifetime is tracked by server notifications such as
+  `process/exited`, not by a client-side process registry
+
+Spawning a local child process is deliberately separate:
+
+- `spawn_local_exec_server(...)`
+
+## Example session
+
+Initialize:
+
+```json
+{"id":1,"method":"initialize","params":{"clientName":"example-client"}}
+{"id":1,"result":{"protocolVersion":"exec-server.v0"}}
+{"method":"initialized","params":{}}
+```
+
+Start a process:
+
+```json
+{"id":2,"method":"process/start","params":{"processId":"proc-1","argv":["bash","-lc","printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"],"cwd":"/tmp","env":{"PATH":"/usr/bin:/bin"},"tty":true,"arg0":null}}
+{"id":2,"result":{"processId":"proc-1"}}
+{"method":"process/output","params":{"processId":"proc-1","stream":"pty","chunk":"cmVhZHkK"}}
+```
+
+Write to the process:
+
+```json
+{"id":3,"method":"process/write","params":{"processId":"proc-1","chunk":"aGVsbG8K"}}
+{"id":3,"result":{"accepted":true}}
+{"method":"process/output","params":{"processId":"proc-1","stream":"pty","chunk":"ZWNobzpoZWxsbwo="}}
+```
+
+Terminate it:
+
+```json
+{"id":4,"method":"process/terminate","params":{"processId":"proc-1"}}
+{"id":4,"result":{"running":true}}
+{"method":"process/exited","params":{"processId":"proc-1","exitCode":0}}
+```

codex-rs/exec-server/src/bin/codex-exec-server.rs

@@ -0,0 +1,23 @@
+use clap::Parser;
+use codex_exec_server::ExecServerTransport;
+
+#[derive(Debug, Parser)]
+struct ExecServerArgs {
+    /// Transport endpoint URL. Supported values: `stdio://` (default),
+    /// `ws://IP:PORT`.
+    #[arg(
+        long = "listen",
+        value_name = "URL",
+        default_value = ExecServerTransport::DEFAULT_LISTEN_URL
+    )]
+    listen: ExecServerTransport,
+}
+
+#[tokio::main]
+async fn main() {
+    let args = ExecServerArgs::parse();
+    if let Err(err) = codex_exec_server::run_main_with_transport(args.listen).await {
+        eprintln!("{err}");
+        std::process::exit(1);
+    }
+}

codex-rs/exec-server/src/client_api.rs

@@ -0,0 +1,27 @@
+use std::time::Duration;
+
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+
+/// Connection options for any exec-server client transport.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExecServerClientConnectOptions {
+    pub client_name: String,
+    pub initialize_timeout: Duration,
+}
+
+/// WebSocket connection arguments for a remote exec-server.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RemoteExecServerConnectArgs {
+    pub websocket_url: String,
+    pub client_name: String,
+    pub connect_timeout: Duration,
+    pub initialize_timeout: Duration,
+}
+
+/// Connection-level server events.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ExecServerEvent {
+    OutputDelta(ExecOutputDeltaNotification),
+    Exited(ExecExitedNotification),
+}

codex-rs/exec-server/src/connection.rs

@@ -0,0 +1,417 @@
+use codex_app_server_protocol::JSONRPCMessage;
+use futures::SinkExt;
+use futures::StreamExt;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::io::BufWriter;
+use tokio::sync::mpsc;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Message;
+
+pub(crate) const CHANNEL_CAPACITY: usize = 128;
+
+#[derive(Debug)]
+pub(crate) enum JsonRpcConnectionEvent {
+    Message(JSONRPCMessage),
+    Disconnected { reason: Option<String> },
+}
+
+pub(crate) struct JsonRpcConnection {
+    outgoing_tx: mpsc::Sender<JSONRPCMessage>,
+    incoming_rx: mpsc::Receiver<JsonRpcConnectionEvent>,
+}
+
+impl JsonRpcConnection {
+    pub(crate) fn from_stdio<R, W>(reader: R, writer: W, connection_label: String) -> Self
+    where
+        R: AsyncRead + Unpin + Send + 'static,
+        W: AsyncWrite + Unpin + Send + 'static,
+    {
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
+
+        let reader_label = connection_label.clone();
+        let incoming_tx_for_reader = incoming_tx.clone();
+        tokio::spawn(async move {
+            let mut lines = BufReader::new(reader).lines();
+            loop {
+                match lines.next_line().await {
+                    Ok(Some(line)) => {
+                        if line.trim().is_empty() {
+                            continue;
+                        }
+                        match serde_json::from_str::<JSONRPCMessage>(&line) {
+                            Ok(message) => {
+                                if incoming_tx_for_reader
+                                    .send(JsonRpcConnectionEvent::Message(message))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_disconnected(
+                                    &incoming_tx_for_reader,
+                                    Some(format!(
+                                        "failed to parse JSON-RPC message from {reader_label}: {err}"
+                                    )),
+                                )
+                                .await;
+                                break;
+                            }
+                        }
+                    }
+                    Ok(None) => {
+                        send_disconnected(&incoming_tx_for_reader, None).await;
+                        break;
+                    }
+                    Err(err) => {
+                        send_disconnected(
+                            &incoming_tx_for_reader,
+                            Some(format!(
+                                "failed to read JSON-RPC message from {reader_label}: {err}"
+                            )),
+                        )
+                        .await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        tokio::spawn(async move {
+            let mut writer = BufWriter::new(writer);
+            while let Some(message) = outgoing_rx.recv().await {
+                if let Err(err) = write_jsonrpc_line_message(&mut writer, &message).await {
+                    send_disconnected(
+                        &incoming_tx,
+                        Some(format!(
+                            "failed to write JSON-RPC message to {connection_label}: {err}"
+                        )),
+                    )
+                    .await;
+                    break;
+                }
+            }
+        });
+
+        Self {
+            outgoing_tx,
+            incoming_rx,
+        }
+    }
+
+    pub(crate) fn from_websocket<S>(stream: WebSocketStream<S>, connection_label: String) -> Self
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    {
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let (mut websocket_writer, mut websocket_reader) = stream.split();
+
+        let reader_label = connection_label.clone();
+        let incoming_tx_for_reader = incoming_tx.clone();
+        tokio::spawn(async move {
+            loop {
+                match websocket_reader.next().await {
+                    Some(Ok(Message::Text(text))) => {
+                        match serde_json::from_str::<JSONRPCMessage>(text.as_ref()) {
+                            Ok(message) => {
+                                if incoming_tx_for_reader
+                                    .send(JsonRpcConnectionEvent::Message(message))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_disconnected(
+                                    &incoming_tx_for_reader,
+                                    Some(format!(
+                                        "failed to parse websocket JSON-RPC message from {reader_label}: {err}"
+                                    )),
+                                )
+                                .await;
+                                break;
+                            }
+                        }
+                    }
+                    Some(Ok(Message::Binary(bytes))) => {
+                        match serde_json::from_slice::<JSONRPCMessage>(bytes.as_ref()) {
+                            Ok(message) => {
+                                if incoming_tx_for_reader
+                                    .send(JsonRpcConnectionEvent::Message(message))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_disconnected(
+                                    &incoming_tx_for_reader,
+                                    Some(format!(
+                                        "failed to parse websocket JSON-RPC message from {reader_label}: {err}"
+                                    )),
+                                )
+                                .await;
+                                break;
+                            }
+                        }
+                    }
+                    Some(Ok(Message::Close(_))) => {
+                        send_disconnected(&incoming_tx_for_reader, None).await;
+                        break;
+                    }
+                    Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}
+                    Some(Ok(_)) => {}
+                    Some(Err(err)) => {
+                        send_disconnected(
+                            &incoming_tx_for_reader,
+                            Some(format!(
+                                "failed to read websocket JSON-RPC message from {reader_label}: {err}"
+                            )),
+                        )
+                        .await;
+                        break;
+                    }
+                    None => {
+                        send_disconnected(&incoming_tx_for_reader, None).await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        tokio::spawn(async move {
+            while let Some(message) = outgoing_rx.recv().await {
+                match serialize_jsonrpc_message(&message) {
+                    Ok(encoded) => {
+                        if let Err(err) = websocket_writer.send(Message::Text(encoded.into())).await
+                        {
+                            send_disconnected(
+                                &incoming_tx,
+                                Some(format!(
+                                    "failed to write websocket JSON-RPC message to {connection_label}: {err}"
+                                )),
+                            )
+                            .await;
+                            break;
+                        }
+                    }
+                    Err(err) => {
+                        send_disconnected(
+                            &incoming_tx,
+                            Some(format!(
+                                "failed to serialize JSON-RPC message for {connection_label}: {err}"
+                            )),
+                        )
+                        .await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        Self {
+            outgoing_tx,
+            incoming_rx,
+        }
+    }
+
+    pub(crate) fn into_parts(
+        self,
+    ) -> (
+        mpsc::Sender<JSONRPCMessage>,
+        mpsc::Receiver<JsonRpcConnectionEvent>,
+    ) {
+        (self.outgoing_tx, self.incoming_rx)
+    }
+}
+
+async fn send_disconnected(
+    incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
+    reason: Option<String>,
+) {
+    let _ = incoming_tx
+        .send(JsonRpcConnectionEvent::Disconnected { reason })
+        .await;
+}
+
+async fn write_jsonrpc_line_message<W>(
+    writer: &mut BufWriter<W>,
+    message: &JSONRPCMessage,
+) -> std::io::Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    let encoded =
+        serialize_jsonrpc_message(message).map_err(|err| std::io::Error::other(err.to_string()))?;
+    writer.write_all(encoded.as_bytes()).await?;
+    writer.write_all(b"\n").await?;
+    writer.flush().await
+}
+
+fn serialize_jsonrpc_message(message: &JSONRPCMessage) -> Result<String, serde_json::Error> {
+    serde_json::to_string(message)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use codex_app_server_protocol::JSONRPCMessage;
+    use codex_app_server_protocol::JSONRPCRequest;
+    use codex_app_server_protocol::JSONRPCResponse;
+    use codex_app_server_protocol::RequestId;
+    use pretty_assertions::assert_eq;
+    use tokio::io::AsyncBufReadExt;
+    use tokio::io::AsyncWriteExt;
+    use tokio::io::BufReader;
+    use tokio::sync::mpsc;
+    use tokio::time::timeout;
+
+    use super::JsonRpcConnection;
+    use super::JsonRpcConnectionEvent;
+    use super::serialize_jsonrpc_message;
+
+    async fn recv_event(
+        incoming_rx: &mut mpsc::Receiver<JsonRpcConnectionEvent>,
+    ) -> JsonRpcConnectionEvent {
+        let recv_result = timeout(Duration::from_secs(1), incoming_rx.recv()).await;
+        let maybe_event = match recv_result {
+            Ok(maybe_event) => maybe_event,
+            Err(err) => panic!("timed out waiting for connection event: {err}"),
+        };
+        match maybe_event {
+            Some(event) => event,
+            None => panic!("connection event stream ended unexpectedly"),
+        }
+    }
+
+    async fn read_jsonrpc_line<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> JSONRPCMessage
+    where
+        R: tokio::io::AsyncRead + Unpin,
+    {
+        let next_line = timeout(Duration::from_secs(1), lines.next_line()).await;
+        let line_result = match next_line {
+            Ok(line_result) => line_result,
+            Err(err) => panic!("timed out waiting for JSON-RPC line: {err}"),
+        };
+        let maybe_line = match line_result {
+            Ok(maybe_line) => maybe_line,
+            Err(err) => panic!("failed to read JSON-RPC line: {err}"),
+        };
+        let line = match maybe_line {
+            Some(line) => line,
+            None => panic!("connection closed before JSON-RPC line arrived"),
+        };
+        match serde_json::from_str::<JSONRPCMessage>(&line) {
+            Ok(message) => message,
+            Err(err) => panic!("failed to parse JSON-RPC line: {err}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn stdio_connection_reads_and_writes_jsonrpc_messages() {
+        let (mut writer_to_connection, connection_reader) = tokio::io::duplex(1024);
+        let (connection_writer, reader_from_connection) = tokio::io::duplex(1024);
+        let connection =
+            JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
+        let (outgoing_tx, mut incoming_rx) = connection.into_parts();
+
+        let incoming_message = JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(7),
+            method: "initialize".to_string(),
+            params: Some(serde_json::json!({ "clientName": "test-client" })),
+            trace: None,
+        });
+        let encoded = match serialize_jsonrpc_message(&incoming_message) {
+            Ok(encoded) => encoded,
+            Err(err) => panic!("failed to serialize incoming message: {err}"),
+        };
+        if let Err(err) = writer_to_connection
+            .write_all(format!("{encoded}\n").as_bytes())
+            .await
+        {
+            panic!("failed to write to connection: {err}");
+        }
+
+        let event = recv_event(&mut incoming_rx).await;
+        match event {
+            JsonRpcConnectionEvent::Message(message) => {
+                assert_eq!(message, incoming_message);
+            }
+            JsonRpcConnectionEvent::Disconnected { reason } => {
+                panic!("unexpected disconnect event: {reason:?}");
+            }
+        }
+
+        let outgoing_message = JSONRPCMessage::Response(JSONRPCResponse {
+            id: RequestId::Integer(7),
+            result: serde_json::json!({ "protocolVersion": "exec-server.v0" }),
+        });
+        if let Err(err) = outgoing_tx.send(outgoing_message.clone()).await {
+            panic!("failed to queue outgoing message: {err}");
+        }
+
+        let mut lines = BufReader::new(reader_from_connection).lines();
+        let message = read_jsonrpc_line(&mut lines).await;
+        assert_eq!(message, outgoing_message);
+    }
+
+    #[tokio::test]
+    async fn stdio_connection_reports_parse_errors() {
+        let (mut writer_to_connection, connection_reader) = tokio::io::duplex(1024);
+        let (connection_writer, _reader_from_connection) = tokio::io::duplex(1024);
+        let connection =
+            JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
+        let (_outgoing_tx, mut incoming_rx) = connection.into_parts();
+
+        if let Err(err) = writer_to_connection.write_all(b"not-json\n").await {
+            panic!("failed to write invalid JSON: {err}");
+        }
+
+        let event = recv_event(&mut incoming_rx).await;
+        match event {
+            JsonRpcConnectionEvent::Disconnected { reason } => {
+                let reason = match reason {
+                    Some(reason) => reason,
+                    None => panic!("expected a parse error reason"),
+                };
+                assert!(
+                    reason.contains("failed to parse JSON-RPC message from test"),
+                    "unexpected disconnect reason: {reason}"
+                );
+            }
+            JsonRpcConnectionEvent::Message(message) => {
+                panic!("unexpected JSON-RPC message: {message:?}");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn stdio_connection_reports_clean_disconnect() {
+        let (writer_to_connection, connection_reader) = tokio::io::duplex(1024);
+        let (connection_writer, _reader_from_connection) = tokio::io::duplex(1024);
+        let connection =
+            JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
+        let (_outgoing_tx, mut incoming_rx) = connection.into_parts();
+        drop(writer_to_connection);
+
+        let event = recv_event(&mut incoming_rx).await;
+        match event {
+            JsonRpcConnectionEvent::Disconnected { reason } => {
+                assert_eq!(reason, None);
+            }
+            JsonRpcConnectionEvent::Message(message) => {
+                panic!("unexpected JSON-RPC message: {message:?}");
+            }
+        }
+    }
+}

codex-rs/exec-server/src/lib.rs

@@ -0,0 +1,30 @@
+mod client;
+mod client_api;
+mod connection;
+mod local;
+mod protocol;
+mod server;
+
+pub use client::ExecServerClient;
+pub use client::ExecServerError;
+pub use client_api::ExecServerClientConnectOptions;
+pub use client_api::ExecServerEvent;
+pub use client_api::RemoteExecServerConnectArgs;
+pub use local::ExecServerLaunchCommand;
+pub use local::SpawnedExecServer;
+pub use local::spawn_local_exec_server;
+pub use protocol::ExecExitedNotification;
+pub use protocol::ExecOutputDeltaNotification;
+pub use protocol::ExecOutputStream;
+pub use protocol::ExecParams;
+pub use protocol::ExecResponse;
+pub use protocol::InitializeParams;
+pub use protocol::InitializeResponse;
+pub use protocol::TerminateParams;
+pub use protocol::TerminateResponse;
+pub use protocol::WriteParams;
+pub use protocol::WriteResponse;
+pub use server::ExecServerTransport;
+pub use server::ExecServerTransportParseError;
+pub use server::run_main;
+pub use server::run_main_with_transport;

codex-rs/exec-server/src/local.rs

@@ -0,0 +1,70 @@
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::sync::Mutex as StdMutex;
+
+use tokio::process::Child;
+use tokio::process::Command;
+
+use crate::client::ExecServerClient;
+use crate::client::ExecServerError;
+use crate::client_api::ExecServerClientConnectOptions;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExecServerLaunchCommand {
+    pub program: PathBuf,
+    pub args: Vec<String>,
+}
+
+pub struct SpawnedExecServer {
+    client: ExecServerClient,
+    child: StdMutex<Option<Child>>,
+}
+
+impl SpawnedExecServer {
+    pub fn client(&self) -> &ExecServerClient {
+        &self.client
+    }
+}
+
+impl Drop for SpawnedExecServer {
+    fn drop(&mut self) {
+        if let Ok(mut child_guard) = self.child.lock()
+            && let Some(child) = child_guard.as_mut()
+        {
+            let _ = child.start_kill();
+        }
+    }
+}
+
+pub async fn spawn_local_exec_server(
+    command: ExecServerLaunchCommand,
+    options: ExecServerClientConnectOptions,
+) -> Result<SpawnedExecServer, ExecServerError> {
+    let mut child = Command::new(&command.program);
+    child.args(&command.args);
+    child.stdin(Stdio::piped());
+    child.stdout(Stdio::piped());
+    child.stderr(Stdio::inherit());
+    child.kill_on_drop(true);
+
+    let mut child = child.spawn().map_err(ExecServerError::Spawn)?;
+    let stdin = child.stdin.take().ok_or_else(|| {
+        ExecServerError::Protocol("exec-server stdin was not captured".to_string())
+    })?;
+    let stdout = child.stdout.take().ok_or_else(|| {
+        ExecServerError::Protocol("exec-server stdout was not captured".to_string())
+    })?;
+
+    let client = match ExecServerClient::connect_stdio(stdin, stdout, options).await {
+        Ok(client) => client,
+        Err(err) => {
+            let _ = child.start_kill();
+            return Err(err);
+        }
+    };
+
+    Ok(SpawnedExecServer {
+        client,
+        child: StdMutex::new(Some(child)),
+    })
+}

codex-rs/exec-server/src/protocol.rs

@@ -0,0 +1,162 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use serde::Deserialize;
+use serde::Serialize;
+
+pub const INITIALIZE_METHOD: &str = "initialize";
+pub const INITIALIZED_METHOD: &str = "initialized";
+pub const EXEC_METHOD: &str = "process/start";
+pub const EXEC_READ_METHOD: &str = "process/read";
+pub const EXEC_WRITE_METHOD: &str = "process/write";
+pub const EXEC_TERMINATE_METHOD: &str = "process/terminate";
+pub const EXEC_OUTPUT_DELTA_METHOD: &str = "process/output";
+pub const EXEC_EXITED_METHOD: &str = "process/exited";
+pub const PROTOCOL_VERSION: &str = "exec-server.v0";
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct ByteChunk(#[serde(with = "base64_bytes")] pub Vec<u8>);
+
+impl ByteChunk {
+    pub fn into_inner(self) -> Vec<u8> {
+        self.0
+    }
+}
+
+impl From<Vec<u8>> for ByteChunk {
+    fn from(value: Vec<u8>) -> Self {
+        Self(value)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InitializeParams {
+    pub client_name: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InitializeResponse {
+    pub protocol_version: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecParams {
+    /// Client-chosen logical process handle scoped to this connection/session.
+    /// This is a protocol key, not an OS pid.
+    pub process_id: String,
+    pub argv: Vec<String>,
+    pub cwd: PathBuf,
+    pub env: HashMap<String, String>,
+    pub tty: bool,
+    pub arg0: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecResponse {
+    pub process_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReadParams {
+    pub process_id: String,
+    pub after_seq: Option<u64>,
+    pub max_bytes: Option<usize>,
+    pub wait_ms: Option<u64>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProcessOutputChunk {
+    pub seq: u64,
+    pub stream: ExecOutputStream,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReadResponse {
+    pub chunks: Vec<ProcessOutputChunk>,
+    pub next_seq: u64,
+    pub exited: bool,
+    pub exit_code: Option<i32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteParams {
+    pub process_id: String,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteResponse {
+    pub accepted: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TerminateParams {
+    pub process_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TerminateResponse {
+    pub running: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecOutputStream {
+    Stdout,
+    Stderr,
+    Pty,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecOutputDeltaNotification {
+    pub process_id: String,
+    pub stream: ExecOutputStream,
+    pub chunk: ByteChunk,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecExitedNotification {
+    pub process_id: String,
+    pub exit_code: i32,
+}
+
+mod base64_bytes {
+    use super::BASE64_STANDARD;
+    use base64::Engine as _;
+    use serde::Deserialize;
+    use serde::Deserializer;
+    use serde::Serializer;
+
+    pub fn serialize<S>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&BASE64_STANDARD.encode(bytes))
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let encoded = String::deserialize(deserializer)?;
+        BASE64_STANDARD
+            .decode(encoded)
+            .map_err(serde::de::Error::custom)
+    }
+}

codex-rs/exec-server/src/server.rs

@@ -0,0 +1,20 @@
+mod handler;
+mod processor;
+mod routing;
+mod transport;
+
+pub(crate) use handler::ExecServerHandler;
+pub(crate) use routing::ExecServerOutboundMessage;
+pub(crate) use routing::ExecServerServerNotification;
+pub use transport::ExecServerTransport;
+pub use transport::ExecServerTransportParseError;
+
+pub async fn run_main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    run_main_with_transport(ExecServerTransport::Stdio).await
+}
+
+pub async fn run_main_with_transport(
+    transport: ExecServerTransport,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    transport::run_transport(transport).await
+}

codex-rs/exec-server/src/server/processor.rs

@@ -0,0 +1,139 @@
+use tokio::sync::mpsc;
+use tracing::debug;
+use tracing::warn;
+
+use crate::connection::CHANNEL_CAPACITY;
+use crate::connection::JsonRpcConnection;
+use crate::connection::JsonRpcConnectionEvent;
+use crate::server::handler::ExecServerHandler;
+use crate::server::routing::ExecServerClientNotification;
+use crate::server::routing::ExecServerInboundMessage;
+use crate::server::routing::ExecServerOutboundMessage;
+use crate::server::routing::ExecServerRequest;
+use crate::server::routing::ExecServerResponseMessage;
+use crate::server::routing::RoutedExecServerMessage;
+use crate::server::routing::encode_outbound_message;
+use crate::server::routing::route_jsonrpc_message;
+
+pub(crate) async fn run_connection(connection: JsonRpcConnection) {
+    let (json_outgoing_tx, mut incoming_rx) = connection.into_parts();
+    let (outgoing_tx, mut outgoing_rx) =
+        mpsc::channel::<ExecServerOutboundMessage>(CHANNEL_CAPACITY);
+    let mut handler = ExecServerHandler::new(outgoing_tx.clone());
+
+    let outbound_task = tokio::spawn(async move {
+        while let Some(message) = outgoing_rx.recv().await {
+            let json_message = match encode_outbound_message(message) {
+                Ok(json_message) => json_message,
+                Err(err) => {
+                    warn!("failed to serialize exec-server outbound message: {err}");
+                    break;
+                }
+            };
+            if json_outgoing_tx.send(json_message).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    while let Some(event) = incoming_rx.recv().await {
+        match event {
+            JsonRpcConnectionEvent::Message(message) => match route_jsonrpc_message(message) {
+                Ok(RoutedExecServerMessage::Inbound(message)) => {
+                    if let Err(err) = dispatch_to_handler(&mut handler, message, &outgoing_tx).await
+                    {
+                        warn!("closing exec-server connection after protocol error: {err}");
+                        break;
+                    }
+                }
+                Ok(RoutedExecServerMessage::ImmediateOutbound(message)) => {
+                    if outgoing_tx.send(message).await.is_err() {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    warn!("closing exec-server connection after protocol error: {err}");
+                    break;
+                }
+            },
+            JsonRpcConnectionEvent::Disconnected { reason } => {
+                if let Some(reason) = reason {
+                    debug!("exec-server connection disconnected: {reason}");
+                }
+                break;
+            }
+        }
+    }
+
+    handler.shutdown().await;
+    drop(handler);
+    drop(outgoing_tx);
+    let _ = outbound_task.await;
+}
+
+async fn dispatch_to_handler(
+    handler: &mut ExecServerHandler,
+    message: ExecServerInboundMessage,
+    outgoing_tx: &mpsc::Sender<ExecServerOutboundMessage>,
+) -> Result<(), String> {
+    match message {
+        ExecServerInboundMessage::Request(request) => {
+            let outbound = match request {
+                ExecServerRequest::Initialize { request_id, .. } => request_outbound(
+                    request_id,
+                    handler
+                        .initialize()
+                        .map(ExecServerResponseMessage::Initialize),
+                ),
+                ExecServerRequest::Exec { request_id, params } => request_outbound(
+                    request_id,
+                    handler
+                        .exec(params)
+                        .await
+                        .map(ExecServerResponseMessage::Exec),
+                ),
+                ExecServerRequest::Read { request_id, params } => request_outbound(
+                    request_id,
+                    handler
+                        .read(params)
+                        .await
+                        .map(ExecServerResponseMessage::Read),
+                ),
+                ExecServerRequest::Write { request_id, params } => request_outbound(
+                    request_id,
+                    handler
+                        .write(params)
+                        .await
+                        .map(ExecServerResponseMessage::Write),
+                ),
+                ExecServerRequest::Terminate { request_id, params } => request_outbound(
+                    request_id,
+                    handler
+                        .terminate(params)
+                        .await
+                        .map(ExecServerResponseMessage::Terminate),
+                ),
+            };
+            outgoing_tx
+                .send(outbound)
+                .await
+                .map_err(|_| "outbound channel closed".to_string())
+        }
+        ExecServerInboundMessage::Notification(ExecServerClientNotification::Initialized) => {
+            handler.initialized()
+        }
+    }
+}
+
+fn request_outbound(
+    request_id: codex_app_server_protocol::RequestId,
+    result: Result<ExecServerResponseMessage, codex_app_server_protocol::JSONRPCErrorError>,
+) -> ExecServerOutboundMessage {
+    match result {
+        Ok(response) => ExecServerOutboundMessage::Response {
+            request_id,
+            response,
+        },
+        Err(error) => ExecServerOutboundMessage::Error { request_id, error },
+    }
+}

codex-rs/exec-server/src/server/routing.rs

@@ -0,0 +1,454 @@
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use serde::de::DeserializeOwned;
+
+use crate::protocol::EXEC_EXITED_METHOD;
+use crate::protocol::EXEC_METHOD;
+use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
+use crate::protocol::EXEC_READ_METHOD;
+use crate::protocol::EXEC_TERMINATE_METHOD;
+use crate::protocol::EXEC_WRITE_METHOD;
+use crate::protocol::ExecExitedNotification;
+use crate::protocol::ExecOutputDeltaNotification;
+use crate::protocol::ExecParams;
+use crate::protocol::ExecResponse;
+use crate::protocol::INITIALIZE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::InitializeParams;
+use crate::protocol::InitializeResponse;
+use crate::protocol::ReadParams;
+use crate::protocol::ReadResponse;
+use crate::protocol::TerminateParams;
+use crate::protocol::TerminateResponse;
+use crate::protocol::WriteParams;
+use crate::protocol::WriteResponse;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum ExecServerInboundMessage {
+    Request(ExecServerRequest),
+    Notification(ExecServerClientNotification),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum ExecServerRequest {
+    Initialize {
+        request_id: RequestId,
+        params: InitializeParams,
+    },
+    Exec {
+        request_id: RequestId,
+        params: ExecParams,
+    },
+    Read {
+        request_id: RequestId,
+        params: ReadParams,
+    },
+    Write {
+        request_id: RequestId,
+        params: WriteParams,
+    },
+    Terminate {
+        request_id: RequestId,
+        params: TerminateParams,
+    },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum ExecServerClientNotification {
+    Initialized,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) enum ExecServerOutboundMessage {
+    Response {
+        request_id: RequestId,
+        response: ExecServerResponseMessage,
+    },
+    Error {
+        request_id: RequestId,
+        error: JSONRPCErrorError,
+    },
+    Notification(ExecServerServerNotification),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum ExecServerResponseMessage {
+    Initialize(InitializeResponse),
+    Exec(ExecResponse),
+    Read(ReadResponse),
+    Write(WriteResponse),
+    Terminate(TerminateResponse),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum ExecServerServerNotification {
+    OutputDelta(ExecOutputDeltaNotification),
+    Exited(ExecExitedNotification),
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) enum RoutedExecServerMessage {
+    Inbound(ExecServerInboundMessage),
+    ImmediateOutbound(ExecServerOutboundMessage),
+}
+
+pub(crate) fn route_jsonrpc_message(
+    message: JSONRPCMessage,
+) -> Result<RoutedExecServerMessage, String> {
+    match message {
+        JSONRPCMessage::Request(request) => route_request(request),
+        JSONRPCMessage::Notification(notification) => route_notification(notification),
+        JSONRPCMessage::Response(response) => Err(format!(
+            "unexpected client response for request id {:?}",
+            response.id
+        )),
+        JSONRPCMessage::Error(error) => Err(format!(
+            "unexpected client error for request id {:?}",
+            error.id
+        )),
+    }
+}
+
+pub(crate) fn encode_outbound_message(
+    message: ExecServerOutboundMessage,
+) -> Result<JSONRPCMessage, serde_json::Error> {
+    match message {
+        ExecServerOutboundMessage::Response {
+            request_id,
+            response,
+        } => Ok(JSONRPCMessage::Response(JSONRPCResponse {
+            id: request_id,
+            result: serialize_response(response)?,
+        })),
+        ExecServerOutboundMessage::Error { request_id, error } => {
+            Ok(JSONRPCMessage::Error(JSONRPCError {
+                id: request_id,
+                error,
+            }))
+        }
+        ExecServerOutboundMessage::Notification(notification) => Ok(JSONRPCMessage::Notification(
+            serialize_notification(notification)?,
+        )),
+    }
+}
+
+pub(crate) fn invalid_request(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32600,
+        data: None,
+        message,
+    }
+}
+
+pub(crate) fn invalid_params(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32602,
+        data: None,
+        message,
+    }
+}
+
+pub(crate) fn internal_error(message: String) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: -32603,
+        data: None,
+        message,
+    }
+}
+
+fn route_request(request: JSONRPCRequest) -> Result<RoutedExecServerMessage, String> {
+    match request.method.as_str() {
+        INITIALIZE_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::Initialize { request_id, params }
+        })),
+        EXEC_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::Exec { request_id, params }
+        })),
+        EXEC_READ_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::Read { request_id, params }
+        })),
+        EXEC_WRITE_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::Write { request_id, params }
+        })),
+        EXEC_TERMINATE_METHOD => Ok(parse_request_params(request, |request_id, params| {
+            ExecServerRequest::Terminate { request_id, params }
+        })),
+        other => Ok(RoutedExecServerMessage::ImmediateOutbound(
+            ExecServerOutboundMessage::Error {
+                request_id: request.id,
+                error: invalid_request(format!("unknown method: {other}")),
+            },
+        )),
+    }
+}
+
+fn route_notification(
+    notification: JSONRPCNotification,
+) -> Result<RoutedExecServerMessage, String> {
+    match notification.method.as_str() {
+        INITIALIZED_METHOD => Ok(RoutedExecServerMessage::Inbound(
+            ExecServerInboundMessage::Notification(ExecServerClientNotification::Initialized),
+        )),
+        other => Err(format!("unexpected notification method: {other}")),
+    }
+}
+
+fn parse_request_params<P, F>(request: JSONRPCRequest, build: F) -> RoutedExecServerMessage
+where
+    P: DeserializeOwned,
+    F: FnOnce(RequestId, P) -> ExecServerRequest,
+{
+    let request_id = request.id;
+    match serde_json::from_value::<P>(request.params.unwrap_or(serde_json::Value::Null)) {
+        Ok(params) => RoutedExecServerMessage::Inbound(ExecServerInboundMessage::Request(build(
+            request_id, params,
+        ))),
+        Err(err) => RoutedExecServerMessage::ImmediateOutbound(ExecServerOutboundMessage::Error {
+            request_id,
+            error: invalid_params(err.to_string()),
+        }),
+    }
+}
+
+fn serialize_response(
+    response: ExecServerResponseMessage,
+) -> Result<serde_json::Value, serde_json::Error> {
+    match response {
+        ExecServerResponseMessage::Initialize(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::Exec(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::Read(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::Write(response) => serde_json::to_value(response),
+        ExecServerResponseMessage::Terminate(response) => serde_json::to_value(response),
+    }
+}
+
+fn serialize_notification(
+    notification: ExecServerServerNotification,
+) -> Result<JSONRPCNotification, serde_json::Error> {
+    match notification {
+        ExecServerServerNotification::OutputDelta(params) => Ok(JSONRPCNotification {
+            method: EXEC_OUTPUT_DELTA_METHOD.to_string(),
+            params: Some(serde_json::to_value(params)?),
+        }),
+        ExecServerServerNotification::Exited(params) => Ok(JSONRPCNotification {
+            method: EXEC_EXITED_METHOD.to_string(),
+            params: Some(serde_json::to_value(params)?),
+        }),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use pretty_assertions::assert_eq;
+    use serde_json::json;
+
+    use super::ExecServerClientNotification;
+    use super::ExecServerInboundMessage;
+    use super::ExecServerOutboundMessage;
+    use super::ExecServerRequest;
+    use super::ExecServerResponseMessage;
+    use super::ExecServerServerNotification;
+    use super::RoutedExecServerMessage;
+    use super::encode_outbound_message;
+    use super::route_jsonrpc_message;
+    use crate::protocol::EXEC_EXITED_METHOD;
+    use crate::protocol::EXEC_METHOD;
+    use crate::protocol::ExecExitedNotification;
+    use crate::protocol::ExecParams;
+    use crate::protocol::ExecResponse;
+    use crate::protocol::INITIALIZE_METHOD;
+    use crate::protocol::INITIALIZED_METHOD;
+    use crate::protocol::InitializeParams;
+    use codex_app_server_protocol::JSONRPCMessage;
+    use codex_app_server_protocol::JSONRPCNotification;
+    use codex_app_server_protocol::JSONRPCRequest;
+    use codex_app_server_protocol::JSONRPCResponse;
+    use codex_app_server_protocol::RequestId;
+
+    #[test]
+    fn routes_initialize_requests_to_typed_variants() {
+        let routed = route_jsonrpc_message(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(1),
+            method: INITIALIZE_METHOD.to_string(),
+            params: Some(json!({ "clientName": "test-client" })),
+            trace: None,
+        }))
+        .expect("initialize request should route");
+
+        assert_eq!(
+            routed,
+            RoutedExecServerMessage::Inbound(ExecServerInboundMessage::Request(
+                ExecServerRequest::Initialize {
+                    request_id: RequestId::Integer(1),
+                    params: InitializeParams {
+                        client_name: "test-client".to_string(),
+                    },
+                },
+            ))
+        );
+    }
+
+    #[test]
+    fn malformed_exec_params_return_immediate_error_outbound() {
+        let routed = route_jsonrpc_message(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(2),
+            method: EXEC_METHOD.to_string(),
+            params: Some(json!({ "processId": "proc-1" })),
+            trace: None,
+        }))
+        .expect("exec request should route");
+
+        let RoutedExecServerMessage::ImmediateOutbound(ExecServerOutboundMessage::Error {
+            request_id,
+            error,
+        }) = routed
+        else {
+            panic!("expected invalid-params error outbound");
+        };
+        assert_eq!(request_id, RequestId::Integer(2));
+        assert_eq!(error.code, -32602);
+    }
+
+    #[test]
+    fn routes_initialized_notifications_to_typed_variants() {
+        let routed = route_jsonrpc_message(JSONRPCMessage::Notification(JSONRPCNotification {
+            method: INITIALIZED_METHOD.to_string(),
+            params: Some(json!({})),
+        }))
+        .expect("initialized notification should route");
+
+        assert_eq!(
+            routed,
+            RoutedExecServerMessage::Inbound(ExecServerInboundMessage::Notification(
+                ExecServerClientNotification::Initialized,
+            ))
+        );
+    }
+
+    #[test]
+    fn serializes_typed_notifications_back_to_jsonrpc() {
+        let message = encode_outbound_message(ExecServerOutboundMessage::Notification(
+            ExecServerServerNotification::Exited(ExecExitedNotification {
+                process_id: "proc-1".to_string(),
+                exit_code: 0,
+            }),
+        ))
+        .expect("notification should serialize");
+
+        assert_eq!(
+            message,
+            JSONRPCMessage::Notification(JSONRPCNotification {
+                method: EXEC_EXITED_METHOD.to_string(),
+                params: Some(json!({
+                    "processId": "proc-1",
+                    "exitCode": 0,
+                })),
+            })
+        );
+    }
+
+    #[test]
+    fn serializes_typed_responses_back_to_jsonrpc() {
+        let message = encode_outbound_message(ExecServerOutboundMessage::Response {
+            request_id: RequestId::Integer(3),
+            response: ExecServerResponseMessage::Exec(ExecResponse {
+                process_id: "proc-1".to_string(),
+            }),
+        })
+        .expect("response should serialize");
+
+        assert_eq!(
+            message,
+            JSONRPCMessage::Response(codex_app_server_protocol::JSONRPCResponse {
+                id: RequestId::Integer(3),
+                result: json!({
+                    "processId": "proc-1",
+                }),
+            })
+        );
+    }
+
+    #[test]
+    fn routes_exec_requests_with_typed_params() {
+        let cwd = std::env::current_dir().expect("cwd");
+        let routed = route_jsonrpc_message(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(4),
+            method: EXEC_METHOD.to_string(),
+            params: Some(json!({
+                "processId": "proc-1",
+                "argv": ["bash", "-lc", "true"],
+                "cwd": cwd,
+                "env": {},
+                "tty": true,
+                "arg0": null,
+            })),
+            trace: None,
+        }))
+        .expect("exec request should route");
+
+        let RoutedExecServerMessage::Inbound(ExecServerInboundMessage::Request(
+            ExecServerRequest::Exec { request_id, params },
+        )) = routed
+        else {
+            panic!("expected typed exec request");
+        };
+        assert_eq!(request_id, RequestId::Integer(4));
+        assert_eq!(
+            params,
+            ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec!["bash".to_string(), "-lc".to_string(), "true".to_string()],
+                cwd: std::env::current_dir().expect("cwd"),
+                env: std::collections::HashMap::new(),
+                tty: true,
+                arg0: None,
+            }
+        );
+    }
+
+    #[test]
+    fn unknown_request_methods_return_immediate_invalid_request_errors() {
+        let routed = route_jsonrpc_message(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(5),
+            method: "process/unknown".to_string(),
+            params: Some(json!({})),
+            trace: None,
+        }))
+        .expect("unknown request should still route");
+
+        assert_eq!(
+            routed,
+            RoutedExecServerMessage::ImmediateOutbound(ExecServerOutboundMessage::Error {
+                request_id: RequestId::Integer(5),
+                error: super::invalid_request("unknown method: process/unknown".to_string()),
+            })
+        );
+    }
+
+    #[test]
+    fn unexpected_client_notifications_are_rejected() {
+        let err = route_jsonrpc_message(JSONRPCMessage::Notification(JSONRPCNotification {
+            method: "process/output".to_string(),
+            params: Some(json!({})),
+        }))
+        .expect_err("unexpected client notification should fail");
+
+        assert_eq!(err, "unexpected notification method: process/output");
+    }
+
+    #[test]
+    fn unexpected_client_responses_are_rejected() {
+        let err = route_jsonrpc_message(JSONRPCMessage::Response(JSONRPCResponse {
+            id: RequestId::Integer(6),
+            result: json!({}),
+        }))
+        .expect_err("unexpected client response should fail");
+
+        assert_eq!(err, "unexpected client response for request id Integer(6)");
+    }
+}

codex-rs/exec-server/src/server/transport.rs

@@ -0,0 +1,166 @@
+use std::net::SocketAddr;
+use std::str::FromStr;
+
+use tokio::net::TcpListener;
+use tokio_tungstenite::accept_async;
+use tracing::warn;
+
+use crate::connection::JsonRpcConnection;
+use crate::server::processor::run_connection;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ExecServerTransport {
+    Stdio,
+    WebSocket { bind_address: SocketAddr },
+}
+
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub enum ExecServerTransportParseError {
+    UnsupportedListenUrl(String),
+    InvalidWebSocketListenUrl(String),
+}
+
+impl std::fmt::Display for ExecServerTransportParseError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ExecServerTransportParseError::UnsupportedListenUrl(listen_url) => write!(
+                f,
+                "unsupported --listen URL `{listen_url}`; expected `stdio://` or `ws://IP:PORT`"
+            ),
+            ExecServerTransportParseError::InvalidWebSocketListenUrl(listen_url) => write!(
+                f,
+                "invalid websocket --listen URL `{listen_url}`; expected `ws://IP:PORT`"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for ExecServerTransportParseError {}
+
+impl ExecServerTransport {
+    pub const DEFAULT_LISTEN_URL: &str = "stdio://";
+
+    pub fn from_listen_url(listen_url: &str) -> Result<Self, ExecServerTransportParseError> {
+        if listen_url == Self::DEFAULT_LISTEN_URL {
+            return Ok(Self::Stdio);
+        }
+
+        if let Some(socket_addr) = listen_url.strip_prefix("ws://") {
+            let bind_address = socket_addr.parse::<SocketAddr>().map_err(|_| {
+                ExecServerTransportParseError::InvalidWebSocketListenUrl(listen_url.to_string())
+            })?;
+            return Ok(Self::WebSocket { bind_address });
+        }
+
+        Err(ExecServerTransportParseError::UnsupportedListenUrl(
+            listen_url.to_string(),
+        ))
+    }
+}
+
+impl FromStr for ExecServerTransport {
+    type Err = ExecServerTransportParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Self::from_listen_url(s)
+    }
+}
+
+pub(crate) async fn run_transport(
+    transport: ExecServerTransport,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    match transport {
+        ExecServerTransport::Stdio => {
+            run_connection(JsonRpcConnection::from_stdio(
+                tokio::io::stdin(),
+                tokio::io::stdout(),
+                "exec-server stdio".to_string(),
+            ))
+            .await;
+            Ok(())
+        }
+        ExecServerTransport::WebSocket { bind_address } => {
+            run_websocket_listener(bind_address).await
+        }
+    }
+}
+
+async fn run_websocket_listener(
+    bind_address: SocketAddr,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let listener = TcpListener::bind(bind_address).await?;
+    let local_addr = listener.local_addr()?;
+    print_websocket_startup_banner(local_addr);
+
+    loop {
+        let (stream, peer_addr) = listener.accept().await?;
+        tokio::spawn(async move {
+            match accept_async(stream).await {
+                Ok(websocket) => {
+                    run_connection(JsonRpcConnection::from_websocket(
+                        websocket,
+                        format!("exec-server websocket {peer_addr}"),
+                    ))
+                    .await;
+                }
+                Err(err) => {
+                    warn!(
+                        "failed to accept exec-server websocket connection from {peer_addr}: {err}"
+                    );
+                }
+            }
+        });
+    }
+}
+
+#[allow(clippy::print_stderr)]
+fn print_websocket_startup_banner(addr: SocketAddr) {
+    eprintln!("codex-exec-server listening on ws://{addr}");
+}
+
+#[cfg(test)]
+mod tests {
+    use pretty_assertions::assert_eq;
+
+    use super::ExecServerTransport;
+
+    #[test]
+    fn exec_server_transport_parses_stdio_listen_url() {
+        let transport =
+            ExecServerTransport::from_listen_url(ExecServerTransport::DEFAULT_LISTEN_URL)
+                .expect("stdio listen URL should parse");
+        assert_eq!(transport, ExecServerTransport::Stdio);
+    }
+
+    #[test]
+    fn exec_server_transport_parses_websocket_listen_url() {
+        let transport = ExecServerTransport::from_listen_url("ws://127.0.0.1:1234")
+            .expect("websocket listen URL should parse");
+        assert_eq!(
+            transport,
+            ExecServerTransport::WebSocket {
+                bind_address: "127.0.0.1:1234".parse().expect("valid socket address"),
+            }
+        );
+    }
+
+    #[test]
+    fn exec_server_transport_rejects_invalid_websocket_listen_url() {
+        let err = ExecServerTransport::from_listen_url("ws://localhost:1234")
+            .expect_err("hostname bind address should be rejected");
+        assert_eq!(
+            err.to_string(),
+            "invalid websocket --listen URL `ws://localhost:1234`; expected `ws://IP:PORT`"
+        );
+    }
+
+    #[test]
+    fn exec_server_transport_rejects_unsupported_listen_url() {
+        let err = ExecServerTransport::from_listen_url("http://127.0.0.1:1234")
+            .expect_err("unsupported scheme should fail");
+        assert_eq!(
+            err.to_string(),
+            "unsupported --listen URL `http://127.0.0.1:1234`; expected `stdio://` or `ws://IP:PORT`"
+        );
+    }
+}

codex-rs/exec-server/tests/stdio_smoke.rs

@@ -0,0 +1,298 @@
+#![cfg(unix)]
+
+use std::process::Stdio;
+use std::time::Duration;
+
+use anyhow::Context;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_exec_server::ExecOutputStream;
+use codex_exec_server::ExecParams;
+use codex_exec_server::ExecServerClient;
+use codex_exec_server::ExecServerClientConnectOptions;
+use codex_exec_server::ExecServerEvent;
+use codex_exec_server::ExecServerLaunchCommand;
+use codex_exec_server::InitializeParams;
+use codex_exec_server::InitializeResponse;
+use codex_exec_server::RemoteExecServerConnectArgs;
+use codex_exec_server::spawn_local_exec_server;
+use codex_utils_cargo_bin::cargo_bin;
+use pretty_assertions::assert_eq;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Command;
+use tokio::sync::broadcast;
+use tokio::time::timeout;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_accepts_initialize_over_stdio() -> anyhow::Result<()> {
+    let binary = cargo_bin("codex-exec-server")?;
+    let mut child = Command::new(binary);
+    child.stdin(Stdio::piped());
+    child.stdout(Stdio::piped());
+    child.stderr(Stdio::inherit());
+    let mut child = child.spawn()?;
+
+    let mut stdin = child.stdin.take().expect("stdin");
+    let stdout = child.stdout.take().expect("stdout");
+    let mut stdout = BufReader::new(stdout).lines();
+
+    let initialize = JSONRPCMessage::Request(JSONRPCRequest {
+        id: RequestId::Integer(1),
+        method: "initialize".to_string(),
+        params: Some(serde_json::to_value(InitializeParams {
+            client_name: "exec-server-test".to_string(),
+        })?),
+        trace: None,
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&initialize)?).as_bytes())
+        .await?;
+
+    let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
+    let response_line = response_line.expect("response line");
+    let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
+    let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = response else {
+        panic!("expected initialize response");
+    };
+    assert_eq!(id, RequestId::Integer(1));
+    let initialize_response: InitializeResponse = serde_json::from_value(result)?;
+    assert_eq!(initialize_response.protocol_version, "exec-server.v0");
+
+    let initialized = JSONRPCMessage::Notification(JSONRPCNotification {
+        method: "initialized".to_string(),
+        params: Some(serde_json::json!({})),
+    });
+    stdin
+        .write_all(format!("{}\n", serde_json::to_string(&initialized)?).as_bytes())
+        .await?;
+
+    child.start_kill()?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_client_streams_output_and_accepts_writes() -> anyhow::Result<()> {
+    let mut env = std::collections::HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+
+    let server = spawn_local_exec_server(
+        ExecServerLaunchCommand {
+            program: cargo_bin("codex-exec-server")?,
+            args: Vec::new(),
+        },
+        ExecServerClientConnectOptions {
+            client_name: "exec-server-test".to_string(),
+            initialize_timeout: Duration::from_secs(5),
+        },
+    )
+    .await?;
+
+    let client = server.client();
+    let mut events = client.event_receiver();
+    let response = client
+        .exec(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
+                    .to_string(),
+            ],
+            cwd: std::env::current_dir()?,
+            env,
+            tty: true,
+            arg0: None,
+        })
+        .await?;
+    let process_id = response.process_id;
+
+    let (stream, ready_output) = recv_until_contains(&mut events, &process_id, "ready").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        ready_output.contains("ready"),
+        "expected initial ready output"
+    );
+
+    client.write(&process_id, b"hello\n".to_vec()).await?;
+
+    let (stream, echoed_output) =
+        recv_until_contains(&mut events, &process_id, "echo:hello").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        echoed_output.contains("echo:hello"),
+        "expected echoed output"
+    );
+
+    client.terminate(&process_id).await?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_server_client_connects_over_websocket() -> anyhow::Result<()> {
+    let mut env = std::collections::HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+
+    let binary = cargo_bin("codex-exec-server")?;
+    let mut child = Command::new(binary);
+    child.args(["--listen", "ws://127.0.0.1:0"]);
+    child.stdin(Stdio::null());
+    child.stdout(Stdio::null());
+    child.stderr(Stdio::piped());
+    let mut child = child.spawn()?;
+    let stderr = child.stderr.take().expect("stderr");
+    let mut stderr_lines = BufReader::new(stderr).lines();
+    let websocket_url = read_websocket_url(&mut stderr_lines).await?;
+
+    let client = ExecServerClient::connect_websocket(RemoteExecServerConnectArgs {
+        websocket_url,
+        client_name: "exec-server-test".to_string(),
+        connect_timeout: Duration::from_secs(5),
+        initialize_timeout: Duration::from_secs(5),
+    })
+    .await?;
+
+    let mut events = client.event_receiver();
+    let response = client
+        .exec(ExecParams {
+            process_id: "proc-1".to_string(),
+            argv: vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
+                    .to_string(),
+            ],
+            cwd: std::env::current_dir()?,
+            env,
+            tty: true,
+            arg0: None,
+        })
+        .await?;
+    let process_id = response.process_id;
+
+    let (stream, ready_output) = recv_until_contains(&mut events, &process_id, "ready").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        ready_output.contains("ready"),
+        "expected initial ready output"
+    );
+
+    client.write(&process_id, b"hello\n".to_vec()).await?;
+
+    let (stream, echoed_output) =
+        recv_until_contains(&mut events, &process_id, "echo:hello").await?;
+    assert_eq!(stream, ExecOutputStream::Pty);
+    assert!(
+        echoed_output.contains("echo:hello"),
+        "expected echoed output"
+    );
+
+    client.terminate(&process_id).await?;
+    child.start_kill()?;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn websocket_disconnect_terminates_processes_for_that_connection() -> anyhow::Result<()> {
+    let mut env = std::collections::HashMap::new();
+    if let Some(path) = std::env::var_os("PATH") {
+        env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
+    }
+
+    let marker_path = std::env::temp_dir().join(format!(
+        "codex-exec-server-disconnect-{}-{}",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)?
+            .as_nanos()
+    ));
+    let _ = std::fs::remove_file(&marker_path);
+
+    let binary = cargo_bin("codex-exec-server")?;
+    let mut child = Command::new(binary);
+    child.args(["--listen", "ws://127.0.0.1:0"]);
+    child.stdin(Stdio::null());
+    child.stdout(Stdio::null());
+    child.stderr(Stdio::piped());
+    let mut child = child.spawn()?;
+    let stderr = child.stderr.take().expect("stderr");
+    let mut stderr_lines = BufReader::new(stderr).lines();
+    let websocket_url = read_websocket_url(&mut stderr_lines).await?;
+
+    {
+        let client = ExecServerClient::connect_websocket(RemoteExecServerConnectArgs {
+            websocket_url,
+            client_name: "exec-server-test".to_string(),
+            connect_timeout: Duration::from_secs(5),
+            initialize_timeout: Duration::from_secs(5),
+        })
+        .await?;
+
+        let _response = client
+            .exec(ExecParams {
+                process_id: "proc-1".to_string(),
+                argv: vec![
+                    "bash".to_string(),
+                    "-lc".to_string(),
+                    format!("sleep 2; printf disconnected > {}", marker_path.display()),
+                ],
+                cwd: std::env::current_dir()?,
+                env,
+                tty: false,
+                arg0: None,
+            })
+            .await?;
+    }
+
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    assert!(
+        !marker_path.exists(),
+        "managed process should be terminated when the websocket client disconnects"
+    );
+
+    child.start_kill()?;
+    let _ = std::fs::remove_file(&marker_path);
+    Ok(())
+}
+
+async fn read_websocket_url<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> anyhow::Result<String>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    let line = timeout(Duration::from_secs(5), lines.next_line()).await??;
+    let line = line.context("missing websocket startup banner")?;
+    let websocket_url = line
+        .split_whitespace()
+        .find(|part| part.starts_with("ws://"))
+        .context("missing websocket URL in startup banner")?;
+    Ok(websocket_url.to_string())
+}
+
+async fn recv_until_contains(
+    events: &mut broadcast::Receiver<ExecServerEvent>,
+    process_id: &str,
+    needle: &str,
+) -> anyhow::Result<(ExecOutputStream, String)> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    let mut collected = String::new();
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let event = timeout(remaining, events.recv()).await??;
+        if let ExecServerEvent::OutputDelta(output_event) = event
+            && output_event.process_id == process_id
+        {
+            collected.push_str(&String::from_utf8_lossy(&output_event.chunk.into_inner()));
+            if collected.contains(needle) {
+                return Ok((output_event.stream, collected));
+            }
+        }
+    }
+}

codex-rs/login/src/lib.rs

@@ -16,7 +16,6 @@ pub use server::run_login_server;
 pub use codex_app_server_protocol::AuthMode;
 pub use codex_core::AuthManager;
 pub use codex_core::CodexAuth;
-pub use codex_core::TokenData;
 pub use codex_core::auth::AuthDotJson;
 pub use codex_core::auth::CLIENT_ID;
 pub use codex_core::auth::CODEX_API_KEY_ENV_VAR;
@@ -24,3 +23,4 @@ pub use codex_core::auth::OPENAI_API_KEY_ENV_VAR;
 pub use codex_core::auth::login_with_api_key;
 pub use codex_core::auth::logout;
 pub use codex_core::auth::save_auth;
+pub use codex_core::token_data::TokenData;

codex-rs/login/src/server.rs

@@ -29,12 +29,12 @@ use base64::Engine;
 use chrono::Utc;
 use codex_app_server_protocol::AuthMode;
 use codex_client::build_reqwest_client_with_custom_ca;
-use codex_core::TokenData;
 use codex_core::auth::AuthCredentialsStoreMode;
 use codex_core::auth::AuthDotJson;
 use codex_core::auth::save_auth;
 use codex_core::default_client::originator;
-use codex_core::parse_chatgpt_jwt_claims;
+use codex_core::token_data::TokenData;
+use codex_core::token_data::parse_chatgpt_jwt_claims;
 use rand::RngCore;
 use serde_json::Value as JsonValue;
 use tiny_http::Header;

codex-rs/tui_app_server/src/app.rs

@@ -69,7 +69,6 @@ use codex_core::config::types::ApprovalsReviewer;
 use codex_core::config::types::ModelAvailabilityNuxConfig;
 use codex_core::config_loader::ConfigLayerStackOrdering;
 use codex_core::features::Feature;
-use codex_core::message_history;
 use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
 use codex_core::models_manager::model_presets::HIDE_GPT_5_1_CODEX_MAX_MIGRATION_PROMPT_CONFIG;
 use codex_core::models_manager::model_presets::HIDE_GPT5_1_MIGRATION_PROMPT_CONFIG;
@@ -87,10 +86,10 @@ use codex_protocol::openai_models::ModelUpgrade;
 use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
 use codex_protocol::protocol::AskForApproval;
 use codex_protocol::protocol::FinalOutput;
-use codex_protocol::protocol::GetHistoryEntryResponseEvent;
 use codex_protocol::protocol::ListSkillsResponseEvent;
 #[cfg(test)]
 use codex_protocol::protocol::McpAuthStatus;
+#[cfg(test)]
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::SandboxPolicy;
 use codex_protocol::protocol::SessionSource;
@@ -458,7 +457,6 @@ struct ThreadEventSnapshot {
 enum ThreadBufferedEvent {
     Notification(ServerNotification),
     Request(ServerRequest),
-    HistoryEntryResponse(GetHistoryEntryResponseEvent),
     LegacyWarning(String),
     LegacyRollback { num_turns: u32 },
 }
@@ -618,7 +616,6 @@ impl ThreadEventStore {
                         .pending_interactive_replay
                         .should_replay_snapshot_request(request),
                     ThreadBufferedEvent::Notification(_)
-                    | ThreadBufferedEvent::HistoryEntryResponse(_)
                     | ThreadBufferedEvent::LegacyWarning(_)
                     | ThreadBufferedEvent::LegacyRollback { .. } => true,
                 })
@@ -1766,21 +1763,8 @@ impl App {
             return Ok(());
         };
 
-        self.submit_thread_op(app_server, thread_id, op).await
-    }
-
-    async fn submit_thread_op(
-        &mut self,
-        app_server: &mut AppServerSession,
-        thread_id: ThreadId,
-        op: AppCommand,
-    ) -> Result<()> {
         crate::session_log::log_outbound_op(&op);
 
-        if self.try_handle_local_history_op(thread_id, &op).await? {
-            return Ok(());
-        }
-
         if self
             .try_resolve_app_server_request(app_server, thread_id, &op)
             .await?
@@ -1793,7 +1777,7 @@ impl App {
             .await?
         {
             if ThreadEventStore::op_can_change_pending_replay_state(&op) {
-                self.note_thread_outbound_op(thread_id, &op).await;
+                self.note_active_thread_outbound_op(&op).await;
                 self.refresh_pending_thread_approvals().await;
             }
             return Ok(());
@@ -1871,66 +1855,6 @@ impl App {
         }
     }
 
-    /// Intercept composer-history operations and handle them locally against
-    /// `$CODEX_HOME/history.jsonl`, bypassing the app-server RPC layer.
-    async fn try_handle_local_history_op(
-        &mut self,
-        thread_id: ThreadId,
-        op: &AppCommand,
-    ) -> Result<bool> {
-        match op.view() {
-            AppCommandView::Other(Op::AddToHistory { text }) => {
-                let text = text.clone();
-                let config = self.chat_widget.config_ref().clone();
-                tokio::spawn(async move {
-                    if let Err(err) =
-                        message_history::append_entry(&text, &thread_id, &config).await
-                    {
-                        tracing::warn!(
-                            thread_id = %thread_id,
-                            error = %err,
-                            "failed to append to message history"
-                        );
-                    }
-                });
-                Ok(true)
-            }
-            AppCommandView::Other(Op::GetHistoryEntryRequest { offset, log_id }) => {
-                let offset = *offset;
-                let log_id = *log_id;
-                let config = self.chat_widget.config_ref().clone();
-                let app_event_tx = self.app_event_tx.clone();
-                tokio::spawn(async move {
-                    let entry_opt = tokio::task::spawn_blocking(move || {
-                        message_history::lookup(log_id, offset, &config)
-                    })
-                    .await
-                    .unwrap_or_else(|err| {
-                        tracing::warn!(error = %err, "history lookup task failed");
-                        None
-                    });
-
-                    app_event_tx.send(AppEvent::ThreadHistoryEntryResponse {
-                        thread_id,
-                        event: GetHistoryEntryResponseEvent {
-                            offset,
-                            log_id,
-                            entry: entry_opt.map(|entry| {
-                                codex_protocol::message_history::HistoryEntry {
-                                    conversation_id: entry.session_id,
-                                    ts: entry.ts,
-                                    text: entry.text,
-                                }
-                            }),
-                        },
-                    });
-                });
-                Ok(true)
-            }
-            _ => Ok(false),
-        }
-    }
-
     async fn try_submit_active_thread_op_via_app_server(
         &mut self,
         app_server: &mut AppServerSession,
@@ -2289,50 +2213,6 @@ impl App {
         Ok(())
     }
 
-    async fn enqueue_thread_history_entry_response(
-        &mut self,
-        thread_id: ThreadId,
-        event: GetHistoryEntryResponseEvent,
-    ) -> Result<()> {
-        let (sender, store) = {
-            let channel = self.ensure_thread_channel(thread_id);
-            (channel.sender.clone(), Arc::clone(&channel.store))
-        };
-
-        let should_send = {
-            let mut guard = store.lock().await;
-            guard
-                .buffer
-                .push_back(ThreadBufferedEvent::HistoryEntryResponse(event.clone()));
-            if guard.buffer.len() > guard.capacity
-                && let Some(removed) = guard.buffer.pop_front()
-                && let ThreadBufferedEvent::Request(request) = &removed
-            {
-                guard
-                    .pending_interactive_replay
-                    .note_evicted_server_request(request);
-            }
-            guard.active
-        };
-
-        if should_send {
-            match sender.try_send(ThreadBufferedEvent::HistoryEntryResponse(event)) {
-                Ok(()) => {}
-                Err(TrySendError::Full(event)) => {
-                    tokio::spawn(async move {
-                        if let Err(err) = sender.send(event).await {
-                            tracing::warn!("thread {thread_id} event channel closed: {err}");
-                        }
-                    });
-                }
-                Err(TrySendError::Closed(_)) => {
-                    tracing::warn!("thread {thread_id} event channel closed");
-                }
-            }
-        }
-        Ok(())
-    }
-
     async fn enqueue_thread_legacy_rollback(
         &mut self,
         thread_id: ThreadId,
@@ -2424,10 +2304,6 @@ impl App {
                 ThreadBufferedEvent::Request(request) => {
                     self.enqueue_thread_request(thread_id, request).await?;
                 }
-                ThreadBufferedEvent::HistoryEntryResponse(event) => {
-                    self.enqueue_thread_history_entry_response(thread_id, event)
-                        .await?;
-                }
                 ThreadBufferedEvent::LegacyWarning(message) => {
                     self.enqueue_thread_legacy_warning(thread_id, message)
                         .await?;
@@ -3589,12 +3465,22 @@ impl App {
                 self.submit_active_thread_op(app_server, op.into()).await?;
             }
             AppEvent::SubmitThreadOp { thread_id, op } => {
-                self.submit_thread_op(app_server, thread_id, op.into())
-                    .await?;
-            }
-            AppEvent::ThreadHistoryEntryResponse { thread_id, event } => {
-                self.enqueue_thread_history_entry_response(thread_id, event)
-                    .await?;
+                let app_command: AppCommand = op.into();
+                if self
+                    .try_resolve_app_server_request(app_server, thread_id, &app_command)
+                    .await?
+                {
+                    return Ok(AppRunControl::Continue);
+                }
+                crate::session_log::log_outbound_op(&app_command);
+                tracing::error!(
+                    thread_id = %thread_id,
+                    op = ?app_command,
+                    "unexpected unresolved thread-scoped app command"
+                );
+                self.chat_widget.add_error_message(format!(
+                    "Thread-scoped request is no longer pending for thread {thread_id}."
+                ));
             }
             AppEvent::DiffResult(text) => {
                 // Clear the in-progress state in the bottom pane
@@ -4753,9 +4639,6 @@ impl App {
                 self.chat_widget
                     .handle_server_request(request, /*replay_kind*/ None);
             }
-            ThreadBufferedEvent::HistoryEntryResponse(event) => {
-                self.chat_widget.handle_history_entry_response(event);
-            }
             ThreadBufferedEvent::LegacyWarning(message) => {
                 self.chat_widget.add_warning_message(message);
             }
@@ -4777,9 +4660,6 @@ impl App {
             ThreadBufferedEvent::Request(request) => self
                 .chat_widget
                 .handle_server_request(request, Some(ReplayKind::ThreadSnapshot)),
-            ThreadBufferedEvent::HistoryEntryResponse(event) => {
-                self.chat_widget.handle_history_entry_response(event)
-            }
             ThreadBufferedEvent::LegacyWarning(message) => {
                 self.chat_widget.add_warning_message(message);
             }
@@ -5640,44 +5520,6 @@ mod tests {
             .expect("listener task drop notification should succeed");
     }
 
-    #[tokio::test]
-    async fn history_lookup_response_is_routed_to_requesting_thread() -> Result<()> {
-        let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
-        let thread_id = ThreadId::new();
-
-        let handled = app
-            .try_handle_local_history_op(
-                thread_id,
-                &Op::GetHistoryEntryRequest {
-                    offset: 0,
-                    log_id: 1,
-                }
-                .into(),
-            )
-            .await?;
-
-        assert!(handled);
-
-        let app_event = tokio::time::timeout(Duration::from_secs(1), app_event_rx.recv())
-            .await
-            .expect("history lookup should emit an app event")
-            .expect("app event channel should stay open");
-
-        let AppEvent::ThreadHistoryEntryResponse {
-            thread_id: routed_thread_id,
-            event,
-        } = app_event
-        else {
-            panic!("expected thread-routed history response");
-        };
-        assert_eq!(routed_thread_id, thread_id);
-        assert_eq!(event.offset, 0);
-        assert_eq!(event.log_id, 1);
-        assert!(event.entry.is_none());
-
-        Ok(())
-    }
-
     #[tokio::test]
     async fn enqueue_thread_event_does_not_block_when_channel_full() -> Result<()> {
         let mut app = make_test_app().await;

codex-rs/tui_app_server/src/app_event.rs

@@ -15,7 +15,6 @@ use codex_chatgpt::connectors::AppInfo;
 use codex_file_search::FileMatch;
 use codex_protocol::ThreadId;
 use codex_protocol::openai_models::ModelPreset;
-use codex_protocol::protocol::GetHistoryEntryResponseEvent;
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::RateLimitSnapshot;
 use codex_utils_approval_presets::ApprovalPreset;
@@ -82,12 +81,6 @@ pub(crate) enum AppEvent {
         op: Op,
     },
 
-    /// Deliver a synthetic history lookup response to a specific thread channel.
-    ThreadHistoryEntryResponse {
-        thread_id: ThreadId,
-        event: GetHistoryEntryResponseEvent,
-    },
-
     /// Start a new session.
     NewSession,
 

codex-rs/tui_app_server/src/app_server_session.rs

@@ -54,7 +54,6 @@ use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::TurnSteerParams;
 use codex_app_server_protocol::TurnSteerResponse;
 use codex_core::config::Config;
-use codex_core::message_history;
 use codex_otel::TelemetryAuthMode;
 use codex_protocol::ThreadId;
 use codex_protocol::openai_models::ModelAvailabilityNux;
@@ -278,7 +277,7 @@ impl AppServerSession {
             })
             .await
             .wrap_err("thread/start failed during TUI bootstrap")?;
-        started_thread_from_start_response(response, config).await
+        started_thread_from_start_response(response)
     }
 
     pub(crate) async fn resume_thread(
@@ -292,14 +291,14 @@ impl AppServerSession {
             .request_typed(ClientRequest::ThreadResume {
                 request_id,
                 params: thread_resume_params_from_config(
-                    config.clone(),
+                    config,
                     thread_id,
                     self.thread_params_mode(),
                 ),
             })
             .await
             .wrap_err("thread/resume failed during TUI bootstrap")?;
-        started_thread_from_resume_response(response, &config).await
+        started_thread_from_resume_response(&response)
     }
 
     pub(crate) async fn fork_thread(
@@ -313,14 +312,14 @@ impl AppServerSession {
             .request_typed(ClientRequest::ThreadFork {
                 request_id,
                 params: thread_fork_params_from_config(
-                    config.clone(),
+                    config,
                     thread_id,
                     self.thread_params_mode(),
                 ),
             })
             .await
             .wrap_err("thread/fork failed during TUI bootstrap")?;
-        started_thread_from_fork_response(response, &config).await
+        started_thread_from_fork_response(&response)
     }
 
     fn thread_params_mode(&self) -> ThreadParamsMode {
@@ -844,12 +843,10 @@ fn thread_cwd_from_config(config: &Config, thread_params_mode: ThreadParamsMode)
     }
 }
 
-async fn started_thread_from_start_response(
+fn started_thread_from_start_response(
     response: ThreadStartResponse,
-    config: &Config,
 ) -> Result<AppServerStartedThread> {
-    let session = thread_session_state_from_thread_start_response(&response, config)
-        .await
+    let session = thread_session_state_from_thread_start_response(&response)
         .map_err(color_eyre::eyre::Report::msg)?;
     Ok(AppServerStartedThread {
         session,
@@ -857,77 +854,66 @@ async fn started_thread_from_start_response(
     })
 }
 
-async fn started_thread_from_resume_response(
-    response: ThreadResumeResponse,
-    config: &Config,
-) -> Result<AppServerStartedThread> {
-    let session = thread_session_state_from_thread_resume_response(&response, config)
-        .await
-        .map_err(color_eyre::eyre::Report::msg)?;
-    Ok(AppServerStartedThread {
-        session,
-        turns: response.thread.turns,
-    })
-}
-
-async fn started_thread_from_fork_response(
-    response: ThreadForkResponse,
-    config: &Config,
-) -> Result<AppServerStartedThread> {
-    let session = thread_session_state_from_thread_fork_response(&response, config)
-        .await
-        .map_err(color_eyre::eyre::Report::msg)?;
-    Ok(AppServerStartedThread {
-        session,
-        turns: response.thread.turns,
-    })
-}
-
-async fn thread_session_state_from_thread_start_response(
-    response: &ThreadStartResponse,
-    config: &Config,
-) -> Result<ThreadSessionState, String> {
-    thread_session_state_from_thread_response(
-        &response.thread.id,
-        response.thread.name.clone(),
-        response.thread.path.clone(),
-        response.model.clone(),
-        response.model_provider.clone(),
-        response.service_tier,
-        response.approval_policy.to_core(),
-        response.approvals_reviewer.to_core(),
-        response.sandbox.to_core(),
-        response.cwd.clone(),
-        response.reasoning_effort,
-        config,
-    )
-    .await
-}
-
-async fn thread_session_state_from_thread_resume_response(
+fn started_thread_from_resume_response(
     response: &ThreadResumeResponse,
-    config: &Config,
-) -> Result<ThreadSessionState, String> {
-    thread_session_state_from_thread_response(
-        &response.thread.id,
-        response.thread.name.clone(),
-        response.thread.path.clone(),
-        response.model.clone(),
-        response.model_provider.clone(),
-        response.service_tier,
-        response.approval_policy.to_core(),
-        response.approvals_reviewer.to_core(),
-        response.sandbox.to_core(),
-        response.cwd.clone(),
-        response.reasoning_effort,
-        config,
-    )
-    .await
+) -> Result<AppServerStartedThread> {
+    let session = thread_session_state_from_thread_resume_response(response)
+        .map_err(color_eyre::eyre::Report::msg)?;
+    Ok(AppServerStartedThread {
+        session,
+        turns: response.thread.turns.clone(),
+    })
 }
 
-async fn thread_session_state_from_thread_fork_response(
+fn started_thread_from_fork_response(
+    response: &ThreadForkResponse,
+) -> Result<AppServerStartedThread> {
+    let session = thread_session_state_from_thread_fork_response(response)
+        .map_err(color_eyre::eyre::Report::msg)?;
+    Ok(AppServerStartedThread {
+        session,
+        turns: response.thread.turns.clone(),
+    })
+}
+
+fn thread_session_state_from_thread_start_response(
+    response: &ThreadStartResponse,
+) -> Result<ThreadSessionState, String> {
+    thread_session_state_from_thread_response(
+        &response.thread.id,
+        response.thread.name.clone(),
+        response.thread.path.clone(),
+        response.model.clone(),
+        response.model_provider.clone(),
+        response.service_tier,
+        response.approval_policy.to_core(),
+        response.approvals_reviewer.to_core(),
+        response.sandbox.to_core(),
+        response.cwd.clone(),
+        response.reasoning_effort,
+    )
+}
+
+fn thread_session_state_from_thread_resume_response(
+    response: &ThreadResumeResponse,
+) -> Result<ThreadSessionState, String> {
+    thread_session_state_from_thread_response(
+        &response.thread.id,
+        response.thread.name.clone(),
+        response.thread.path.clone(),
+        response.model.clone(),
+        response.model_provider.clone(),
+        response.service_tier,
+        response.approval_policy.to_core(),
+        response.approvals_reviewer.to_core(),
+        response.sandbox.to_core(),
+        response.cwd.clone(),
+        response.reasoning_effort,
+    )
+}
+
+fn thread_session_state_from_thread_fork_response(
     response: &ThreadForkResponse,
-    config: &Config,
 ) -> Result<ThreadSessionState, String> {
     thread_session_state_from_thread_response(
         &response.thread.id,
@@ -941,9 +927,7 @@ async fn thread_session_state_from_thread_fork_response(
         response.sandbox.to_core(),
         response.cwd.clone(),
         response.reasoning_effort,
-        config,
     )
-    .await
 }
 
 fn review_target_to_app_server(
@@ -969,7 +953,7 @@ fn review_target_to_app_server(
     clippy::too_many_arguments,
     reason = "session mapping keeps explicit fields"
 )]
-async fn thread_session_state_from_thread_response(
+fn thread_session_state_from_thread_response(
     thread_id: &str,
     thread_name: Option<String>,
     rollout_path: Option<PathBuf>,
@@ -981,12 +965,9 @@ async fn thread_session_state_from_thread_response(
     sandbox_policy: SandboxPolicy,
     cwd: PathBuf,
     reasoning_effort: Option<codex_protocol::openai_models::ReasoningEffort>,
-    config: &Config,
 ) -> Result<ThreadSessionState, String> {
     let thread_id = ThreadId::from_string(thread_id)
         .map_err(|err| format!("thread id `{thread_id}` is invalid: {err}"))?;
-    let (history_log_id, history_entry_count) = message_history::history_metadata(config).await;
-    let history_entry_count = u64::try_from(history_entry_count).unwrap_or(u64::MAX);
 
     Ok(ThreadSessionState {
         thread_id,
@@ -1000,8 +981,8 @@ async fn thread_session_state_from_thread_response(
         sandbox_policy,
         cwd,
         reasoning_effort,
-        history_log_id,
-        history_entry_count,
+        history_log_id: 0,
+        history_entry_count: 0,
         network_proxy: None,
         rollout_path,
     })
@@ -1103,10 +1084,8 @@ mod tests {
         assert_eq!(fork.model_provider, None);
     }
 
-    #[tokio::test]
-    async fn resume_response_restores_turns_from_thread_items() {
-        let temp_dir = tempfile::tempdir().expect("tempdir");
-        let config = build_config(&temp_dir).await;
+    #[test]
+    fn resume_response_restores_turns_from_thread_items() {
         let thread_id = ThreadId::new();
         let response = ThreadResumeResponse {
             thread: codex_app_server_protocol::Thread {
@@ -1156,44 +1135,9 @@ mod tests {
             reasoning_effort: None,
         };
 
-        let started = started_thread_from_resume_response(response.clone(), &config)
-            .await
-            .expect("resume response should map");
+        let started =
+            started_thread_from_resume_response(&response).expect("resume response should map");
         assert_eq!(started.turns.len(), 1);
         assert_eq!(started.turns[0], response.thread.turns[0]);
     }
-
-    #[tokio::test]
-    async fn session_configured_populates_history_metadata() {
-        let temp_dir = tempfile::tempdir().expect("tempdir");
-        let config = build_config(&temp_dir).await;
-        let thread_id = ThreadId::new();
-
-        message_history::append_entry("older", &thread_id, &config)
-            .await
-            .expect("history append should succeed");
-        message_history::append_entry("newer", &thread_id, &config)
-            .await
-            .expect("history append should succeed");
-
-        let session = thread_session_state_from_thread_response(
-            &thread_id.to_string(),
-            Some("restore".to_string()),
-            None,
-            "gpt-5.4".to_string(),
-            "openai".to_string(),
-            None,
-            AskForApproval::Never,
-            codex_protocol::config_types::ApprovalsReviewer::User,
-            SandboxPolicy::new_read_only_policy(),
-            PathBuf::from("/tmp/project"),
-            None,
-            &config,
-        )
-        .await
-        .expect("session should map");
-
-        assert_ne!(session.history_log_id, 0);
-        assert_eq!(session.history_entry_count, 2);
-    }
 }

codex-rs/tui_app_server/src/bottom_pane/chat_composer.rs

@@ -740,6 +740,7 @@ impl ChatComposer {
     /// composer rehydrates the entry immediately. This path intentionally routes through
     /// [`Self::apply_history_entry`] so cursor placement remains aligned with keyboard history
     /// recall semantics.
+    #[cfg(test)]
     pub(crate) fn on_history_entry_response(
         &mut self,
         log_id: u64,

codex-rs/tui_app_server/src/bottom_pane/chat_composer_history.rs

@@ -4,9 +4,10 @@ use std::path::PathBuf;
 use crate::app_event::AppEvent;
 use crate::app_event_sender::AppEventSender;
 use crate::bottom_pane::MentionBinding;
+use crate::history_cell;
 use crate::mention_codec::decode_history_mentions;
-use codex_protocol::protocol::Op;
 use codex_protocol::user_input::TextElement;
+use tracing::warn;
 
 /// A composer history entry that can rehydrate draft state.
 #[derive(Debug, Clone, PartialEq)]
@@ -236,6 +237,7 @@ impl ChatComposerHistory {
     }
 
     /// Integrate a GetHistoryEntryResponse event.
+    #[cfg(test)]
     pub fn on_entry_response(
         &mut self,
         log_id: u64,
@@ -278,10 +280,16 @@ impl ChatComposerHistory {
             self.last_history_text = Some(entry.text.clone());
             return Some(entry);
         } else if let Some(log_id) = self.history_log_id {
-            app_event_tx.send(AppEvent::CodexOp(Op::GetHistoryEntryRequest {
-                offset: global_idx,
+            warn!(
                 log_id,
-            }));
+                offset = global_idx,
+                "composer history fetch is unavailable in app-server TUI"
+            );
+            app_event_tx.send(AppEvent::InsertHistoryCell(Box::new(
+                history_cell::new_error_event(
+                    "Composer history fetch: Not available in app-server TUI yet.".to_string(),
+                ),
+            )));
         }
         None
     }
@@ -336,18 +344,17 @@ mod tests {
         assert!(history.should_handle_navigation("", 0));
         assert!(history.navigate_up(&tx).is_none()); // don't replace the text yet
 
-        // Verify that a history lookup request was sent.
+        // Verify that the app-server TUI emits an explicit user-facing stub error instead.
         let event = rx.try_recv().expect("expected AppEvent to be sent");
-        let AppEvent::CodexOp(op) = event else {
+        let AppEvent::InsertHistoryCell(cell) = event else {
             panic!("unexpected event variant");
         };
-        assert_eq!(
-            Op::GetHistoryEntryRequest {
-                log_id: 1,
-                offset: 2,
-            },
-            op
-        );
+        let rendered = cell
+            .display_lines(80)
+            .into_iter()
+            .map(|line| line.to_string())
+            .collect::<String>();
+        assert!(rendered.contains("Composer history fetch: Not available in app-server TUI yet."));
 
         // Inject the async response.
         assert_eq!(
@@ -358,18 +365,17 @@ mod tests {
         // Next Up should move to offset 1.
         assert!(history.navigate_up(&tx).is_none()); // don't replace the text yet
 
-        // Verify second lookup request for offset 1.
+        // Verify second stub error for offset 1.
         let event2 = rx.try_recv().expect("expected second event");
-        let AppEvent::CodexOp(op) = event2 else {
+        let AppEvent::InsertHistoryCell(cell) = event2 else {
             panic!("unexpected event variant");
         };
-        assert_eq!(
-            Op::GetHistoryEntryRequest {
-                log_id: 1,
-                offset: 1,
-            },
-            op
-        );
+        let rendered = cell
+            .display_lines(80)
+            .into_iter()
+            .map(|line| line.to_string())
+            .collect::<String>();
+        assert!(rendered.contains("Composer history fetch: Not available in app-server TUI yet."));
 
         assert_eq!(
             Some(HistoryEntry::new("older".to_string())),

codex-rs/tui_app_server/src/bottom_pane/mod.rs

@@ -1073,6 +1073,7 @@ impl BottomPane {
             || self.composer.is_in_paste_burst()
     }
 
+    #[cfg(test)]
     pub(crate) fn on_history_entry_response(
         &mut self,
         log_id: u64,

codex-rs/tui_app_server/src/chatwidget.rs

@@ -46,8 +46,6 @@ use crate::audio_device::list_realtime_audio_device_names;
 use crate::bottom_pane::StatusLineItem;
 use crate::bottom_pane::StatusLinePreviewData;
 use crate::bottom_pane::StatusLineSetupView;
-use crate::mention_codec::LinkedMention;
-use crate::mention_codec::encode_history_mentions;
 use crate::model_catalog::ModelCatalog;
 use crate::multi_agents;
 use crate::status::RateLimitWindowDisplay;
@@ -3476,7 +3474,8 @@ impl ChatWidget {
         }
     }
 
-    pub(crate) fn handle_history_entry_response(
+    #[cfg(test)]
+    fn on_get_history_entry_response(
         &mut self,
         event: codex_protocol::protocol::GetHistoryEntryResponseEvent,
     ) {
@@ -5317,19 +5316,9 @@ impl ChatWidget {
             return;
         }
 
-        // Persist the text to cross-session message history. Mentions are
-        // encoded into placeholder syntax so recall can reconstruct the
-        // mention bindings in a future session.
+        // Persist the text to cross-session message history.
         if !text.is_empty() {
-            let encoded_mentions = mention_bindings
-                .iter()
-                .map(|binding| LinkedMention {
-                    mention: binding.mention.clone(),
-                    path: binding.path.clone(),
-                })
-                .collect::<Vec<_>>();
-            let history_text = encode_history_mentions(&text, &encoded_mentions);
-            self.submit_op(Op::AddToHistory { text: history_text });
+            warn!("skipping composer history persistence in app-server TUI");
         }
 
         if let Some(pending_steer) = pending_steer {
@@ -6451,7 +6440,7 @@ impl ChatWidget {
             EventMsg::McpToolCallEnd(ev) => self.on_mcp_tool_call_end(ev),
             EventMsg::WebSearchBegin(ev) => self.on_web_search_begin(ev),
             EventMsg::WebSearchEnd(ev) => self.on_web_search_end(ev),
-            EventMsg::GetHistoryEntryResponse(ev) => self.handle_history_entry_response(ev),
+            EventMsg::GetHistoryEntryResponse(ev) => self.on_get_history_entry_response(ev),
             EventMsg::McpListToolsResponse(ev) => self.on_list_mcp_tools(ev),
             EventMsg::ListCustomPromptsResponse(_) => {
                 tracing::warn!(

codex-rs/tui_app_server/src/local_chatgpt_auth.rs

@@ -2,10 +2,8 @@ use std::path::Path;
 
 use codex_app_server_protocol::AuthMode;
 use codex_app_server_protocol::ChatgptAuthTokensRefreshResponse;
-use codex_core::TokenData;
 use codex_core::auth::AuthCredentialsStoreMode;
 use codex_core::auth::load_auth_dot_json;
-use codex_core::parse_chatgpt_jwt_claims;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct LocalChatgptAuth {
@@ -71,10 +69,10 @@ mod tests {
     use base64::Engine;
     use chrono::Utc;
     use codex_app_server_protocol::AuthMode;
-    use codex_core::TokenData;
     use codex_core::auth::AuthDotJson;
     use codex_core::auth::login_with_chatgpt_auth_tokens;
     use codex_core::auth::save_auth;
+    use codex_core::token_data::TokenData;
     use pretty_assertions::assert_eq;
     use serde::Serialize;
     use serde_json::json;
@@ -112,7 +110,8 @@ mod tests {
             auth_mode: Some(AuthMode::Chatgpt),
             openai_api_key: None,
             tokens: Some(TokenData {
-                id_token: parse_chatgpt_jwt_claims(&id_token).expect("id token should parse"),
+                id_token: codex_core::token_data::parse_chatgpt_jwt_claims(&id_token)
+                    .expect("id token should parse"),
                 access_token,
                 refresh_token: "refresh-token".to_string(),
                 account_id: Some("workspace-1".to_string()),