Unify path identity fallbacks

core/metadata_audit.py

@@ -178,15 +178,7 @@ def _format_signed_size(bytes_value: int) -> str:
 
 
 def _model_key(model_path: str, metadata_path: str) -> Tuple[str, str]:
-    try:
-        model_identity = get_path_identity(model_path)
-    except (OSError, ValueError):
-        model_identity = os.path.normcase(os.path.abspath(model_path))
-    try:
-        metadata_identity = get_path_identity(metadata_path)
-    except (OSError, ValueError):
-        metadata_identity = os.path.normcase(os.path.abspath(metadata_path))
-    return model_identity, metadata_identity
+    return get_path_identity(model_path), get_path_identity(metadata_path)
 
 
 def _get_batch_size(

core/resolver.py

@@ -218,10 +218,7 @@ def _build_local_hash_match_cache(
             if not normalized_hash:
                 continue
 
-            try:
-                model_identity = get_path_identity(model_path)
-            except (OSError, ValueError):
-                model_identity = os.path.normcase(os.path.abspath(model_path))
+            model_identity = get_path_identity(model_path)
             entry_key = (normalized_hash, model_identity or model_path)
             if entry_key in seen_entries:
                 continue

core/scanner.py

@@ -27,16 +27,6 @@ _MODEL_FILES_CACHE_AT: float = 0.0
 _MODEL_FILES_CACHE_TTL_SECONDS = 2.0
 
 
-def _path_identity(path: str) -> str:
-    """Return a stable identity for dedupe across symlinks/junctions."""
-    return get_path_identity(path)
-
-
-def _directory_identity(path: str) -> str:
-    """Return a stable identity for loop detection across symlinks/junctions."""
-    return _path_identity(path)
-
-
 def get_model_directories() -> Dict[str, Tuple[List[str], set]]:
     """
     Get all configured model directories from folder_paths.
@@ -95,10 +85,7 @@ def scan_directory(
 
         # Walk through directory recursively
         for root, dirs, files in os.walk(base_directory, followlinks=True):
-            try:
-                root_identity = _directory_identity(root)
-            except (OSError, ValueError):
-                root_identity = os.path.normcase(os.path.abspath(root))
+            root_identity = get_path_identity(root)
 
             if root_identity in visited_dirs:
                 dirs[:] = []
@@ -113,10 +100,7 @@ def scan_directory(
                     continue
 
                 child_path = os.path.join(root, dirname)
-                try:
-                    child_identity = _directory_identity(child_path)
-                except (OSError, ValueError):
-                    child_identity = os.path.normcase(os.path.abspath(child_path))
+                child_identity = get_path_identity(child_path)
 
                 if child_identity in visited_dirs:
                     continue
@@ -232,10 +216,7 @@ def scan_all_directories() -> List[Dict[str, str]]:
 
         for directory_path in paths:
             try:
-                try:
-                    root_identity = _directory_identity(directory_path)
-                except (OSError, ValueError):
-                    root_identity = os.path.normcase(os.path.abspath(directory_path))
+                root_identity = get_path_identity(directory_path)
 
                 root_key = (category, root_identity)
                 if root_key in seen_scan_roots:

tests/test_metadata_model_helpers.py

@@ -3,8 +3,11 @@ import tempfile
 import unittest
 from unittest.mock import patch
 
+from core.metadata_audit import _model_key
 from core.metadata_model_utils import dedupe_models, is_model_file_path
 from core.path_utils import get_model_path_identity
+from core.resolver import _build_local_hash_match_cache
+from core.scanner import scan_directory
 
 
 class MetadataModelHelperTests(unittest.TestCase):
@@ -57,6 +60,47 @@ class MetadataModelHelperTests(unittest.TestCase):
         with patch("core.path_utils.os.path.realpath", side_effect=OSError):
             self.assertEqual(expected, get_model_path_identity(model_path))
 
+    def test_scanner_audit_and_resolver_share_path_identity_fallback(self):
+        model_path = os.path.join("models", "model.safetensors")
+        metadata_path = os.path.join("models", "model.metadata.json")
+        expected_model = os.path.normcase(os.path.abspath(model_path))
+        expected_metadata = os.path.normcase(os.path.abspath(metadata_path))
+
+        with patch("core.path_utils.os.path.realpath", side_effect=OSError):
+            self.assertEqual(
+                (expected_model, expected_metadata),
+                _model_key(model_path, metadata_path),
+            )
+
+            with patch("core.scanner.os.path.exists", return_value=True), patch(
+                "core.scanner.os.path.isdir", return_value=True
+            ), patch(
+                "core.scanner.os.walk",
+                return_value=[
+                    ("models", [], ["model.safetensors"]),
+                ],
+            ):
+                models = scan_directory("models", {".safetensors"}, "checkpoints")
+
+        self.assertEqual(["model.safetensors"], [model["filename"] for model in models])
+
+        with patch("core.resolver.os.path.isdir", return_value=False), patch(
+                "core.resolver.find_metadata_sidecar_path",
+                return_value=metadata_path,
+            ), patch(
+                "core.resolver.read_merged_model_metadata",
+                return_value={"sha256": "a" * 64},
+            ), patch(
+                "core.resolver._extract_model_sha256_from_metadata",
+                return_value=["a" * 64],
+        ):
+            index = _build_local_hash_match_cache(
+                [{"path": model_path, "filename": "model.safetensors"}]
+            )
+
+        self.assertEqual(1, len(index["a" * 64]))
+        self.assertEqual(model_path, index["a" * 64][0]["model"]["path"])
+
     def test_model_path_identity_rejects_empty_and_whitespace_paths(self):
         self.assertEqual("", get_model_path_identity(None))
         self.assertEqual("", get_model_path_identity(""))