Merge 29dd204720 into sapling-pr-archive-bolinfest

codex-rs/Cargo.lock

@@ -1409,6 +1409,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "assert_matches",
+ "async-channel",
  "async-trait",
  "base64 0.22.1",
  "bytes",
@@ -1842,6 +1843,7 @@ dependencies = [
 name = "codex-code-mode"
 version = "0.0.0"
 dependencies = [
+ "async-channel",
  "async-trait",
  "codex-protocol",
  "deno_core_icudata",

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json

@@ -10695,6 +10695,44 @@
             ],
             "title": "LocalPluginSource",
             "type": "object"
+          },
+          {
+            "properties": {
+              "path": {
+                "type": [
+                  "string",
+                  "null"
+                ]
+              },
+              "refName": {
+                "type": [
+                  "string",
+                  "null"
+                ]
+              },
+              "sha": {
+                "type": [
+                  "string",
+                  "null"
+                ]
+              },
+              "type": {
+                "enum": [
+                  "git"
+                ],
+                "title": "GitPluginSourceType",
+                "type": "string"
+              },
+              "url": {
+                "type": "string"
+              }
+            },
+            "required": [
+              "type",
+              "url"
+            ],
+            "title": "GitPluginSource",
+            "type": "object"
           }
         ]
       },

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json

@@ -7227,6 +7227,44 @@
           ],
           "title": "LocalPluginSource",
           "type": "object"
+        },
+        {
+          "properties": {
+            "path": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "refName": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "sha": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "type": {
+              "enum": [
+                "git"
+              ],
+              "title": "GitPluginSourceType",
+              "type": "string"
+            },
+            "url": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "type",
+            "url"
+          ],
+          "title": "GitPluginSource",
+          "type": "object"
         }
       ]
     },

codex-rs/app-server-protocol/schema/json/v2/PluginListResponse.json

@@ -204,6 +204,44 @@
           ],
           "title": "LocalPluginSource",
           "type": "object"
+        },
+        {
+          "properties": {
+            "path": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "refName": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "sha": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "type": {
+              "enum": [
+                "git"
+              ],
+              "title": "GitPluginSourceType",
+              "type": "string"
+            },
+            "url": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "type",
+            "url"
+          ],
+          "title": "GitPluginSource",
+          "type": "object"
         }
       ]
     },

codex-rs/app-server-protocol/schema/json/v2/PluginReadResponse.json

@@ -224,6 +224,44 @@
           ],
           "title": "LocalPluginSource",
           "type": "object"
+        },
+        {
+          "properties": {
+            "path": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "refName": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "sha": {
+              "type": [
+                "string",
+                "null"
+              ]
+            },
+            "type": {
+              "enum": [
+                "git"
+              ],
+              "title": "GitPluginSourceType",
+              "type": "string"
+            },
+            "url": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "type",
+            "url"
+          ],
+          "title": "GitPluginSource",
+          "type": "object"
         }
       ]
     },

codex-rs/app-server-protocol/schema/typescript/v2/PluginSource.ts

@@ -3,4 +3,4 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { AbsolutePathBuf } from "../AbsolutePathBuf";
 
-export type PluginSource = { "type": "local", path: AbsolutePathBuf, };
+export type PluginSource = { "type": "local", path: AbsolutePathBuf, } | { "type": "git", url: string, path: string | null, refName: string | null, sha: string | null, };

codex-rs/app-server-protocol/src/protocol/v2.rs

@@ -3802,6 +3802,14 @@ pub enum PluginSource {
     #[serde(rename_all = "camelCase")]
     #[ts(rename_all = "camelCase")]
     Local { path: AbsolutePathBuf },
+    #[serde(rename_all = "camelCase")]
+    #[ts(rename_all = "camelCase")]
+    Git {
+        url: String,
+        path: Option<String>,
+        ref_name: Option<String>,
+        sha: Option<String>,
+    },
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]

codex-rs/app-server/src/codex_message_processor.rs

@@ -8944,6 +8944,17 @@ fn plugin_interface_to_info(interface: PluginManifestInterface) -> PluginInterfa
 fn marketplace_plugin_source_to_info(source: MarketplacePluginSource) -> PluginSource {
     match source {
         MarketplacePluginSource::Local { path } => PluginSource::Local { path },
+        MarketplacePluginSource::Git {
+            url,
+            path,
+            ref_name,
+            sha,
+        } => PluginSource::Git {
+            url,
+            path,
+            ref_name,
+            sha,
+        },
     }
 }
 

codex-rs/code-mode/Cargo.toml

@@ -13,6 +13,7 @@ path = "src/lib.rs"
 workspace = true
 
 [dependencies]
+async-channel = { workspace = true }
 async-trait = { workspace = true }
 codex-protocol = { workspace = true }
 deno_core_icudata = { workspace = true }

codex-rs/code-mode/src/service.rs

