fix: deduplicate Windows Conda casing aliases (#519)

## Summary
- preserve the canonical executable path on cache hits
- keep caller-facing aliases valid in the current working context
- invalidate stale in-memory entries when tracked executables disappear
- deduplicate Windows Conda casing aliases while preserving on-disk
spelling
- version performance inventory semantics so the intentional v1-to-v2
deduplication transition is explicit and same-schema count mismatches
remain blocking

Fixes #518
Related: microsoft/vscode-python-environments#1703

## Validation
- `cargo test -p pet-conda --test environment_locations_test`
- `cargo test --features ci-perf --test e2e_performance
test_performance_summary --no-run`
- `python -B -m unittest discover -s scripts/tests -p 'test_*.py' -v`
(47 passed)
- `./scripts/rust-precommit.ps1`
- replayed the failed Windows snapshot against its exact base: legacy
comparison fails at 8/1 vs 10/2; inventory schema v2 passes with all
latency metrics within budget

---------

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

Author: karthiknadig

Cargo.lock

@@ -21,5 +21,8 @@ env_logger = "0.10.2"
 yaml-rust2 = "0.8.1"
 rayon = "1.11.0"
 
+[dev-dependencies]
+tempfile = "3.13"
+
 [features]
 ci = []

crates/pet-conda/Cargo.toml

@@ -319,6 +319,25 @@ pub fn get_conda_envs_from_environment_txt(env_vars: &EnvVariables) -> Vec<PathB
     envs
 }
 
+#[cfg(windows)]
+fn restore_existing_leaf_case(path: PathBuf) -> PathBuf {
+    let Some(parent) = path.parent() else {
+        return path;
+    };
+    let Some(file_name) = path.file_name() else {
+        return path;
+    };
+    let Ok(entries) = fs::read_dir(parent) else {
+        return path;
+    };
+
+    entries
+        .filter_map(Result::ok)
+        .find(|entry| entry.file_name().eq_ignore_ascii_case(file_name))
+        .map(|entry| entry.path())
+        .unwrap_or(path)
+}
+
 #[cfg(windows)]
 pub fn get_known_conda_install_locations(
     env_vars: &EnvVariables,
@@ -416,15 +435,6 @@ pub fn get_known_conda_install_locations(
                 .join("conda"),
         );
     }
-    known_paths.sort();
-    known_paths.dedup();
-    // Ensure the casing of the paths are correct.
-    // Its possible the actual path is in a different case.
-    // E.g. instead of C:\username\miniconda it might bt C:\username\Miniconda
-    // We use lower cases above, but it could be in any case on disc.
-    // We do not want to have duplicates in different cases.
-    // & we'd like to preserve the case of the original path as on disc.
-    known_paths = known_paths.iter().map(norm_case).collect();
     if let Some(conda_dir) = get_conda_dir_from_exe(conda_executable) {
         known_paths.push(conda_dir);
     }
@@ -436,6 +446,13 @@ pub fn get_known_conda_install_locations(
     if let Some(mamba_dir) = get_conda_dir_from_exe(&find_mamba_binary(env_vars)) {
         known_paths.push(mamba_dir);
     }
+
+    known_paths = known_paths
+        .into_iter()
+        .filter(|path| path.exists())
+        .map(norm_case)
+        .map(restore_existing_leaf_case)
+        .collect();
     known_paths.sort();
     known_paths.dedup();
 

crates/pet-conda/src/environment_locations.rs

@@ -205,3 +205,46 @@ fn skips_path_lookup_when_conda_executable_provided() {
         locations
     );
 }
+
+#[cfg(windows)]
+#[test]
+fn deduplicates_windows_install_aliases_and_preserves_disk_casing() {
+    use common::create_env_variables;
+    use pet_conda::environment_locations::get_conda_environment_paths;
+    use pet_fs::path::norm_case;
+    use std::fs;
+
+    let temp_dir = tempfile::tempdir().expect("failed to create temporary test directory");
+    let home = temp_dir.path();
+    let install = home.join("Miniconda3");
+    let child = install.join("envs").join("MyEnv");
+
+    fs::create_dir_all(install.join("conda-meta"))
+        .expect("failed to create base conda-meta directory");
+    fs::create_dir_all(install.join("condabin")).expect("failed to create base condabin directory");
+    fs::create_dir_all(child.join("conda-meta"))
+        .expect("failed to create child conda-meta directory");
+
+    let conda_state = home.join(".conda");
+    fs::create_dir_all(&conda_state).expect("failed to create .conda directory");
+    fs::write(
+        conda_state.join("environments.txt"),
+        format!("{}\n{}\n", install.display(), child.display()),
+    )
+    .expect("failed to write environments.txt");
+
+    let mut env = create_env_variables(home.to_path_buf(), home.to_path_buf());
+    env.userprofile = Some(home.to_string_lossy().into_owned());
+
+    let environments = get_conda_environment_paths(&env, &None);
+    let normalized_home = norm_case(home);
+    let mut local_environments = environments
+        .into_iter()
+        .filter(|path| path.starts_with(&normalized_home))
+        .collect::<Vec<_>>();
+    local_environments.sort();
+
+    let mut expected = vec![norm_case(install), norm_case(child)];
+    expected.sort();
+    assert_eq!(local_environments, expected);
+}

