test: validate session resource and cache invariants

Allow ambient cache additions while preserving original entries, validate derived resource metrics exactly, and surface Linux descriptor enumeration errors. Address PR #563 feedback and native CI failures.

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

Author: karthiknadig

crates/pet/tests/session_performance.rs

@@ -3,7 +3,9 @@
 
 use pet_fs::path::norm_case;
 use serde_json::json;
+use std::collections::BTreeMap;
 use std::fs;
+use std::io;
 use std::path::{Path, PathBuf};
 use std::process::Command;
 use std::thread;
@@ -420,6 +422,55 @@ fn directory_usage(root: &Path) -> (usize, u64) {
     (files, bytes)
 }
 
+fn cache_contents(root: &Path) -> io::Result<BTreeMap<PathBuf, Vec<u8>>> {
+    let mut pending = vec![(PathBuf::new(), root.to_path_buf())];
+    let mut contents = BTreeMap::new();
+    while let Some((relative_directory, directory)) = pending.pop() {
+        for entry in fs::read_dir(directory)? {
+            let entry = entry?;
+            let file_type = entry.file_type()?;
+            let relative_path = relative_directory.join(entry.file_name());
+            if file_type.is_dir() {
+                pending.push((relative_path, entry.path()));
+            } else if file_type.is_file() {
+                contents.insert(relative_path, fs::read(entry.path())?);
+            }
+        }
+    }
+    Ok(contents)
+}
+
+fn original_cache_entries_unchanged(
+    original: &BTreeMap<PathBuf, Vec<u8>>,
+    current: &BTreeMap<PathBuf, Vec<u8>>,
+) -> bool {
+    original
+        .iter()
+        .all(|(path, bytes)| current.get(path) == Some(bytes))
+}
+
+#[test]
+fn cache_preservation_allows_additions_but_rejects_original_entry_changes() {
+    let original = BTreeMap::from([
+        (PathBuf::from("flat.json"), b"original".to_vec()),
+        (
+            PathBuf::from("nested").join("entry.json"),
+            b"nested".to_vec(),
+        ),
+    ]);
+    let mut with_addition = original.clone();
+    with_addition.insert(PathBuf::from("ambient.json"), b"ambient".to_vec());
+    assert!(original_cache_entries_unchanged(&original, &with_addition));
+
+    let mut changed = with_addition.clone();
+    changed.insert(PathBuf::from("flat.json"), b"modified".to_vec());
+    assert!(!original_cache_entries_unchanged(&original, &changed));
+
+    let mut deleted = with_addition;
+    deleted.remove(&PathBuf::from("nested").join("entry.json"));
+    assert!(!original_cache_entries_unchanged(&original, &deleted));
+}
+
 fn observe_resources(pid: u32) -> ResourceSample {
     let samples = (0..3)
         .map(|_| {
@@ -534,8 +585,8 @@ fn sample_process(pid: u32) -> ResourceSample {
         descriptors: Some(
             fs::read_dir(format!("/proc/{pid}/fd"))
                 .expect("failed to read process-specific descriptor directory")
-                .map(|entry| entry.expect("failed to read process descriptor entry"))
-                .count(),
+                .try_fold(0, |count, entry| entry.map(|_| count + 1))
+                .expect("failed to read process descriptor entry"),
         ),
     }
 }
@@ -901,9 +952,10 @@ fn long_lived_session_benchmark() {
 
     fixture.clear_barrier();
     let pre_resolve_resources = observe_resources(client.process_id());
-    let original_cache_before_overlap = directory_usage(&fixture.cache);
+    let original_cache_before_overlap =
+        cache_contents(&fixture.cache).expect("failed to capture pre-overlap cache contents");
     assert!(
-        original_cache_before_overlap.0 > 0 && original_cache_before_overlap.1 > 0,
+        !original_cache_before_overlap.is_empty(),
         "warm-up resolve did not populate the original process cache"
     );
     let overlap_workspace = fixture._root.path().join("overlap-workspace");
@@ -972,10 +1024,11 @@ fn long_lived_session_benchmark() {
     max_ambient_environment_count =
         max_ambient_environment_count.max(overlap_ambient_environment_count);
     max_ambient_manager_count = max_ambient_manager_count.max(overlap_ambient_manager_count);
-    assert_eq!(
-        directory_usage(&fixture.cache),
-        original_cache_before_overlap,
-        "overlap reconfiguration did not retain the original cache contents"
+    let cache_after_overlap =
+        cache_contents(&fixture.cache).expect("failed to capture post-overlap cache contents");
+    assert!(
+        original_cache_entries_unchanged(&original_cache_before_overlap, &cache_after_overlap),
+        "overlap reconfiguration modified or deleted an original cache entry"
     );
     assert_eq!(
         fixture.released_count(),

docs/SESSION_BENCHMARKS.md

@@ -53,8 +53,11 @@ The workflow keeps Cargo output in a target directory inside its checkout and
 uploads only the privacy-safe `session-metrics.json` artifact, never fixture
 paths or raw benchmark output. A dedicated parser rejects missing, malformed,
 duplicate, mode-inconsistent, count-inconsistent, or shape-inconsistent
-payloads. Artifact writes are atomic, and the timeout fallback replaces corrupt
-artifacts rather than uploading invalid JSON. Failed runs retain validated
+payloads. It also recomputes the observed resource peak from every reported
+snapshot and the RSS delta from the pre-resolve and final snapshots, rejecting
+either derived value unless it matches exactly. Artifact writes are atomic, and
+the timeout fallback replaces corrupt artifacts rather than uploading invalid
+JSON. Failed runs retain validated
 measurements when available; failures without usable metrics have explicit failed
 status and zero counts. The workflow preserves the original benchmark failure.
 
@@ -64,8 +67,10 @@ real paths keep macOS temporary-directory aliases equivalent, while ambient
 interpreters bypass it. During the proof pass,
 every distinct interpreter process records entry before the client issues
 `info`, reconfigures to a new workspace while retaining the process's original
-cache directory, and refreshes that known inventory. The barrier remains held
-until those responsiveness checks complete.
+cache directory, and refreshes that known inventory. Every cache file captured
+before reconfiguration must still exist with identical bytes afterward; cache
+files added for unrelated global discoveries are allowed. The barrier remains
+held until those responsiveness checks complete.
 Platform-global locators may also report host installations and managers. The
 benchmark converts only configured workspace entries to strict fixture
 identities, validates that fixture-scoped managers remain empty, and records

scripts/session_metrics.py

@@ -91,14 +91,27 @@ def require_integer_list(value: Any, name: str, expected_length: int) -> list[in
     return [require_integer(item, f"{name}[{index}]") for index, item in enumerate(value)]
 
 
-def validate_resource(value: Any, name: str) -> None:
+def validate_resource(value: Any, name: str) -> dict[str, Any]:
     resource = require_object(
         value, name, {"residentBytes", "threads", "handlesOrDescriptors"}
     )
     require_integer(resource["residentBytes"], f"{name}.residentBytes")
     for key in ("threads", "handlesOrDescriptors"):
         if resource[key] is not None:
             require_integer(resource[key], f"{name}.{key}")
+    return resource
+
+
+def observed_resource_peak(samples: Sequence[dict[str, Any]]) -> dict[str, Any]:
+    def optional_max(key: str) -> int | None:
+        known = [sample[key] for sample in samples if sample[key] is not None]
+        return max(known) if known else None
+
+    return {
+        "residentBytes": max(sample["residentBytes"] for sample in samples),
+        "threads": optional_max("threads"),
+        "handlesOrDescriptors": optional_max("handlesOrDescriptors"),
+    }
 
 
 def validate_cache_value(value: Any, name: str) -> dict[str, Any]:
@@ -299,6 +312,7 @@ def validate_success_metrics(value: Any, expected_mode: str) -> dict[str, Any]:
     if not isinstance(inventory_resources, list) or len(inventory_resources) != len(expected_sizes):
         raise MetricsError("inventoryResourceSamples must contain one entry per size")
     seen_resource_sizes = set()
+    resource_samples = []
     for index, value in enumerate(inventory_resources):
         sample = require_object(
             value,
@@ -311,7 +325,11 @@ def validate_success_metrics(value: Any, expected_mode: str) -> dict[str, Any]:
                 f"inventoryResourceSamples[{index}] has an invalid or duplicate size"
             )
         seen_resource_sizes.add(size)
-        validate_resource(sample["resources"], f"inventoryResourceSamples[{index}].resources")
+        resource_samples.append(
+            validate_resource(
+                sample["resources"], f"inventoryResourceSamples[{index}].resources"
+            )
+        )
 
     batch_resources = metrics["resolveBatchResourceSamples"]
     if not isinstance(batch_resources, list) or len(batch_resources) != expected_batches:
@@ -329,17 +347,42 @@ def validate_success_metrics(value: Any, expected_mode: str) -> dict[str, Any]:
                 f"resolveBatchResourceSamples[{index}] has an invalid or duplicate batch"
             )
         seen_batches.add(batch)
-        validate_resource(sample["resources"], f"resolveBatchResourceSamples[{index}].resources")
+        resource_samples.append(
+            validate_resource(
+                sample["resources"], f"resolveBatchResourceSamples[{index}].resources"
+            )
+        )
 
     for name in (
         "preResolveResources",
         "barrierObservedResources",
         "postOverlapResources",
-        "observedResourcePeak",
         "resourceAfter",
     ):
-        validate_resource(metrics[name], name)
-    require_integer(metrics["rssDeltaFromPreResolveBytes"], "rssDeltaFromPreResolveBytes", minimum=-(2**63))
+        resource_samples.append(validate_resource(metrics[name], name))
+    reported_peak = validate_resource(
+        metrics["observedResourcePeak"], "observedResourcePeak"
+    )
+    expected_peak = observed_resource_peak(resource_samples)
+    if reported_peak != expected_peak:
+        raise MetricsError(
+            f"observedResourcePeak must equal {expected_peak}, got {reported_peak}"
+        )
+
+    reported_delta = require_integer(
+        metrics["rssDeltaFromPreResolveBytes"],
+        "rssDeltaFromPreResolveBytes",
+        minimum=-(2**63),
+    )
+    expected_delta = (
+        metrics["resourceAfter"]["residentBytes"]
+        - metrics["preResolveResources"]["residentBytes"]
+    )
+    if reported_delta != expected_delta:
+        raise MetricsError(
+            "rssDeltaFromPreResolveBytes must equal "
+            f"{expected_delta}, got {reported_delta}"
+        )
     return metrics
 
 

scripts/tests/test_session_metrics.py

@@ -19,11 +19,11 @@
                              failed_metrics, write_json)  # noqa: E402
 
 
-def resource():
+def resource(resident=100, threads=2, descriptors=3):
     return {
-        "residentBytes": 100,
-        "threads": 2,
-        "handlesOrDescriptors": 3,
+        "residentBytes": resident,
+        "threads": threads,
+        "handlesOrDescriptors": descriptors,
     }
 
 
@@ -100,11 +100,11 @@ def valid_metrics():
         "resolveBatchResourceSamples": [
             {"batch": batch, "resources": resource()} for batch in range(2)
         ],
-        "preResolveResources": resource(),
+        "preResolveResources": resource(resident=100),
         "barrierObservedResources": resource(),
         "postOverlapResources": resource(),
         "observedResourcePeak": resource(),
-        "resourceAfter": resource(),
+        "resourceAfter": resource(resident=90),
         "rssDeltaFromPreResolveBytes": -10,
     }
 
@@ -163,6 +163,13 @@ def write_payloads(self, *payloads):
     def artifact(self):
         return json.loads(self.output.read_text(encoding="utf-8"))
 
+    def assert_metrics_rejected(self, metrics, message):
+        self.write_payloads(json.dumps(metrics))
+        with self.assertRaisesRegex(MetricsError, message):
+            extract_metrics(self.input, self.output, 0, "fast")
+        self.assertEqual(self.artifact()["status"], "failed")
+        self.assertFalse(self.artifact()["metricsProduced"])
+
     def test_no_metrics_fails_closed_and_persists_failure(self):
         self.input.write_text("benchmark output only\n", encoding="utf-8")
         with self.assertRaisesRegex(MetricsError, "exactly one"):
@@ -224,6 +231,83 @@ def test_success_preserves_validated_metrics(self):
         self.assertTrue(metrics["metricsProduced"])
         self.assertEqual(len(metrics["refreshScenarioSamples"]), 9)
 
+    def test_resource_peak_rejects_low_and_high_values_for_every_field(self):
+        for field in ("residentBytes", "threads", "handlesOrDescriptors"):
+            for difference in (-1, 1):
+                with self.subTest(field=field, difference=difference):
+                    metrics = valid_metrics()
+                    metrics["observedResourcePeak"][field] += difference
+                    self.assert_metrics_rejected(metrics, "observedResourcePeak")
+
+    def test_resource_peak_includes_every_sample_category(self):
+        categories = (
+            ("inventory", lambda metrics: metrics["inventoryResourceSamples"][0]["resources"]),
+            ("pre-resolve", lambda metrics: metrics["preResolveResources"]),
+            ("barrier", lambda metrics: metrics["barrierObservedResources"]),
+            ("post-overlap", lambda metrics: metrics["postOverlapResources"]),
+            ("resolve-batch", lambda metrics: metrics["resolveBatchResourceSamples"][0]["resources"]),
+            ("after", lambda metrics: metrics["resourceAfter"]),
+        )
+        for name, select_sample in categories:
+            with self.subTest(category=name):
+                metrics = valid_metrics()
+                sample = select_sample(metrics)
+                sample.update(
+                    residentBytes=1000,
+                    threads=1001,
+                    handlesOrDescriptors=1002,
+                )
+                metrics["observedResourcePeak"] = resource(1000, 1001, 1002)
+                metrics["rssDeltaFromPreResolveBytes"] = (
+                    metrics["resourceAfter"]["residentBytes"]
+                    - metrics["preResolveResources"]["residentBytes"]
+                )
+                self.write_payloads(json.dumps(metrics))
+                extract_metrics(self.input, self.output, 0, "fast")
+
+    def test_resource_peak_optional_fields_support_all_null_and_mixed_samples(self):
+        metrics = valid_metrics()
+        samples = [
+            *(entry["resources"] for entry in metrics["inventoryResourceSamples"]),
+            metrics["preResolveResources"],
+            metrics["barrierObservedResources"],
+            metrics["postOverlapResources"],
+            *(entry["resources"] for entry in metrics["resolveBatchResourceSamples"]),
+            metrics["resourceAfter"],
+        ]
+        for sample in samples:
+            sample["threads"] = None
+            sample["handlesOrDescriptors"] = None
+        metrics["observedResourcePeak"]["threads"] = None
+        metrics["observedResourcePeak"]["handlesOrDescriptors"] = None
+        self.write_payloads(json.dumps(metrics))
+        extract_metrics(self.input, self.output, 0, "fast")
+
+        samples[0]["threads"] = 7
+        samples[-1]["threads"] = 5
+        samples[1]["handlesOrDescriptors"] = 8
+        samples[-2]["handlesOrDescriptors"] = 6
+        metrics["observedResourcePeak"]["threads"] = 7
+        metrics["observedResourcePeak"]["handlesOrDescriptors"] = 8
+        self.write_payloads(json.dumps(metrics))
+        extract_metrics(self.input, self.output, 0, "fast")
+
+    def test_rss_delta_accepts_positive_negative_and_zero_values(self):
+        for after, expected_delta in ((110, 10), (90, -10), (100, 0)):
+            with self.subTest(expected_delta=expected_delta):
+                metrics = valid_metrics()
+                metrics["resourceAfter"]["residentBytes"] = after
+                metrics["rssDeltaFromPreResolveBytes"] = expected_delta
+                if after > metrics["observedResourcePeak"]["residentBytes"]:
+                    metrics["observedResourcePeak"]["residentBytes"] = after
+                self.write_payloads(json.dumps(metrics))
+                extract_metrics(self.input, self.output, 0, "fast")
+
+    def test_inconsistent_rss_delta_fails_closed_with_failure_artifact(self):
+        metrics = valid_metrics()
+        metrics["rssDeltaFromPreResolveBytes"] = -9
+        self.assert_metrics_rejected(metrics, "rssDeltaFromPreResolveBytes")
+
     def test_malformed_nested_types_fail_closed_with_failure_artifacts(self):
         fields = [
             ("sizes", 0), ("samplesPerSize",), ("resolveConcurrency",),