feat(metadata): add CivitAI AutoV3 hash support across all storage layers

- Three-state autov3 field (not-checked / checked-unavailable / 12-hex value)
  in .metadata.json sidecars, in-memory ModelHashIndex, and SQLite
  (models.autov3 column + autov3_index table) with column-presence migration
- Background self-terminating backfill for legacy rows: per-model-type
  concurrency guard, executor-offloaded I/O, Civitai-first resolution
  (SHA256-matched version file) falling back to the embedded safetensors
  header hash
- Civitai-first propagation on metadata refresh, scan, and download paths;
  reject the empty-string SHA256 placeholder and strip OneTrainer 0x prefix
- List API hash filters and hash index lookups accept 12-char AutoV3
- Cap safetensors header reads at 64 MiB to prevent crafted-file allocation
- Prevent stale AutoV3 mappings on file replacement while preserving them on
  same-file re-registration (lazy-hash completion)

tests/utils/test_file_utils.py

@@ -1,15 +1,26 @@
 import hashlib
+import json
 import os
+import struct
 
 import pytest
 
+from py.utils.constants import MAX_SAFETENSORS_HEADER_BYTES
 from py.utils.file_utils import (
+    calculate_autov3,
     calculate_sha256,
     find_preview_file,
     get_preview_extension,
 )
 
 
+def _write_safetensors(path, metadata, payload=b"payload-bytes"):
+    """Write a minimal real safetensors file: 8-byte little-endian header
+    length, a JSON header containing ``__metadata__``, then arbitrary payload."""
+    header = json.dumps({"__metadata__": metadata}).encode("utf-8")
+    path.write_bytes(struct.pack("<Q", len(header)) + header + payload)
+
+
 @pytest.mark.asyncio
 async def test_calculate_sha256(tmp_path):
     file_path = tmp_path / "sample.bin"
@@ -49,3 +60,126 @@ def test_find_preview_file_supports_example_extension(tmp_path):
 )
 def test_get_preview_extension(preview_name, expected):
     assert get_preview_extension(preview_name) == expected
