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

Move connector runtime into connectors crate

Browse Source
This commit is contained in:
Matthew Zeng
2026-07-09 22:38:52 -07:00
parent d15ebdd08d
commit 4f6fb1bbf1
11 changed files with 515 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+5
View File
@@ -2580,15 +2580,20 @@ name = "codex-connectors"
version = "0.0.0"
dependencies = [
"anyhow",
"arc-swap",
"codex-config",
"codex-otel",
"codex-plugin",
"codex-protocol",
"indexmap 2.14.0",
"pretty_assertions",
"serde",
"serde_json",
"sha1 0.10.6",
"tempfile",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"tracing",
"urlencoding",
]
codex-rs/codex-mcp/src/connection_manager_tests.rs
+71 -70
View File
@@ -76,14 +76,23 @@ fn create_codex_apps_tools_cache_context(
) -> CodexAppsToolsCacheContext {
CodexAppsToolsCache::default().context(
codex_home,
CodexAppsToolsCacheKey {
account_id: account_id.map(ToOwned::to_owned),
chatgpt_user_id: chatgpt_user_id.map(ToOwned::to_owned),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
account_id.map(ToOwned::to_owned),
chatgpt_user_id.map(ToOwned::to_owned),
),
)
}
fn store_current_tools(cache_context: &CodexAppsToolsCacheContext, tools: Vec<ToolInfo>) {
cache_context
.publish_if_newest_accepted(
cache_context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools,
)
.expect("publish cached tools");
}
fn create_test_server_info(title: &str) -> McpServerInfo {
McpServerInfo {
name: "codex-apps".to_string(),
@@ -200,7 +209,7 @@ fn create_test_manager_with_failed_apps_startup(
Some("reconnect-test-account"),
Some("reconnect-test-user"),
);
cache_context.store_current_tools_for_test(cached_tools);
store_current_tools(&cache_context, cached_tools);
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
let permission_profile = Constrained::allow_any(PermissionProfile::default());
let mut manager = McpConnectionManager::new_uninitialized(
@@ -812,10 +821,13 @@ async fn list_all_tools_uses_shared_codex_apps_cache_while_client_is_pending() {
Some("account-one"),
Some("user-one"),
);
cache_context.store_current_tools_for_test(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"calendar_create_event",
)]);
store_current_tools(
&cache_context,
vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"calendar_create_event",
)],
);
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
.boxed()
.shared();
@@ -1116,7 +1128,7 @@ async fn list_all_tools_does_not_block_when_shared_codex_apps_cache_is_empty() {
Some("account-one"),
Some("user-one"),
);
cache_context.store_current_tools_for_test(Vec::new());
store_current_tools(&cache_context, Vec::new());
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
.boxed()
.shared();
@@ -1157,10 +1169,13 @@ async fn list_all_tools_uses_shared_codex_apps_cache_when_client_startup_fails()
Some("account-one"),
Some("user-one"),
);
cache_context.store_current_tools_for_test(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"calendar_create_event",
)]);
store_current_tools(
&cache_context,
vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"calendar_create_event",
)],
);
let server_info = create_test_server_info("Codex Apps");
let failed_client = futures::future::ready::<Result<ManagedClient, StartupOutcomeError>>(Err(
StartupOutcomeError::Failed {
@@ -1219,11 +1234,10 @@ async fn context_discard_while_checking_failed_startup_does_not_reconnect() {
let cache = CodexAppsToolsCache::default();
let context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-one".to_string()),
Some("user-one".to_string()),
),
);
let (failure_started_tx, failure_started_rx) = tokio::sync::oneshot::channel();
let release_failure = Arc::new(tokio::sync::Notify::new());
@@ -1272,11 +1286,10 @@ async fn context_discard_while_checking_failed_startup_does_not_reconnect() {
.expect("failed startup check should begin");
let _new_context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-two".to_string()),
Some("user-two".to_string()),
),
);
release_failure.notify_one();
reconnect_task.await.expect("reconnect check should finish");
@@ -1566,11 +1579,10 @@ async fn discarded_codex_apps_context_hides_tools_and_rejects_calls() {
let cache = CodexAppsToolsCache::default();
let context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-one".to_string()),
Some("user-one".to_string()),
),
);
let mut client = create_ready_async_managed_client(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
@@ -1594,11 +1606,10 @@ async fn discarded_codex_apps_context_hides_tools_and_rejects_calls() {
let _new_context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-two".to_string()),
Some("user-two".to_string()),
),
);
assert!(manager.list_all_tools().await.is_empty());
@@ -1623,11 +1634,10 @@ async fn context_discard_during_startup_hides_codex_apps_tools() {
let cache = CodexAppsToolsCache::default();
let context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-one".to_string()),
Some("user-one".to_string()),
),
);
let (client, startup_started, release_startup) = create_blocked_codex_apps_client(
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "old_tool")],
@@ -1656,11 +1666,10 @@ async fn context_discard_during_startup_hides_codex_apps_tools() {
.expect("tool listing should await startup");
let _new_context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-two".to_string()),
Some("user-two".to_string()),
),
);
release_startup.notify_one();
@@ -1677,11 +1686,10 @@ async fn context_discard_during_startup_is_not_reported_ready() {
let cache = CodexAppsToolsCache::default();
let context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-one".to_string()),
Some("user-one".to_string()),
),
);
let (client, startup_started, release_startup) =
create_blocked_codex_apps_client(Vec::new(), context, ToolFilter::default()).await;
@@ -1710,11 +1718,10 @@ async fn context_discard_during_startup_is_not_reported_ready() {
.expect("readiness check should await startup");
let _new_context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-two".to_string()),
Some("user-two".to_string()),
),
);
release_startup.notify_one();
@@ -1732,11 +1739,10 @@ async fn context_discard_during_startup_rejects_codex_apps_calls() {
let cache = CodexAppsToolsCache::default();
let context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-one".to_string()),
Some("user-one".to_string()),
),
);
let (client, startup_started, release_startup) = create_blocked_codex_apps_client(
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "old_tool")],
@@ -1777,11 +1783,10 @@ async fn context_discard_during_startup_rejects_codex_apps_calls() {
.expect("tool call should await startup");
let _new_context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(
Some("account-two".to_string()),
Some("user-two".to_string()),
),
);
release_startup.notify_one();
@@ -1934,11 +1939,7 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() {
),
codex_home.path().to_path_buf(),
CodexAppsToolsCache::default(),
CodexAppsToolsCacheKey {
account_id: None,
chatgpt_user_id: None,
is_workspace_account: false,
},
CodexAppsToolsCacheKey::personal(/*account_id*/ None, /*chatgpt_user_id*/ None),
/*prefix_mcp_tool_names*/ true,
ElicitationCapability::default(),
/*supports_openai_form_elicitation*/ false,
codex-rs/codex-mcp/src/connector_runtime/mod.rs
+52
View File
@@ -0,0 +1,52 @@
//! MCP-facing aliases for the connector-owned runtime.
use codex_login::CodexAuth;
use codex_protocol::mcp::McpServerInfo;
use crate::tools::ToolInfo;
impl codex_connectors::ConnectorRuntimePayload for ToolInfo {
const TOOLS_CACHE_DIR: &'static str = "cache/codex_apps_tools";
const TOOLS_CACHE_SCHEMA_VERSION: u8 = 4;
const SERVER_INFO_CACHE_DIR: &'static str = "cache/codex_apps_server_info";
const SERVER_INFO_CACHE_SCHEMA_VERSION: u8 = 1;
}
pub type ConnectorRuntimeContextKey = codex_connectors::ConnectorRuntimeContextKey;
pub type ConnectorRuntimeManager = codex_connectors::ConnectorRuntimeManager<ToolInfo>;
pub type ConnectorRuntimeSnapshot = codex_connectors::ConnectorRuntimeSnapshot<ToolInfo>;
/// Compatibility alias for existing cache call sites.
pub type CodexAppsToolsCacheKey = ConnectorRuntimeContextKey;
/// Compatibility alias for existing cache call sites.
pub type CodexAppsToolsCache = ConnectorRuntimeManager;
pub(crate) type CodexAppsToolsCacheContext = codex_connectors::ConnectorRuntimeContext<ToolInfo>;
pub(crate) use codex_connectors::ConnectorRuntimeFetchSource as CodexAppsToolsFetchSource;
/// Builds the CodexAuth-backed connector runtime context key.
pub fn connector_runtime_context_key(auth: Option<&CodexAuth>) -> ConnectorRuntimeContextKey {
let account_id = auth.and_then(CodexAuth::get_account_id);
let chatgpt_user_id = auth.and_then(CodexAuth::get_chatgpt_user_id);
if auth.is_some_and(CodexAuth::is_workspace_account) {
ConnectorRuntimeContextKey::workspace(account_id, chatgpt_user_id)
} else {
ConnectorRuntimeContextKey::personal(account_id, chatgpt_user_id)
}
}
/// Compatibility helper for existing cache call sites.
pub fn codex_apps_tools_cache_key(auth: Option<&CodexAuth>) -> CodexAppsToolsCacheKey {
connector_runtime_context_key(auth)
}
pub(crate) fn load_startup_cached_codex_apps_server_info(
cache_context: &CodexAppsToolsCacheContext,
) -> Option<McpServerInfo> {
cache_context.cached_server_info()
}
#[cfg(test)]
#[path = "tests.rs"]
mod tests;
codex-rs/codex-mcp/src/connector_runtime/tests.rs
+98
View File
@@ -0,0 +1,98 @@
use std::sync::Arc;
use codex_protocol::mcp::McpServerInfo;
use pretty_assertions::assert_eq;
use rmcp::model::JsonObject;
use rmcp::model::Tool;
use tempfile::tempdir;
use super::*;
const LEGACY_CACHE_KEY_SHA1: &str = "3ee2e36a9f9ee95357626b2c0ec2c72f069c2258";
#[test]
fn pre_move_v4_tool_info_cache_cold_loads_from_legacy_path() {
let codex_home = tempdir().expect("tempdir");
let tools_cache_dir = codex_home.path().join("cache/codex_apps_tools");
let server_info_cache_dir = codex_home.path().join("cache/codex_apps_server_info");
std::fs::create_dir_all(&tools_cache_dir).expect("create tools cache directory");
std::fs::create_dir_all(&server_info_cache_dir).expect("create server info cache directory");
std::fs::write(
tools_cache_dir.join(format!("{LEGACY_CACHE_KEY_SHA1}.json")),
r#"{
"schema_version": 4,
"tools": [
{
"server_name": "codex_apps",
"supports_parallel_tool_calls": true,
"server_origin": "https://chatgpt.com",
"tool_name": "calendar_search",
"tool_namespace": "codex_apps",
"namespace_description": "Calendar",
"tool": {
"name": "calendar.search",
"description": "Search the calendar",
"inputSchema": {}
},
"connector_id": "connector-calendar",
"connector_name": "Calendar",
"plugin_display_names": ["Calendar plugin"]
}
]
}"#,
)
.expect("write legacy tools cache");
std::fs::write(
server_info_cache_dir.join(format!("{LEGACY_CACHE_KEY_SHA1}.json")),
r#"{
"schema_version": 1,
"server_info": {
"name": "codex-apps",
"title": "Codex Apps",
"version": "1.0.0",
"description": null,
"icons": null,
"websiteUrl": null
}
}"#,
)
.expect("write legacy server info cache");
let expected_server_info = McpServerInfo {
name: "codex-apps".to_string(),
title: Some("Codex Apps".to_string()),
version: "1.0.0".to_string(),
description: None,
icons: None,
website_url: None,
};
let expected_tool = ToolInfo {
server_name: "codex_apps".to_string(),
supports_parallel_tool_calls: true,
server_origin: Some("https://chatgpt.com".to_string()),
callable_name: "calendar_search".to_string(),
callable_namespace: "codex_apps".to_string(),
namespace_description: Some("Calendar".to_string()),
tool: Tool::new(
"calendar.search".to_string(),
"Search the calendar".to_string(),
Arc::new(JsonObject::default()),
),
connector_id: Some("connector-calendar".to_string()),
connector_name: Some("Calendar".to_string()),
plugin_display_names: vec!["Calendar plugin".to_string()],
};
let reloaded = ConnectorRuntimeManager::default().context(
codex_home.path().to_path_buf(),
ConnectorRuntimeContextKey::personal(
Some("account-one".to_string()),
Some("user-one".to_string()),
),
);
assert_eq!(
serde_json::to_value(reloaded.current_tools()).expect("serialize reloaded tools"),
serde_json::to_value(Some(vec![expected_tool])).expect("serialize expected tools"),
);
assert_eq!(reloaded.cached_server_info(), Some(expected_server_info));
}
codex-rs/codex-mcp/src/lib.rs
-1
View File
@@ -84,7 +84,6 @@ mod catalog;
pub(crate) mod codex_apps;
pub(crate) mod connection_manager;
mod connector_runtime;
mod connector_runtime_persistence;
pub(crate) mod elicitation;
pub(crate) mod mcp;
mod plugin_config;
codex-rs/codex-mcp/src/tools.rs
-3
View File
@@ -22,9 +22,6 @@ use tracing::warn;
use crate::mcp::sanitize_responses_api_tool_name;
pub(crate) const MCP_TOOLS_CACHE_WRITE_DURATION_METRIC: &str =
"codex.mcp.tools.cache_write.duration_ms";
const LEGACY_MCP_TOOL_NAME_PREFIX: &str = "mcp__";
#[derive(Debug, Clone, Serialize, Deserialize)]
codex-rs/connectors/Cargo.toml
+6 -1
View File
@@ -9,18 +9,23 @@ workspace = true
[dependencies]
anyhow = { workspace = true }
arc-swap = { workspace = true }
codex-config = { workspace = true }
codex-otel = { workspace = true }
codex-plugin = { workspace = true }
codex-protocol = { workspace = true }
indexmap = { workspace = true, features = ["serde"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
sha1 = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
tokio-util = { workspace = true, features = ["rt"] }
tracing = { workspace = true }
urlencoding = { workspace = true }
[dev-dependencies]
pretty_assertions = { workspace = true }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
[lib]
codex-rs/codex-mcp/src/connector_runtime.rs → codex-rs/connectors/src/connector_runtime/mod.rs
+145 -133
View File
@@ -1,9 +1,9 @@
//! Shared runtime snapshot for the host-owned Codex Apps MCP server.
//! Shared runtime snapshot for connector-backed MCP tools.
//!
//! Runtime snapshots are process-local live state scoped by the active Codex
//! auth context. Disk is best-effort cold-start persistence; a context reads it
//! once when activated and never rereads it. Full connector metadata is owned by
//! the connector metadata store, not by this module.
//! Runtime snapshots are process-local live state scoped by the active account
//! and workspace. Disk is best-effort cold-start persistence; a context reads
//! it once when activated and never rereads it. Full connector metadata is
//! owned by the connector metadata store, not by this module.
use std::path::PathBuf;
use std::sync::Arc;
@@ -15,34 +15,69 @@ use std::time::Instant;
use std::time::SystemTime;
use arc_swap::ArcSwapOption;
use codex_login::CodexAuth;
use codex_protocol::mcp::McpServerInfo;
use serde::Deserialize;
use serde::Serialize;
use serde::de::DeserializeOwned;
use tokio_util::sync::CancellationToken;
use crate::connector_runtime_persistence::load_cached_codex_apps_server_info;
use crate::connector_runtime_persistence::load_cached_connector_runtime_for_identity;
use crate::connector_runtime_persistence::persist_codex_apps_cache;
use crate::connector_runtime_persistence::server_info_cache_path;
use crate::connector_runtime_persistence::tools_cache_path;
use crate::runtime::emit_duration;
use crate::tools::ToolInfo;
use self::persistence::load_cached_codex_apps_server_info;
use self::persistence::load_cached_connector_runtime_for_identity;
use self::persistence::persist_codex_apps_cache;
use self::persistence::server_info_cache_path;
use self::persistence::tools_cache_path;
const MCP_TOOLS_CACHE_PUBLISH_DURATION_METRIC: &str = "codex.mcp.tools.cache_publish.duration_ms";
/// Defines the stable on-disk contract for one connector runtime payload.
///
/// Implementations must use cache directories that are unique to their payload
/// shape and bump the corresponding schema version whenever that shape changes.
pub trait ConnectorRuntimePayload: Clone + Serialize + DeserializeOwned {
const TOOLS_CACHE_DIR: &'static str;
const TOOLS_CACHE_SCHEMA_VERSION: u8;
const SERVER_INFO_CACHE_DIR: &'static str;
const SERVER_INFO_CACHE_SCHEMA_VERSION: u8;
}
/// The account and workspace identity of a connector runtime catalog.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ConnectorRuntimeContextKey {
account_id: Option<String>,
chatgpt_user_id: Option<String>,
is_workspace_account: bool,
}
impl ConnectorRuntimeContextKey {
pub fn personal(account_id: Option<String>, chatgpt_user_id: Option<String>) -> Self {
Self {
account_id,
chatgpt_user_id,
is_workspace_account: false,
}
}
pub fn workspace(account_id: Option<String>, chatgpt_user_id: Option<String>) -> Self {
Self {
account_id,
chatgpt_user_id,
is_workspace_account: true,
}
}
}
/// One atomically published connector runtime state.
///
/// Tools remain raw and in response order. Local and managed configuration is
/// intentionally applied by readers rather than persisted in this snapshot.
#[derive(Debug, Clone)]
pub struct ConnectorRuntimeSnapshot {
pub(crate) tools: Vec<ToolInfo>,
pub(crate) refreshed_at: SystemTime,
pub struct ConnectorRuntimeSnapshot<T> {
tools: Vec<T>,
refreshed_at: SystemTime,
}
impl ConnectorRuntimeSnapshot {
pub fn tools(&self) -> &[ToolInfo] {
impl<T> ConnectorRuntimeSnapshot<T> {
pub fn tools(&self) -> &[T] {
&self.tools
}
@@ -57,61 +92,45 @@ impl ConnectorRuntimeSnapshot {
}
}
/// The CodexAuth bits that identify a connector runtime catalog.
///
/// Debug bearer-token overrides bypass the shared runtime manager, so shared
/// snapshots only need the CodexAuth-backed identity.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ConnectorRuntimeContextKey {
pub(crate) account_id: Option<String>,
pub(crate) chatgpt_user_id: Option<String>,
pub(crate) is_workspace_account: bool,
}
/// Builds the CodexAuth-backed connector runtime context key.
pub fn connector_runtime_context_key(auth: Option<&CodexAuth>) -> ConnectorRuntimeContextKey {
ConnectorRuntimeContextKey {
account_id: auth.and_then(CodexAuth::get_account_id),
chatgpt_user_id: auth.and_then(CodexAuth::get_chatgpt_user_id),
is_workspace_account: auth.is_some_and(CodexAuth::is_workspace_account),
}
}
/// Compatibility alias for existing cache call sites.
pub type CodexAppsToolsCacheKey = ConnectorRuntimeContextKey;
/// Compatibility helper for existing cache call sites.
pub fn codex_apps_tools_cache_key(auth: Option<&CodexAuth>) -> CodexAppsToolsCacheKey {
connector_runtime_context_key(auth)
}
/// Process-scoped owner of the active account/workspace connector runtime.
///
/// Activating a different context discards the prior in-memory entry. Handles
/// to a discarded context can no longer read or publish its snapshot, which
/// prevents account A state from bleeding into account B.
#[derive(Clone, Default)]
pub struct ConnectorRuntimeManager {
active: Arc<Mutex<Option<Arc<ConnectorRuntimeEntry>>>>,
pub struct ConnectorRuntimeManager<T: ConnectorRuntimePayload> {
active: Arc<Mutex<Option<Arc<ConnectorRuntimeEntry<T>>>>>,
}
/// Compatibility alias for existing cache call sites.
pub type CodexAppsToolsCache = ConnectorRuntimeManager;
impl<T: ConnectorRuntimePayload> Clone for ConnectorRuntimeManager<T> {
fn clone(&self) -> Self {
Self {
active: Arc::clone(&self.active),
}
}
}
impl ConnectorRuntimeManager {
impl<T: ConnectorRuntimePayload> Default for ConnectorRuntimeManager<T> {
fn default() -> Self {
Self {
active: Arc::new(Mutex::new(None)),
}
}
}
impl<T: ConnectorRuntimePayload> ConnectorRuntimeManager<T> {
pub fn current_snapshot(
&self,
codex_home: PathBuf,
key: ConnectorRuntimeContextKey,
) -> Option<Arc<ConnectorRuntimeSnapshot>> {
) -> Option<Arc<ConnectorRuntimeSnapshot<T>>> {
self.context(codex_home, key).current_snapshot()
}
pub(crate) fn context(
pub fn context(
&self,
codex_home: PathBuf,
key: ConnectorRuntimeContextKey,
) -> ConnectorRuntimeContext {
) -> ConnectorRuntimeContext<T> {
let identity = ConnectorRuntimeIdentity { codex_home, key };
let mut active = lock_unpoisoned(&self.active);
if let Some(active) = active.as_ref()
@@ -136,50 +155,39 @@ impl ConnectorRuntimeManager {
}
/// Handle to the active account/workspace connector runtime.
#[derive(Clone)]
pub(crate) struct ConnectorRuntimeContext {
active: Arc<Mutex<Option<Arc<ConnectorRuntimeEntry>>>>,
pub(crate) entry: Arc<ConnectorRuntimeEntry>,
pub struct ConnectorRuntimeContext<T: ConnectorRuntimePayload> {
active: Arc<Mutex<Option<Arc<ConnectorRuntimeEntry<T>>>>>,
entry: Arc<ConnectorRuntimeEntry<T>>,
}
/// Compatibility alias for existing cache call sites.
pub(crate) type CodexAppsToolsCacheContext = ConnectorRuntimeContext;
impl<T: ConnectorRuntimePayload> Clone for ConnectorRuntimeContext<T> {
fn clone(&self) -> Self {
Self {
active: Arc::clone(&self.active),
entry: Arc::clone(&self.entry),
}
}
}
impl ConnectorRuntimeContext {
pub fn current_snapshot(&self) -> Option<Arc<ConnectorRuntimeSnapshot>> {
impl<T: ConnectorRuntimePayload> ConnectorRuntimeContext<T> {
pub fn current_snapshot(&self) -> Option<Arc<ConnectorRuntimeSnapshot<T>>> {
let active = lock_unpoisoned(&self.active);
self.matches_active(active.as_ref())
.then(|| self.entry.current_snapshot.load_full())
.flatten()
}
pub(crate) fn tools_cache_path(&self) -> PathBuf {
tools_cache_path(&self.entry.identity)
}
pub(crate) fn server_info_cache_path(&self) -> PathBuf {
server_info_cache_path(&self.entry.identity)
}
pub(crate) fn current_tools(&self) -> Option<Vec<ToolInfo>> {
self.current_snapshot()
.map(|snapshot| snapshot.tools.clone())
}
pub(crate) fn has_current_tools(&self) -> bool {
pub fn has_current_tools(&self) -> bool {
self.current_snapshot().is_some()
}
/// Shared by MCP clients attached to this context and cancelled before an identity switch.
pub(crate) fn routing_cancellation_token(&self) -> CancellationToken {
/// Shared by clients attached to this context and cancelled before an identity switch.
pub fn routing_cancellation_token(&self) -> CancellationToken {
self.entry.routing_cancellation_token.clone()
}
pub(crate) fn begin_fetch(
&self,
source: CodexAppsToolsFetchSource,
) -> CodexAppsToolsFetchTicket {
CodexAppsToolsFetchTicket {
pub fn begin_fetch(&self, source: ConnectorRuntimeFetchSource) -> ConnectorRuntimeFetchTicket {
ConnectorRuntimeFetchTicket {
generation: self
.entry
.next_fetch_generation
@@ -189,12 +197,33 @@ impl ConnectorRuntimeContext {
}
}
pub(crate) fn publish_runtime_if_newest_accepted(
pub fn cached_server_info(&self) -> Option<McpServerInfo> {
load_cached_codex_apps_server_info(self)
}
fn tools_cache_path(&self) -> PathBuf {
tools_cache_path::<T>(&self.entry.identity)
}
fn server_info_cache_path(&self) -> PathBuf {
server_info_cache_path::<T>(&self.entry.identity)
}
fn matches_active(&self, active: Option<&Arc<ConnectorRuntimeEntry<T>>>) -> bool {
active.is_some_and(|active| Arc::ptr_eq(active, &self.entry))
}
pub fn current_tools(&self) -> Option<Vec<T>> {
self.current_snapshot()
.map(|snapshot| snapshot.tools.clone())
}
pub fn publish_runtime_if_newest_accepted(
&self,
ticket: CodexAppsToolsFetchTicket,
ticket: ConnectorRuntimeFetchTicket,
server_info: &McpServerInfo,
tools: Vec<ToolInfo>,
) -> Result<Arc<ConnectorRuntimeSnapshot>, ConnectorRuntimeContextDiscarded> {
tools: Vec<T>,
) -> Result<Arc<ConnectorRuntimeSnapshot<T>>, ConnectorRuntimeDiscarded> {
self.publish_runtime_if_newest_accepted_with(
ticket,
server_info,
@@ -205,11 +234,11 @@ impl ConnectorRuntimeContext {
fn publish_runtime_if_newest_accepted_with(
&self,
ticket: CodexAppsToolsFetchTicket,
ticket: ConnectorRuntimeFetchTicket,
server_info: &McpServerInfo,
tools: Vec<ToolInfo>,
persist: impl FnOnce(&ConnectorRuntimeContext, &McpServerInfo, &ConnectorRuntimeSnapshot),
) -> Result<Arc<ConnectorRuntimeSnapshot>, ConnectorRuntimeContextDiscarded> {
tools: Vec<T>,
persist: impl FnOnce(&ConnectorRuntimeContext<T>, &McpServerInfo, &ConnectorRuntimeSnapshot<T>),
) -> Result<Arc<ConnectorRuntimeSnapshot<T>>, ConnectorRuntimeDiscarded> {
let publish_start = Instant::now();
let active = lock_unpoisoned(&self.active);
if !self.matches_active(active.as_ref()) {
@@ -219,7 +248,7 @@ impl ConnectorRuntimeContext {
publish_start.elapsed(),
&[("source", ticket.source.as_str()), ("result", "discarded")],
);
return Err(ConnectorRuntimeContextDiscarded);
return Err(ConnectorRuntimeDiscarded);
}
let mut last_accepted_generation = lock_unpoisoned(&self.entry.last_accepted_generation);
@@ -231,9 +260,7 @@ impl ConnectorRuntimeContext {
publish_start.elapsed(),
&[("source", ticket.source.as_str()), ("result", "stale")],
);
return self
.current_snapshot()
.ok_or(ConnectorRuntimeContextDiscarded);
return self.current_snapshot().ok_or(ConnectorRuntimeDiscarded);
}
let snapshot = Arc::new(ConnectorRuntimeSnapshot {
@@ -258,40 +285,24 @@ impl ConnectorRuntimeContext {
Ok(snapshot)
}
pub(crate) fn publish_if_newest_accepted(
pub fn publish_if_newest_accepted(
&self,
ticket: CodexAppsToolsFetchTicket,
ticket: ConnectorRuntimeFetchTicket,
server_info: &McpServerInfo,
tools: Vec<ToolInfo>,
) -> Result<Vec<ToolInfo>, ConnectorRuntimeContextDiscarded> {
tools: Vec<T>,
) -> Result<Vec<T>, ConnectorRuntimeDiscarded> {
self.publish_runtime_if_newest_accepted(ticket, server_info, tools)
.map(|snapshot| snapshot.tools.clone())
}
#[cfg(test)]
pub(crate) fn store_current_tools_for_test(&self, tools: Vec<ToolInfo>) {
if !self.matches_active(lock_unpoisoned(&self.active).as_ref()) {
return;
}
let snapshot = ConnectorRuntimeSnapshot {
tools,
refreshed_at: SystemTime::now(),
};
self.entry.current_snapshot.store(Some(Arc::new(snapshot)));
}
fn matches_active(&self, active: Option<&Arc<ConnectorRuntimeEntry>>) -> bool {
active.is_some_and(|active| Arc::ptr_eq(active, &self.entry))
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum CodexAppsToolsFetchSource {
pub enum ConnectorRuntimeFetchSource {
Startup,
HardRefresh,
}
impl CodexAppsToolsFetchSource {
impl ConnectorRuntimeFetchSource {
fn as_str(self) -> &'static str {
match self {
Self::Startup => "startup",
@@ -300,25 +311,25 @@ impl CodexAppsToolsFetchSource {
}
}
pub(crate) struct CodexAppsToolsFetchTicket {
pub struct ConnectorRuntimeFetchTicket {
generation: u64,
source: CodexAppsToolsFetchSource,
source: ConnectorRuntimeFetchSource,
}
#[derive(Debug, Clone, Copy, thiserror::Error)]
#[error("connector runtime context was discarded")]
pub(crate) struct ConnectorRuntimeContextDiscarded;
pub struct ConnectorRuntimeDiscarded;
/// All live state owned by one activated connector identity.
pub(crate) struct ConnectorRuntimeEntry {
pub(crate) identity: ConnectorRuntimeIdentity,
pub(crate) current_snapshot: ArcSwapOption<ConnectorRuntimeSnapshot>,
struct ConnectorRuntimeEntry<T: ConnectorRuntimePayload> {
identity: ConnectorRuntimeIdentity,
current_snapshot: ArcSwapOption<ConnectorRuntimeSnapshot<T>>,
routing_cancellation_token: CancellationToken,
next_fetch_generation: AtomicU64,
last_accepted_generation: Mutex<u64>,
}
impl ConnectorRuntimeEntry {
impl<T: ConnectorRuntimePayload> ConnectorRuntimeEntry<T> {
fn new(identity: ConnectorRuntimeIdentity) -> Self {
let current_snapshot = load_cached_connector_runtime_for_identity(&identity).map(Arc::new);
Self {
@@ -336,15 +347,15 @@ impl ConnectorRuntimeEntry {
/// The auth key says whose runtime catalog we are reading. `codex_home` keeps
/// the persisted cache under the right home directory.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) struct ConnectorRuntimeIdentity {
pub(crate) codex_home: PathBuf,
pub(crate) key: ConnectorRuntimeContextKey,
struct ConnectorRuntimeIdentity {
codex_home: PathBuf,
key: ConnectorRuntimeContextKey,
}
pub(crate) fn load_startup_cached_codex_apps_server_info(
cache_context: &CodexAppsToolsCacheContext,
) -> Option<McpServerInfo> {
load_cached_codex_apps_server_info(cache_context)
fn emit_duration(metric: &str, duration: Duration, tags: &[(&str, &str)]) {
if let Some(metrics) = codex_otel::global() {
let _ = metrics.record_duration(metric, duration, tags);
}
}
fn lock_unpoisoned<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
@@ -353,6 +364,7 @@ fn lock_unpoisoned<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
mod persistence;
#[cfg(test)]
#[path = "connector_runtime_tests.rs"]
mod tests;
codex-rs/codex-mcp/src/connector_runtime_persistence.rs → codex-rs/connectors/src/connector_runtime/persistence.rs
+63 -53
View File
@@ -21,25 +21,25 @@ use sha1::Sha1;
use tempfile::NamedTempFile;
use tracing::instrument;
use crate::connector_runtime::CodexAppsToolsCacheContext;
use crate::connector_runtime::ConnectorRuntimeIdentity;
use crate::connector_runtime::ConnectorRuntimeSnapshot;
use crate::runtime::emit_duration;
use crate::tools::MCP_TOOLS_CACHE_WRITE_DURATION_METRIC;
use crate::tools::ToolInfo;
use super::ConnectorRuntimeContext;
use super::ConnectorRuntimeIdentity;
use super::ConnectorRuntimePayload;
use super::ConnectorRuntimeSnapshot;
use super::emit_duration;
const MCP_TOOLS_CACHE_WRITE_DURATION_METRIC: &str = "codex.mcp.tools.cache_write.duration_ms";
pub(crate) const CODEX_APPS_TOOLS_CACHE_MAX_BYTES: u64 = 32 * 1024 * 1024;
const CODEX_APPS_TOOLS_CACHE_DIR: &str = "cache/codex_apps_tools";
pub(crate) const CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION: u8 = 4;
const CODEX_APPS_SERVER_INFO_CACHE_DIR: &str = "cache/codex_apps_server_info";
const CODEX_APPS_SERVER_INFO_CACHE_SCHEMA_VERSION: u8 = 1;
pub(crate) fn tools_cache_path(identity: &ConnectorRuntimeIdentity) -> PathBuf {
cache_path_in(identity, CODEX_APPS_TOOLS_CACHE_DIR)
pub(crate) fn tools_cache_path<T: ConnectorRuntimePayload>(
identity: &ConnectorRuntimeIdentity,
) -> PathBuf {
cache_path_in(identity, T::TOOLS_CACHE_DIR)
}
pub(crate) fn server_info_cache_path(identity: &ConnectorRuntimeIdentity) -> PathBuf {
cache_path_in(identity, CODEX_APPS_SERVER_INFO_CACHE_DIR)
pub(crate) fn server_info_cache_path<T: ConnectorRuntimePayload>(
identity: &ConnectorRuntimeIdentity,
) -> PathBuf {
cache_path_in(identity, T::SERVER_INFO_CACHE_DIR)
}
fn cache_path_in(identity: &ConnectorRuntimeIdentity, cache_dir: &str) -> PathBuf {
@@ -54,27 +54,28 @@ fn cache_path_in(identity: &ConnectorRuntimeIdentity, cache_dir: &str) -> PathBu
}
#[instrument(level = "trace", skip_all)]
pub(crate) fn load_cached_connector_runtime_for_identity(
pub(crate) fn load_cached_connector_runtime_for_identity<T: ConnectorRuntimePayload>(
identity: &ConnectorRuntimeIdentity,
) -> Option<ConnectorRuntimeSnapshot> {
let cache_path = tools_cache_path(identity);
) -> Option<ConnectorRuntimeSnapshot<T>> {
let cache_path = tools_cache_path::<T>(identity);
let (bytes, modified_at) = read_bounded_cache_file(&cache_path).ok()?;
let cache: CodexAppsToolsDiskCache = serde_json::from_slice(&bytes).ok()?;
(cache.schema_version == CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION).then_some(
ConnectorRuntimeSnapshot {
tools: cache.tools,
refreshed_at: modified_at,
},
)
let cache: CodexAppsToolsDiskCache<T> = serde_json::from_slice(&bytes).ok()?;
(cache.schema_version == T::TOOLS_CACHE_SCHEMA_VERSION).then_some(ConnectorRuntimeSnapshot {
tools: cache.tools,
refreshed_at: modified_at,
})
}
pub(crate) fn write_cached_connector_runtime(
cache_context: &CodexAppsToolsCacheContext,
snapshot: &ConnectorRuntimeSnapshot,
) -> anyhow::Result<()> {
pub(crate) fn write_cached_connector_runtime<T>(
cache_context: &ConnectorRuntimeContext<T>,
snapshot: &ConnectorRuntimeSnapshot<T>,
) -> anyhow::Result<()>
where
T: ConnectorRuntimePayload,
{
let cache_path = cache_context.tools_cache_path();
let bytes = serde_json::to_vec_pretty(&CodexAppsToolsDiskCache {
schema_version: CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION,
schema_version: T::TOOLS_CACHE_SCHEMA_VERSION,
tools: snapshot.tools.clone(),
})
.context("failed to serialize connector runtime cache")?;
@@ -82,33 +83,34 @@ pub(crate) fn write_cached_connector_runtime(
}
#[instrument(level = "trace", skip_all)]
pub(crate) fn load_cached_codex_apps_server_info(
cache_context: &CodexAppsToolsCacheContext,
pub(crate) fn load_cached_codex_apps_server_info<T: ConnectorRuntimePayload>(
cache_context: &ConnectorRuntimeContext<T>,
) -> Option<McpServerInfo> {
let (bytes, _) = read_bounded_cache_file(&cache_context.server_info_cache_path()).ok()?;
let cache: CodexAppsServerInfoDiskCache = serde_json::from_slice(&bytes).ok()?;
(cache.schema_version == CODEX_APPS_SERVER_INFO_CACHE_SCHEMA_VERSION)
.then_some(cache.server_info)
(cache.schema_version == T::SERVER_INFO_CACHE_SCHEMA_VERSION).then_some(cache.server_info)
}
fn write_cached_codex_apps_server_info(
cache_context: &CodexAppsToolsCacheContext,
fn write_cached_codex_apps_server_info<T: ConnectorRuntimePayload>(
cache_context: &ConnectorRuntimeContext<T>,
server_info: &McpServerInfo,
) -> anyhow::Result<()> {
let cache_path = cache_context.server_info_cache_path();
let bytes = serde_json::to_vec_pretty(&CodexAppsServerInfoDiskCache {
schema_version: CODEX_APPS_SERVER_INFO_CACHE_SCHEMA_VERSION,
schema_version: T::SERVER_INFO_CACHE_SCHEMA_VERSION,
server_info: server_info.clone(),
})
.context("failed to serialize Codex Apps server info cache")?;
write_codex_apps_cache_file(&cache_path, "server info", bytes)
}
pub(crate) fn persist_codex_apps_cache(
cache_context: &CodexAppsToolsCacheContext,
pub(crate) fn persist_codex_apps_cache<T>(
cache_context: &ConnectorRuntimeContext<T>,
server_info: &McpServerInfo,
snapshot: &ConnectorRuntimeSnapshot,
) {
snapshot: &ConnectorRuntimeSnapshot<T>,
) where
T: ConnectorRuntimePayload,
{
let cache_write_start = Instant::now();
let tools_result = write_cached_connector_runtime(cache_context, snapshot);
if let Err(err) = &tools_result {
@@ -199,9 +201,9 @@ fn write_codex_apps_cache_file(
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct CodexAppsToolsDiskCache {
struct CodexAppsToolsDiskCache<T> {
schema_version: u8,
tools: Vec<ToolInfo>,
tools: Vec<T>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -218,11 +220,13 @@ fn sha1_hex(s: &str) -> String {
}
#[cfg(test)]
pub(crate) fn write_cached_codex_apps_tools_for_test(
cache_context: &CodexAppsToolsCacheContext,
pub(crate) fn write_cached_codex_apps_tools_for_test<T>(
cache_context: &ConnectorRuntimeContext<T>,
server_info: &McpServerInfo,
tools: &[ToolInfo],
) {
tools: &[T],
) where
T: ConnectorRuntimePayload,
{
let snapshot = ConnectorRuntimeSnapshot {
tools: tools.to_vec(),
refreshed_at: SystemTime::now(),
@@ -235,18 +239,24 @@ pub(crate) fn write_cached_codex_apps_tools_for_test(
}
#[cfg(test)]
pub(crate) fn read_cached_codex_apps_tools(
cache_context: &CodexAppsToolsCacheContext,
) -> Option<Vec<ToolInfo>> {
pub(crate) fn read_cached_codex_apps_tools<T>(
cache_context: &ConnectorRuntimeContext<T>,
) -> Option<Vec<T>>
where
T: ConnectorRuntimePayload,
{
load_cached_connector_runtime_for_identity(&cache_context.entry.identity)
.map(|snapshot| snapshot.tools)
}
#[cfg(test)]
pub(crate) fn write_cached_codex_apps_tools(
cache_context: &CodexAppsToolsCacheContext,
tools: &[ToolInfo],
) -> anyhow::Result<()> {
pub(crate) fn write_cached_codex_apps_tools<T>(
cache_context: &ConnectorRuntimeContext<T>,
tools: &[T],
) -> anyhow::Result<()>
where
T: ConnectorRuntimePayload,
{
let snapshot = ConnectorRuntimeSnapshot {
tools: tools.to_vec(),
refreshed_at: SystemTime::now(),
codex-rs/codex-mcp/src/connector_runtime_tests.rs → codex-rs/connectors/src/connector_runtime/tests.rs
+66 -90
View File
@@ -1,39 +1,41 @@
use super::persistence::CODEX_APPS_TOOLS_CACHE_MAX_BYTES;
use super::persistence::read_cached_codex_apps_tools;
use super::persistence::write_cached_codex_apps_tools;
use super::persistence::write_cached_codex_apps_tools_for_test;
use super::*;
use crate::connector_runtime_persistence::CODEX_APPS_TOOLS_CACHE_MAX_BYTES;
use crate::connector_runtime_persistence::CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION;
use crate::connector_runtime_persistence::read_cached_codex_apps_tools;
use crate::connector_runtime_persistence::write_cached_codex_apps_tools;
use crate::connector_runtime_persistence::write_cached_codex_apps_tools_for_test;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::tools::ToolInfo;
use codex_protocol::ToolName;
use codex_protocol::mcp::McpServerInfo;
use pretty_assertions::assert_eq;
use rmcp::model::JsonObject;
use rmcp::model::Tool;
use std::collections::HashSet;
use serde::Deserialize;
use serde::Serialize;
#[cfg(unix)]
use std::os::unix::ffi::OsStringExt;
use std::path::PathBuf;
use std::sync::Arc;
use tempfile::tempdir;
fn create_test_tool(server_name: &str, tool_name: &str) -> ToolInfo {
ToolInfo {
const CODEX_APPS_MCP_SERVER_NAME: &str = "codex_apps";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct TestTool {
server_name: String,
callable_name: String,
connector_id: Option<String>,
connector_name: Option<String>,
}
impl ConnectorRuntimePayload for TestTool {
const TOOLS_CACHE_DIR: &'static str = "cache/connector_runtime_test_tools";
const TOOLS_CACHE_SCHEMA_VERSION: u8 = 1;
const SERVER_INFO_CACHE_DIR: &'static str = "cache/connector_runtime_test_server_info";
const SERVER_INFO_CACHE_SCHEMA_VERSION: u8 = 1;
}
fn create_test_tool(server_name: &str, tool_name: &str) -> TestTool {
TestTool {
server_name: server_name.to_string(),
supports_parallel_tool_calls: false,
server_origin: None,
callable_name: tool_name.to_string(),
callable_namespace: server_name.to_string(),
namespace_description: None,
tool: Tool::new(
tool_name.to_string(),
format!("Test tool: {tool_name}"),
Arc::new(JsonObject::default()),
),
connector_id: None,
connector_name: None,
plugin_display_names: Vec::new(),
}
}
@@ -42,7 +44,7 @@ fn create_test_tool_with_connector(
tool_name: &str,
connector_id: &str,
connector_name: Option<&str>,
) -> ToolInfo {
) -> TestTool {
let mut tool = create_test_tool(server_name, tool_name);
tool.connector_id = Some(connector_id.to_string());
tool.connector_name = connector_name.map(ToOwned::to_owned);
@@ -53,10 +55,10 @@ fn create_codex_apps_tools_cache_context(
codex_home: PathBuf,
account_id: Option<&str>,
chatgpt_user_id: Option<&str>,
) -> CodexAppsToolsCacheContext {
CodexAppsToolsCache::default().context(
) -> ConnectorRuntimeContext<TestTool> {
ConnectorRuntimeManager::<TestTool>::default().context(
codex_home,
CodexAppsToolsCacheKey {
ConnectorRuntimeContextKey {
account_id: account_id.map(ToOwned::to_owned),
chatgpt_user_id: chatgpt_user_id.map(ToOwned::to_owned),
is_workspace_account: false,
@@ -75,13 +77,6 @@ fn create_test_server_info(title: &str) -> McpServerInfo {
}
}
fn model_tool_names(tools: &[ToolInfo]) -> HashSet<ToolName> {
tools
.iter()
.map(ToolInfo::canonical_tool_name)
.collect::<HashSet<_>>()
}
#[test]
fn codex_apps_tools_cache_is_overwritten_by_last_write() {
let codex_home = tempdir().expect("tempdir");
@@ -193,7 +188,7 @@ fn codex_apps_tools_cache_is_ignored_when_schema_version_mismatches() {
std::fs::create_dir_all(parent).expect("create parent");
}
let bytes = serde_json::to_vec_pretty(&serde_json::json!({
"schema_version": CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION + 1,
"schema_version": TestTool::TOOLS_CACHE_SCHEMA_VERSION + 1,
"tools": [create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "one")],
}))
.expect("serialize");
@@ -242,7 +237,7 @@ fn startup_cached_codex_apps_tools_loads_from_disk_cache() {
let startup_tools = cache_context
.current_tools()
.expect("expected startup snapshot to load from cache");
let cached_server_info = load_startup_cached_codex_apps_server_info(&cache_context);
let cached_server_info = cache_context.cached_server_info();
assert_eq!(startup_tools.len(), 1);
assert_eq!(startup_tools[0].server_name, CODEX_APPS_MCP_SERVER_NAME);
@@ -263,7 +258,7 @@ fn startup_cached_codex_apps_tools_loads_without_server_info_cache() {
std::fs::create_dir_all(parent).expect("create parent");
}
let bytes = serde_json::to_vec_pretty(&serde_json::json!({
"schema_version": CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION,
"schema_version": TestTool::TOOLS_CACHE_SCHEMA_VERSION,
"tools": [create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "calendar_search")],
}))
.expect("serialize");
@@ -277,7 +272,7 @@ fn startup_cached_codex_apps_tools_loads_without_server_info_cache() {
let startup_tools = cache_context
.current_tools()
.expect("legacy startup snapshot should remain available");
let cached_server_info = load_startup_cached_codex_apps_server_info(&cache_context);
let cached_server_info = cache_context.cached_server_info();
assert_eq!(startup_tools.len(), 1);
assert_eq!(startup_tools[0].callable_name, "calendar_search");
@@ -307,7 +302,7 @@ fn codex_apps_server_info_cache_survives_legacy_tools_cache_write() {
std::fs::create_dir_all(parent).expect("create parent");
}
let bytes = serde_json::to_vec_pretty(&serde_json::json!({
"schema_version": CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION - 1,
"schema_version": TestTool::TOOLS_CACHE_SCHEMA_VERSION - 1,
"tools": [create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "calendar_search")],
}))
.expect("serialize");
@@ -319,7 +314,7 @@ fn codex_apps_server_info_cache_survives_legacy_tools_cache_write() {
);
assert_eq!(
load_startup_cached_codex_apps_server_info(&startup_cache_context),
startup_cache_context.cached_server_info(),
Some(server_info)
);
assert!(startup_cache_context.current_tools().is_none());
@@ -359,10 +354,10 @@ fn codex_apps_tools_cache_context_does_not_reread_disk_after_creation() {
#[test]
fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
let codex_home = tempdir().expect("tempdir");
let cache = CodexAppsToolsCache::default();
let cache = ConnectorRuntimeManager::<TestTool>::default();
let cache_context_1 = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
ConnectorRuntimeContextKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
@@ -370,14 +365,14 @@ fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
);
let cache_context_2 = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
ConnectorRuntimeContextKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
);
let older_ticket = cache_context_1.begin_fetch(CodexAppsToolsFetchSource::Startup);
let newer_ticket = cache_context_2.begin_fetch(CodexAppsToolsFetchSource::HardRefresh);
let older_ticket = cache_context_1.begin_fetch(ConnectorRuntimeFetchSource::Startup);
let newer_ticket = cache_context_2.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh);
let server_info = create_test_server_info("Codex Apps");
let newer_tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newer")];
let older_tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "older")];
@@ -385,14 +380,7 @@ fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
let published_tools = cache_context_2
.publish_if_newest_accepted(newer_ticket, &server_info, newer_tools)
.expect("publish newer tools");
assert_eq!(
model_tool_names(&published_tools),
model_tool_names(
&cache_context_1
.current_tools()
.expect("new snapshot should publish")
)
);
assert_eq!(cache_context_1.current_tools(), Some(published_tools));
let current_tools = cache_context_1
.publish_if_newest_accepted(older_ticket, &server_info, older_tools)
.expect("stale publish should return current tools");
@@ -414,9 +402,9 @@ fn codex_apps_tools_cache_keeps_live_publish_when_disk_persistence_fails() {
let codex_home = tempdir().expect("tempdir");
let codex_home_file = codex_home.path().join("not-a-directory");
std::fs::write(&codex_home_file, b"occupied").expect("create codex home file");
let cache_context = CodexAppsToolsCache::default().context(
let cache_context = ConnectorRuntimeManager::<TestTool>::default().context(
codex_home_file,
CodexAppsToolsCacheKey {
ConnectorRuntimeContextKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
@@ -425,17 +413,14 @@ fn codex_apps_tools_cache_keeps_live_publish_when_disk_persistence_fails() {
let tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "live")];
let published_tools = cache_context
.publish_if_newest_accepted(
cache_context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh),
cache_context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools.clone(),
)
.expect("live publish should survive persistence failure");
assert_eq!(model_tool_names(&published_tools), model_tool_names(&tools));
assert_eq!(
model_tool_names(&cache_context.current_tools().expect("live snapshot")),
model_tool_names(&tools)
);
assert_eq!(published_tools, tools);
assert_eq!(cache_context.current_tools(), Some(tools));
}
#[cfg(unix)]
@@ -444,10 +429,10 @@ fn codex_apps_tools_cache_scopes_non_utf8_home_disk_paths() {
let codex_home = PathBuf::from(std::ffi::OsString::from_vec(
b"/tmp/codex-home-\xff".to_vec(),
));
let cache = CodexAppsToolsCache::default();
let cache = ConnectorRuntimeManager::<TestTool>::default();
let user_one_context = cache.context(
codex_home.clone(),
CodexAppsToolsCacheKey {
ConnectorRuntimeContextKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
@@ -455,7 +440,7 @@ fn codex_apps_tools_cache_scopes_non_utf8_home_disk_paths() {
);
let user_two_context = cache.context(
codex_home,
CodexAppsToolsCacheKey {
ConnectorRuntimeContextKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
@@ -466,16 +451,13 @@ fn codex_apps_tools_cache_scopes_non_utf8_home_disk_paths() {
user_two_context.tools_cache_path(),
];
assert_eq!(
cache_paths.iter().collect::<HashSet<_>>().len(),
cache_paths.len()
);
assert_ne!(cache_paths[0], cache_paths[1]);
}
#[test]
fn activating_new_context_discards_old_context_without_state_bleed() {
let codex_home = tempdir().expect("tempdir");
let manager = ConnectorRuntimeManager::default();
let manager = ConnectorRuntimeManager::<TestTool>::default();
let context_a = manager.context(
codex_home.path().to_path_buf(),
ConnectorRuntimeContextKey {
@@ -487,7 +469,7 @@ fn activating_new_context_discards_old_context_without_state_bleed() {
let tools_a = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "tool-a")];
let snapshot_a = context_a
.publish_runtime_if_newest_accepted(
context_a.begin_fetch(CodexAppsToolsFetchSource::HardRefresh),
context_a.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools_a.clone(),
)
@@ -519,7 +501,7 @@ fn activating_new_context_discards_old_context_without_state_bleed() {
assert!(
context_a
.publish_runtime_if_newest_accepted(
context_a.begin_fetch(CodexAppsToolsFetchSource::HardRefresh),
context_a.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "stale-a")],
)
@@ -529,20 +511,14 @@ fn activating_new_context_discards_old_context_without_state_bleed() {
let tools_b = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "tool-b")];
let snapshot_b = context_b
.publish_runtime_if_newest_accepted(
context_b.begin_fetch(CodexAppsToolsFetchSource::HardRefresh),
context_b.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools_b.clone(),
)
.expect("publish context B");
assert_eq!(
model_tool_names(snapshot_a.tools()),
model_tool_names(&tools_a)
);
assert_eq!(
model_tool_names(snapshot_b.tools()),
model_tool_names(&tools_b)
);
assert_eq!(snapshot_a.tools(), &tools_a);
assert_eq!(snapshot_b.tools(), &tools_b);
assert!(Arc::ptr_eq(
&snapshot_b,
&context_b.current_snapshot().expect("context B snapshot")
@@ -552,7 +528,7 @@ fn activating_new_context_discards_old_context_without_state_bleed() {
#[test]
fn discarded_compatibility_publish_reports_an_error() {
let codex_home = tempdir().expect("tempdir");
let manager = ConnectorRuntimeManager::default();
let manager = ConnectorRuntimeManager::<TestTool>::default();
let context_a = manager.context(
codex_home.path().to_path_buf(),
ConnectorRuntimeContextKey {
@@ -561,7 +537,7 @@ fn discarded_compatibility_publish_reports_an_error() {
is_workspace_account: false,
},
);
let ticket = context_a.begin_fetch(CodexAppsToolsFetchSource::HardRefresh);
let ticket = context_a.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh);
let _context_b = manager.context(
codex_home.path().to_path_buf(),
ConnectorRuntimeContextKey {
@@ -577,7 +553,7 @@ fn discarded_compatibility_publish_reports_an_error() {
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "stale-a")],
),
Err(ConnectorRuntimeContextDiscarded)
Err(ConnectorRuntimeDiscarded)
));
}
@@ -626,7 +602,7 @@ fn cold_loaded_snapshot_uses_cache_modification_time() {
);
let snapshot = reloaded.current_snapshot().expect("cold-loaded snapshot");
assert_eq!(model_tool_names(snapshot.tools()), model_tool_names(&tools));
assert_eq!(snapshot.tools(), &tools);
assert_eq!(snapshot.refreshed_at(), modified_at);
}
#[test]
@@ -637,8 +613,8 @@ fn accepted_generations_finish_persistence_in_order() {
Some("account-one"),
Some("user-one"),
);
let older_ticket = context.begin_fetch(CodexAppsToolsFetchSource::Startup);
let newer_ticket = context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh);
let older_ticket = context.begin_fetch(ConnectorRuntimeFetchSource::Startup);
let newer_ticket = context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh);
let (older_persisting_tx, older_persisting_rx) = std::sync::mpsc::channel();
let (release_older_tx, release_older_rx) = std::sync::mpsc::channel();
let older_context = context.clone();
@@ -700,7 +676,7 @@ fn accepted_generations_finish_persistence_in_order() {
#[test]
fn workspace_context_change_discards_matching_account_and_user_ids() {
let codex_home = tempdir().expect("tempdir");
let manager = ConnectorRuntimeManager::default();
let manager = ConnectorRuntimeManager::<TestTool>::default();
let personal_context = manager.context(
codex_home.path().to_path_buf(),
ConnectorRuntimeContextKey {
@@ -711,7 +687,7 @@ fn workspace_context_change_discards_matching_account_and_user_ids() {
);
personal_context
.publish_runtime_if_newest_accepted(
personal_context.begin_fetch(CodexAppsToolsFetchSource::Startup),
personal_context.begin_fetch(ConnectorRuntimeFetchSource::Startup),
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "personal")],
)
@@ -738,8 +714,8 @@ fn live_publish_sets_timestamp_and_stale_publish_preserves_it() {
Some("account-one"),
Some("user-one"),
);
let stale_ticket = context.begin_fetch(CodexAppsToolsFetchSource::Startup);
let current_ticket = context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh);
let stale_ticket = context.begin_fetch(ConnectorRuntimeFetchSource::Startup);
let current_ticket = context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh);
let before = SystemTime::now();
let current = context
.publish_runtime_if_newest_accepted(
codex-rs/connectors/src/lib.rs
+9
View File
@@ -11,6 +11,7 @@ use serde::Serialize;
pub mod accessible;
mod app_info;
mod app_tool_policy;
mod connector_runtime;
mod directory_cache;
pub mod filter;
pub mod merge;
@@ -28,6 +29,14 @@ pub use app_tool_policy::AppToolPolicyEvaluator;
pub use app_tool_policy::AppToolPolicyInput;
pub use app_tool_policy::app_is_enabled;
pub use app_tool_policy::apps_config_from_layer_stack;
pub use connector_runtime::ConnectorRuntimeContext;
pub use connector_runtime::ConnectorRuntimeContextKey;
pub use connector_runtime::ConnectorRuntimeDiscarded;
pub use connector_runtime::ConnectorRuntimeFetchSource;
pub use connector_runtime::ConnectorRuntimeFetchTicket;
pub use connector_runtime::ConnectorRuntimeManager;
pub use connector_runtime::ConnectorRuntimePayload;
pub use connector_runtime::ConnectorRuntimeSnapshot;
pub use directory_cache::ConnectorDirectoryCacheContext;
pub use plugin_config::parse_plugin_app_config;
pub use plugin_config::parse_plugin_app_config_value;