Wait for agent shutdown before releasing thread

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

@@ -4,7 +4,6 @@ use codex_config::remove_user_marketplace_config;
 use codex_plugin::validate_plugin_segment;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use std::fs;
-use std::io;
 use std::path::Path;
 use std::path::PathBuf;
 
@@ -75,16 +74,9 @@ fn remove_marketplace_sync(
 }
 
 fn remove_marketplace_root(root: &Path) -> Result<Option<AbsolutePathBuf>, MarketplaceRemoveError> {
-    let metadata = match fs::symlink_metadata(root) {
-        Ok(metadata) => metadata,
-        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(None),
-        Err(err) => {
-            return Err(MarketplaceRemoveError::Internal(format!(
-                "failed to inspect installed marketplace root {}: {err}",
-                root.display()
-            )));
-        }
-    };
+    if !root.exists() {
+        return Ok(None);
+    }
 
     let removed_root = AbsolutePathBuf::try_from(root.to_path_buf()).map_err(|err| {
         MarketplaceRemoveError::Internal(format!(
@@ -92,16 +84,16 @@ fn remove_marketplace_root(root: &Path) -> Result<Option<AbsolutePathBuf>, Marke
             root.display()
         ))
     })?;
-    let file_type = metadata.file_type();
-    let remove_result = if file_type.is_dir() {
-        fs::remove_dir_all(root)
-    } else if file_type.is_file() || file_type.is_symlink() {
-        fs::remove_file(root)
-    } else {
-        return Err(MarketplaceRemoveError::Internal(format!(
-            "installed marketplace root {} is neither a file nor a directory",
+    let metadata = fs::symlink_metadata(root).map_err(|err| {
+        MarketplaceRemoveError::Internal(format!(
+            "failed to inspect installed marketplace root {}: {err}",
             root.display()
-        )));
+        ))
+    })?;
+    let remove_result = if metadata.is_dir() {
+        fs::remove_dir_all(root)
+    } else {
+        fs::remove_file(root)
     };
     remove_result.map_err(|err| {
         MarketplaceRemoveError::Internal(format!(
@@ -287,98 +279,6 @@ mod tests {
         assert!(!config.contains("[marketplaces.debug]"));
     }
 
-    #[cfg(unix)]
-    #[test]
-    fn remove_marketplace_sync_removes_dangling_symlink_installed_root() {
-        let codex_home = TempDir::new().unwrap();
-        record_user_marketplace(
-            codex_home.path(),
-            "debug",
-            &MarketplaceConfigUpdate {
-                last_updated: "2026-04-13T00:00:00Z",
-                last_revision: None,
-                source_type: "git",
-                source: "https://github.com/owner/repo.git",
-                ref_name: Some("main"),
-                sparse_paths: &[],
-            },
-        )
-        .unwrap();
-        let installed_root = marketplace_install_root(codex_home.path()).join("debug");
-        fs::create_dir_all(installed_root.parent().unwrap()).unwrap();
-        let missing_target = codex_home.path().join("missing-target");
-        std::os::unix::fs::symlink(&missing_target, &installed_root).unwrap();
-
-        let outcome = remove_marketplace_sync(
-            codex_home.path(),
-            MarketplaceRemoveRequest {
-                marketplace_name: "debug".to_string(),
-            },
-        )
-        .unwrap();
-
-        assert_eq!(
-            outcome,
-            MarketplaceRemoveOutcome {
-                marketplace_name: "debug".to_string(),
-                removed_installed_root: Some(
-                    AbsolutePathBuf::try_from(installed_root.clone()).unwrap()
-                ),
-            }
-        );
-        assert!(fs::symlink_metadata(&installed_root).is_err());
-        let config =
-            fs::read_to_string(codex_home.path().join(codex_config::CONFIG_TOML_FILE)).unwrap();
-        assert!(!config.contains("[marketplaces.debug]"));
-    }
-
-    #[cfg(unix)]
-    #[test]
-    fn remove_marketplace_sync_removes_symlink_installed_root() {
-        let codex_home = TempDir::new().unwrap();
-        record_user_marketplace(
-            codex_home.path(),
-            "debug",
-            &MarketplaceConfigUpdate {
-                last_updated: "2026-04-13T00:00:00Z",
-                last_revision: None,
-                source_type: "git",
-                source: "https://github.com/owner/repo.git",
-                ref_name: Some("main"),
-                sparse_paths: &[],
-            },
-        )
-        .unwrap();
-        let installed_root = marketplace_install_root(codex_home.path()).join("debug");
-        fs::create_dir_all(installed_root.parent().unwrap()).unwrap();
-        let target_root = codex_home.path().join("debug-target");
-        fs::create_dir_all(&target_root).unwrap();
-        std::os::unix::fs::symlink(&target_root, &installed_root).unwrap();
-
-        let outcome = remove_marketplace_sync(
-            codex_home.path(),
-            MarketplaceRemoveRequest {
-                marketplace_name: "debug".to_string(),
-            },
-        )
-        .unwrap();
-
-        assert_eq!(
-            outcome,
-            MarketplaceRemoveOutcome {
-                marketplace_name: "debug".to_string(),
-                removed_installed_root: Some(
-                    AbsolutePathBuf::try_from(installed_root.clone()).unwrap()
-                ),
-            }
-        );
-        assert!(!installed_root.exists());
-        assert!(target_root.exists());
-        let config =
-            fs::read_to_string(codex_home.path().join(codex_config::CONFIG_TOML_FILE)).unwrap();
-        assert!(!config.contains("[marketplaces.debug]"));
-    }
-
     #[test]
     fn remove_marketplace_sync_removes_inline_config_entry() {
         let codex_home = TempDir::new().unwrap();

codex-rs/core/src/agent/control.rs

@@ -693,9 +693,10 @@ impl AgentControl {
             thread.codex.session.ensure_rollout_materialized().await;
             thread.codex.session.flush_rollout().await?;
             if matches!(thread.agent_status().await, AgentStatus::Shutdown) {
+                thread.wait_until_terminated().await;
                 Ok(String::new())
             } else {
-                state.send_op(agent_id, Op::Shutdown {}).await
+                thread.shutdown_and_wait().await.map(|()| String::new())
             }
         } else {
             state.send_op(agent_id, Op::Shutdown {}).await