fix: honor stat-aware cache lookup contract

modelaudit/cache/cache_manager.py

@@ -87,6 +87,7 @@ def get_cached_result_with_stat(
         cached_result, file_identity = self.get_cached_result_with_identity(
             file_path,
             version_context=version_context,
+            file_stat=stat_result,
             include_private_metadata=include_private_metadata,
         )
         try:
@@ -100,17 +101,21 @@ def get_cached_result_with_identity(
         file_path: str,
         version_context: dict[str, Any] | None = None,
         *,
+        file_stat: os.stat_result | None = None,
         include_private_metadata: bool = False,
     ) -> tuple[dict[str, Any] | None, ScannedFileIdentity | None]:
         """Return a cache lookup and retain the monitored identity for a miss scan."""
         if not self.enabled or not self.cache:
             return None, None
 
-        cached_result, file_identity = self.cache.get_cached_result_with_identity(
-            file_path,
-            version_context=version_context,
-            include_private_metadata=include_private_metadata,
-        )
+        lookup_kwargs: dict[str, Any] = {
+            "version_context": version_context,
+            "include_private_metadata": include_private_metadata,
+        }
+        if file_stat is not None:
+            lookup_kwargs["file_stat"] = file_stat
+
+        cached_result, file_identity = self.cache.get_cached_result_with_identity(file_path, **lookup_kwargs)
         if cached_result is not None and not cached_scan_result_dependencies_available(cached_result):
             logger.debug(f"Bypassing cached result with unavailable scanner dependencies: {Path(file_path).name}")
             return None, file_identity

modelaudit/cache/scan_results_cache.py

@@ -460,6 +460,7 @@ def get_cached_result_with_stat(
         cached_result, file_identity = self.get_cached_result_with_identity(
             file_path,
             version_context=version_context,
+            file_stat=file_stat,
             include_private_metadata=include_private_metadata,
         )
         if file_identity is not None:
@@ -471,6 +472,7 @@ def get_cached_result_with_identity(
         file_path: str,
         version_context: dict[str, Any] | None = None,
         *,
+        file_stat: os.stat_result | None = None,
         include_private_metadata: bool = False,
     ) -> tuple[dict[str, Any] | None, ScannedFileIdentity | None]:
         """Return a cache lookup plus the monitored identity reusable by a miss scan."""
@@ -480,7 +482,10 @@ def get_cached_result_with_identity(
                 logger.debug("Bypassing scan-result cache lookup for symlinked path %s", file_path)
                 return None, None
 
-            file_identity = self.capture_file_identity(file_path)
+            if file_stat is None:
+                file_identity = self.capture_file_identity(file_path)
+            else:
+                file_identity = self.capture_file_identity(file_path, file_stat=file_stat)
             file_stat, file_hash, _change_token, _ancestor_identity = file_identity
             cache_key, _content_hash = self._generate_cache_key_material(
                 file_path,
@@ -854,14 +859,20 @@ def store_result(
                 temporary_cache_path.unlink(missing_ok=True)
             self.release_ancestor_identity(expected_ancestor_identity)
 
-    def capture_file_identity(self, file_path: str) -> ScannedFileIdentity:
-        """Capture a stable stat, content hash, and platform change token before scanning."""
+    def capture_file_identity(
+        self,
+        file_path: str,
+        file_stat: os.stat_result | None = None,
+    ) -> ScannedFileIdentity:
+        """Capture a stable file identity, reusing a caller stat snapshot for the first probe when available."""
         if self._path_has_symlink_component(file_path):
             raise ValueError(f"Symlinked paths are not cacheable: {file_path}")
 
         last_change_error: ValueError | None = None
+        stat_hint = file_stat
         for _capture_attempt in range(_MAX_IDENTITY_CAPTURE_ATTEMPTS):
-            preliminary_stat = os.stat(file_path)
+            preliminary_stat = stat_hint if stat_hint is not None else os.stat(file_path)
+            stat_hint = None
             probe = self._get_change_clock_probe(file_path, preliminary_stat.st_dev)
             preliminary_change_token = self._get_file_change_token(file_path, preliminary_stat)
             preliminary_ancestor_identity = self._capture_ancestor_identity(file_path)

tests/cache/test_cache_correctness.py

@@ -671,6 +671,79 @@ def capture_once(path: str) -> Any:
     assert capture_calls == 1
 
 
+def test_scan_results_cache_get_cached_result_with_stat_reuses_provided_stat(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    file_path = _make_cacheable_file(tmp_path, name="scan-results-stat.cache")
+    cache = ScanResultsCache(str(tmp_path / "scan-cache"))
+    version_context = build_cache_version_context({"timeout": 30})
+    expected = {"checks": [], "issues": [], "metadata": {}, "scanner": "test", "success": True}
+    file_identity = cache.capture_file_identity(str(file_path))
+    recorded_stats: list[os.stat_result | None] = []
+    provided_stat = os.stat(file_path)
+
+    def lookup_with_record(
+        path: str,
+        version_context: dict[str, Any] | None = None,
+        *,
+        file_stat: os.stat_result | None = None,
+        include_private_metadata: bool = False,
+    ) -> Any:
+        assert path == str(file_path)
+        assert version_context is not None
+        assert include_private_metadata is False
+        recorded_stats.append(file_stat)
+        return expected, file_identity
+
+    monkeypatch.setattr(cache, "get_cached_result_with_identity", lookup_with_record)
+
+    cached_result = cache.get_cached_result_with_stat(str(file_path), provided_stat, version_context=version_context)
+
+    assert cached_result == expected
+    assert len(recorded_stats) == 1
+    assert recorded_stats[0] is provided_stat
+
+
+def test_cache_manager_get_cached_result_with_stat_reuses_provided_stat(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    file_path = _make_cacheable_file(tmp_path, name="cache-manager-stat.cache")
+    cache_manager = get_cache_manager(str(tmp_path / "cache"), enabled=True)
+    assert cache_manager.cache is not None
+    version_context = build_cache_version_context({"timeout": 30})
+    expected = {"checks": [], "issues": [], "metadata": {}, "scanner": "test", "success": True}
+    file_identity = cache_manager.cache.capture_file_identity(str(file_path))
+    recorded_stats: list[os.stat_result | None] = []
+    provided_stat = os.stat(file_path)
+
+    def lookup_with_record(
+        path: str,
+        version_context: dict[str, Any] | None = None,
+        *,
+        file_stat: os.stat_result | None = None,
+        include_private_metadata: bool = False,
+    ) -> Any:
+        assert path == str(file_path)
+        assert version_context is not None
+        assert include_private_metadata is False
+        recorded_stats.append(file_stat)
+        return expected, file_identity
+
+    monkeypatch.setattr(cache_manager.cache, "get_cached_result_with_identity", lookup_with_record)
+
+    cached_result = cache_manager.get_cached_result_with_stat(
+        str(file_path),
+        provided_stat,
+        version_context=version_context,
+    )
+
+    assert cached_result == expected
+    assert len(recorded_stats) == 1
+    assert recorded_stats[0] is provided_stat
+
+
 def test_cached_scan_does_not_store_result_after_post_scan_replacement(tmp_path: Path) -> None:
     file_path = _make_cacheable_file(tmp_path, name="race.dat")
     clean_payload = b"clean:" + (b"x" * 2042)