fix: publish coherent request snapshots (Fixes #536)

Build configured locator graphs off-lock and atomically publish request state. Preserve compatible caches and rebind nested Conda dependencies. Keep callback panics from poisoning publication gates, and prove snapshot, cache, and generation boundaries with deterministic tests.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: karthiknadig

crates/pet-poetry/src/lib.rs

@@ -375,6 +375,8 @@ impl Locator for Poetry {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use pet_core::env::PythonEnv;
+    use pet_core::manager::{EnvManager, EnvManagerType};
     use pet_core::os_environment::EnvironmentApi;
 
     fn path_from_components(components: &[&str]) -> PathBuf {
@@ -598,4 +600,64 @@ mod tests {
 
         assert!(poetry.poetry_executable.read().unwrap().is_none());
     }
+
+    #[test]
+    fn identical_configure_preserves_cached_identification_fidelity() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let prefix = temp_dir.path().join("project-env");
+        let executable = prefix.join(if cfg!(windows) {
+            "Scripts/python.exe"
+        } else {
+            "bin/python"
+        });
+        std::fs::create_dir_all(executable.parent().unwrap()).unwrap();
+        std::fs::write(prefix.join("pyvenv.cfg"), "version = 3.12.0").unwrap();
+        std::fs::write(&executable, "").unwrap();
+        std::fs::write(executable.parent().unwrap().join("activate"), "").unwrap();
+
+        let environment = EnvironmentApi::new();
+        let poetry = Poetry::from(&environment);
+        let workspace = temp_dir.path().join("removed-project");
+        let manager = EnvManager::new(
+            temp_dir.path().join("removed-poetry"),
+            EnvManagerType::Poetry,
+            Some("1.8.0".to_string()),
+        );
+        let config = Configuration {
+            workspace_directories: Some(vec![workspace.clone()]),
+            poetry_executable: Some(manager.executable.clone()),
+            ..Default::default()
+        };
+        poetry.configure(&config);
+        poetry
+            .search_result
+            .write()
+            .unwrap()
+            .replace(LocatorResult {
+                managers: vec![manager.clone()],
+                environments: vec![PythonEnvironment {
+                    executable: Some(executable.clone()),
+                    prefix: Some(prefix.clone()),
+                    kind: Some(PythonEnvironmentKind::Poetry),
+                    manager: Some(manager.clone()),
+                    project: Some(workspace.clone()),
+                    symlinks: Some(vec![executable.clone()]),
+                    ..Default::default()
+                }],
+            });
+
+        assert!(!workspace.exists());
+        assert!(!manager.executable.exists());
+        poetry.configure(&config);
+
+        let identified = poetry
+            .try_from(&PythonEnv::new(
+                executable,
+                Some(prefix),
+                Some("3.12.0".to_string()),
+            ))
+            .expect("cached Poetry environment must remain identifiable");
+        assert_eq!(identified.manager, Some(manager));
+        assert_eq!(identified.project, Some(workspace));
+    }
 }

crates/pet-pyenv/src/lib.rs

@@ -47,6 +47,14 @@ impl PyEnv {
             versions_dir: Arc::new(Mutex::new(None)),
         }
     }