+
+
+class TestCalculateAutov3:
+    def test_returns_first_12_chars_from_sshs_model_hash(self, tmp_path):
+        file_path = tmp_path / "lora.safetensors"
+        _write_safetensors(file_path, {"sshs_model_hash": "abcdef1234567890abcdef"})
+
+        assert calculate_autov3(str(file_path)) == "abcdef123456"
+
+    def test_lowercases_uppercase_embedded_hash(self, tmp_path):
+        file_path = tmp_path / "upper.safetensors"
+        _write_safetensors(file_path, {"sshs_model_hash": "ABCDEF1234567890ABCDEF"})
+
+        assert calculate_autov3(str(file_path)) == "abcdef123456"
+
+    def test_returns_first_12_chars_from_modelspec_hash_sha256(self, tmp_path):
+        file_path = tmp_path / "model.safetensors"
+        _write_safetensors(file_path, {"modelspec.hash_sha256": "00112233445566778899aabb"})
+
+        assert calculate_autov3(str(file_path)) == "001122334455"
+
+    def test_prefers_sshs_model_hash_over_modelspec_hash(self, tmp_path):
+        file_path = tmp_path / "model.safetensors"
+        _write_safetensors(
+            file_path,
+            {
+                "sshs_model_hash": "aaaaaaaaaaaaaaaaaaaaaaaa",
+                "modelspec.hash_sha256": "bbbbbbbbbbbbbbbbbbbbbbbb",
+            },
+        )
+
+        assert calculate_autov3(str(file_path)) == "aaaaaaaaaaaa"
+
+    def test_returns_none_for_non_safetensors_file(self, tmp_path):
+        file_path = tmp_path / "plain.bin"
+        file_path.write_bytes(b"just some plain bytes, not a safetensors file")
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_for_empty_file(self, tmp_path):
+        file_path = tmp_path / "empty.safetensors"
+        file_path.write_bytes(b"")
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_for_header_length_shorter_than_8_bytes(self, tmp_path):
+        file_path = tmp_path / "short.safetensors"
+        file_path.write_bytes(b"\x10\x00")
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_for_truncated_json_header(self, tmp_path):
+        file_path = tmp_path / "truncated.safetensors"
+        file_path.write_bytes(struct.pack("<Q", 100) + b'{"__metadata__":')
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_when_metadata_lacks_recognized_hash(self, tmp_path):
+        file_path = tmp_path / "model.safetensors"
+        _write_safetensors(file_path, {"ss_model_name": "something-else"})
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_when_embedded_hash_is_too_short(self, tmp_path):
+        file_path = tmp_path / "model.safetensors"
+        _write_safetensors(file_path, {"sshs_model_hash": "abc123"})
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_when_embedded_hash_is_not_a_string(self, tmp_path):
+        file_path = tmp_path / "model.safetensors"
+        _write_safetensors(file_path, {"sshs_model_hash": 123456})
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_strips_0x_prefix_from_modelspec_hash(self, tmp_path):
+        # OneTrainer writes modelspec.hash_sha256 with a "0x" prefix.
+        file_path = tmp_path / "onetrainer.safetensors"
+        _write_safetensors(
+            file_path, {"modelspec.hash_sha256": "0x1585b50b9d7d66778899aabb"}
+        )
+
+        assert calculate_autov3(str(file_path)) == "1585b50b9d7d"
+
+    def test_returns_none_for_empty_string_sha256_placeholder(self, tmp_path):
+        # Repackaging tools sometimes write the SHA256-of-empty placeholder
+        # instead of a real hash; it must not be treated as a valid AutoV3.
+        file_path = tmp_path / "broken.safetensors"
+        _write_safetensors(
+            file_path,
+            {"sshs_model_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"},
+        )
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_strips_0x_prefix_from_empty_hash_placeholder(self, tmp_path):
+        file_path = tmp_path / "onetrainer_broken.safetensors"
+        _write_safetensors(
+            file_path,
+            {"modelspec.hash_sha256": "0xe3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"},
+        )
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_returns_none_for_header_length_over_limit(self, tmp_path):
+        # A crafted file whose 64-bit header length exceeds the cap must not
+        # trigger a giant allocation; it is rejected before any read.
+        file_path = tmp_path / "huge_header.safetensors"
+        file_path.write_bytes(
+            struct.pack("<Q", MAX_SAFETENSORS_HEADER_BYTES + 1) + b'{"__metadata__": {}}'
+        )
+
+        assert calculate_autov3(str(file_path)) is None
+
+    def test_reads_header_at_exact_limit_boundary(self, tmp_path):
+        # A header length exactly at the cap is still valid.
+        file_path = tmp_path / "at_limit.safetensors"
+        file_path.write_bytes(
+            struct.pack("<Q", MAX_SAFETENSORS_HEADER_BYTES) + b'{"__metadata__": {"sshs_model_hash": "abcdef1234567890abcdef"}}'
+        )
+
+        # The read returns {} because the actual bytes are shorter than the
+        # claimed length (short-read guard), without allocating anything huge.
+        assert calculate_autov3(str(file_path)) is None

tests/utils/test_metadata_defaults.py

@@ -1,10 +1,18 @@
 import json
+import struct
 import pytest
 
 from py.utils.metadata_manager import MetadataManager
 from py.utils.models import BaseModelMetadata
 
 
