Clean up failed and abandoned marketplace staging directories (#55689)

## Why

Failed marketplace adds can leave temporary checkouts behind, and interrupted adds or upgrades can leave abandoned staging directories.

## What changed

- Share a staging guard across marketplace adds and upgrades so temporary checkouts are removed on completion or failure, including config-write rollback.
- Reclaim up to four abandoned directories per cleanup pass during add and upgrade operations. Use file locks to preserve active checkouts and operating-system-specific names to avoid reclaiming another platform's staging directories.
- Skip symlinks and unrelated entries, and report cleanup failures as warnings.

## Testing

Add regression coverage for failed adds, CLI and automatic-upgrade cleanup, cross-process ownership and abandonment, lock contention, bounded cleanup, and symlink handling.

GitOrigin-RevId: 20999da0eab861dfb8025aad5e9fba2bcaf818b8

codex-rs/app-server/tests/suite/v2/marketplace_upgrade.rs

@@ -271,6 +271,12 @@ async fn automatic_upgrade_isolates_git_while_explicit_install_preserves_configu
     let manual_plugin_alias = "https://manual.example/plugin.git";
     let manual_plugin_rewrite = format!("url.{}.insteadOf", plugin_source.path().display());
     let config_before = std::fs::read_to_string(&config_path)?;
+    let abandoned = marketplace_install_root(codex_home.path())
+        .join(".staging")
+        .join(format!("marketplace-{}-abandoned", std::env::consts::OS));
+    std::fs::create_dir_all(abandoned.join("checkout"))?;
+    std::fs::write(abandoned.join(".lock"), "")?;
+    std::fs::write(abandoned.join("checkout/payload"), "abandoned upgrade")?;
     let mut builder = TestAppServer::builder()
         .with_codex_home(codex_home.path())
         .with_plugin_startup_tasks()
@@ -305,6 +311,7 @@ async fn automatic_upgrade_isolates_git_while_explicit_install_preserves_configu
         }
     })
     .await?;