crates/pet-conda/tests/environment_locations_test.rs

@@ -29,6 +29,7 @@ static REQUEST_ID: AtomicU32 = AtomicU32::new(1);
 /// Number of iterations for statistical tests
 const STAT_ITERATIONS: usize = 10;
 const PERFORMANCE_METRICS_SCHEMA_VERSION: u8 = 2;
+const PERFORMANCE_INVENTORY_SCHEMA_VERSION: u8 = 2;
 const STDERR_TAIL_LINES: usize = 100;
 
 /// Statistical metrics with percentile calculations
@@ -1571,6 +1572,7 @@ fn test_performance_summary() {
     // Existing top-level refresh fields remain warm-cache values for schema compatibility.
     let json_output = serde_json::to_string_pretty(&json!({
         "metrics_schema_version": PERFORMANCE_METRICS_SCHEMA_VERSION,
+        "inventory_schema_version": PERFORMANCE_INVENTORY_SCHEMA_VERSION,
         "server_startup_ms": startup_stats.p50().unwrap_or(0),
         "full_refresh_ms": warm_refresh_stats.p50().unwrap_or(0),
         "cold_refresh_ms": cold_refresh_stats.p50().unwrap_or(0),

crates/pet/tests/e2e_performance.rs

@@ -7,7 +7,7 @@ PET uses pull-request snapshots to prevent performance and coverage drift. Each
 The performance workflow runs 10 paired cache-cold/cache-warm JSON-RPC iterations on Linux, Windows, and macOS, plus 10 untimed cache-cold diagnostic iterations. A comparison is valid only when:
 
 - current and baseline metrics contain at least five samples for every required distribution;
-- environment and manager counts match exactly; and
+- environment and manager counts match exactly within the same inventory schema; and
 - the benchmark command and JSON extraction both succeed.
 
 A metric blocks when it exceeds both its absolute and relative budget:
@@ -32,6 +32,8 @@ The Windows warm full-refresh P50 budget was recalibrated in issue #513 from fiv
 
 Schema v2 records `full_refresh` and `time_to_first_env` from the warm member of each pair and adds cold refresh/time-to-first distributions. During its one-time rollout, comparisons against a schema-v1 base checked cold P50 against explicit absolute ceilings of 500ms on Linux, 750ms on Windows, and 1,000ms on macOS. Schema-v2-to-v2 comparisons use the table's dual budgets.
 
+Inventory schema v2 treats Windows Conda installation paths that differ only by on-disk casing as one logical workload entry. During the one-time v1-to-v2 transition, the report explicitly identifies the schema change and permits the expected count mismatch. Once the v2 baseline is published, exact environment and manager count matching resumes automatically.
+
 The cold P50 budgets were calibrated in issue #509 using two unchanged-head all-platform runs and the final pull-request validation.
 
 The dual budget avoids failing on tiny percentage changes while still blocking material latency regressions. Warm tail metrics remain mandatory; cold P95 remains diagnostic because a single host event can dominate it, while cold P50 blocks delays that affect the independent cold iterations consistently.

docs/QUALITY_SNAPSHOTS.md

@@ -106,6 +106,7 @@ def regressed(self) -> bool:
     ),
 }
 PERFORMANCE_METRICS_SCHEMA_VERSION = 2
+PERFORMANCE_INVENTORY_SCHEMA_VERSION = 2
 COLD_REFRESH_SPEC = MetricSpec('Cold refresh P50', 'cold_refresh', 'p50')
 COLD_DIAGNOSTIC_SPECS = (
     MetricSpec('Cold refresh P95', 'cold_refresh', 'p95'),
@@ -196,6 +197,20 @@ def performance_schema_version(snapshot: dict[str, Any], source: str) -> int:
     return version
 
 
+def inventory_schema_version(snapshot: dict[str, Any], source: str) -> int:
+    version = require_integer(
+        snapshot.get('inventory_schema_version', 1),
+        f'{source}.inventory_schema_version',
+        minimum=1,
+    )
+    if version > PERFORMANCE_INVENTORY_SCHEMA_VERSION:
+        raise SnapshotError(
+            f'{source}.inventory_schema_version {version} is newer than supported version '
+            f'{PERFORMANCE_INVENTORY_SCHEMA_VERSION}'
+        )
+    return version
+
+
 def cold_refresh_budget(platform: str) -> RegressionBudget:
     key = platform_key(platform)
     try:
@@ -230,16 +245,25 @@ def compare_performance(
             f'{baseline_version}'
         )
 
+    current_inventory_version = inventory_schema_version(current, 'current')
+    baseline_inventory_version = inventory_schema_version(baseline, 'baseline')
+    if current_inventory_version < baseline_inventory_version:
+        raise SnapshotError(
+            f'Current inventory schema {current_inventory_version} is older than baseline '
+            f'inventory schema {baseline_inventory_version}'
+        )
+
     current_envs = require_integer(current.get('environments_count'), 'current.environments_count', minimum=1)
     baseline_envs = require_integer(baseline.get('environments_count'), 'baseline.environments_count', minimum=1)
     current_managers = require_integer(current.get('managers_count'), 'current.managers_count')
     baseline_managers = require_integer(baseline.get('managers_count'), 'baseline.managers_count')
 
     failures: list[str] = []
-    if current_envs != baseline_envs:
-        failures.append(f'Environment inventory changed: current={current_envs}, baseline={baseline_envs}')
-    if current_managers != baseline_managers:
-        failures.append(f'Manager inventory changed: current={current_managers}, baseline={baseline_managers}')
+    if current_inventory_version == baseline_inventory_version:
+        if current_envs != baseline_envs:
+            failures.append(f'Environment inventory changed: current={current_envs}, baseline={baseline_envs}')
+        if current_managers != baseline_managers:
+            failures.append(f'Manager inventory changed: current={current_managers}, baseline={baseline_managers}')
 
     comparisons: list[PerformanceComparison] = [
         MetricComparison(
@@ -353,6 +377,8 @@ def performance_report(
     current: dict[str, Any],
     baseline: dict[str, Any],
 ) -> str:
+    current_inventory_version = inventory_schema_version(current, 'current')
+    baseline_inventory_version = inventory_schema_version(baseline, 'baseline')
     rows = []
     has_legacy_cold_baseline = False
     for comparison in comparisons:
@@ -392,12 +418,19 @@ def performance_report(
             '',
             '> Cold refresh uses a platform absolute ceiling while the exact base has legacy metrics.',
         ])
+    if current_inventory_version > baseline_inventory_version:
+        report.extend([
+            '',
+            '### Inventory schema transition',
+            f'- Inventory schema transitioned from v{baseline_inventory_version} to '
+            f'v{current_inventory_version}; exact count matching is skipped for this comparison.',
+        ])
     if failures:
         report.extend(['', '### Blocking findings', *[f'- {failure}' for failure in failures]])
     report.extend([
         '',
         '> A regression must exceed both the documented absolute and relative budget. '
-        'Environment and manager inventories must match exactly.',
+        'Environment and manager inventories must match exactly within the same inventory schema.',
     ])
     return '\n'.join(report) + '\n'
 

scripts/quality_snapshot.py

@@ -27,7 +27,7 @@ def performance_snapshot(
     *, refresh_p50=100, refresh_p95=500, startup_p50=10, startup_p95=20,
     first_p50=15, first_p95=30, cold_p50=200, cold_p95=500,
     cold_first_p50=25, cold_first_p95=50, environments=5, managers=1,
-    schema_version=1
+    schema_version=1, inventory_schema_version=None
 ):
     snapshot = {
         'server_startup_ms': startup_p50,
@@ -55,6 +55,8 @@ def performance_snapshot(
             'p50': cold_first_p50,
             'p95': cold_first_p95,
         }
+    if inventory_schema_version is not None:
+        snapshot['inventory_schema_version'] = inventory_schema_version
     return snapshot
 
 
@@ -282,6 +284,44 @@ def test_inventory_mismatch_fails(self):
         self.assertTrue(any('Environment inventory changed' in failure for failure in failures))
         self.assertTrue(any('Manager inventory changed' in failure for failure in failures))
 
+    def test_inventory_schema_transition_allows_count_change(self):
+        current = performance_snapshot(
+            environments=6,
+            managers=1,
+            inventory_schema_version=2,
+        )
+        baseline = performance_snapshot(environments=8, managers=2)
+
+        comparisons, failures = compare_performance(current, baseline, 'Windows')
+        report = performance_report('Windows', comparisons, failures, current, baseline)
+
+        self.assertEqual(failures, [])
+        self.assertIn('Inventory schema transitioned from v1 to v2', report)
+
+    def test_same_inventory_schema_still_requires_matching_counts(self):
+        current = performance_snapshot(environments=6, inventory_schema_version=2)
+        baseline = performance_snapshot(environments=8, inventory_schema_version=2)
+
+        _, failures = compare_performance(current, baseline, 'Windows')
+
+        self.assertTrue(any('Environment inventory changed' in failure for failure in failures))
+
+    def test_older_current_inventory_schema_is_invalid(self):
+        with self.assertRaisesRegex(SnapshotError, 'older than baseline inventory schema'):
+            compare_performance(
+                performance_snapshot(),
+                performance_snapshot(inventory_schema_version=2),
+                'Windows',
+            )
+
+    def test_newer_inventory_schema_is_invalid(self):
+        with self.assertRaisesRegex(SnapshotError, 'newer than supported version'):
+            compare_performance(
+                performance_snapshot(inventory_schema_version=3),
+                performance_snapshot(),
+                'Windows',
+            )
+
     def test_missing_metric_is_invalid(self):
         current = performance_snapshot()
         del current['stats']['full_refresh']['p95']