openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-24 18:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 5e37371586 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-06-27 17:36:46 -07:00
committed by GitHub
parent 476954a680 5e37371586
commit 5816a9a8d6
54 changed files with 1838 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+14 -8
View File
@@ -2683,6 +2683,12 @@
"name": {
"type": "string"
},
"namespace": {
"type": [
"string",
"null"
]
},
"status": {
"type": [
"string",
@@ -3631,6 +3637,13 @@
"ephemeral": {
"type": "boolean"
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -3673,13 +3686,6 @@
}
],
"description": "Optional client-supplied analytics source classification for this forked thread."
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -6873,4 +6879,4 @@
}
],
"title": "ClientRequest"
}
}
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+14 -8
View File
@@ -15510,6 +15510,12 @@
"name": {
"type": "string"
},
"namespace": {
"type": [
"string",
"null"
]
},
"status": {
"type": [
"string",
@@ -17279,6 +17285,13 @@
"ephemeral": {
"type": "boolean"
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -17321,13 +17334,6 @@
}
],
"description": "Optional client-supplied analytics source classification for this forked thread."
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -20874,4 +20880,4 @@
},
"title": "CodexAppServerProtocol",
"type": "object"
}
}
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+14 -8
View File
@@ -11914,6 +11914,12 @@
"name": {
"type": "string"
},
"namespace": {
"type": [
"string",
"null"
]
},
"status": {
"type": [
"string",
@@ -15058,6 +15064,13 @@
"ephemeral": {
"type": "boolean"
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -15100,13 +15113,6 @@
}
],
"description": "Optional client-supplied analytics source classification for this forked thread."
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -18652,4 +18658,4 @@
},
"title": "CodexAppServerProtocolV2",
"type": "object"
}
}
codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json
Generated
+6
View File
@@ -751,6 +751,12 @@
"name": {
"type": "string"
},
"namespace": {
"type": [
"string",
"null"
]
},
"status": {
"type": [
"string",
codex-rs/app-server-protocol/schema/json/v2/ThreadForkParams.json
Generated
+8 -8
View File
@@ -126,6 +126,13 @@
"ephemeral": {
"type": "boolean"
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -168,13 +175,6 @@
}
],
"description": "Optional client-supplied analytics source classification for this forked thread."
},
"lastTurnId": {
"description": "Optional last turn id to fork through, inclusive.\n\nWhen specified, turns after `last_turn_id` are omitted from the fork. The referenced turn cannot be in progress.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -182,4 +182,4 @@
],
"title": "ThreadForkParams",
"type": "object"
}
}
codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json
Generated
+6
View File
@@ -821,6 +821,12 @@
"name": {
"type": "string"
},
"namespace": {
"type": [
"string",
"null"
]
},
"status": {
"type": [
"string",
codex-rs/app-server-protocol/schema/typescript/ResponseItem.ts
Generated
+1 -1
View File
@@ -20,4 +20,4 @@ id?: string,
/**
* Set when using the Responses API.
*/
call_id: string | null, status: LocalShellStatus, action: LocalShellAction, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "function_call", id?: string, name: string, namespace?: string, arguments: string, call_id: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "tool_search_call", id?: string, call_id: string | null, status?: string, execution: string, arguments: unknown, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "function_call_output", id?: string, call_id: string, output: FunctionCallOutputBody, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "custom_tool_call", id?: string, status?: string, call_id: string, name: string, input: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "custom_tool_call_output", id?: string, call_id: string, name?: string, output: FunctionCallOutputBody, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "tool_search_output", id?: string, call_id: string | null, status: string, execution: string, tools: unknown[], internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "web_search_call", id?: string, status?: string, action?: WebSearchAction, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "image_generation_call", id?: string, status: string, revised_prompt?: string, result: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "compaction", id?: string, encrypted_content: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "compaction_trigger", } | { "type": "context_compaction", id?: string, encrypted_content?: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "other" };
call_id: string | null, status: LocalShellStatus, action: LocalShellAction, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "function_call", id?: string, name: string, namespace?: string, arguments: string, call_id: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "tool_search_call", id?: string, call_id: string | null, status?: string, execution: string, arguments: unknown, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "function_call_output", id?: string, call_id: string, output: FunctionCallOutputBody, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "custom_tool_call", id?: string, status?: string, call_id: string, name: string, namespace?: string, input: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "custom_tool_call_output", id?: string, call_id: string, name?: string, output: FunctionCallOutputBody, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "tool_search_output", id?: string, call_id: string | null, status: string, execution: string, tools: unknown[], internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "web_search_call", id?: string, status?: string, action?: WebSearchAction, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "image_generation_call", id?: string, status: string, revised_prompt?: string, result: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "compaction", id?: string, encrypted_content: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "compaction_trigger", } | { "type": "context_compaction", id?: string, encrypted_content?: string, internal_chat_message_metadata_passthrough?: InternalChatMessageMetadataPassthrough, } | { "type": "other" };
codex-rs/app-server-protocol/src/export.rs
+2
View File
@@ -43,6 +43,8 @@ pub(crate) const GENERATED_TS_HEADER: &str = "// GENERATED CODE! DO NOT MODIFY B
const IGNORED_DEFINITIONS: &[&str] = &["Option<()>"];
const JSON_V1_ALLOWLIST: &[&str] = &["InitializeParams", "InitializeResponse"];
const EXPERIMENTAL_CLIENT_METHOD_DEPENDENCY_TYPES: &[&str] = &[
"EnvironmentShellInfo",
"PathUri",
"RemoteControlClient",
"RemoteControlClientsListOrder",
"ThreadBackgroundTerminal",
codex-rs/app-server-protocol/src/protocol/common.rs
+7
View File
@@ -943,6 +943,13 @@ client_request_definitions! {
serialization: global("environment"),
response: v2::EnvironmentAddResponse,
},
#[experimental("environment/info")]
/// Reads information from a configured execution environment.
EnvironmentInfo => "environment/info" {
params: v2::EnvironmentInfoParams,
serialization: global_shared_read("environment"),
response: v2::EnvironmentInfoResponse,
},
McpServerOauthLogin => "mcpServer/oauth/login" {
params: v2::McpServerOauthLoginParams,
codex-rs/app-server-protocol/src/protocol/v2/environment.rs
+27
View File
@@ -1,3 +1,4 @@
use codex_utils_path_uri::PathUri;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
@@ -19,3 +20,29 @@ pub struct EnvironmentAddParams {
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct EnvironmentAddResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct EnvironmentInfoParams {
pub environment_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct EnvironmentInfoResponse {
pub shell: EnvironmentShellInfo,
/// Default working directory reported by the environment, as a canonical file URI.
pub cwd: Option<PathUri>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct EnvironmentShellInfo {
/// Stable shell name, for example `zsh`, `bash`, `powershell`, `sh`, or `cmd`.
pub name: String,
/// Target-native shell executable path or command name.
pub path: String,
}
codex-rs/app-server/README.md
+1
View File
@@ -205,6 +205,7 @@ Example with notification opt-out:
- `permissionProfile/list` — beta; list available permission profile ids with optional display `description` text and an `allowed` flag reflecting effective requirements, using cursor pagination. Pass `cwd` when the caller needs project-local `[permissions.<id>]` entries to be included in the current catalog view.
- `experimentalFeature/enablement/set` — patch the in-memory process-wide runtime feature enablement for currently supported feature keys. For each feature, precedence is: cloud requirements > --enable <feature_name> > config.toml > experimentalFeature/enablement/set (new) > code default. Invalid keys will be ignored.
- `environment/add` — experimental; add or replace a named remote environment by `environmentId` and `execServerUrl` for later selection by `thread/start` or `turn/start`; optional `connectTimeoutMs` overrides the WebSocket connection timeout; returns `{}` and does not change the default environment.
- `environment/info` — experimental; connect to a configured environment by `environmentId` and return its detected `shell` plus its default `cwd` as a canonical environment-native `file:` URI. Connection failures are returned as request errors.
- `collaborationMode/list` — list available collaboration mode presets (experimental, no pagination). Built-in presets do not select a model; the Plan preset selects medium reasoning effort. This response omits built-in developer instructions; clients should either pass `settings.developer_instructions: null` when setting a mode to use Codex's built-in instructions, or provide their own instructions explicitly.
- `skills/list` — list skills for one or more `cwd` values (optional `forceReload`).
- `skills/extraRoots/set` — replace the app-server process runtime extra standalone skill roots. The roots are not persisted; missing directories are accepted and simply load no skills.
codex-rs/app-server/src/current_time.rs
+1 -1
View File
@@ -22,7 +22,7 @@ use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingMessageSender;
use crate::thread_state::ThreadStateManager;
const CURRENT_TIME_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
const CURRENT_TIME_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
const CURRENT_TIME_POLL_INTERVAL: Duration = Duration::from_secs(1);
pub(crate) fn app_server_time_provider(
codex-rs/app-server/src/message_processor.rs
+3
View File
@@ -1037,6 +1037,9 @@ impl MessageProcessor {
ClientRequest::EnvironmentAdd { params, .. } => {
self.environment_processor.environment_add(params).await
}
ClientRequest::EnvironmentInfo { params, .. } => {
self.environment_processor.environment_info(params).await
}
ClientRequest::FsReadFile { params, .. } => self
.fs_processor
.read_file(params)
codex-rs/app-server/src/request_processors.rs
+3
View File
@@ -62,6 +62,9 @@ use codex_app_server_protocol::DynamicToolNamespaceTool;
use codex_app_server_protocol::DynamicToolSpec;
use codex_app_server_protocol::EnvironmentAddParams;
use codex_app_server_protocol::EnvironmentAddResponse;
use codex_app_server_protocol::EnvironmentInfoParams;
use codex_app_server_protocol::EnvironmentInfoResponse;
use codex_app_server_protocol::EnvironmentShellInfo;
use codex_app_server_protocol::ExperimentalFeature as ApiExperimentalFeature;
use codex_app_server_protocol::ExperimentalFeatureListParams;
use codex_app_server_protocol::ExperimentalFeatureListResponse;
codex-rs/app-server/src/request_processors/environment_processor.rs
+26
View File
@@ -26,4 +26,30 @@ impl EnvironmentRequestProcessor {
.map_err(|err| invalid_request(err.to_string()))?;
Ok(Some(EnvironmentAddResponse {}.into()))
}
pub(crate) async fn environment_info(
&self,
params: EnvironmentInfoParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
let environment_id = params.environment_id;
let environment = self
.environment_manager
.get_environment(&environment_id)
.ok_or_else(|| invalid_request(format!("unknown environment id `{environment_id}`")))?;
let info = environment.info().await.map_err(|err| {
internal_error(format!(
"failed to get info for environment `{environment_id}`: {err}"
))
})?;
Ok(Some(
EnvironmentInfoResponse {
shell: EnvironmentShellInfo {
name: info.shell.name,
path: info.shell.path,
},
cwd: info.cwd,
}
.into(),
))
}
}
codex-rs/app-server/tests/suite/v2/environment_info.rs
+208
View File
@@ -0,0 +1,208 @@
use std::time::Duration;
use anyhow::Result;
use app_test_support::TestAppServer;
use app_test_support::to_response;
use codex_app_server_protocol::EnvironmentAddResponse;
use codex_app_server_protocol::EnvironmentInfoResponse;
use codex_app_server_protocol::EnvironmentShellInfo;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_utils_path_uri::PathUri;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::time::timeout;
use super::exec_server_test_support::accept_exec_server_environment;
const RPC_TIMEOUT: Duration = Duration::from_secs(10);
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
const INTERNAL_ERROR_CODE: i64 = -32603;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn environment_info_returns_remote_environment_info() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let exec_server_url = format!("ws://{}", listener.local_addr()?);
let exec_server = tokio::spawn(async move {
accept_exec_server_environment(
listener,
json!({
"shell": {"name": "zsh", "path": "/bin/zsh"},
"cwd": "file:///workspace",
}),
)
.await?;
Ok::<_, anyhow::Error>(())
});
let codex_home = TempDir::new()?;
let mut app_server = TestAppServer::new(codex_home.path()).await?;
timeout(RPC_TIMEOUT, app_server.initialize()).await??;
add_environment(
&mut app_server,
&exec_server_url,
/*connect_timeout_ms*/ None,
)
.await?;
let request_id = app_server
.send_raw_request(
"environment/info",
Some(json!({"environmentId": "remote-a"})),
)
.await?;
let response: JSONRPCResponse = timeout(
RPC_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
assert_eq!(
to_response::<EnvironmentInfoResponse>(response)?,
EnvironmentInfoResponse {
shell: EnvironmentShellInfo {
name: "zsh".to_string(),
path: "/bin/zsh".to_string(),
},
cwd: Some(PathUri::parse("file:///workspace")?),
}
);
timeout(RPC_TIMEOUT, exec_server).await???;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn environment_info_accepts_missing_cwd() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let exec_server_url = format!("ws://{}", listener.local_addr()?);
let exec_server = tokio::spawn(async move {
accept_exec_server_environment(
listener,
json!({"shell": {"name": "zsh", "path": "/bin/zsh"}}),
)
.await?;
Ok::<_, anyhow::Error>(())
});
let codex_home = TempDir::new()?;
let mut app_server = TestAppServer::new(codex_home.path()).await?;
timeout(RPC_TIMEOUT, app_server.initialize()).await??;
add_environment(
&mut app_server,
&exec_server_url,
/*connect_timeout_ms*/ None,
)
.await?;
let request_id = app_server
.send_raw_request(
"environment/info",
Some(json!({"environmentId": "remote-a"})),
)
.await?;
let response: JSONRPCResponse = timeout(
RPC_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
assert_eq!(
to_response::<EnvironmentInfoResponse>(response)?,
EnvironmentInfoResponse {
shell: EnvironmentShellInfo {
name: "zsh".to_string(),
path: "/bin/zsh".to_string(),
},
cwd: None,
}
);
timeout(RPC_TIMEOUT, exec_server).await???;
Ok(())
}
#[tokio::test]
async fn environment_info_rejects_unknown_environment() -> Result<()> {
let codex_home = TempDir::new()?;
let mut app_server = TestAppServer::new(codex_home.path()).await?;
timeout(RPC_TIMEOUT, app_server.initialize()).await??;
let request_id = app_server
.send_raw_request(
"environment/info",
Some(json!({"environmentId": "missing"})),
)
.await?;
let error = timeout(
RPC_TIMEOUT,
app_server.read_stream_until_error_message(RequestId::Integer(request_id)),
)
.await??;
assert_eq!(
error,
JSONRPCError {
id: RequestId::Integer(request_id),
error: JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "unknown environment id `missing`".to_string(),
data: None,
},
}
);
Ok(())
}
#[tokio::test]
async fn environment_info_reports_connection_failure() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let exec_server_url = format!("ws://{}", listener.local_addr()?);
let codex_home = TempDir::new()?;
let mut app_server = TestAppServer::new(codex_home.path()).await?;
timeout(RPC_TIMEOUT, app_server.initialize()).await??;
add_environment(&mut app_server, &exec_server_url, Some(50)).await?;
let request_id = app_server
.send_raw_request(
"environment/info",
Some(json!({"environmentId": "remote-a"})),
)
.await?;
let error = timeout(
RPC_TIMEOUT,
app_server.read_stream_until_error_message(RequestId::Integer(request_id)),
)
.await??;
assert_eq!(error.error.code, INTERNAL_ERROR_CODE);
assert!(
error
.error
.message
.contains("failed to get info for environment `remote-a`")
);
Ok(())
}
async fn add_environment(
app_server: &mut TestAppServer,
exec_server_url: &str,
connect_timeout_ms: Option<u64>,
) -> Result<()> {
let request_id = app_server
.send_raw_request(
"environment/add",
Some(json!({
"environmentId": "remote-a",
"execServerUrl": exec_server_url,
"connectTimeoutMs": connect_timeout_ms,
})),
)
.await?;
let response: JSONRPCResponse = timeout(
RPC_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let _: EnvironmentAddResponse = to_response(response)?;
Ok(())
}
codex-rs/app-server/tests/suite/v2/exec_server_test_support.rs
+73
View File
@@ -0,0 +1,73 @@
use anyhow::Result;
use futures::SinkExt;
use futures::StreamExt;
use serde_json::Value;
use serde_json::json;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
pub(crate) async fn accept_exec_server_environment(
listener: TcpListener,
environment_info: Value,
) -> Result<WebSocketStream<TcpStream>> {
let mut websocket = accept_initialized_exec_server(listener).await?;
let request = read_exec_server_json(&mut websocket).await?;
assert_eq!(request["method"], "environment/info");
websocket
.send(Message::Text(
json!({
"id": request["id"],
"result": environment_info,
})
.to_string()
.into(),
))
.await?;
Ok(websocket)
}
pub(crate) async fn accept_initialized_exec_server(
listener: TcpListener,
) -> Result<WebSocketStream<TcpStream>> {
let (stream, _) = listener.accept().await?;
let mut websocket = accept_async(stream).await?;
let initialize = read_exec_server_json(&mut websocket).await?;
assert_eq!(initialize["method"], "initialize");
websocket
.send(Message::Text(
json!({
"id": initialize["id"],
"result": {"sessionId": "test-session"},
})
.to_string()
.into(),
))
.await?;
let initialized = read_exec_server_json(&mut websocket).await?;
assert_eq!(initialized["method"], "initialized");
Ok(websocket)
}
pub(crate) async fn read_exec_server_json(
websocket: &mut WebSocketStream<TcpStream>,
) -> Result<Value> {
loop {
match websocket
.next()
.await
.ok_or_else(|| anyhow::anyhow!("exec-server websocket closed"))??
{
Message::Text(text) => return Ok(serde_json::from_str(text.as_ref())?),
Message::Binary(bytes) => return Ok(serde_json::from_slice(bytes.as_ref())?),
Message::Ping(_) | Message::Pong(_) => {}
message => anyhow::bail!("expected JSON-RPC message, got {message:?}"),
}
}
}
codex-rs/app-server/tests/suite/v2/mcp_tool.rs
+8 -50
View File
@@ -38,7 +38,6 @@ use codex_utils_path_uri::PathUri;
use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP;
use core_test_support::responses;
use futures::SinkExt;
use futures::StreamExt;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
use rmcp::model::BooleanSchema;
@@ -64,14 +63,14 @@ use rmcp::transport::streamable_http_server::session::local::LocalSessionManager
use serde_json::json;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio::time::timeout;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
use super::exec_server_test_support::accept_exec_server_environment;
use super::exec_server_test_support::read_exec_server_json;
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
const TEST_SERVER_NAME: &str = "tool_server";
const TEST_TOOL_NAME: &str = "echo_tool";
@@ -841,35 +840,11 @@ async fn serve_environment_until_shutdown(
filesystem_request_tx: oneshot::Sender<()>,
mut shutdown_rx: oneshot::Receiver<()>,
) -> Result<()> {
let (stream, _) = listener.accept().await?;
let mut websocket = accept_async(stream).await?;
let initialize = read_exec_server_json(&mut websocket).await?;
assert_eq!(initialize["method"], "initialize");
websocket
.send(Message::Text(
json!({
"id": initialize["id"],
"result": {"sessionId": "test-session"},
})
.to_string()
.into(),
))
.await?;
let initialized = read_exec_server_json(&mut websocket).await?;
assert_eq!(initialized["method"], "initialized");
let environment_info = read_exec_server_json(&mut websocket).await?;
assert_eq!(environment_info["method"], "environment/info");
websocket
.send(Message::Text(
json!({
"id": environment_info["id"],
"result": {"shell": {"name": "zsh", "path": "/bin/zsh"}},
})
.to_string()
.into(),
))
.await?;
let mut websocket = accept_exec_server_environment(
listener,
json!({"shell": {"name": "zsh", "path": "/bin/zsh"}}),
)
.await?;
let mut filesystem_request_tx = Some(filesystem_request_tx);
loop {
@@ -899,23 +874,6 @@ async fn serve_environment_until_shutdown(
}
}
async fn read_exec_server_json(
websocket: &mut WebSocketStream<TcpStream>,
) -> Result<serde_json::Value> {
loop {
match websocket
.next()
.await
.ok_or_else(|| anyhow::anyhow!("exec-server websocket closed"))??
{
Message::Text(text) => return Ok(serde_json::from_str(text.as_ref())?),
Message::Binary(bytes) => return Ok(serde_json::from_slice(bytes.as_ref())?),
Message::Ping(_) | Message::Pong(_) => {}
message => anyhow::bail!("expected JSON-RPC message, got {message:?}"),
}
}
}
async fn wait_for_mcp_tool_call_completed(
mcp: &mut TestAppServer,
call_id: &str,
codex-rs/app-server/tests/suite/v2/mod.rs
+2
View File
@@ -15,6 +15,8 @@ mod connection_handling_websocket_unix;
mod current_time;
mod dynamic_tools;
mod environment_add;
mod environment_info;
mod exec_server_test_support;
#[cfg(not(target_os = "windows"))]
mod executor_mcp;
mod executor_skills;
codex-rs/cli/src/marketplace_cmd.rs
+1 -1
View File
@@ -341,7 +341,7 @@ fn configured_marketplace_sources_by_root(
codex_home: &Path,
plugins_input: &PluginsConfigInput,
) -> HashMap<PathBuf, JsonMarketplaceSource> {
let marketplace_sources = configured_marketplace_sources(plugins_input);
let marketplace_sources = configured_marketplace_sources(plugins_input, codex_home);
let Some(user_config) = plugins_input.config_layer_stack.effective_user_config() else {
return HashMap::new();
};
codex-rs/cli/src/plugin_cmd.rs
+11 -1
View File
@@ -10,6 +10,7 @@ use codex_core_plugins::PluginInstallOutcome;
use codex_core_plugins::PluginInstallRequest;
use codex_core_plugins::PluginsConfigInput;
use codex_core_plugins::PluginsManager;
use codex_core_plugins::allowed_configured_marketplace_names;
use codex_core_plugins::installed_marketplaces::marketplace_install_root;
use codex_core_plugins::installed_marketplaces::resolve_configured_marketplace_root;
use codex_core_plugins::marketplace::MarketplaceListError;
@@ -228,7 +229,7 @@ pub async fn run_plugin_list(
.is_none_or(|name| marketplace.name == *name)
})
.collect::<Vec<_>>();
let marketplace_sources = configured_marketplace_sources(&plugins_input);
let marketplace_sources = configured_marketplace_sources(&plugins_input, codex_home.as_path());
if args.json {
let output = JsonPluginListOutput::from_marketplaces(
@@ -480,6 +481,7 @@ pub(crate) struct JsonMarketplaceSource {
pub(crate) fn configured_marketplace_sources(
plugins_input: &PluginsConfigInput,
codex_home: &Path,
) -> HashMap<String, JsonMarketplaceSource> {
let Some(user_config) = plugins_input.config_layer_stack.effective_user_config() else {
return HashMap::new();
@@ -490,9 +492,12 @@ pub(crate) fn configured_marketplace_sources(
else {
return HashMap::new();
};
let allowed_marketplace_names =
allowed_configured_marketplace_names(&plugins_input.config_layer_stack, codex_home);
marketplaces
.iter()
.filter(|(marketplace_name, _)| allowed_marketplace_names.contains(*marketplace_name))
.filter_map(|(marketplace_name, marketplace)| {
let source_type = marketplace
.get("source_type")
@@ -746,11 +751,16 @@ pub(crate) fn configured_marketplace_snapshot_issues(
else {
return Vec::new();
};
let allowed_marketplace_names =
allowed_configured_marketplace_names(&plugins_input.config_layer_stack, codex_home);
let default_install_root = marketplace_install_root(codex_home);
let mut manifest_paths = Vec::new();
let mut issues = Vec::new();
for (configured_name, marketplace) in configured_marketplaces {
if !allowed_marketplace_names.contains(configured_name) {
continue;
}
if marketplace_name.is_some_and(|name| configured_name != name) {
continue;
}
codex-rs/cli/tests/plugin_cli.rs
+3 -1
View File
@@ -205,6 +205,7 @@ fn setup_local_marketplace_with_implicit_system_roots() -> Result<(TempDir, Temp
let cache_home = TempDir::new()?;
let runtime_root = cache_home
.path()
.join(".cache")
.join("codex-runtimes")
.join("codex-primary-runtime")
.join("plugins")
@@ -844,7 +845,8 @@ async fn plugin_list_ignores_implicit_system_marketplace_roots_without_manifests
let (codex_home, source, cache_home) = setup_local_marketplace_with_implicit_system_roots()?;
codex_command(codex_home.path())?
.env("XDG_CACHE_HOME", cache_home.path())
.env("HOME", cache_home.path())
.env("USERPROFILE", cache_home.path())
.args(["plugin", "list"])
.assert()
.success()
codex-rs/core-plugins/src/installed_marketplaces.rs
+2 -1
View File
@@ -7,6 +7,7 @@ use codex_utils_absolute_path::AbsolutePathBuf;
use tracing::warn;
use crate::marketplace::find_marketplace_manifest_path;
use crate::marketplace_policy::project_effective_user_config;
pub const INSTALLED_MARKETPLACES_DIR: &str = ".tmp/marketplaces";
@@ -18,7 +19,7 @@ pub fn installed_marketplace_roots_from_layer_stack(
config_layer_stack: &ConfigLayerStack,
codex_home: &Path,
) -> Vec<AbsolutePathBuf> {
let Some(user_config) = config_layer_stack.effective_user_config() else {
let Some(user_config) = project_effective_user_config(config_layer_stack, codex_home) else {
return Vec::new();
};
let Some(marketplaces_value) = user_config.get("marketplaces") else {
codex-rs/core-plugins/src/lib.rs
+1
View File
@@ -55,6 +55,7 @@ pub use manager::PluginUninstallError;
pub use manager::PluginsConfigInput;
pub use manager::PluginsManager;
pub use manager::RecommendedPluginCandidatesInput;
pub use marketplace_policy::allowed_configured_marketplace_names;
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeError as PluginMarketplaceUpgradeError;
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome as PluginMarketplaceUpgradeOutcome;
pub use provider::ExecutorPluginProvider;
codex-rs/core-plugins/src/loader.rs
+142 -71
View File
@@ -8,8 +8,9 @@ use crate::manifest::PluginManifestPaths;
use crate::manifest::load_plugin_manifest;
use crate::marketplace::MarketplacePluginSource;
use crate::marketplace::find_marketplace_plugin;
use crate::marketplace::list_marketplaces;
use crate::marketplace::list_marketplaces_with_home;
use crate::marketplace::load_marketplace;
use crate::marketplace_policy::configured_plugins_from_stack;
use crate::npm_source::materialize_npm_plugin_source;
use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;
use crate::remote::RemoteInstalledPlugin;
@@ -84,6 +85,18 @@ enum NonCuratedCacheRefreshMode {
ForceReinstall,
}
#[derive(Debug)]
pub(crate) struct NonCuratedCacheRefreshOutcome {
pub(crate) cache_refreshed: bool,
pub(crate) errors: Vec<NonCuratedCacheRefreshError>,
}
#[derive(Debug)]
pub(crate) struct NonCuratedCacheRefreshError {
pub(crate) marketplace_name: String,
pub(crate) message: String,
}
pub(crate) fn log_plugin_load_errors(plugins: &[LoadedPlugin<McpServerConfig>]) {
for plugin in plugins.iter().filter(|plugin| plugin.error.is_some()) {
if let Some(error) = plugin.error.as_deref() {
@@ -129,7 +142,7 @@ async fn load_plugins_from_layer_stack_with_scope(
scope: PluginLoadScope<'_>,
) -> Vec<LoadedPlugin<McpServerConfig>> {
let configured_plugins = merge_configured_plugins_with_remote_installed(
configured_plugins_from_stack(config_layer_stack),
configured_plugins_from_stack(config_layer_stack, store.codex_home().as_path()),
extra_plugins,
store,
remote_global_catalog_active,
@@ -410,24 +423,54 @@ pub fn curated_plugin_cache_version(plugin_version: &str) -> String {
}
}
pub fn refresh_non_curated_plugin_cache(
#[cfg(test)]
pub(crate) fn refresh_non_curated_plugin_cache(
codex_home: &Path,
additional_roots: &[AbsolutePathBuf],
configured_plugin_keys: &[String],
) -> Result<bool, String> {
collapse_non_curated_cache_refresh(refresh_non_curated_plugin_cache_detailed(
codex_home,
additional_roots,
configured_plugin_keys,
))
}
pub(crate) fn refresh_non_curated_plugin_cache_detailed(
codex_home: &Path,
additional_roots: &[AbsolutePathBuf],
configured_plugin_keys: &[String],
) -> Result<NonCuratedCacheRefreshOutcome, String> {
refresh_non_curated_plugin_cache_with_mode(
codex_home,
additional_roots,
configured_plugin_keys,
NonCuratedCacheRefreshMode::IfVersionChanged,
)
}
pub fn refresh_non_curated_plugin_cache_force_reinstall(
#[cfg(test)]
pub(crate) fn refresh_non_curated_plugin_cache_force_reinstall(
codex_home: &Path,
additional_roots: &[AbsolutePathBuf],
configured_plugin_keys: &[String],
) -> Result<bool, String> {
collapse_non_curated_cache_refresh(refresh_non_curated_plugin_cache_force_reinstall_detailed(
codex_home,
additional_roots,
configured_plugin_keys,
))
}
pub(crate) fn refresh_non_curated_plugin_cache_force_reinstall_detailed(
codex_home: &Path,
additional_roots: &[AbsolutePathBuf],
configured_plugin_keys: &[String],
) -> Result<NonCuratedCacheRefreshOutcome, String> {
refresh_non_curated_plugin_cache_with_mode(
codex_home,
additional_roots,
configured_plugin_keys,
NonCuratedCacheRefreshMode::ForceReinstall,
)
}
@@ -435,16 +478,32 @@ pub fn refresh_non_curated_plugin_cache_force_reinstall(
fn refresh_non_curated_plugin_cache_with_mode(
codex_home: &Path,
additional_roots: &[AbsolutePathBuf],
configured_plugin_keys: &[String],
mode: NonCuratedCacheRefreshMode,
) -> Result<bool, String> {
let configured_non_curated_plugin_ids =
non_curated_plugin_ids_from_config_keys(configured_plugins_from_codex_home(
codex_home,
"failed to read user config while refreshing non-curated plugin cache",
"failed to parse user config while refreshing non-curated plugin cache",
));
) -> Result<NonCuratedCacheRefreshOutcome, String> {
let mut configured_non_curated_plugin_ids = configured_plugin_keys
.iter()
.filter_map(|plugin_key| match PluginId::parse(plugin_key) {
Ok(plugin_id) if !is_openai_curated_marketplace_name(&plugin_id.marketplace_name) => {
Some(plugin_id)
}
Ok(_) => None,
Err(err) => {
warn!(
plugin_key,
error = %err,
"ignoring invalid plugin key during non-curated cache refresh setup"
);
None
}
})
.collect::<Vec<_>>();
configured_non_curated_plugin_ids.sort_unstable_by_key(PluginId::as_key);
if configured_non_curated_plugin_ids.is_empty() {
return Ok(false);
return Ok(NonCuratedCacheRefreshOutcome {
cache_refreshed: false,
errors: Vec::new(),
});
}
let configured_non_curated_plugin_keys = configured_non_curated_plugin_ids
.iter()
@@ -452,7 +511,7 @@ fn refresh_non_curated_plugin_cache_with_mode(
.collect::<HashSet<_>>();
let store = PluginStore::try_new(codex_home.to_path_buf()).map_err(|err| err.to_string())?;
let marketplace_outcome = list_marketplaces(additional_roots)
let marketplace_outcome = list_marketplaces_with_home(additional_roots, /*home_dir*/ None)
.map_err(|err| format!("failed to discover marketplaces for cache refresh: {err}"))?;
let mut plugin_sources = HashMap::<String, (MarketplacePluginSource, Option<String>)>::new();
@@ -462,14 +521,18 @@ fn refresh_non_curated_plugin_cache_with_mode(
}
for plugin in marketplace.plugins {
let plugin_id =
PluginId::new(plugin.name.clone(), marketplace.name.clone()).map_err(|err| {
match err {
PluginIdError::Invalid(message) => {
format!("failed to prepare non-curated plugin cache refresh: {message}")
}
}
})?;
let plugin_id = match PluginId::new(plugin.name.clone(), marketplace.name.clone()) {
Ok(plugin_id) => plugin_id,
Err(PluginIdError::Invalid(message)) => {
warn!(
plugin = plugin.name,
marketplace = marketplace.name,
error = %message,
"ignoring invalid plugin entry during cache refresh"
);
continue;
}
};
let plugin_key = plugin_id.as_key();
if !configured_non_curated_plugin_keys.contains(&plugin_key) {
continue;
@@ -504,6 +567,7 @@ fn refresh_non_curated_plugin_cache_with_mode(
}
let mut cache_refreshed = false;
let mut refresh_errors = Vec::new();
for plugin_id in configured_non_curated_plugin_ids {
let plugin_key = plugin_id.as_key();
let Some((source, manifest_fallback_contents)) = plugin_sources.get(&plugin_key).cloned()
@@ -515,49 +579,70 @@ fn refresh_non_curated_plugin_cache_with_mode(
);
continue;
};
let materialized =
materialize_marketplace_plugin_source(codex_home, &source).map_err(|err| {
format!("failed to materialize plugin source for {plugin_key}: {err}")
})?;
let source_path = materialized.path.clone();
let plugin_version = match manifest_fallback_contents.as_deref() {
Some(manifest_contents) => plugin_version_for_source_with_fallback_manifest(
source_path.as_path(),
manifest_contents,
),
None => plugin_version_for_source(source_path.as_path()),
}
.map_err(|err| format!("failed to read plugin version for {plugin_key}: {err}"))?;
let refresh_result = (|| -> Result<bool, String> {
let materialized =
materialize_marketplace_plugin_source(codex_home, &source).map_err(|err| {
format!("failed to materialize plugin source for {plugin_key}: {err}")
})?;
let source_path = materialized.path;
let plugin_version = match manifest_fallback_contents.as_deref() {
Some(manifest_contents) => plugin_version_for_source_with_fallback_manifest(
source_path.as_path(),
manifest_contents,
),
None => plugin_version_for_source(source_path.as_path()),
}
.map_err(|err| format!("failed to read plugin version for {plugin_key}: {err}"))?;
if mode == NonCuratedCacheRefreshMode::IfVersionChanged
&& store.active_plugin_version(&plugin_id).as_deref() == Some(plugin_version.as_str())
{
continue;
}
if mode == NonCuratedCacheRefreshMode::IfVersionChanged
&& store.active_plugin_version(&plugin_id).as_deref()
== Some(plugin_version.as_str())
{
return Ok(false);
}
match manifest_fallback_contents.as_deref() {
Some(manifest_contents) => store.install_with_version_and_fallback_manifest(
source_path,
plugin_id.clone(),
plugin_version,
manifest_contents,
),
None => store.install_with_version(source_path, plugin_id.clone(), plugin_version),
match manifest_fallback_contents.as_deref() {
Some(manifest_contents) => store.install_with_version_and_fallback_manifest(
source_path,
plugin_id.clone(),
plugin_version,
manifest_contents,
),
None => store.install_with_version(source_path, plugin_id.clone(), plugin_version),
}
.map_err(|err| format!("failed to refresh plugin cache for {plugin_key}: {err}"))?;
Ok(true)
})();
match refresh_result {
Ok(refreshed) => cache_refreshed |= refreshed,
Err(message) => refresh_errors.push(NonCuratedCacheRefreshError {
marketplace_name: plugin_id.marketplace_name,
message,
}),
}
.map_err(|err| format!("failed to refresh plugin cache for {plugin_key}: {err}"))?;
cache_refreshed = true;
}
Ok(cache_refreshed)
Ok(NonCuratedCacheRefreshOutcome {
cache_refreshed,
errors: refresh_errors,
})
}
fn configured_plugins_from_stack(
config_layer_stack: &ConfigLayerStack,
) -> HashMap<String, PluginConfig> {
let Some(user_config) = config_layer_stack.effective_user_config() else {
return HashMap::new();
};
configured_plugins_from_user_config_value(&user_config)
#[cfg(test)]
fn collapse_non_curated_cache_refresh(
outcome: Result<NonCuratedCacheRefreshOutcome, String>,
) -> Result<bool, String> {
let outcome = outcome?;
if outcome.errors.is_empty() {
Ok(outcome.cache_refreshed)
} else {
Err(outcome
.errors
.into_iter()
.map(|error| error.message)
.collect::<Vec<_>>()
.join("; "))
}
}
fn is_full_git_sha(value: &str) -> bool {
@@ -647,20 +732,6 @@ fn curated_plugin_ids_from_config_keys(
configured_curated_plugin_ids
}
fn non_curated_plugin_ids_from_config_keys(
configured_plugins: HashMap<String, PluginConfig>,
) -> Vec<PluginId> {
let mut configured_non_curated_plugin_ids = configured_plugin_ids(
configured_plugins,
"ignoring invalid plugin key during non-curated cache refresh setup",
)
.into_iter()
.filter(|plugin_id| !is_openai_curated_marketplace_name(&plugin_id.marketplace_name))
.collect::<Vec<_>>();
configured_non_curated_plugin_ids.sort_unstable_by_key(PluginId::as_key);
configured_non_curated_plugin_ids
}
pub fn configured_curated_plugin_ids_from_codex_home(codex_home: &Path) -> Vec<PluginId> {
curated_plugin_ids_from_config_keys(configured_plugins_from_codex_home(
codex_home,
codex-rs/core-plugins/src/loader_tests.rs
+1 -1
View File
@@ -43,7 +43,7 @@ fn configured_plugins_from_stack_merges_user_layers() {
)
.expect("valid config layer stack");
let plugins = configured_plugins_from_stack(&stack);
let plugins = configured_plugins_from_stack(&stack, temp_dir.path());
assert_eq!(
plugins,
codex-rs/core-plugins/src/manager.rs
+166 -63
View File
@@ -16,8 +16,8 @@ use crate::loader::log_plugin_load_errors;
use crate::loader::materialize_marketplace_plugin_source;
use crate::loader::plugin_capability_summary_from_root;
use crate::loader::refresh_curated_plugin_cache;
use crate::loader::refresh_non_curated_plugin_cache;
use crate::loader::refresh_non_curated_plugin_cache_force_reinstall;
use crate::loader::refresh_non_curated_plugin_cache_detailed;
use crate::loader::refresh_non_curated_plugin_cache_force_reinstall_detailed;
use crate::loader::remote_installed_plugins_to_config;
use crate::manifest::PluginManifestInterface;
use crate::manifest::load_plugin_manifest;
@@ -32,9 +32,12 @@ use crate::marketplace::MarketplacePluginSource;
use crate::marketplace::ResolvedMarketplacePlugin;
use crate::marketplace::find_installable_marketplace_plugin;
use crate::marketplace::find_marketplace_plugin;
use crate::marketplace::list_marketplaces;
use crate::marketplace::home_dir;
use crate::marketplace::list_marketplaces_with_home;
use crate::marketplace::plugin_interface_with_marketplace_category;
use crate::marketplace_policy::MarketplacePolicy;
use crate::marketplace_policy::allowed_configured_marketplace_names;
use crate::marketplace_policy::configured_plugins_from_stack;
use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeError;
use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome;
use crate::marketplace_upgrade::upgrade_configured_git_marketplaces;
@@ -84,6 +87,7 @@ use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_plugins::PluginSkillRoot;
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::RwLock;
@@ -195,6 +199,7 @@ pub struct PluginListBackgroundTaskOptions {
#[derive(Clone, PartialEq, Eq)]
struct NonCuratedCacheRefreshRequest {
roots: Vec<AbsolutePathBuf>,
configured_plugin_keys: Vec<String>,
mode: NonCuratedCacheRefreshMode,
}
@@ -388,9 +393,16 @@ struct PluginLoadCacheKey {
}
impl PluginLoadCacheKey {
fn from_config(config: &PluginsConfigInput, remote_global_catalog_active: bool) -> Self {
fn from_config(
config: &PluginsConfigInput,
codex_home: &Path,
remote_global_catalog_active: bool,
) -> Self {
Self {
configured_plugins: configured_plugins_from_stack(&config.config_layer_stack),
configured_plugins: configured_plugins_from_stack(
&config.config_layer_stack,
codex_home,
),
skill_config_rules: skill_config_rules_from_stack(&config.config_layer_stack),
remote_global_catalog_active,
}
@@ -494,8 +506,11 @@ impl PluginsManager {
if !config.plugins_enabled {
return None;
}
let key =
PluginLoadCacheKey::from_config(config, self.remote_global_catalog_active(config));
let key = PluginLoadCacheKey::from_config(
config,
self.codex_home.as_path(),
self.remote_global_catalog_active(config),
);
self.loaded_plugins_cache
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
@@ -525,7 +540,11 @@ impl PluginsManager {
}
let remote_global_catalog_active = self.remote_global_catalog_active(config);
let cache_key = PluginLoadCacheKey::from_config(config, remote_global_catalog_active);
let cache_key = PluginLoadCacheKey::from_config(
config,
self.codex_home.as_path(),
remote_global_catalog_active,
);
if !force_reload && let Some(plugins) = self.cached_loaded_plugins(&cache_key) {
return self.resolve_loaded_plugins_for_auth(plugins);
}
@@ -1007,7 +1026,7 @@ impl PluginsManager {
options: PluginListBackgroundTaskOptions,
on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
) {
self.maybe_start_non_curated_plugin_cache_refresh(roots);
self.maybe_start_non_curated_plugin_cache_refresh(config, roots);
if options.refresh_global_remote_catalog_cache {
self.maybe_start_global_remote_catalog_cache_refresh(config, auth.clone());
}
@@ -1547,7 +1566,7 @@ impl PluginsManager {
let (installed_plugins, enabled_plugins) = self.configured_plugin_states(config);
let marketplace_roots =
self.marketplace_roots(config, additional_roots, include_openai_curated);
let marketplace_outcome = list_marketplaces(&marketplace_roots)?;
let marketplace_outcome = self.list_marketplaces_with_policy(config, &marketplace_roots)?;
let mut seen_plugin_keys = HashSet::new();
let marketplaces = marketplace_outcome
.marketplaces
@@ -1636,11 +1655,12 @@ impl PluginsManager {
return Ok(MarketplaceListOutcome::default());
}
list_marketplaces(&self.marketplace_roots(
let marketplace_roots = self.marketplace_roots(
config,
additional_roots,
/*include_openai_curated*/ true,
))
);
self.list_marketplaces_with_policy(config, &marketplace_roots)
}
pub(crate) async fn tool_suggest_metadata_for_marketplace_plugin(
@@ -1666,6 +1686,17 @@ impl PluginsManager {
}
let plugin = find_marketplace_plugin(&request.marketplace_path, &request.plugin_name)?;
MarketplacePolicy::from_requirements(config.config_layer_stack.requirements())
.validate_install(
&config.config_layer_stack,
self.codex_home.as_path(),
&request.marketplace_path,
&plugin.plugin_id.marketplace_name,
)
.map_err(|message| MarketplaceError::InvalidMarketplaceFile {
path: request.marketplace_path.to_path_buf(),
message,
})?;
if !self.restriction_product_matches(plugin.policy.products.as_deref()) {
return Err(MarketplaceError::PluginNotFound {
plugin_name: plugin.plugin_id.plugin_name,
@@ -2023,12 +2054,30 @@ impl PluginsManager {
));
}
if !outcome.upgraded_roots.is_empty() {
match refresh_non_curated_plugin_cache_force_reinstall(
let mut configured_plugin_keys = configured_plugins_from_stack(
&config.config_layer_stack,
self.codex_home.as_path(),
)
.into_keys()
.collect::<Vec<_>>();
configured_plugin_keys.sort_unstable();
match refresh_non_curated_plugin_cache_force_reinstall_detailed(
self.codex_home.as_path(),
&outcome.upgraded_roots,
&configured_plugin_keys,
) {
Ok(cache_refreshed) => {
self.clear_caches_after_marketplace_source_refresh(cache_refreshed);
Ok(refresh_outcome) => {
self.clear_caches_after_marketplace_source_refresh(
refresh_outcome.cache_refreshed,
);
outcome
.errors
.extend(refresh_outcome.errors.into_iter().map(|error| {
ConfiguredMarketplaceUpgradeError {
marketplace_name: error.marketplace_name,
message: error.message,
}
}));
}
Err(err) => {
self.clear_cache();
@@ -2048,9 +2097,11 @@ impl PluginsManager {
pub fn maybe_start_non_curated_plugin_cache_refresh(
self: &Arc<Self>,
config: &PluginsConfigInput,
roots: &[AbsolutePathBuf],
) {
self.schedule_non_curated_plugin_cache_refresh(
config,
roots,
NonCuratedCacheRefreshMode::IfVersionChanged,
);
@@ -2127,38 +2178,81 @@ impl PluginsManager {
fn schedule_non_curated_plugin_cache_refresh(
self: &Arc<Self>,
config: &PluginsConfigInput,
roots: &[AbsolutePathBuf],
mode: NonCuratedCacheRefreshMode,
) {
let mut roots = roots.to_vec();
let marketplace_roots =
self.marketplace_roots(config, roots, /*include_openai_curated*/ false);
let outcome = match self.list_marketplaces_with_policy(config, &marketplace_roots) {
Ok(outcome) => outcome,
Err(err) => {
warn!("failed to prepare non-curated plugin cache refresh: {err}");
return;
}
};
let policy = MarketplacePolicy::from_requirements(config.config_layer_stack.requirements());
let mut roots = outcome
.marketplaces
.into_iter()
.filter(|marketplace| !is_openai_curated_marketplace_name(&marketplace.name))
.filter_map(|marketplace| {
match policy.validate_install(
&config.config_layer_stack,
self.codex_home.as_path(),
&marketplace.path,
&marketplace.name,
) {
Ok(()) => Some(marketplace.path),
Err(err) => {
warn!(
marketplace = marketplace.name,
path = %marketplace.path.display(),
error = %err,
"skipping marketplace source during plugin cache refresh"
);
None
}
}
})
.collect::<Vec<_>>();
roots.sort_unstable();
roots.dedup();
if roots.is_empty() {
let mut configured_plugin_keys =
configured_plugins_from_stack(&config.config_layer_stack, self.codex_home.as_path())
.into_keys()
.collect::<Vec<_>>();
configured_plugin_keys.sort_unstable();
if roots.is_empty() || configured_plugin_keys.is_empty() {
return;
}
let request = NonCuratedCacheRefreshRequest { roots, mode };
let mut request = NonCuratedCacheRefreshRequest {
roots,
configured_plugin_keys,
mode,
};
let should_spawn = {
let mut state = match self.non_curated_cache_refresh_state.write() {
Ok(state) => state,
Err(err) => err.into_inner(),
};
// Collapse repeated plugin/list requests onto one worker and only queue another pass
// when the requested roots set actually changes. Forced reinstall requests are not
// deduped against the last completed pass because the same marketplace root path can
// point at newly activated files after an auto-upgrade.
if state.requested.as_ref() == Some(&request)
|| (mode == NonCuratedCacheRefreshMode::IfVersionChanged
&& !state.in_flight
&& state.last_refreshed.as_ref() == Some(&request))
{
return;
}
if mode == NonCuratedCacheRefreshMode::IfVersionChanged
if request.mode == NonCuratedCacheRefreshMode::IfVersionChanged
&& state.requested.as_ref().is_some_and(|requested| {
requested.mode == NonCuratedCacheRefreshMode::ForceReinstall
&& requested.roots == request.roots
})
{
request.mode = NonCuratedCacheRefreshMode::ForceReinstall;
}
// Collapse repeated plugin/list requests onto one worker and only queue another pass
// when the roots or configured plugin set changes. Forced reinstall requests are not
// deduped against the last completed pass because the same marketplace root path can
// point at newly activated files after an auto-upgrade.
if state.requested.as_ref() == Some(&request)
|| (request.mode == NonCuratedCacheRefreshMode::IfVersionChanged
&& !state.in_flight
&& state.last_refreshed.as_ref() == Some(&request))
{
return;
}
@@ -2349,21 +2443,33 @@ impl PluginsManager {
let refresh_result = match request.mode {
NonCuratedCacheRefreshMode::IfVersionChanged => {
refresh_non_curated_plugin_cache(self.codex_home.as_path(), &request.roots)
}
NonCuratedCacheRefreshMode::ForceReinstall => {
refresh_non_curated_plugin_cache_force_reinstall(
refresh_non_curated_plugin_cache_detailed(
self.codex_home.as_path(),
&request.roots,
&request.configured_plugin_keys,
)
}
NonCuratedCacheRefreshMode::ForceReinstall => {
refresh_non_curated_plugin_cache_force_reinstall_detailed(
self.codex_home.as_path(),
&request.roots,
&request.configured_plugin_keys,
)
}
};
let refreshed = match refresh_result {
Ok(cache_refreshed) => {
if cache_refreshed {
Ok(refresh_outcome) => {
if refresh_outcome.cache_refreshed {
self.clear_cache();
}
true
for error in &refresh_outcome.errors {
warn!(
marketplace = error.marketplace_name,
error = %error.message,
"failed to refresh configured plugin cache"
);
}
refresh_outcome.errors.is_empty()
}
Err(err) => {
self.clear_cache();
@@ -2391,7 +2497,8 @@ impl PluginsManager {
&self,
config: &PluginsConfigInput,
) -> (HashSet<String>, HashSet<String>) {
let configured_plugins = configured_plugins_from_stack(&config.config_layer_stack);
let configured_plugins =
configured_plugins_from_stack(&config.config_layer_stack, self.codex_home.as_path());
let installed_plugins = configured_plugins
.keys()
.filter(|plugin_key| {
@@ -2448,6 +2555,27 @@ impl PluginsManager {
roots.dedup();
roots
}
fn list_marketplaces_with_policy(
&self,
config: &PluginsConfigInput,
roots: &[AbsolutePathBuf],
) -> Result<MarketplaceListOutcome, MarketplaceError> {
let mut outcome = list_marketplaces_with_home(roots, home_dir().as_deref())?;
let policy = MarketplacePolicy::from_requirements(config.config_layer_stack.requirements());
if !policy.is_restricted() {
return Ok(outcome);
}
let allowed_marketplace_names = allowed_configured_marketplace_names(
&config.config_layer_stack,
self.codex_home.as_path(),
);
outcome.marketplaces.retain(|marketplace| {
is_openai_curated_marketplace_name(&marketplace.name)
|| allowed_marketplace_names.contains(&marketplace.name)
});
Ok(outcome)
}
}
pub(crate) fn remote_plugin_install_required_description(
@@ -2612,31 +2740,6 @@ impl PluginUninstallError {
}
}
pub(crate) fn configured_plugins_from_stack(
config_layer_stack: &ConfigLayerStack,
) -> HashMap<String, PluginConfig> {
// Plugin entries remain persisted user config only.
let Some(user_config) = config_layer_stack.effective_user_config() else {
return HashMap::new();
};
configured_plugins_from_user_config_value(&user_config)
}
fn configured_plugins_from_user_config_value(
user_config: &toml::Value,
) -> HashMap<String, PluginConfig> {
let Some(plugins_value) = user_config.get("plugins") else {
return HashMap::new();
};
match plugins_value.clone().try_into() {
Ok(plugins) => plugins,
Err(err) => {
warn!("invalid plugins config: {err}");
HashMap::new()
}
}
}
#[cfg(test)]
#[path = "manager_tests.rs"]
mod tests;
codex-rs/core-plugins/src/manager_tests.rs
+249 -3
View File
@@ -100,6 +100,19 @@ fn config_layer_stack_with_requirements(
.expect("build config layer stack")
}
fn plugins_config_input_with_requirements(
codex_home: &Path,
user_config: &str,
requirements: &str,
) -> PluginsConfigInput {
PluginsConfigInput::new(
config_layer_stack_with_requirements(codex_home, user_config, requirements),
/*plugins_enabled*/ true,
/*remote_plugin_enabled*/ false,
String::new(),
)
}
#[test]
fn plugins_manager_tracks_auth_mode() {
let tmp = TempDir::new().unwrap();
@@ -122,6 +135,184 @@ fn plugins_manager_tracks_auth_mode() {
assert_eq!(manager_with_auth.auth_mode(), Some(AuthMode::Chatgpt));
}
#[tokio::test]
async fn marketplace_policy_projection_disables_installed_plugin_and_invalidates_cache() {
let codex_home = TempDir::new().expect("create Codex home");
write_plugin(
&codex_home.path().join("plugins/cache/company"),
"sample/local",
"sample",
);
let user_config = r#"
[marketplaces.company]
source_type = "git"
source = "https://github.com/example/company.git"
[plugins."sample@company"]
enabled = true
"#;
let allowed = plugins_config_input_with_requirements(
codex_home.path(),
user_config,
r#"
[marketplaces]
restrict_to_allowed_sources = true
[marketplaces.allowed_sources.company]
source = "git"
url = "https://github.com/example/company.git"
"#,
);
let blocked = plugins_config_input_with_requirements(
codex_home.path(),
user_config,
r#"
[marketplaces]
restrict_to_allowed_sources = true
[marketplaces.allowed_sources.other]
source = "git"
url = "https://github.com/example/other.git"
"#,
);
let manager = PluginsManager::new(codex_home.path().to_path_buf());
let allowed_outcome = manager.plugins_for_config(&allowed).await;
assert_eq!(allowed_outcome.plugins().len(), 1);
assert_eq!(allowed_outcome.plugins()[0].config_name, "sample@company");
let blocked_outcome = manager.plugins_for_config(&blocked).await;
assert_eq!(blocked_outcome, PluginLoadOutcome::default());
}
#[tokio::test]
async fn plugin_read_rejects_marketplace_blocked_by_requirements() {
let codex_home = TempDir::new().expect("create Codex home");
let marketplace_root = codex_home.path().join("marketplace");
write_plugin(&marketplace_root, "sample", "sample");
write_file(
&marketplace_root.join(".agents/plugins/marketplace.json"),
r#"{
"name": "company",
"plugins": [
{
"name": "sample",
"source": {"source": "local", "path": "./sample"}
}
]
}"#,
);
let config = plugins_config_input_with_requirements(
codex_home.path(),
"",
r#"
[marketplaces]
restrict_to_allowed_sources = true
"#,
);
let marketplace_path =
AbsolutePathBuf::try_from(marketplace_root.join(".agents/plugins/marketplace.json"))
.expect("absolute marketplace path");
let err = PluginsManager::new(codex_home.path().to_path_buf())
.read_plugin_for_config(
&config,
&PluginReadRequest {
plugin_name: "sample".to_string(),
marketplace_path,
},
)
.await
.expect_err("blocked marketplace should not be readable");
assert!(matches!(
err,
MarketplaceError::InvalidMarketplaceFile { .. }
));
}
#[test]
fn marketplace_policy_filters_discovered_marketplaces_by_configured_name() {
let codex_home = TempDir::new().expect("create Codex home");
let repo_root = codex_home.path().join("repo");
let subdirectory = repo_root.join("worktree/subdirectory");
fs::create_dir_all(&subdirectory).expect("create input subdirectory");
write_plugin(&repo_root, "sample", "sample");
write_file(
&repo_root.join(".agents/plugins/marketplace.json"),
r#"{
"name": "company",
"plugins": [
{
"name": "sample",
"source": {"source": "local", "path": "./sample"}
}
]
}"#,
);
init_git_repo(&repo_root);
let repo_root = AbsolutePathBuf::try_from(repo_root).expect("absolute repository root");
let subdirectory =
AbsolutePathBuf::try_from(subdirectory).expect("absolute input subdirectory");
let manager = PluginsManager::new(codex_home.path().to_path_buf());
let user_config = format!(
r#"
[marketplaces.company]
source_type = "local"
source = {:?}
"#,
repo_root.as_path()
);
let allowed = plugins_config_input_with_requirements(
codex_home.path(),
&user_config,
&format!(
r#"
[marketplaces]
restrict_to_allowed_sources = true
[marketplaces.allowed_sources.company]
source = "local"
path = {:?}
"#,
repo_root.as_path()
),
);
let blocked = plugins_config_input_with_requirements(
codex_home.path(),
&user_config,
&format!(
r#"
[marketplaces]
restrict_to_allowed_sources = true
[marketplaces.allowed_sources.subdirectory]
source = "local"
path = {:?}
"#,
subdirectory.as_path()
),
);
let allowed_outcome = manager
.list_marketplaces_for_config(
&allowed,
std::slice::from_ref(&subdirectory),
/*include_openai_curated*/ false,
)
.expect("list allowed marketplace");
assert_eq!(allowed_outcome.marketplaces.len(), 1);
assert_eq!(allowed_outcome.marketplaces[0].name, "company");
let blocked_outcome = manager
.list_marketplaces_for_config(
&blocked,
std::slice::from_ref(&subdirectory),
/*include_openai_curated*/ false,
)
.expect("list blocked marketplace");
assert_eq!(blocked_outcome.marketplaces, Vec::new());
}
fn write_auth_projection_plugin(codex_home: &Path, name: &str, include_app: bool) {
let plugin_root = codex_home
.join("plugins/cache")
@@ -3348,7 +3539,11 @@ plugins = true
.unwrap()
.marketplaces
.into_iter()
.find(|marketplace| marketplace.name == "debug")
.find(|marketplace| {
marketplace.path
== AbsolutePathBuf::try_from(repo_root.join(".agents/plugins/marketplace.json"))
.unwrap()
})
.unwrap()
.plugins
.into_iter()
@@ -3785,7 +3980,7 @@ enabled = true
let marketplaces = PluginsManager::new(tmp.path().to_path_buf())
.list_marketplaces_for_config(
&config,
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&[AbsolutePathBuf::try_from(repo_root.clone()).unwrap()],
/*include_openai_curated*/ true,
)
.unwrap()
@@ -3793,7 +3988,11 @@ enabled = true
let marketplace = marketplaces
.into_iter()
.find(|marketplace| marketplace.name == "debug")
.find(|marketplace| {
marketplace.path
== AbsolutePathBuf::try_from(repo_root.join(".agents/plugins/marketplace.json"))
.unwrap()
})
.expect("debug marketplace should be listed");
let mut plugins = marketplace.plugins;
@@ -5256,6 +5455,7 @@ enabled = true
refresh_non_curated_plugin_cache(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["sample-plugin@debug".to_string()],
)
.expect("cache refresh should succeed")
);
@@ -5308,6 +5508,7 @@ enabled = true
refresh_non_curated_plugin_cache(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["sample-plugin@debug".to_string()],
)
.expect("cache refresh should reinstall missing configured plugin")
);
@@ -5367,6 +5568,7 @@ enabled = true
refresh_non_curated_plugin_cache(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["sample-plugin@debug".to_string()],
)
.expect("cache refresh should materialize configured Git plugin")
);
@@ -5420,6 +5622,7 @@ enabled = true
!refresh_non_curated_plugin_cache(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["sample-plugin@debug".to_string()],
)
.expect("cache refresh should be a no-op when configured plugins are current")
);
@@ -5473,6 +5676,7 @@ enabled = true
refresh_non_curated_plugin_cache_force_reinstall(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["sample-plugin@debug".to_string()],
)
.expect("cache refresh should reinstall unchanged local version")
);
@@ -5531,6 +5735,7 @@ enabled = true
refresh_non_curated_plugin_cache(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["sample-plugin@debug".to_string()],
)
.expect("cache refresh should ignore unrelated invalid plugin manifests")
);
@@ -5542,6 +5747,47 @@ enabled = true
);
}
#[test]
fn refresh_non_curated_plugin_cache_continues_after_plugin_error() {
let tmp = tempfile::tempdir().unwrap();
let repo_root = tmp.path().join("repo");
fs::create_dir_all(repo_root.join(".git")).unwrap();
fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
write_plugin_with_version(&repo_root, "z-good", "z-good", Some("1.2.3"));
write_file(
&repo_root.join(".agents/plugins/marketplace.json"),
r#"{
"name": "debug",
"plugins": [
{
"name": "a-broken",
"source": {
"source": "local",
"path": "./missing"
}
},
{
"name": "z-good",
"source": {
"source": "local",
"path": "./z-good"
}
}
]
}"#,
);
let err = refresh_non_curated_plugin_cache(
tmp.path(),
&[AbsolutePathBuf::try_from(repo_root).unwrap()],
&["a-broken@debug".to_string(), "z-good@debug".to_string()],
)
.expect_err("broken plugin should be reported after refreshing the remaining plugins");
assert!(err.contains("a-broken@debug"));
assert!(tmp.path().join("plugins/cache/debug/z-good/1.2.3").is_dir());
}
#[tokio::test]
async fn load_plugins_ignores_project_config_files() {
let codex_home = TempDir::new().unwrap();
codex-rs/core-plugins/src/marketplace_policy.rs
+118 -1
View File
@@ -18,9 +18,13 @@ use codex_config::MarketplaceAllowedSourceToml;
use codex_config::RequirementSource;
use codex_config::types::MarketplaceConfig;
use codex_config::types::MarketplaceSourceType;
use codex_config::types::PluginConfig;
use codex_plugin::PluginId;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path::paths_match_after_normalization;
use regex::Regex;
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use url::Url;
@@ -70,7 +74,7 @@ impl MarketplacePolicy {
}
}
fn is_restricted(&self) -> bool {
pub(crate) fn is_restricted(&self) -> bool {
self.restricted.is_some()
}
@@ -198,6 +202,104 @@ impl AllowedMarketplaceSource {
}
}
pub(crate) fn project_effective_user_config(
config_layer_stack: &ConfigLayerStack,
codex_home: &Path,
) -> Option<toml::Value> {
let mut user_config = config_layer_stack.effective_user_config()?;
let policy = MarketplacePolicy::from_requirements(config_layer_stack.requirements());
if !policy.is_restricted() {
return Some(user_config);
}
let allowed_marketplace_names =
allowed_configured_marketplace_names_with_policy(&user_config, &policy, codex_home);
let configured_marketplace_names = user_config
.get("marketplaces")
.and_then(toml::Value::as_table)
.map(|marketplaces| marketplaces.keys().cloned().collect::<HashSet<_>>())
.unwrap_or_default();
if let Some(marketplaces) = user_config
.get_mut("marketplaces")
.and_then(toml::Value::as_table_mut)
{
marketplaces
.retain(|marketplace_name, _| allowed_marketplace_names.contains(marketplace_name));
}
if let Some(plugins) = user_config
.get_mut("plugins")
.and_then(toml::Value::as_table_mut)
{
plugins.retain(|plugin_key, _| {
let Ok(plugin_id) = PluginId::parse(plugin_key) else {
return false;
};
(is_openai_curated_marketplace_name(&plugin_id.marketplace_name)
&& !configured_marketplace_names.contains(&plugin_id.marketplace_name))
|| allowed_marketplace_names.contains(&plugin_id.marketplace_name)
});
}
Some(user_config)
}
pub fn allowed_configured_marketplace_names(
config_layer_stack: &ConfigLayerStack,
codex_home: &Path,
) -> HashSet<String> {
let Some(user_config) = config_layer_stack.effective_user_config() else {
return HashSet::new();
};
let policy = MarketplacePolicy::from_requirements(config_layer_stack.requirements());
allowed_configured_marketplace_names_with_policy(&user_config, &policy, codex_home)
}
fn allowed_configured_marketplace_names_with_policy(
user_config: &toml::Value,
policy: &MarketplacePolicy,
codex_home: &Path,
) -> HashSet<String> {
let Some(marketplaces) = user_config
.get("marketplaces")
.and_then(toml::Value::as_table)
else {
return HashSet::new();
};
if !policy.is_restricted() {
return marketplaces.keys().cloned().collect();
}
marketplaces
.iter()
.filter_map(|(marketplace_name, marketplace)| {
let allowed = match managed_marketplace_config_name(codex_home, marketplace) {
Some(expected_name) => expected_name == marketplace_name,
None => policy
.validate_configured_marketplace(marketplace_name, marketplace)
.is_ok(),
};
allowed.then(|| marketplace_name.clone())
})
.collect()
}
pub(crate) fn configured_plugins_from_stack(
config_layer_stack: &ConfigLayerStack,
codex_home: &Path,
) -> HashMap<String, PluginConfig> {
let Some(user_config) = project_effective_user_config(config_layer_stack, codex_home) else {
return HashMap::new();
};
let Some(plugins_value) = user_config.get("plugins") else {
return HashMap::new();
};
match plugins_value.clone().try_into() {
Ok(plugins) => plugins,
Err(err) => {
tracing::warn!("invalid plugins config: {err}");
HashMap::new()
}
}
}
pub(crate) fn validate_marketplace_source_for_add(
codex_home: &Path,
requirements: &ConfigRequirements,
@@ -358,6 +460,21 @@ fn managed_marketplace_name(
managed_local_marketplace_name(codex_home, root.as_path())
}
fn managed_marketplace_config_name(
codex_home: &Path,
marketplace: &toml::Value,
) -> Option<&'static str> {
if marketplace.get("source_type").and_then(toml::Value::as_str) != Some("local") {
return None;
}
let path = marketplace
.get("source")
.and_then(toml::Value::as_str)
.map(Path::new)
.filter(|path| path.is_absolute())?;
managed_local_marketplace_name(codex_home, path)
}
fn managed_local_marketplace_name(codex_home: &Path, root: &Path) -> Option<&'static str> {
for marketplace_name in [
OPENAI_BUNDLED_MARKETPLACE_NAME,
codex-rs/core-plugins/src/marketplace_policy_tests.rs
+191
View File
@@ -383,6 +383,197 @@ restrict_to_allowed_sources = true
assert!(validate_marketplace_name_for_add(expected_name, "other").is_err());
}
#[test]
fn projected_user_config_removes_blocked_marketplaces_and_plugins() {
let codex_home = TempDir::new().expect("create Codex home");
let config_file = AbsolutePathBuf::try_from(codex_home.path().join("config.toml"))
.expect("absolute config path");
let stack = config_layer_stack_with_user_config(
r#"
[marketplaces]
restrict_to_allowed_sources = true
[marketplaces.allowed_sources.company]
source = "git"
url = "https://github.com/example/allowed.git"
"#,
Some((
r#"
[marketplaces.allowed]
source_type = "git"
source = "https://github.com/example/allowed.git"
[marketplaces.blocked]
source_type = "git"
source = "https://github.com/example/blocked.git"
[plugins."sample@allowed"]
enabled = true
[plugins."sample@blocked"]
enabled = true
"#,
config_file,
)),
);
let projected =
project_effective_user_config(&stack, codex_home.path()).expect("project user config");
assert_eq!(
projected["marketplaces"]
.as_table()
.expect("projected marketplaces")
.keys()
.cloned()
.collect::<Vec<_>>(),
vec!["allowed".to_string()]
);
assert_eq!(
configured_plugins_from_stack(&stack, codex_home.path())
.into_keys()
.collect::<Vec<_>>(),
vec!["sample@allowed".to_string()]
);
let raw = stack.effective_user_config().expect("raw user config");
assert!(raw["marketplaces"]["blocked"].is_table());
assert!(raw["plugins"]["sample@blocked"].is_table());
}
#[test]
fn managed_bundled_config_is_retained_only_at_its_owned_path() {
let codex_home = TempDir::new().expect("create Codex home");
let bundled_root = codex_home
.path()
.join(".tmp/bundled-marketplaces")
.join(crate::OPENAI_BUNDLED_MARKETPLACE_NAME);
let config_file = AbsolutePathBuf::try_from(codex_home.path().join("config.toml"))
.expect("absolute config path");
let stack = config_layer_stack_with_user_config(
r#"
[marketplaces]
restrict_to_allowed_sources = true
"#,
Some((
&format!(
r#"
[marketplaces.openai-bundled]
source_type = "local"
source = {bundled_root:?}
[marketplaces.openai-bundled-alpha]
source_type = "local"
source = "/tmp/not-managed"
[marketplaces.evil]
source_type = "local"
source = {bundled_root:?}
[plugins."sample@openai-bundled"]
enabled = true
[plugins."sample@openai-bundled-alpha"]
enabled = true
[plugins."sample@evil"]
enabled = true
"#
),
config_file,
)),
);
let projected =
project_effective_user_config(&stack, codex_home.path()).expect("project user config");
assert_eq!(
projected["marketplaces"]
.as_table()
.expect("projected marketplaces")
.keys()
.cloned()
.collect::<Vec<_>>(),
vec![crate::OPENAI_BUNDLED_MARKETPLACE_NAME.to_string()]
);
assert_eq!(
projected["plugins"]
.as_table()
.expect("projected plugins")
.keys()
.cloned()
.collect::<Vec<_>>(),
vec![format!("sample@{}", crate::OPENAI_BUNDLED_MARKETPLACE_NAME)]
);
}
#[test]
fn allowlisted_config_names_are_not_globally_reserved() {
let codex_home = TempDir::new().expect("create Codex home");
let source_root = TempDir::new().expect("create marketplace root");
let source_root = source_root
.path()
.canonicalize()
.expect("canonical marketplace root");
let config_file = AbsolutePathBuf::try_from(codex_home.path().join("config.toml"))
.expect("absolute config path");
let stack = config_layer_stack_with_user_config(
&format!(
r#"
[marketplaces]
restrict_to_allowed_sources = true
[marketplaces.allowed_sources.local]
source = "local"
path = {source_root:?}
"#
),
Some((
&format!(
r#"
[marketplaces.openai-bundled]
source_type = "local"
source = {source_root:?}
[marketplaces.openai-curated]
source_type = "local"
source = {source_root:?}
[plugins."sample@openai-bundled"]
enabled = true
[plugins."sample@openai-curated"]
enabled = true
"#
),
config_file,
)),
);
let projected =
project_effective_user_config(&stack, codex_home.path()).expect("project user config");
assert_eq!(
projected["marketplaces"]
.as_table()
.expect("projected marketplaces")
.keys()
.cloned()
.collect::<Vec<_>>(),
vec!["openai-bundled".to_string(), "openai-curated".to_string()]
);
assert_eq!(
projected["plugins"]
.as_table()
.expect("projected plugins")
.keys()
.cloned()
.collect::<Vec<_>>(),
vec![
"sample@openai-bundled".to_string(),
"sample@openai-curated".to_string()
]
);
}
#[test]
fn blocked_upgrade_is_rejected_before_marketplace_installation() {
let codex_home = TempDir::new().expect("create Codex home");
codex-rs/core-plugins/src/store.rs
+12 -1
View File
@@ -37,6 +37,7 @@ pub struct PluginInstallResult {
#[derive(Debug, Clone)]
pub struct PluginStore {
codex_home: AbsolutePathBuf,
root: AbsolutePathBuf,
data_root: AbsolutePathBuf,
}
@@ -59,14 +60,24 @@ impl PluginStore {
let data_root =
AbsolutePathBuf::from_absolute_path_checked(codex_home.join(PLUGINS_DATA_DIR))
.map_err(|err| PluginStoreError::io("failed to resolve plugin data root", err))?;
let codex_home = AbsolutePathBuf::from_absolute_path_checked(codex_home)
.map_err(|err| PluginStoreError::io("failed to resolve Codex home", err))?;
Ok(Self { root, data_root })
Ok(Self {
codex_home,
root,
data_root,
})
}
pub fn root(&self) -> &AbsolutePathBuf {
&self.root
}
pub(crate) fn codex_home(&self) -> &AbsolutePathBuf {
&self.codex_home
}
pub fn plugin_base_root(&self, plugin_id: &PluginId) -> AbsolutePathBuf {
self.root
.join(&plugin_id.marketplace_name)
codex-rs/core/src/codex_delegate_tests.rs
+1
View File
@@ -80,6 +80,7 @@ async fn forward_events_cancelled_while_send_blocked_shuts_down_delegate() {
status: None,
call_id: "call-1".to_string(),
name: "tool".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
},
codex-rs/core/src/context_manager/history_tests.rs
+52
View File
@@ -513,6 +513,7 @@ fn for_prompt_strips_images_when_model_does_not_support_images() {
status: None,
call_id: "tool-1".to_string(),
name: "js_repl".to_string(),
namespace: None,
input: "view_image".to_string(),
internal_chat_message_metadata_passthrough: None,
},
@@ -582,6 +583,7 @@ fn for_prompt_strips_images_when_model_does_not_support_images() {
status: None,
call_id: "tool-1".to_string(),
name: "js_repl".to_string(),
namespace: None,
input: "view_image".to_string(),
internal_chat_message_metadata_passthrough: None,
},
@@ -1090,6 +1092,7 @@ fn remove_first_item_handles_custom_tool_pair() {
status: None,
call_id: "tool-1".to_string(),
name: "my_tool".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
},
@@ -1378,6 +1381,7 @@ fn normalize_adds_missing_output_for_custom_tool_call() {
status: None,
call_id: "tool-x".to_string(),
name: "custom".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
}];
@@ -1393,6 +1397,7 @@ fn normalize_adds_missing_output_for_custom_tool_call() {
status: None,
call_id: "tool-x".to_string(),
name: "custom".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
},
@@ -1512,6 +1517,7 @@ fn normalize_mixed_inserts_and_removals() {
status: None,
call_id: "t1".to_string(),
name: "tool".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
},
@@ -1556,6 +1562,7 @@ fn normalize_mixed_inserts_and_removals() {
status: None,
call_id: "t1".to_string(),
name: "tool".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
},
@@ -1622,6 +1629,49 @@ fn normalize_adds_missing_output_for_function_call_inserts_output() {
);
}
#[test]
fn for_prompt_assigns_stable_id_to_synthetic_output_without_reordering_history() {
let items = vec![
ResponseItem::FunctionCall {
id: Some("fc_existing".to_string()),
name: "do_it".to_string(),
namespace: None,
arguments: "{}".to_string(),
call_id: "call-x".to_string(),
internal_chat_message_metadata_passthrough: None,
},
ResponseItem::Message {
id: Some("msg_later".to_string()),
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "later turn".to_string(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
},
];
let first = create_history_with_items(items.clone()).for_prompt(&default_input_modalities());
let second = create_history_with_items(items).for_prompt(&default_input_modalities());
assert_eq!(
first, second,
"repeated prompt projections should assign the same ID to the synthetic output"
);
let [
ResponseItem::FunctionCall { .. },
ResponseItem::FunctionCallOutput { id: Some(id), .. },
ResponseItem::Message { .. },
] = first.as_slice()
else {
panic!("expected the synthetic output between its call and the later message");
};
assert!(
id.starts_with("fco_"),
"the synthetic function call output should use the Responses API output ID prefix"
);
}
#[test]
fn normalize_adds_missing_output_for_tool_search_call() {
let items = vec![ResponseItem::ToolSearchCall {
@@ -1668,6 +1718,7 @@ fn normalize_adds_missing_output_for_custom_tool_call_panics_in_debug() {
status: None,
call_id: "tool-x".to_string(),
name: "custom".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
}];
@@ -1810,6 +1861,7 @@ fn normalize_mixed_inserts_and_removals_panics_in_debug() {
status: None,
call_id: "t1".to_string(),
name: "tool".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
},
codex-rs/core/src/context_manager/normalize.rs
+26 -6
View File
@@ -4,12 +4,15 @@ use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::InputModality;
use std::collections::HashSet;
use uuid::Uuid;
use crate::util::error_or_panic;
use tracing::info;
const IMAGE_CONTENT_OMITTED_PLACEHOLDER: &str =
"image content omitted because you do not support image input";
// Changing this value would change model-visible IDs and invalidate prompt caches.
const SYNTHETIC_OUTPUT_ID_NAMESPACE: Uuid = Uuid::from_u128(0x90d38d3e_6a5b_4d52_bfe2_2f1e634bfac4);
pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
let mut function_output_ids = HashSet::new();
@@ -40,14 +43,14 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
for (idx, item) in items.iter().enumerate() {
match item {
ResponseItem::FunctionCall { call_id, .. }
ResponseItem::FunctionCall { id, call_id, .. }
if !function_output_ids.contains(call_id.as_str()) =>
{
info!("Function call output is missing for call id: {call_id}");
missing_outputs_to_insert.push((
idx,
ResponseItem::FunctionCallOutput {
id: None,
id: synthetic_output_id("fco", id.as_deref()),
call_id: call_id.clone(),
output: FunctionCallOutputPayload::from_text("aborted".to_string()),
internal_chat_message_metadata_passthrough: None,
@@ -55,6 +58,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
));
}
ResponseItem::ToolSearchCall {
id,
call_id: Some(call_id),
..
} if !tool_search_output_ids.contains(call_id.as_str()) => {
@@ -62,7 +66,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
missing_outputs_to_insert.push((
idx,
ResponseItem::ToolSearchOutput {
id: None,
id: synthetic_output_id("tso", id.as_deref()),
call_id: Some(call_id.clone()),
status: "completed".to_string(),
execution: "client".to_string(),
@@ -71,7 +75,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
},
));
}
ResponseItem::CustomToolCall { call_id, .. }
ResponseItem::CustomToolCall { id, call_id, .. }
if !custom_tool_output_ids.contains(call_id.as_str()) =>
{
error_or_panic(format!(
@@ -80,7 +84,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
missing_outputs_to_insert.push((
idx,
ResponseItem::CustomToolCallOutput {
id: None,
id: synthetic_output_id("ctco", id.as_deref()),
call_id: call_id.clone(),
name: None,
output: FunctionCallOutputPayload::from_text("aborted".to_string()),
@@ -90,6 +94,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
}
// LocalShellCall is represented in upstream streams by a FunctionCallOutput
ResponseItem::LocalShellCall {
id,
call_id: Some(call_id),
..
} if !function_output_ids.contains(call_id.as_str()) => {
@@ -99,7 +104,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
missing_outputs_to_insert.push((
idx,
ResponseItem::FunctionCallOutput {
id: None,
id: synthetic_output_id("fco", id.as_deref()),
call_id: call_id.clone(),
output: FunctionCallOutputPayload::from_text("aborted".to_string()),
internal_chat_message_metadata_passthrough: None,
@@ -121,6 +126,21 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
}
}
/// Derives a stable ID for a prompt-only output from its source call's item ID.
///
/// Prompt normalization can run repeatedly without persisting its synthetic
/// outputs, so the namespace and name format must remain stable across retries
/// and resumes to preserve prompt-cache reuse. Returning `None` when the source
/// call has no ID preserves the legacy behavior for older history items.
fn synthetic_output_id(prefix: &str, item_id: Option<&str>) -> Option<String> {
let source_id = item_id.filter(|id| !id.is_empty())?;
let name = format!("{prefix}:{source_id}");
Some(format!(
"{prefix}_{}",
Uuid::new_v5(&SYNTHETIC_OUTPUT_ID_NAMESPACE, name.as_bytes())
))
}
pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItem>) {
let function_call_ids: HashSet<String> = items
.iter()
codex-rs/core/src/session/tests.rs
+1
View File
@@ -10482,6 +10482,7 @@ async fn fatal_tool_error_stops_turn_and_reports_error() {
status: None,
call_id: "call-1".to_string(),
name: "shell_command".to_string(),
namespace: None,
input: "{}".to_string(),
internal_chat_message_metadata_passthrough: None,
};
codex-rs/core/src/session/turn.rs
+8 -2
View File
@@ -2080,8 +2080,14 @@ async fn try_run_sampling_request(
}
}
ResponseEvent::OutputItemAdded(item) => {
if let ResponseItem::CustomToolCall { call_id, name, .. } = &item {
let tool_name = ToolName::plain(name.as_str());
if let ResponseItem::CustomToolCall {
call_id,
name,
namespace,
..
} = &item
{
let tool_name = ToolName::new(namespace.clone(), name.as_str());
active_tool_argument_diff_consumer = tool_runtime
.create_diff_consumer(&tool_name)
.map(|consumer| (call_id.clone(), consumer));
codex-rs/core/src/stream_events_utils_tests.rs
+1
View File
@@ -116,6 +116,7 @@ fn external_context_pollution_items_exclude_local_tool_calls() {
status: None,
call_id: "custom-1".to_string(),
name: "apply_patch".to_string(),
namespace: None,
input: "*** Begin Patch\n*** End Patch\n".to_string(),
internal_chat_message_metadata_passthrough: None,
},
codex-rs/core/src/tools/router.rs
+3 -2
View File
@@ -25,7 +25,7 @@ use tracing::instrument;
pub use crate::tools::context::ToolCallSource;
#[derive(Clone, Debug)]
#[derive(Clone, Debug, PartialEq)]
pub struct ToolCall {
pub tool_name: ToolName,
pub call_id: String,
@@ -147,11 +147,12 @@ impl ToolRouter {
ResponseItem::ToolSearchCall { .. } => Ok(None),
ResponseItem::CustomToolCall {
name,
namespace,
input,
call_id,
..
} => Ok(Some(ToolCall {
tool_name: ToolName::plain(name),
tool_name: ToolName::new(namespace, name),
call_id,
payload: ToolPayload::Custom { input },
})),
codex-rs/core/src/tools/router_tests.rs
+29
View File
@@ -179,6 +179,35 @@ async fn build_tool_call_uses_namespace_for_registry_name() -> anyhow::Result<()
Ok(())
}
#[tokio::test]
async fn build_custom_tool_call_uses_namespace_for_registry_name() -> anyhow::Result<()> {
let tool_name = "exec".to_string();
let call = ToolRouter::build_tool_call(ResponseItem::CustomToolCall {
id: None,
status: None,
call_id: "call-namespace".to_string(),
name: tool_name.clone(),
namespace: Some("mcp__python".to_string()),
input: "print('hello')".to_string(),
internal_chat_message_metadata_passthrough: None,
})?
.expect("custom_tool_call should produce a tool call");
assert_eq!(
call,
ToolCall {
tool_name: ToolName::namespaced("mcp__python", tool_name),
call_id: "call-namespace".to_string(),
payload: ToolPayload::Custom {
input: "print('hello')".to_string(),
},
}
);
Ok(())
}
#[tokio::test]
async fn mcp_parallel_support_uses_handler_data() -> anyhow::Result<()> {
let (_, turn) = make_session_and_context().await;
codex-rs/core/src/turn_timing_tests.rs
+1
View File
@@ -121,6 +121,7 @@ fn response_item_records_turn_ttft_for_first_output_signals() {
status: None,
call_id: "call-2".to_string(),
name: "custom".to_string(),
namespace: None,
input: "echo hi".to_string(),
internal_chat_message_metadata_passthrough: None,
}
codex-rs/core/tests/common/responses.rs
+18
View File
@@ -895,6 +895,24 @@ pub fn ev_custom_tool_call(call_id: &str, name: &str, input: &str) -> Value {
})
}
pub fn ev_custom_tool_call_with_namespace(
call_id: &str,
namespace: &str,
name: &str,
input: &str,
) -> Value {
serde_json::json!({
"type": "response.output_item.done",
"item": {
"type": "custom_tool_call",
"call_id": call_id,
"namespace": namespace,
"name": name,
"input": input
}
})
}
pub fn ev_local_shell_call(call_id: &str, status: &str, command: Vec<&str>) -> Value {
serde_json::json!({
"type": "response.output_item.done",
codex-rs/core/tests/suite/client.rs
+104
View File
@@ -363,6 +363,107 @@ async fn response_item_ids_persist_across_resume_and_preserve_server_ids() -> an
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<()> {
let function_call_id = "missing-output-call";
let thread_id = ThreadId::default();
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
parent_thread_id: None,
timestamp: "2024-01-01T00:00:00Z".to_string(),
cwd: ".".into(),
originator: "test_originator".to_string(),
cli_version: "test_version".to_string(),
model_provider: Some("test-provider".to_string()),
..Default::default()
},
git: None,
}),
},
RolloutLine {
timestamp: "2024-01-01T00:00:01.000Z".to_string(),
item: RolloutItem::ResponseItem(ResponseItem::FunctionCall {
id: Some("fc_existing".to_string()),
name: "do_it".to_string(),
namespace: None,
arguments: "{}".to_string(),
call_id: function_call_id.to_string(),
internal_chat_message_metadata_passthrough: None,
}),
},
];
let tmpdir = TempDir::new()?;
let session_path = tmpdir.path().join("normalized-call-output-item-id.jsonl");
let mut file = std::fs::File::create(&session_path)?;
for line in rollout {
writeln!(file, "{}", serde_json::to_string(&line)?)?;
}
let server = MockServer::start().await;
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]),
],
)
.await;
let codex_home = Arc::new(TempDir::new()?);
let mut builder = test_codex().with_config(|config| {
let _ = config.features.enable(Feature::ItemIds);
});
let first = builder
.resume(&server, Arc::clone(&codex_home), session_path.clone())
.await?;
first.submit_turn("first resume").await?;
first.codex.submit(Op::Shutdown).await?;
wait_for_event(&first.codex, |event| {
matches!(event, EventMsg::ShutdownComplete)
})
.await;
assert!(
!std::fs::read_to_string(&session_path)?.contains("\"type\":\"function_call_output\""),
"prompt-only repair should not be persisted to the rollout"
);
builder = builder.with_config(|config| {
let _ = config.features.enable(Feature::ItemIds);
});
let second = builder.resume(&server, codex_home, session_path).await?;
second.submit_turn("second resume").await?;
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
let first_output = requests[0].function_call_output(function_call_id);
let first_output_id = first_output
.get("id")
.and_then(serde_json::Value::as_str)
.expect("reconstructed output should have an item ID")
.to_string();
let first_output_uuid = first_output_id
.strip_prefix("fco_")
.expect("synthetic output should use the Responses API prefix");
assert_eq!(
Uuid::parse_str(first_output_uuid)?.get_version(),
Some(uuid::Version::Sha1)
);
assert_eq!(
requests[1]
.function_call_output(function_call_id)
.get("id")
.and_then(serde_json::Value::as_str),
Some(first_output_id.as_str())
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn response_item_ids_are_sent_for_all_remote_v2_compaction_requests() -> anyhow::Result<()> {
let server = MockServer::start().await;
@@ -785,6 +886,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
status: None,
call_id: "legacy-js-call".to_string(),
name: "js_repl".to_string(),
namespace: None,
input: "console.log('legacy image flow')".to_string(),
internal_chat_message_metadata_passthrough: None,
};
@@ -976,6 +1078,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
status: Some("completed".to_string()),
call_id: custom_call_id.to_string(),
name: "js_repl".to_string(),
namespace: None,
input: "console.log('image flow')".to_string(),
internal_chat_message_metadata_passthrough: None,
}),
@@ -2912,6 +3015,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
status: Some("completed".into()),
call_id: "custom-tool-call-id".into(),
name: "custom_tool".into(),
namespace: None,
input: "{}".into(),
internal_chat_message_metadata_passthrough: None,
});
codex-rs/core/tests/suite/tools.rs
+75
View File
@@ -20,6 +20,7 @@ use core_test_support::assert_regex_match;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_custom_tool_call;
use core_test_support::responses::ev_custom_tool_call_with_namespace;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once;
@@ -177,6 +178,80 @@ async fn custom_tool_unknown_returns_custom_output_error() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn namespaced_custom_tool_call_preserves_namespace_through_dispatch_and_replay() -> Result<()>
{
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build(&server).await?;
let call_id = "custom-namespaced";
let namespace = "test_namespace::";
let tool_name = "unsupported_tool";
let input = "\"payload\"";
mount_sse_once(
&server,
sse(vec![
ev_response_created("resp-1"),
ev_custom_tool_call_with_namespace(call_id, namespace, tool_name, input),
ev_completed("resp-1"),
]),
)
.await;
let mock = mount_sse_once(
&server,
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
)
.await;
test.submit_turn_with_approval_and_permission_profile(
"invoke namespaced custom tool",
AskForApproval::Never,
PermissionProfile::Disabled,
)
.await?;
let request = mock.single_request();
let custom_tool_calls = request.inputs_of_type("custom_tool_call");
let turn_id = custom_tool_calls
.first()
.and_then(|item| item.pointer("/internal_chat_message_metadata_passthrough/turn_id"))
.and_then(Value::as_str)
.map(str::to_string)
.expect("custom tool call should include turn metadata");
assert_eq!(
(custom_tool_calls, request.custom_tool_call_output(call_id),),
(
vec![json!({
"type": "custom_tool_call",
"call_id": call_id,
"namespace": namespace,
"name": tool_name,
"input": input,
"internal_chat_message_metadata_passthrough": {
"turn_id": turn_id,
},
})],
json!({
"type": "custom_tool_call_output",
"call_id": call_id,
"output": format!("unsupported custom tool call: {namespace}{tool_name}"),
"internal_chat_message_metadata_passthrough": {
"turn_id": turn_id,
},
}),
)
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shell_command_escalated_permissions_rejected_then_ok() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/exec-server/src/client.rs
+22 -14
View File
@@ -110,6 +110,7 @@ mod recovery;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(10);
const ENVIRONMENT_INFO_TIMEOUT: Duration = Duration::from_secs(30);
const PROCESS_EVENT_CHANNEL_CAPACITY: usize = 256;
const PROCESS_EVENT_RETAINED_BYTES: usize = 1024 * 1024;
@@ -510,7 +511,12 @@ impl ExecServerClient {
}
pub async fn environment_info(&self) -> Result<EnvironmentInfo, ExecServerError> {
self.call(ENVIRONMENT_INFO_METHOD, &()).await
let rpc_client = self.inner.rpc_client().await?;
map_rpc_call_result(
rpc_client
.call_with_timeout(ENVIRONMENT_INFO_METHOD, &(), ENVIRONMENT_INFO_TIMEOUT)
.await,
)
}
pub async fn read(&self, params: ReadParams) -> Result<ReadResponse, ExecServerError> {
@@ -780,22 +786,21 @@ impl ExecServerClient {
P: serde::Serialize,
T: serde::de::DeserializeOwned,
{
match rpc_client.call(method, params).await {
Ok(response) => Ok(response),
Err(error) => {
let error = ExecServerError::from(error);
if is_transport_closed_error(&error) {
Err(ExecServerError::Disconnected(disconnected_message(
/*reason*/ None,
)))
} else {
Err(error)
}
}
}
map_rpc_call_result(rpc_client.call(method, params).await)
}
}
fn map_rpc_call_result<T>(result: Result<T, RpcCallError>) -> Result<T, ExecServerError> {
result.map_err(|error| {
let error = ExecServerError::from(error);
if is_transport_closed_error(&error) {
ExecServerError::Disconnected(disconnected_message(/*reason*/ None))
} else {
error
}
})
}
async fn cleanup_process_start(
client: &ExecServerClient,
process_id: &ProcessId,
@@ -822,6 +827,9 @@ impl From<RpcCallError> for ExecServerError {
code: error.code,
message: error.message,
},
RpcCallError::TimedOut { method, timeout } => Self::Protocol(format!(
"timed out waiting for exec-server `{method}` response after {timeout:?}"
)),
}
}
}
codex-rs/exec-server/src/rpc.rs
+84 -2
View File
@@ -5,6 +5,7 @@ use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use std::time::Duration;
use codex_exec_server_protocol::JSONRPCError;
use codex_exec_server_protocol::JSONRPCErrorError;
@@ -21,6 +22,7 @@ use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::sync::watch;
use tokio::task::JoinHandle;
use tokio::time::timeout;
use crate::connection::JsonRpcConnection;
use crate::connection::JsonRpcConnectionEvent;
@@ -36,6 +38,8 @@ pub(crate) enum RpcCallError {
Json(serde_json::Error),
/// The executor returned a JSON-RPC error response for this call.
Server(JSONRPCErrorError),
/// The executor did not return a response before the caller's deadline.
TimedOut { method: String, timeout: Duration },
}
type PendingRequest = oneshot::Sender<Result<Value, RpcCallError>>;
@@ -46,6 +50,11 @@ type RequestRoute<S> = Box<
type NotificationRoute<S> =
Box<dyn Fn(Arc<S>, JSONRPCNotification) -> BoxFuture<Result<(), String>> + Send + Sync>;
enum RpcCallTimeout {
None,
After(Duration),
}
#[derive(Debug)]
pub(crate) enum RpcClientEvent {
Notification(JSONRPCNotification),
@@ -334,6 +343,33 @@ impl RpcClient {
}
pub(crate) async fn call<P, T>(&self, method: &str, params: &P) -> Result<T, RpcCallError>
where
P: Serialize,
T: DeserializeOwned,
{
self.call_inner(method, params, RpcCallTimeout::None).await
}
pub(crate) async fn call_with_timeout<P, T>(
&self,
method: &str,
params: &P,
call_timeout: Duration,
) -> Result<T, RpcCallError>
where
P: Serialize,
T: DeserializeOwned,
{
self.call_inner(method, params, RpcCallTimeout::After(call_timeout))
.await
}
async fn call_inner<P, T>(
&self,
method: &str,
params: &P,
call_timeout: RpcCallTimeout,
) -> Result<T, RpcCallError>
where
P: Serialize,
T: DeserializeOwned,
@@ -379,8 +415,20 @@ impl RpcClient {
// still-pending requests. Awaiting this receiver preserves that order:
// responses already read before EOF still win, and truly pending calls
// are failed once the reader observes the disconnect.
let result: Result<Value, RpcCallError> =
response_rx.await.map_err(|_| RpcCallError::Closed)?;
let response = match call_timeout {
RpcCallTimeout::None => response_rx.await,
RpcCallTimeout::After(call_timeout) => match timeout(call_timeout, response_rx).await {
Ok(response) => response,
Err(_) => {
self.pending.lock().await.remove(&request_id);
return Err(RpcCallError::TimedOut {
method: method.to_string(),
timeout: call_timeout,
});
}
},
};
let result: Result<Value, RpcCallError> = response.map_err(|_| RpcCallError::Closed)?;
let response = match result {
Ok(response) => response,
Err(error) => return Err(error),
@@ -673,6 +721,40 @@ mod tests {
}
}
#[tokio::test]
async fn rpc_client_timeout_removes_pending_request() {
let (client_stdin, server_reader) = tokio::io::duplex(4096);
let (server_writer, client_stdout) = tokio::io::duplex(4096);
let (release_server_tx, release_server_rx) = tokio::sync::oneshot::channel();
let connection =
JsonRpcConnection::from_stdio(client_stdout, client_stdin, "test-rpc".to_string());
let (client, _events_rx) = RpcClient::new(connection);
let server = tokio::spawn(async move {
let mut lines = BufReader::new(server_reader).lines();
let request = read_jsonrpc_line(&mut lines).await;
assert!(matches!(request, JSONRPCMessage::Request(_)));
let _server_writer = server_writer;
let _ = release_server_rx.await;
});
let call_timeout = Duration::from_millis(10);
let result = client
.call_with_timeout::<_, serde_json::Value>("slow", &serde_json::json!({}), call_timeout)
.await;
assert!(matches!(
result,
Err(super::RpcCallError::TimedOut { method, timeout })
if method == "slow" && timeout == call_timeout
));
assert_eq!(client.pending_request_count().await, 0);
let _ = release_server_tx.send(());
if let Err(err) = server.await {
panic!("server task failed: {err}");
}
}
#[tokio::test(flavor = "current_thread")]
async fn rpc_client_propagates_current_trace_context() {
let span_exporter = InMemorySpanExporter::default();
codex-rs/ext/image-generation/src/tests.rs
+1
View File
@@ -97,6 +97,7 @@ async fn recent_image_fallback_selects_newest_images_in_chronological_order() {
status: Some("completed".to_string()),
call_id: "code-mode-call".to_string(),
name: "exec".to_string(),
namespace: None,
input: String::new(),
internal_chat_message_metadata_passthrough: None,
},
codex-rs/protocol/src/models.rs
+3
View File
@@ -1052,6 +1052,9 @@ pub enum ResponseItem {
call_id: String,
name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
namespace: Option<String>,
input: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
codex-rs/tools/src/tool_payload.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use std::borrow::Cow;
use codex_protocol::models::SearchToolCallParams;
/// Canonical payload shapes accepted by model-visible tool runtimes.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, PartialEq)]
pub enum ToolPayload {
Function { arguments: String },
ToolSearch { arguments: SearchToolCallParams },
codex-rs/tui/src/chatwidget/tests/app_server.rs
+3 -2
View File
@@ -1151,8 +1151,8 @@ async fn live_app_server_cyber_policy_error_renders_dedicated_notice() {
let cells = drain_insert_history(&mut rx);
assert_eq!(cells.len(), 1);
let rendered = lines_to_single_string(&cells[0]);
assert!(rendered.contains("This chat was flagged for possible cybersecurity risk"));
assert!(rendered.contains("Trusted Access for Cyber"));
assert!(rendered.contains("This content can't be shown"));
assert!(rendered.contains("extra caution with cybersecurity requests"));
assert!(!rendered.contains("server fallback message"));
assert!(!chat.bottom_pane.is_task_running());
}
@@ -1171,6 +1171,7 @@ async fn app_server_safety_access_errors_render_dedicated_notice() {
assert_eq!(cells.len(), 1);
let rendered = lines_to_single_string(&cells[0]);
assert!(rendered.contains("This content can't be shown"));
assert!(rendered.contains("biological research"));
assert!(!rendered.contains("Invalid prompt:"));
}
}
codex-rs/tui/src/chatwidget/tests/status_and_layout.rs
+2 -2
View File
@@ -54,8 +54,8 @@ async fn app_server_cyber_policy_error_renders_dedicated_notice() {
let cells = drain_insert_history(&mut rx);
assert_eq!(cells.len(), 1);
let rendered = lines_to_single_string(&cells[0]);
assert!(rendered.contains("This chat was flagged for possible cybersecurity risk"));
assert!(rendered.contains("Trusted Access for Cyber"));
assert!(rendered.contains("This content can't be shown"));
assert!(rendered.contains("extra caution with cybersecurity requests"));
assert!(!rendered.contains("server fallback message"));
}
codex-rs/tui/src/history_cell/notices.rs
+59 -74
View File
@@ -86,91 +86,76 @@ pub(crate) fn new_warning_event(message: String) -> PrefixedWrappedHistoryCell {
}
#[derive(Debug)]
pub(crate) struct SafetyAccessBlockCell;
pub(crate) struct SafetyAccessBlockCell {
body: &'static str,
trusted_access_url: &'static str,
}
const SAFETY_ACCESS_BLOCK_TITLE: &str = "This content can't be shown";
const SAFETY_ACCESS_BLOCK_BODY: &str = "We've limited access to this content for safety reasons. This type of information may be used to benefit or to harm people.";
const SAFETY_ACCESS_BLOCK_LEARN_MORE_URL: &str = "https://help.openai.com/en/articles/20001326";
pub(crate) fn new_safety_access_block_event() -> SafetyAccessBlockCell {
SafetyAccessBlockCell
SafetyAccessBlockCell {
body: "We take extra caution with requests involving biological research and applications that could pose safety risks. If you’re a researcher at an approved organization, you may be able to apply for Trusted Access.",
trusted_access_url: "https://openai.com/form/trusted-access-for-life-sciences",
}
}
pub(crate) fn new_cyber_policy_error_event() -> SafetyAccessBlockCell {
SafetyAccessBlockCell {
body: "We take extra caution with cybersecurity requests. If you’re a security professional, you may be able to apply for Trusted Access.",
trusted_access_url: "https://openai.com/form/enterprise-trusted-access-for-cyber/",
}
}
impl HistoryCell for SafetyAccessBlockCell {
fn display_lines(&self, width: u16) -> Vec<Line<'static>> {
let mut lines = vec![vec!["ⓘ ".cyan(), SAFETY_ACCESS_BLOCK_TITLE.bold()].into()];
let body = Line::from(vec![" ".into(), SAFETY_ACCESS_BLOCK_BODY.dim()]);
let wrap_width = width.saturating_sub(2).max(1) as usize;
let wrapped = adaptive_wrap_line(
&body,
RtOptions::new(wrap_width).subsequent_indent(" ".into()),
);
push_owned_lines(&wrapped, &mut lines);
lines
}
fn raw_lines(&self) -> Vec<Line<'static>> {
vec![
Line::from(SAFETY_ACCESS_BLOCK_TITLE),
Line::from(SAFETY_ACCESS_BLOCK_BODY),
]
}
}
const TRUSTED_ACCESS_FOR_CYBER_URL: &str = "https://chatgpt.com/cyber";
#[derive(Debug)]
pub(crate) struct CyberPolicyNoticeCell;
pub(crate) fn new_cyber_policy_error_event() -> CyberPolicyNoticeCell {
CyberPolicyNoticeCell
}
impl HistoryCell for CyberPolicyNoticeCell {
fn display_lines(&self, width: u16) -> Vec<Line<'static>> {
let mut lines: Vec<Line<'static>> = Vec::new();
lines.push(
vec![
"ⓘ ".cyan(),
"This chat was flagged for possible cybersecurity risk".bold(),
]
.into(),
);
let wrap_width = width.saturating_sub(2).max(1) as usize;
let body = Line::from(vec![
" If this seems wrong, try rephrasing your request. To get authorized for security work, join the "
.dim(),
"Trusted Access for Cyber".cyan().underlined(),
" program.".dim(),
]);
let wrapped = adaptive_wrap_line(
&body,
RtOptions::new(wrap_width).subsequent_indent(" ".into()),
);
push_owned_lines(&wrapped, &mut lines);
lines.push(
vec![
" ".into(),
TRUSTED_ACCESS_FOR_CYBER_URL.cyan().underlined(),
]
.into(),
);
lines
}
fn raw_lines(&self) -> Vec<Line<'static>> {
vec![
Line::from("This chat was flagged for possible cybersecurity risk"),
Line::from(
"If this seems wrong, try rephrasing your request. To get authorized for security work, join the Trusted Access for Cyber program.",
),
Line::from(TRUSTED_ACCESS_FOR_CYBER_URL),
]
visible_lines(self.display_hyperlink_lines(width))
}
fn display_hyperlink_lines(&self, width: u16) -> Vec<HyperlinkLine> {
crate::terminal_hyperlinks::annotate_web_urls(self.display_lines(width))
let mut lines = vec![HyperlinkLine::new(
vec!["ⓘ ".cyan(), SAFETY_ACCESS_BLOCK_TITLE.bold()].into(),
)];
let body = Line::from(vec![" ".into(), self.body.dim()]);
let wrap_width = width.saturating_sub(2).max(1) as usize;
let wrapped = adaptive_wrap_line(
&body,
RtOptions::new(wrap_width).subsequent_indent(" ".into()),
);
let mut wrapped_body = Vec::new();
push_owned_lines(&wrapped, &mut wrapped_body);
lines.extend(plain_hyperlink_lines(wrapped_body));
for (label, url) in [
("Trusted Access", self.trusted_access_url),
("Learn more", SAFETY_ACCESS_BLOCK_LEARN_MORE_URL),
] {
let source = crate::terminal_hyperlinks::annotate_web_urls_in_line(
vec![format!(" {label}: ").dim(), url.cyan().underlined()].into(),
);
let wrapped = crate::wrapping::word_wrap_line(
&source.line,
RtOptions::new(wrap_width).subsequent_indent(" ".into()),
);
let mut wrapped_links = Vec::new();
push_owned_lines(&wrapped, &mut wrapped_links);
lines.extend(crate::terminal_hyperlinks::remap_wrapped_line(
&source,
wrapped_links,
));
}
lines
}
fn raw_lines(&self) -> Vec<Line<'static>> {
let trusted_access_url = self.trusted_access_url;
vec![
Line::from(SAFETY_ACCESS_BLOCK_TITLE),
Line::from(self.body),
Line::from(format!("Trusted Access: {trusted_access_url}")),
Line::from(format!("Learn more: {SAFETY_ACCESS_BLOCK_LEARN_MORE_URL}")),
]
}
fn transcript_hyperlink_lines(&self, width: u16) -> Vec<HyperlinkLine> {
codex-rs/tui/src/history_cell/snapshots/codex_tui__history_cell__tests__cyber_policy_error_event_narrow_snapshot.snap
+13 -9
View File
@@ -1,12 +1,16 @@
---
source: tui/src/history_cell.rs
assertion_line: 3312
source: tui/src/history_cell/tests.rs
expression: rendered
---
ⓘ This chat was flagged for possible cybersecurity risk
If this seems wrong, try
rephrasing your request. To get
authorized for security work,
join the Trusted Access for
Cyber program.
https://chatgpt.com/cyber
ⓘ This content can't be shown
We take extra caution with
cybersecurity requests. If
you’re a security professional,
you may be able to apply for
Trusted Access.
Trusted Access: https://
openai.com/form/enterprise-
trusted-access-for-cyber/
Learn more: https://
help.openai.com/en/
articles/20001326
codex-rs/tui/src/history_cell/snapshots/codex_tui__history_cell__tests__cyber_policy_error_event_snapshot.snap
+5 -4
View File
@@ -3,7 +3,8 @@ source: tui/src/history_cell.rs
assertion_line: 3305
expression: rendered
---
ⓘ This chat was flagged for possible cybersecurity risk
If this seems wrong, try rephrasing your request. To get authorized for
security work, join the Trusted Access for Cyber program.
https://chatgpt.com/cyber
ⓘ This content can't be shown
We take extra caution with cybersecurity requests. If you’re a security
professional, you may be able to apply for Trusted Access.
Trusted Access: https://openai.com/form/enterprise-trusted-access-for-cyber/
Learn more: https://help.openai.com/en/articles/20001326
codex-rs/tui/src/history_cell/snapshots/codex_tui__history_cell__tests__safety_access_block_event_snapshot.snap
+5 -2
View File
@@ -3,5 +3,8 @@ source: tui/src/history_cell/tests.rs
expression: rendered
---
ⓘ This content can't be shown
We've limited access to this content for safety reasons. This type of
information may be used to benefit or to harm people.
We take extra caution with requests involving biological research and
applications that could pose safety risks. If you’re a researcher at an
approved organization, you may be able to apply for Trusted Access.
Trusted Access: https://openai.com/form/trusted-access-for-life-sciences
Learn more: https://help.openai.com/en/articles/20001326