+    assert!(!abandoned.exists());
 
     if invalid_user_config {
         // Subsequent explicit operations must pick up a repaired user config normally.

codex-rs/cli/tests/marketplace_add.rs

@@ -40,6 +40,104 @@ fn write_marketplace_source(source: &Path, marker: &str) -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn marketplace_add_git_source_cleans_staging_after_failures() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let source = TempDir::new()?;
+    write_marketplace_source(source.path(), "git source")?;
+    for args in [
+        vec!["init", "--quiet"],
+        vec!["add", "."],
+        vec![
+            "-c",
+            "user.name=Codex Tests",
+            "-c",
+            "user.email=codex@example.com",
+            "-c",
+            "commit.gpgsign=false",
+            "commit",
+            "-m",
+            "marketplace",
+        ],
+    ] {
+        let output = std::process::Command::new("git")
+            .current_dir(source.path())
+            .args(args)
+            .output()?;
+        anyhow::ensure!(
+            output.status.success(),
+            "git fixture failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+
+    // Exercise real Git clones without using the network or changing the process environment.
+    let source_url = "https://example.com/marketplace.git";
+    let rewrite_key = format!("url.{}.insteadOf", source.path().display());
+    let install_root = marketplace_install_root(codex_home.path());
+    let staging_root = install_root.join(".staging");
+    let abandoned = staging_root.join(format!("marketplace-{}-abandoned", std::env::consts::OS));
+    std::fs::create_dir_all(abandoned.join("checkout"))?;
+    std::fs::write(abandoned.join(".lock"), "")?;
+    std::fs::write(abandoned.join("checkout/payload"), "abandoned clone")?;
+    for (args, error) in [
+        (["--ref", "missing-ref"], "git checkout missing-ref failed"),
+        (
+            ["--sparse", "plugins/sample"],
+            "marketplace root does not contain a supported manifest",
+        ),
+    ] {
+        for _ in 0..2 {
+            codex_command(codex_home.path())?
+                .current_dir(codex_home.path())
+                .env("GIT_CONFIG_COUNT", "1")
+                .env("GIT_CONFIG_KEY_0", &rewrite_key)
+                .env("GIT_CONFIG_VALUE_0", source_url)
+                .args(["plugin", "marketplace", "add", source_url])
+                .args(args)
+                .assert()
+                .failure()
+                .stderr(contains(error));
+
+            assert_eq!(std::fs::read_dir(&staging_root)?.count(), 0);
+            assert!(!install_root.join("debug").exists());
+            assert!(!codex_home.path().join(CONFIG_TOML_FILE).exists());
+        }
+    }
+
+    codex_command(codex_home.path())?
+        .current_dir(codex_home.path())
+        .env("GIT_CONFIG_COUNT", "1")
+        .env("GIT_CONFIG_KEY_0", &rewrite_key)
+        .env("GIT_CONFIG_VALUE_0", source_url)
+        .args(["plugin", "marketplace", "add", source_url])
+        .assert()
+        .success();
+    let config = std::fs::read_to_string(codex_home.path().join(CONFIG_TOML_FILE))?;
+
+    // A different ref bypasses the already-added shortcut and conflicts after cloning.
+    codex_command(codex_home.path())?
+        .current_dir(codex_home.path())
+        .env("GIT_CONFIG_COUNT", "1")
+        .env("GIT_CONFIG_KEY_0", &rewrite_key)
+        .env("GIT_CONFIG_VALUE_0", source_url)
+        .args(["plugin", "marketplace", "add", source_url, "--ref", "HEAD"])
+        .assert()
+        .failure()
+        .stderr(contains("already added from a different source"));
+
+    assert_eq!(std::fs::read_dir(&staging_root)?.count(), 0);
+    assert_eq!(
+        std::fs::read_to_string(install_root.join("debug/plugins/sample/marker.txt"))?,
+        "git source"
+    );
+    assert_eq!(
+        std::fs::read_to_string(codex_home.path().join(CONFIG_TOML_FILE))?,
+        config
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn marketplace_add_local_directory_source() -> Result<()> {
     let codex_home = TempDir::new()?;

codex-rs/cli/tests/marketplace_upgrade.rs

@@ -16,6 +16,12 @@ fn codex_command(codex_home: &Path) -> Result<assert_cmd::Command> {
 #[tokio::test]
 async fn marketplace_upgrade_runs_under_plugin() -> Result<()> {
     let codex_home = TempDir::new()?;
+    let abandoned = marketplace_install_root(codex_home.path())
+        .join(".staging")
+        .join(format!("marketplace-{}-abandoned", std::env::consts::OS));
+    std::fs::create_dir_all(abandoned.join("checkout"))?;
+    std::fs::write(abandoned.join(".lock"), "")?;
+    std::fs::write(abandoned.join("checkout/payload"), "abandoned clone")?;
 
     codex_command(codex_home.path())?
         .args(["plugin", "marketplace", "upgrade"])
@@ -23,6 +29,8 @@ async fn marketplace_upgrade_runs_under_plugin() -> Result<()> {
         .success()
         .stdout(contains("No configured Git marketplaces to upgrade."));
 
+    assert!(!abandoned.exists());
+
     Ok(())
 }
 
@@ -88,6 +96,10 @@ async fn marketplace_upgrade_json_prints_upgrade_outcome() -> Result<()> {
         manifest
     );
     assert!(!codex_home.path().join("config.toml").exists());
+    assert_eq!(
+        std::fs::read_dir(marketplace_install_root(codex_home.path()).join(".staging"))?.count(),
+        0
+    );
 
     Ok(())
 }

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

@@ -18,6 +18,7 @@ pub mod marketplace;
 pub mod marketplace_add;
 mod marketplace_policy;
 pub mod marketplace_remove;
+mod marketplace_staging;
 pub mod marketplace_upgrade;
 mod npm_source;
 mod plugin_bundle_archive;

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

@@ -1,12 +1,13 @@
 use crate::installed_marketplaces::marketplace_install_root;
 use crate::marketplace_policy::validate_marketplace_name_for_add;
 use crate::marketplace_policy::validate_marketplace_source_for_add;
+use crate::marketplace_staging::MarketplaceStagingDir;
+use crate::marketplace_staging::cleanup_marketplace_staging;
 use codex_config::ConfigRequirements;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use std::fs;
 use std::path::Path;
 use std::path::PathBuf;
-use tempfile::Builder;
 
 mod install;
 mod metadata;
@@ -14,7 +15,6 @@ mod source;
 
 use install::clone_git_source;
 use install::ensure_marketplace_destination_is_inside_install_root;
-use install::marketplace_staging_root;
 use install::replace_marketplace_root;
 use install::safe_marketplace_dir_name;
 use metadata::MarketplaceInstallMetadata;
@@ -111,6 +111,10 @@ where
         ))
     })?;
 
+    // Remove abandoned staging directories from earlier operations.
+    // This is generally fast and rare operation so it doesn't deserve a separate cleanup task
+    cleanup_marketplace_staging(&install_root);
+
     let install_metadata = MarketplaceInstallMetadata::from_source(&source, &sparse_paths);
     if let Some(existing_root) =
         installed_marketplace_root_for_source(codex_home, &install_root, &install_metadata)?
@@ -153,27 +157,18 @@ where
         });
     }
 
