openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into codex/viyatb/pr14718-fix

Browse Source 
# Conflicts:
#	codex-rs/core/src/codex_tests.rs
#	codex-rs/core/src/config/mod.rs
#	codex-rs/core/src/config_loader/mod.rs
This commit is contained in:
viyatb-oai
2026-04-06 12:39:25 -07:00
parent c90928c22a 46b7e4fb2c
commit 36eb261eac
646 changed files with 40689 additions and 25431 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/BUILD.bazel
+1
View File
@@ -3,5 +3,6 @@ load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "app-server",
crate_name = "codex_app_server",
integration_test_timeout = "long",
test_tags = ["no-sandbox"],
)
codex-rs/app-server/Cargo.toml
+5
View File
@@ -29,8 +29,10 @@ axum = { workspace = true, default-features = false, features = [
"tokio",
"ws",
] }
codex-analytics = { workspace = true }
codex-arg0 = { workspace = true }
codex-cloud-requirements = { workspace = true }
codex-config = { workspace = true }
codex-core = { workspace = true }
codex-exec-server = { workspace = true }
codex-features = { workspace = true }
@@ -43,10 +45,13 @@ codex-backend-client = { workspace = true }
codex-file-search = { workspace = true }
codex-chatgpt = { workspace = true }
codex-login = { workspace = true }
codex-mcp = { workspace = true }
codex-models-manager = { workspace = true }
codex-protocol = { workspace = true }
codex-app-server-protocol = { workspace = true }
codex-feedback = { workspace = true }
codex-rmcp-client = { workspace = true }
codex-rollout = { workspace = true }
codex-sandboxing = { workspace = true }
codex-state = { workspace = true }
codex-tools = { workspace = true }
codex-rs/app-server/README.md
+6 -6
View File
@@ -132,9 +132,9 @@ Example with notification opt-out:
## API Overview
- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread.
- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`.
- `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it.
- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread.
- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. The returned `thread.forkedFromId` points at the source thread when known. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread.
- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
- `thread/loaded/list` — list the thread ids currently loaded in memory.
- `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
@@ -273,7 +273,7 @@ Experimental API: `thread/start`, `thread/resume`, and `thread/fork` accept `per
- `modelProviders` — restrict results to specific providers; unset, null, or an empty array will include all providers.
- `sourceKinds` — restrict results to specific sources; omit or pass `[]` for interactive sessions only (`cli`, `vscode`).
- `archived` — when `true`, list archived threads only. When `false` or `null`, list non-archived threads (default).
- `cwd` — restrict results to threads whose session cwd exactly matches this path.
- `cwd` — restrict results to threads whose session cwd exactly matches this path. Relative paths are resolved against the app-server process cwd before matching.
- `searchTerm` — restrict results to threads whose extracted title contains this substring (case-sensitive).
- Responses include `agentNickname` and `agentRole` for AgentControl-spawned thread sub-agents when available.
@@ -913,10 +913,10 @@ All items emit shared lifecycle events:
- `item/started` — emits the full `item` when a new unit of work begins so the UI can render it immediately; the `item.id` in this payload matches the `itemId` used by deltas.
- `item/completed` — sends the final `item` once that work itself finishes (for example, after a tool call or message completes); treat this as the authoritative execution/result state.
- `item/autoApprovalReview/started` — [UNSTABLE] temporary guardian notification carrying `{threadId, turnId, targetItemId, review, action?}` when guardian approval review begins. This shape is expected to change soon.
- `item/autoApprovalReview/completed` — [UNSTABLE] temporary guardian notification carrying `{threadId, turnId, targetItemId, review, action?}` when guardian approval review resolves. This shape is expected to change soon.
- `item/autoApprovalReview/started` — [UNSTABLE] temporary guardian notification carrying `{threadId, turnId, targetItemId, review, action}` when guardian approval review begins. This shape is expected to change soon.
- `item/autoApprovalReview/completed` — [UNSTABLE] temporary guardian notification carrying `{threadId, turnId, targetItemId, review, action}` when guardian approval review resolves. This shape is expected to change soon.
`review` is [UNSTABLE] and currently has `{status, riskScore?, riskLevel?, rationale?}`, where `status` is one of `inProgress`, `approved`, `denied`, or `aborted`. `action` is the guardian action summary payload from core when available and is intended to support temporary standalone pending-review UI. These notifications are separate from the target item's own `item/completed` lifecycle and are intentionally temporary while the guardian app protocol is still being designed.
`review` is [UNSTABLE] and currently has `{status, riskScore?, riskLevel?, rationale?}`, where `status` is one of `inProgress`, `approved`, `denied`, or `aborted`. `action` is a tagged union with `type: "command" | "execve" | "applyPatch" | "networkAccess" | "mcpToolCall"`. Command-like actions include a `source` discriminator (`"shell"` or `"unifiedExec"`). These notifications are separate from the target item's own `item/completed` lifecycle and are intentionally temporary while the guardian app protocol is still being designed.
There are additional item-specific events:
codex-rs/app-server/src/bespoke_event_handling.rs
+599 -227
View File
File diff suppressed because it is too large Load Diff
codex-rs/app-server/src/codex_message_processor.rs
+372 -33
View File
@@ -20,6 +20,7 @@ use crate::thread_status::resolve_thread_status;
use chrono::DateTime;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::Account;
use codex_app_server_protocol::AccountLoginCompletedNotification;
use codex_app_server_protocol::AccountUpdatedNotification;
@@ -28,10 +29,12 @@ use codex_app_server_protocol::AppsListParams;
use codex_app_server_protocol::AppsListResponse;
use codex_app_server_protocol::AskForApproval;
use codex_app_server_protocol::AuthMode;
use codex_app_server_protocol::AuthMode as CoreAuthMode;
use codex_app_server_protocol::CancelLoginAccountParams;
use codex_app_server_protocol::CancelLoginAccountResponse;
use codex_app_server_protocol::CancelLoginAccountStatus;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::CodexErrorInfo as AppServerCodexErrorInfo;
use codex_app_server_protocol::CollaborationModeListParams;
use codex_app_server_protocol::CollaborationModeListResponse;
@@ -179,8 +182,7 @@ use codex_arg0::Arg0DispatchPaths;
use codex_backend_client::Client as BackendClient;
use codex_chatgpt::connectors;
use codex_cloud_requirements::cloud_requirements_loader;
use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_config::types::McpServerTransportConfig;
use codex_core::CodexThread;
use codex_core::Cursor as RolloutCursor;
use codex_core::ForkSnapshot;
@@ -191,23 +193,16 @@ use codex_core::SteerInputError;
use codex_core::ThreadConfigSnapshot;
use codex_core::ThreadManager;
use codex_core::ThreadSortKey as CoreThreadSortKey;
use codex_core::auth::AuthMode as CoreAuthMode;
use codex_core::auth::CLIENT_ID;
use codex_core::auth::login_with_api_key;
use codex_core::config::Config;
use codex_core::config::ConfigOverrides;
use codex_core::config::NetworkProxyAuditMetadata;
use codex_core::config::edit::ConfigEdit;
use codex_core::config::edit::ConfigEditsBuilder;
use codex_core::config::types::McpServerTransportConfig;
use codex_core::config_loader::CloudRequirementsLoadError;
use codex_core::config_loader::CloudRequirementsLoadErrorCode;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_core::config_loader::load_config_layers_state;
use codex_core::default_client::set_default_client_residency_requirement;
use codex_core::error::CodexErr;
use codex_core::error::Result as CodexResult;
use codex_core::exec::ExecCapturePolicy;
use codex_core::exec::ExecExpiration;
use codex_core::exec::ExecParams;
@@ -216,11 +211,6 @@ use codex_core::find_archived_thread_path_by_id_str;
use codex_core::find_thread_name_by_id;
use codex_core::find_thread_names_by_ids;
use codex_core::find_thread_path_by_id_str;
use codex_core::mcp::auth::discover_supported_scopes;
use codex_core::mcp::auth::resolve_oauth_scopes;
use codex_core::mcp::collect_mcp_snapshot;
use codex_core::mcp::group_tools_by_server;
use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_core::parse_cursor;
use codex_core::plugins::MarketplaceError;
use codex_core::plugins::MarketplacePluginSource;
@@ -235,9 +225,6 @@ use codex_core::read_head_for_summary;
use codex_core::read_session_meta_line;
use codex_core::rollout_date_parts;
use codex_core::sandboxing::SandboxPermissions;
use codex_core::state_db::StateDbHandle;
use codex_core::state_db::get_state_db;
use codex_core::state_db::reconcile_rollout;
use codex_core::windows_sandbox::WindowsSandboxLevelExt;
use codex_core::windows_sandbox::WindowsSandboxSetupMode as CoreWindowsSandboxSetupMode;
use codex_core::windows_sandbox::WindowsSandboxSetupRequest;
@@ -246,18 +233,33 @@ use codex_features::Feature;
use codex_features::Stage;
use codex_feedback::CodexFeedback;
use codex_git_utils::git_diff_to_remote;
use codex_git_utils::resolve_root_git_project_for_trust;
use codex_login::AuthManager;
use codex_login::CLIENT_ID;
use codex_login::CodexAuth;
use codex_login::ServerOptions as LoginServerOptions;
use codex_login::ShutdownHandle;
use codex_login::auth::login_with_chatgpt_auth_tokens;
use codex_login::complete_device_code_login;
use codex_login::default_client::set_default_client_residency_requirement;
use codex_login::login_with_api_key;
use codex_login::request_device_code;
use codex_login::run_login_server;
use codex_mcp::mcp::auth::discover_supported_scopes;
use codex_mcp::mcp::auth::resolve_oauth_scopes;
use codex_mcp::mcp::collect_mcp_snapshot;
use codex_mcp::mcp::effective_mcp_servers;
use codex_mcp::mcp::qualified_mcp_tool_name_prefix;
use codex_models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ForcedLoginMethod;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::TrustLevel;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::dynamic_tools::DynamicToolSpec as CoreDynamicToolSpec;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AgentStatus;
@@ -282,6 +284,9 @@ use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
use codex_protocol::user_input::UserInput as CoreInputItem;
use codex_rmcp_client::perform_oauth_login_return_url;
use codex_rollout::state_db::StateDbHandle;
use codex_rollout::state_db::get_state_db;
use codex_rollout::state_db::reconcile_rollout;
use codex_state::StateRuntime;
use codex_state::ThreadMetadata;
use codex_state::ThreadMetadataBuilder;
@@ -403,6 +408,7 @@ pub(crate) struct CodexMessageProcessor {
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
@@ -433,6 +439,8 @@ struct ListenerTaskContext {
thread_manager: Arc<ThreadManager>,
thread_state_manager: ThreadStateManager,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
general_analytics_enabled: bool,
thread_watch_manager: ThreadWatchManager,
fallback_model_provider: String,
codex_home: PathBuf,
@@ -455,6 +463,7 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) thread_manager: Arc<ThreadManager>,
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) analytics_events_client: AnalyticsEventsClient,
pub(crate) arg0_paths: Arg0DispatchPaths,
pub(crate) config: Arc<Config>,
pub(crate) cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
@@ -519,6 +528,7 @@ impl CodexMessageProcessor {
auth_manager,
thread_manager,
outgoing,
analytics_events_client,
arg0_paths,
config,
cli_overrides,
@@ -531,6 +541,7 @@ impl CodexMessageProcessor {
auth_manager,
thread_manager,
outgoing: outgoing.clone(),
analytics_events_client,
arg0_paths,
config,
cli_overrides,
@@ -680,6 +691,7 @@ impl CodexMessageProcessor {
connection_id: ConnectionId,
request: ClientRequest,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
request_context: RequestContext,
) {
let to_connection_request_id = |request_id| ConnectionRequestId {
@@ -696,6 +708,8 @@ impl CodexMessageProcessor {
self.thread_start(
to_connection_request_id(request_id),
params,
app_server_client_name.clone(),
app_server_client_version.clone(),
request_context,
)
.await;
@@ -800,6 +814,7 @@ impl CodexMessageProcessor {
to_connection_request_id(request_id),
params,
app_server_client_name.clone(),
app_server_client_version.clone(),
)
.await;
}
@@ -1015,7 +1030,7 @@ impl CodexMessageProcessor {
&mut self,
params: &LoginApiKeyParams,
) -> std::result::Result<(), JSONRPCErrorError> {
if self.auth_manager.is_external_auth_active() {
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
@@ -1094,7 +1109,7 @@ impl CodexMessageProcessor {
) -> std::result::Result<LoginServerOptions, JSONRPCErrorError> {
let config = self.config.as_ref();
if self.auth_manager.is_external_auth_active() {
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
@@ -1531,7 +1546,7 @@ impl CodexMessageProcessor {
}
async fn refresh_token_if_requested(&self, do_refresh: bool) -> RefreshTokenRequestOutcome {
if self.auth_manager.is_external_auth_active() {
if self.auth_manager.is_external_chatgpt_auth_active() {
return RefreshTokenRequestOutcome::NotAttemptedOrSucceeded;
}
if do_refresh && let Err(err) = self.auth_manager.refresh_token().await {
@@ -2046,6 +2061,8 @@ impl CodexMessageProcessor {
&self,
request_id: ConnectionRequestId,
params: ThreadStartParams,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
request_context: RequestContext,
) {
let ThreadStartParams {
@@ -2086,6 +2103,8 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -2099,6 +2118,8 @@ impl CodexMessageProcessor {
runtime_feature_enablement,
cloud_requirements,
request_id,
app_server_client_name,
app_server_client_version,
config,
typesafe_overrides,
dynamic_tools,
@@ -2172,6 +2193,8 @@ impl CodexMessageProcessor {
runtime_feature_enablement: BTreeMap<String, bool>,
cloud_requirements: CloudRequirementsLoader,
request_id: ConnectionRequestId,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
config_overrides: Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: ConfigOverrides,
dynamic_tools: Option<Vec<ApiDynamicToolSpec>>,
@@ -2180,10 +2203,11 @@ impl CodexMessageProcessor {
experimental_raw_events: bool,
request_trace: Option<W3cTraceContext>,
) {
let config = match derive_config_from_params(
let requested_cwd = typesafe_overrides.cwd.clone();
let mut config = match derive_config_from_params(
&cli_overrides,
config_overrides,
typesafe_overrides,
config_overrides.clone(),
typesafe_overrides.clone(),
&cloud_requirements,
&listener_task_context.codex_home,
&runtime_feature_enablement,
@@ -2201,6 +2225,70 @@ impl CodexMessageProcessor {
}
};
// The user may have requested WorkspaceWrite or DangerFullAccess via
// the command line, though in the process of deriving the Config, it
// could be downgraded to ReadOnly (perhaps there is no sandbox
// available on Windows or the enterprise config disallows it). The cwd
// should still be considered "trusted" in this case.
let requested_sandbox_trusts_project = matches!(
typesafe_overrides.sandbox_mode,
Some(
codex_protocol::config_types::SandboxMode::WorkspaceWrite
| codex_protocol::config_types::SandboxMode::DangerFullAccess
)
);
if requested_cwd.is_some()
&& !config.active_project.is_trusted()
&& (requested_sandbox_trusts_project
|| matches!(
config.permissions.sandbox_policy.get(),
codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { .. }
| codex_protocol::protocol::SandboxPolicy::DangerFullAccess
| codex_protocol::protocol::SandboxPolicy::ExternalSandbox { .. }
))
{
let trust_target = resolve_root_git_project_for_trust(config.cwd.as_path())
.unwrap_or_else(|| config.cwd.to_path_buf());
if let Err(err) = codex_core::config::set_project_trust_level(
&listener_task_context.codex_home,
trust_target.as_path(),
TrustLevel::Trusted,
) {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to persist trusted project state: {err}"),
data: None,
};
listener_task_context
.outgoing
.send_error(request_id, error)
.await;
return;
}
config = match derive_config_from_params(
&cli_overrides,
config_overrides,
typesafe_overrides,
&cloud_requirements,
&listener_task_context.codex_home,
&runtime_feature_enablement,
)
.await
{
Ok(config) => config,
Err(err) => {
let error = config_load_error(&err);
listener_task_context
.outgoing
.send_error(request_id, error)
.await;
return;
}
};
}
let dynamic_tools = dynamic_tools.unwrap_or_default();
let core_dynamic_tools = if dynamic_tools.is_empty() {
Vec::new()
@@ -2253,6 +2341,19 @@ impl CodexMessageProcessor {
session_configured,
..
} = new_conv;
if let Err(error) = Self::set_app_server_client_info(
thread.as_ref(),
app_server_client_name,
app_server_client_version,
)
.await
{
listener_task_context
.outgoing
.send_error(request_id, error)
.await;
return;
}
let config_snapshot = thread
.config_snapshot()
.instrument(tracing::info_span!(
@@ -2318,6 +2419,17 @@ impl CodexMessageProcessor {
sandbox: config_snapshot.sandbox_policy.into(),
reasoning_effort: config_snapshot.reasoning_effort,
};
if listener_task_context.general_analytics_enabled {
listener_task_context
.analytics_events_client
.track_response(
request_id.connection_id.0,
ClientResponse::ThreadStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
listener_task_context
.outgoing
@@ -2642,6 +2754,24 @@ impl CodexMessageProcessor {
if state_db_ctx.is_none() {
state_db_ctx = get_state_db(&self.config).await;
}
if state_db_ctx.is_none() {
match StateRuntime::init(
self.config.sqlite_home.clone(),
self.config.model_provider_id.clone(),
)
.await
{
Ok(ctx) => {
state_db_ctx = Some(ctx);
}
Err(err) => {
warn!(
"failed to initialize state db for thread metadata update at {}: {err}",
self.config.sqlite_home.display()
);
}
}
}
let Some(state_db_ctx) = state_db_ctx else {
self.send_internal_error(
request_id,
@@ -3275,6 +3405,13 @@ impl CodexMessageProcessor {
cwd,
search_term,
} = params;
let cwd = match normalize_thread_list_cwd_filter(cwd) {
Ok(cwd) => cwd,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let requested_page_size = limit
.map(|value| value as usize)
@@ -3293,7 +3430,7 @@ impl CodexMessageProcessor {
model_providers,
source_kinds,
archived: archived.unwrap_or(false),
cwd: cwd.map(PathBuf::from),
cwd,
search_term,
},
)
@@ -3502,6 +3639,11 @@ impl CodexMessageProcessor {
}
build_thread_from_snapshot(thread_uuid, &config_snapshot, loaded_rollout_path)
};
if thread.forked_from_id.is_none()
&& let Some(rollout_path) = rollout_path.as_ref()
{
thread.forked_from_id = forked_from_id_from_rollout(rollout_path).await;
}
self.attach_thread_name(thread_uuid, &mut thread).await;
if include_turns && let Some(rollout_path) = rollout_path.as_ref() {
@@ -3796,6 +3938,15 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadResume {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
}
@@ -4322,7 +4473,12 @@ impl CodexMessageProcessor {
)
.await
{
Ok(summary) => summary_to_thread(summary),
Ok(summary) => {
let mut thread = summary_to_thread(summary);
thread.forked_from_id =
forked_from_id_from_rollout(fork_rollout_path.as_path()).await;
thread
}
Err(err) => {
self.send_internal_error(
request_id,
@@ -4356,6 +4512,14 @@ impl CodexMessageProcessor {
}
};
thread.preview = preview_from_rollout_items(&history_items);
thread.forked_from_id = source_thread_id
.or_else(|| {
history_items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id),
_ => None,
})
})
.map(|id| id.to_string());
if let Err(message) = populate_thread_turns(
&mut thread,
ThreadTurnSource::HistoryItems(&history_items),
@@ -4403,6 +4567,15 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadFork {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
@@ -5000,9 +5173,12 @@ impl CodexMessageProcessor {
return;
}
};
let mcp_config = config.to_mcp_config(self.thread_manager.plugins_manager().as_ref());
let auth = self.auth_manager.auth().await;
tokio::spawn(async move {
Self::list_mcp_server_status_task(outgoing, request, params, config).await;
Self::list_mcp_server_status_task(outgoing, request, params, config, mcp_config, auth)
.await;
});
}
@@ -5011,15 +5187,66 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: ListMcpServerStatusParams,
config: Config,
mcp_config: codex_mcp::mcp::McpConfig,
auth: Option<CodexAuth>,
) {
let snapshot = collect_mcp_snapshot(&config).await;
let snapshot = collect_mcp_snapshot(
&mcp_config,
auth.as_ref(),
request_id.request_id.to_string(),
)
.await;
let tools_by_server = group_tools_by_server(&snapshot.tools);
// Rebuild the tool list per original server name instead of using
// `group_tools_by_server()`: qualified tool names are sanitized for the
// Responses API, so a config key like `some-server` is encoded as the
// `mcp__some_server__` prefix. Matching with the original server name's
// sanitized prefix preserves `/mcp` output for hyphenated names.
let effective_servers = effective_mcp_servers(&mcp_config, auth.as_ref());
let mut sanitized_prefix_counts = HashMap::<String, usize>::new();
for name in effective_servers.keys() {
let prefix = qualified_mcp_tool_name_prefix(name);
*sanitized_prefix_counts.entry(prefix).or_default() += 1;
}
let tools_by_server = effective_servers
.keys()
.map(|name| {
let prefix = qualified_mcp_tool_name_prefix(name);
// If multiple server names normalize to the same prefix, the
// qualified tool namespace is ambiguous (for example
// `some-server` and `some_server` both become
// `mcp__some_server__`). In that case, avoid attributing the
// same tools to multiple servers.
let tools = if sanitized_prefix_counts
.get(&prefix)
.copied()
.unwrap_or_default()
== 1
{
snapshot
.tools
.iter()
.filter_map(|(qualified_name, tool)| {
qualified_name
.strip_prefix(&prefix)
.map(|tool_name| (tool_name.to_string(), tool.clone()))
})
.collect::<HashMap<_, _>>()
} else {
HashMap::new()
};
(name.clone(), tools)
})
.collect::<HashMap<_, _>>();
let mut server_names: Vec<String> = config
.mcp_servers
.keys()
.cloned()
// Include built-in/plugin MCP servers that are present in the
// effective runtime config even when they are not user-declared in
// `config.mcp_servers`.
.chain(effective_servers.keys().cloned())
.chain(snapshot.auth_statuses.keys().cloned())
.chain(snapshot.resources.keys().cloned())
.chain(snapshot.resource_templates.keys().cloned())
@@ -6288,6 +6515,7 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: TurnStartParams,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
) {
if let Err(error) = Self::validate_v2_input_limit(&params.input) {
self.outgoing.send_error(request_id, error).await;
@@ -6300,8 +6528,12 @@ impl CodexMessageProcessor {
return;
}
};
if let Err(error) =
Self::set_app_server_client_name(thread.as_ref(), app_server_client_name).await
if let Err(error) = Self::set_app_server_client_info(
thread.as_ref(),
app_server_client_name,
app_server_client_version,
)
.await
{
self.outgoing.send_error(request_id, error).await;
return;
@@ -6395,16 +6627,17 @@ impl CodexMessageProcessor {
}
}
async fn set_app_server_client_name(
async fn set_app_server_client_info(
thread: &CodexThread,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
) -> Result<(), JSONRPCErrorError> {
thread
.set_app_server_client_name(app_server_client_name)
.set_app_server_client_info(app_server_client_name, app_server_client_version)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to set app server client name: {err}"),
message: format!("failed to set app server client info: {err}"),
data: None,
})
}
@@ -6984,6 +7217,8 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -7071,6 +7306,8 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -7102,6 +7339,8 @@ impl CodexMessageProcessor {
outgoing,
thread_manager,
thread_state_manager,
analytics_events_client: _,
general_analytics_enabled: _,
thread_watch_manager,
fallback_model_provider,
codex_home,
@@ -7536,6 +7775,57 @@ impl CodexMessageProcessor {
}
}
fn normalize_thread_list_cwd_filter(
cwd: Option<String>,
) -> Result<Option<PathBuf>, JSONRPCErrorError> {
let Some(cwd) = cwd else {
return Ok(None);
};
AbsolutePathBuf::relative_to_current_dir(cwd.as_str())
.map(AbsolutePathBuf::into_path_buf)
.map(Some)
.map_err(|err| JSONRPCErrorError {
code: INVALID_PARAMS_ERROR_CODE,
message: format!("invalid thread/list cwd filter `{cwd}`: {err}"),
data: None,
})
}
#[cfg(test)]
mod thread_list_cwd_filter_tests {
use super::normalize_thread_list_cwd_filter;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use std::path::PathBuf;
#[test]
fn normalize_thread_list_cwd_filter_preserves_absolute_paths() {
let cwd = if cfg!(windows) {
String::from(r"C:\srv\repo-b")
} else {
String::from("/srv/repo-b")
};
assert_eq!(
normalize_thread_list_cwd_filter(Some(cwd.clone())).expect("cwd filter should parse"),
Some(PathBuf::from(cwd))
);
}
#[test]
fn normalize_thread_list_cwd_filter_resolves_relative_paths_against_server_cwd()
-> std::io::Result<()> {
let expected = AbsolutePathBuf::relative_to_current_dir("repo-b")?.to_path_buf();
assert_eq!(
normalize_thread_list_cwd_filter(Some(String::from("repo-b")))
.expect("cwd filter should parse"),
Some(expected)
);
Ok(())
}
}
#[allow(clippy::too_many_arguments)]
async fn handle_thread_listener_command(
conversation_id: ThreadId,
@@ -8511,6 +8801,7 @@ async fn load_thread_summary_for_rollout(
rollout_path.display()
)
})?;
thread.forked_from_id = forked_from_id_from_rollout(rollout_path).await;
if let Some(persisted_metadata) = persisted_metadata {
merge_mutable_thread_metadata(
&mut thread,
@@ -8522,6 +8813,14 @@ async fn load_thread_summary_for_rollout(
Ok(thread)
}
async fn forked_from_id_from_rollout(path: &Path) -> Option<String> {
read_session_meta_line(path)
.await
.ok()
.and_then(|meta_line| meta_line.meta.forked_from_id)
.map(|thread_id| thread_id.to_string())
}
fn merge_mutable_thread_metadata(thread: &mut Thread, persisted_thread: Thread) {
thread.git_info = persisted_thread.git_info;
}
@@ -8602,6 +8901,7 @@ fn build_thread_from_snapshot(
let now = time::OffsetDateTime::now_utc().unix_timestamp();
Thread {
id: thread_id.to_string(),
forked_from_id: None,
preview: String::new(),
ephemeral: config_snapshot.ephemeral,
model_provider: config_snapshot.model_provider_id.clone(),
@@ -8644,6 +8944,7 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread {
Thread {
id: conversation_id.to_string(),
forked_from_id: None,
preview,
ephemeral: false,
model_provider,
@@ -9140,6 +9441,44 @@ mod tests {
Ok(())
}
#[tokio::test]
async fn read_summary_from_rollout_preserves_forked_from_id() -> Result<()> {
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMetaLine;
use std::fs;
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let forked_from_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?;
let timestamp = "2025-09-05T16:53:11.850Z".to_string();
let session_meta = SessionMeta {
id: conversation_id,
forked_from_id: Some(forked_from_id),
timestamp: timestamp.clone(),
model_provider: Some("test-provider".to_string()),
..SessionMeta::default()
};
let line = RolloutLine {
timestamp,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: session_meta,
git: None,
}),
};
fs::write(&path, format!("{}\n", serde_json::to_string(&line)?))?;
assert_eq!(
forked_from_id_from_rollout(path.as_path()).await,
Some(forked_from_id.to_string())
);
Ok(())
}
#[tokio::test]
async fn aborting_pending_request_clears_pending_state() -> Result<()> {
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
codex-rs/app-server/src/codex_message_processor/plugin_mcp_oauth.rs
+5 -5
View File
@@ -3,12 +3,12 @@ use std::sync::Arc;
use codex_app_server_protocol::McpServerOauthLoginCompletedNotification;
use codex_app_server_protocol::ServerNotification;
use codex_config::types::McpServerConfig;
use codex_core::config::Config;
use codex_core::config::types::McpServerConfig;
use codex_core::mcp::auth::McpOAuthLoginSupport;
use codex_core::mcp::auth::oauth_login_support;
use codex_core::mcp::auth::resolve_oauth_scopes;
use codex_core::mcp::auth::should_retry_without_scopes;
use codex_mcp::mcp::auth::McpOAuthLoginSupport;
use codex_mcp::mcp::auth::oauth_login_support;
use codex_mcp::mcp::auth::resolve_oauth_scopes;
use codex_mcp::mcp::auth::should_retry_without_scopes;
use codex_rmcp_client::perform_oauth_login_silent;
use tracing::warn;
codex-rs/app-server/src/command_exec.rs
+1 -1
View File
@@ -18,12 +18,12 @@ use codex_app_server_protocol::CommandExecWriteParams;
use codex_app_server_protocol::CommandExecWriteResponse;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::ServerNotification;
use codex_core::bytes_to_string_smart;
use codex_core::config::StartedNetworkProxy;
use codex_core::exec::DEFAULT_EXEC_COMMAND_TIMEOUT_MS;
use codex_core::exec::ExecExpiration;
use codex_core::exec::IO_DRAIN_TIMEOUT_MS;
use codex_core::sandboxing::ExecRequest;
use codex_protocol::exec_output::bytes_to_string_smart;
use codex_sandboxing::SandboxType;
use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP;
use codex_utils_pty::ProcessHandle;
codex-rs/app-server/src/config_api.rs
+24 -4
View File
@@ -1,6 +1,7 @@
use crate::error_code::INTERNAL_ERROR_CODE;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use async_trait::async_trait;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::ConfigBatchWriteParams;
use codex_app_server_protocol::ConfigReadParams;
use codex_app_server_protocol::ConfigReadResponse;
@@ -16,7 +17,6 @@ use codex_app_server_protocol::NetworkDomainPermission;
use codex_app_server_protocol::NetworkRequirements;
use codex_app_server_protocol::NetworkUnixSocketPermission;
use codex_app_server_protocol::SandboxMode;
use codex_core::AnalyticsEventsClient;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::config::ConfigService;
@@ -366,6 +366,12 @@ fn map_requirements_toml_to_api(requirements: ConfigRequirementsToml) -> ConfigR
.map(codex_app_server_protocol::AskForApproval::from)
.collect()
}),
allowed_approvals_reviewers: requirements.allowed_approvals_reviewers.map(|reviewers| {
reviewers
.into_iter()
.map(codex_app_server_protocol::ApprovalsReviewer::from)
.collect()
}),
allowed_sandbox_modes: requirements.allowed_sandbox_modes.map(|modes| {
modes
.into_iter()
@@ -510,15 +516,16 @@ fn config_write_error(code: ConfigWriteErrorCode, message: impl Into<String>) ->
#[cfg(test)]
mod tests {
use super::*;
use codex_core::AnalyticsEventsClient;
use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_analytics::AnalyticsEventsClient;
use codex_core::config_loader::NetworkDomainPermissionToml as CoreNetworkDomainPermissionToml;
use codex_core::config_loader::NetworkDomainPermissionsToml as CoreNetworkDomainPermissionsToml;
use codex_core::config_loader::NetworkRequirementsToml as CoreNetworkRequirementsToml;
use codex_core::config_loader::NetworkUnixSocketPermissionToml as CoreNetworkUnixSocketPermissionToml;
use codex_core::config_loader::NetworkUnixSocketPermissionsToml as CoreNetworkUnixSocketPermissionsToml;
use codex_features::Feature;
use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_protocol::config_types::ApprovalsReviewer as CoreApprovalsReviewer;
use codex_protocol::protocol::AskForApproval as CoreAskForApproval;
use pretty_assertions::assert_eq;
use serde_json::json;
@@ -545,6 +552,10 @@ mod tests {
CoreAskForApproval::Never,
CoreAskForApproval::OnRequest,
]),
allowed_approvals_reviewers: Some(vec![
CoreApprovalsReviewer::User,
CoreApprovalsReviewer::GuardianSubagent,
]),
allowed_sandbox_modes: Some(vec![
CoreSandboxModeRequirement::ReadOnly,
CoreSandboxModeRequirement::ExternalSandbox,
@@ -602,6 +613,13 @@ mod tests {
codex_app_server_protocol::AskForApproval::OnRequest,
])
);
assert_eq!(
mapped.allowed_approvals_reviewers,
Some(vec![
codex_app_server_protocol::ApprovalsReviewer::User,
codex_app_server_protocol::ApprovalsReviewer::GuardianSubagent,
])
);
assert_eq!(
mapped.allowed_sandbox_modes,
Some(vec![SandboxMode::ReadOnly]),
@@ -651,6 +669,7 @@ mod tests {
fn map_requirements_toml_to_api_omits_unix_socket_none_entries_from_legacy_network_fields() {
let requirements = ConfigRequirementsToml {
allowed_approval_policies: None,
allowed_approvals_reviewers: None,
allowed_sandbox_modes: None,
allowed_web_search_modes: None,
guardian_developer_instructions: None,
@@ -707,6 +726,7 @@ mod tests {
fn map_requirements_toml_to_api_normalizes_allowed_web_search_modes() {
let requirements = ConfigRequirementsToml {
allowed_approval_policies: None,
allowed_approvals_reviewers: None,
allowed_sandbox_modes: None,
allowed_web_search_modes: Some(Vec::new()),
guardian_developer_instructions: None,
codex-rs/app-server/src/in_process.rs
+2
View File
@@ -64,6 +64,7 @@ use crate::outgoing_message::QueuedOutgoingMessage;
use crate::transport::CHANNEL_CAPACITY;
use crate::transport::OutboundConnectionState;
use crate::transport::route_outgoing_envelope;
use codex_analytics::AppServerRpcTransport;
use codex_app_server_protocol::ClientNotification;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ConfigWarningNotification;
@@ -393,6 +394,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
config_warnings: args.config_warnings,
session_source: args.session_source,
enable_codex_api_key_env: args.enable_codex_api_key_env,
rpc_transport: AppServerRpcTransport::InProcess,
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut session = ConnectionSessionState::default();
codex-rs/app-server/src/lib.rs
+13 -2
View File
@@ -2,7 +2,6 @@
use codex_arg0::Arg0DispatchPaths;
use codex_cloud_requirements::cloud_requirements_loader;
use codex_core::AuthManager;
use codex_core::config::Config;
use codex_core::config::ConfigBuilder;
use codex_core::config_loader::CloudRequirementsLoader;
@@ -31,6 +30,7 @@ use crate::transport::auth::policy_from_settings;
use crate::transport::route_outgoing_envelope;
use crate::transport::start_stdio_connection;
use crate::transport::start_websocket_acceptor;
use codex_analytics::AppServerRpcTransport;
use codex_app_server_protocol::ConfigLayerSource;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
@@ -42,6 +42,7 @@ use codex_core::config_loader::ConfigLoadError;
use codex_core::config_loader::TextRange as CoreTextRange;
use codex_exec_server::EnvironmentManager;
use codex_feedback::CodexFeedback;
use codex_login::AuthManager;
use codex_protocol::protocol::SessionSource;
use codex_state::log_db;
use tokio::sync::mpsc;
@@ -451,7 +452,9 @@ pub async fn run_main_with_transport(
range: None,
});
}
if let Some(warning) = codex_core::config::system_bwrap_warning() {
if let Some(warning) =
codex_core::config::system_bwrap_warning(config.permissions.sandbox_policy.get())
{
config_warnings.push(ConfigWarningNotification {
summary: warning,
details: None,
@@ -618,6 +621,7 @@ pub async fn run_main_with_transport(
config_warnings,
session_source,
enable_codex_api_key_env: false,
rpc_transport: analytics_rpc_transport(transport),
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
@@ -841,6 +845,13 @@ pub async fn run_main_with_transport(
Ok(())
}
fn analytics_rpc_transport(transport: AppServerTransport) -> AppServerRpcTransport {
match transport {
AppServerTransport::Stdio => AppServerRpcTransport::Stdio,
AppServerTransport::WebSocket { .. } => AppServerRpcTransport::Websocket,
}
}
#[cfg(test)]
mod tests {
use super::LogFormat;
codex-rs/app-server/src/message_processor.rs
+45 -22
View File
@@ -20,7 +20,10 @@ use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::RequestContext;
use crate::transport::AppServerTransport;
use async_trait::async_trait;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppServerRpcTransport;
use codex_app_server_protocol::AppListUpdatedNotification;
use codex_app_server_protocol::AuthMode as LoginAuthMode;
use codex_app_server_protocol::ChatgptAuthTokensRefreshParams;
use codex_app_server_protocol::ChatgptAuthTokensRefreshReason;
use codex_app_server_protocol::ChatgptAuthTokensRefreshResponse;
@@ -55,25 +58,24 @@ use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::experimental_required_message;
use codex_arg0::Arg0DispatchPaths;
use codex_chatgpt::connectors;
use codex_core::AnalyticsEventsClient;
use codex_core::AuthManager;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_core::default_client::SetOriginatorError;
use codex_core::default_client::USER_AGENT_SUFFIX;
use codex_core::default_client::get_codex_user_agent;
use codex_core::default_client::set_default_client_residency_requirement;
use codex_core::default_client::set_default_originator;
use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_exec_server::EnvironmentManager;
use codex_features::Feature;
use codex_feedback::CodexFeedback;
use codex_login::AuthManager;
use codex_login::auth::ExternalAuth;
use codex_login::auth::ExternalAuthRefreshContext;
use codex_login::auth::ExternalAuthRefreshReason;
use codex_login::auth::ExternalAuthRefresher;
use codex_login::auth::ExternalAuthTokens;
use codex_login::default_client::SetOriginatorError;
use codex_login::default_client::USER_AGENT_SUFFIX;
use codex_login::default_client::get_codex_user_agent;
use codex_login::default_client::set_default_client_residency_requirement;
use codex_login::default_client::set_default_originator;
use codex_models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::W3cTraceContext;
@@ -102,7 +104,11 @@ impl ExternalAuthRefreshBridge {
}
#[async_trait]
impl ExternalAuthRefresher for ExternalAuthRefreshBridge {
impl ExternalAuth for ExternalAuthRefreshBridge {
fn auth_mode(&self) -> LoginAuthMode {
LoginAuthMode::Chatgpt
}
async fn refresh(
&self,
context: ExternalAuthRefreshContext,
@@ -144,11 +150,11 @@ impl ExternalAuthRefresher for ExternalAuthRefreshBridge {
let response: ChatgptAuthTokensRefreshResponse =
serde_json::from_value(result).map_err(std::io::Error::other)?;
Ok(ExternalAuthTokens {
access_token: response.access_token,
chatgpt_account_id: response.chatgpt_account_id,
chatgpt_plan_type: response.chatgpt_plan_type,
})
Ok(ExternalAuthTokens::chatgpt(
response.access_token,
response.chatgpt_account_id,
response.chatgpt_plan_type,
))
}
}
@@ -159,9 +165,11 @@ pub(crate) struct MessageProcessor {
external_agent_config_api: ExternalAgentConfigApi,
fs_api: FsApi,
auth_manager: Arc<AuthManager>,
analytics_events_client: AnalyticsEventsClient,
fs_watch_manager: FsWatchManager,
config: Arc<Config>,
config_warnings: Arc<Vec<ConfigWarningNotification>>,
rpc_transport: AppServerRpcTransport,
}
#[derive(Clone, Debug, Default)]
@@ -186,6 +194,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) config_warnings: Vec<ConfigWarningNotification>,
pub(crate) session_source: SessionSource,
pub(crate) enable_codex_api_key_env: bool,
pub(crate) rpc_transport: AppServerRpcTransport,
}
impl MessageProcessor {
@@ -205,11 +214,15 @@ impl MessageProcessor {
config_warnings,
session_source,
enable_codex_api_key_env,
rpc_transport,
} = args;
let auth_manager = AuthManager::shared(
let auth_manager = AuthManager::shared_with_external_auth(
config.codex_home.clone(),
enable_codex_api_key_env,
config.cli_auth_credentials_store_mode,
Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(),
}),
);
let thread_manager = Arc::new(ThreadManager::new(
config.as_ref(),
@@ -223,9 +236,6 @@ impl MessageProcessor {
environment_manager,
));
auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone());
auth_manager.set_external_auth_refresher(Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(),
}));
let analytics_events_client = AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
@@ -242,6 +252,7 @@ impl MessageProcessor {
auth_manager: auth_manager.clone(),
thread_manager: Arc::clone(&thread_manager),
outgoing: outgoing.clone(),
analytics_events_client: analytics_events_client.clone(),
arg0_paths,
config: Arc::clone(&config),
cli_overrides: cli_overrides.clone(),
@@ -262,7 +273,7 @@ impl MessageProcessor {
loader_overrides,
cloud_requirements,
thread_manager,
analytics_events_client,
analytics_events_client.clone(),
);
let external_agent_config_api = ExternalAgentConfigApi::new(config.codex_home.clone());
let fs_api = FsApi::default();
@@ -275,14 +286,16 @@ impl MessageProcessor {
external_agent_config_api,
fs_api,
auth_manager,
analytics_events_client,
fs_watch_manager,
config,
config_warnings: Arc::new(config_warnings),
rpc_transport,
}
}
pub(crate) fn clear_runtime_references(&self) {
self.auth_manager.clear_external_auth_refresher();
self.auth_manager.clear_external_auth();
}
pub(crate) async fn process_request(
@@ -547,6 +560,7 @@ impl MessageProcessor {
// shared thread when another connected client did not opt into
// experimental API). Proposed direction is instance-global first-write-wins
// with initialize-time mismatch rejection.
let analytics_initialize_params = params.clone();
let (experimental_api_enabled, opt_out_notification_methods) =
match params.capabilities {
Some(capabilities) => (
@@ -568,7 +582,7 @@ impl MessageProcessor {
session.app_server_client_name = Some(name.clone());
session.client_version = Some(version.clone());
let originator = name.clone();
if let Err(error) = set_default_originator(originator) {
if let Err(error) = set_default_originator(originator.clone()) {
match error {
SetOriginatorError::InvalidHeaderValue => {
let error = JSONRPCErrorError {
@@ -591,6 +605,14 @@ impl MessageProcessor {
}
}
}
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_initialize(
connection_id.0,
analytics_initialize_params,
originator,
self.rpc_transport,
);
}
set_default_client_residency_requirement(self.config.enforce_residency.value());
let user_agent_suffix = format!("{name}; {version}");
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
@@ -828,6 +850,7 @@ impl MessageProcessor {
connection_id,
other,
session.app_server_client_name.clone(),
session.client_version.clone(),
request_context,
)
.boxed()
codex-rs/app-server/src/message_processor/tracing_tests.rs
+2
View File
@@ -7,6 +7,7 @@ use crate::transport::AppServerTransport;
use anyhow::Result;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::write_mock_responses_config_toml;
use codex_analytics::AppServerRpcTransport;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::InitializeCapabilities;
@@ -246,6 +247,7 @@ fn build_test_processor(
config_warnings: Vec::new(),
session_source: SessionSource::VSCode,
enable_codex_api_key_env: false,
rpc_transport: AppServerRpcTransport::Stdio,
});
(processor, outgoing_rx)
}
codex-rs/app-server/src/models.rs
+1 -1
View File
@@ -4,7 +4,7 @@ use codex_app_server_protocol::Model;
use codex_app_server_protocol::ModelUpgradeInfo;
use codex_app_server_protocol::ReasoningEffortOption;
use codex_core::ThreadManager;
use codex_core::models_manager::manager::RefreshStrategy;
use codex_models_manager::manager::RefreshStrategy;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::openai_models::ReasoningEffortPreset;
codex-rs/app-server/src/thread_state.rs
+33
View File
@@ -16,6 +16,7 @@ use std::sync::Weak;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tracing::error;
type PendingInterruptQueue = Vec<(
ConnectionRequestId,
@@ -116,6 +117,38 @@ impl ThreadState {
}
}
pub(crate) async fn resolve_server_request_on_thread_listener(
thread_state: &Arc<Mutex<ThreadState>>,
request_id: RequestId,
) {
let (completion_tx, completion_rx) = oneshot::channel();
let listener_command_tx = {
let state = thread_state.lock().await;
state.listener_command_tx()
};
let Some(listener_command_tx) = listener_command_tx else {
error!("failed to remove pending client request: thread listener is not running");
return;
};
if listener_command_tx
.send(ThreadListenerCommand::ResolveServerRequest {
request_id,
completion_tx,
})
.is_err()
{
error!(
"failed to remove pending client request: thread listener command channel is closed"
);
return;
}
if let Err(err) = completion_rx.await {
error!("failed to remove pending client request: {err}");
}
}
struct ThreadEntry {
state: Arc<Mutex<ThreadState>>,
connection_ids: HashSet<ConnectionId>,
codex-rs/app-server/src/thread_status.rs
+1
View File
@@ -792,6 +792,7 @@ mod tests {
fn test_thread(thread_id: &str, source: codex_app_server_protocol::SessionSource) -> Thread {
Thread {
id: thread_id.to_string(),
forked_from_id: None,
preview: String::new(),
ephemeral: false,
model_provider: "mock-provider".to_string(),
codex-rs/app-server/tests/common/Cargo.toml
+4
View File
@@ -7,6 +7,9 @@ license.workspace = true
[lib]
path = "lib.rs"
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
base64 = { workspace = true }
@@ -15,6 +18,7 @@ codex-app-server-protocol = { workspace = true }
codex-core = { workspace = true }
codex-features = { workspace = true }
codex-login = { workspace = true }
codex-models-manager = { workspace = true }
codex-protocol = { workspace = true }
codex-utils-cargo-bin = { workspace = true }
serde = { workspace = true }
codex-rs/app-server/tests/common/auth_fixtures.rs
+3 -3
View File
@@ -7,9 +7,9 @@ use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use chrono::DateTime;
use chrono::Utc;
use codex_app_server_protocol::AuthMode;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::auth::AuthDotJson;
use codex_core::auth::save_auth;
use codex_login::AuthCredentialsStoreMode;
use codex_login::AuthDotJson;
use codex_login::save_auth;
use codex_login::token_data::TokenData;
use codex_login::token_data::parse_chatgpt_jwt_claims;
use serde_json::json;
codex-rs/app-server/tests/common/mcp_process.rs
+11 -1
View File
@@ -45,6 +45,7 @@ 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::ListMcpServerStatusParams;
use codex_app_server_protocol::LoginAccountParams;
use codex_app_server_protocol::MockExperimentalMethodParams;
use codex_app_server_protocol::ModelListParams;
@@ -79,7 +80,7 @@ use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnSteerParams;
use codex_app_server_protocol::WindowsSandboxSetupStartParams;
use codex_core::default_client::CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR;
use codex_login::default_client::CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR;
use tokio::process::Command;
pub struct McpProcess {
@@ -526,6 +527,15 @@ impl McpProcess {
self.send_request("plugin/read", params).await
}
/// Send an `mcpServerStatus/list` JSON-RPC request.
pub async fn send_list_mcp_server_status_request(
&mut self,
params: ListMcpServerStatusParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("mcpServerStatus/list", params).await
}
/// Send a JSON-RPC request with raw params for protocol-level validation tests.
pub async fn send_raw_request(
&mut self,
codex-rs/app-server/tests/common/models_cache.rs
+3 -2
View File
@@ -1,6 +1,7 @@
use chrono::DateTime;
use chrono::Utc;
use codex_core::test_support::all_model_presets;
use codex_models_manager::client_version_to_whole;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::openai_models::ConfigShellToolType;
use codex_protocol::openai_models::ModelInfo;
@@ -27,7 +28,7 @@ fn preset_to_info(preset: &ModelPreset, priority: i32) -> ModelInfo {
},
supported_in_api: preset.supported_in_api,
priority,
upgrade: preset.upgrade.as_ref().map(|u| u.into()),
upgrade: preset.upgrade.as_ref().map(Into::into),
base_instructions: "base instructions".to_string(),
model_messages: None,
supports_reasoning_summaries: false,
@@ -84,7 +85,7 @@ pub fn write_models_cache_with_models(
let cache_path = codex_home.join("models_cache.json");
// DateTime<Utc> serializes to RFC3339 format by default with serde
let fetched_at: DateTime<Utc> = Utc::now();
let client_version = codex_core::models_manager::client_version_to_whole();
let client_version = client_version_to_whole();
let cache = json!({
"fetched_at": fetched_at,
"etag": null,
codex-rs/app-server/tests/suite/auth.rs
+2 -2
View File
@@ -12,8 +12,8 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::LoginAccountResponse;
use codex_app_server_protocol::RequestId;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_login::AuthCredentialsStoreMode;
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use pretty_assertions::assert_eq;
use std::path::Path;
use tempfile::TempDir;
codex-rs/app-server/tests/suite/fuzzy_file_search.rs
+24
View File
@@ -366,6 +366,30 @@ async fn test_fuzzy_file_search_session_streams_updates() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_fuzzy_file_search_session_update_is_case_insensitive() -> Result<()> {
let codex_home = TempDir::new()?;
let root = TempDir::new()?;
std::fs::write(root.path().join("alpha.txt"), "contents")?;
let mut mcp = initialized_mcp(&codex_home).await?;
let root_path = root.path().to_string_lossy().to_string();
let session_id = "session-case-insensitive";
mcp.start_fuzzy_file_search_session(session_id, vec![root_path.clone()])
.await?;
mcp.update_fuzzy_file_search_session(session_id, "ALP")
.await?;
let payload =
wait_for_session_updated(&mut mcp, session_id, "ALP", FileExpectation::NonEmpty).await?;
assert_eq!(payload.files.len(), 1);
assert_eq!(payload.files[0].root, root_path);
assert_eq!(payload.files[0].path, "alpha.txt");
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_fuzzy_file_search_session_no_updates_after_complete_until_query_edited() -> Result<()>
{
codex-rs/app-server/tests/suite/v2/account.rs
+1 -1
View File
@@ -28,7 +28,7 @@ use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStatus;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use codex_login::login_with_api_key;
use codex_protocol::account::PlanType as AccountPlanType;
use core_test_support::responses;
codex-rs/app-server/tests/suite/v2/analytics.rs
+103 -2
View File
@@ -1,10 +1,23 @@
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::DEFAULT_CLIENT_NAME;
use app_test_support::write_chatgpt_auth;
use codex_config::types::OtelExporterKind;
use codex_config::types::OtelHttpProtocol;
use codex_core::config::ConfigBuilder;
use codex_core::config::types::OtelExporterKind;
use codex_core::config::types::OtelHttpProtocol;
use codex_login::AuthCredentialsStoreMode;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::HashMap;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
const SERVICE_VERSION: &str = "0.0.0-test";
@@ -65,3 +78,91 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
assert_eq!(has_metrics, true);
Ok(())
}
pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
Mock::given(method("POST"))
.and(path("/codex/analytics-events/events"))
.respond_with(ResponseTemplate::new(200))
.mount(server)
.await;
write_chatgpt_auth(
codex_home,
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
Ok(())
}
pub(crate) async fn wait_for_analytics_payload(
server: &MockServer,
read_timeout: Duration,
) -> Result<Value> {
let body = timeout(read_timeout, async {
loop {
let Some(requests) = server.received_requests().await else {
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
};
if let Some(request) = requests.iter().find(|request| {
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
}) {
break request.body.clone();
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await?;
serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))
}
pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> {
let events = payload["events"]
.as_array()
.ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?;
events
.iter()
.find(|event| event["event_type"] == "codex_thread_initialized")
.ok_or_else(|| anyhow::anyhow!("codex_thread_initialized event should be present"))
}
pub(crate) fn assert_basic_thread_initialized_event(
event: &Value,
thread_id: &str,
expected_model: &str,
initialization_mode: &str,
) {
assert_eq!(event["event_params"]["thread_id"], thread_id);
assert_eq!(
event["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_CLIENT_NAME
);
assert_eq!(
event["event_params"]["app_server_client"]["client_name"],
DEFAULT_CLIENT_NAME
);
assert_eq!(
event["event_params"]["app_server_client"]["rpc_transport"],
"stdio"
);
assert_eq!(event["event_params"]["model"], expected_model);
assert_eq!(event["event_params"]["ephemeral"], false);
assert_eq!(event["event_params"]["thread_source"], "user");
assert_eq!(
event["event_params"]["subagent_source"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["parent_thread_id"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["initialization_mode"],
initialization_mode
);
assert!(event["event_params"]["created_at"].as_u64().is_some());
}
codex-rs/app-server/tests/suite/v2/app_list.rs
+3 -3
View File
@@ -35,9 +35,9 @@ use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::auth::AuthDotJson;
use codex_core::auth::save_auth;
use codex_login::AuthCredentialsStoreMode;
use codex_login::AuthDotJson;
use codex_login::save_auth;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
use rmcp::model::JsonObject;
codex-rs/app-server/tests/suite/v2/compaction.rs
+1 -1
View File
@@ -28,7 +28,7 @@ use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use core_test_support::responses;
codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs
+1 -1
View File
@@ -31,7 +31,7 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
codex-rs/app-server/tests/suite/v2/mcp_server_status.rs
+217
View File
@@ -0,0 +1,217 @@
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::to_response;
use app_test_support::write_mock_responses_config_toml;
use axum::Router;
use codex_app_server_protocol::ListMcpServerStatusParams;
use codex_app_server_protocol::ListMcpServerStatusResponse;
use codex_app_server_protocol::RequestId;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
use rmcp::model::JsonObject;
use rmcp::model::ListToolsResult;
use rmcp::model::ServerCapabilities;
use rmcp::model::ServerInfo;
use rmcp::model::Tool;
use rmcp::model::ToolAnnotations;
use rmcp::transport::StreamableHttpServerConfig;
use rmcp::transport::StreamableHttpService;
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use serde_json::json;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::task::JoinHandle;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
#[tokio::test]
async fn mcp_server_status_list_returns_tools_for_hyphenated_server_names() -> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let (mcp_server_url, mcp_server_handle) = start_mcp_server("lookup").await?;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml(
codex_home.path(),
&server.uri(),
&BTreeMap::new(),
/*auto_compact_limit*/ 1024,
/*requires_openai_auth*/ None,
"mock_provider",
"compact",
)?;
let config_path = codex_home.path().join("config.toml");
let mut config_toml = std::fs::read_to_string(&config_path)?;
config_toml.push_str(&format!(
r#"
[mcp_servers.some-server]
url = "{mcp_server_url}/mcp"
"#
));
std::fs::write(config_path, config_toml)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
cursor: None,
limit: None,
})
.await?;
let response = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let response: ListMcpServerStatusResponse = to_response(response)?;
assert_eq!(response.next_cursor, None);
assert_eq!(response.data.len(), 1);
let status = &response.data[0];
assert_eq!(status.name, "some-server");
assert_eq!(
status.tools.keys().cloned().collect::<BTreeSet<_>>(),
BTreeSet::from(["lookup".to_string()])
);
mcp_server_handle.abort();
let _ = mcp_server_handle.await;
Ok(())
}
#[derive(Clone)]
struct McpStatusServer {
tool_name: Arc<String>,
}
impl ServerHandler for McpStatusServer {
fn get_info(&self) -> ServerInfo {
ServerInfo {
capabilities: ServerCapabilities::builder().enable_tools().build(),
..ServerInfo::default()
}
}
async fn list_tools(
&self,
_request: Option<rmcp::model::PaginatedRequestParams>,
_context: rmcp::service::RequestContext<rmcp::service::RoleServer>,
) -> Result<ListToolsResult, rmcp::ErrorData> {
let input_schema: JsonObject = serde_json::from_value(json!({
"type": "object",
"additionalProperties": false
}))
.map_err(|err| rmcp::ErrorData::internal_error(err.to_string(), None))?;
let mut tool = Tool::new(
Cow::Owned(self.tool_name.as_ref().clone()),
Cow::Borrowed("Look up test data."),
Arc::new(input_schema),
);
tool.annotations = Some(ToolAnnotations::new().read_only(true));
Ok(ListToolsResult {
tools: vec![tool],
next_cursor: None,
meta: None,
})
}
}
#[tokio::test]
async fn mcp_server_status_list_does_not_duplicate_tools_for_sanitized_name_collisions()
-> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let (dash_server_url, dash_server_handle) = start_mcp_server("dash_lookup").await?;
let (underscore_server_url, underscore_server_handle) =
start_mcp_server("underscore_lookup").await?;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml(
codex_home.path(),
&server.uri(),
&BTreeMap::new(),
/*auto_compact_limit*/ 1024,
/*requires_openai_auth*/ None,
"mock_provider",
"compact",
)?;
let config_path = codex_home.path().join("config.toml");
let mut config_toml = std::fs::read_to_string(&config_path)?;
config_toml.push_str(&format!(
r#"
[mcp_servers.some-server]
url = "{dash_server_url}/mcp"
[mcp_servers.some_server]
url = "{underscore_server_url}/mcp"
"#
));
std::fs::write(config_path, config_toml)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
cursor: None,
limit: None,
})
.await?;
let response = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let response: ListMcpServerStatusResponse = to_response(response)?;
assert_eq!(response.next_cursor, None);
assert_eq!(response.data.len(), 2);
let status_tools = response
.data
.iter()
.map(|status| (status.name.as_str(), status.tools.keys().count()))
.collect::<BTreeMap<_, _>>();
assert_eq!(
status_tools,
BTreeMap::from([("some-server", 0), ("some_server", 0)])
);
dash_server_handle.abort();
let _ = dash_server_handle.await;
underscore_server_handle.abort();
let _ = underscore_server_handle.await;
Ok(())
}
async fn start_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
let tool_name = Arc::new(tool_name.to_string());
let mcp_service = StreamableHttpService::new(
move || {
Ok(McpStatusServer {
tool_name: Arc::clone(&tool_name),
})
},
Arc::new(LocalSessionManager::default()),
StreamableHttpServerConfig::default(),
);
let router = Router::new().nest_service("/mcp", mcp_service);
let handle = tokio::spawn(async move {
let _ = axum::serve(listener, router).await;
});
Ok((format!("http://{addr}"), handle))
}
codex-rs/app-server/tests/suite/v2/mod.rs
+1
View File
@@ -15,6 +15,7 @@ mod experimental_feature_list;
mod fs;
mod initialize;
mod mcp_server_elicitation;
mod mcp_server_status;
mod model_list;
mod output_schema;
mod plan_item;
codex-rs/app-server/tests/suite/v2/plugin_install.rs
+1 -1
View File
@@ -25,7 +25,7 @@ use codex_app_server_protocol::PluginAuthPolicy;
use codex_app_server_protocol::PluginInstallParams;
use codex_app_server_protocol::PluginInstallResponse;
use codex_app_server_protocol::RequestId;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
codex-rs/app-server/tests/suite/v2/plugin_list.rs
+1 -1
View File
@@ -15,8 +15,8 @@ use codex_app_server_protocol::PluginMarketplaceEntry;
use codex_app_server_protocol::PluginSource;
use codex_app_server_protocol::PluginSummary;
use codex_app_server_protocol::RequestId;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::config::set_project_trust_level;
use codex_login::AuthCredentialsStoreMode;
use codex_protocol::config_types::TrustLevel;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
codex-rs/app-server/tests/suite/v2/plugin_read.rs
+1 -1
View File
@@ -23,7 +23,7 @@ use codex_app_server_protocol::PluginInstallPolicy;
use codex_app_server_protocol::PluginReadParams;
use codex_app_server_protocol::PluginReadResponse;
use codex_app_server_protocol::RequestId;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs
+1 -1
View File
@@ -11,7 +11,7 @@ use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::PluginUninstallParams;
use codex_app_server_protocol::PluginUninstallResponse;
use codex_app_server_protocol::RequestId;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
codex-rs/app-server/tests/suite/v2/rate_limits.rs
+1 -1
View File
@@ -10,7 +10,7 @@ use codex_app_server_protocol::LoginAccountResponse;
use codex_app_server_protocol::RateLimitSnapshot;
use codex_app_server_protocol::RateLimitWindow;
use codex_app_server_protocol::RequestId;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_login::AuthCredentialsStoreMode;
use codex_protocol::account::PlanType as AccountPlanType;
use pretty_assertions::assert_eq;
use serde_json::json;
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+62 -2
View File
@@ -24,8 +24,8 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_login::AuthCredentialsStoreMode;
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
@@ -38,6 +38,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -107,6 +112,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
);
assert_ne!(thread.id, conversation_id);
assert_eq!(thread.forked_from_id, Some(conversation_id.clone()));
assert_eq!(thread.preview, preview);
assert_eq!(thread.model_provider, "mock_provider");
assert_eq!(thread.status, ThreadStatus::Idle);
@@ -177,6 +183,50 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "forked");
Ok(())
}
#[tokio::test]
async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -249,6 +299,7 @@ async fn thread_fork_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -509,7 +560,13 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
String::new()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -522,6 +579,9 @@ chatgpt_base_url = "{chatgpt_base_url}"
model_provider = "mock_provider"
[features]
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs
+54 -1
View File
@@ -19,10 +19,10 @@ use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_core::ARCHIVED_SESSIONS_SUBDIR;
use codex_core::state_db::reconcile_rollout;
use codex_git_utils::GitSha;
use codex_protocol::ThreadId;
use codex_protocol::protocol::GitInfo as RolloutGitInfo;
use codex_rollout::state_db::reconcile_rollout;
use codex_state::StateRuntime;
use pretty_assertions::assert_eq;
use serde_json::Value;
@@ -223,6 +223,59 @@ async fn thread_metadata_update_repairs_missing_sqlite_row_for_stored_thread() -
Ok(())
}
#[tokio::test]
async fn thread_metadata_update_repairs_stored_thread_before_backfill_completes() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let _state_db =
StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()).await?;
let preview = "Stored thread preview before backfill";
let thread_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-30-00",
"2025-01-05T12:30:00Z",
preview,
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread_id.clone(),
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: Some(Some("feature/pending-backfill".to_string())),
origin_url: None,
}),
})
.await?;
let update_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(update_id)),
)
.await??;
let ThreadMetadataUpdateResponse { thread: updated } =
to_response::<ThreadMetadataUpdateResponse>(update_resp)?;
assert_eq!(updated.id, thread_id);
assert_eq!(updated.preview, preview);
assert_eq!(
updated.git_info,
Some(GitInfo {
sha: None,
branch: Some("feature/pending-backfill".to_string()),
origin_url: None,
})
);
Ok(())
}
#[tokio::test]
async fn thread_metadata_update_repairs_loaded_thread_without_resetting_summary() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/app-server/tests/suite/v2/thread_read.rs
+52
View File
@@ -7,6 +7,8 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
@@ -152,6 +154,56 @@ async fn thread_read_can_include_turns() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_read_returns_forked_from_id_for_forked_threads() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
vec![],
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse { thread: forked, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let read_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: forked.id,
include_turns: false,
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.forked_from_id, Some(conversation_id));
Ok(())
}
#[tokio::test]
async fn thread_read_loaded_thread_returns_precomputed_path_before_materialization() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+61 -2
View File
@@ -38,8 +38,8 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_login::AuthCredentialsStoreMode;
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_protocol::ThreadId;
use codex_protocol::config_types::Personality;
use codex_protocol::models::ContentItem;
@@ -70,6 +70,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
@@ -150,6 +155,51 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Vec::new(),
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "gpt-5.2-codex", "resumed");
Ok(())
}
#[tokio::test]
async fn thread_resume_returns_rollout_history() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -1449,6 +1499,7 @@ async fn thread_resume_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -1827,6 +1878,7 @@ model_provider = "mock_provider"
[features]
personality = true
general_analytics = true
[model_providers.mock_provider]
name = "Mock provider for test"
@@ -1843,7 +1895,13 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &std::path::Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
String::new()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -1858,6 +1916,7 @@ model_provider = "mock_provider"
[features]
personality = true
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
codex-rs/app-server/tests/suite/v2/thread_shell_command.rs
+31 -7
View File
@@ -26,6 +26,7 @@ use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_core::shell::default_user_shell;
use codex_features::FEATURES;
use codex_features::Feature;
use pretty_assertions::assert_eq;
@@ -67,11 +68,12 @@ async fn thread_shell_command_runs_as_standalone_turn_and_persists_history() ->
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let (shell_command, expected_output) = current_shell_output_command("hello from bang")?;
let shell_id = mcp
.send_thread_shell_command_request(ThreadShellCommandParams {
thread_id: thread.id.clone(),
command: "printf 'hello from bang\\n'".to_string(),
command: shell_command,
})
.await?;
let shell_resp: JSONRPCResponse = timeout(
@@ -93,7 +95,10 @@ async fn thread_shell_command_runs_as_standalone_turn_and_persists_history() ->
assert_eq!(status, &CommandExecutionStatus::InProgress);
let delta = wait_for_command_execution_output_delta(&mut mcp, &command_id).await?;
assert_eq!(delta.delta, "hello from bang\n");
assert_eq!(
delta.delta.trim_end_matches(['\r', '\n']),
expected_output.trim_end_matches(['\r', '\n'])
);
let completed = wait_for_command_execution_completed(&mut mcp, Some(&command_id)).await?;
let ThreadItem::CommandExecution {
@@ -110,7 +115,7 @@ async fn thread_shell_command_runs_as_standalone_turn_and_persists_history() ->
assert_eq!(id, &command_id);
assert_eq!(source, &CommandExecutionSource::UserShell);
assert_eq!(status, &CommandExecutionStatus::Completed);
assert_eq!(aggregated_output.as_deref(), Some("hello from bang\n"));
assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str()));
assert_eq!(*exit_code, Some(0));
timeout(
@@ -147,7 +152,7 @@ async fn thread_shell_command_runs_as_standalone_turn_and_persists_history() ->
};
assert_eq!(source, &CommandExecutionSource::UserShell);
assert_eq!(status, &CommandExecutionStatus::Completed);
assert_eq!(aggregated_output.as_deref(), Some("hello from bang\n"));
assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str()));
Ok(())
}
@@ -196,6 +201,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let (shell_command, expected_output) = current_shell_output_command("active turn bang")?;
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
@@ -240,7 +246,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
let shell_id = mcp
.send_thread_shell_command_request(ThreadShellCommandParams {
thread_id: thread.id.clone(),
command: "printf 'active turn bang\\n'".to_string(),
command: shell_command,
})
.await?;
let shell_resp: JSONRPCResponse = timeout(
@@ -269,7 +275,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
unreachable!("helper returns command execution item");
};
assert_eq!(source, &CommandExecutionSource::UserShell);
assert_eq!(aggregated_output.as_deref(), Some("active turn bang\n"));
assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str()));
mcp.send_response(
request_id,
@@ -309,7 +315,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
source: CommandExecutionSource::UserShell,
aggregated_output,
..
} if aggregated_output.as_deref() == Some("active turn bang\n")
} if aggregated_output.as_deref() == Some(expected_output.as_str())
)
}),
"expected active-turn shell command to be persisted on the existing turn"
@@ -318,6 +324,24 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
Ok(())
}
fn current_shell_output_command(text: &str) -> Result<(String, String)> {
let command_and_output = match default_user_shell().name() {
"powershell" => {
let escaped_text = text.replace('\'', "''");
(
format!("Write-Output '{escaped_text}'"),
format!("{text}\r\n"),
)
}
"cmd" => (format!("echo {text}"), format!("{text}\r\n")),
_ => {
let quoted_text = shlex::try_quote(text)?;
(format!("printf '%s\\n' {quoted_text}"), format!("{text}\n"))
}
};
Ok(command_and_output)
}
async fn wait_for_command_execution_started(
mcp: &mut McpProcess,
expected_id: Option<&str>,
codex-rs/app-server/tests/suite/v2/thread_start.rs
+306 -11
View File
@@ -4,21 +4,24 @@ use app_test_support::McpProcess;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
use app_test_support::write_chatgpt_auth;
use codex_app_server_protocol::AskForApproval;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::McpServerStartupState;
use codex_app_server_protocol::McpServerStatusUpdatedNotification;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SandboxMode;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadStatusChangedNotification;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_core::config::set_project_trust_level;
use codex_git_utils::resolve_root_git_project_for_trust;
use codex_login::AuthCredentialsStoreMode;
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::config_types::TrustLevel;
use codex_protocol::openai_models::ReasoningEffort;
@@ -26,6 +29,7 @@ use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
@@ -34,6 +38,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -42,7 +51,7 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
// Start server and initialize.
let mut mcp = McpProcess::new(codex_home.path()).await?;
@@ -153,12 +162,79 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
assert_eq!(payload["events"].as_array().expect("events array").len(), 1);
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "new");
Ok(())
}
#[tokio::test]
async fn thread_start_does_not_track_thread_initialized_analytics_without_feature() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ false,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let _ = to_response::<ThreadStartResponse>(resp)?;
let payload = wait_for_analytics_payload(&server, Duration::from_millis(250)).await;
assert!(
payload.is_err(),
"thread analytics should be gated off when general_analytics is disabled"
);
Ok(())
}
#[tokio::test]
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let workspace = TempDir::new()?;
let project_config_dir = workspace.path().join(".codex");
@@ -199,7 +275,7 @@ async fn thread_start_accepts_flex_service_tier() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -227,7 +303,7 @@ async fn thread_start_accepts_metrics_service_name() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -254,7 +330,7 @@ async fn thread_start_accepts_metrics_service_name() -> Result<()> {
async fn thread_start_ephemeral_remains_pathless() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -455,6 +531,7 @@ async fn thread_start_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -510,16 +587,225 @@ async fn thread_start_surfaces_cloud_requirements_load_errors() -> Result<()> {
Ok(())
}
// Helper to create a config.toml pointing at the mock model server.
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
#[tokio::test]
async fn thread_start_with_elevated_sandbox_trusts_project_and_followup_loads_project_config()
-> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let workspace = TempDir::new()?;
let project_config_dir = workspace.path().join(".codex");
std::fs::create_dir_all(&project_config_dir)?;
std::fs::write(
project_config_dir.join("config.toml"),
r#"
model_reasoning_effort = "high"
"#,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let first_request = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(workspace.path().display().to_string()),
sandbox: Some(SandboxMode::WorkspaceWrite),
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(first_request)),
)
.await??;
let second_request = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(workspace.path().display().to_string()),
..Default::default()
})
.await?;
let second_response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(second_request)),
)
.await??;
let ThreadStartResponse {
approval_policy,
reasoning_effort,
..
} = to_response::<ThreadStartResponse>(second_response)?;
assert_eq!(approval_policy, AskForApproval::OnRequest);
assert_eq!(reasoning_effort, Some(ReasoningEffort::High));
let config_toml = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
let trusted_root = resolve_root_git_project_for_trust(workspace.path())
.unwrap_or_else(|| workspace.path().to_path_buf());
assert!(config_toml.contains(&persisted_trust_path(&trusted_root)));
assert!(config_toml.contains("trust_level = \"trusted\""));
Ok(())
}
#[tokio::test]
async fn thread_start_with_nested_git_cwd_trusts_repo_root() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let repo_root = TempDir::new()?;
std::fs::create_dir(repo_root.path().join(".git"))?;
let nested = repo_root.path().join("nested/project");
std::fs::create_dir_all(&nested)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(nested.display().to_string()),
sandbox: Some(SandboxMode::WorkspaceWrite),
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let config_toml = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
let trusted_root =
resolve_root_git_project_for_trust(&nested).expect("git root should resolve");
assert!(config_toml.contains(&persisted_trust_path(&trusted_root)));
assert!(!config_toml.contains(&persisted_trust_path(&nested)));
Ok(())
}
#[tokio::test]
async fn thread_start_with_read_only_sandbox_does_not_persist_project_trust() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let workspace = TempDir::new()?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(workspace.path().display().to_string()),
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let config_toml = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
assert!(!config_toml.contains("trust_level = \"trusted\""));
assert!(!config_toml.contains(&workspace.path().display().to_string()));
Ok(())
}
#[tokio::test]
async fn thread_start_skips_trust_write_when_project_is_already_trusted() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
let workspace = TempDir::new()?;
let project_config_dir = workspace.path().join(".codex");
std::fs::create_dir_all(&project_config_dir)?;
std::fs::write(
project_config_dir.join("config.toml"),
r#"
model_reasoning_effort = "high"
"#,
)?;
set_project_trust_level(codex_home.path(), workspace.path(), TrustLevel::Trusted)?;
let config_before = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(workspace.path().display().to_string()),
sandbox: Some(SandboxMode::WorkspaceWrite),
..Default::default()
})
.await?;
let response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let ThreadStartResponse {
approval_policy,
reasoning_effort,
..
} = to_response::<ThreadStartResponse>(response)?;
assert_eq!(approval_policy, AskForApproval::OnRequest);
assert_eq!(reasoning_effort, Some(ReasoningEffort::High));
let config_after = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
assert_eq!(config_after, config_before);
Ok(())
}
fn create_config_toml_without_approval_policy(
codex_home: &Path,
server_uri: &str,
) -> std::io::Result<()> {
create_config_toml_with_optional_approval_policy(
codex_home, server_uri, /*approval_policy*/ None,
)
}
fn persisted_trust_path(project_path: &Path) -> String {
let project_path =
std::fs::canonicalize(project_path).unwrap_or_else(|_| project_path.to_path_buf());
let project_path = project_path.display().to_string();
if let Some(project_path) = project_path.strip_prefix(r"\\?\UNC\") {
return format!(r"\\{project_path}");
}
project_path
.strip_prefix(r"\\?\")
.unwrap_or(&project_path)
.to_string()
}
fn create_config_toml_with_optional_approval_policy(
codex_home: &Path,
server_uri: &str,
approval_policy: Option<&str>,
) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
let approval_policy = approval_policy
.map(|policy| format!("approval_policy = \"{policy}\"\n"))
.unwrap_or_default();
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
{approval_policy}sandbox_mode = "read-only"
model_provider = "mock_provider"
@@ -538,7 +824,13 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
String::new()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -551,6 +843,9 @@ chatgpt_base_url = "{chatgpt_base_url}"
model_provider = "mock_provider"
[features]
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
codex-rs/app-server/tests/suite/v2/turn_start.rs
+61
View File
@@ -2523,6 +2523,67 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn turn_start_with_elevated_override_does_not_persist_project_trust() -> Result<()> {
let responses = vec![create_final_assistant_message_sse_response("Done")?];
let server = create_mock_responses_server_sequence_unchecked(responses).await;
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
&server.uri(),
"never",
&BTreeMap::from([(Feature::Personality, true)]),
)?;
let workspace = TempDir::new()?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_request = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(workspace.path().display().to_string()),
..Default::default()
})
.await?;
let thread_response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(thread_request)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_response)?;
let turn_request = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id,
cwd: Some(workspace.path().to_path_buf()),
sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess),
input: vec![V2UserInput::Text {
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_request)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let config_toml = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
assert!(!config_toml.contains("trust_level = \"trusted\""));
assert!(!config_toml.contains(&workspace.path().display().to_string()));
Ok(())
}
// Helper to create a config.toml pointing at the mock model server.
fn create_config_toml(
codex_home: &Path,