+    pub fn with_conda_locator(&self, conda_locator: Arc<dyn CondaLocator>) -> PyEnv {
+        PyEnv {
+            env_vars: self.env_vars.clone(),
+            conda_locator,
+            manager: self.manager.clone(),
+            versions_dir: self.versions_dir.clone(),
+        }
+    }
     fn clear(&self) {
         self.manager.lock().expect("manager mutex poisoned").take();
         self.versions_dir

crates/pet-python-utils/src/cache.rs

@@ -46,6 +46,10 @@ pub fn get_cache_directory() -> Option<PathBuf> {
 }
 
 pub fn set_cache_directory(cache_dir: PathBuf) {
+    CACHE.set_cache_directory(cache_dir);
+}
+
+pub fn set_cache_directory_and_get_effective(cache_dir: PathBuf) -> PathBuf {
     CACHE.set_cache_directory(cache_dir)
 }
 
@@ -75,24 +79,18 @@ impl CacheImpl {
 
     /// Once a cache directory has been set, you cannot change it.
     /// No point supporting such a scenario.
-    fn set_cache_directory(&self, cache_dir: PathBuf) {
-        if let Some(cache_dir) = self
-            .cache_dir
-            .lock()
-            .expect("cache_dir mutex poisoned")
-            .clone()
-        {
+    fn set_cache_directory(&self, cache_dir: PathBuf) -> PathBuf {
+        let mut effective = self.cache_dir.lock().expect("cache_dir mutex poisoned");
+        if let Some(existing) = effective.as_ref() {
             warn!(
                 "Cache directory has already been set to {:?}. Cannot change it now.",
-                cache_dir
+                existing
             );
-            return;
+            return existing.clone();
         }
         trace!("Setting cache directory to {:?}", cache_dir);
-        self.cache_dir
-            .lock()
-            .expect("cache_dir mutex poisoned")
-            .replace(cache_dir);
+        effective.replace(cache_dir.clone());
+        cache_dir
     }
     fn clear(&self) -> io::Result<()> {
         trace!("Clearing cache");
@@ -389,7 +387,11 @@ impl CacheEntry for CacheEntryImpl {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        mpsc, Barrier,
+    };
+    use std::time::Duration;
     use tempfile::tempdir_in;
 
     fn environment(executable: PathBuf, aliases: Vec<PathBuf>) -> ResolvedPythonEnv {
@@ -523,4 +525,75 @@ mod tests {
 
         assert!(entry.get().is_none());
     }
+
+    #[test]
+    fn concurrent_first_cache_directory_set_has_one_effective_path() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let cache = Arc::new(CacheImpl::new(None));
+        let barrier = Arc::new(Barrier::new(3));
+        let (result_tx, result_rx) = mpsc::channel();
+        let mut workers = Vec::new();
+
+        for name in ["cache-a", "cache-b"] {
+            let cache = cache.clone();
+            let barrier = barrier.clone();
+            let result_tx = result_tx.clone();
+            let requested = temp_dir.path().join(name);
+            workers.push(std::thread::spawn(move || {
+                barrier.wait();
+                result_tx
+                    .send(cache.set_cache_directory(requested))
+                    .unwrap();
+            }));
+        }
+
+        barrier.wait();
+        let first = result_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+        let second = result_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+        for worker in workers {
+            worker.join().unwrap();
+        }
+
+        assert_eq!(first, second);
+        assert_eq!(cache.get_cache_directory(), Some(first));
+    }
+
+    #[test]
+    fn in_flight_entry_can_repopulate_persistent_cache_after_clear() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let cache_directory = temp_dir.path().join("cache");
+        std::fs::create_dir_all(&cache_directory).unwrap();
+        let executable = temp_dir.path().join("python");
+        std::fs::write(&executable, "python").unwrap();
+        let cache = Arc::new(CacheImpl::new(Some(cache_directory)));
+        let entry = cache.create_cache(executable.clone());
+        let entry_guard = entry.lock().unwrap();
+        let (clear_done_tx, clear_done_rx) = mpsc::channel();
+        let clear_cache = cache.clone();
+
+        let clear_worker = std::thread::spawn(move || {
+            clear_done_tx.send(clear_cache.clear()).unwrap();
+        });
+        clear_done_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("clear must complete while the old entry is held")
+            .unwrap();
+
+        let expected = environment(executable.clone(), vec![executable.clone()]);
+        entry_guard.store(expected.clone());
+        drop(entry_guard);
+        clear_worker.join().unwrap();
+
+        let next_entry = cache.create_cache(executable);
+        assert!(!Arc::ptr_eq(&entry, &next_entry));
+        let repopulated = next_entry
+            .lock()
+            .unwrap()
+            .get()
+            .expect("post-clear write must be visible to the next entry");
+        assert_eq!(repopulated.executable, expected.executable);
+        assert_eq!(repopulated.prefix, expected.prefix);
+        assert_eq!(repopulated.version, expected.version);
+        assert_eq!(repopulated.symlinks, expected.symlinks);
+    }
 }

crates/pet-windows-registry/src/lib.rs

@@ -44,6 +44,12 @@ impl WindowsRegistry {
             search_result: Arc::new(Mutex::new(None)),
         }
     }
+    pub fn with_conda_locator(&self, conda_locator: Arc<dyn CondaLocator>) -> WindowsRegistry {
+        WindowsRegistry {
+            conda_locator,
+            search_result: self.search_result.clone(),
+        }
+    }
     #[cfg(windows)]
     fn find_with_cache(
         &self,
@@ -521,6 +527,67 @@ mod tests {
         );
     }
 
+    #[cfg(windows)]
+    #[test]
+    fn test_rebound_registry_replays_cached_conda_dirs_to_new_locator() {
+        use pet_core::manager::EnvManager;
+        use pet_core::python_environment::PythonEnvironment;
+        use pet_core::telemetry::TelemetryEvent;
+
+        struct NoopReporter;
+        impl Reporter for NoopReporter {
+            fn report_manager(&self, _manager: &EnvManager) {}
+            fn report_environment(&self, _env: &PythonEnvironment) {}
+            fn report_telemetry(&self, _event: &TelemetryEvent) {}
+        }
+
+        let temp = TempDir::new().unwrap();
+        let conda_prefix = temp.path().join("registry-conda");
+        fs::create_dir_all(conda_prefix.join("conda-meta")).unwrap();
+        fs::create_dir_all(conda_prefix.join("envs")).unwrap();
+        fs::create_dir_all(conda_prefix.join("Scripts")).unwrap();
+        fs::write(conda_prefix.join("conda-meta").join("history"), "").unwrap();
+        fs::write(
+            conda_prefix
+                .join("conda-meta")
+                .join("conda-24.1.0-py312_0.json"),
+            r#"{"version":"24.1.0"}"#,
+        )
+        .unwrap();
+        fs::write(
+            conda_prefix
+                .join("conda-meta")
+                .join("python-3.12.0-h123_0.json"),
+            r#"{"version":"3.12.0"}"#,
+        )
+        .unwrap();
+        fs::write(conda_prefix.join("Scripts").join("conda.bat"), "").unwrap();
+        fs::write(conda_prefix.join("python.exe"), "").unwrap();
+
+        let environment = EnvironmentApi::new();
+        let retired_conda = Arc::new(Conda::from(&environment));
+        let replacement_conda = Arc::new(Conda::from(&environment));
+        let registry = WindowsRegistry::from(retired_conda.clone());
+        registry
+            .search_result
+            .lock()
+            .unwrap()
+            .replace(Arc::new(CachedRegistryWalk {
+                result: LocatorResult {
+                    managers: Vec::new(),
+                    environments: Vec::new(),
+                },
+                conda_install_dirs: vec![conda_prefix.clone()],
+            }));
+        let rebound = registry.with_conda_locator(replacement_conda.clone());
+
+        rebound.find(&NoopReporter);
+
+        assert!(Arc::ptr_eq(&registry.search_result, &rebound.search_result));
+        assert!(replacement_conda.environments.contains_key(&conda_prefix));
+        assert!(!retired_conda.environments.contains_key(&conda_prefix));
+    }
+
     /// Smoke test: on a fresh locator (empty cache), `find()` runs the new
     /// parallel walk through HKLM and HKCU and never panics or deadlocks.
     /// The discovered environment list may legitimately be empty on a CI