-    let staging_root = marketplace_staging_root(&install_root);
-    fs::create_dir_all(&staging_root).map_err(|err| {
+    let staged_dir = MarketplaceStagingDir::new(&install_root).map_err(|err| {
         MarketplaceAddError::Internal(format!(
-            "failed to create marketplace staging directory {}: {err}",
-            staging_root.display()
+            "failed to create temporary marketplace directory in {}: {err}",
+            install_root.display()
         ))
     })?;
-    let staged_root = Builder::new()
-        .prefix("marketplace-add-")
-        .tempdir_in(&staging_root)
-        .map_err(|err| {
-            MarketplaceAddError::Internal(format!(
-                "failed to create temporary marketplace directory in {}: {err}",
-                staging_root.display()
-            ))
-        })?;
-    let staged_root = staged_root.keep();
+    // Retain the cleanup guard through validation, installation, and config-write rollback.
+    let staged_root = staged_dir.path();
 
-    stage_marketplace_source(&source, &sparse_paths, &staged_root, clone_source)?;
+    stage_marketplace_source(&source, &sparse_paths, staged_root, clone_source)?;
 
-    let marketplace_name = validate_marketplace_source_root(&staged_root)?;
+    let marketplace_name = validate_marketplace_source_root(staged_root)?;
     validate_marketplace_name_for_add(managed_marketplace_name, &marketplace_name)
         .map_err(MarketplaceAddError::InvalidRequest)?;
 
@@ -184,7 +179,7 @@ where
             "marketplace '{marketplace_name}' is already added from a different source; remove it before adding this source"
         )));
     }