@@ -44,8 +44,8 @@ struct SessionHandle {
 struct Inner {
     stored_values: Mutex<HashMap<String, JsonValue>>,
     sessions: Mutex<HashMap<String, SessionHandle>>,
-    turn_message_tx: mpsc::UnboundedSender<TurnMessage>,
-    turn_message_rx: Arc<Mutex<mpsc::UnboundedReceiver<TurnMessage>>>,
+    turn_message_tx: async_channel::Sender<TurnMessage>,
+    turn_message_rx: async_channel::Receiver<TurnMessage>,
     next_cell_id: AtomicU64,
 }
 
@@ -55,14 +55,14 @@ pub struct CodeModeService {
 
 impl CodeModeService {
     pub fn new() -> Self {
-        let (turn_message_tx, turn_message_rx) = mpsc::unbounded_channel();
+        let (turn_message_tx, turn_message_rx) = async_channel::unbounded();
 
         Self {
             inner: Arc::new(Inner {
                 stored_values: Mutex::new(HashMap::new()),
                 sessions: Mutex::new(HashMap::new()),
                 turn_message_tx,
-                turn_message_rx: Arc::new(Mutex::new(turn_message_rx)),
+                turn_message_rx,
                 next_cell_id: AtomicU64::new(1),
             }),
         }
@@ -146,16 +146,13 @@ impl CodeModeService {
     pub fn start_turn_worker(&self, host: Arc<dyn CodeModeTurnHost>) -> CodeModeTurnWorker {
         let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
         let inner = Arc::clone(&self.inner);
-        let turn_message_rx = Arc::clone(&self.inner.turn_message_rx);
+        let turn_message_rx = self.inner.turn_message_rx.clone();
 
         tokio::spawn(async move {
             loop {
                 let next_message = tokio::select! {
                     _ = &mut shutdown_rx => break,
-                    message = async {
-                        let mut turn_message_rx = turn_message_rx.lock().await;
-                        turn_message_rx.recv().await
-                    } => message,
+                    message = turn_message_rx.recv() => message.ok(),
                 };
                 let Some(next_message) = next_message else {
                     break;
@@ -361,7 +358,7 @@ async fn run_session_control(
                             cell_id: cell_id.clone(),
                             call_id,
                             text,
-                        });
+                        }).await;
                     }
                     RuntimeEvent::ToolCall { id, name, input } => {
                         let _ = inner.turn_message_tx.send(TurnMessage::ToolCall {
@@ -369,7 +366,7 @@ async fn run_session_control(
                             id,
                             name,
                             input,
-                        });
+                        }).await;
                     }
                     RuntimeEvent::Result {
                         stored_values,
@@ -500,12 +497,12 @@ mod tests {
     }
 
     fn test_inner() -> Arc<Inner> {
-        let (turn_message_tx, turn_message_rx) = mpsc::unbounded_channel();
+        let (turn_message_tx, turn_message_rx) = async_channel::unbounded();
         Arc::new(Inner {
             stored_values: Mutex::new(HashMap::new()),
             sessions: Mutex::new(HashMap::new()),
             turn_message_tx,
-            turn_message_rx: Arc::new(Mutex::new(turn_message_rx)),
+            turn_message_rx,
             next_cell_id: AtomicU64::new(1),
         })
     }

codex-rs/codex-api/Cargo.toml

@@ -5,6 +5,7 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+async-channel = { workspace = true }
 async-trait = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }

codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs

@@ -65,9 +65,9 @@ enum WsCommand {
 impl WsStream {
     fn new(
         inner: WebSocketStream<MaybeTlsStream<TcpStream>>,
-    ) -> (Self, mpsc::UnboundedReceiver<Result<Message, WsError>>) {
+    ) -> (Self, async_channel::Receiver<Result<Message, WsError>>) {
         let (tx_command, mut rx_command) = mpsc::channel::<WsCommand>(32);
-        let (tx_message, rx_message) = mpsc::unbounded_channel::<Result<Message, WsError>>();
+        let (tx_message, rx_message) = async_channel::unbounded::<Result<Message, WsError>>();
 
         let pump_task = tokio::spawn(async move {
             let mut inner = inner;
@@ -110,7 +110,7 @@ impl WsStream {
                                 trace!(payload_len = payload.len(), "realtime websocket received ping");
                                 if let Err(err) = inner.send(Message::Pong(payload)).await {
                                     error!("realtime websocket failed to send pong: {err}");
-                                    let _ = tx_message.send(Err(err));
+                                    let _ = tx_message.send(Err(err)).await;
                                     break;
                                 }
                             }
@@ -138,7 +138,7 @@ impl WsStream {
                                     }
                                     Message::Ping(_) | Message::Pong(_) => {}
                                 }
-                                if tx_message.send(Ok(message)).is_err() {
+                                if tx_message.send(Ok(message)).await.is_err() {
                                     break;
                                 }
                                 if is_close {
@@ -147,7 +147,7 @@ impl WsStream {
                             }
                             Err(err) => {
                                 error!("realtime websocket receive failed: {err}");
-                                let _ = tx_message.send(Err(err));
+                                let _ = tx_message.send(Err(err)).await;
                                 break;
                             }
                         }
@@ -208,7 +208,7 @@ pub struct RealtimeWebsocketWriter {
 
 #[derive(Clone)]
 pub struct RealtimeWebsocketEvents {
-    rx_message: Arc<Mutex<mpsc::UnboundedReceiver<Result<Message, WsError>>>>,
+    rx_message: async_channel::Receiver<Result<Message, WsError>>,
     active_transcript: Arc<Mutex<ActiveTranscriptState>>,
     event_parser: RealtimeEventParser,
     is_closed: Arc<AtomicBool>,
@@ -256,7 +256,7 @@ impl RealtimeWebsocketConnection {
 
     fn new(
         stream: WsStream,
-        rx_message: mpsc::UnboundedReceiver<Result<Message, WsError>>,
+        rx_message: async_channel::Receiver<Result<Message, WsError>>,
         event_parser: RealtimeEventParser,
     ) -> Self {
         let stream = Arc::new(stream);
@@ -268,7 +268,7 @@ impl RealtimeWebsocketConnection {
                 event_parser,
             },
             events: RealtimeWebsocketEvents {
-                rx_message: Arc::new(Mutex::new(rx_message)),
+                rx_message,
                 active_transcript: Arc::new(Mutex::new(ActiveTranscriptState::default())),
                 event_parser,
                 is_closed,
@@ -369,16 +369,16 @@ impl RealtimeWebsocketEvents {
         }
 
         loop {
-            let msg = match self.rx_message.lock().await.recv().await {
-                Some(Ok(msg)) => msg,
-                Some(Err(err)) => {
+            let msg = match self.rx_message.recv().await {
+                Ok(Ok(msg)) => msg,
+                Ok(Err(err)) => {
                     self.is_closed.store(true, Ordering::SeqCst);
                     error!("realtime websocket read failed: {err}");
                     return Err(ApiError::Stream(format!(
                         "failed to read websocket message: {err}"
                     )));
                 }
-                None => {
+                Err(_) => {
                     self.is_closed.store(true, Ordering::SeqCst);
                     info!("realtime websocket event stream ended");
                     return Ok(None);

codex-rs/core-plugins/src/loader.rs

@@ -32,7 +32,9 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fs;
 use std::path::Path;
+use std::process::Command;
 use std::sync::Arc;
+use tempfile::TempDir;
 use tracing::warn;
 
 const DEFAULT_SKILLS_DIR_NAME: &str = "skills";
@@ -150,6 +152,14 @@ pub fn refresh_curated_plugin_cache(
         }
         let source_path = match plugin.source {
             MarketplacePluginSource::Local { path } => path,
+            MarketplacePluginSource::Git { .. } => {
+                warn!(
+                    plugin = plugin_name,
+                    marketplace = OPENAI_CURATED_MARKETPLACE_NAME,
+                    "skipping remote curated plugin source during cache refresh"
+                );
+                continue;
+            }
         };
         plugin_sources.insert(plugin_name, source_path);
     }
@@ -227,7 +237,7 @@ fn refresh_non_curated_plugin_cache_with_mode(
     let store = PluginStore::new(codex_home.to_path_buf());
     let marketplace_outcome = list_marketplaces(additional_roots)
         .map_err(|err| format!("failed to discover marketplaces for cache refresh: {err}"))?;
-    let mut plugin_sources = HashMap::<String, (AbsolutePathBuf, String)>::new();
+    let mut plugin_sources = HashMap::<String, MarketplacePluginSource>::new();
 
     for marketplace in marketplace_outcome.marketplaces {
         if marketplace.name == OPENAI_CURATED_MARKETPLACE_NAME {
@@ -256,19 +266,14 @@ fn refresh_non_curated_plugin_cache_with_mode(
                 continue;
             }
 
-            let source_path = match plugin.source {
-                MarketplacePluginSource::Local { path } => path,
-            };
-            let plugin_version = plugin_version_for_source(source_path.as_path())
-                .map_err(|err| format!("failed to read plugin version for {plugin_key}: {err}"))?;
-            plugin_sources.insert(plugin_key, (source_path, plugin_version));
+            plugin_sources.insert(plugin_key, plugin.source);
         }
     }
 
     let mut cache_refreshed = false;
     for plugin_id in configured_non_curated_plugin_ids {
         let plugin_key = plugin_id.as_key();
-        let Some((source_path, plugin_version)) = plugin_sources.get(&plugin_key).cloned() else {
+        let Some(source) = plugin_sources.get(&plugin_key).cloned() else {
             warn!(
                 plugin = plugin_id.plugin_name,
                 marketplace = plugin_id.marketplace_name,
@@ -276,6 +281,13 @@ fn refresh_non_curated_plugin_cache_with_mode(
             );
             continue;
         };
+        let materialized =
+            materialize_marketplace_plugin_source(codex_home, &source).map_err(|err| {
+                format!("failed to materialize plugin source for {plugin_key}: {err}")
+            })?;
+        let source_path = materialized.path.clone();
+        let plugin_version = plugin_version_for_source(source_path.as_path())
+            .map_err(|err| format!("failed to read plugin version for {plugin_key}: {err}"))?;
 
         if mode == NonCuratedCacheRefreshMode::IfVersionChanged
             && store.active_plugin_version(&plugin_id).as_deref() == Some(plugin_version.as_str())
@@ -836,3 +848,186 @@ fn normalize_plugin_mcp_server_value(
 struct PluginMcpDiscovery {
     mcp_servers: HashMap<String, McpServerConfig>,
 }
+
+#[derive(Debug)]
+pub struct MaterializedMarketplacePluginSource {
+    pub path: AbsolutePathBuf,
+    _tempdir: Option<TempDir>,
+}
+
+pub fn materialize_marketplace_plugin_source(
+    codex_home: &Path,
+    source: &MarketplacePluginSource,
+) -> Result<MaterializedMarketplacePluginSource, String> {
+    match source {
+        MarketplacePluginSource::Local { path } => Ok(MaterializedMarketplacePluginSource {
+            path: path.clone(),
+            _tempdir: None,
+        }),
+        MarketplacePluginSource::Git {
+            url,
+            path,
+            ref_name,
+            sha,
+        } => {
+            let staging_root = codex_home.join("plugins/.marketplace-plugin-source-staging");
+            fs::create_dir_all(&staging_root).map_err(|err| {
+                format!(
+                    "failed to create marketplace plugin source staging directory {}: {err}",
+                    staging_root.display()
+                )
+            })?;
+            let tempdir = tempfile::Builder::new()
+                .prefix("marketplace-plugin-source-")
+                .tempdir_in(&staging_root)
+                .map_err(|err| {
+                    format!(
+                        "failed to create marketplace plugin source staging directory in {}: {err}",
+                        staging_root.display()
+                    )
+                })?;
+            clone_git_plugin_source(
+                url,
+                ref_name.as_deref(),
+                sha.as_deref(),
+                path.as_deref(),
+                tempdir.path(),
+            )?;
+            let path = if let Some(path) = path {
+                AbsolutePathBuf::try_from(tempdir.path().join(path)).map_err(|err| {
+                    format!("failed to resolve materialized plugin source path: {err}")
+                })?
+            } else {
+                AbsolutePathBuf::try_from(tempdir.path().to_path_buf()).map_err(|err| {
+                    format!("failed to resolve materialized plugin source path: {err}")
+                })?
+            };
+            Ok(MaterializedMarketplacePluginSource {
+                path,
+                _tempdir: Some(tempdir),
+            })
+        }
+    }
+}
+
+fn clone_git_plugin_source(
+    url: &str,
+    ref_name: Option<&str>,
+    sha: Option<&str>,
+    sparse_checkout_path: Option<&str>,
+    destination: &Path,
+) -> Result<(), String> {
+    if let Some(sparse_checkout_path) = sparse_checkout_path {
+        run_git(
+            &[
+                "clone",
+                "--filter=blob:none",
+                "--sparse",
+                "--no-checkout",
+                url,
+                destination.to_string_lossy().as_ref(),
+            ],
+            /*cwd*/ None,
+        )?;
+        run_git(
+            &[
+                "sparse-checkout",
+                "set",
+                "--no-cone",
+                "--",
+                sparse_checkout_path,
+            ],
+            Some(destination),
+        )?;
+    } else {
+        run_git(
+            &["clone", url, destination.to_string_lossy().as_ref()],
+            /*cwd*/ None,
+        )?;
+    }
+    if let Some(target) = sha.or(ref_name) {
+        run_git(&["checkout", target], Some(destination))?;
+    } else if sparse_checkout_path.is_some() {
+        run_git(&["checkout"], Some(destination))?;
+    }
+    Ok(())
+}
+
+fn run_git(args: &[&str], cwd: Option<&Path>) -> Result<(), String> {
+    let mut command = Command::new("git");
+    command.args(args);
+    command.env("GIT_TERMINAL_PROMPT", "0");
+    if let Some(cwd) = cwd {
+        command.current_dir(cwd);
+    }
+
+    let output = command
+        .output()
+        .map_err(|err| format!("failed to run git {}: {err}", args.join(" ")))?;
+    if output.status.success() {
+        return Ok(());
+    }
+
+    Err(format!(
+        "git {} failed with status {}\nstdout:\n{}\nstderr:\n{}",
+        args.join(" "),
+        output.status,
+        String::from_utf8_lossy(&output.stdout).trim(),
+        String::from_utf8_lossy(&output.stderr).trim()
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn materialize_git_subdir_uses_sparse_checkout() {
+        let codex_home = tempfile::tempdir().expect("create codex home");
+        let repo = tempfile::tempdir().expect("create git repo");
+        let plugin_dir = repo.path().join("plugins/toolkit");
+        fs::create_dir_all(&plugin_dir).expect("create plugin directory");
+        fs::create_dir_all(repo.path().join("plugins/other")).expect("create other plugin");
+        fs::write(plugin_dir.join("marker.txt"), "toolkit").expect("write plugin marker");
+        fs::write(repo.path().join("plugins/other/marker.txt"), "other")
+            .expect("write other marker");
+        fs::write(repo.path().join("root.txt"), "root").expect("write root marker");
+
+        run_git(&["init"], Some(repo.path())).expect("init git repo");
+        run_git(
+            &["config", "user.email", "test@example.com"],
+            Some(repo.path()),
+        )
+        .expect("configure git email");
+        run_git(&["config", "user.name", "Test User"], Some(repo.path()))
+            .expect("configure git name");
+        run_git(&["add", "."], Some(repo.path())).expect("stage git repo");
+        run_git(&["commit", "-m", "init"], Some(repo.path())).expect("commit git repo");
+
+        let materialized = materialize_marketplace_plugin_source(
+            codex_home.path(),
+            &MarketplacePluginSource::Git {
+                url: repo.path().display().to_string(),
+                path: Some("plugins/toolkit".to_string()),
+                ref_name: None,
+                sha: None,
+            },
+        )
+        .expect("materialize git source");
+
+        assert_eq!(
+            plugin_dir.file_name(),
+            materialized.path.as_path().file_name()
+        );
+        assert!(materialized.path.as_path().join("marker.txt").is_file());
+        let checkout_root = materialized
+            .path
+            .as_path()
+            .parent()
+            .and_then(Path::parent)
+            .expect("materialized path should be nested under checkout root");
+        assert!(!checkout_root.join("root.txt").exists());
+        assert!(!checkout_root.join("plugins/other/marker.txt").exists());
+    }
+}

codex-rs/core-plugins/src/marketplace.rs

@@ -9,7 +9,6 @@ use codex_protocol::protocol::Product;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use dirs::home_dir;
 use serde::Deserialize;
-use serde::Deserializer;
 use serde_json::Value as JsonValue;
 use std::fs;
 use std::io;
@@ -67,7 +66,15 @@ pub struct MarketplacePlugin {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum MarketplacePluginSource {
-    Local { path: AbsolutePathBuf },
+    Local {
+        path: AbsolutePathBuf,
+    },
+    Git {
+        url: String,
+        path: Option<String>,
+        ref_name: Option<String>,
+        sha: Option<String>,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -386,27 +393,26 @@ fn resolve_marketplace_plugin_entry(
         policy,
         category,
     } = plugin;
-    let Some(source_path) = resolve_supported_plugin_source_path(marketplace_path, &name, source)
-    else {
+    let Some(source) = resolve_supported_plugin_source(marketplace_path, &name, source) else {
         return Ok(None);
     };
 
-    let manifest = load_plugin_manifest(source_path.as_path());
-    let mut interface = manifest
-        .as_ref()
-        .and_then(|manifest| manifest.interface.clone());
-    if let Some(category) = category {
-        // Marketplace taxonomy wins when both sources provide a category.
-        interface
-            .get_or_insert_with(PluginManifestInterface::default)
-            .category = Some(category);
-    }
+    let manifest = match &source {
+        MarketplacePluginSource::Local { path } => load_plugin_manifest(path.as_path()),
+        MarketplacePluginSource::Git { .. } => None,
+    };
+    let interface = plugin_interface_with_marketplace_category(
+        manifest
+            .as_ref()
+            .and_then(|manifest| manifest.interface.clone()),
+        category,
+    );
 
     Ok(Some(ResolvedMarketplacePlugin {
         plugin_id: PluginId::new(name, marketplace_name.to_string()).map_err(|err| match err {
             PluginIdError::Invalid(message) => MarketplaceError::InvalidPlugin(message),
         })?,
-        source: MarketplacePluginSource::Local { path: source_path },
+        source,
         policy: MarketplacePluginPolicy {
             installation: policy.installation,
             authentication: policy.authentication,
@@ -417,27 +423,13 @@ fn resolve_marketplace_plugin_entry(
     }))
 }
 
-fn resolve_supported_plugin_source_path(
+fn resolve_supported_plugin_source(
     marketplace_path: &AbsolutePathBuf,
     plugin_name: &str,
     source: RawMarketplaceManifestPluginSource,
-) -> Option<AbsolutePathBuf> {
+) -> Option<MarketplacePluginSource> {
     match source {
-        RawMarketplaceManifestPluginSource::Local { path } => {
-            match resolve_local_plugin_source_path(marketplace_path, &path) {
-                Ok(path) => Some(path),
-                Err(err) => {
-                    warn!(
-                        path = %marketplace_path.display(),
-                        plugin = plugin_name,
-                        error = %err,
-                        "skipping marketplace plugin that failed to resolve"
-                    );
-                    None
-                }
-            }
-        }
-        RawMarketplaceManifestPluginSource::Unsupported => {
+        RawMarketplaceManifestPluginSource::Unsupported(_) => {
             warn!(
                 path = %marketplace_path.display(),
                 plugin = plugin_name,
@@ -445,27 +437,85 @@ fn resolve_supported_plugin_source_path(
             );
             None
         }
+        source => match resolve_plugin_source(marketplace_path, source) {
+            Ok(source) => Some(source),
+            Err(err) => {
+                warn!(
+                    path = %marketplace_path.display(),
+                    plugin = plugin_name,
+                    error = %err,
+                    "skipping marketplace plugin that failed to resolve"
+                );
+                None
+            }
+        },
+    }
+}
+
+fn resolve_plugin_source(
+    marketplace_path: &AbsolutePathBuf,
+    source: RawMarketplaceManifestPluginSource,
+) -> Result<MarketplacePluginSource, MarketplaceError> {
+    match source {
+        RawMarketplaceManifestPluginSource::Path(path)
+        | RawMarketplaceManifestPluginSource::Object(
+            RawMarketplaceManifestPluginSourceObject::Local { path },
+        ) => Ok(MarketplacePluginSource::Local {
+            path: resolve_local_plugin_source_path(marketplace_path, &path)?,
+        }),
+        RawMarketplaceManifestPluginSource::Object(
+            RawMarketplaceManifestPluginSourceObject::Url {
+                url,
+                path,
+                ref_name,
+                sha,
+            },
+        ) => Ok(MarketplacePluginSource::Git {
+            url: normalize_git_plugin_source_url(marketplace_path, &url)?,
+            path: path
+                .as_deref()
+                .map(|path| normalize_remote_plugin_subdir(marketplace_path, path))
+                .transpose()?,
+            ref_name: normalize_optional_git_selector(&ref_name),
+            sha: normalize_optional_git_selector(&sha),
+        }),
+        RawMarketplaceManifestPluginSource::Object(
+            RawMarketplaceManifestPluginSourceObject::GitSubdir {
+                url,
+                path,
+                ref_name,
+                sha,
+            },
+        ) => Ok(MarketplacePluginSource::Git {
+            url: normalize_git_plugin_source_url(marketplace_path, &url)?,
+            path: Some(normalize_remote_plugin_subdir(marketplace_path, &path)?),
+            ref_name: normalize_optional_git_selector(&ref_name),
+            sha: normalize_optional_git_selector(&sha),
+        }),
+        RawMarketplaceManifestPluginSource::Unsupported(_) => {
+            unreachable!("unsupported plugin sources should be filtered before resolution")
+        }
     }
 }
 
 fn resolve_local_plugin_source_path(
     marketplace_path: &AbsolutePathBuf,
-    source_path: &str,
+    path: &str,
 ) -> Result<AbsolutePathBuf, MarketplaceError> {
-    let Some(source_path) = source_path.strip_prefix("./") else {
+    let Some(path) = path.strip_prefix("./") else {
         return Err(MarketplaceError::InvalidMarketplaceFile {
             path: marketplace_path.to_path_buf(),
             message: "local plugin source path must start with `./`".to_string(),
         });
     };
-    if source_path.is_empty() {
+    if path.is_empty() {
         return Err(MarketplaceError::InvalidMarketplaceFile {
             path: marketplace_path.to_path_buf(),
             message: "local plugin source path must not be empty".to_string(),
         });
     }
 
-    let relative_source_path = Path::new(source_path);
+    let relative_source_path = Path::new(path);
     if relative_source_path
         .components()
         .any(|component| !matches!(component, Component::Normal(_)))
@@ -481,6 +531,151 @@ fn resolve_local_plugin_source_path(
     Ok(marketplace_root_dir(marketplace_path)?.join(relative_source_path))
 }
 
+fn normalize_remote_plugin_subdir(
+    marketplace_path: &AbsolutePathBuf,
+    path: &str,
+) -> Result<String, MarketplaceError> {
+    let path = path.trim();
+    let path = path.strip_prefix("./").unwrap_or(path);
+    if path.is_empty() {
+        return Err(MarketplaceError::InvalidMarketplaceFile {
+            path: marketplace_path.to_path_buf(),
+            message: "git plugin source path must not be empty".to_string(),
+        });
+    }
+    let relative_path = Path::new(path);
+    if relative_path
+        .components()
+        .any(|component| !matches!(component, Component::Normal(_)))
+    {
+        return Err(MarketplaceError::InvalidMarketplaceFile {
+            path: marketplace_path.to_path_buf(),
+            message: "git plugin source path must stay within the repository root".to_string(),
+        });
+    }
+    Ok(path.to_string())
+}
+
+fn normalize_git_plugin_source_url(
+    marketplace_path: &AbsolutePathBuf,
+    url: &str,
+) -> Result<String, MarketplaceError> {
+    let url = url.trim();
+    if url.is_empty() {
+        return Err(MarketplaceError::InvalidMarketplaceFile {
+            path: marketplace_path.to_path_buf(),
+            message: "git plugin source url must not be empty".to_string(),
+        });
+    }
+    if url.starts_with("http://") || url.starts_with("https://") {
+        return Ok(normalize_github_git_url(url));
+    }
+    if url.starts_with("./")
+        || url.starts_with("../")
+        || url.starts_with(".\\")
+        || url.starts_with("..\\")
+    {
+        return normalize_relative_git_plugin_source_url(marketplace_path, url);
+    }
+    if url.starts_with("file://") || url.starts_with('/') {
+        return Ok(url.to_string());
+    }
+    if url.starts_with("ssh://") || url.starts_with("git@") && url.contains(':') {
+        return Ok(url.to_string());
+    }
+    if let Some(url) = normalize_github_shorthand_url(url) {
+        return Ok(url);
+    }
+
+    Err(MarketplaceError::InvalidMarketplaceFile {
+        path: marketplace_path.to_path_buf(),
+        message: format!("invalid git plugin source url: {url}"),
+    })
+}
+
+fn normalize_relative_git_plugin_source_url(
+    marketplace_path: &AbsolutePathBuf,
+    url: &str,
+) -> Result<String, MarketplaceError> {
+    let mut normalized = marketplace_root_dir(marketplace_path)?
+        .as_path()
+        .to_path_buf();
+    for segment in url.split(['/', '\\']) {
+        match segment {
+            "" | "." => {}
+            ".." => {
+                return Err(MarketplaceError::InvalidMarketplaceFile {
+                    path: marketplace_path.to_path_buf(),
+                    message: "relative git plugin source url must stay within the marketplace root"
+                        .to_string(),
+                });
+            }
+            segment => normalized.push(segment),
+        }
+    }
+    Ok(normalized.display().to_string())
+}
+
+fn normalize_optional_git_selector(value: &Option<String>) -> Option<String> {
+    value
+        .as_deref()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(str::to_string)
+}
+
+fn normalize_github_git_url(url: &str) -> String {
+    if url.starts_with("https://github.com/") && !url.ends_with(".git") {
+        format!("{url}.git")
+    } else {
+        url.to_string()
+    }
+}
+
+fn normalize_github_shorthand_url(source: &str) -> Option<String> {
+    if !looks_like_github_shorthand(source) {
+        return None;
+    }
+    let mut segments = source.split('/');
+    let owner = segments.next()?;
+    let repo = segments.next()?;
+    let repo = repo.strip_suffix(".git").unwrap_or(repo);
+    if repo.is_empty() {
+        return None;
+    }
+    Some(format!("https://github.com/{owner}/{repo}.git"))
+}
+
+fn looks_like_github_shorthand(source: &str) -> bool {
+    let mut segments = source.split('/');
+    let owner = segments.next();
+    let repo = segments.next();
+    let extra = segments.next();
+    owner.is_some_and(is_github_shorthand_segment)
+        && repo.is_some_and(is_github_shorthand_segment)
+        && extra.is_none()
+}
+
+fn is_github_shorthand_segment(segment: &str) -> bool {
+    !segment.is_empty()
+        && segment
+            .chars()
+            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
+}
+
+pub fn plugin_interface_with_marketplace_category(
+    mut interface: Option<PluginManifestInterface>,
+    category: Option<String>,
+) -> Option<PluginManifestInterface> {
+    if let Some(category) = category {
+        // Marketplace taxonomy wins when both sources provide a category.
+        interface
+            .get_or_insert_with(PluginManifestInterface::default)
+            .category = Some(category);
+    }
+    interface
+}
+
 fn marketplace_root_dir(
     marketplace_path: &AbsolutePathBuf,
 ) -> Result<AbsolutePathBuf, MarketplaceError> {
@@ -533,33 +728,36 @@ struct RawMarketplaceManifestPluginPolicy {
     products: Option<Vec<Product>>,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
 enum RawMarketplaceManifestPluginSource {
-    Local { path: String },
-    // Mixed-source marketplaces should still contribute the local plugins we can load.
-    Unsupported,
+    Path(String),
+    Object(RawMarketplaceManifestPluginSourceObject),
+    #[allow(dead_code)]
+    Unsupported(JsonValue),
 }
 
-impl<'de> Deserialize<'de> for RawMarketplaceManifestPluginSource {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        let source = JsonValue::deserialize(deserializer)?;
-        Ok(match source {
-            JsonValue::String(path) => Self::Local { path },
-            JsonValue::Object(object) => match object.get("source").and_then(JsonValue::as_str) {
-                Some("local") => match object.get("path").and_then(JsonValue::as_str) {
-                    Some(path) => Self::Local {
-                        path: path.to_string(),
-                    },
-                    None => Self::Unsupported,
-                },
-                _ => Self::Unsupported,
-            },
-            _ => Self::Unsupported,
-        })
-    }
+#[derive(Debug, Deserialize)]
+#[serde(tag = "source", rename_all = "lowercase")]
+enum RawMarketplaceManifestPluginSourceObject {
+    Local {
+        path: String,
+    },
+    Url {
+        url: String,
+        path: Option<String>,
+        #[serde(rename = "ref")]
+        ref_name: Option<String>,
+        sha: Option<String>,
+    },
+    #[serde(rename = "git-subdir")]
+    GitSubdir {
+        url: String,
+        path: String,
+        #[serde(rename = "ref")]
+        ref_name: Option<String>,
+        sha: Option<String>,
+    },
 }
 
 fn resolve_marketplace_interface(

codex-rs/core-plugins/src/marketplace_tests.rs

@@ -112,6 +112,214 @@ fn find_marketplace_plugin_supports_alternate_layout_and_string_local_source() {
     );
 }
 
+#[test]
+fn find_marketplace_plugin_supports_git_subdir_sources() {
+    let tmp = tempdir().unwrap();
+    let repo_root = tmp.path().join("repo");
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
+    fs::write(
+        repo_root.join(".agents/plugins/marketplace.json"),
+        r#"{
+  "name": "codex-curated",
+  "plugins": [
+    {
+      "name": "remote-plugin",
+      "source": {
+        "source": "git-subdir",
+        "url": "openai/joey_marketplace3",
+        "path": "plugins/toolkit",
+        "ref": "main",
+        "sha": "abc123"
+      }
+    }
+  ]
+}"#,
+    )
+    .unwrap();
+
+    let resolved = find_marketplace_plugin(
+        &AbsolutePathBuf::try_from(repo_root.join(".agents/plugins/marketplace.json")).unwrap(),
+        "remote-plugin",
+    )
+    .unwrap();
+
+    assert_eq!(
+        resolved,
+        ResolvedMarketplacePlugin {
+            plugin_id: PluginId::new("remote-plugin".to_string(), "codex-curated".to_string())
+                .unwrap(),
+            source: MarketplacePluginSource::Git {
+                url: "https://github.com/openai/joey_marketplace3.git".to_string(),
+                path: Some("plugins/toolkit".to_string()),
+                ref_name: Some("main".to_string()),
+                sha: Some("abc123".to_string()),
+            },
+            policy: MarketplacePluginPolicy {
+                installation: MarketplacePluginInstallPolicy::Available,
+                authentication: MarketplacePluginAuthPolicy::OnInstall,
+                products: None,
+            },
+            interface: None,
+            manifest: None,
+        }
+    );
+}
+
+#[test]
+fn find_marketplace_plugin_normalizes_github_shorthand_with_dot_git_suffix() {
+    let tmp = tempdir().unwrap();
+    let repo_root = tmp.path().join("repo");
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
+    fs::write(
+        repo_root.join(".agents/plugins/marketplace.json"),
+        r#"{
+  "name": "codex-curated",
+  "plugins": [
+    {
+      "name": "remote-plugin",
+      "source": {
+        "source": "git-subdir",
+        "url": "openai/toolkit.git",
+        "path": "plugins/toolkit"
+      }
+    }
+  ]
+}"#,
+    )
+    .unwrap();
+
+    let resolved = find_marketplace_plugin(
+        &AbsolutePathBuf::try_from(repo_root.join(".agents/plugins/marketplace.json")).unwrap(),
+        "remote-plugin",
+    )
+    .unwrap();
+
+    assert_eq!(
+        resolved.source,
+        MarketplacePluginSource::Git {
+            url: "https://github.com/openai/toolkit.git".to_string(),
+            path: Some("plugins/toolkit".to_string()),
+            ref_name: None,
+            sha: None,
+        }
+    );
+}
+
+#[test]
+fn find_marketplace_plugin_normalizes_relative_git_source_urls_to_marketplace_root() {
+    for source_url in ["./remotes/toolkit.git", ".\\remotes\\toolkit.git"] {
+        let tmp = tempdir().unwrap();
+        let repo_root = tmp.path().join("repo");
+        let remote_repo = repo_root.join("remotes").join("toolkit.git");
+        fs::create_dir_all(repo_root.join(".git")).unwrap();
+        fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
+        fs::create_dir_all(&remote_repo).unwrap();
+        fs::write(
+            repo_root.join(".agents/plugins/marketplace.json"),
+            format!(
+                r#"{{
+  "name": "codex-curated",
+  "plugins": [
+    {{
+      "name": "remote-plugin",
+      "source": {{
+        "source": "git-subdir",
+        "url": "{}",
+        "path": "plugins/toolkit"
+      }}
+    }}
+  ]
+}}"#,
+                source_url.replace('\\', "\\\\")
+            ),
+        )
+        .unwrap();
+
+        let resolved = find_marketplace_plugin(
+            &AbsolutePathBuf::try_from(repo_root.join(".agents/plugins/marketplace.json")).unwrap(),
+            "remote-plugin",
+        )
+        .unwrap();
+
+        assert_eq!(
+            resolved.source,
+            MarketplacePluginSource::Git {
+                url: remote_repo.display().to_string(),
+                path: Some("plugins/toolkit".to_string()),
+                ref_name: None,
+                sha: None,
+            }
+        );
+    }
+}
+
+#[test]
+fn normalize_relative_git_plugin_source_url_rejects_parent_traversal() {
+    for source_url in [
+        "../toolkit.git",
+        "./../toolkit.git",
+        "..\\toolkit.git",
+        ".\\..\\toolkit.git",
+    ] {
+        let tmp = tempdir().unwrap();
+        let repo_root = tmp.path().join("repo");
+        let marketplace_path = repo_root.join(".agents/plugins/marketplace.json");
+        let marketplace_path = AbsolutePathBuf::try_from(marketplace_path).unwrap();
+        let err =
+            normalize_relative_git_plugin_source_url(&marketplace_path, source_url).unwrap_err();
+
+        assert_eq!(
+            err.to_string(),
+            format!(
+                "invalid marketplace file `{}`: relative git plugin source url must stay within the marketplace root",
+                marketplace_path.display()
+            )
+        );
+    }
+}
+
+#[test]
+fn find_marketplace_plugin_skips_root_equivalent_git_subdir_paths() {
+    for path in [".", "./", "plugins/.."] {
+        let tmp = tempdir().unwrap();
+        let repo_root = tmp.path().join("repo");
+        fs::create_dir_all(repo_root.join(".git")).unwrap();
+        fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
+        fs::write(
+            repo_root.join(".agents/plugins/marketplace.json"),
+            format!(
+                r#"{{
+  "name": "codex-curated",
+  "plugins": [
+    {{
+      "name": "remote-plugin",
+      "source": {{
+        "source": "git-subdir",
+        "url": "openai/toolkit",
+        "path": "{path}"
+      }}
+    }}
+  ]
+}}"#
+            ),
+        )
+        .unwrap();
+
+        let err = find_marketplace_plugin(
+            &AbsolutePathBuf::try_from(repo_root.join(".agents/plugins/marketplace.json")).unwrap(),
+            "remote-plugin",
+        )
+        .unwrap_err();
+
+        assert_eq!(
+            err.to_string(),
+            "plugin `remote-plugin` was not found in marketplace `codex-curated`"
+        );
+    }
+}
+
 #[test]
 fn find_marketplace_plugin_reports_missing_plugin() {
     let tmp = tempdir().unwrap();
@@ -827,7 +1035,7 @@ fn list_marketplaces_reports_marketplace_load_errors() {
 }
 
 #[test]
-fn list_marketplaces_skips_unsupported_plugin_sources_but_keeps_local_plugins() {
+fn list_marketplaces_keeps_remote_and_local_plugin_sources() {
     let tmp = tempdir().unwrap();
     let repo_root = tmp.path().join("repo");
 
@@ -845,7 +1053,7 @@ fn list_marketplaces_skips_unsupported_plugin_sources_but_keeps_local_plugins()
       "name": "url-plugin",
       "source": {
         "source": "url",
-        "url": "https://github.com/example/plugin.git"
+        "url": "https://github.com/example/plugin"
       }
     },
     {
@@ -854,7 +1062,8 @@ fn list_marketplaces_skips_unsupported_plugin_sources_but_keeps_local_plugins()
         "source": "git-subdir",
         "url": "owner/repo",
         "path": "plugins/example",
-        "ref": "main"
+        "ref": "main",
+        "sha": "abc123"
       }
     }
   ]
@@ -869,14 +1078,53 @@ fn list_marketplaces_skips_unsupported_plugin_sources_but_keeps_local_plugins()
     .marketplaces;
 
     assert_eq!(marketplaces.len(), 1);
-    assert_eq!(marketplaces[0].name, "mixed-source-marketplace");
-    assert_eq!(marketplaces[0].plugins.len(), 1);
-    assert_eq!(marketplaces[0].plugins[0].name, "local-plugin");
     assert_eq!(
-        marketplaces[0].plugins[0].source,
-        MarketplacePluginSource::Local {
-            path: AbsolutePathBuf::try_from(repo_root.join("plugins/local-plugin")).unwrap(),
-        }
+        marketplaces[0].plugins,
+        vec![
+            MarketplacePlugin {
+                name: "local-plugin".to_string(),
+                source: MarketplacePluginSource::Local {
+                    path: AbsolutePathBuf::try_from(repo_root.join("plugins/local-plugin"))
+                        .unwrap(),
+                },
+                policy: MarketplacePluginPolicy {
+                    installation: MarketplacePluginInstallPolicy::Available,
+                    authentication: MarketplacePluginAuthPolicy::OnInstall,
+                    products: None,
+                },
+                interface: None,
+            },
+            MarketplacePlugin {
+                name: "url-plugin".to_string(),
+                source: MarketplacePluginSource::Git {
+                    url: "https://github.com/example/plugin.git".to_string(),
+                    path: None,
+                    ref_name: None,
+                    sha: None,
+                },
+                policy: MarketplacePluginPolicy {
+                    installation: MarketplacePluginInstallPolicy::Available,
+                    authentication: MarketplacePluginAuthPolicy::OnInstall,
+                    products: None,
+                },
+                interface: None,
+            },
+            MarketplacePlugin {
+                name: "git-subdir-plugin".to_string(),
+                source: MarketplacePluginSource::Git {
+                    url: "https://github.com/owner/repo.git".to_string(),
+                    path: Some("plugins/example".to_string()),
+                    ref_name: Some("main".to_string()),
+                    sha: Some("abc123".to_string()),
+                },
+                policy: MarketplacePluginPolicy {
+                    installation: MarketplacePluginInstallPolicy::Available,
+                    authentication: MarketplacePluginAuthPolicy::OnInstall,
+                    products: None,
+                },
+                interface: None,
+            },
+        ]
     );
 }
 
@@ -1126,35 +1374,6 @@ fn find_marketplace_plugin_skips_invalid_local_paths() {
     );
 }
 
-#[test]
-fn find_marketplace_plugin_skips_unsupported_sources() {
-    let tmp = tempdir().unwrap();
-    let repo_root = tmp.path().join("repo");
-    fs::create_dir_all(repo_root.join(".git")).unwrap();
-    let marketplace_path = write_alternate_marketplace(
-        &repo_root,
-        r#"{
-  "name": "alternate-marketplace",
-  "plugins": [
-    {
-      "name": "remote-plugin",
-      "source": {
-        "source": "url",
-        "url": "https://github.com/example/plugin.git"
-      }
-    }
-  ]
-}"#,
-    );
-
-    let err = find_marketplace_plugin(&marketplace_path, "remote-plugin").unwrap_err();
-
-    assert_eq!(
-        err.to_string(),
-        "plugin `remote-plugin` was not found in marketplace `alternate-marketplace`"
-    );
-}
-
 #[test]
 fn find_marketplace_plugin_uses_first_duplicate_entry() {
     let tmp = tempdir().unwrap();

codex-rs/core/src/agent_identity.rs

@@ -22,7 +22,7 @@ use rand::TryRngCore;
 use rand::rngs::OsRng;
 use serde::Deserialize;
 use serde::Serialize;
-use tokio::sync::Mutex;
+use tokio::sync::Semaphore;
 use tracing::debug;
 use tracing::info;
 use tracing::warn;
@@ -42,7 +42,7 @@ pub(crate) struct AgentIdentityManager {
     chatgpt_base_url: String,
     feature_enabled: bool,
     abom: AgentBillOfMaterials,
-    ensure_lock: Arc<Mutex<()>>,
+    ensure_lock: Arc<Semaphore>,
 }
 
 impl std::fmt::Debug for AgentIdentityManager {
@@ -110,7 +110,7 @@ impl AgentIdentityManager {
             chatgpt_base_url: config.chatgpt_base_url.clone(),
             feature_enabled: config.features.enabled(Feature::UseAgentIdentity),
             abom: build_abom(session_source),
-            ensure_lock: Arc::new(Mutex::new(())),
+            ensure_lock: Arc::new(Semaphore::new(1)),
         }
     }
 
@@ -137,7 +137,11 @@ impl AgentIdentityManager {
         auth: &CodexAuth,
         binding: &AgentIdentityBinding,
     ) -> Result<StoredAgentIdentity> {
-        let _guard = self.ensure_lock.lock().await;
+        let _guard = self
+            .ensure_lock
+            .acquire()
+            .await
+            .map_err(|_| anyhow::anyhow!("agent identity ensure semaphore closed"))?;
 
         if let Some(stored_identity) = self.load_stored_identity(auth, binding)? {
             info!(
@@ -339,7 +343,7 @@ impl AgentIdentityManager {
             chatgpt_base_url,
             feature_enabled,
             abom: build_abom(session_source),
-            ensure_lock: Arc::new(Mutex::new(())),
+            ensure_lock: Arc::new(Semaphore::new(1)),
         }
     }
 }

codex-rs/core/src/exec_policy.rs

@@ -27,6 +27,7 @@ use codex_shell_command::is_dangerous_command::command_might_be_dangerous;
 use codex_shell_command::is_safe_command::is_known_safe_command;
 use thiserror::Error;
 use tokio::fs;
+use tokio::sync::Semaphore;
 use tokio::task::spawn_blocking;
 use tracing::instrument;
 
@@ -190,7 +191,7 @@ pub enum ExecPolicyUpdateError {
 
 pub(crate) struct ExecPolicyManager {
     policy: ArcSwap<Policy>,
-    update_lock: tokio::sync::Mutex<()>,
+    update_lock: Semaphore,
 }
 
 pub(crate) struct ExecApprovalRequest<'a> {
@@ -206,7 +207,7 @@ impl ExecPolicyManager {
     pub(crate) fn new(policy: Arc<Policy>) -> Self {
         Self {
             policy: ArcSwap::from(policy),
-            update_lock: tokio::sync::Mutex::new(()),
+            update_lock: Semaphore::new(1),
         }
     }
 
@@ -324,7 +325,15 @@ impl ExecPolicyManager {
         codex_home: &Path,
         amendment: &ExecPolicyAmendment,
     ) -> Result<(), ExecPolicyUpdateError> {
-        let _update_guard = self.update_lock.lock().await;
+        let _update_guard =
+            self.update_lock
+                .acquire()
+                .await
+                .map_err(|_| ExecPolicyUpdateError::AddRule {
+                    source: ExecPolicyRuleError::InvalidRule(
+                        "exec policy update semaphore closed".to_string(),
+                    ),
+                })?;
         let policy_path = default_policy_path(codex_home);
         spawn_blocking({
             let policy_path = policy_path.clone();
@@ -369,7 +378,15 @@ impl ExecPolicyManager {
         decision: Decision,
         justification: Option<String>,
     ) -> Result<(), ExecPolicyUpdateError> {
-        let _update_guard = self.update_lock.lock().await;
+        let _update_guard =
+            self.update_lock
+                .acquire()
+                .await
+                .map_err(|_| ExecPolicyUpdateError::AddRule {
+                    source: ExecPolicyRuleError::InvalidRule(
+                        "exec policy update semaphore closed".to_string(),
+                    ),
+                })?;
         let policy_path = default_policy_path(codex_home);
         let host = host.to_string();
         spawn_blocking({

codex-rs/core/src/guardian/review_session.rs

@@ -19,6 +19,7 @@ use codex_protocol::protocol::SandboxPolicy;
 use codex_protocol::protocol::SubAgentSource;
 use serde_json::Value;
 use tokio::sync::Mutex;
+use tokio::sync::Semaphore;
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
@@ -91,7 +92,7 @@ struct GuardianReviewSession {
     codex: Codex,
     cancel_token: CancellationToken,
     reuse_key: GuardianReviewSessionReuseKey,
-    review_lock: Mutex<()>,
+    review_lock: Semaphore,
     state: Mutex<GuardianReviewState>,
 }
 
@@ -281,7 +282,7 @@ impl GuardianReviewSessionManager {
             Ok(mut state) => {
                 if let Some(trunk) = state.trunk.as_ref()
                     && trunk.reuse_key != next_reuse_key
-                    && trunk.review_lock.try_lock().is_ok()
+                    && trunk.review_lock.try_acquire().is_ok()
                 {
                     stale_trunk_to_shutdown = state.trunk.take();
                 }
@@ -336,7 +337,7 @@ impl GuardianReviewSessionManager {
             .await;
         }
 
-        let trunk_guard = match trunk.review_lock.try_lock() {
+        let trunk_guard = match trunk.review_lock.try_acquire() {
             Ok(trunk_guard) => trunk_guard,
             Err(_) => {
                 return Box::pin(self.run_ephemeral_review(
@@ -375,7 +376,7 @@ impl GuardianReviewSessionManager {
             reuse_key,
             codex,
             cancel_token: CancellationToken::new(),
-            review_lock: Mutex::new(()),
+            review_lock: Semaphore::new(1),
             state: Mutex::new(GuardianReviewState {
                 prior_review_count: 0,
                 last_reviewed_transcript_cursor: None,
@@ -397,7 +398,7 @@ impl GuardianReviewSessionManager {
                 reuse_key,
                 codex,
                 cancel_token: CancellationToken::new(),
-                review_lock: Mutex::new(()),
+                review_lock: Semaphore::new(1),
                 state: Mutex::new(GuardianReviewState {
                     prior_review_count: 0,
                     last_reviewed_transcript_cursor: None,
@@ -531,7 +532,7 @@ async fn spawn_guardian_review_session(
         codex,
         cancel_token,
         reuse_key,
-        review_lock: Mutex::new(()),
+        review_lock: Semaphore::new(1),
         state: Mutex::new(GuardianReviewState {
             prior_review_count,
             last_reviewed_transcript_cursor: initial_transcript_cursor,

codex-rs/core/src/plugins/manager.rs

@@ -22,6 +22,7 @@ use codex_core_plugins::loader::load_plugin_mcp_servers;
 use codex_core_plugins::loader::load_plugin_skills;
 use codex_core_plugins::loader::load_plugins_from_layer_stack;
 use codex_core_plugins::loader::log_plugin_load_errors;
+use codex_core_plugins::loader::materialize_marketplace_plugin_source;
 use codex_core_plugins::loader::plugin_telemetry_metadata_from_root;
 use codex_core_plugins::loader::refresh_curated_plugin_cache;
 use codex_core_plugins::loader::refresh_non_curated_plugin_cache;
@@ -39,6 +40,7 @@ use codex_core_plugins::marketplace::find_installable_marketplace_plugin;
 use codex_core_plugins::marketplace::find_marketplace_plugin;
 use codex_core_plugins::marketplace::list_marketplaces;
 use codex_core_plugins::marketplace::load_marketplace;
+use codex_core_plugins::marketplace::plugin_interface_with_marketplace_category;
 use codex_core_plugins::marketplace_upgrade::ConfiguredMarketplaceUpgradeError;
 use codex_core_plugins::marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome;
 use codex_core_plugins::marketplace_upgrade::configured_git_marketplace_names;
@@ -68,7 +70,7 @@ use std::sync::RwLock;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::time::Instant;
-use tokio::sync::Mutex;
+use tokio::sync::Semaphore;
 use toml_edit::value;
 use tracing::info;
 use tracing::warn;
@@ -188,6 +190,12 @@ pub struct PluginDetail {
     pub disabled_skill_paths: HashSet<AbsolutePathBuf>,
     pub apps: Vec<AppConnectorId>,
     pub mcp_server_names: Vec<String>,
+    pub details_unavailable_reason: Option<PluginDetailsUnavailableReason>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PluginDetailsUnavailableReason {
+    InstallRequiredForRemoteSource,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -341,7 +349,7 @@ pub struct PluginsManager {
     configured_marketplace_upgrade_state: RwLock<ConfiguredMarketplaceUpgradeState>,
     non_curated_cache_refresh_state: RwLock<NonCuratedCacheRefreshState>,
     cached_enabled_outcome: RwLock<Option<PluginLoadOutcome>>,
-    remote_sync_lock: Mutex<()>,
+    remote_sync_lock: Semaphore,
     restriction_product: Option<Product>,
     analytics_events_client: RwLock<Option<AnalyticsEventsClient>>,
 }
@@ -371,7 +379,7 @@ impl PluginsManager {
             ),
             non_curated_cache_refresh_state: RwLock::new(NonCuratedCacheRefreshState::default()),
             cached_enabled_outcome: RwLock::new(None),
-            remote_sync_lock: Mutex::new(()),
+            remote_sync_lock: Semaphore::new(1),
             restriction_product,
             analytics_events_client: RwLock::new(None),
         }
@@ -588,8 +596,12 @@ impl PluginsManager {
                 None
             };
         let store = self.store.clone();
+        let codex_home = self.codex_home.clone();
         let result: StorePluginInstallResult = tokio::task::spawn_blocking(move || {
-            let MarketplacePluginSource::Local { path: source_path } = resolved.source;
+            let materialized =
+                materialize_marketplace_plugin_source(codex_home.as_path(), &resolved.source)
+                    .map_err(PluginStoreError::Invalid)?;
+            let source_path = materialized.path;
             if let Some(plugin_version) = plugin_version {
                 store.install_with_version(source_path, resolved.plugin_id, plugin_version)
             } else {
@@ -696,7 +708,9 @@ impl PluginsManager {
         auth: Option<&CodexAuth>,
         additive_only: bool,
     ) -> Result<RemotePluginSyncResult, PluginRemoteSyncError> {
-        let _remote_sync_guard = self.remote_sync_lock.lock().await;
+        let _remote_sync_guard = self.remote_sync_lock.acquire().await.map_err(|_| {
+            PluginRemoteSyncError::Config(anyhow::anyhow!("remote plugin sync semaphore closed"))
+        })?;
 
         if !config.features.enabled(Feature::Plugins) {
             return Ok(RemotePluginSyncResult::default());
@@ -754,6 +768,14 @@ impl PluginsManager {
             let plugin_key = plugin_id.as_key();
             let source_path = match plugin.source {
                 MarketplacePluginSource::Local { path } => path,
+                MarketplacePluginSource::Git { .. } => {
+                    warn!(
+                        plugin = plugin_name,
+                        marketplace = %marketplace_name,
+                        "skipping remote plugin source during remote sync"
+                    );
+                    continue;
+                }
             };
             let current_enabled = configured_plugins
                 .get(&plugin_key)
@@ -1032,9 +1054,55 @@ impl PluginsManager {
             });
         }
 
-        let source_path = match &plugin.source {
-            MarketplacePluginSource::Local { path } => path.clone(),
-        };
+        let plugin_id =
+            PluginId::new(plugin.name.clone(), marketplace_name.to_string()).map_err(|err| {
+                match err {
+                    PluginIdError::Invalid(message) => MarketplaceError::InvalidPlugin(message),
+                }
+            })?;
+        let plugin_key = plugin_id.as_key();
+        if matches!(plugin.source, MarketplacePluginSource::Git { .. }) && !plugin.installed {
+            return Ok(PluginDetail {
+                id: plugin_key,
+                name: plugin.name,
+                description: None,
+                source: plugin.source,
+                policy: plugin.policy,
+                interface: plugin.interface,
+                installed: plugin.installed,
+                enabled: plugin.enabled,
+                skills: Vec::new(),
+                disabled_skill_paths: HashSet::new(),
+                apps: Vec::new(),
+                mcp_server_names: Vec::new(),
+                details_unavailable_reason: Some(
+                    PluginDetailsUnavailableReason::InstallRequiredForRemoteSource,
+                ),
+            });
+        }
+
+        let source_path =
+            if matches!(plugin.source, MarketplacePluginSource::Git { .. }) && plugin.installed {
+                self.store.active_plugin_root(&plugin_id).ok_or_else(|| {
+                    MarketplaceError::InvalidPlugin(format!(
+                        "installed plugin cache entry is missing for {plugin_key}"
+                    ))
+                })?
+            } else {
+                let codex_home = self.codex_home.clone();
+                let source = plugin.source.clone();
+                let materialized = tokio::task::spawn_blocking(move || {
+                    materialize_marketplace_plugin_source(codex_home.as_path(), &source)
+                })
+                .await
+                .map_err(|err| {
+                    MarketplaceError::InvalidPlugin(format!(
+                        "failed to materialize plugin source: {err}"
+                    ))
+                })?
+                .map_err(MarketplaceError::InvalidPlugin)?;
+                materialized.path.clone()
+            };
         if !source_path.as_path().is_dir() {
             return Err(MarketplaceError::InvalidPlugin(
                 "path does not exist or is not a directory".to_string(),
@@ -1044,6 +1112,14 @@ impl PluginsManager {
             MarketplaceError::InvalidPlugin("missing or invalid plugin.json".to_string())
         })?;
         let description = manifest.description.clone();
+        let marketplace_category = plugin
+            .interface
+            .as_ref()
+            .and_then(|interface| interface.category.clone());
+        let interface = plugin_interface_with_marketplace_category(
+            manifest.interface.clone(),
+            marketplace_category,
+        );
         let resolved_skills = load_plugin_skills(
             &source_path,
             &manifest.paths,
@@ -1067,13 +1143,14 @@ impl PluginsManager {
             description,
             source: plugin.source,
             policy: plugin.policy,
-            interface: plugin.interface,
+            interface,
             installed: plugin.installed,
             enabled: plugin.enabled,
             skills: resolved_skills.skills,
             disabled_skill_paths: resolved_skills.disabled_skill_paths,
             apps,
             mcp_server_names,
+            details_unavailable_reason: None,
         })
     }
 

codex-rs/core/src/plugins/manager_tests.rs

@@ -66,6 +66,31 @@ fn write_plugin(root: &Path, dir_name: &str, manifest_name: &str) {
     );
 }
 
+fn init_git_repo(repo: &Path) {
+    run_git(repo, &["init"]);
+    run_git(repo, &["config", "user.email", "codex-test@example.com"]);
+    run_git(repo, &["config", "user.name", "Codex Test"]);
+    run_git(repo, &["add", "."]);
+    run_git(repo, &["commit", "-m", "initial"]);
+}
+
+fn run_git(repo: &Path, args: &[&str]) {
+    let output = std::process::Command::new("git")
+        .arg("-C")
+        .arg(repo)
+        .args(args)
+        .output()
+        .unwrap_or_else(|err| panic!("git should run: {err}"));
+    assert!(
+        output.status.success(),
+        "git -C {} {} failed\nstdout:\n{}\nstderr:\n{}",
+        repo.display(),
+        args.join(" "),
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+}
+
 fn plugin_config_toml(enabled: bool, plugins_feature_enabled: bool) -> String {
     let mut root = toml::map::Map::new();
 
@@ -1050,6 +1075,113 @@ async fn install_plugin_uses_manifest_version_for_non_curated_plugins() {
     );
 }
 
+#[tokio::test]
+async fn install_plugin_supports_git_subdir_marketplace_sources() {
+    let tmp = tempfile::tempdir().unwrap();
+    let repo_root = tmp.path().join("marketplace");
+    let remote_repo = tmp.path().join("remote-plugin-repo");
+    let remote_repo_url = url::Url::from_directory_path(&remote_repo)
+        .unwrap()
+        .to_string();
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
+    write_plugin(&remote_repo, "plugins/toolkit", "toolkit");
+    init_git_repo(&remote_repo);
+    fs::write(
+        repo_root.join(".agents/plugins/marketplace.json"),
+        format!(
+            r#"{{
+  "name": "debug",
+  "plugins": [
+    {{
+      "name": "toolkit",
+      "source": {{
+        "source": "git-subdir",
+        "url": "{remote_repo_url}",
+        "path": "plugins/toolkit"
+      }}
+    }}
+  ]
+}}"#
+        ),
+    )
+    .unwrap();
+
+    let result = PluginsManager::new(tmp.path().to_path_buf())
+        .install_plugin(PluginInstallRequest {
+            plugin_name: "toolkit".to_string(),
+            marketplace_path: AbsolutePathBuf::try_from(
+                repo_root.join(".agents/plugins/marketplace.json"),
+            )
+            .unwrap(),
+        })
+        .await
+        .unwrap();
+
+    let installed_path = tmp.path().join("plugins/cache/debug/toolkit/local");
+    assert_eq!(
+        result,
+        PluginInstallOutcome {
+            plugin_id: PluginId::new("toolkit".to_string(), "debug".to_string()).unwrap(),
+            plugin_version: "local".to_string(),
+            installed_path: AbsolutePathBuf::try_from(installed_path.clone()).unwrap(),
+            auth_policy: MarketplacePluginAuthPolicy::OnInstall,
+        }
+    );
+    assert!(installed_path.join(".codex-plugin/plugin.json").is_file());
+}
+
+#[tokio::test]
+async fn install_plugin_supports_relative_git_subdir_marketplace_sources() {
+    let tmp = tempfile::tempdir().unwrap();
+    let repo_root = tmp.path().join("marketplace");
+    let remote_repo = repo_root.join("remote-plugin-repo");
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    fs::create_dir_all(repo_root.join(".agents/plugins")).unwrap();
+    write_plugin(&remote_repo, "plugins/toolkit", "toolkit");
+    init_git_repo(&remote_repo);
+    fs::write(
+        repo_root.join(".agents/plugins/marketplace.json"),
+        r#"{
+  "name": "debug",
+  "plugins": [
+    {
+      "name": "toolkit",
+      "source": {
+        "source": "git-subdir",
+        "url": "./remote-plugin-repo",
+        "path": "plugins/toolkit"
+      }
+    }
+  ]
+}"#,
+    )
+    .unwrap();
+
+    let result = PluginsManager::new(tmp.path().to_path_buf())
+        .install_plugin(PluginInstallRequest {
+            plugin_name: "toolkit".to_string(),
+            marketplace_path: AbsolutePathBuf::try_from(
+                repo_root.join(".agents/plugins/marketplace.json"),
+            )
+            .unwrap(),
+        })
+        .await
+        .unwrap();
+
+    let installed_path = tmp.path().join("plugins/cache/debug/toolkit/local");
+    assert_eq!(
+        result,
+        PluginInstallOutcome {
+            plugin_id: PluginId::new("toolkit".to_string(), "debug".to_string()).unwrap(),
+            plugin_version: "local".to_string(),
+            installed_path: AbsolutePathBuf::try_from(installed_path.clone()).unwrap(),
+            auth_policy: MarketplacePluginAuthPolicy::OnInstall,
+        }
+    );
+    assert!(installed_path.join(".codex-plugin/plugin.json").is_file());
+}
+
 #[tokio::test]
 async fn uninstall_plugin_removes_cache_and_config_entry() {
     let tmp = tempfile::tempdir().unwrap();
@@ -1433,6 +1565,179 @@ enabled = false
     assert!(outcome.plugin.disabled_skill_paths.is_empty());
 }
 
+#[tokio::test]
+async fn read_plugin_for_config_uninstalled_git_source_requires_install_without_cloning() {
+    let tmp = tempfile::tempdir().unwrap();
+    let repo_root = tmp.path().join("repo");
+    let missing_remote_repo = tmp.path().join("missing-remote-plugin-repo");
+    let missing_remote_repo_url = url::Url::from_directory_path(&missing_remote_repo)
+        .unwrap()
+        .to_string();
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    write_file(
+        &repo_root.join(".agents/plugins/marketplace.json"),
+        &format!(
+            r#"{{
+  "name": "debug",
+  "plugins": [
+    {{
+      "name": "toolkit",
+      "source": {{
+        "source": "git-subdir",
+        "url": "{missing_remote_repo_url}",
+        "path": "plugins/toolkit"
+      }},
+      "policy": {{
+        "installation": "AVAILABLE",
+        "authentication": "ON_INSTALL"
+      }}
+    }}
+  ]
+}}"#
+        ),
+    );
+    write_file(
+        &tmp.path().join(CONFIG_TOML_FILE),
+        r#"[features]
+plugins = true
+"#,
+    );
+
+    let config = load_config(tmp.path(), &repo_root).await;
+    let outcome = PluginsManager::new(tmp.path().to_path_buf())
+        .read_plugin_for_config(
+            &config,
+            &PluginReadRequest {
+                plugin_name: "toolkit".to_string(),
+                marketplace_path: AbsolutePathBuf::try_from(
+                    repo_root.join(".agents/plugins/marketplace.json"),
+                )
+                .unwrap(),
+            },
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(
+        outcome.plugin.details_unavailable_reason,
+        Some(PluginDetailsUnavailableReason::InstallRequiredForRemoteSource)
+    );
+    assert!(!outcome.plugin.installed);
+    assert!(outcome.plugin.description.is_none());
+    assert!(outcome.plugin.skills.is_empty());
+    assert!(outcome.plugin.apps.is_empty());
+    assert!(outcome.plugin.mcp_server_names.is_empty());
+    assert!(
+        !tmp.path()
+            .join("plugins/.marketplace-plugin-source-staging")
+            .exists()
+    );
+}
+
+#[tokio::test]
+async fn read_plugin_for_config_installed_git_source_reads_from_cache_without_cloning() {
+    let tmp = tempfile::tempdir().unwrap();
+    let repo_root = tmp.path().join("repo");
+    let missing_remote_repo = tmp.path().join("missing-remote-plugin-repo");
+    let missing_remote_repo_url = url::Url::from_directory_path(&missing_remote_repo)
+        .unwrap()
+        .to_string();
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    write_file(
+        &repo_root.join(".agents/plugins/marketplace.json"),
+        &format!(
+            r#"{{
+  "name": "debug",
+  "plugins": [
+    {{
+      "name": "toolkit",
+      "source": {{
+        "source": "git-subdir",
+        "url": "{missing_remote_repo_url}",
+        "path": "plugins/toolkit"
+      }},
+      "category": "Developer Tools"
+    }}
+  ]
+}}"#
+        ),
+    );
+    let cached_plugin_root = tmp.path().join("plugins/cache/debug/toolkit/local");
+    write_file(
+        &cached_plugin_root.join(".codex-plugin/plugin.json"),
+        r#"{
+  "name": "toolkit",
+  "description": "Cached toolkit plugin",
+  "interface": {
+    "displayName": "Toolkit"
+  }
+}"#,
+    );
+    write_file(
+        &cached_plugin_root.join("skills/search/SKILL.md"),
+        "---\nname: search\ndescription: search cached data\n---\n",
+    );
+    write_file(
+        &cached_plugin_root.join(".app.json"),
+        r#"{"apps":{"calendar":{"id":"connector_calendar"}}}"#,
+    );
+    write_file(
+        &cached_plugin_root.join(".mcp.json"),
+        r#"{"mcpServers":{"toolkit":{"command":"toolkit-mcp"}}}"#,
+    );
+    write_file(
+        &tmp.path().join(CONFIG_TOML_FILE),
+        r#"[features]
+plugins = true
+
+[plugins."toolkit@debug"]
+enabled = true
+"#,
+    );
+
+    let config = load_config(tmp.path(), &repo_root).await;
+    let outcome = PluginsManager::new(tmp.path().to_path_buf())
+        .read_plugin_for_config(
+            &config,
+            &PluginReadRequest {
+                plugin_name: "toolkit".to_string(),
+                marketplace_path: AbsolutePathBuf::try_from(
+                    repo_root.join(".agents/plugins/marketplace.json"),
+                )
+                .unwrap(),
+            },
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(outcome.plugin.details_unavailable_reason, None);
+    assert_eq!(
+        outcome.plugin.description.as_deref(),
+        Some("Cached toolkit plugin")
+    );
+    assert_eq!(
+        outcome.plugin.interface,
+        Some(PluginManifestInterface {
+            display_name: Some("Toolkit".to_string()),
+            category: Some("Developer Tools".to_string()),
+            ..Default::default()
+        })
+    );
+    assert!(outcome.plugin.installed);
+    assert_eq!(outcome.plugin.skills.len(), 1);
+    assert_eq!(outcome.plugin.skills[0].name, "toolkit:search");
+    assert_eq!(
+        outcome.plugin.apps,
+        vec![AppConnectorId("connector_calendar".to_string())]
+    );
+    assert_eq!(outcome.plugin.mcp_server_names, vec!["toolkit".to_string()]);
+    assert!(
+        !tmp.path()
+            .join("plugins/.marketplace-plugin-source-staging")
+            .exists()
+    );
+}
+
 #[tokio::test]
 async fn sync_plugins_from_remote_returns_default_when_feature_disabled() {
     let tmp = tempfile::tempdir().unwrap();
@@ -2656,6 +2961,65 @@ enabled = true
     );
 }
 
+#[test]
+fn refresh_non_curated_plugin_cache_refreshes_configured_git_source() {
+    let tmp = tempfile::tempdir().unwrap();
+    let repo_root = tmp.path().join("repo");
+    let remote_repo = tmp.path().join("remote-plugin-repo");
+    let remote_repo_url = url::Url::from_directory_path(&remote_repo)
+        .unwrap()
+        .to_string();
+    fs::create_dir_all(repo_root.join(".git")).unwrap();
+    write_plugin_with_version(
+        &remote_repo,
+        "plugins/sample-plugin",
+        "sample-plugin",
+        Some("1.2.3"),
+    );
+    init_git_repo(&remote_repo);
+    write_file(
+        &repo_root.join(".agents/plugins/marketplace.json"),
+        &format!(
+            r#"{{
+  "name": "debug",
+  "plugins": [
+    {{
+      "name": "sample-plugin",
+      "source": {{
+        "source": "git-subdir",
+        "url": "{remote_repo_url}",
+        "path": "plugins/sample-plugin"
+      }}
+    }}
+  ]
+}}"#
+        ),
+    );
+    write_file(
+        &tmp.path().join(CONFIG_TOML_FILE),
+        r#"[features]
+plugins = true
+
+[plugins."sample-plugin@debug"]
+enabled = true
+"#,
+    );
+
+    assert!(
+        refresh_non_curated_plugin_cache(
+            tmp.path(),
+            &[AbsolutePathBuf::try_from(repo_root).unwrap()],
+        )
+        .expect("cache refresh should materialize configured Git plugin")
+    );
+
+    assert!(
+        tmp.path()
+            .join("plugins/cache/debug/sample-plugin/1.2.3")
+            .is_dir()
+    );
+}
+
 #[test]
 fn refresh_non_curated_plugin_cache_returns_false_when_configured_plugins_are_current() {
     let tmp = tempfile::tempdir().unwrap();

codex-rs/core/src/plugins/mod.rs

@@ -35,6 +35,7 @@ pub use manager::ConfiguredMarketplacePlugin;
 pub use manager::OPENAI_BUNDLED_MARKETPLACE_NAME;
 pub use manager::OPENAI_CURATED_MARKETPLACE_NAME;
 pub use manager::PluginDetail;
+pub use manager::PluginDetailsUnavailableReason;
 pub use manager::PluginInstallError;
 pub use manager::PluginInstallOutcome;
 pub use manager::PluginInstallRequest;

codex-rs/core/src/session/mod.rs

@@ -892,7 +892,10 @@ impl Session {
         let Some(started_proxy) = self.services.network_proxy.as_ref() else {
             return;
         };
-        let _refresh_guard = self.managed_network_proxy_refresh_lock.lock().await;
+        let Ok(_refresh_guard) = self.managed_network_proxy_refresh_lock.acquire().await else {
+            error!("managed network proxy refresh semaphore closed");
+            return;
+        };
         let session_configuration = {
             let state = self.state.lock().await;
             state.session_configuration.clone()
@@ -1743,7 +1746,11 @@ impl Session {
         amendment: &NetworkPolicyAmendment,
         network_approval_context: &NetworkApprovalContext,
     ) -> anyhow::Result<()> {
-        let _refresh_guard = self.managed_network_proxy_refresh_lock.lock().await;
+        let _refresh_guard = self
+            .managed_network_proxy_refresh_lock
+            .acquire()
+            .await
+            .map_err(|_| anyhow::anyhow!("managed network proxy refresh semaphore closed"))?;
         let host =
             Self::validated_network_policy_amendment_host(amendment, network_approval_context)?;
         let codex_home = self

codex-rs/core/src/session/session.rs

@@ -1,4 +1,5 @@
 use super::*;
+use tokio::sync::Semaphore;
 
 /// Context for an initialized model agent
 ///
@@ -11,7 +12,7 @@ pub(crate) struct Session {
     pub(super) state: Mutex<SessionState>,
     /// Serializes rebuild/apply cycles for the running proxy; each cycle
     /// rebuilds from the current SessionState while holding this lock.
-    pub(super) managed_network_proxy_refresh_lock: Mutex<()>,
+    pub(super) managed_network_proxy_refresh_lock: Semaphore,
     /// The set of enabled features should be invariant for the lifetime of the
     /// session.
     pub(super) features: ManagedFeatures,
@@ -695,7 +696,7 @@ impl Session {
             agent_status,
             out_of_band_elicitation_paused,
             state: Mutex::new(state),
-            managed_network_proxy_refresh_lock: Mutex::new(()),
+            managed_network_proxy_refresh_lock: Semaphore::new(1),
             features: config.features.clone(),
             pending_mcp_server_refresh_config: Mutex::new(None),
             conversation: Arc::new(RealtimeConversationManager::new()),

codex-rs/core/src/session/tests.rs

@@ -108,6 +108,7 @@ use opentelemetry::trace::TraceContextExt;
 use opentelemetry::trace::TraceId;
 use std::path::Path;
 use std::time::Duration;
+use tokio::sync::Semaphore;
 use tokio::time::sleep;
 use tokio::time::timeout;
 use tracing_opentelemetry::OpenTelemetrySpanExt;
@@ -3015,7 +3016,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         agent_status: agent_status_tx,
         out_of_band_elicitation_paused: watch::channel(false).0,
         state: Mutex::new(state),
-        managed_network_proxy_refresh_lock: Mutex::new(()),
+        managed_network_proxy_refresh_lock: Semaphore::new(1),
         features: config.features.clone(),
         pending_mcp_server_refresh_config: Mutex::new(None),
         conversation: Arc::new(RealtimeConversationManager::new()),
@@ -3978,7 +3979,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
         agent_status: agent_status_tx,
         out_of_band_elicitation_paused: watch::channel(false).0,
         state: Mutex::new(state),
-        managed_network_proxy_refresh_lock: Mutex::new(()),
+        managed_network_proxy_refresh_lock: Semaphore::new(1),
         features: config.features.clone(),
         pending_mcp_server_refresh_config: Mutex::new(None),
         conversation: Arc::new(RealtimeConversationManager::new()),

codex-rs/login/src/auth/manager.rs

@@ -12,7 +12,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::sync::RwLock;
-use tokio::sync::Mutex as AsyncMutex;
+use tokio::sync::Semaphore;
 use tokio::sync::watch;
 
 use codex_app_server_protocol::AuthMode;
@@ -1178,7 +1178,7 @@ pub struct AuthManager {
     enable_codex_api_key_env: bool,
     auth_credentials_store_mode: AuthCredentialsStoreMode,
     forced_chatgpt_workspace_id: RwLock<Option<String>>,
-    refresh_lock: AsyncMutex<()>,
+    refresh_lock: Semaphore,
     external_auth: RwLock<Option<Arc<dyn ExternalAuth>>>,
     auth_state_tx: watch::Sender<()>,
 }
@@ -1246,7 +1246,7 @@ impl AuthManager {
             enable_codex_api_key_env,
             auth_credentials_store_mode,
             forced_chatgpt_workspace_id: RwLock::new(None),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(None),
             auth_state_tx,
         }
@@ -1266,7 +1266,7 @@ impl AuthManager {
             enable_codex_api_key_env: false,
             auth_credentials_store_mode: AuthCredentialsStoreMode::File,
             forced_chatgpt_workspace_id: RwLock::new(None),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(None),
             auth_state_tx,
         })
@@ -1285,7 +1285,7 @@ impl AuthManager {
             enable_codex_api_key_env: false,
             auth_credentials_store_mode: AuthCredentialsStoreMode::File,
             forced_chatgpt_workspace_id: RwLock::new(None),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(None),
             auth_state_tx,
         })
@@ -1302,7 +1302,7 @@ impl AuthManager {
             enable_codex_api_key_env: false,
             auth_credentials_store_mode: AuthCredentialsStoreMode::File,
             forced_chatgpt_workspace_id: RwLock::new(None),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(Some(
                 Arc::new(BearerTokenRefresher::new(config)) as Arc<dyn ExternalAuth>
             )),
@@ -1572,7 +1572,12 @@ impl AuthManager {
     /// can assume that some other instance already refreshed it. If the persisted
     /// token is the same as the cached, then ask the token authority to refresh.
     pub async fn refresh_token(&self) -> Result<(), RefreshTokenError> {
-        let _refresh_guard = self.refresh_lock.lock().await;
+        let _refresh_guard = self.refresh_lock.acquire().await.map_err(|_| {
+            RefreshTokenError::Permanent(RefreshTokenFailedError::new(
+                RefreshTokenFailedReason::Other,
+                REFRESH_TOKEN_UNKNOWN_MESSAGE.to_string(),
+            ))
+        })?;
         let auth_before_reload = self.auth_cached();
         if auth_before_reload
             .as_ref()
@@ -1604,7 +1609,12 @@ impl AuthManager {
     /// observe refreshed token. If the token refresh fails, returns the error to
     /// the caller.
     pub async fn refresh_token_from_authority(&self) -> Result<(), RefreshTokenError> {
-        let _refresh_guard = self.refresh_lock.lock().await;
+        let _refresh_guard = self.refresh_lock.acquire().await.map_err(|_| {
+            RefreshTokenError::Permanent(RefreshTokenFailedError::new(
+                RefreshTokenFailedReason::Other,
+                REFRESH_TOKEN_UNKNOWN_MESSAGE.to_string(),
+            ))
+        })?;
         self.refresh_token_from_authority_impl().await
     }
 

codex-rs/rmcp-client/src/rmcp_client.rs

@@ -67,6 +67,7 @@ use tokio::io::AsyncBufReadExt;
 use tokio::io::BufReader;
 use tokio::process::Command;
 use tokio::sync::Mutex;
+use tokio::sync::Semaphore;
 use tokio::sync::watch;
 use tokio::time;
 use tracing::info;
@@ -563,7 +564,7 @@ pub struct RmcpClient {
     state: Mutex<ClientState>,
     transport_recipe: TransportRecipe,
     initialize_context: Mutex<Option<InitializeContext>>,
-    session_recovery_lock: Mutex<()>,
+    session_recovery_lock: Semaphore,
     elicitation_pause_state: ElicitationPauseState,
 }
 
@@ -592,7 +593,7 @@ impl RmcpClient {
             }),
             transport_recipe,
             initialize_context: Mutex::new(None),
-            session_recovery_lock: Mutex::new(()),
+            session_recovery_lock: Semaphore::new(1),
             elicitation_pause_state: ElicitationPauseState::new(),
         })
     }
@@ -621,7 +622,7 @@ impl RmcpClient {
             }),
             transport_recipe,
             initialize_context: Mutex::new(None),
-            session_recovery_lock: Mutex::new(()),
+            session_recovery_lock: Semaphore::new(1),
             elicitation_pause_state: ElicitationPauseState::new(),
         })
     }
@@ -1227,7 +1228,11 @@ impl RmcpClient {
         &self,
         failed_service: &Arc<RunningService<RoleClient, ElicitationClientService>>,
     ) -> Result<()> {
-        let _recovery_guard = self.session_recovery_lock.lock().await;
+        let _recovery_guard = self
+            .session_recovery_lock
+            .acquire()
+            .await
+            .map_err(|_| anyhow!("MCP client recovery semaphore closed"))?;
 
         {
             let guard = self.state.lock().await;