openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-14 21:40:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refresh plugin recommendations when tooling state changes (#61387)

Browse Source 
## Why

Cached recommendations can keep offering a plugin after installation. Refreshing the shared catalog during step preparation can also leave the recommendation prompt and install handler with different candidates.

## What changed

- Check `tooling_changed_at` before reusing recommendations, with a one-second reuse window and a 250 ms deadline that includes authentication. Refresh the catalog when the marker changes, preserving cached recommendations on failure and retrying on a later poll.
- Scope recommendation and featured-plugin caches by product, backend, and account identity, and share concurrent recommendation refreshes.
- Use the same recommendation snapshot for a step's prompt and install handler, while allowing later steps to observe updated recommendations.

## Testing

Add coverage for marker changes, missing markers, refresh failures, concurrent and cancelled requests, and the authentication deadline. Extend integration tests to verify consistent step candidates and stop offering an installed plugin while preserving the original recommendation block in conversation history.

GitOrigin-RevId: 44fa6723a623d9604f0c768118bc6889d2d3b693
This commit is contained in:
Alex Zamoshchin
2026-11-13 22:16:32 +00:00
committed by copyberry
parent 826b6e8199
commit 42f950cd11
7 changed files with 460 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core-plugins/src/manager.rs
+108 -102
View File
@@ -162,6 +162,7 @@ const FEATURED_PLUGIN_IDS_CACHE_TTL: std::time::Duration =
std::time::Duration::from_secs(60 * 60 * 3);
const REMOTE_INSTALLED_PLUGIN_SYNC_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
const LOADED_PLUGINS_CACHE_CAPACITY: usize = 8;
const TOOLING_STATE_REUSE_WINDOW: Duration = Duration::from_secs(1);
type EffectivePluginsChangedCallback = Arc<dyn Fn(EffectivePluginsChange) + Send + Sync + 'static>;
@@ -299,22 +300,31 @@ pub struct RecommendedPluginCandidatesInput<'a> {
pub app_server_client_name: Option<&'a str>,
}
#[derive(Clone, PartialEq, Eq)]
struct FeaturedPluginIdsCacheKey {
#[derive(Clone, Hash, PartialEq, Eq)]
struct PluginServiceCacheKey {
product_sku: String,
chatgpt_base_url: String,
account_id: Option<String>,
chatgpt_user_id: Option<String>,
is_workspace_account: bool,
}
#[derive(Clone, Hash, PartialEq, Eq)]
struct RecommendedPluginsCacheKey {
chatgpt_base_url: String,
#[derive(Clone)]
struct CachedRecommendations {
mode: RecommendedPluginsMode,
tooling_changed_at: Option<String>,
checked_at: Instant,
}
#[derive(Default)]
struct RecommendationCacheEntry {
cached: Option<CachedRecommendations>,
in_flight: Option<Arc<OnceCell<RecommendedPluginsMode>>>,
}
#[derive(Clone)]
struct CachedFeaturedPluginIds {
key: FeaturedPluginIdsCacheKey,
key: PluginServiceCacheKey,
expires_at: Instant,
featured_plugin_ids: Vec<String>,
}
@@ -488,11 +498,16 @@ struct ConfiguredMarketplaceUpgradeState {
in_flight: bool,
}
fn featured_plugin_ids_cache_key(
fn plugin_service_cache_key(
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
) -> FeaturedPluginIdsCacheKey {
FeaturedPluginIdsCacheKey {
) -> PluginServiceCacheKey {
PluginServiceCacheKey {
product_sku: config
.product_sku
.as_deref()
.unwrap_or(CODEX_PRODUCT_SKU)
.to_owned(),
chatgpt_base_url: config.chatgpt_base_url.clone(),
account_id: auth.and_then(CodexAuth::get_account_id),
chatgpt_user_id: auth.and_then(CodexAuth::get_chatgpt_user_id),
@@ -500,12 +515,6 @@ fn featured_plugin_ids_cache_key(
}
}
fn recommended_plugins_cache_key(config: &PluginsConfigInput) -> RecommendedPluginsCacheKey {
RecommendedPluginsCacheKey {
chatgpt_base_url: config.chatgpt_base_url.clone(),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginInstallRequest {
pub plugin_name: String,
@@ -625,9 +634,7 @@ pub struct PluginsManager {
codex_home: PathBuf,
store: PluginStore,
featured_plugin_ids_cache: RwLock<Option<CachedFeaturedPluginIds>>,
recommended_plugins_cache: RwLock<HashMap<RecommendedPluginsCacheKey, RecommendedPluginsMode>>,
recommended_plugins_refreshes:
RwLock<HashMap<RecommendedPluginsCacheKey, Arc<OnceCell<RecommendedPluginsMode>>>>,
recommended_plugins_cache: Mutex<HashMap<PluginServiceCacheKey, RecommendationCacheEntry>>,
configured_marketplace_upgrade_state: RwLock<ConfiguredMarketplaceUpgradeState>,
non_curated_cache_refresh_lock: Semaphore,
non_curated_cache_refresh_state: RwLock<NonCuratedCacheRefreshState>,
@@ -826,8 +833,7 @@ impl PluginsManager {
codex_home,
store,
featured_plugin_ids_cache: RwLock::new(None),
recommended_plugins_cache: RwLock::new(HashMap::new()),
recommended_plugins_refreshes: RwLock::new(HashMap::new()),
recommended_plugins_cache: Mutex::new(HashMap::new()),
configured_marketplace_upgrade_state: RwLock::new(
ConfiguredMarketplaceUpgradeState::default(),
),
@@ -1078,16 +1084,10 @@ impl PluginsManager {
}
pub fn clear_recommended_plugins_cache(&self) {
let mut refreshes = match self.recommended_plugins_refreshes.write() {
Ok(refreshes) => refreshes,
Err(err) => err.into_inner(),
};
refreshes.clear();
let mut cache = match self.recommended_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
cache.clear();
self.recommended_plugins_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clear();
}
fn clear_loaded_plugins_cache(&self) {
@@ -2022,10 +2022,7 @@ impl PluginsManager {
);
}
fn cached_featured_plugin_ids(
&self,
cache_key: &FeaturedPluginIdsCacheKey,
) -> Option<Vec<String>> {
fn cached_featured_plugin_ids(&self, cache_key: &PluginServiceCacheKey) -> Option<Vec<String>> {
{
let cache = match self.featured_plugin_ids_cache.read() {
Ok(cache) => cache,
@@ -2056,7 +2053,7 @@ impl PluginsManager {
fn write_featured_plugin_ids_cache(
&self,
cache_key: FeaturedPluginIdsCacheKey,
cache_key: PluginServiceCacheKey,
featured_plugin_ids: &[String],
) {
let mut cache = match self.featured_plugin_ids_cache.write() {
@@ -2079,7 +2076,7 @@ impl PluginsManager {
return Ok(Vec::new());
}
let cache_key = featured_plugin_ids_cache_key(config, auth);
let cache_key = plugin_service_cache_key(config, auth);
if let Some(featured_plugin_ids) = self.cached_featured_plugin_ids(&cache_key) {
return Ok(featured_plugin_ids);
}
@@ -2113,73 +2110,52 @@ impl PluginsManager {
return RecommendedPluginsMode::Legacy;
}
let cache_key = recommended_plugins_cache_key(config);
if let Some(cached) = self.cached_recommended_plugins_mode(&cache_key) {
return cached;
}
let refresh = {
let mut refreshes = match self.recommended_plugins_refreshes.write() {
Ok(refreshes) => refreshes,
Err(err) => err.into_inner(),
};
if let Some(cached) = self.cached_recommended_plugins_mode(&cache_key) {
return cached;
let cache_key = plugin_service_cache_key(config, auth);
let (cached, refresh) = {
let mut cache = self
.recommended_plugins_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let entry = cache.entry(cache_key.clone()).or_default();
if let Some(cached) = &entry.cached
&& cached.checked_at.elapsed() < TOOLING_STATE_REUSE_WINDOW
{
return cached.mode.clone();
}
refreshes
.entry(cache_key.clone())
.or_insert_with(|| Arc::new(OnceCell::new()))
.clone()
(
entry.cached.clone(),
Arc::clone(entry.in_flight.get_or_insert_default()),
)
};
let mode = refresh
refresh
.get_or_init(|| async {
match crate::remote::fetch_recommended_plugins(
&self.remote_plugin_service_config(config),
auth,
)
.await
let updated = self.refresh_recommendations(config, auth, cached).await;
let mode = updated
.as_ref()
.map_or(RecommendedPluginsMode::Legacy, |cached| cached.mode.clone());
let mut cache = self
.recommended_plugins_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
// A clear may have discarded this request while it was in flight.
if let Some(entry) = cache.get_mut(&cache_key)
&& entry
.in_flight
.as_ref()
.is_some_and(|current| Arc::ptr_eq(current, &refresh))
{
Ok(mode) => {
let refreshes = match self.recommended_plugins_refreshes.read() {
Ok(refreshes) => refreshes,
Err(err) => err.into_inner(),
};
// Hold the refresh lock through publication, matching cache-clear lock
// order. An invalidated initializer must not repopulate the cache.
if refreshes
.get(&cache_key)
.is_some_and(|current| Arc::ptr_eq(current, &refresh))
{
let mut cache = match self.recommended_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
cache.insert(cache_key.clone(), mode.clone());
}
mode
}
Err(err) => {
warn!(error = %err, "failed to load recommended plugins");
RecommendedPluginsMode::Legacy
if let Some(cached) = updated {
entry.cached = Some(cached);
entry.in_flight = None;
} else {
cache.remove(&cache_key);
}
}
mode
})
.await
.clone();
let mut refreshes = match self.recommended_plugins_refreshes.write() {
Ok(refreshes) => refreshes,
Err(err) => err.into_inner(),
};
if refreshes
.get(&cache_key)
.is_some_and(|current| Arc::ptr_eq(current, &refresh))
{
refreshes.remove(&cache_key);
}
mode
.clone()
}
/// Returns endpoint recommendations eligible for installation in the current client.
@@ -2227,15 +2203,45 @@ impl PluginsManager {
))
}
fn cached_recommended_plugins_mode(
async fn refresh_recommendations(
&self,
cache_key: &RecommendedPluginsCacheKey,
) -> Option<RecommendedPluginsMode> {
let cache = match self.recommended_plugins_cache.read() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
cache.get(cache_key).cloned()
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
mut cached: Option<CachedRecommendations>,
) -> Option<CachedRecommendations> {
let remote = self.remote_plugin_service_config(config);
// Observe the marker first so a change during the catalog read is seen on the next poll.
let marker = crate::remote::fetch_tooling_changed_at(&remote, auth)
.await
.unwrap_or_else(|err| {
tracing::debug!(
sub_error_type = err.sub_error_type(),
"failed to read recommendation tooling state"
);
None
});
let checked_at = Instant::now();
if let Some(entry) = &mut cached {
entry.checked_at = checked_at;
if marker.is_none() || marker == entry.tooling_changed_at {
return cached;
}
}
match crate::remote::fetch_recommended_plugins(&remote, auth).await {
Ok(mode) => Some(CachedRecommendations {
mode,
tooling_changed_at: marker,
checked_at,
}),
Err(err) => {
warn!(
sub_error_type = err.sub_error_type(),
"failed to load recommended plugins"
);
// Keep the previous marker so the next poll retries the changed catalog.
cached
}
}
}
pub async fn install_plugin(
codex-rs/core-plugins/src/manager_tests.rs
+122 -111
View File
@@ -6142,6 +6142,9 @@ plugins = true
Mock::given(method("GET"))
.and(path("/ps/plugins/suggested/codex"))
.and(query_param("scope", "GLOBAL"))
.and(header("authorization", "Bearer Access Token"))
.and(header("chatgpt-account-id", "account_id"))
.and(header("OAI-Product-Sku", "codex"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"enabled": true,
"plugins": []
@@ -6156,7 +6159,11 @@ plugins = true
config.chatgpt_base_url = server.uri();
let manager = std::sync::Arc::new(test_plugins_manager(tmp.path().to_path_buf()));
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
let cache_key = recommended_plugins_cache_key(&config);
let cache_key = plugin_service_cache_key(&config, Some(&auth));
let cached_mode = || {
let cache = manager.recommended_plugins_cache.lock().unwrap();
Some(cache.get(&cache_key)?.cached.as_ref()?.mode.clone())
};
manager.maybe_start_remote_plugin_caches_refresh(
&config,
@@ -6166,7 +6173,7 @@ plugins = true
let mode = tokio::time::timeout(Duration::from_secs(2), async {
loop {
if let Some(mode) = manager.cached_recommended_plugins_mode(&cache_key) {
if let Some(mode) = cached_mode() {
break mode;
}
tokio::time::sleep(Duration::from_millis(10)).await;
@@ -6187,78 +6194,113 @@ plugins = true
mode
);
manager.clear_recommended_plugins_cache();
assert_eq!(manager.cached_recommended_plugins_mode(&cache_key), None);
assert!(cached_mode().is_none());
}
async fn mock_recommendation_responses(
server: &MockServer,
route: &str,
responses: impl IntoIterator<Item = ResponseTemplate>,
) {
for response in responses {
Mock::given(method("GET"))
.and(path(route))
.respond_with(response)
.up_to_n_times(1)
.expect(1)
.mount(server)
.await;
}
Mock::given(method("GET"))
.and(path(route))
.respond_with(ResponseTemplate::new(500))
.with_priority(255)
.expect(0)
.mount(server)
.await;
}
#[tokio::test]
async fn recommended_plugins_mode_deduplicates_concurrent_cache_misses() {
let tmp = tempfile::tempdir().unwrap();
write_file(
&tmp.path().join(CONFIG_TOML_FILE),
r#"[features]
plugins = true
"#,
);
async fn recommended_plugins_stay_cached_until_a_changed_marker_refresh_succeeds() {
let tmp = TempDir::new().unwrap();
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/ps/plugins/suggested/codex"))
.and(query_param("scope", "GLOBAL"))
.and(header("authorization", "Bearer Access Token"))
.and(header("chatgpt-account-id", "account_id"))
.and(header("OAI-Product-Sku", "codex"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({
"enabled": true,
"plugins": [
{
"id": "plugin_slack",
"name": "slack",
"display_name": "Slack"
},
{
"id": "plugin_github",
"name": "github",
"display_name": "GitHub"
}
]
}))
.set_delay(Duration::from_millis(100)),
)
.expect(1)
.mount(&server)
.await;
let mut config = load_config(tmp.path(), tmp.path()).await;
config.chatgpt_base_url = server.uri();
let manager = test_plugins_manager(tmp.path().to_path_buf());
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
let expected = RecommendedPluginsMode::Endpoint {
plugins: vec![
RecommendedPlugin {
config_id: "github@openai-curated-remote".to_string(),
remote_plugin_id: "plugin_github".to_string(),
display_name: "GitHub".to_string(),
},
RecommendedPlugin {
config_id: "slack@openai-curated-remote".to_string(),
remote_plugin_id: "plugin_slack".to_string(),
display_name: "Slack".to_string(),
},
],
let recommendations = || manager.recommended_plugins_mode_for_config(&config, Some(&auth));
let marker = |value: Option<&str>| {
ResponseTemplate::new(200).set_body_json(serde_json::json!({"tooling_changed_at": value}))
};
mock_recommendation_responses(
&server,
"/ps/plugins/tooling-state/turtle",
[
marker(Some("100")),
marker(None),
ResponseTemplate::new(503),
marker(Some("100")),
marker(Some("101")),
marker(Some("101")),
marker(Some("101")),
],
)
.await;
mock_recommendation_responses(
&server,
"/ps/plugins/suggested/codex",
[
ResponseTemplate::new(200).set_body_json(serde_json::json!({
"enabled": true,
"plugins": [{"id": "plugin_slack", "name": "slack", "display_name": "Slack"}]
})),
ResponseTemplate::new(503),
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"enabled": false, "plugins": []})),
],
)
.await;
let (left, right) = tokio::join!(
manager.recommended_plugins_mode_for_config(&config, Some(&auth)),
manager.recommended_plugins_mode_for_config(&config, Some(&auth)),
);
assert_eq!((left, right), (expected.clone(), expected.clone()));
let original = RecommendedPluginsMode::Endpoint {
plugins: vec![RecommendedPlugin {
config_id: "slack@openai-curated-remote".to_string(),
remote_plugin_id: "plugin_slack".to_string(),
display_name: "Slack".to_string(),
}],
};
let (left, right) = tokio::join!(recommendations(), recommendations());
assert_eq!((left, right), (original.clone(), original.clone()));
assert_eq!(recommendations().await, original, "fresh cache hit");
let key = plugin_service_cache_key(&config, Some(&auth));
let poll_cutoff = Instant::now() - TOOLING_STATE_REUSE_WINDOW;
let stale_check = poll_cutoff
.checked_sub(Duration::from_secs(24 * 60 * 60))
.unwrap_or(poll_cutoff);
let age_tooling_check = || {
let mut cache = manager.recommended_plugins_cache.lock().unwrap();
let cached = cache
.get_mut(&key)
.and_then(|entry| entry.cached.as_mut())
.expect("recommendations were cached");
cached.checked_at = stale_check;
};
for reason in ["null marker", "failed marker request", "unchanged marker"] {
age_tooling_check();
assert_eq!(recommendations().await, original, "{reason}");
}
// Concurrent callers share the failed catalog refresh. The next poll retries the same marker.
age_tooling_check();
let (left, right) = tokio::join!(recommendations(), recommendations());
assert_eq!((left, right), (original.clone(), original));
age_tooling_check();
assert_eq!(recommendations().await, RecommendedPluginsMode::Legacy);
age_tooling_check();
assert_eq!(
manager
.recommended_plugins_mode_for_config(&config, Some(&auth))
.await,
expected
recommendations().await,
RecommendedPluginsMode::Legacy,
"an unchanged marker reuses the explicitly disabled response"
);
}
@@ -6414,45 +6456,6 @@ plugins = true
);
}
#[tokio::test]
async fn recommended_plugins_mode_caches_explicit_false() {
let tmp = tempfile::tempdir().unwrap();
write_file(
&tmp.path().join(CONFIG_TOML_FILE),
r#"[features]
plugins = true
"#,
);
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/ps/plugins/suggested/codex"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"enabled": false,
"plugins": []
})))
.expect(1)
.mount(&server)
.await;
let mut config = load_config(tmp.path(), tmp.path()).await;
config.chatgpt_base_url = server.uri();
let manager = test_plugins_manager(tmp.path().to_path_buf());
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
assert_eq!(
manager
.recommended_plugins_mode_for_config(&config, Some(&auth))
.await,
RecommendedPluginsMode::Legacy
);
assert_eq!(
manager
.recommended_plugins_mode_for_config(&config, Some(&auth))
.await,
RecommendedPluginsMode::Legacy
);
}
#[tokio::test]
async fn recommended_plugins_mode_retries_after_fetch_failure() {
let tmp = tempfile::tempdir().unwrap();
@@ -6475,13 +6478,23 @@ plugins = true
config.chatgpt_base_url = server.uri();
let manager = test_plugins_manager(tmp.path().to_path_buf());
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
assert_eq!(
manager
.recommended_plugins_mode_for_config(&config, Some(&auth))
.await,
RecommendedPluginsMode::Legacy
);
let lookup = || manager.recommended_plugins_mode_for_config(&config, Some(&auth));
let mut first = Box::pin(lookup());
assert!(futures::poll!(first.as_mut()).is_pending());
let mut waiting = Box::pin(lookup());
assert!(futures::poll!(waiting.as_mut()).is_pending());
drop(first);
// A waiting caller takes over a cancelled fetch. Both callers share the cold failure.
let (left, right) = tokio::join!(waiting, lookup());
assert_eq!(
(left, right),
(
RecommendedPluginsMode::Legacy,
RecommendedPluginsMode::Legacy
)
);
server.verify().await;
server.reset().await;
Mock::given(method("GET"))
.and(path("/ps/plugins/suggested/codex"))
@@ -6494,9 +6507,7 @@ plugins = true
.await;
assert_eq!(
manager
.recommended_plugins_mode_for_config(&config, Some(&auth))
.await,
lookup().await,
RecommendedPluginsMode::Endpoint {
plugins: Vec::new()
}
codex-rs/core-plugins/src/remote.rs
+27
View File
@@ -117,6 +117,7 @@ const REMOTE_INSTALLED_PLUGINS_FETCH_TIMEOUT: Duration = Duration::from_secs(60)
// Allow 20,000 installed plugins without unbounded cursor history.
const MAX_REMOTE_INSTALLED_PLUGIN_PAGES: usize = 100;
const RECOMMENDED_PLUGINS_TIMEOUT: Duration = Duration::from_secs(5);
const TOOLING_STATE_TIMEOUT: Duration = Duration::from_millis(250);
const REMOTE_PLUGIN_LIST_PAGE_LIMIT: u32 = 200;
const MAX_RECOMMENDED_PLUGINS: usize = 50;
const MAX_RECOMMENDED_PLUGIN_NAME_LEN: usize = 64;
@@ -897,6 +898,11 @@ struct RemotePluginListResponse {
pagination: RemotePluginPagination,
}
#[derive(Deserialize)]
struct PluginToolingStateResponse {
tooling_changed_at: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
struct RecommendedPluginsResponse {
enabled: bool,
@@ -1147,6 +1153,27 @@ pub fn invalidate_cached_remote_plugin_catalog_scopes(
}
}
#[instrument(level = "trace", skip_all)]
pub(crate) async fn fetch_tooling_changed_at(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<Option<String>, RemotePluginCatalogError> {
let auth = ensure_chatgpt_auth(auth)?;
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/tooling-state/turtle");
let poll = async {
let request = authenticated_request(config, Method::GET, &url, auth).await?;
let response: PluginToolingStateResponse = send_and_decode(request, &url).await?;
Ok(response.tooling_changed_at)
};
tokio::time::timeout(TOOLING_STATE_TIMEOUT, poll)
.await
.unwrap_or_else(|_| {
tracing::debug!("timed out reading recommendation tooling state");
Ok(None)
})
}
#[instrument(level = "trace", skip_all)]
pub async fn fetch_recommended_plugins(
config: &RemotePluginServiceConfig,
codex-rs/core-plugins/src/remote_runtime_auth_tests.rs
+44
View File
@@ -228,6 +228,50 @@ async fn plugin_runtime_auth_shares_registration_and_strips_cross_origin_auth()
}
}
#[tokio::test]
async fn tooling_state_deadline_includes_pending_runtime_registration() {
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
let runtime = Arc::new(RuntimeAuthManager::new(
AuthManager::from_auth_for_testing(auth.clone()),
AgentHostAuthorization::InstallationId {
installation_id: Uuid::nil(),
},
RuntimeRegistrationMetadata {
harness: Some("codex".into()),
harness_version: None,
role: RuntimeRole::Orchestrator,
},
format!("{base}/register").parse().unwrap(),
));
let config = RemotePluginServiceConfig::new_with_selected_runtime_auth(
format!("{base}/backend-api"),
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
/*product_sku*/ None,
(
Arc::new(std::sync::OnceLock::from(Arc::clone(&runtime))),
RuntimeAuthMode::DualHeader,
),
);
// Hold registration open before starting the marker clock.
let registration = runtime.auth();
tokio::pin!(registration);
let (_connection, _) = tokio::select! {
connection = listener.accept() => connection.unwrap(),
_ = &mut registration => panic!("registration finished before receiving a response"),
};
tokio::time::pause();
let poll = fetch_tooling_changed_at(&config, Some(&auth));
let marker = tokio::time::timeout(TOOLING_STATE_TIMEOUT * 2, poll)
.await
.expect("the marker deadline must include runtime registration")
.unwrap();
assert_eq!(marker, None);
}
#[tokio::test]
async fn custom_plugin_backend_keeps_legacy_auth_with_runtime_enabled() {
for mode in [RuntimeAuthMode::DualHeader, RuntimeAuthMode::AssertionOnly] {
codex-rs/core/src/session/mod.rs
+27 -20
View File
@@ -309,6 +309,7 @@ use self::session::SessionSettingsCommit;
pub(crate) use self::session::SessionSettingsUpdate;
#[cfg(test)]
use self::turn::AssistantMessageStreamParsers;
use self::turn::PreparedToolRecommendations;
use self::turn::RealtimeEventText;
use self::turn::agent_message_text;
#[cfg(test)]
@@ -347,7 +348,9 @@ use crate::tools::network_approval::build_blocked_request_observer;
use crate::tools::network_approval::build_network_policy_decider;
#[cfg(test)]
use crate::tools::parallel::ToolCallRuntime;
use crate::tools::router::ToolSuggestCandidates;
use crate::tools::sandboxing::ApprovalStore;
use crate::tools::spec_plan::tool_suggest_enabled;
use crate::turn_timing::TurnTimingState;
use crate::turn_timing::record_turn_ttfm_metric;
use crate::unified_exec::UnifiedExecProcessManager;
@@ -356,7 +359,6 @@ use crate::windows_sandbox::local_binding_policy_for_sandbox;
use crate::windows_sandbox::managed_proxy_routing_for_windows_sandbox;
use codex_core_plugins::PluginCommandAttribution;
use codex_core_plugins::PluginsManager;
use codex_core_plugins::RecommendedPluginCandidatesInput;
use codex_git_utils::get_git_repo_root;
use codex_history::CodexHarnessMetadata;
use codex_history::CompactedItem;
@@ -4488,7 +4490,13 @@ impl Session {
&required_servers,
required_plugins,
)),
turn::prepare_tool_recommendations(self.as_ref(), turn_context.as_ref()),
async {
if tool_suggest_enabled(&turn_context) {
Some(PreparedToolRecommendations::load(self, &turn_context).await)
} else {
None
}
},
);
let search = crate::model_search::resolve(
turn_context.as_ref(),
@@ -5134,25 +5142,25 @@ impl Session {
developer_sections
.push(DeveloperInstructions::new(developer_instructions).render_fragment());
}
let recommended_plugin_candidates = if turn_context
.config
.features
.plugin_recommendations_enabled()
let step_candidates = step_context.extension_data.get::<ToolSuggestCandidates>();
// Prompt-only recommendations have no install handler to capture candidates for the step.
let prompt_only_candidates = if !tool_suggest_enabled(turn_context)
&& turn_context
.config
.features
.plugin_recommendations_enabled()
{
let auth = self.services.auth_manager.auth().await;
let plugins_config = turn_context.config.plugins_config_input();
self.services
.plugins_manager
.recommended_plugin_candidates_for_config(RecommendedPluginCandidatesInput {
plugins_config: &plugins_config,
auth: auth.as_ref(),
disabled_tools: &turn_context.config.tool_suggest.disabled_tools,
app_server_client_name: turn_context.app_server_client_name.as_deref(),
})
.await
match PreparedToolRecommendations::load(self, turn_context).await {
PreparedToolRecommendations::Endpoint(tools) => Some(tools),
PreparedToolRecommendations::Legacy(_) => None,
}
} else {
None
};
let recommended_plugin_candidates = step_candidates
.as_deref()
.map(|candidates| candidates.tools.as_slice())
.or(prompt_only_candidates.as_deref());
let context_contributors = self.services.extensions.context_contributors().to_vec();
for contributor in &context_contributors {
for fragment in contributor
@@ -5301,9 +5309,8 @@ impl Session {
}
if next_item.is_none()
&& let Some(recommended_plugins) = recommended_plugin_candidates
.as_deref()
.and_then(|plugins| {
&& let Some(recommended_plugins) =
recommended_plugin_candidates.and_then(|plugins| {
RecommendedPluginsInstructions::from_plugins(
plugins,
&ResolvedModelMessages::from_model(&step_context.settings.model_info)
codex-rs/core/src/session/turn.rs
+61 -83
View File
@@ -67,7 +67,6 @@ use crate::tools::router::ToolSuggestCandidates;
use crate::tools::router::ToolSuggestPresentation;
use crate::tools::router::record_registration_metrics;
use crate::tools::spec_plan::build_tool_router;
use crate::tools::spec_plan::tool_suggest_enabled;
use crate::turn_diff_tracker::TurnDiffTracker;
use crate::turn_timing::record_turn_ttft_metric;
use crate::util::error_or_panic;
@@ -2095,40 +2094,29 @@ async fn run_sampling_request(
}
}
pub(crate) struct PreparedToolRecommendations {
auth: Option<CodexAuth>,
endpoint_candidates: Option<Vec<DiscoverableTool>>,
pub(super) enum PreparedToolRecommendations {
Endpoint(Vec<DiscoverableTool>),
Legacy(Option<CodexAuth>),
}
#[instrument(level = "trace", skip_all)]
pub(crate) async fn prepare_tool_recommendations(
sess: &Session,
turn_context: &TurnContext,
) -> PreparedToolRecommendations {
let tool_suggest_is_enabled = tool_suggest_enabled(turn_context);
let auth = if tool_suggest_is_enabled {
sess.services.auth_manager.auth().await
} else {
None
};
let endpoint_candidates = if tool_suggest_is_enabled {
let plugins_config = turn_context.config.plugins_config_input();
sess.services
impl PreparedToolRecommendations {
#[instrument(level = "trace", skip_all)]
pub(super) async fn load(sess: &Session, turn_context: &TurnContext) -> Self {
let auth = sess.services.auth_manager.auth().await;
let endpoint_candidates = sess
.services
.plugins_manager
.recommended_plugin_candidates_for_config(RecommendedPluginCandidatesInput {
plugins_config: &plugins_config,
plugins_config: &turn_context.config.plugins_config_input(),
auth: auth.as_ref(),
disabled_tools: &turn_context.config.tool_suggest.disabled_tools,
app_server_client_name: turn_context.app_server_client_name.as_deref(),
})
.await
} else {
None
};
PreparedToolRecommendations {
auth,
endpoint_candidates,
.await;
match endpoint_candidates {
Some(tools) => Self::Endpoint(tools),
None => Self::Legacy(auth),
}
}
}
@@ -2141,77 +2129,61 @@ pub(crate) async fn prepare_tool_recommendations(
apps_enabled = turn_context.apps_enabled()
)
)]
pub(crate) async fn built_tools(
pub(super) async fn built_tools(
sess: &Session,
turn_context: &TurnContext,
model_info: &codex_protocol::openai_models::ModelInfo,
environments: &TurnEnvironmentSnapshot,
mcp: &Arc<codex_mcp::McpBinding>,
step_store: &ExtensionData,
prepared_recommendations: PreparedToolRecommendations,
prepared_recommendations: Option<PreparedToolRecommendations>,
dynamic_tools: &[codex_protocol::dynamic_tools::DynamicToolSpec],
) -> CodexResult<Arc<ToolRouter>> {
let all_mcp_tools = mcp.tools();
let connector_snapshot = mcp.config().connector_snapshot.clone();
let apps_enabled = turn_context.apps_enabled();
let accessible_connectors =
apps_enabled.then(|| connectors::accessible_connectors_from_mcp_tools(all_mcp_tools));
let tool_suggest_is_enabled = tool_suggest_enabled(turn_context);
let PreparedToolRecommendations {
auth,
endpoint_candidates: endpoint_recommended_plugin_candidates,
} = prepared_recommendations;
let tool_suggest_candidates =
if let Some(recommended_plugin_candidates) = endpoint_recommended_plugin_candidates {
Some(ToolSuggestCandidates {
tools: recommended_plugin_candidates,
presentation: ToolSuggestPresentation::RecommendationContext,
})
} else {
let loaded_plugin_app_connector_ids = connector_snapshot
let tool_suggest_candidates = match prepared_recommendations {
Some(PreparedToolRecommendations::Endpoint(tools)) => Some(ToolSuggestCandidates {
tools,
presentation: ToolSuggestPresentation::RecommendationContext,
}),
Some(PreparedToolRecommendations::Legacy(auth)) if apps_enabled => {
let accessible_connectors =
connectors::accessible_connectors_from_mcp_tools(mcp.tools());
let loaded_plugin_app_connector_ids = mcp
.config()
.connector_snapshot
.connector_ids()
.iter()
.map(|connector_id| connector_id.0.clone())
.collect::<Vec<_>>();
async {
if apps_enabled && tool_suggest_is_enabled {
if let Some(accessible_connectors) = accessible_connectors.as_ref() {
match connectors::list_tool_suggest_discoverable_tools_with_auth(
&turn_context.config,
sess.services.plugins_manager.as_ref(),
auth.as_ref(),
accessible_connectors.as_slice(),
&loaded_plugin_app_connector_ids,
)
.await
.map(|discoverable_tools| {
filter_request_plugin_install_discoverable_tools_for_client(
discoverable_tools,
turn_context.app_server_client_name.as_deref(),
)
}) {
Ok(discoverable_tools) if discoverable_tools.is_empty() => None,
Ok(discoverable_tools) => Some(ToolSuggestCandidates {
tools: discoverable_tools,
presentation: ToolSuggestPresentation::ListTool,
}),
Err(err) => {
warn!("failed to load discoverable tool suggestions: {err:#}");
None
}
}
} else {
None
}
} else {
match connectors::list_tool_suggest_discoverable_tools_with_auth(
&turn_context.config,
sess.services.plugins_manager.as_ref(),
auth.as_ref(),
&accessible_connectors,
&loaded_plugin_app_connector_ids,
)
.instrument(trace_span!("built_tools.load_discoverable_tools"))
.await
{
Ok(tools) => {
let tools = filter_request_plugin_install_discoverable_tools_for_client(
tools,
turn_context.app_server_client_name.as_deref(),
);
(!tools.is_empty()).then_some(ToolSuggestCandidates {
tools,
presentation: ToolSuggestPresentation::ListTool,
})
}
Err(err) => {
warn!("failed to load discoverable tool suggestions: {err:#}");
None
}
}
.instrument(trace_span!("built_tools.load_discoverable_tools"))
.await
};
Ok(Arc::new(build_tool_router(
}
Some(PreparedToolRecommendations::Legacy(_)) | None => None,
};
let router = build_tool_router(
sess,
turn_context,
model_info,
@@ -2221,7 +2193,13 @@ pub(crate) async fn built_tools(
step_store,
tool_suggest_candidates.as_ref(),
dynamic_tools,
)?))
)?;
if let Some(candidates) = tool_suggest_candidates
&& candidates.presentation == ToolSuggestPresentation::RecommendationContext
{
step_store.insert(candidates);
}
Ok(Arc::new(router))
}
#[derive(Debug)]
codex-rs/core/tests/suite/request_plugin_install.rs
+71 -11
View File
@@ -382,19 +382,38 @@ async fn resolve_install_elicitation(
Ok(())
}
async fn mount_remote_calendar_recommendation(server: &wiremock::MockServer) {
async fn mount_remote_calendar_recommendation(
server: &wiremock::MockServer,
installed: Arc<AtomicBool>,
) {
let installed_for_marker = Arc::clone(&installed);
Mock::given(method("GET"))
.and(path("/ps/plugins/tooling-state/turtle"))
.respond_with(move |_: &wiremock::Request| {
let marker = installed_for_marker
.load(Ordering::SeqCst)
.then_some("1700000000000");
ResponseTemplate::new(200).set_body_json(json!({ "tooling_changed_at": marker }))
})
.mount(server)
.await;
Mock::given(method("GET"))
.and(path("/ps/plugins/suggested/codex"))
.and(query_param("scope", "GLOBAL"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"enabled": true,
"plugins": [{
"id": REMOTE_CALENDAR_PLUGIN_ID,
"name": "calendar",
"display_name": "Calendar"
}]
})))
.expect(1)
.and(query_param("includeInstalled", "false"))
.respond_with(move |_: &wiremock::Request| {
let plugins = if installed.load(Ordering::SeqCst) {
json!([])
} else {
json!([{
"id": REMOTE_CALENDAR_PLUGIN_ID,
"name": "calendar",
"display_name": "Calendar"
}])
};
ResponseTemplate::new(200).set_body_json(json!({ "enabled": true, "plugins": plugins }))
})
.expect(2)
.mount(server)
.await;
Mock::given(method("GET"))
@@ -788,6 +807,18 @@ async fn mcp_discovery_overlaps_endpoint_plugin_recommendations() -> Result<()>
response.requests().is_empty(),
"sampling should wait for the complete MCP catalog"
);
// Change the shared catalog after the step's recommendation request, while MCP is blocked.
Mock::given(method("GET"))
.and(path("/ps/plugins/suggested/codex"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"enabled": true, "plugins": []
})))
.with_priority(1)
.mount(&server)
.await;
test.thread_manager
.plugins_manager()
.clear_recommended_plugins_cache();
test.fs()
.write_file(
&barrier,
@@ -809,6 +840,11 @@ async fn mcp_discovery_overlaps_endpoint_plugin_recommendations() -> Result<()>
.contains("github@openai-curated-remote"),
"the completed request should preserve endpoint recommendations"
);
assert!(
tool_names(&request.body_json())
.iter()
.any(|name| name == REQUEST_PLUGIN_INSTALL_TOOL_NAME)
);
assert!(
request
.tool_by_name("mcp__step_preparation", "echo")
@@ -1541,7 +1577,8 @@ async fn run_remote_plugin_install_refresh_case(refreshed_tools: RefreshedAppsTo
}
RefreshedAppsTools::Missing => AppsTestServer::mount_without_tools(&server).await?,
};
mount_remote_calendar_recommendation(&server).await;
let installed = Arc::new(AtomicBool::new(false));
mount_remote_calendar_recommendation(&server, Arc::clone(&installed)).await;
let initial_remote_installed_plugins = mount_empty_remote_installed_plugins(&server).await;
let install_call_id = "install-calendar";
@@ -1579,6 +1616,7 @@ async fn run_remote_plugin_install_refresh_case(refreshed_tools: RefreshedAppsTo
let elicitation = start_install_turn(&test, "use Calendar").await?;
mount_remote_calendar_installed_plugins(&server).await;
drop(initial_remote_installed_plugins);
installed.store(true, Ordering::SeqCst);
tools_available.store(
matches!(refreshed_tools, RefreshedAppsTools::Available),
Ordering::SeqCst,
@@ -1623,10 +1661,32 @@ async fn run_remote_plugin_install_refresh_case(refreshed_tools: RefreshedAppsTo
.codex
.refresh_runtime_config(current_config, test.config.clone())
.await;
// Let the one-second marker poll window expire before the next turn.
tokio::time::sleep(Duration::from_secs(1)).await;
test.submit_turn("check whether Calendar is still installed")
.await?;
let requests = mock.requests();
assert_eq!(requests.len(), 3);
let recommendation_blocks = |index: usize| {
requests[index]
.message_input_texts("developer")
.into_iter()
.filter(|text| text.starts_with("<recommended_plugins>"))
.collect::<Vec<_>>()
};
let initial = recommendation_blocks(0);
assert_eq!(initial.len(), 1);
assert_eq!(
recommendation_blocks(2),
initial,
"refreshing recommendations must preserve the original block"
);
assert!(
!tool_names(&requests[2].body_json())
.iter()
.any(|name| name == REQUEST_PLUGIN_INSTALL_TOOL_NAME),
"the updated recommendations must stop offering to install Calendar"
);
assert_eq!(
requests[2]
.tool_by_name(CALENDAR_NAMESPACE, CALENDAR_CREATE_EVENT_TOOL)