-    replace_marketplace_root(&staged_root, &destination).map_err(|err| {
+    replace_marketplace_root(staged_root, &destination).map_err(|err| {
         MarketplaceAddError::Internal(format!(
             "failed to install marketplace at {}: {err}",
             destination.display()
@@ -193,7 +188,7 @@ where
     if let Err(err) =
         record_added_marketplace_entry(codex_home, &marketplace_name, &install_metadata)
     {
-        if let Err(rollback_err) = fs::rename(&destination, &staged_root) {
+        if let Err(rollback_err) = fs::rename(&destination, staged_root) {
             return Err(MarketplaceAddError::Internal(format!(
                 "{err}; additionally failed to roll back installed marketplace at {}: {rollback_err}",
                 destination.display()

codex-rs/core-plugins/src/marketplace_add/install.rs

@@ -1,7 +1,6 @@
 use super::MarketplaceAddError;
 use std::fs;
 use std::path::Path;
-use std::path::PathBuf;
 
 pub(super) fn clone_git_source(
     url: &str,
@@ -105,10 +104,6 @@ pub(super) fn replace_marketplace_root(
     fs::rename(staged_root, destination)
 }
 
-pub(super) fn marketplace_staging_root(install_root: &Path) -> PathBuf {
-    install_root.join(".staging")
-}
-
 fn run_git(args: &[&str], cwd: Option<&Path>) -> Result<(), MarketplaceAddError> {
     let mut command = codex_utils_process::background_command("git");
     command

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

@@ -0,0 +1,161 @@
+//! Owns temporary marketplace checkouts and reclaims abandoned ones.
+//! A coordination lock protects publication of each checkout's lifetime lock. Checkouts
+//! live below their owner directory so Git and activation cannot replace the lock file.
+//! Cleanup claims a limited batch under coordination, then deletes it outside that lock.
+
+use codex_utils_path::symlink_metadata_if_exists;
+use std::fs;
+use std::fs::File;
+use std::io;
+use std::path::Path;
+use std::path::PathBuf;
+use tempfile::TempDir;
+use tracing::warn;
+
+const MAX_CLEANUP_DIRECTORIES: usize = 4;
+const COORDINATION_LOCK: &str = ".staging.lock";
+const CLEANUP_LOCK: &str = ".staging-cleanup.lock";
+const OWNER_LOCK: &str = ".lock";
+
+pub(crate) struct MarketplaceStagingDir {
+    // Close the lock before TempDir removes its directory, including on Windows.
+    _lock: File,
+    _directory: TempDir,
+    checkout: PathBuf,
+}
+
+impl MarketplaceStagingDir {
+    pub(crate) fn new(install_root: &Path) -> io::Result<Self> {
+        let staging_root = Self::staging_root(install_root);
+        fs::create_dir_all(&staging_root)?;
+        let coordination = open_lock(&install_root.join(COORDINATION_LOCK))?;
+        coordination.lock()?;
+        let directory = tempfile::Builder::new()
+            .prefix(&Self::directory_prefix())
+            .tempdir_in(&staging_root)?;
+        let lock = File::options()
+            .read(true)
+            .write(true)
+            .create_new(true)
+            .open(directory.path().join(OWNER_LOCK))?;
+        lock.lock()?;
+        let checkout = directory.path().join("checkout");
+        Ok(Self {
+            _lock: lock,
+            _directory: directory,
+            checkout,
+        })
+    }
+
+    pub(crate) fn path(&self) -> &Path {
+        &self.checkout
+    }
+
+    fn staging_root(install_root: &Path) -> PathBuf {
+        install_root.join(".staging")
+    }
+
+    fn directory_prefix() -> String {
+        // Windows and WSL may share a home without sharing a file-lock namespace.
+        format!("marketplace-{}-", std::env::consts::OS)
+    }
+}
+
+pub(crate) fn cleanup_marketplace_staging(install_root: &Path) {
+    if let Err(err) = remove_stale_staging_dirs(install_root) {
+        warn!("failed to clean up marketplace staging directories: {err}");
+    }
+}
+
+fn open_lock(path: &Path) -> io::Result<File> {
+    File::options()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(path)
+}
+
+fn try_lock(lock: File) -> io::Result<Option<File>> {
+    match lock.try_lock() {
+        Ok(()) => Ok(Some(lock)),
+        Err(fs::TryLockError::WouldBlock) => Ok(None),
+        Err(fs::TryLockError::Error(err)) => Err(err),
+    }
+}
+
+fn remove_stale_staging_dirs(install_root: &Path) -> io::Result<()> {
+    let staging_root = MarketplaceStagingDir::staging_root(install_root);
+    let Some(metadata) = symlink_metadata_if_exists(&staging_root)? else {
+        return Ok(());
+    };
+    if !metadata.is_dir() {
+        return Ok(());
+    }
+    // Skip this pass if another process creates a checkout or removes a batch.
+    let Some(_cleanup) = try_lock(open_lock(&install_root.join(CLEANUP_LOCK))?)? else {
+        return Ok(());
+    };
+    let Some(coordination) = try_lock(open_lock(&install_root.join(COORDINATION_LOCK))?)? else {
+        return Ok(());
+    };
+    let prefix = MarketplaceStagingDir::directory_prefix();
+    let mut abandoned = Vec::new();
+    for entry in fs::read_dir(&staging_root)? {
+        let entry = entry?;
+        let path = entry.path();
+        // Continue past an unreadable entry so one failure cannot prevent reclamation.
+        match is_abandoned_staging_dir(&entry, &prefix) {
+            Ok(true) => abandoned.push(path),
+            Ok(false) => {}
+            Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+            Err(err) => {
+                warn!(path = %path.display(), "failed to inspect marketplace staging directory: {err}");
+            }
+        }
+        if abandoned.len() == MAX_CLEANUP_DIRECTORIES {
+            break;
+        }
+    }
+    // Unique checkout names cannot acquire a new owner after publication.
+    // Keep the cleanup lock, but let new checkouts start while deletion runs.
+    drop(coordination);
+    for path in abandoned {
+        if let Err(err) = fs::remove_dir_all(&path)
+            && err.kind() != io::ErrorKind::NotFound
+        {
+            warn!(path = %path.display(), "failed to remove abandoned marketplace staging directory: {err}");
+        }
+    }
+    Ok(())
+}
+
+fn is_abandoned_staging_dir(entry: &fs::DirEntry, prefix: &str) -> io::Result<bool> {
+    if !entry.file_type()?.is_dir() {
+        return Ok(false);
+    }
+    let name = entry.file_name();
+    let Some(name) = name.to_str() else {
+        return Ok(false);
+    };
+    if !name.starts_with(prefix) {
+        return Ok(false);
+    }
+    let owner_lock = entry.path().join(OWNER_LOCK);
+    let Some(metadata) = symlink_metadata_if_exists(&owner_lock)? else {
+        // A crash before lock publication can leave an empty directory.
+        return Ok(true);
+    };
+    if !metadata.is_file() {
+        return Ok(false);
+    }
+    // An unlocked checkout has no Codex owner to activate it. A surviving Git
+    // process may still write here; deletion can interrupt that abandoned clone.
+    let lock = File::options().read(true).write(true).open(owner_lock)?;
+    // Close the probe's lock before cleanup removes the directory on Windows.
+    Ok(try_lock(lock)?.is_some())
+}
+
+#[cfg(test)]
+#[path = "marketplace_staging_tests.rs"]
+mod tests;

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

@@ -0,0 +1,159 @@
+//! Exercises staging ownership across processes and reclamation of abandoned checkouts.
+
+use super::*;
+use anyhow::Result;
+use pretty_assertions::assert_eq;
+use std::io::BufRead;
+use std::io::Read;
+use std::io::Write;
+use std::process::Stdio;
+
+#[test]
+fn cleanup_preserves_live_process_and_reclaims_its_abandoned_checkout() -> Result<()> {
+    const CHILD_HOME: &str = "CODEX_MARKETPLACE_STAGING_CHILD_HOME";
+    if let Some(home) = std::env::var_os(CHILD_HOME) {
+        let staging = MarketplaceStagingDir::new(Path::new(&home))?;
+        fs::create_dir_all(staging.path())?;
+        fs::write(staging.path().join("payload"), "in-flight clone")?;
+        println!("staging-ready");
+        std::io::stdout().flush()?;
+        std::io::stdin().read_exact(&mut [0])?;
+        // Model process exit while a background upgrade still owns a checkout.
+        std::process::exit(0);
+    }
+
+    let home = tempfile::tempdir()?;
+    let mut child = codex_utils_process::background_command(std::env::current_exe()?)
+        .args([
+            "--exact",
+            "marketplace_staging::tests::cleanup_preserves_live_process_and_reclaims_its_abandoned_checkout",
+            "--nocapture",
+        ])
+        .env(CHILD_HOME, home.path())
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .spawn()?;
+    let stdout = std::io::BufReader::new(child.stdout.take().unwrap());
+    let mut ready = false;
+    for line in stdout.lines() {
+        if line?.contains("staging-ready") {
+            ready = true;
+            break;
+        }
+    }
+    assert!(ready, "child should publish its checkout before cleanup");
+    let staging_root = home.path().join(".staging");
+    let active = fs::read_dir(&staging_root)?.next().unwrap()?.path();
+    let payload = active.join("checkout/payload");
+
+    // Cleanup must preserve a checkout owned by another process.
+    remove_stale_staging_dirs(home.path())?;
+    assert_eq!(fs::read_to_string(&payload)?, "in-flight clone");
+    let concurrent = MarketplaceStagingDir::new(home.path())?;
+    fs::create_dir_all(concurrent.path())?;
+    fs::write(concurrent.path().join("payload"), "second clone")?;
+
+    child.stdin.take().unwrap().write_all(b"x")?;
+    assert!(child.wait()?.success());
+    assert!(payload.exists(), "process exit must bypass TempDir cleanup");
+    cleanup_marketplace_staging(home.path());
+    assert!(!active.exists());
+    assert_eq!(
+        fs::read_to_string(concurrent.path().join("payload"))?,
+        "second clone"
+    );
+    drop(concurrent);
+    assert_eq!(fs::read_dir(staging_root)?.count(), 0);
+    Ok(())
+}
+
+#[test]
+fn cleanup_reclaims_unpublished_and_preserves_unrelated_entries() -> Result<()> {
+    let home = tempfile::tempdir()?;
+    let staging_root = home.path().join(".staging");
+    let unrelated = staging_root.join("unrelated");
+    let foreign = staging_root.join("marketplace-other-os-active");
+    let unpublished = staging_root.join(format!(
+        "{}unpublished",
+        MarketplaceStagingDir::directory_prefix()
+    ));
+    for path in [&unrelated, &foreign, &unpublished] {
+        fs::create_dir_all(path)?;
+    }
+    let unrelated_file =
+        staging_root.join(format!("{}file", MarketplaceStagingDir::directory_prefix()));
+    fs::write(&unrelated_file, "keep")?;
+
+    remove_stale_staging_dirs(home.path())?;
+    assert!(!unpublished.exists());
+    assert!(unrelated.exists());
+    assert!(foreign.exists());
+    assert_eq!(fs::read_to_string(unrelated_file)?, "keep");
+    Ok(())
+}
+
+#[cfg(unix)]
+#[test]
+fn cleanup_does_not_follow_symlinks() -> Result<()> {
+    let home = tempfile::tempdir()?;
+    let outside = tempfile::tempdir()?;
+    let staging_root = home.path().join(".staging");
+    fs::create_dir_all(&staging_root)?;
+    fs::write(outside.path().join("payload"), "keep")?;
+    let alias = staging_root.join(format!(
+        "{}directory-link",
+        MarketplaceStagingDir::directory_prefix()
+    ));
+    std::os::unix::fs::symlink(outside.path(), &alias)?;
+    let invalid_owner =
+        staging_root.join(format!("{}link", MarketplaceStagingDir::directory_prefix()));
+    fs::create_dir_all(&invalid_owner)?;
+    std::os::unix::fs::symlink(
+        outside.path().join("payload"),
+        invalid_owner.join(OWNER_LOCK),
+    )?;
+
+    remove_stale_staging_dirs(home.path())?;
+    assert_eq!(fs::read_to_string(outside.path().join("payload"))?, "keep");
+    assert!(alias.is_symlink());
+    assert!(invalid_owner.exists());
+    Ok(())
+}
+
+#[test]
+fn cleanup_limits_each_pass_and_skips_contended_locks() -> Result<()> {
+    let home = tempfile::tempdir()?;
+    let staging_root = home.path().join(".staging");
+    let count = MAX_CLEANUP_DIRECTORIES + 1;
+    for index in 0..count {
+        fs::create_dir_all(staging_root.join(format!(
+            "{}{index}",
+            MarketplaceStagingDir::directory_prefix()
+        )))?;
+    }
+
+    for name in [CLEANUP_LOCK, COORDINATION_LOCK] {
+        let lock = open_lock(&home.path().join(name))?;
+        lock.lock()?;
+        cleanup_marketplace_staging(home.path());
+        assert_eq!(fs::read_dir(&staging_root)?.count(), count);
+        if name == CLEANUP_LOCK {
+            // A cleanup worker must not prevent another operation from creating a checkout.
+            drop(MarketplaceStagingDir::new(home.path())?);
+        }
+    }
+
+    for remaining in [1, 0] {
+        // Other tests can fork while cleanup holds a lock. The child retains
+        // that lock until exec, even after this process closes its file.
+        // Wait for both locks, then release them before testing the batch size.
+        for name in [CLEANUP_LOCK, COORDINATION_LOCK] {
+            let lock = open_lock(&home.path().join(name))?;
+            lock.lock()?;
+            lock.unlock()?;
+        }
+        cleanup_marketplace_staging(home.path());
+        assert_eq!(fs::read_dir(&staging_root)?.count(), remaining);
+    }
+    Ok(())
+}

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

@@ -13,6 +13,8 @@ use crate::marketplace::validate_marketplace_root;
 use crate::marketplace_add::MarketplaceSource;
 use crate::marketplace_policy::MarketplacePolicy;
 use crate::marketplace_policy::validate_marketplace_name_for_add;
+use crate::marketplace_staging::MarketplaceStagingDir;
+use crate::marketplace_staging::cleanup_marketplace_staging;
 use codex_config::ConfigLayerStack;
 use codex_config::types::MarketplaceConfig;
 use codex_config::types::MarketplaceSourceType;
@@ -93,6 +95,8 @@ pub(crate) fn upgrade_configured_git_marketplaces_with_mode(
     mode: PluginGitMode,
     reload_config: &ConfigLayerReload,
 ) -> ConfiguredMarketplaceUpgradeOutcome {
+    let install_root = marketplace_install_root(codex_home);
+    cleanup_marketplace_staging(&install_root);
     let loaded = load_configured_git_marketplaces(config_layer_stack);
     let marketplaces = loaded
         .marketplaces
@@ -108,7 +112,6 @@ pub(crate) fn upgrade_configured_git_marketplaces_with_mode(
         return ConfiguredMarketplaceUpgradeOutcome::default();
     }
 
-    let install_root = marketplace_install_root(codex_home);
     let mut selected_marketplaces = marketplaces
         .iter()
         .map(|marketplace| marketplace.name.clone())
@@ -254,22 +257,12 @@ fn upgrade_configured_git_marketplace(
         return Ok(None);
     }
 
-    let staging_parent = install_root.join(".staging");
-    std::fs::create_dir_all(&staging_parent).map_err(|err| {
+    let staged_dir = MarketplaceStagingDir::new(install_root).map_err(|err| {
         format!(
-            "failed to create marketplace upgrade staging directory {}: {err}",
-            staging_parent.display()
+            "failed to create temporary marketplace upgrade directory in {}: {err}",
+            install_root.display()
         )
     })?;
-    let staged_dir = tempfile::Builder::new()
-        .prefix("marketplace-upgrade-")
-        .tempdir_in(&staging_parent)
-        .map_err(|err| {
-            format!(
-                "failed to create temporary marketplace upgrade directory in {}: {err}",
-                staging_parent.display()
-            )
-        })?;
 
     let activated_revision = clone_git_source(
         codex_home,
@@ -289,7 +282,7 @@ fn upgrade_configured_git_marketplace(
         ));
     }
     write_installed_marketplace_metadata(staged_dir.path(), marketplace, &activated_revision)?;
-    activate_marketplace_root(&destination, staged_dir, &previous_snapshot, || {
+    activate_marketplace_root(&destination, staged_dir.path(), &previous_snapshot, || {
         ensure_configured_git_marketplace_unchanged(reload_config, marketplace)
     })?;
 

codex-rs/core-plugins/src/marketplace_upgrade/activation.rs

@@ -4,7 +4,6 @@ use serde::Deserialize;
 use serde::Serialize;
 use std::path::Path;
 use std::path::PathBuf;
-use tempfile::TempDir;
 use tracing::warn;
 
 const MARKETPLACE_INSTALL_METADATA_FILE: &str = ".codex-marketplace-install.json";
@@ -87,11 +86,10 @@ pub(super) fn write_installed_marketplace_metadata(
 
 pub(super) fn activate_marketplace_root(
     destination: &Path,
-    staged_dir: TempDir,
+    staged_root: &Path,
     previous_snapshot: &InstalledMarketplaceSnapshot,
     after_activate: impl FnOnce() -> Result<(), String>,
 ) -> Result<(), String> {
-    let staged_root = staged_dir.path();
     let Some(parent) = destination.parent() else {
         return Err(format!(
             "failed to determine marketplace install parent for {}",

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

@@ -311,6 +311,12 @@ fn up_to_date_fast_path_validates_marketplace_name() {
     .expect_err("mismatched marketplace name must not use the up-to-date fast path");
 
     assert!(err.contains("git clone marketplace source failed"));
+    assert_eq!(
+        std::fs::read_dir(install_root.join(".staging"))
+            .expect("read marketplace staging directory")
+            .count(),
+        0
+    );
 }
 
 #[test]
@@ -347,10 +353,9 @@ fn stale_activation_restores_newer_concurrently_installed_marketplace() {
     )
     .expect("write newer installed marketplace metadata");
 
-    let staged_dir = tempfile::Builder::new()
-        .prefix("marketplace-upgrade-")
-        .tempdir_in(&install_root)
-        .expect("create stale upgrade staging directory");
+    let staged_dir =
+        MarketplaceStagingDir::new(&install_root).expect("create stale upgrade staging directory");
+    std::fs::create_dir_all(staged_dir.path()).expect("create stale upgrade checkout");
     std::fs::write(staged_dir.path().join("marker.txt"), "stale snapshot")
         .expect("write stale staged marketplace snapshot");
     super::activation::write_installed_marketplace_metadata(
@@ -362,7 +367,7 @@ fn stale_activation_restores_newer_concurrently_installed_marketplace() {
 
     let err = super::activation::activate_marketplace_root(
         &destination,
-        staged_dir,
+        staged_dir.path(),
         &previous_snapshot,
         || Ok(()),
     )
@@ -384,6 +389,13 @@ fn stale_activation_restores_newer_concurrently_installed_marketplace() {
         &marketplace,
         NEWER_REVISION
     ));
+    drop(staged_dir);
+    assert_eq!(
+        std::fs::read_dir(install_root.join(".staging"))
+            .expect("read marketplace staging directory")
+            .count(),
+        0
+    );
 }
 
 fn config_reloader(codex_home: &Path) -> ConfigLayerReload {

codex-rs/utils/path-utils/src/lib.rs

@@ -1,4 +1,4 @@
-//! Path normalization, replacement, symlink resolution, and atomic writes.
+//! Path normalization, replacement, metadata inspection, symlink resolution, and atomic writes.
 
 pub(crate) mod env;
 pub use env::is_wsl;
@@ -15,6 +15,16 @@ use std::path::Path;
 use std::path::PathBuf;
 use tempfile::NamedTempFile;
 
+/// Read metadata without following the final symlink, returning `None` for a missing path.
+/// Other I/O errors are returned to the caller.
+pub fn symlink_metadata_if_exists(path: &Path) -> io::Result<Option<std::fs::Metadata>> {
+    match std::fs::symlink_metadata(path) {
+        Ok(metadata) => Ok(Some(metadata)),
+        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
+        Err(err) => Err(err),
+    }
+}
+
 pub fn normalize_for_path_comparison(path: impl AsRef<Path>) -> std::io::Result<PathBuf> {
     let canonical = path.as_ref().canonicalize()?;
     Ok(normalize_for_wsl(canonical))

codex-rs/utils/path-utils/src/path_utils_tests.rs

@@ -1,7 +1,33 @@
 use super::replace_path_and_deduplicate;
+use super::symlink_metadata_if_exists;
 use pretty_assertions::assert_eq;
 use std::path::Path;
 
+#[test]
+fn symlink_metadata_if_exists_distinguishes_files_directories_and_missing_paths()
+-> std::io::Result<()> {
+    let directory = tempfile::tempdir()?;
+    let file = directory.path().join("file");
+    std::fs::write(&file, "contents")?;
+
+    let directory_metadata =
+        symlink_metadata_if_exists(directory.path())?.expect("directory exists");
+    let file_metadata = symlink_metadata_if_exists(&file)?.expect("file exists");
+    assert_eq!(
+        (directory_metadata.is_dir(), file_metadata.is_file()),
+        (true, true)
+    );
+    assert!(symlink_metadata_if_exists(&directory.path().join("missing"))?.is_none());
+    Ok(())
+}
+
+#[test]
+fn symlink_metadata_if_exists_preserves_io_errors() {
+    let error = symlink_metadata_if_exists(Path::new("invalid\0path"))
+        .expect_err("an invalid path must not be treated as missing");
+    assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
+}
+
 #[test]
 fn replace_path_and_deduplicate_preserves_other_paths_and_order() {
     let old_path = Path::new("old");
@@ -22,9 +48,27 @@ fn replace_path_and_deduplicate_preserves_other_paths_and_order() {
 #[cfg(unix)]
 mod symlinks {
     use super::super::resolve_symlink_write_paths;
+    use super::super::symlink_metadata_if_exists;
     use pretty_assertions::assert_eq;
     use std::os::unix::fs::symlink;
 
+    #[test]
+    fn symlink_metadata_if_exists_preserves_links_with_and_without_targets() -> std::io::Result<()>
+    {
+        let directory = tempfile::tempdir()?;
+        let target = directory.path().join("target");
+        let link = directory.path().join("link");
+        std::fs::write(&target, "contents")?;
+        symlink(&target, &link)?;
+
+        let linked_metadata = symlink_metadata_if_exists(&link)?.expect("symlink exists");
+        std::fs::remove_file(target)?;
+        let dangling_metadata = symlink_metadata_if_exists(&link)?.expect("symlink still exists");
+        assert!(linked_metadata.is_symlink());
+        assert_eq!(dangling_metadata.file_type(), linked_metadata.file_type());
+        Ok(())
+    }
+
     #[test]
     fn symlink_cycles_fall_back_to_root_write_path() -> std::io::Result<()> {
         let dir = tempfile::tempdir()?;