mirror of
https://github.com/openai/codex.git
synced 2026-11-14 21:40:39 +00:00
Anthony Tafoya
a309f9206a
## What changed - Build executor skill catalogs from discovery V2 inventories without reopening host paths or falling back to legacy discovery. - Merge Cloud and executor skills by remote plugin identity and skill name, retaining Cloud metadata and accepted provider read routes. Apply plugin selection and disablement before merging. - Share a stable catalog across each step's rendering, list/read tools, and explicit and implicit skill invocations. - Preserve unmatched executor entries when Cloud catalogs are unavailable or partial; only complete Cloud catalogs disable unmatched remote entries. - Bind discovery caches to captured environments and executor sessions, and refresh MCP when V2 inventories change. Preserve pending invalidations when forced MCP refreshes fail. ## Testing Add regression coverage for V2 inventory sharing and refresh, environment isolation, catalog merging, partial Cloud results, disabled plugins, and catalog consistency across steps and read routes. GitOrigin-RevId: 842bdcbc9e19364dd69a9abf9e48330b7425a76c
61 lines
1.5 KiB
Rust
61 lines
1.5 KiB
Rust
use std::sync::atomic::AtomicBool;
|
|
use std::sync::atomic::Ordering;
|
|
use tokio::sync::AcquireError;
|
|
use tokio::sync::Semaphore;
|
|
use tokio::sync::SemaphorePermit;
|
|
|
|
/// Owns MCP invalidation and the single gate used to publish runtime updates.
|
|
pub(super) struct McpRefresh {
|
|
pending: AtomicBool,
|
|
gate: Semaphore,
|
|
}
|
|
|
|
impl McpRefresh {
|
|
pub(super) fn new() -> Self {
|
|
Self {
|
|
pending: AtomicBool::new(false),
|
|
gate: Semaphore::new(/*permits*/ 1),
|
|
}
|
|
}
|
|
|
|
pub(super) fn invalidate(&self) {
|
|
self.pending.store(true, Ordering::Release);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(super) fn is_pending(&self) -> bool {
|
|
self.pending.load(Ordering::Acquire)
|
|
}
|
|
|
|
pub(super) fn claim(&self) -> bool {
|
|
self.pending.swap(false, Ordering::AcqRel)
|
|
}
|
|
|
|
#[tracing::instrument(name = "mcp.runtime.refresh_wait", skip_all)]
|
|
pub(super) async fn acquire(&self) -> Result<SemaphorePermit<'_>, AcquireError> {
|
|
self.gate.acquire().await
|
|
}
|
|
|
|
pub(super) fn close(&self) {
|
|
self.gate.close();
|
|
}
|
|
}
|
|
|
|
/// Restores a claimed refresh when its task is cancelled before publication.
|
|
pub(super) struct McpRefreshInvalidationGuard<'a> {
|
|
pub(super) refresh: &'a McpRefresh,
|
|
pub(super) published: bool,
|
|
}
|
|
|
|
impl Drop for McpRefreshInvalidationGuard<'_> {
|
|
fn drop(&mut self) {
|
|
if !self.published {
|
|
self.refresh.invalidate();
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "mcp_refresh_tests.rs"]
|
|
mod tests;
|