+def _write_safetensors(path, metadata, payload=b"payload-bytes"):
+    """Write a minimal real safetensors file: 8-byte little-endian header
+    length, a JSON header containing ``__metadata__``, then arbitrary payload."""
+    header = json.dumps({"__metadata__": metadata}).encode("utf-8")
+    path.write_bytes(struct.pack("<Q", len(header)) + header + payload)
+
+
 @pytest.mark.asyncio
 async def test_base_model_metadata_sets_empty_civitai_dict():
     metadata = BaseModelMetadata(
@@ -35,3 +43,35 @@ async def test_create_default_metadata_uses_empty_civitai(tmp_path):
     payload = json.loads(metadata_path.read_text(encoding="utf-8"))
 
     assert payload.get("civitai") == {}
+
+
+@pytest.mark.asyncio
+async def test_create_default_metadata_writes_autov3_from_safetensors(tmp_path):
+    model_path = tmp_path / "example.safetensors"
+    _write_safetensors(model_path, {"sshs_model_hash": "abcdef1234567890abcdef"})
+
+    metadata = await MetadataManager.create_default_metadata(str(model_path))
+
+    assert metadata is not None
+    assert metadata.autov3 == "abcdef123456"
+
+    metadata_path = model_path.with_suffix(".metadata.json")
+    payload = json.loads(metadata_path.read_text(encoding="utf-8"))
+
+    assert payload.get("autov3") == "abcdef123456"
+
+
+@pytest.mark.asyncio
+async def test_create_default_metadata_writes_autov3_null_for_non_safetensors(tmp_path):
+    model_path = tmp_path / "example.safetensors"
+    model_path.write_bytes(b"stub")
+
+    metadata = await MetadataManager.create_default_metadata(str(model_path))
+
+    assert metadata is not None
+    assert metadata.autov3 == ""
+
+    metadata_path = model_path.with_suffix(".metadata.json")
+    payload = json.loads(metadata_path.read_text(encoding="utf-8"))
+
+    assert payload.get("autov3") is None

tests/utils/test_models_sub_type.py

@@ -125,3 +125,200 @@ class TestLoraMetadataConsistency:
         
         # Type is stored in civitai dict
         assert metadata.civitai.get("model", {}).get("type") == "Lora"
+
+
+class TestBaseModelMetadataAutov3:
+    """Three-state autov3 semantics on BaseModelMetadata (None/''/12-hex)."""
+
+    def _make_dict(self, **overrides):
+        data = {
+            "file_name": "model",
+            "model_name": "Model",
+            "file_path": "/tmp/model.safetensors",
+            "size": 0,
+            "modified": 0.0,
+            "sha256": "deadbeef",
+            "base_model": "Unknown",
+            "preview_url": "",
+        }
+        data.update(overrides)
+        return data
+
+    def test_from_dict_null_autov3_normalizes_to_empty_string(self):
+        metadata = BaseModelMetadata.from_dict(self._make_dict(autov3=None))
+
+        assert metadata.autov3 == ""
+
+    def test_to_dict_emits_autov3_null_when_checked_unavailable(self):
+        metadata = BaseModelMetadata.from_dict(self._make_dict(autov3=None))
+
+        payload = metadata.to_dict()
+
+        assert "autov3" in payload
+        assert payload["autov3"] is None
+
+    def test_from_dict_absent_autov3_stays_none(self):
+        metadata = BaseModelMetadata.from_dict(self._make_dict())
+
+        assert metadata.autov3 is None
+
+    def test_to_dict_omits_autov3_when_not_checked(self):
+        metadata = BaseModelMetadata.from_dict(self._make_dict())
+
+        assert "autov3" not in metadata.to_dict()
+
+    def test_from_dict_preserves_lowercase_autov3_value(self):
+        metadata = BaseModelMetadata.from_dict(self._make_dict(autov3="abcdef123456"))
+
+        assert metadata.autov3 == "abcdef123456"
+        assert metadata.to_dict()["autov3"] == "abcdef123456"
+
+    @pytest.mark.parametrize("model_cls", [LoraMetadata, CheckpointMetadata, EmbeddingMetadata])
+    def test_from_civitai_info_extracts_autov3(self, model_cls):
+        # Civitai versions can ship multiple files; AutoV3 is taken from the
+        # file whose SHA256 matches the downloaded file.
+        version_info = {
+            "baseModel": "SDXL",
+            "model": {"name": "Test", "description": "", "tags": []},
+            "files": [
+                {"name": "other.safetensors", "sizeKB": 100, "hashes": {"SHA256": "zzz999"}},
+                {
+                    "name": "model.safetensors",
+                    "sizeKB": 1000,
+                    "hashes": {"SHA256": "abc123", "AutoV3": "ABCDEF1234567890ABCDEF"},
+                },
+            ],
+        }
+        file_info = {
+            "name": "model.safetensors",
+            "sizeKB": 1000,
+            "hashes": {"SHA256": "abc123"},
+        }
+
+        metadata = model_cls.from_civitai_info(version_info, file_info, "/test/model.safetensors")
+
+        assert metadata.autov3 == "abcdef123456"
+
+    @pytest.mark.parametrize(
+        "hashes",
+        [
+            {},
+            {"SHA256": "abc123"},
+            {"SHA256": "abc123", "AutoV3": "abc"},
+            {"SHA256": "abc123", "AutoV3": 123},
+            {"SHA256": "abc123", "AutoV3": None},
+        ],
+    )
+    @pytest.mark.parametrize("model_cls", [LoraMetadata, CheckpointMetadata, EmbeddingMetadata])
+    def test_from_civitai_info_autov3_none_when_missing_or_invalid(self, model_cls, hashes):
+        version_info = {
+            "baseModel": "SDXL",
+            "model": {"name": "Test", "description": "", "tags": []},
+            "files": [{"name": "model.safetensors", "sizeKB": 1000, "hashes": hashes}],
+        }
+        file_info = {"name": "model.safetensors", "sizeKB": 1000, "hashes": {"SHA256": "abc123"}}
+
+        metadata = model_cls.from_civitai_info(version_info, file_info, "/test/model.safetensors")
+
+        assert metadata.autov3 is None
+
+    def test_autov3_from_civitai_files_matches_sha256_case_insensitively(self):
+        from py.utils.models import autov3_from_civitai_files
+
+        civitai = {
+            "files": [
+                {"name": "a.safetensors", "hashes": {"SHA256": "111AAA"}},
+                {"name": "b.safetensors", "hashes": {"SHA256": "222BBB", "AutoV3": "ABCDEF123456"}},
+            ]
+        }
+        assert autov3_from_civitai_files(civitai, "222bbb") == "abcdef123456"
+        assert autov3_from_civitai_files(civitai, "111aaa") is None
+        assert autov3_from_civitai_files(civitai, "999999") is None
+        assert autov3_from_civitai_files(None, "222bbb") is None
+        assert autov3_from_civitai_files(civitai, "") is None
+
+    def test_autov3_from_civitai_files_ignores_files_without_sha256(self):
+        from py.utils.models import autov3_from_civitai_files
+
+        civitai = {
+            "files": [
+                {"name": "a.safetensors", "hashes": {"AutoV3": "ABCDEF123456"}},
+                {"name": "b.safetensors", "hashes": {"SHA256": "222BBB", "AutoV3": "123456ABCDEF"}},
+            ]
+        }
+        assert autov3_from_civitai_files(civitai, "222bbb") == "123456abcdef"
+
+    def test_autov3_from_civitai_files_rejects_empty_hash_placeholder(self):
+        from py.utils.models import autov3_from_civitai_files
+
+        civitai = {
+            "files": [
+                {"name": "b.safetensors", "hashes": {"SHA256": "222BBB", "AutoV3": "E3B0C44298FC1C149AFB..."}},
+            ]
+        }
+        # The empty-string SHA256 placeholder must never be adopted as a value.
+        assert autov3_from_civitai_files(civitai, "222bbb") is None
+
+    def test_update_civitai_info_populates_autov3_from_matching_file(self):
+        metadata = CheckpointMetadata(
+            file_name="kreamania",
+            model_name="Kreamania",
+            file_path="/test/kreamania_variant5.safetensors",
+            size=1000,
+            modified=1234567890.0,
+            sha256="111AABBF94DD9E59C05D842FCCF57BEC915B2A3C237F6B54F8D614E40858D717",
+            base_model="FLUX.1 D",
+            preview_url="",
+            autov3="",
+        )
+        version_info = {
+            "files": [
+                {
+                    "name": "kreamania_variant5.safetensors",
+                    "hashes": {"SHA256": "111aabbf94dd9e59c05d842fccf57bec915b2a3c237f6b54f8d614e40858d717", "AutoV3": "8A582E901D7F"},
+                }
+            ]
+        }
+
+        metadata.update_civitai_info(version_info)
+
+        assert metadata.autov3 == "8a582e901d7f"
+
+    def test_update_civitai_info_prefers_civitai_over_header_value(self):
+        metadata = BaseModelMetadata(
+            file_name="model",
+            model_name="Model",
+            file_path="/test/model.safetensors",
+            size=1000,
+            modified=1234567890.0,
+            sha256="abc123",
+            base_model="SDXL",
+            preview_url="",
+            autov3="e3b0c44298fc",  # stale header-extracted value
+        )
+        version_info = {
+            "files": [
+                {"name": "model.safetensors", "hashes": {"SHA256": "abc123", "AutoV3": "DEF456ABC789"}}
+            ]
+        }
+
+        metadata.update_civitai_info(version_info)
+
+        assert metadata.autov3 == "def456abc789"
+
+    def test_update_civitai_info_keeps_autov3_when_no_match(self):
+        metadata = BaseModelMetadata(
+            file_name="model",
+            model_name="Model",
+            file_path="/test/model.safetensors",
+            size=1000,
+            modified=1234567890.0,
+            sha256="abc123",
+            base_model="SDXL",
+            preview_url="",
+            autov3="",
+        )
+
+        metadata.update_civitai_info({"files": [{"name": "other.safetensors", "hashes": {"SHA256": "zzz999", "AutoV3": "DEF456ABC789"}}]})
+
+        assert metadata.autov3 == ""